package edu.washington.escience.myria.io;

import java.io.IOException;
import java.io.OutputStream;
import java.net.URI;
import java.util.Objects;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import com.fasterxml.jackson.annotation.JsonProperty;

import edu.washington.escience.myria.coordinator.CatalogException;

public class UriSink implements DataSink {
  /** Required for Java serialization. */
  private static final long serialVersionUID = 1L;

  @JsonProperty
  private final URI uri;

  public UriSink(@JsonProperty(value = "uri", required = true) final String uri) throws CatalogException {
    this.uri = URI.create(Objects.requireNonNull(uri, "Parameter uri cannot be null"));
    if (!this.uri.getScheme().equals("hdfs")) {
      throw new CatalogException("URI must be an HDFS URI");
    }
  }

  @Override
  public OutputStream getOutputStream() throws IOException {
    Configuration conf = new Configuration();

    FileSystem fs = FileSystem.get(uri, conf);
    Path rootPath = new Path(uri);
    return fs.create(rootPath);
  }
}

@@ -625,22 +625,16 @@
  = true)
- final
  String 
@@ -660,24 +660,98 @@
 Exception %7B%0A
+    if (uri.contains(%22s3%22)) %7B%0A      uri = uri.replace(%22s3%22, %22s3n%22);%0A    %7D%0A
     this.uri
@@ -870,16 +870,55 @@
 (%22hdfs%22)
+ && !this.uri.getScheme().equals(%22s3n%22)
 ) %7B%0A    
@@ -967,16 +967,22 @@
 an HDFS 
+or S3 
 URI%22);%0A 

add s3 option