CALCITE-2159: unnest to support dynamic row type

core/src/main/java/org/apache/calcite/rel/core/Uncollect.java

@@ -23,9 +23,12 @@
 import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.rel.RelWriter;
 import org.apache.calcite.rel.SingleRel;
+import org.apache.calcite.rel.type.DynamicRecordType;
+import org.apache.calcite.rel.type.DynamicRecordTypeImpl;
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rel.type.RelDataTypeFactory;
 import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.rel.type.RelDataTypeFieldImpl;
 import org.apache.calcite.sql.SqlUnnestOperator;
 import org.apache.calcite.sql.SqlUtil;
 import org.apache.calcite.sql.type.MapSqlType;
@@ -123,14 +126,24 @@ public static RelDataType deriveUncollectRowType(RelNode rel,
     RelDataType inputType = rel.getRowType();
     assert inputType.isStruct() : inputType + " is not a struct";
     final List<RelDataTypeField> fields = inputType.getFieldList();
-    final RelDataTypeFactory.Builder builder =
-        rel.getCluster().getTypeFactory().builder();
+    final RelDataTypeFactory typeFactory = rel.getCluster().getTypeFactory();
+    final RelDataTypeFactory.Builder builder = typeFactory.builder();
     for (RelDataTypeField field : fields) {
       if (field.getType() instanceof MapSqlType) {
         builder.add(SqlUnnestOperator.MAP_KEY_COLUMN_NAME, field.getType().getKeyType());
         builder.add(SqlUnnestOperator.MAP_VALUE_COLUMN_NAME, field.getType().getValueType());
       } else {
         RelDataType ret = field.getType().getComponentType();
+        if (ret == null) {
+          DynamicRecordType dynRet = new DynamicRecordTypeImpl(typeFactory);
+          RelDataTypeField dynField = new RelDataTypeFieldImpl(
+              DynamicRecordType.DYNAMIC_STAR_PREFIX,
+              fields.size(),
+              typeFactory.createTypeWithNullability(
+                  typeFactory.createSqlType(SqlTypeName.DYNAMIC_STAR), true));
+          dynRet.getFieldList().add(dynField);
+          ret = dynRet;
+        }
         assert null != ret;
         if (ret.isStruct()) {
           builder.addAll(ret.getFieldList());

core/src/main/java/org/apache/calcite/rel/type/RelDataTypeHolder.java

@@ -57,6 +57,10 @@ Pair<RelDataTypeField, Boolean> getFieldOrInsert(String fieldName, boolean caseS
       if (Util.matches(caseSensitive, f.getName(), fieldName)) {
         return Pair.of(f, false);
       }
+      //dynamic star field match with any field.
+      if (f.getType().getSqlTypeName() == SqlTypeName.DYNAMIC_STAR) {
+        return Pair.of(f, false);
+      }
     }
 
     final SqlTypeName typeName = DynamicRecordType.isDynamicStarColName(fieldName)

core/src/main/java/org/apache/calcite/sql/SqlUnnestOperator.java

@@ -17,8 +17,12 @@
 package org.apache.calcite.sql;
 
 
+import org.apache.calcite.rel.type.DynamicRecordType;
+import org.apache.calcite.rel.type.DynamicRecordTypeImpl;
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.rel.type.RelDataTypeFieldImpl;
 import org.apache.calcite.sql.type.ArraySqlType;
 import org.apache.calcite.sql.type.MapSqlType;
 import org.apache.calcite.sql.type.MultisetSqlType;
@@ -67,7 +71,19 @@ public SqlUnnestOperator(boolean withOrdinality) {
       RelDataType type = opBinding.getOperandType(operand);
       if (type.isStruct()) {
         type = type.getFieldList().get(0).getType();
+      } else if (type.getSqlTypeName() == SqlTypeName.ANY) {
+        //when it is ANY, we assume there is only one reldata type
+        final RelDataTypeFactory factory =  opBinding.getTypeFactory();
+        DynamicRecordType dynRet = new DynamicRecordTypeImpl(factory);
+        RelDataTypeField dynField = new RelDataTypeFieldImpl(
+            DynamicRecordType.DYNAMIC_STAR_PREFIX,
+            0,
+            factory.createTypeWithNullability(
+                factory.createSqlType(SqlTypeName.DYNAMIC_STAR), true));
+        dynRet.getFieldList().add(dynField);
+        return dynRet;
       }
+
       assert type instanceof ArraySqlType || type instanceof MultisetSqlType
           || type instanceof MapSqlType;
       if (type instanceof MapSqlType) {

core/src/main/java/org/apache/calcite/sql/type/SqlTypeUtil.java

@@ -962,6 +962,14 @@ private static boolean flattenFields(
                     field.getType().getComponentType(),
                     null),
                 -1);
+        if (field.getType() instanceof ArraySqlType) {
+          flattenedCollectionType = typeFactory.createArrayType(
+              flattenRecordType(
+                  typeFactory,
+                  field.getType().getComponentType(),
+                  null),
+              -1);
+        }
         field =
             new RelDataTypeFieldImpl(
                 field.getName(),

core/src/main/java/org/apache/calcite/sql/validate/UnnestNamespace.java

@@ -16,16 +16,19 @@
  */
 package org.apache.calcite.sql.validate;
 
+import org.apache.calcite.jdbc.CalciteSchema;
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.sql.SqlCall;
+import org.apache.calcite.sql.SqlIdentifier;
 import org.apache.calcite.sql.SqlNode;
 import org.apache.calcite.sql.SqlUnnestOperator;
 import org.apache.calcite.sql.type.MultisetSqlType;
 
+
 /**
  * Namespace for UNNEST.
  */
-class UnnestNamespace extends AbstractNamespace {
+public class UnnestNamespace extends AbstractNamespace {
   //~ Instance fields --------------------------------------------------------
 
   private final SqlCall unnest;
@@ -47,6 +50,30 @@ class UnnestNamespace extends AbstractNamespace {
 
   //~ Methods ----------------------------------------------------------------
 
+  public void setType(RelDataType type) {
+    this.type = type;
+    this.rowType = convertToStruct(type);
+  }
+
+  private SqlValidatorTable getOrInsertTableToSchema(CalciteSchema schema) {
+    SqlNode toUnnest = this.unnest.operand(0);
+    if (toUnnest instanceof SqlIdentifier) {
+      final SqlIdentifier id = (SqlIdentifier) toUnnest;
+      SqlQualified qualified = this.scope.fullyQualify(id);
+      //List<String> fullTableName = qualified.namespace.getTable().getQualifiedName();
+      return qualified.namespace.getTable();
+    }
+    return null;
+  }
+
+  public SqlValidatorTable getTable() {
+    if (unnest.operand(0) instanceof SqlIdentifier) {
+      //when operand of SqlIdentifier type does not have struct, fake a table for UnnestNameSpace
+      return getOrInsertTableToSchema(validator.catalogReader.getRootSchema());
+    }
+    return null;
+  }
+
   protected RelDataType validateImpl(RelDataType targetRowType) {
     // Validate the call and its arguments, and infer the return type.
     validator.validateCall(unnest, scope);

core/src/test/java/org/apache/calcite/test/SqlToRelConverterTest.java

@@ -67,7 +67,7 @@ protected DiffRepository getDiffRepos() {
   /** Sets the SQL statement for a test. */
   public final Sql sql(String sql) {
     return new Sql(sql, true, true, tester, false,
-        SqlToRelConverter.Config.DEFAULT, SqlConformanceEnum.DEFAULT);
+        SqlToRelConverter.Config.DEFAULT, tester.getConformance());
   }
 
   protected final void check(
@@ -1055,6 +1055,30 @@ protected final void check(
     sql(sql).ok();
   }
 
+  @Test public void testUnnestArrayAggPlan() {
+    final String sql = "select d.deptno, e2.empno_avg\n"
+        + "from dept_nested as d outer apply\n"
+        + " (select avg(e.empno) as empno_avg from UNNEST(d.employees) as e) e2";
+
+    sql(sql).with(getLenienTester()).ok();
+  }
+
+  @Test public void testUnnestArrayPlan() {
+    final String sql = "select d.deptno, e2.empno\n"
+        + "from dept_nested as d,\n"
+        + " UNNEST(d.employees) e2";
+
+    sql(sql).with(getExtendedTester()).ok();
+  }
+
+  @Test public void testUnnestApplyDefaultConformance() {
+    final String sql = "select d.deptno, e2.empno_avg\n"
+        + "from dept_nested as d outer apply\n"
+        + " (select avg(e.empno) as empno_avg from UNNEST(d.employees) as e) e2";
+
+    sql(sql).with(getExtendedTester()).ok();
+  }
+
   @Test public void testArrayOfRecord() {
     sql("select employees[1].skills[2+3].desc from dept_nested").ok();
   }
@@ -2409,6 +2433,44 @@ public void testDynamicStarInTableJoin() throws Exception {
     sql(sql).with(getTesterWithDynamicTable()).ok();
   }
 
+  @Test
+  public void testDynamicNestedColumn() throws Exception {
+
+    final String sql1 = "select t2.fake_col as fake2 from SALES.CUSTOMER as t2";
+    sql(sql1).with(getTesterWithDynamicTable()).anyPlan();
+
+    final String sql2 = "select t2.fake_col['fake_col2'] as fake2 from SALES.CUSTOMER as t2";
+    sql(sql2).with(getTesterWithDynamicTable()).anyPlan();
+
+    final String sql3 = "select t3.fake_q1['fake_col2'] as fake2 "
+        + "from (select t2.fake_col as fake_q1 from SALES.CUSTOMER as t2) as t3";
+    sql(sql3).with(getTesterWithDynamicTable()).anyPlan();
+
+    final String sql4 = "select t3.fake_q1['fake_col2'] as fake2 "
+        + "from (select t2.fake_col as fake_q1 from SALES.CUSTOMER as t2) as t3";
+    sql(sql4).with(getTesterWithDynamicTable()).anyPlan();
+  }
+
+  @Test
+  public void testDynamicSchemaUnnest() throws Exception {
+
+    final String sql3 = "select t1.c_nationkey, t3.fake_col3 "
+        + "from SALES.CUSTOMER as t1, "
+        + "lateral (select t2.fake_col2 as fake_col3 from unnest(t1.fake_col) as t2) as t3";
+
+    sql(sql3).with(getTesterWithDynamicTable()).anyPlan();
+
+  }
+
+  @Test
+  public void testDynamicSchemaSubquery() throws Exception {
+
+    final String sql3 = "select t1.fake_col, t1.fake_col2 as fake_col2 "
+        + "from (select * from SALES.CUSTOMER ) as t1";
+
+    sql(sql3).with(getTesterWithDynamicTable()).anyPlan();
+
+  }
   /**
    * Test case for Dynamic Table / Dynamic Star support
    * <a href="https://issues.apache.org/jira/browse/CALCITE-1150">[CALCITE-1150]</a>
@@ -2514,6 +2576,16 @@ public Prepare.CatalogReader apply(RelDataTypeFactory typeFactory) {
       });
   }
 
+  private Tester getLenienTester() {
+    return lenienTester.withCatalogReaderFactory(
+        new Function<RelDataTypeFactory, Prepare.CatalogReader>() {
+          public Prepare.CatalogReader apply(RelDataTypeFactory typeFactory) {
+            return new MockCatalogReader(typeFactory, true)
+                .init().init2();
+          }
+        });
+  }
+
   private Tester getTesterWithDynamicTable() {
     return tester.withCatalogReaderFactory(
         new Function<RelDataTypeFactory, Prepare.CatalogReader>() {
@@ -2546,6 +2618,7 @@ public Prepare.CatalogReader apply(RelDataTypeFactory typeFactory) {
                 regionTable.addColumn("R_NAME", varcharType);
                 regionTable.addColumn("R_COMMENT", varcharType);
                 registerTable(regionTable);
+
                 return this;
               }
               // CHECKSTYLE: IGNORE 1
@@ -2781,11 +2854,13 @@ public class Sql {
     public void ok() {
       convertsTo("${plan}");
     }
-
+    public void anyPlan() {
+      convertsTo("ANYPLAN");
+    }
     public void convertsTo(String plan) {
       tester.withExpand(expand)
           .withDecorrelation(decorrelate)
-          .withConformance(conformance)
+          .withConformance(tester.getConformance())
           .withConfig(config)
           .assertConvertsTo(sql, plan, trim);
     }

core/src/test/java/org/apache/calcite/test/SqlToRelTestBase.java

@@ -84,10 +84,12 @@ public abstract class SqlToRelTestBase {
 
   protected static final String NL = System.getProperty("line.separator");
 
+  protected static final SqlConformance DEFAULT_CONFORMANCE = SqlConformanceEnum.DEFAULT;
   //~ Instance fields --------------------------------------------------------
 
-  protected final Tester tester = createTester();
+  protected final Tester tester = createTester(SqlConformanceEnum.DEFAULT);
 
+  protected final Tester lenienTester = createTester(SqlConformanceEnum.LENIENT);
   //~ Methods ----------------------------------------------------------------
 
   public SqlToRelTestBase() {
@@ -96,9 +98,13 @@ public SqlToRelTestBase() {
 
   protected Tester createTester() {
     return new TesterImpl(getDiffRepos(), false, false, true, false,
-        null, null);
+        null, null, SqlToRelConverter.Config.DEFAULT, DEFAULT_CONFORMANCE);
   }
 
+  protected Tester createTester(SqlConformance conformance) {
+    return new TesterImpl(getDiffRepos(), false, false, true, false,
+        null, null, SqlToRelConverter.Config.DEFAULT, conformance);
+  }
   /**
    * Returns the default diff repository for this test, or null if there is
    * no repository.
@@ -627,7 +633,8 @@ protected final RelOptPlanner getPlanner() {
     }
 
     public SqlNode parseQuery(String sql) throws Exception {
-      SqlParser parser = SqlParser.create(sql);
+      SqlParser parser = SqlParser.create(sql,
+          SqlParser.configBuilder().setConformance(getConformance()).build());
       return parser.parseQuery();
     }
 
@@ -705,7 +712,9 @@ public void assertConvertsTo(
       // that plans come out nicely stacked instead of first
       // line immediately after CDATA start
       String actual = NL + RelOptUtil.toString(rel);
-      diffRepos.assertEquals("plan", plan, actual);
+      if (plan.compareToIgnoreCase("ANYPLAN") != 0) {
+        diffRepos.assertEquals("plan", plan, actual);
+      }
     }
 
     /**

core/src/test/resources/org/apache/calcite/test/SqlToRelConverterTest.xml

@@ -5066,4 +5066,42 @@ LogicalProject(A=[$0], B=[$1])
 ]]>
         </Resource>
     </TestCase>
+    <TestCase name="testUnnestArrayAggPlan">
+        <Resource name="sql">
+            <![CDATA[select  d.deptno, e2.empno_avg
+            from dept_nested as d outer apply
+           (select avg(e.empno) as empno_avg from UNNEST(d.employees) as e) e2]]>
+        </Resource>
+        <Resource name="plan">
+            <![CDATA[
+LogicalProject(DEPTNO=[$0], EMPNO_AVG=[$3])
+  LogicalProject(DEPTNO=[$0], NAME=[$1], EMPLOYEES=[$2], EMPNO_AVG=[$3])
+    LogicalCorrelate(correlation=[$cor0], joinType=[left], requiredColumns=[{2}])
+      LogicalTableScan(table=[[CATALOG, SALES, DEPT_NESTED]])
+      LogicalAggregate(group=[{}], EMPNO_AVG=[AVG($0)])
+        LogicalProject(EMPNO=[$0])
+          Uncollect
+            LogicalProject(EMPLOYEES=[$cor0.EMPLOYEES_2])
+              LogicalValues(tuples=[[{ 0 }]])
+]]>
+        </Resource>
+    </TestCase>
+    <TestCase name="testUnnestArrayPlan">
+        <Resource name="sql">
+            <![CDATA[select  d.deptno, e2.empno_avg
+            from dept_nested as d, 
+            UNNEST(d.employees) e2]]>
+        </Resource>
+        <Resource name="plan">
+            <![CDATA[
+LogicalProject(DEPTNO=[$0], EMPNO=[$3])
+  LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
+    LogicalTableScan(table=[[CATALOG, SALES, DEPT_NESTED]])
+    Uncollect
+      LogicalProject(EMPLOYEES=[$cor0.EMPLOYEES_2])
+        LogicalValues(tuples=[[{ 0 }]])
+]]>
+        </Resource>
+    </TestCase>
+
 </Root>