[FLINK-36861][table-planner] CREATE TABLE AS with METADATA columns do…

…esn't work
reswqa · Dec 11, 2024 · 5ca9eef · 5ca9eef
1 parent 529f640
commit 5ca9eef
Show file tree

Hide file tree

Showing 5 changed files with 486 additions and 316 deletions.
diff --git a/...ble-planner/src/main/java/org/apache/flink/table/planner/operations/MergeTableAsUtil.java b/...ble-planner/src/main/java/org/apache/flink/table/planner/operations/MergeTableAsUtil.java
@@ -26,6 +26,7 @@
 import org.apache.flink.sql.parser.ddl.constraint.SqlTableConstraint;
 import org.apache.flink.table.api.Schema;
 import org.apache.flink.table.api.Schema.UnresolvedColumn;
+import org.apache.flink.table.api.Schema.UnresolvedMetadataColumn;
 import org.apache.flink.table.api.Schema.UnresolvedPhysicalColumn;
 import org.apache.flink.table.api.TableException;
 import org.apache.flink.table.api.ValidationException;
@@ -386,19 +387,31 @@ private void validateImplicitCastCompatibility(
                 int columnPos,
                 UnresolvedColumn sourceColumn,
                 UnresolvedColumn sinkColumn) {
-            if (!(sinkColumn instanceof UnresolvedPhysicalColumn)) {
+            LogicalType sinkColumnType;
+
+            if (sinkColumn instanceof UnresolvedPhysicalColumn) {
+                sinkColumnType = getLogicalType(((UnresolvedPhysicalColumn) sinkColumn));
+            } else if ((sinkColumn instanceof UnresolvedMetadataColumn)) {
+                if (((UnresolvedMetadataColumn) sinkColumn).isVirtual()) {
+                    throw new ValidationException(
+                            String.format(
+                                    "A column named '%s' already exists in the source schema. "
+                                            + "Virtual metadata columns cannot overwrite "
+                                            + "columns from source.",
+                                    columnName));
+                }
+
+                sinkColumnType = getLogicalType(((UnresolvedMetadataColumn) sinkColumn));
+            } else {
                 throw new ValidationException(
                         String.format(
                                 "A column named '%s' already exists in the source schema. "
-                                        + "Computed and metadata columns cannot overwrite "
-                                        + "regular columns.",
+                                        + "Computed columns cannot overwrite columns from source.",
                                 columnName));
             }
 
             LogicalType sourceColumnType =
                     getLogicalType(((UnresolvedPhysicalColumn) sourceColumn));
-            LogicalType sinkColumnType = getLogicalType(((UnresolvedPhysicalColumn) sinkColumn));
-
             if (!supportsImplicitCast(sourceColumnType, sinkColumnType)) {
                 throw new ValidationException(
                         String.format(

diff --git a/...le-planner/src/main/java/org/apache/flink/table/planner/operations/SchemaBuilderUtil.java b/...le-planner/src/main/java/org/apache/flink/table/planner/operations/SchemaBuilderUtil.java
@@ -267,6 +267,14 @@ LogicalType getLogicalType(UnresolvedPhysicalColumn column) {
         return dataTypeFactory.createDataType(column.getDataType()).getLogicalType();
     }
 
+    /**
+     * Gets the column data type of {@link UnresolvedMetadataColumn} column and convert it to a
+     * {@link LogicalType}.
+     */
+    LogicalType getLogicalType(UnresolvedMetadataColumn column) {
+        return dataTypeFactory.createDataType(column.getDataType()).getLogicalType();
+    }
+
     Optional<String> getComment(SqlTableColumn column) {
         return column.getComment().map(c -> ((SqlLiteral) c).getValueAs(String.class));
     }