TAJO-1723: INSERT INTO statement should allow nested fields as target columns.

tajo-algebra/src/main/java/org/apache/tajo/algebra/Insert.java

@@ -32,7 +32,7 @@ public class Insert extends Expr {
   @Expose @SerializedName("TableName")
   private String tableName;
   @Expose @SerializedName("TargetColumns")
-  private String [] targetColumns;
+  private ColumnReferenceExpr [] targetColumns;
   @Expose @SerializedName("StorageType")
   private String storageType;
   @Expose @SerializedName("Location")
@@ -70,11 +70,11 @@ public boolean hasTargetColumns() {
     return targetColumns != null;
   }
 
-  public String [] getTargetColumns() {
+  public ColumnReferenceExpr [] getTargetColumns() {
     return targetColumns;
   }
 
-  public void setTargetColumns(String [] targets) {
+  public void setTargetColumns(ColumnReferenceExpr [] targets) {
     this.targetColumns = targets;
   }
 

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

@@ -19,11 +19,11 @@
 package org.apache.tajo.catalog;
 
 import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Lists;
 import org.apache.tajo.common.TajoDataTypes.Type;
 
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
+import java.util.*;
 
 /**
  * Utility methods for nested field
@@ -65,24 +65,6 @@ public static String makePath(String[] parts, int startIndex, int depth) {
     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.
@@ -106,4 +88,5 @@ private static Column lookupColumnInternal(Column currentColumn, String [] paths
       throw new NoSuchFieldError(makePath(paths));
     }
   }
+
 }

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

@@ -18,8 +18,11 @@
 
 package org.apache.tajo.catalog;
 
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Lists;
 import org.apache.tajo.util.TUtil;
 
+import java.util.HashMap;
 import java.util.List;
 
 import static org.apache.tajo.common.TajoDataTypes.DataType;
@@ -113,11 +116,44 @@ public static <T extends Schema> T clone(Schema schema) {
     return names;
   }
 
+  public static String [] convertColumnsToPaths(Iterable<Column> columns, boolean onlyLeaves) {
+    List<String> paths = Lists.newArrayList();
+
+    for (Column c : columns) {
+      if (onlyLeaves && c.getDataType().getType() == Type.RECORD) {
+        continue;
+      }
+      paths.add(c.getSimpleName());
+    }
+
+    return paths.toArray(new String [paths.size()]);
+  }
+
+  public static ImmutableMap<String, Type> buildTypeMap(Iterable<Column> schema, String [] targetPaths) {
+
+    HashMap<String, Type> builder = new HashMap<String, Type>();
+    for (Column column : schema) {
+
+      // Keep types which only belong to projected paths
+      // For example, assume that a projected path is 'name/first_name', where name is RECORD and first_name is TEXT.
+      // In this case, we should keep two types:
+      // * name - RECORD
+      // * name/first_name TEXT
+      for (String p : targetPaths) {
+        if (p.startsWith(column.getSimpleName())) {
+          builder.put(column.getSimpleName(), column.getDataType().getType());
+        }
+      }
+    }
+
+    return ImmutableMap.copyOf(builder);
+  }
+
   /**
    * Column visitor interface
    */
-  public static interface ColumnVisitor {
-    public void visit(int depth, List<String> path, Column column);
+  public interface ColumnVisitor {
+    void visit(int depth, List<String> path, Column column);
   }
 
   /**

tajo-core/src/main/antlr4/org/apache/tajo/engine/parser/SQLParser.g4

@@ -1593,7 +1593,7 @@ null_ordering
 */
 
 insert_statement
-  : INSERT (OVERWRITE)? INTO table_name (LEFT_PAREN column_name_list RIGHT_PAREN)? query_expression
+  : INSERT (OVERWRITE)? INTO table_name (LEFT_PAREN column_reference_list RIGHT_PAREN)? query_expression
   | INSERT (OVERWRITE)? INTO LOCATION path=Character_String_Literal (USING storage_type=identifier (param_clause)?)? query_expression
   ;
 

tajo-core/src/main/java/org/apache/tajo/engine/parser/SQLAnalyzer.java

@@ -1621,10 +1621,12 @@ public Expr visitInsert_statement(SQLParser.Insert_statementContext ctx) {
     if (ctx.table_name() != null) {
       insertExpr.setTableName(ctx.table_name().getText());
 
-      if (ctx.column_name_list() != null) {
-        String[] targetColumns = new String[ctx.column_name_list().identifier().size()];
+      if (ctx.column_reference_list() != null) {
+        ColumnReferenceExpr [] targetColumns =
+            new ColumnReferenceExpr[ctx.column_reference_list().column_reference().size()];
+
         for (int i = 0; i < targetColumns.length; i++) {
-          targetColumns[i] = ctx.column_name_list().identifier().get(i).getText();
+          targetColumns[i] = visitColumn_reference(ctx.column_reference_list().column_reference(i));
         }
 
         insertExpr.setTargetColumns(targetColumns);

tajo-core/src/test/java/org/apache/tajo/engine/query/TestSelectNestedRecord.java

@@ -30,7 +30,7 @@
 public class TestSelectNestedRecord extends QueryTestCaseBase {
 
   @Test
-  public final void testSelect1() throws Exception {
+  public final void testSelect0() throws Exception {
     List<String> tables = executeDDL("sample1_ddl.sql", "sample1", "sample1");
     assertEquals(TUtil.newList("sample1"), tables);
 
@@ -39,6 +39,16 @@ public final void testSelect1() throws Exception {
     cleanupQuery(res);
   }
 
+  @Test
+  public final void testSelect1() throws Exception {
+    List<String> tables = executeDDL("sample1_ddl.sql", "sample1", "sample2");
+    assertEquals(TUtil.newList("sample2"), tables);
+
+    ResultSet res = executeQuery();
+    assertResultSet(res);
+    cleanupQuery(res);
+  }
+
   @Test
   public final void testSelect2() throws Exception {
     List<String> tables = executeDDL("tweets_ddl.sql", "tweets", "tweets");
@@ -68,4 +78,32 @@ public final void testNestedFieldAsJoinKey1() throws Exception {
     assertResultSet(res);
     cleanupQuery(res);
   }
+
+  @Test
+  public final void testInsertType1() throws Exception {
+    // all columns
+    List<String> tables = executeDDL("sample1_ddl.sql", "sample1", "sample3");
+    assertEquals(TUtil.newList("sample3"), tables);
+
+    executeString("CREATE TABLE clone (title TEXT, name RECORD (first_name TEXT, last_name TEXT)) USING JSON;").close();
+
+    executeString("INSERT INTO clone (title, name.first_name, name.last_name) SELECT title, name.first_name, name.last_name from sample3").close();
+    ResultSet res = executeString("select title, name.first_name, name.last_name from clone");
+    assertResultSet(res);
+    res.close();
+  }
+
+  @Test
+  public final void testInsertType2() throws Exception {
+    // some columns
+    List<String> tables = executeDDL("sample1_ddl.sql", "sample1", "sample4");
+    assertEquals(TUtil.newList("sample4"), tables);
+
+    executeString("CREATE TABLE clone2 (title TEXT, name RECORD (first_name TEXT, last_name TEXT)) USING JSON;").close();
+
+    executeString("INSERT INTO clone2 (title, name.last_name) SELECT title, name.last_name from sample4").close();
+    ResultSet res = executeString("select title, name.first_name, name.last_name from clone2");
+    assertResultSet(res);
+    res.close();
+  }
 }

tajo-core/src/test/resources/queries/TestSelectNestedRecord/sample0_ddl.sql

@@ -0,0 +1 @@
+CREATE TABLE clone (title TEXT, name RECORD (first_name TEXT, last_name TEXT)) USING JSON;

tajo-core/src/test/resources/queries/TestSelectNestedRecord/testSelect0.sql

@@ -0,0 +1 @@
+SELECT title, name.first_name, name.last_name FROM sample1;

tajo-core/src/test/resources/results/TestSelectNestedRecord/testInsertType1.result

@@ -0,0 +1,5 @@
+title,name/first_name,name/last_name
+-------------------------------
+Hand of the King,Eddard,Stark
+Assassin,Arya,Stark
+Dancing Master,Syrio,Forel

tajo-core/src/test/resources/results/TestSelectNestedRecord/testInsertType2.result

@@ -0,0 +1,5 @@
+title,name/first_name,name/last_name
+-------------------------------
+Hand of the King,null,Stark
+Assassin,null,Stark
+Dancing Master,null,Forel

tajo-core/src/test/resources/results/TestSelectNestedRecord/testSelect0.result

@@ -0,0 +1,5 @@
+title,name/first_name,name/last_name
+-------------------------------
+Hand of the King,Eddard,Stark
+Assassin,Arya,Stark
+Dancing Master,Syrio,Forel

tajo-plan/src/main/java/org/apache/tajo/plan/LogicalPlanner.java

@@ -56,7 +56,6 @@
 import org.apache.tajo.plan.rewrite.rules.ProjectionPushDownRule;
 import org.apache.tajo.plan.util.ExprFinder;
 import org.apache.tajo.plan.util.PlannerUtil;
-import org.apache.tajo.plan.verifier.VerifyException;
 import org.apache.tajo.storage.StorageService;
 import org.apache.tajo.util.KeyValueSet;
 import org.apache.tajo.util.Pair;
@@ -1656,10 +1655,10 @@ private InsertNode buildInsertIntoTablePlan(PlanContext context, InsertNode inse
 
       // See PreLogicalPlanVerifier.visitInsert.
       // It guarantees that the equivalence between the numbers of target and projected columns.
-      String [] targets = expr.getTargetColumns();
+      ColumnReferenceExpr [] targets = expr.getTargetColumns();
       Schema targetColumns = new Schema();
       for (int i = 0; i < targets.length; i++) {
-        Column targetColumn = desc.getLogicalSchema().getColumn(targets[i]);
+        Column targetColumn = desc.getLogicalSchema().getColumn(targets[i].getCanonicalName().replace(".", "/"));
 
         if (targetColumn == null) {
           throw makeSyntaxError("column '" + targets[i] + "' of relation '" + desc.getName() + "' does not exist");
@@ -1702,15 +1701,20 @@ private void buildProjectedInsert(PlanContext context, InsertNode insertNode) {
     }
 
     if (child instanceof Projectable) {
-      Projectable projectionNode = (Projectable)insertNode.getChild();
+      Projectable projectionNode = insertNode.getChild();
 
       // Modifying projected columns by adding NULL constants
       // It is because that table appender does not support target columns to be written.
       List<Target> targets = TUtil.newList();
-      for (int i = 0; i < tableSchema.size(); i++) {
-        Column column = tableSchema.getColumn(i);
 
+      for (Column column : tableSchema.getAllColumns()) {
         int idxInProjectionNode = targetColumns.getIndex(column);
+
+        // record type itself cannot be projected yet.
+        if (column.getDataType().getType() == TajoDataTypes.Type.RECORD) {
+          continue;
+        }
+
         if (idxInProjectionNode >= 0 && idxInProjectionNode < projectionNode.getTargets().length) {
           targets.add(projectionNode.getTargets()[idxInProjectionNode]);
         } else {

tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/json/JsonLineDeserializer.java

@@ -19,6 +19,7 @@
 package org.apache.tajo.storage.json;
 
 
+import com.facebook.presto.hive.shaded.com.google.common.collect.Lists;
 import io.netty.buffer.ByteBuf;
 import net.minidev.json.JSONObject;
 import net.minidev.json.parser.JSONParser;
@@ -35,57 +36,29 @@
 import org.apache.tajo.storage.Tuple;
 import org.apache.tajo.storage.text.TextLineDeserializer;
 import org.apache.tajo.storage.text.TextLineParsingError;
-import org.apache.tajo.util.TUtil;
 
 import java.io.IOException;
 import java.util.Map;
 
 public class JsonLineDeserializer extends TextLineDeserializer {
   private JSONParser parser;
+
   // Full Path -> Type
-  private Map<String, Type> types;
-  private String [] projectedPaths;
+  private final Map<String, Type> types;
+  private final String [] projectedPaths;
 
   public JsonLineDeserializer(Schema schema, TableMeta meta, Column [] projected) {
     super(schema, meta);
 
-    projectedPaths = new String[projected.length];
-    for (int i = 0; i < projected.length; i++) {
-      this.projectedPaths[i] = projected[i].getSimpleName();
-    }
+    projectedPaths = SchemaUtil.convertColumnsToPaths(Lists.newArrayList(projected), true);
+    types = SchemaUtil.buildTypeMap(schema.getAllColumns(), projectedPaths);
   }
 
   @Override
   public void init() {
-    types = TUtil.newHashMap();
-    for (Column column : schema.getAllColumns()) {
-
-      // Keep types which only belong to projected paths
-      // For example, assume that a projected path is 'name/first_name', where name is RECORD and first_name is TEXT.
-      // In this case, we should keep two types:
-      // * name - RECORD
-      // * name/first_name TEXT
-      for (String p :projectedPaths) {
-        if (p.startsWith(column.getSimpleName())) {
-          types.put(column.getSimpleName(), column.getDataType().getType());
-        }
-      }
-    }
     parser = new JSONParser(JSONParser.MODE_JSON_SIMPLE | JSONParser.IGNORE_CONTROL_CHAR);
   }
 
-  private static String makePath(String [] path, int depth) {
-    StringBuilder sb = new StringBuilder();
-    for (int i = 0; i <= depth; i++) {
-      sb.append(path[i]);
-      if (i < depth) {
-        sb.append(NestedPathUtil.PATH_DELIMITER);
-      }
-    }
-
-    return sb.toString();
-  }
-
   /**
    *
    *