[WIP] Supporting CTAS queries for Hive to Spark query translations

coral-common/src/main/java/com/linkedin/coral/common/ToRelConverter.java

@@ -118,7 +118,6 @@ protected ToRelConverter(Map<String, Map<String, List<String>>> localMetaStore)
   public RelNode convertSql(String sql) {
     return toRel(toSqlNode(sql));
   }
-
   /**
    * Similar to {@link #convertSql(String)} but converts hive view definition stored
    * in the hive metastore to corresponding {@link RelNode} implementation.
@@ -133,8 +132,6 @@ public RelNode convertView(String hiveDbName, String hiveViewName) {
     return toRel(sqlNode);
   }
 
-  // TODO change back to protected once the relevant tests move to the common package
-  @VisibleForTesting
   public SqlNode toSqlNode(String sql) {
     return toSqlNode(sql, null);
   }
@@ -161,8 +158,7 @@ public SqlNode processView(String dbName, String tableName) {
     return toSqlNode(stringViewExpandedText, table);
   }
 
-  @VisibleForTesting
-  protected RelNode toRel(SqlNode sqlNode) {
+  public RelNode toRel(SqlNode sqlNode) {
     RelRoot root = getSqlToRelConverter().convertQuery(sqlNode, true, true);
     return standardizeRel(root.rel);
   }

coral-common/src/main/java/com/linkedin/coral/common/calcite/sql/SqlCommand.java

@@ -0,0 +1,10 @@
+package com.linkedin.coral.common.calcite.sql;
+
+import org.apache.calcite.sql.SqlNode;
+
+public interface SqlCommand {
+
+    public SqlNode getSelectQuery();
+
+    public void setSelectQuery(SqlNode selectQuery);
+}

coral-common/src/main/java/com/linkedin/coral/common/calcite/sql/SqlCreateTable.java

@@ -0,0 +1,101 @@
+package com.linkedin.coral.common.calcite.sql;
+
+import com.linkedin.coral.javax.annotation.Nullable;
+import org.apache.calcite.sql.*;
+import org.apache.calcite.sql.parser.SqlParserPos;
+import org.apache.calcite.util.ImmutableNullableList;
+
+import java.util.List;
+import java.util.Objects;
+
+
+public class SqlCreateTable extends SqlCreate implements SqlCommand{
+    //name of the table to be created
+    private final SqlIdentifier name;
+    // column details like column name, data type, etc. This may be null, like in case of CTAS
+    private final @Nullable SqlNodeList columnList;
+    // select query node in case of "CREATE TABLE ... AS query"; else may be null
+    private @Nullable SqlNode selectQuery;
+    // specifying serde property
+    private final @Nullable SqlNode serDe;
+    // specifying file format such as Parquet, ORC, etc.
+    private final @Nullable SqlNodeList fileFormat;
+    // specifying delimiter fields for row format
+    private final @Nullable SqlCharStringLiteral rowFormat;
+
+    private static final SqlOperator OPERATOR =
+            new SqlSpecialOperator("CREATE TABLE", SqlKind.CREATE_TABLE);
+
+    /** Creates a SqlCreateTable. */
+    public SqlCreateTable(SqlParserPos pos, boolean replace, boolean ifNotExists,
+                          SqlIdentifier name, @Nullable SqlNodeList columnList, @Nullable SqlNode selectQuery,
+                          @Nullable SqlNode serDe, @Nullable SqlNodeList fileFormat, @Nullable SqlCharStringLiteral rowFormat) {
+        super(OPERATOR, pos, replace, ifNotExists);
+        this.name = Objects.requireNonNull(name, "name");
+        this.columnList = columnList;
+        this.selectQuery = selectQuery;
+        this.serDe = serDe;
+        this.fileFormat = fileFormat;
+        this.rowFormat = rowFormat;
+    }
+
+    @SuppressWarnings("nullness")
+    @Override public List<SqlNode> getOperandList() {
+        return ImmutableNullableList.of(name, columnList, selectQuery, serDe, fileFormat, rowFormat);
+    }
+
+    @Override public void unparse(SqlWriter writer, int leftPrec, int rightPrec) {
+        writer.keyword("CREATE");
+        writer.keyword("TABLE");
+        if (ifNotExists) {
+            writer.keyword("IF NOT EXISTS");
+        }
+        name.unparse(writer, leftPrec, rightPrec);
+        if (columnList != null) {
+            SqlWriter.Frame frame = writer.startList("(", ")");
+            for (SqlNode c : columnList) {
+                writer.sep(",");
+                c.unparse(writer, 0, 0);
+            }
+            writer.endList(frame);
+        }
+        if(serDe != null){
+            writer.keyword("ROW FORMAT SERDE");
+            serDe.unparse(writer, 0, 0);
+            writer.newlineAndIndent();
+        }
+        if(rowFormat != null){
+            writer.keyword("ROW FORMAT DELIMITED FIELDS TERMINATED BY");
+            rowFormat.unparse(writer, 0, 0);
+            writer.newlineAndIndent();
+        }
+        if(fileFormat != null){
+            if(fileFormat.size() == 1){
+                writer.keyword("STORED AS");
+                fileFormat.get(0).unparse(writer, 0, 0);
+                writer.newlineAndIndent();
+            } else {
+                writer.keyword("STORED AS INPUTFORMAT");
+                fileFormat.get(0).unparse(writer, 0, 0);
+                writer.keyword("OUTPUTFORMAT");
+                fileFormat.get(1).unparse(writer, 0, 0);
+                writer.newlineAndIndent();
+            }
+        }
+        if (selectQuery != null) {
+            writer.keyword("AS");
+            writer.newlineAndIndent();
+            selectQuery.unparse(writer, 0, 0);
+        }
+    }
+
+    @Override
+    public SqlNode getSelectQuery() {
+        return selectQuery;
+    }
+
+    @Override
+    public void setSelectQuery(SqlNode query) {
+        this.selectQuery = query;
+    }
+}

