TAJO-1359: Add nested field projector and language extension to project nested record.

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

@@ -171,6 +171,16 @@ public static boolean isDelimited(String identifier) {
     return openQuote && closeQuote;
   }
 
+  /**
+   * True if a given name is a simple identifier, meaning is not a dot-chained name.
+   *
+   * @param columnOrTableName Column or Table name to be checked
+   * @return True if a given name is a simple identifier. Otherwise, it will return False.
+   */
+  public static boolean isSimpleIdentifier(String columnOrTableName) {
+    return columnOrTableName.split(CatalogConstants.IDENTIFIER_DELIMITER_REGEXP).length == 1;
+  }
+
   public static boolean isFQColumnName(String tableName) {
     return tableName.split(CatalogConstants.IDENTIFIER_DELIMITER_REGEXP).length == 3;
   }
@@ -662,7 +672,7 @@ public static DataType getWidestType(DataType...types) {
       if (types[i].getType() != Type.NULL_TYPE) {
         Type candidate = TUtil.getFromNestedMap(OPERATION_CASTING_MAP, widest.getType(), types[i].getType());
         if (candidate == null) {
-          throw new InvalidOperationException("No matched operation for those types: " + TUtil.arrayToString
+          throw new InvalidOperationException("No matched operation for those types: " + StringUtils.join
               (types));
         }
         widest = newSimpleDataType(candidate);

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

@@ -0,0 +1,109 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.catalog;
+
+import com.google.common.base.Preconditions;
+import org.apache.tajo.common.TajoDataTypes.Type;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Utility methods for nested field
+ */
+public class NestedPathUtil {
+  public static final String PATH_DELIMITER = "/";
+
+  public static final List<String> ROOT_PATH = Collections.unmodifiableList(new ArrayList<String>());
+
+  public static boolean isPath(String name) {
+    return name.indexOf(PATH_DELIMITER.charAt(0)) >= 0;
+  }
+
+  public static String makePath(String[] parts) {
+    return makePath(parts, 0);
+  }
+
+  public static String makePath(String[] parts, int startIndex) {
+    return makePath(parts, startIndex, parts.length);
+  }
+
+  /**
+   * Make a nested field path
+   *
+   * @param parts path parts
+   * @param startIndex startIndex
+   * @param depth Depth
+   * @return Path
+   */
+  public static String makePath(String[] parts, int startIndex, int depth) {
+    Preconditions.checkArgument(startIndex <= (parts.length - 1));
+
+    StringBuilder sb = new StringBuilder();
+    for (int i = startIndex; i < depth; i++) {
+      sb.append(PATH_DELIMITER);
+      sb.append(parts[i].toString());
+    }
+
+    return sb.toString();
+  }
+
+  /**
+   * Lookup the actual column corresponding to a given path.
+   * We assume that a path starts with the slash '/' and it
+   * does not include the root field.
+   *
+   * @param nestedField Nested column
+   * @param path Path which starts with '/';
+   * @return Column corresponding to the path
+   */
+  public static Column lookupPath(Column nestedField, String path) {
+    Preconditions.checkArgument(path.charAt(0) == PATH_DELIMITER.charAt(0),
+        "A nested field path must start with slash '/'.");
+
+    // We assume that path starts with '/', causing an empty string "" at 0 in the path splits.
+    // So, we should start the index from 1 instead of 0.
+    return lookupPath(nestedField, path.split(PATH_DELIMITER));
+  }
+
+  public static Column lookupPath(Column nestedField, String [] paths) {
+    // We assume that path starts with '/', causing an empty string "" at 0 in the path splits.
+    // So, we should start the index from 1 instead of 0.
+    return lookupColumnInternal(nestedField, paths, 1);
+  }
+
+  private static Column lookupColumnInternal(Column currentColumn, String [] paths, int depth) {
+    Column found = null;
+
+    if (currentColumn.getDataType().getType() == Type.RECORD) {
+      found = currentColumn.typeDesc.nestedRecordSchema.getColumn(paths[depth]);
+    }
+
+    if (found != null) {
+      if (found.getDataType().getType() == Type.RECORD) {
+        return lookupColumnInternal(found, paths, depth + 1);
+      } else {
+        return found;
+      }
+    } else {
+      throw new NoSuchFieldError(makePath(paths));
+    }
+  }
+}

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

@@ -32,6 +32,7 @@
 import org.apache.tajo.common.TajoDataTypes.DataType;
 import org.apache.tajo.common.TajoDataTypes.Type;
 import org.apache.tajo.json.GsonObject;
+import org.apache.tajo.util.StringUtils;
 import org.apache.tajo.util.TUtil;
 
 import java.util.*;
@@ -135,7 +136,8 @@ private void init() {
    * @param qualifier The qualifier
    */
   public void setQualifier(String qualifier) {
-    List<Column> columns = getColumns();
+    // only change root fields, and must keep each nested field simple name
+    List<Column> columns = getRootColumns();
 
     fields.clear();
     fieldsByQualifiedName.clear();
@@ -180,14 +182,39 @@ public int getIndex(Column column) {
    * @return The column matched to a given column name.
    */
   public Column getColumn(String name) {
-    String [] parts = name.split("\\.");
-    // Some of the string can includes database name and table name and column name.
-    // For example, it can be 'default.table1.id'.
-    // Therefore, spilt string array length can be 3.
-    if (parts.length >= 2) {
-      return getColumnByQName(name);
+
+    if (NestedPathUtil.isPath(name)) {
+
+      // TODO - to be refactored
+      if (fieldsByQualifiedName.containsKey(name)) {
+        Column flattenColumn = fields.get(fieldsByQualifiedName.get(name));
+        if (flattenColumn != null) {
+          return flattenColumn;
+        }
+      }
+
+      String [] paths = name.split(NestedPathUtil.PATH_DELIMITER);
+      Column column = getColumn(paths[0]);
+      if (column == null) {
+        return null;
+      }
+      Column actualColumn = NestedPathUtil.lookupPath(column, paths);
+
+      Column columnPath = new Column(
+          column.getQualifiedName() + NestedPathUtil.makePath(paths, 1),
+          actualColumn.typeDesc);
+
+      return columnPath;
     } else {
-      return getColumnByName(name);
+      String[] parts = name.split("\\.");
+      // Some of the string can includes database name and table name and column name.
+      // For example, it can be 'default.table1.id'.
+      // Therefore, spilt string array length can be 3.
+      if (parts.length >= 2) {
+        return getColumnByQName(name);
+      } else {
+        return getColumnByName(name);
+      }
     }
   }
 
@@ -268,12 +295,46 @@ public int getColumnIdByName(String colName) {
     }
     return -1;
   }
-
-	public List<Column> getColumns() {
+
+  /**
+   * Get root columns, meaning all columns except for nested fields.
+   *
+   * @return A list of root columns
+   */
+	public List<Column> getRootColumns() {
 		return ImmutableList.copyOf(fields);
 	}
 
+  /**
+   * Get all columns, including all nested fields
+   *
+   * @return A list of all columns
+   */
+  public List<Column> getAllColumns() {
+    final List<Column> columnList = TUtil.newList();
+
+    SchemaUtil.visitSchema(this, new ColumnVisitor() {
+      @Override
+      public void visit(int depth, List<String> path, Column column) {
+        if (path.size() > 0) {
+          String parentPath = StringUtils.join(path, NestedPathUtil.PATH_DELIMITER);
+          String currentPath = parentPath + NestedPathUtil.PATH_DELIMITER + column.getSimpleName();
+          columnList.add(new Column(currentPath, column.getTypeDesc()));
+        } else {
+          columnList.add(column);
+        }
+      }
+    });
+
+    return columnList;
+  }
+
   public boolean contains(String name) {
+    // TODO - It's a hack
+    if (NestedPathUtil.isPath(name)) {
+      return (getColumn(name) != null);
+    }
+
     if (fieldsByQualifiedName.containsKey(name)) {
       return true;
     }
@@ -288,6 +349,11 @@ public boolean contains(String name) {
   }
 
   public boolean contains(Column column) {
+    // TODO - It's a hack
+    if (NestedPathUtil.isPath(column.getQualifiedName())) {
+      return (getColumn(column.getQualifiedName()) != null);
+    }
+
     if (column.hasQualifier()) {
       return fieldsByQualifiedName.containsKey(column.getQualifiedName());
     } else {
@@ -314,7 +380,24 @@ public boolean containsByName(String colName) {
   }
 
   public boolean containsAll(Collection<Column> columns) {
-    return fields.containsAll(columns);
+    boolean containFlag = true;
+
+    for (Column c :columns) {
+      if (NestedPathUtil.isPath(c.getSimpleName())) {
+        if (contains(c.getQualifiedName())) {
+          containFlag &= true;
+        } else {
+          String[] paths = c.getQualifiedName().split("/");
+          boolean existRootPath = contains(paths[0]);
+          boolean existLeafPath = getColumn(c.getSimpleName()) != null;
+          containFlag &= existRootPath && existLeafPath;
+        }
+      } else {
+        containFlag &= fields.contains(c);
+      }
+    }
+
+    return containFlag;
   }
 
   public synchronized Schema addColumn(String name, TypeDesc typeDesc) {
@@ -351,7 +434,7 @@ public synchronized void addColumn(Column column) {
 	}
 
 	public synchronized void addColumns(Schema schema) {
-    for(Column column : schema.getColumns()) {
+    for(Column column : schema.getRootColumns()) {
       addColumn(column);
     }
   }
@@ -396,7 +479,7 @@ public SchemaProtoBuilder(SchemaProto.Builder builder) {
     }
 
     @Override
-    public void visit(int depth, Column column) {
+    public void visit(int depth, List<String> path, Column column) {
 
       if (column.getDataType().getType() == Type.RECORD) {
         DataType.Builder updatedType = DataType.newBuilder(column.getDataType());

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

@@ -18,6 +18,10 @@
 
 package org.apache.tajo.catalog;
 
+import org.apache.tajo.util.TUtil;
+
+import java.util.List;
+
 import static org.apache.tajo.common.TajoDataTypes.DataType;
 import static org.apache.tajo.common.TajoDataTypes.Type;
 
@@ -34,12 +38,12 @@ public class SchemaUtil {
   static int tmpColumnSeq = 0;
   public static Schema merge(Schema left, Schema right) {
     Schema merged = new Schema();
-    for(Column col : left.getColumns()) {
+    for(Column col : left.getRootColumns()) {
       if (!merged.containsByQualifiedName(col.getQualifiedName())) {
         merged.addColumn(col);
       }
     }
-    for(Column col : right.getColumns()) {
+    for(Column col : right.getRootColumns()) {
       if (merged.containsByQualifiedName(col.getQualifiedName())) {
         merged.addColumn("?fake" + (tmpColumnSeq++), col.getDataType());
       } else {
@@ -59,7 +63,7 @@ public static Schema merge(Schema left, Schema right) {
    */
   public static Schema getNaturalJoinColumns(Schema left, Schema right) {
     Schema common = new Schema();
-    for (Column outer : left.getColumns()) {
+    for (Column outer : left.getRootColumns()) {
       if (!common.containsByName(outer.getSimpleName()) && right.containsByName(outer.getSimpleName())) {
         common.addColumn(new Column(outer.getSimpleName(), outer.getDataType()));
       }
@@ -113,7 +117,7 @@ public static <T extends Schema> T clone(Schema schema) {
    * Column visitor interface
    */
   public static interface ColumnVisitor {
-    public void visit(int depth, Column column);
+    public void visit(int depth, List<String> path, Column column);
   }
 
   /**
@@ -122,8 +126,8 @@ public static interface ColumnVisitor {
    * @param function
    */
   public static void visitSchema(Schema schema, ColumnVisitor function) {
-      for(Column col : schema.getColumns()) {
-        visitInDepthFirstOrder(0, function, col);
+      for(Column col : schema.getRootColumns()) {
+        visitInDepthFirstOrder(0, NestedPathUtil.ROOT_PATH, function, col);
       }
   }
 
@@ -134,14 +138,21 @@ public static void visitSchema(Schema schema, ColumnVisitor function) {
    * @param function Visitor
    * @param column Current visiting column
    */
-  private static void visitInDepthFirstOrder(int depth, ColumnVisitor function, Column column) {
+  private static void visitInDepthFirstOrder(int depth,
+                                             final List<String> path,
+                                             ColumnVisitor function,
+                                             Column column) {
+
     if (column.getDataType().getType() == Type.RECORD) {
-      for (Column nestedColumn : column.typeDesc.nestedRecordSchema.getColumns()) {
-        visitInDepthFirstOrder(depth + 1, function, nestedColumn);
+      for (Column nestedColumn : column.typeDesc.nestedRecordSchema.getRootColumns()) {
+        List<String> newPath = TUtil.newList(path);
+        newPath.add(column.getQualifiedName());
+
+        visitInDepthFirstOrder(depth + 1, newPath, function, nestedColumn);
       }
-      function.visit(depth, column);
+      function.visit(depth, path, column);
     } else {
-      function.visit(depth, column);
+      function.visit(depth, path, column);
     }
   }