[FLINK-35195][table] Convert CatalogMaterializedTable to CatalogTable…

… to generate execution plan for planner
liuyongvs · May 7, 2024 · d8491c0 · d8491c0
1 parent b037f56
commit d8491c0
Show file tree

Hide file tree

Showing 4 changed files with 44 additions and 2 deletions.
diff --git a/...ink-table-api-java/src/main/java/org/apache/flink/table/catalog/ContextResolvedTable.java b/...ink-table-api-java/src/main/java/org/apache/flink/table/catalog/ContextResolvedTable.java
@@ -142,6 +142,20 @@ public <T extends CatalogBaseTable> T getTable() {
         return (T) resolvedTable.getOrigin();
     }
 
+    /**
+     * Convert the {@link ResolvedCatalogMaterializedTable} in {@link ContextResolvedTable} to
+     * {@link ResolvedCatalogTable }.
+     */
+    public ContextResolvedTable toCatalogTable() {
+        if (resolvedTable.getTableKind() == CatalogBaseTable.TableKind.MATERIALIZED_TABLE) {
+            return ContextResolvedTable.permanent(
+                    objectIdentifier,
+                    catalog,
+                    ((ResolvedCatalogMaterializedTable) resolvedTable).toResolvedCatalogTable());
+        }
+        return this;
+    }
+
     /**
      * Copy the {@link ContextResolvedTable}, replacing the underlying {@link CatalogTable} options.
      */
@@ -150,6 +164,12 @@ public ContextResolvedTable copy(Map<String, String> newOptions) {
             throw new ValidationException(
                     String.format("View '%s' cannot be enriched with new options.", this));
         }
+        if (resolvedTable.getTableKind() == CatalogBaseTable.TableKind.MATERIALIZED_TABLE) {
+            return ContextResolvedTable.permanent(
+                    objectIdentifier,
+                    catalog,
+                    ((ResolvedCatalogMaterializedTable) resolvedTable).copy(newOptions));
+        }
         return new ContextResolvedTable(
                 objectIdentifier,
                 catalog,
@@ -159,6 +179,12 @@ public ContextResolvedTable copy(Map<String, String> newOptions) {
 
     /** Copy the {@link ContextResolvedTable}, replacing the underlying {@link ResolvedSchema}. */
     public ContextResolvedTable copy(ResolvedSchema newSchema) {
+        if (resolvedTable.getTableKind() == CatalogBaseTable.TableKind.MATERIALIZED_TABLE) {
+            throw new ValidationException(
+                    String.format(
+                            "Materialized table '%s' cannot be copied with new schema %s.",
+                            this, newSchema));
+        }
         return new ContextResolvedTable(
                 objectIdentifier,
                 catalog,

diff --git a/...common/src/main/java/org/apache/flink/table/catalog/ResolvedCatalogMaterializedTable.java b/...common/src/main/java/org/apache/flink/table/catalog/ResolvedCatalogMaterializedTable.java
@@ -182,4 +182,17 @@ public String toString() {
                 + resolvedSchema
                 + '}';
     }
+
+    /** Convert this object to a {@link ResolvedCatalogTable} object for planner optimize query. */
+    public ResolvedCatalogTable toResolvedCatalogTable() {
+        return new ResolvedCatalogTable(
+                CatalogTable.newBuilder()
+                        .schema(getUnresolvedSchema())
+                        .comment(getComment())
+                        .partitionKeys(getPartitionKeys())
+                        .options(getOptions())
+                        .snapshot(getSnapshot().orElse(null))
+                        .build(),
+                getResolvedSchema());
+    }
 }
diff --git a/...e-planner/src/main/java/org/apache/flink/table/planner/catalog/DatabaseCalciteSchema.java b/...e-planner/src/main/java/org/apache/flink/table/planner/catalog/DatabaseCalciteSchema.java
@@ -90,7 +90,7 @@ public Table getTable(String tableName) {
         return table.map(
                         lookupResult ->
                                 new CatalogSchemaTable(
-                                        lookupResult,
+                                        lookupResult.toCatalogTable(),
                                         getStatistic(lookupResult, identifier),
                                         isStreamingMode))
                 .orElse(null);
@@ -102,6 +102,7 @@ private FlinkStatistic getStatistic(
                 contextResolvedTable.getResolvedTable();
         switch (resolvedBaseTable.getTableKind()) {
             case TABLE:
+            case MATERIALIZED_TABLE:
                 return FlinkStatistic.unknown(resolvedBaseTable.getResolvedSchema())
                         .tableStats(extractTableStats(contextResolvedTable, identifier))
                         .build();

diff --git a/...src/main/java/org/apache/flink/table/planner/operations/SqlNodeToOperationConversion.java b/...src/main/java/org/apache/flink/table/planner/operations/SqlNodeToOperationConversion.java
@@ -734,7 +734,9 @@ private Operation convertSqlInsert(RichSqlInsert insert) {
 
         UnresolvedIdentifier unresolvedIdentifier = UnresolvedIdentifier.of(targetTablePath);
         ObjectIdentifier identifier = catalogManager.qualifyIdentifier(unresolvedIdentifier);
-        ContextResolvedTable contextResolvedTable = catalogManager.getTableOrError(identifier);
+        // If it is materialized table, convert it to catalog table for query optimize
+        ContextResolvedTable contextResolvedTable =
+                catalogManager.getTableOrError(identifier).toCatalogTable();
 
         PlannerQueryOperation query =
                 (PlannerQueryOperation)