lsyldliu commented on code in PR #25880:
URL: https://github.com/apache/flink/pull/25880#discussion_r1900832678
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/materializedtable/MaterializedTableManager.java:
##########
@@ -804,6 +808,78 @@ protected static String getRefreshStatement(
return insertStatement.toString();
}
+ private ResultFetcher callAlterMaterializedTableAsQueryOperation(
+ OperationExecutor operationExecutor,
+ OperationHandle handle,
+ AlterMaterializedTableAsQueryOperation op) {
+ ObjectIdentifier tableIdentifier = op.getTableIdentifier();
+ CatalogMaterializedTable materializedTable =
+ getCatalogMaterializedTable(operationExecutor,
tableIdentifier);
+
+ // 1. suspend the materialized table
+ if (CatalogMaterializedTable.RefreshStatus.SUSPENDED
+ != materializedTable.getRefreshStatus()) {
+ if (CatalogMaterializedTable.RefreshMode.CONTINUOUS
+ == materializedTable.getRefreshMode()) {
+ suspendContinuousRefreshJob(
+ operationExecutor, handle, tableIdentifier,
materializedTable);
+ } else {
+ suspendRefreshWorkflow(
+ operationExecutor, handle, tableIdentifier,
materializedTable);
+ }
+ }
+
+ // 2. replace query definition and resume the materialized table
+ // alter materialized table schema
+ operationExecutor.callExecutableOperation(handle, op);
Review Comment:
We shouldn't alter the materialized table metadata firstly, only the refresh
job or workflow resume sucessfully then we can alter the materialized table
definition query.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
