Add explicit table

snuyanzin · Oct 6, 2024 · ae980a7 · ae980a7
1 parent 7d091b1
commit ae980a7
Show file tree

Hide file tree

Showing 3 changed files with 13 additions and 4 deletions.
diff --git a/...lanner/src/main/java/org/apache/flink/table/planner/calcite/FlinkCalciteSqlValidator.java b/...lanner/src/main/java/org/apache/flink/table/planner/calcite/FlinkCalciteSqlValidator.java
@@ -419,7 +419,7 @@ protected void addToSelectList(
      * A special {@link SqlSelect} to capture the origin of a {@link SqlKind#EXPLICIT_TABLE} within
      * TVF operands.
      */
-    private static class ExplicitTableSqlSelect extends SqlSelect {
+    static class ExplicitTableSqlSelect extends SqlSelect {
 
         private final List<SqlIdentifier> descriptors;
 

diff --git a/...able-planner/src/main/scala/org/apache/flink/table/planner/calcite/SqlRewriterUtils.scala b/...able-planner/src/main/scala/org/apache/flink/table/planner/calcite/SqlRewriterUtils.scala
@@ -19,6 +19,7 @@ package org.apache.flink.table.planner.calcite
 
 import org.apache.flink.sql.parser.`type`.SqlMapTypeNameSpec
 import org.apache.flink.table.api.ValidationException
+import org.apache.flink.table.planner.calcite.FlinkCalciteSqlValidator.ExplicitTableSqlSelect
 import org.apache.flink.table.planner.calcite.PreValidateReWriter.{newValidationError, notSupported}
 import org.apache.flink.table.planner.calcite.SqlRewriterUtils.{rewriteSqlCall, rewriteSqlSelect, rewriteSqlValues}
 import org.apache.flink.util.Preconditions.checkArgument
@@ -160,9 +161,14 @@ object SqlRewriterUtils {
           operands.get(1).asInstanceOf[SqlNodeList],
           operands.get(2),
           operands.get(3))
+      case SqlKind.EXPLICIT_TABLE =>
+        val operands = call.getOperandList
+        val expTable = new ExplicitTableSqlSelect(
+          operands.get(0).asInstanceOf[SqlIdentifier],
+          Collections.emptyList())
+        rewriterUtils.rewriteSelect(expTable, targetRowType, assignedFields, targetPosition)
       // Not support:
       // case SqlKind.WITH =>
-      // case SqlKind.EXPLICIT_TABLE =>
       case _ => throw new ValidationException(unsupportedErrorMessage())
     }
   }

diff --git a/...er/src/test/java/org/apache/flink/table/planner/calcite/FlinkCalciteSqlValidatorTest.java b/...er/src/test/java/org/apache/flink/table/planner/calcite/FlinkCalciteSqlValidatorTest.java
@@ -78,7 +78,9 @@ void testUpsertInto() {
                 "INSERT INTO t2 (a, b) SELECT * FROM t1",
                 "INSERT INTO t2 (a, b) SELECT *, 42 FROM t2_copy",
                 "INSERT INTO t2 (a, b) SELECT 42, * FROM t2_copy",
-                "INSERT INTO t2 (a, b) SELECT * FROM t_nested"
+                "INSERT INTO t2 (a, b) SELECT * FROM t_nested",
+                "INSERT INTO t2 (a, b) TABLE t_nested",
+                "INSERT INTO t2 (a, b) SELECT * FROM (TABLE t_nested)"
             })
     void testInvalidNumberOfColumnsWhileInsertInto(String sql) {
         assertThatThrownBy(() -> plannerMocks.getParser().parse(sql))
@@ -95,7 +97,8 @@ void testInvalidNumberOfColumnsWhileInsertInto(String sql) {
                 "INSERT INTO t2 (a, b) SELECT * FROM t2_copy",
                 "INSERT INTO t2 (a, b) SELECT *, 42 FROM t1",
                 "INSERT INTO t2 (a, b) SELECT 42, * FROM t1",
-                "INSERT INTO t2 (a, b) SELECT f.* FROM t_nested"
+                "INSERT INTO t2 (a, b) SELECT f.* FROM t_nested",
+                "INSERT INTO t2 (a, b) TABLE t2_copy"
             })
     void validInsertIntoTest(final String sql) {
         assertDoesNotThrow(