TAJO-2111: Optimize Partition Table Split Computation for Amazon S3

CHANGES

@@ -2,6 +2,7 @@ Tajo Change Log
 
 Release 0.12.0 - unreleased
 
+
   NEW FEATURES
 
     TAJO-1686: Allow Tajo to use Hive UDF. (jihoon)

tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogUtil.java

@@ -332,6 +332,11 @@ public static TableDesc newTableDesc(String tableName, Schema schema, TableMeta
     return new TableDesc(tableName, schema, meta, path.toUri());
   }
 
+  public static TableDesc newTableDesc(String tableName, Schema schema, TableMeta meta, Path path
+    , PartitionMethodDesc partitionMethodDesc) {
+    return new TableDesc(tableName, schema, meta, path.toUri(), partitionMethodDesc);
+  }
+
   public static TableDesc newTableDesc(TableDescProto proto) {
     return new TableDesc(proto);
   }

tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/TableDesc.java

@@ -51,32 +51,46 @@ public class TableDesc implements ProtoObject<TableDescProto>, GsonObject, Clone
 	public TableDesc() {
 	}
 
+  public TableDesc(String tableName, @Nullable Schema schema, TableMeta meta, @Nullable URI uri, boolean external) {
+    this(tableName, schema, meta, uri, null, external);
+  }
+
   public TableDesc(String tableName, @Nullable Schema schema, TableMeta meta,
-                   @Nullable URI uri, boolean external) {
+                   @Nullable URI uri, @Nullable PartitionMethodDesc partitionMethodDesc, boolean external) {
     this.tableName = tableName;
     this.schema = schema;
     this.meta = meta;
     this.uri = uri;
+    this.partitionMethodDesc = partitionMethodDesc;
     this.external = external;
   }
 
 	public TableDesc(String tableName, @Nullable Schema schema, TableMeta meta, @Nullable URI path) {
-		this(tableName, schema, meta, path, true);
+		this(tableName, schema, meta, path, null, true);
 	}
-
+
+  public TableDesc(String tableName, @Nullable Schema schema, TableMeta meta, @Nullable URI path,
+                   @Nullable PartitionMethodDesc partitionMethodDesc) {
+    this(tableName, schema, meta, path, partitionMethodDesc, true);
+  }
+
 	public TableDesc(String tableName, @Nullable Schema schema, String dataFormat, KeyValueSet options,
                    @Nullable URI path) {
 	  this(tableName, schema, new TableMeta(dataFormat, options), path);
 	}
-
-	public TableDesc(TableDescProto proto) {
+
+  public TableDesc(String tableName, @Nullable Schema schema, String dataFormat, KeyValueSet options,
+                   @Nullable URI path, @Nullable PartitionMethodDesc partitionMethodDesc) {
+    this(tableName, schema, new TableMeta(dataFormat, options), path, partitionMethodDesc);
+  }
+
+  public TableDesc(TableDescProto proto) {
 	  this(proto.getTableName(), proto.hasSchema() ? SchemaFactory.newV1(proto.getSchema()) : null,
-        new TableMeta(proto.getMeta()), proto.hasPath() ? URI.create(proto.getPath()) : null, proto.getIsExternal());
+        new TableMeta(proto.getMeta()), proto.hasPath() ? URI.create(proto.getPath()) : null,
+      proto.hasPartition() ? new PartitionMethodDesc(proto.getPartition()) : null,
+      proto.getIsExternal());
     if(proto.hasStats()) {
       this.stats = new TableStats(proto.getStats());
-    }
-    if (proto.hasPartition()) {
-      this.partitionMethodDesc = new PartitionMethodDesc(proto.getPartition());
     }
 	}
 

tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java

@@ -232,6 +232,15 @@ public static enum ConfVars implements ConfigKey {
     // for RCFile
     HIVEUSEEXPLICITRCFILEHEADER("tajo.exec.rcfile.use.explicit.header", true, Validators.bool()),
 
+    // S3 Configuration --------------------------------------------------
+    S3_MAX_ERROR_RETRIES("tajo.s3.max-error-retries", 10),
+    S3_SSL_ENABLED("tajo.s3.ssl.enabled", true),
+    S3_CONNECT_TIMEOUT("tajo.s3.connect-timeout", "5m"),
+    S3_SOCKET_TIMEOUT("tajo.s3.socket-timeout", "5m"),
+    S3_MAX_CONNECTIONS("tajo.s3.max-connections", 500),
+    S3_USE_INSTANCE_CREDENTIALS("tajo.s3.use-instance-credentials", true),
+    S3_PIN_CLIENT_TO_CURRENT_REGION("tajo.s3.pin-client-to-current-region", false),
+
     // RPC --------------------------------------------------------------------
     //  Internal RPC Client
     INTERNAL_RPC_CLIENT_WORKER_THREAD_NUM("tajo.internal.rpc.client.worker-thread-num",

tajo-common/src/main/java/org/apache/tajo/util/FileUtil.java

@@ -142,4 +142,42 @@ public static void cleanupAndthrowIfFailed(java.io.Closeable... closeables) thro
       throw ioe;
     }
   }
+
+  public static String getCommonPrefix(Path... paths){
+    String prefix = "";
+    String[][] folders = new String[paths.length][];
+
+    // split on file separator
+    for(int i = 0; i < paths.length; i++){
+      folders[i] = paths[i].toString().split("/");
+    }
+
+    for(int j = 0; j < folders[0].length; j++){
+      // grab the next folder name in the first path
+      String thisFolder = folders[0][j];
+      // assume all have matched in case there are no more paths
+      boolean allMatched = true;
+
+      // look at the other paths
+      for(int i = 1; i < folders.length && allMatched; i++){
+        // if there is no folder here
+        if(folders[i].length < j){
+          allMatched = false;
+          // stop looking because we've gone as far as we can
+          break;
+        }
+        // check if it matched
+        allMatched &= folders[i][j].equals(thisFolder);
+      }
+      // if they all matched this folder name
+      if(allMatched) {
+        // add it to the answer
+        prefix += thisFolder + "/";
+      } else {
+        break;
+      }
+    }
+
+    return prefix;
+  }
 }