coral-hive/src/main/java/com/linkedin/coral/hive/hive2rel/parsetree/AbstractASTVisitor.java

@@ -302,12 +302,93 @@ protected R visit(ASTNode node, C ctx) {
       case HiveParser.KW_CURRENT:
         return visitCurrentRow(node, ctx);
 
+      case HiveParser.TOK_CREATETABLE:
+        return visitCreateTable(node, ctx);
+      case HiveParser.TOK_LIKETABLE:
+        return visitLikeTable(node, ctx);
+      case HiveParser.TOK_IFNOTEXISTS:
+        return visitIfNotExists(node, ctx);
+      case HiveParser.TOK_TABCOLLIST:
+        return visitColumnList(node, ctx);
+      case HiveParser.TOK_TABCOL:
+        return visitColumn(node, ctx);
+      case HiveParser.TOK_FILEFORMAT_GENERIC:
+        return visitFileFormatGeneric(node, ctx);
+      case HiveParser.TOK_TABLEFILEFORMAT:
+        return visitTableFileFormat(node, ctx);
+      case HiveParser.TOK_TABLESERIALIZER:
+        return visitTableSerializer(node, ctx);
+      case HiveParser.TOK_SERDENAME:
+        return visitSerdeName(node, ctx);
+      case HiveParser.TOK_TABLEROWFORMAT:
+        return visitTableRowFormat(node, ctx);
+      case HiveParser.TOK_SERDEPROPS:
+        return visitSerdeProps(node, ctx);
+      case HiveParser.TOK_TABLEROWFORMATFIELD:
+        return visitTableRowFormatField(node, ctx);
       default:
         // return visitChildren(node, ctx);
         throw new UnhandledASTTokenException(node);
     }
   }
 
+  protected R visitTableRowFormatField(ASTNode node, C ctx) {
+    return visitChildren(node, ctx).get(0);
+  }
+
+  protected R visitSerdeProps(ASTNode node, C ctx) {
+    return visitChildren(node, ctx).get(0);
+  }
+
+  protected R visitTableRowFormat(ASTNode node, C ctx) {
+    return visitChildren(node, ctx).get(0);
+  }
+
+  protected R visitSerdeName(ASTNode node, C ctx) {
+    return visitChildren(node, ctx).get(0);
+  }
+
+  protected R visitTableSerializer(ASTNode node, C ctx) {
+    return visitChildren(node, ctx).get(0);
+  }
+
+  protected R visitTableFileFormat(ASTNode node, C ctx) {
+    return visitChildren(node, ctx).get(0);
+  }
+
+  protected R visitFileFormatGeneric(ASTNode node, C ctx) {
+    return visitChildren(node, ctx).get(0);
+  }
+
+  protected R visitColumn(ASTNode node, C ctx) {
+    if(node.getChildren() != null){
+      return visitChildren(node, ctx).get(0);
+    }
+    return null;
+  }
+
+  protected R visitColumnList(ASTNode node, C ctx) {
+    return visitChildren(node, ctx).get(0);
+  }
+
+  protected R visitIfNotExists(ASTNode node, C ctx) {
+    if(node.getChildren() != null){
+      return visitChildren(node, ctx).get(0);
+    }
+    return null;
+  }
+
+  protected R visitLikeTable(ASTNode node, C ctx){
+    if(node.getChildren() != null){
+      return visitChildren(node, ctx).get(0);
+    }
+    return null;
+  }
+
+  protected R visitCreateTable(ASTNode node, C ctx){
+    return visitChildren(node, ctx).get(0);
+  }
+
   protected R visitKeywordLiteral(ASTNode node, C ctx) {
     return visitChildren(node, ctx).get(0);
   }

coral-hive/src/main/java/com/linkedin/coral/hive/hive2rel/parsetree/ParseTreeBuilder.java

@@ -9,10 +9,12 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
+import java.util.Arrays;
 import java.util.stream.Collectors;
 
 import javax.annotation.Nullable;
 
+import com.linkedin.coral.common.calcite.sql.SqlCreateTable;
 import org.apache.calcite.avatica.util.TimeUnit;
 import org.apache.calcite.sql.JoinConditionType;
 import org.apache.calcite.sql.JoinType;
