[FLINK-39957][table] Support EXPLAIN of CREATE [OR ALTER] MATERIALIZED TABLE

flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/materializedtable/MaterializedTableManager.java

@@ -954,7 +954,8 @@ private ResultFetcher callAlterMaterializedTableChangeOperation(
                     new AlterMaterializedTableChangeOperation(
                             op.getTableIdentifier(),
                             oldTable -> op.getTableChanges(),
-                            suspendMaterializedTable);
+                            suspendMaterializedTable,
+                            op.getSinkModifyQuery());
             operationExecutor.callExecutableOperation(
                     handle, alterMaterializedTableChangeOperation);
 
@@ -1010,7 +1011,10 @@ private ResultFetcher callAlterMaterializedTableChangeOperation(
 
             AlterMaterializedTableChangeOperation alterMaterializedTableChangeOperation =
                     new AlterMaterializedTableChangeOperation(
-                            tableIdentifier, oldTable -> tableChanges, oldMaterializedTable);
+                            tableIdentifier,
+                            oldTable -> tableChanges,
+                            oldMaterializedTable,
+                            op.getSinkModifyQuery());
 
             operationExecutor.callExecutableOperation(
                     handle, alterMaterializedTableChangeOperation);
@@ -1029,7 +1033,10 @@ private AlterMaterializedTableChangeOperation generateRollbackAlterMaterializedT
             AlterMaterializedTableChangeOperation op) {
 
         return new AlterMaterializedTableChangeOperation(
-                op.getTableIdentifier(), oldTable -> List.of(), oldMaterializedTable);
+                op.getTableIdentifier(),
+                oldTable -> List.of(),
+                oldMaterializedTable,
+                op.getSinkModifyQuery());
     }
 
     private TableChange.ModifyRefreshHandler generateResetSavepointTableChange(

flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/operation/OperationExecutor.java

@@ -494,7 +494,8 @@ private ResultFetcher executeOperationInStatementSetState(
             TableEnvironmentInternal tableEnv, OperationHandle handle, Operation operation) {
         if (operation instanceof EndStatementSetOperation) {
             return callEndStatementSetOperation(tableEnv, handle);
-        } else if (operation instanceof ModifyOperation) {
+        } else if (operation instanceof ModifyOperation
+                && !(operation instanceof MaterializedTableOperation)) {
             sessionContext.addStatementSetOperation((ModifyOperation) operation);
             return ResultFetcher.fromTableResult(handle, TABLE_RESULT_OK, false);
         } else {
@@ -518,7 +519,7 @@ private ResultFetcher executeOperation(
         } else if (op instanceof EndStatementSetOperation) {
             throw new SqlExecutionException(
                     "No Statement Set to submit. 'END' statement should be used after 'BEGIN STATEMENT SET'.");
-        } else if (op instanceof ModifyOperation) {
+        } else if (op instanceof ModifyOperation && !(op instanceof MaterializedTableOperation)) {
             return callModifyOperations(
                     tableEnv, handle, Collections.singletonList((ModifyOperation) op));
         } else if (op instanceof CompileAndExecutePlanOperation

flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/MaterializedTableConversionITCase.java

@@ -57,6 +57,7 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicLong;
 
+import static org.apache.flink.table.catalog.CatalogBaseTable.TableKind.TABLE;
 import static org.apache.flink.table.catalog.CommonCatalogOptions.TABLE_CATALOG_STORE_KIND;
 import static org.apache.flink.table.factories.FactoryUtil.WORKFLOW_SCHEDULER_TYPE;
 import static org.apache.flink.table.gateway.service.MaterializedTableTestUtils.executeStatement;
@@ -293,6 +294,40 @@ void testConvertTableLeftSuspendedWhenRefreshJobFailsToStart() throws Exception
         assertThat(suspendedMaterializedTable.getRefreshStatus()).isSameAs(RefreshStatus.SUSPENDED);
     }
 
+    @Test
+    void testExplainConvertTableToMaterializedTable() throws Exception {
+        // pre-create a regular table that CREATE OR ALTER would convert in place.
+        String createRegularTableDDL =
+                "CREATE TABLE users_shops (\n"
+                        + "  user_id BIGINT,\n"
+                        + "  shop_id BIGINT,\n"
+                        + "  payment_amount_cents BIGINT\n"
+                        + ")\n"
+                        + "WITH ('connector' = 'values')";
+        OperationHandle createTableHandle =
+                executeStatement(service, sessionHandle, createRegularTableDDL);
+        awaitOperationTermination(service, sessionHandle, createTableHandle);
+
+        String explainDDL =
+                "EXPLAIN CREATE OR ALTER MATERIALIZED TABLE users_shops"
+                        + " FRESHNESS = INTERVAL '30' SECOND"
+                        + " AS SELECT user_id, shop_id, payment_amount_cents FROM datagenSource";
+
+        // EXPLAIN succeeds and returns a single non-empty plan. The plan content itself is asserted
+        // by the planner tests in TableSinkTest.
+        OperationHandle explainHandle = executeStatement(service, sessionHandle, explainDDL);
+        awaitOperationTermination(service, sessionHandle, explainHandle);
+        List<RowData> explainResults = fetchAllResults(service, sessionHandle, explainHandle);
+        assertThat(explainResults).hasSize(1);
+        assertThat(explainResults.get(0).getString(0).toString()).isNotBlank();
+
+        // EXPLAIN is side-effect-free: the entry is still a regular table, not converted.
+        assertThat(
+                        service.getTable(sessionHandle, getObjectIdentifier("users_shops"))
+                                .getTableKind())
+                .isSameAs(TABLE);
+    }
+
     private SessionHandle initializeSession() {
         SessionHandle sessionHandle = service.openSession(defaultSessionEnvironment);
         String catalogDDL =

flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/MaterializedTableStatementITCase.java

@@ -1858,6 +1858,83 @@ void testRefreshMaterializedTableWithInvalidParameterInContinuousMode() throws E
         dropMaterializedTable(getObjectIdentifier("my_materialized_table"));
     }
 
+    @Test
+    void testExplainCreateOrAlterMaterializedTableOnExistingTable() throws Exception {
+        createAndVerifyCreateMaterializedTableWithData(
+                "users_shops", List.of(), Map.of(), RefreshMode.FULL);
+        ObjectIdentifier userShopsIdentifier = getObjectIdentifier("users_shops");
+        ResolvedCatalogMaterializedTable oldTable = getTable(userShopsIdentifier);
+
+        // CREATE OR ALTER on an existing table behaves like an alter-with-query. Re-state the same
+        // definition so EXPLAIN plans the sink over the (unchanged) table.
+        String explainDDL =
+                "EXPLAIN CREATE OR ALTER MATERIALIZED TABLE users_shops"
+                        + " PARTITIONED BY (ds)"
+                        + " WITH ('format' = 'debezium-json')"
+                        + " FRESHNESS = INTERVAL '30' SECOND"
+                        + " REFRESH_MODE = FULL"
+                        + " AS SELECT"
+                        + "  user_id,"
+                        + "  shop_id,"
+                        + "  ds,"
+                        + "  COUNT(order_id) AS order_cnt"
+                        + " FROM ("
+                        + "    SELECT user_id, shop_id, order_created_at AS ds, order_id FROM my_source"
+                        + " ) AS tmp"
+                        + " GROUP BY (user_id, shop_id, ds)";
+
+        // EXPLAIN succeeds (the storage options carried over from the existing table avoid a
+        // "Missing required options" failure) and returns a single non-empty plan. The plan content
+        // itself is asserted by the planner tests in TableSinkTest.
+        OperationHandle explainHandle = executeStatement(explainDDL);
+        awaitOperationTermination(service, sessionHandle, explainHandle);
+        List<RowData> explainResults = fetchAllResults(service, sessionHandle, explainHandle);
+        assertThat(explainResults).hasSize(1);
+        assertThat(explainResults.get(0).getString(0).toString()).isNotBlank();
+
+        // EXPLAIN is side-effect-free: the existing materialized table is unchanged.
+        ResolvedCatalogMaterializedTable newTable = getTable(userShopsIdentifier);
+        assertThat(newTable.getResolvedSchema()).isEqualTo(oldTable.getResolvedSchema());
+        assertThat(newTable.getExpandedQuery()).isEqualTo(oldTable.getExpandedQuery());
+
+        dropMaterializedTable(userShopsIdentifier);
+    }
+
+    @Test
+    void testExplainAlterMaterializedTableAsQuery() throws Exception {
+        createAndVerifyCreateMaterializedTableWithData(
+                "users_shops", List.of(), Map.of(), RefreshMode.FULL);
+        ObjectIdentifier userShopsIdentifier = getObjectIdentifier("users_shops");
+        ResolvedCatalogMaterializedTable oldTable = getTable(userShopsIdentifier);
+
+        String explainDDL =
+                "EXPLAIN ALTER MATERIALIZED TABLE users_shops"
+                        + " AS SELECT"
+                        + "  user_id,"
+                        + "  shop_id,"
+                        + "  ds,"
+                        + "  COUNT(order_id) AS order_cnt"
+                        + " FROM ("
+                        + "    SELECT user_id, shop_id, order_created_at AS ds, order_id FROM my_source"
+                        + " ) AS tmp"
+                        + " GROUP BY (user_id, shop_id, ds)";
+
+        // EXPLAIN succeeds and returns a single non-empty plan. The plan content itself is asserted
+        // by the planner tests in TableSinkTest.
+        OperationHandle explainHandle = executeStatement(explainDDL);
+        awaitOperationTermination(service, sessionHandle, explainHandle);
+        List<RowData> explainResults = fetchAllResults(service, sessionHandle, explainHandle);
+        assertThat(explainResults).hasSize(1);
+        assertThat(explainResults.get(0).getString(0).toString()).isNotBlank();
+
+        // EXPLAIN is side-effect-free: the materialized table definition is unchanged.
+        ResolvedCatalogMaterializedTable newTable = getTable(userShopsIdentifier);
+        assertThat(newTable.getResolvedSchema()).isEqualTo(oldTable.getResolvedSchema());
+        assertThat(newTable.getExpandedQuery()).isEqualTo(oldTable.getExpandedQuery());
+
+        dropMaterializedTable(userShopsIdentifier);
+    }
+
     private void setupSavepointDir(Path temporaryPath) throws Exception {
         String savepointDir = "file://" + temporaryPath.toAbsolutePath();
         String setupSavepointDDL =

flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl

@@ -3138,16 +3138,25 @@ SqlNode SqlRichExplain() :
         stmt = RichSqlInsert()
         |
         stmt = SqlCreate()
+        |
+        stmt = SqlAlterMaterializedTable()
     )
     {
         if ((stmt instanceof SqlCreate)
                 && !(stmt instanceof SqlCreateTableAs)
-                && !(stmt instanceof SqlReplaceTableAs)) {
+                && !(stmt instanceof SqlReplaceTableAs)
+                && !(stmt instanceof SqlCreateOrAlterMaterializedTable)) {
             throw SqlUtil.newContextException(
                 getPos(),
                 ParserResource.RESOURCE.explainCreateOrReplaceStatementUnsupported());
         }
 
+        if ((stmt instanceof SqlAlterMaterializedTable) && !(stmt instanceof SqlAlterMaterializedTableAsQuery)) {
+            throw SqlUtil.newContextException(
+                getPos(),
+                ParserResource.RESOURCE.explainAlterMaterializedTableUnsupported());
+        }
+
         return new SqlRichExplain(getPos(), stmt, explainDetails);
     }
 }

flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/utils/ParserResource.java

@@ -53,6 +53,10 @@ public interface ParserResource {
             "Unsupported CREATE OR REPLACE statement for EXPLAIN. The statement must define a query using the AS clause (i.e. CTAS/RTAS statements).")
     Resources.ExInst<ParseException> explainCreateOrReplaceStatementUnsupported();
 
+    @Resources.BaseMessage(
+            "Unsupported ALTER MATERIALIZED TABLE statement for EXPLAIN. The statement must define a query using the AS clause.")
+    Resources.ExInst<ParseException> explainAlterMaterializedTableUnsupported();
+
     @Resources.BaseMessage(
             "Columns identifiers without types in the schema are supported on CTAS/RTAS statements only.")
     Resources.ExInst<ParseException> columnsIdentifiersUnsupported();

flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java

@@ -3170,29 +3170,40 @@ void testAnalyzeTable() {
                 .fails("(?s).*Encountered \"\\,\" at line 1, column 32.\n.*");
     }
 
-    @Test
-    void testExplainCreateTableNoSupported() {
-        this.sql("EXPLAIN CREATE TABLE t (id int^)^")
+    @ParameterizedTest
+    @CsvSource({"CREATE", "CREATE OR REPLACE"})
+    void testExplainCreateTableNotSupported(final String operation) {
+        this.sql(String.format("EXPLAIN %s TABLE t (id int^)^", operation))
                 .fails(
                         "Unsupported CREATE OR REPLACE statement for EXPLAIN\\. The statement must define a query using the AS clause \\(i\\.e\\. CTAS/RTAS statements\\)\\.");
     }
 
-    @Test
-    void testExplainCreateTableAsSelect() {
-        this.sql("EXPLAIN CREATE TABLE t AS SELECT * FROM b")
-                .ok("EXPLAIN CREATE TABLE `T`\nAS\nSELECT *\nFROM `B`");
-    }
-
-    @Test
-    void testExplainCreateOrReplaceTableAsSelect() {
-        this.sql("EXPLAIN CREATE OR REPLACE TABLE t AS SELECT * FROM b")
-                .ok("EXPLAIN CREATE OR REPLACE TABLE `T`\nAS\nSELECT *\nFROM `B`");
+    @ParameterizedTest
+    @CsvSource({
+        "ALTER MATERIALIZED TABLE t ^SUSPEND^",
+        "ALTER MATERIALIZED TABLE t ^RESUME^",
+        "ALTER MATERIALIZED TABLE t ADD category_id STRING METADATA ^VIRTUAL^",
+        "ALTER MATERIALIZED TABLE t MODIFY measurement double ^METADATA^"
+    })
+    void testExplainAlterMaterializedTableWithoutAsQueryNotSupported(
+            final String alterMaterializedTable) {
+        this.sql(String.format("EXPLAIN %s", alterMaterializedTable))
+                .fails(
+                        "Unsupported ALTER MATERIALIZED TABLE statement for EXPLAIN\\. The statement must define a query using the AS clause.");
     }
 
-    @Test
-    void testExplainReplaceTableAsSelect() {
-        this.sql("EXPLAIN REPLACE TABLE t AS SELECT * FROM b")
-                .ok("EXPLAIN REPLACE TABLE `T`\nAS\nSELECT *\nFROM `B`");
+    @ParameterizedTest
+    @CsvSource({
+        "CREATE",
+        "CREATE OR REPLACE",
+        "REPLACE",
+        "CREATE MATERIALIZED",
+        "ALTER MATERIALIZED",
+        "CREATE OR ALTER MATERIALIZED",
+    })
+    void testExplain(final String operation) {
+        this.sql(String.format("EXPLAIN %s TABLE t AS SELECT * FROM b", operation))
+                .ok(String.format("EXPLAIN %s TABLE `T`\nAS\nSELECT *\nFROM `B`", operation));
     }
 
     @Test

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

@@ -19,6 +19,9 @@
 package org.apache.flink.table.operations;
 
 import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.operations.materializedtable.AlterMaterializedTableChangeOperation;
+import org.apache.flink.table.operations.materializedtable.ConvertTableToMaterializedTableOperation;
+import org.apache.flink.table.operations.materializedtable.CreateMaterializedTableOperation;
 
 /**
  * Class that implements visitor pattern. It allows type safe logic on top of tree of {@link
@@ -39,4 +42,10 @@ public interface ModifyOperationVisitor<T> {
     T visit(CreateTableASOperation ctas);
 
     T visit(ReplaceTableAsOperation rtas);
+
+    T visit(CreateMaterializedTableOperation createMaterializedTableOperation);
+
+    T visit(AlterMaterializedTableChangeOperation alterMaterializedTableChangeOperation);
+
+    T visit(ConvertTableToMaterializedTableOperation convertTableToMaterializedTableOperation);
 }

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

@@ -23,6 +23,7 @@
 import org.apache.flink.table.catalog.ObjectIdentifier;
 import org.apache.flink.table.catalog.ResolvedCatalogMaterializedTable;
 import org.apache.flink.table.catalog.TableChange;
+import org.apache.flink.table.operations.QueryOperation;
 
 import java.util.List;
 import java.util.function.Function;
@@ -40,8 +41,9 @@ public class AlterMaterializedTableAsQueryOperation extends AlterMaterializedTab
     public AlterMaterializedTableAsQueryOperation(
             ObjectIdentifier tableIdentifier,
             Function<ResolvedCatalogMaterializedTable, List<TableChange>> tableChangesForTable,
-            ResolvedCatalogMaterializedTable oldTable) {
-        super(tableIdentifier, tableChangesForTable, oldTable);
+            ResolvedCatalogMaterializedTable oldTable,
+            QueryOperation sinkModifyQuery) {
+        super(tableIdentifier, tableChangesForTable, oldTable, sinkModifyQuery);
     }
 
     @Override

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

@@ -37,6 +37,9 @@
 import org.apache.flink.table.catalog.TableChange.ModifyRefreshHandler;
 import org.apache.flink.table.catalog.TableChange.ModifyRefreshStatus;
 import org.apache.flink.table.catalog.TableChange.ModifyStartMode;
+import org.apache.flink.table.operations.ModifyOperation;
+import org.apache.flink.table.operations.ModifyOperationVisitor;
+import org.apache.flink.table.operations.QueryOperation;
 import org.apache.flink.table.operations.ddl.AlterTableChangeOperation;
 
 import java.util.ArrayList;
@@ -50,9 +53,11 @@
  * Alter materialized table with new table definition and table changes represents the modification.
  */
 @Internal
-public class AlterMaterializedTableChangeOperation extends AlterMaterializedTableOperation {
+public class AlterMaterializedTableChangeOperation extends AlterMaterializedTableOperation
+        implements ModifyOperation {
 
     private final Function<ResolvedCatalogMaterializedTable, List<TableChange>> tableChangeForTable;
+    private final QueryOperation sinkModifyQuery;
     private ResolvedCatalogMaterializedTable oldTable;
     private MaterializedTableChangeHandler handler;
     private CatalogMaterializedTable newTable;
@@ -62,9 +67,23 @@ public AlterMaterializedTableChangeOperation(
             ObjectIdentifier tableIdentifier,
             Function<ResolvedCatalogMaterializedTable, List<TableChange>> tableChangeForTable,
             ResolvedCatalogMaterializedTable oldTable) {
+        // Metadata-only changes (options, schema, distribution, ...) carry no sink-modifying query.
+        this(tableIdentifier, tableChangeForTable, oldTable, null);
+    }
+
+    public AlterMaterializedTableChangeOperation(
+            ObjectIdentifier tableIdentifier,
+            Function<ResolvedCatalogMaterializedTable, List<TableChange>> tableChangeForTable,
+            ResolvedCatalogMaterializedTable oldTable,
+            QueryOperation sinkModifyQuery) {
         super(tableIdentifier);
         this.tableChangeForTable = tableChangeForTable;
         this.oldTable = oldTable;
+        this.sinkModifyQuery = sinkModifyQuery;
+    }
+
+    public QueryOperation getSinkModifyQuery() {
+        return sinkModifyQuery;
     }
 
     public List<TableChange> getTableChanges() {
@@ -76,7 +95,7 @@ public List<TableChange> getTableChanges() {
 
     public AlterMaterializedTableChangeOperation copyAsTableChangeOperation() {
         return new AlterMaterializedTableChangeOperation(
-                tableIdentifier, tableChangeForTable, oldTable);
+                tableIdentifier, tableChangeForTable, oldTable, sinkModifyQuery);
     }
 
     public CatalogMaterializedTable getNewTable() {
@@ -144,6 +163,16 @@ public String asSummaryString() {
                 "%s %s\n%s", getOperationName(), tableIdentifier.asSummaryString(), changes);
     }
 
+    @Override
+    public QueryOperation getChild() {
+        return this.sinkModifyQuery;
+    }
+
+    @Override
+    public <T> T accept(final ModifyOperationVisitor<T> visitor) {
+        return visitor.visit(this);
+    }
+
     /** Hook for subclasses to provide a different new-table builder. */
     protected CatalogMaterializedTable computeNewTable() {
         return MaterializedTableChangeHandler.buildNewMaterializedTable(getHandlerWithChanges());