[FLINK-37827][table] Fix use SHOW CREATE TABLE for MATERIALIZED_TABLE

Author: dujie

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/ShowCreateUtil.java

@@ -22,11 +22,13 @@
 import org.apache.flink.table.api.TableException;
 import org.apache.flink.table.catalog.CatalogBaseTable;
 import org.apache.flink.table.catalog.CatalogDescriptor;
+import org.apache.flink.table.catalog.CatalogMaterializedTable;
 import org.apache.flink.table.catalog.CatalogView;
 import org.apache.flink.table.catalog.Column;
 import org.apache.flink.table.catalog.ObjectIdentifier;
 import org.apache.flink.table.catalog.QueryOperationCatalogView;
 import org.apache.flink.table.catalog.ResolvedCatalogBaseTable;
+import org.apache.flink.table.catalog.ResolvedCatalogMaterializedTable;
 import org.apache.flink.table.catalog.ResolvedCatalogModel;
 import org.apache.flink.table.catalog.ResolvedCatalogTable;
 import org.apache.flink.table.catalog.ResolvedSchema;
@@ -87,6 +89,9 @@ public static String buildShowCreateTableRow(
                     String.format(
                             "SHOW CREATE TABLE is only supported for tables, but %s is a view. Please use SHOW CREATE VIEW instead.",
                             tableIdentifier.asSerializableString()));
+        } else if (table.getTableKind() == CatalogBaseTable.TableKind.MATERIALIZED_TABLE) {
+            return buildShowCreateMaterializedTableRow(
+                    (ResolvedCatalogMaterializedTable) table, tableIdentifier, isTemporary);
         }
         StringBuilder sb =
                 new StringBuilder()
@@ -110,6 +115,37 @@ public static String buildShowCreateTableRow(
         return sb.toString();
     }
 
+    private static String buildShowCreateMaterializedTableRow(
+            ResolvedCatalogMaterializedTable table,
+            ObjectIdentifier tableIdentifier,
+            boolean isTemporary) {
+        StringBuilder sb =
+                new StringBuilder()
+                        .append(
+                                buildCreateMaterializedTableFormattedPrefix(
+                                        isTemporary, tableIdentifier));
+        extractFormattedPrimaryKey(table, PRINT_INDENT)
+                .ifPresent(pk -> sb.append(" (").append(pk).append(")\n"));
+
+        extractComment(table).ifPresent(c -> sb.append(formatComment(c)).append("\n"));
+
+        extractFormattedPartitionedInfo(table)
+                .ifPresent(
+                        partitionedInfoFormatted ->
+                                sb.append("PARTITIONED BY (")
+                                        .append(partitionedInfoFormatted)
+                                        .append(")\n"));
+
+        extractFormattedOptions(table.getOptions(), PRINT_INDENT)
+                .ifPresent(v -> sb.append("WITH (\n").append(v).append("\n)\n"));
+
+        sb.append(extractMaterializedTablefreshNess(table)).append("\n");
+        extractMaterializedTableRefreshMode(table).ifPresent(mode -> sb.append(mode).append("\n"));
+
+        sb.append("AS ").append(table.getDefinitionQuery());
+        return sb.toString();
+    }
+
     /** Show create view statement only for views. */
     public static String buildShowCreateViewRow(
             ResolvedCatalogBaseTable<?> view,
@@ -161,6 +197,15 @@ static String buildCreateFormattedPrefix(
                 System.lineSeparator());
     }
 
+    static String buildCreateMaterializedTableFormattedPrefix(
+            boolean isTemporary, ObjectIdentifier identifier) {
+        return String.format(
+                "CREATE %sMATERIALIZED TABLE %s %s",
+                isTemporary ? "TEMPORARY " : "",
+                identifier.asSerializableString(),
+                System.lineSeparator());
+    }
+
     static Optional<String> extractFormattedPrimaryKey(
             ResolvedCatalogBaseTable<?> table, String printIndent) {
         Optional<UniqueConstraint> primaryKey = table.getResolvedSchema().getPrimaryKey();
@@ -269,6 +314,31 @@ static Optional<String> extractFormattedPartitionedInfo(ResolvedCatalogTable cat
                         .collect(Collectors.joining(", ")));
     }
 
+    static Optional<String> extractFormattedPartitionedInfo(
+            ResolvedCatalogMaterializedTable catalogTable) {
+        if (!catalogTable.isPartitioned()) {
+            return Optional.empty();
+        }
+        return Optional.of(
+                catalogTable.getPartitionKeys().stream()
+                        .map(EncodingUtils::escapeIdentifier)
+                        .collect(Collectors.joining(", ")));
+    }
+
+    static String extractMaterializedTablefreshNess(ResolvedCatalogMaterializedTable table) {
+        return String.format("FRESHNESS = %s ", table.getDefinitionFreshness().toString());
+    }
+
+    static Optional<String> extractMaterializedTableRefreshMode(
+            ResolvedCatalogMaterializedTable table) {
+        CatalogMaterializedTable.LogicalRefreshMode logicalRefreshMode =
+                table.getLogicalRefreshMode();
+        if (logicalRefreshMode == CatalogMaterializedTable.LogicalRefreshMode.AUTOMATIC) {
+            return Optional.empty();
+        }
+        return Optional.of(String.format("REFRESH_MODE = %s", table.getRefreshMode().name()));
+    }
+
     static Optional<String> extractFormattedOptions(Map<String, String> conf, String printIndent) {
         if (Objects.isNull(conf) || conf.isEmpty()) {
             return Optional.empty();