@@ -39,6 +41,7 @@
 import org.apache.calcite.sql.fun.SqlStdOperatorTable;
 import org.apache.calcite.sql.parser.SqlParserPos;
 import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.calcite.sql.SqlCharStringLiteral;
 import org.apache.hadoop.hive.metastore.api.Table;
 
 import com.linkedin.coral.com.google.common.collect.ImmutableList;
@@ -608,6 +611,96 @@ protected SqlNode visitSelect(ASTNode node, ParseContext ctx) {
     return ctx.selects;
   }
 
+  @Override
+  protected SqlNode visitCreateTable(ASTNode node, ParseContext ctx){
+    CreateTableOptions ctOptions = new CreateTableOptions();
+    for(Node child: node.getChildren()){
+      ASTNode ast = (ASTNode) child;
+      switch (ast.getType()){
+        case HiveParser.TOK_TABNAME:
+          ctOptions.name = (SqlIdentifier) visitTabnameNode(ast, ctx);
+          break;
+        case HiveParser.TOK_IFNOTEXISTS:
+          ctOptions.ifNotExists = true;
+          break;
+        case HiveParser.TOK_TABCOLLIST:
+          ctOptions.columnList = (SqlNodeList) visitColumnList(ast, ctx);
+          break;
+        case HiveParser.TOK_QUERY:
+          ctOptions.query = visitQueryNode(ast, ctx);
+          break;
+        case HiveParser.TOK_TABLESERIALIZER:
+          ctOptions.tableSerializer = visitTableSerializer(ast, ctx);
+          break;
+        case HiveParser.TOK_TABLEFILEFORMAT:
+          ctOptions.tableFileFormat = (SqlNodeList) visitTableFileFormat(ast, ctx);
+          break;
+        case HiveParser.TOK_FILEFORMAT_GENERIC:
+          ctOptions.tableFileFormat = (SqlNodeList) visitFileFormatGeneric(ast, ctx);
+          break;
+        case HiveParser.TOK_TABLEROWFORMAT:
+          ctOptions.tableRowFormat = (SqlCharStringLiteral) visitTableRowFormat(ast, ctx);
+          break;
+        default:
+          break;
+      }
+    }
+    return new SqlCreateTable(ZERO, false, ctOptions.ifNotExists != null ? ctOptions.ifNotExists : false, ctOptions.name,
+            ctOptions.columnList, ctOptions.query, ctOptions.tableSerializer, ctOptions.tableFileFormat, ctOptions.tableRowFormat);
+  }
+
+  @Override
+  protected SqlNode visitColumnList(ASTNode node, ParseContext ctx){
+    List<SqlNode> sqlNodeList = visitChildren(node, ctx);
+    return new SqlNodeList(sqlNodeList, ZERO);
+  }
+
+  @Override
+  protected SqlNode visitColumn(ASTNode node, ParseContext ctx) {
+    return visitChildren(node, ctx).get(0);
+  }
+
+  @Override
+  protected SqlNode visitIfNotExists(ASTNode node, ParseContext ctx) {
+    return SqlLiteral.createBoolean(true, ZERO);
+  }
+
+  @Override
+  protected SqlNode visitTableRowFormat(ASTNode node, ParseContext ctx){
+    return visitChildren(node, ctx).get(0);
+  }
+
+  @Override
+  protected SqlNode visitSerdeName(ASTNode node, ParseContext ctx) {
+    return visit((ASTNode) node.getChildren().get(0), ctx);
+  }
+
+  @Override
+  protected SqlNode visitTableSerializer(ASTNode node, ParseContext ctx) {
+    return visitChildren(node, ctx).get(0);
+  }
+
+  @Override
+  protected SqlNode visitTableFileFormat(ASTNode node, ParseContext ctx) {
+    List<SqlNode> sqlNodeList = visitChildren(node, ctx);
+    return new SqlNodeList(sqlNodeList, ZERO);
+  }
+
+  @Override
+  protected SqlNode visitFileFormatGeneric(ASTNode node, ParseContext ctx) {
+    return new SqlNodeList(Arrays.asList(visitChildren(node, ctx).get(0)), ZERO);
+  }
+
+  @Override
+  protected SqlNode visitSerdeProps(ASTNode node, ParseContext ctx) {
+    return visitChildren(node, ctx).get(0);
+  }
+
+  @Override
+  protected SqlNode visitTableRowFormatField(ASTNode node, ParseContext ctx) {
+    return visitChildren(node, ctx).get(0);
+  }
+
   @Override
   protected SqlNode visitTabRefNode(ASTNode node, ParseContext ctx) {
     List<SqlNode> sqlNodes = visitChildren(node, ctx);
@@ -1059,4 +1152,14 @@ Optional<Table> getHiveTable() {
       return hiveTable;
     }
   }
-}
+
+  class CreateTableOptions {
+    SqlIdentifier name;
+    SqlNodeList columnList;
+    SqlNode query;
+    Boolean ifNotExists;
+    SqlNode tableSerializer;
+    SqlNodeList tableFileFormat;
+    SqlCharStringLiteral tableRowFormat;
+  }
+}