[FLINK-38683][table] Add schema to SHOW CREATE MATERIALIZED TABLE output

This closes #27236.

Author: snuyanzin

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

@@ -116,18 +116,20 @@ public static String buildShowCreateTableRow(
     public static String buildShowCreateMaterializedTableRow(
             ResolvedCatalogMaterializedTable table,
             ObjectIdentifier tableIdentifier,
-            boolean isTemporary) {
+            boolean isTemporary,
+            SqlFactory sqlFactory) {
         validateTableKind(table, tableIdentifier, TableKind.MATERIALIZED_TABLE);
-        Optional<String> primaryKeys = extractFormattedPrimaryKey(table, PRINT_INDENT);
         StringBuilder sb =
                 new StringBuilder()
                         .append(
                                 buildCreateFormattedPrefix(
-                                        "MATERIALIZED TABLE",
-                                        isTemporary,
-                                        tableIdentifier,
-                                        primaryKeys.isPresent()));
-        primaryKeys.ifPresent(s -> sb.append(s).append("\n)\n"));
+                                        "MATERIALIZED TABLE", isTemporary, tableIdentifier, true));
+        sb.append(extractFormattedColumns(table, PRINT_INDENT));
+        extractFormattedWatermarkSpecs(table, PRINT_INDENT, sqlFactory)
+                .ifPresent(watermarkSpecs -> sb.append(",\n").append(watermarkSpecs));
+        extractFormattedPrimaryKey(table, PRINT_INDENT)
+                .ifPresent(pk -> sb.append(",\n").append(pk));
+        sb.append("\n)\n");
         extractComment(table).ifPresent(c -> sb.append(formatComment(c)).append("\n"));
         table.getDistribution()
                 .map(TableDistribution::toString)

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ShowCreateMaterializedTableOperation.java

@@ -55,7 +55,10 @@ public TableResultInternal execute(Context ctx) {
                                                         tableIdentifier.asSerializableString())));
         String resultRow =
                 ShowCreateUtil.buildShowCreateMaterializedTableRow(
-                        table.getResolvedTable(), tableIdentifier, table.isTemporary());
+                        table.getResolvedTable(),
+                        tableIdentifier,
+                        table.isTemporary(),
+                        ctx.getCatalogManager().getSqlFactory());
 
         return buildStringArrayResult("result", new String[] {resultRow});
     }

flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/api/internal/ShowCreateUtilTest.java

