[FLINK-38968][table] Add toCatalogTable method to CatalogMaterializedTable interface

Author: raminqaf

flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/CatalogBaseTableResolutionTest.java

@@ -354,6 +354,45 @@ void testInvalidDistributionBucketCount() {
                                 + "distributed table must be at least 1.");
     }
 
+    @Test
+    void testCatalogMaterializedTableToCatalogTable() {
+        final CatalogMaterializedTable materializedTable = catalogMaterializedTable();
+
+        final CatalogTable catalogTable = materializedTable.toCatalogTable();
+
+        // Verify the conversion preserves properties
+        assertThat(catalogTable.getUnresolvedSchema())
+                .isEqualTo(materializedTable.getUnresolvedSchema());
+        assertThat(catalogTable.getComment()).isEqualTo(materializedTable.getComment());
+        assertThat(catalogTable.getPartitionKeys()).isEqualTo(materializedTable.getPartitionKeys());
+        assertThat(catalogTable.getOptions()).isEqualTo(materializedTable.getOptions());
+        assertThat(catalogTable.getDistribution()).isEqualTo(materializedTable.getDistribution());
+        assertThat(catalogTable.getSnapshot()).isEqualTo(materializedTable.getSnapshot());
+    }
+
+    @Test
+    void testResolvedCatalogMaterializedTableToResolvedCatalogTable() {
+        final CatalogMaterializedTable materializedTable = catalogMaterializedTable();
+
+        final ResolvedCatalogMaterializedTable resolvedMaterializedTable =
+                resolveCatalogBaseTable(ResolvedCatalogMaterializedTable.class, materializedTable);
+
+        final ResolvedCatalogTable resolvedCatalogTable =
+                resolvedMaterializedTable.toResolvedCatalogTable();
+
+        // Verify schema is preserved
+        assertThat(resolvedCatalogTable.getResolvedSchema())
+                .isEqualTo(resolvedMaterializedTable.getResolvedSchema());
+
+        // Verify origin properties are preserved
+        assertThat(resolvedCatalogTable.getComment())
+                .isEqualTo(resolvedMaterializedTable.getComment());
+        assertThat(resolvedCatalogTable.getPartitionKeys())
+                .isEqualTo(resolvedMaterializedTable.getPartitionKeys());
+        assertThat(resolvedCatalogTable.getOptions())
+                .isEqualTo(resolvedMaterializedTable.getOptions());
+    }
+
     // --------------------------------------------------------------------------------------------
     // Utilities
     // --------------------------------------------------------------------------------------------

flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogMaterializedTable.java

@@ -180,6 +180,18 @@ default String getDefinitionQuery() {
     @Nullable
     byte[] getSerializedRefreshHandler();
 
+    /** Convert this object to a {@link CatalogTable} object for planner optimize query. */
+    default CatalogTable toCatalogTable() {
+        return CatalogTable.newBuilder()
+                .schema(getUnresolvedSchema())
+                .comment(getComment())
+                .distribution(getDistribution().orElse(null))
+                .partitionKeys(getPartitionKeys())
+                .options(getOptions())
+                .snapshot(getSnapshot().orElse(null))
+                .build();
+    }
+
     /** The logical refresh mode of materialized table. */
     @PublicEvolving
     enum LogicalRefreshMode {

flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/ResolvedCatalogMaterializedTable.java

@@ -205,15 +205,6 @@ public String toString() {
 
     /** Convert this object to a {@link ResolvedCatalogTable} object for planner optimize query. */
     public ResolvedCatalogTable toResolvedCatalogTable() {
-        return new ResolvedCatalogTable(
-                CatalogTable.newBuilder()
-                        .schema(getUnresolvedSchema())
-                        .comment(getComment())
-                        .distribution(getDistribution().orElse(null))
-                        .partitionKeys(getPartitionKeys())
-                        .options(getOptions())
-                        .snapshot(getSnapshot().orElse(null))
-                        .build(),
-                getResolvedSchema());
+        return new ResolvedCatalogTable(origin.toCatalogTable(), getResolvedSchema());
     }
 }