[FLINK-36994][table] Support ALTER MATERIALIZED TABLE As <Query> statement #25880

hackergin · 2025-01-02T03:30:21Z

What is the purpose of the change

Add support query modification for materialized table

Brief change log

**

Add support parse materialized table query modification statements
Add support convert materialized table node to operation
Add support execution of query modification of materialized table

Verifying this change

Some unit test and ITCase will be added in MaterializedTableStatementITCase to verify these changes.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
The serializers: (no)
The runtime per-record code paths (performance sensitive): (no)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
The S3 file system connector: (no)

Documentation

Does this pull request introduce a new feature? (yes)
If yes, how is the feature documented? (Will be added in a separated pr)

flinkbot · 2025-01-02T03:34:13Z

CI report:

480c394 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

lsyldliu

@hackergin Thanks for your contribution, I left some comments.

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

.../apache/flink/table/operations/materializedtable/AlterMaterializedTableAsQueryOperation.java

.../java/org/apache/flink/table/gateway/service/materializedtable/MaterializedTableManager.java

.../apache/flink/table/planner/operations/SqlMaterializedTableNodeToOperationConverterTest.java

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

.../apache/flink/table/operations/materializedtable/AlterMaterializedTableAsQueryOperation.java

lsyldliu · 2025-01-03T02:01:45Z

.../apache/flink/table/operations/materializedtable/AlterMaterializedTableAsQueryOperation.java

+            }
+        }
+        for (int i = oldSchema.getColumns().size(); i < newSchema.getColumns().size(); i++) {
+            Column newColumn = newSchema.getColumns().get(i);


If the new added column is duplicated with old column, what behavior is? I think you should a test case to verify it.

Based on my testing, if duplicate field names appear in a query, the latter field names will automatically have suffixes added, which seems consistent with the logic of the Calcite framework. For example, if there are two fields named a, the second a will automatically be renamed to a0.

…ement

…ode to operation

…eration

lsyldliu

Thanks for updating, I left some comments.

lsyldliu · 2025-01-06T11:09:21Z

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

@@ -1374,4 +1391,24 @@ public String toString() {
                    + '}';
        }
    }
+
+    /* A table change to modify the definition query. */


Suggested change

/* A table change to modify the definition query. */

/** A table change to modify the definition query. */

lsyldliu · 2025-01-06T11:13:39Z

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