@@ -68,7 +68,9 @@ class ShowCreateUtilTest {
 
     private static final ResolvedSchema ONE_COLUMN_SCHEMA_WITH_PRIMARY_KEY =
             new ResolvedSchema(
-                    List.of(Column.physical("id", DataTypes.INT())),
+                    List.of(
+                            Column.physical("id", DataTypes.INT()),
+                            Column.metadata("mt_column", DataTypes.STRING(), null, true)),
                     List.of(),
                     UniqueConstraint.primaryKey("pk", List.of("id")),
                     List.of());
@@ -104,12 +106,13 @@ void showCreateView(
     @ParameterizedTest(name = "{index}: {2}")
     @MethodSource("argsForShowCreateMaterializedTable")
     void showCreateMaterializedTable(
-            ResolvedCatalogMaterializedTable materializedTable,
-            boolean isTemporary,
-            String expected) {
+            ResolvedCatalogMaterializedTable materializedTable, String expected) {
         final String createMaterializedTableString =
                 ShowCreateUtil.buildShowCreateMaterializedTableRow(
-                        materializedTable, MATERIALIZED_TABLE_IDENTIFIER, isTemporary);
+                        materializedTable,
+                        MATERIALIZED_TABLE_IDENTIFIER,
+                        false,
+                        DefaultSqlFactory.INSTANCE);
         assertThat(createMaterializedTableString).isEqualTo(expected);
     }
 
@@ -274,55 +277,62 @@ private static Collection<Arguments> argsForShowCreateTable() {
 
     private static Collection<Arguments> argsForShowCreateMaterializedTable() {
         final Collection<Arguments> argList = new ArrayList<>();
-        addTemporaryAndPermanent(
-                argList,
-                createResolvedMaterialized(
-                        ONE_COLUMN_SCHEMA,
-                        null,
-                        List.of(),
-                        null,
-                        IntervalFreshness.ofMinute("1"),
-                        RefreshMode.CONTINUOUS,
-                        "SELECT 1"),
-                "CREATE %sMATERIALIZED TABLE `catalogName`.`dbName`.`materializedTableName`\n"
-                        + "FRESHNESS = INTERVAL '1' MINUTE\n"
-                        + "REFRESH_MODE = CONTINUOUS\n"
-                        + "AS SELECT 1\n");
+        argList.add(
+                Arguments.of(
+                        createResolvedMaterialized(
+                                ONE_COLUMN_SCHEMA,
+                                null,
+                                List.of(),
+                                null,
+                                IntervalFreshness.ofMinute("1"),
+                                RefreshMode.CONTINUOUS,
+                                "SELECT 1"),
+                        "CREATE MATERIALIZED TABLE `catalogName`.`dbName`.`materializedTableName` (\n"
+                                + "  `id` INT\n"
+                                + ")\n"
+                                + "FRESHNESS = INTERVAL '1' MINUTE\n"
+                                + "REFRESH_MODE = CONTINUOUS\n"
+                                + "AS SELECT 1\n"));
 
-        addTemporaryAndPermanent(
-                argList,
-                createResolvedMaterialized(
-                        ONE_COLUMN_SCHEMA_WITH_PRIMARY_KEY,
-                        null,
-                        List.of(),
-                        null,
-                        IntervalFreshness.ofMinute("1"),
-                        RefreshMode.CONTINUOUS,
-                        "SELECT 1"),
-                "CREATE %sMATERIALIZED TABLE `catalogName`.`dbName`.`materializedTableName` (\n"
-                        + "  CONSTRAINT `pk` PRIMARY KEY (`id`) NOT ENFORCED\n"
-                        + ")\n"
-                        + "FRESHNESS = INTERVAL '1' MINUTE\n"
-                        + "REFRESH_MODE = CONTINUOUS\n"
-                        + "AS SELECT 1\n");
+        argList.add(
+                Arguments.of(
+                        createResolvedMaterialized(
+                                ONE_COLUMN_SCHEMA_WITH_PRIMARY_KEY,
+                                null,
+                                List.of(),
+                                null,
+                                IntervalFreshness.ofMinute("1"),
+                                RefreshMode.CONTINUOUS,
+                                "SELECT 1"),
+                        "CREATE MATERIALIZED TABLE `catalogName`.`dbName`.`materializedTableName` (\n"
+                                + "  `id` INT,\n"
+                                + "  `mt_column` VARCHAR(2147483647) METADATA VIRTUAL,\n"
+                                + "  CONSTRAINT `pk` PRIMARY KEY (`id`) NOT ENFORCED\n"
+                                + ")\n"
+                                + "FRESHNESS = INTERVAL '1' MINUTE\n"
+                                + "REFRESH_MODE = CONTINUOUS\n"
+                                + "AS SELECT 1\n"));
 
-        addTemporaryAndPermanent(
-                argList,
-                createResolvedMaterialized(
-                        TWO_COLUMNS_SCHEMA,
-                        "Materialized table comment",
-                        List.of("id"),
-                        TableDistribution.of(TableDistribution.Kind.HASH, 5, List.of("id")),
-                        IntervalFreshness.ofMinute("3"),
-                        RefreshMode.FULL,
-                        "SELECT id, name FROM tbl_a"),
-                "CREATE %sMATERIALIZED TABLE `catalogName`.`dbName`.`materializedTableName`\n"
-                        + "COMMENT 'Materialized table comment'\n"
-                        + "DISTRIBUTED BY HASH(`id`) INTO 5 BUCKETS\n"
-                        + "PARTITIONED BY (`id`)\n"
-                        + "FRESHNESS = INTERVAL '3' MINUTE\n"
-                        + "REFRESH_MODE = FULL\n"
-                        + "AS SELECT id, name FROM tbl_a\n");
+        argList.add(
+                Arguments.of(
+                        createResolvedMaterialized(
+                                TWO_COLUMNS_SCHEMA,
+                                "Materialized table comment",
+                                List.of("id"),
+                                TableDistribution.of(TableDistribution.Kind.HASH, 5, List.of("id")),
+                                IntervalFreshness.ofMinute("3"),
+                                RefreshMode.FULL,
+                                "SELECT id, name FROM tbl_a"),
+                        "CREATE MATERIALIZED TABLE `catalogName`.`dbName`.`materializedTableName` (\n"
+                                + "  `id` INT,\n"
+                                + "  `name` VARCHAR(2147483647)\n"
+                                + ")\n"
+                                + "COMMENT 'Materialized table comment'\n"
+                                + "DISTRIBUTED BY HASH(`id`) INTO 5 BUCKETS\n"
+                                + "PARTITIONED BY (`id`)\n"
+                                + "FRESHNESS = INTERVAL '3' MINUTE\n"
+                                + "REFRESH_MODE = FULL\n"
+                                + "AS SELECT id, name FROM tbl_a\n"));
 
         return argList;
     }