@@ -394,7 +404,7 @@ static ModifyRefreshHandler modifyRefreshHandler(
     * </pre>
     */
    @PublicEvolving
-    class AddColumn implements TableChange {
+    class AddColumn implements CatalogTableChange, MaterializedTableChange {


Please give more java docs to explain it also support materialized table

lsyldliu · 2025-01-06T11:17:06Z

.../apache/flink/table/operations/materializedtable/AlterMaterializedTableAsQueryOperation.java

+@Internal
+public class AlterMaterializedTableAsQueryOperation extends AlterMaterializedTableOperation {
+
+    private final List<TableChange> columnChanges;


Use MaterializedTableChange

lsyldliu · 2025-01-06T11:17:12Z

.../apache/flink/table/operations/materializedtable/AlterMaterializedTableAsQueryOperation.java

+
+    private final List<TableChange> columnChanges;
+
+    private final TableChange definitionQueryChange;


We can put it the above list, If you need it, you can filter it from the list.

lsyldliu · 2025-01-06T11:17:56Z

.../apache/flink/table/operations/materializedtable/AlterMaterializedTableAsQueryOperation.java

+                "ALTER MATERIALIZED TABLE",
+                params,
+                Collections.emptyList(),
+                Operation::asSummaryString);


I think we only need to print the definition query here.

lsyldliu · 2025-01-06T12:18:59Z

.../java/org/apache/flink/table/gateway/service/materializedtable/MaterializedTableManager.java

+                TableChange.AddColumn addColumn = (TableChange.AddColumn) tableChange;
+                rollbackChanges.add(TableChange.dropColumn(addColumn.getColumn().getName()));
+            } else if (tableChange instanceof TableChange.ModifyRefreshHandler) {
+                rollbackChanges.add(


This is not correct, we should use the refresh handler of rollback refresh job

lsyldliu · 2025-01-06T12:36:08Z

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

+                .isEqualTo(
+                        Collections.singletonList(
+                                Column.physical("order_amount_sum", DataTypes.INT())));
+        assertThat(oldTable.getDefinitionQuery()).isNotEqualTo(newTable.getDefinitionQuery());


I think we should verify the definition query is equals.

lsyldliu · 2025-01-06T12:37:33Z

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

+                .isEqualTo(
+                        Collections.singletonList(
+                                Column.physical("order_amount_sum", DataTypes.INT())));
+        assertThat(oldTable.getDefinitionQuery()).isNotEqualTo(newTable.getDefinitionQuery());


lsyldliu · 2025-01-06T12:38:35Z

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

+                .isEqualTo(oldTable.getResolvedSchema().getPrimaryKey());
+        assertThat(newTable.getResolvedSchema().getWatermarkSpecs())
+                .isEqualTo(oldTable.getResolvedSchema().getWatermarkSpecs());
+        assertThat(oldTable.getDefinitionQuery()).isNotEqualTo(newTable.getDefinitionQuery());


lsyldliu · 2025-01-06T12:38:50Z

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

+
+        assertThat(getAddedColumns(newTable.getResolvedSchema(), oldTable.getResolvedSchema()))
+                .isEqualTo(Collections.singletonList(Column.physical("pv", DataTypes.INT())));
+        assertThat(oldTable.getDefinitionQuery()).isNotEqualTo(newTable.getDefinitionQuery());


hackergin changed the title ~~Drat: [FLIP-492] Support query modification for materialized table~~ Draft: [FLIP-492] Support query modification for materialized table Jan 2, 2025

hackergin force-pushed the FLIP-492-poc branch from f2205e0 to 9e31130 Compare January 2, 2025 11:09

lsyldliu reviewed Jan 2, 2025

View reviewed changes

lsyldliu reviewed Jan 3, 2025

View reviewed changes

hackergin force-pushed the FLIP-492-poc branch from 9e31130 to 9a19d23 Compare January 5, 2025 14:07

hackergin changed the title ~~Draft: [FLIP-492] Support query modification for materialized table~~ [FLINK-36993][table] Support ALTER MATERIALIZED TABLE As <Query> statement Jan 5, 2025

hackergin force-pushed the FLIP-492-poc branch 2 times, most recently from 90c372c to a204971 Compare January 6, 2025 03:14

hackergin changed the title ~~[FLINK-36993][table] Support ALTER MATERIALIZED TABLE As <Query> statement~~ [FLINK-36994][table] Support ALTER MATERIALIZED TABLE As <Query> statement Jan 6, 2025

hackergin added 3 commits January 6, 2025 13:51

[FLINK-36994][table] Support parsing ALTER MATERIALIZED TABLE AS stat…

1c611fe

…ement

[FLINK-36994][table] Support converting ALTER MATERIALIZED TABLE AS n…

2c6cedc

…ode to operation

[FLINK-36994][table] Support executing ALTER MATERIALIZED TABLE AS op…

480c394

…eration

hackergin force-pushed the FLIP-492-poc branch from a204971 to 480c394 Compare January 6, 2025 05:52

lsyldliu reviewed Jan 6, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-36994][table] Support ALTER MATERIALIZED TABLE As <Query> statement #25880

[FLINK-36994][table] Support ALTER MATERIALIZED TABLE As <Query> statement #25880

hackergin commented Jan 2, 2025

flinkbot commented Jan 2, 2025 •

edited

Loading

lsyldliu left a comment

lsyldliu Jan 3, 2025

hackergin Jan 6, 2025

lsyldliu left a comment

lsyldliu Jan 6, 2025

lsyldliu Jan 6, 2025

lsyldliu Jan 6, 2025

lsyldliu Jan 6, 2025

lsyldliu Jan 6, 2025

lsyldliu Jan 6, 2025

lsyldliu Jan 6, 2025

lsyldliu Jan 6, 2025

lsyldliu Jan 6, 2025

lsyldliu Jan 6, 2025

	/* A table change to modify the definition query. */
	/** A table change to modify the definition query. */


		private final List<TableChange> columnChanges;

		private final TableChange definitionQueryChange;

[FLINK-36994][table] Support ALTER MATERIALIZED TABLE As <Query> statement #25880

Are you sure you want to change the base?

[FLINK-36994][table] Support ALTER MATERIALIZED TABLE As <Query> statement #25880

Conversation

hackergin commented Jan 2, 2025

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

flinkbot commented Jan 2, 2025 • edited Loading

CI report:

lsyldliu left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lsyldliu left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

flinkbot commented Jan 2, 2025 •

edited

Loading