From 5351ed767725ea0201d545194f9e22ae28818666 Mon Sep 17 00:00:00 2001 From: xuyang Date: Mon, 6 Jul 2026 19:11:37 +0800 Subject: [PATCH] [flink] Support ALTER MATERIALIZED TABLE ... AS in FlinkCatalog --- .../flink/adapter/TableChangeAdapter.java | 60 +++++ .../Flink22TableChangeAdapterTest.java | 59 +++++ .../Flink22MaterializedTableITCase.java | 205 +++++++++++++++++- .../utils/Flink22FlinkConversionsTest.java | 49 +++++ .../flink/adapter/TableChangeAdapter.java | 54 +++++ .../fluss/flink/utils/FlinkConversions.java | 7 + 6 files changed, 433 insertions(+), 1 deletion(-) create mode 100644 fluss-flink/fluss-flink-2.2/src/main/java/org/apache/fluss/flink/adapter/TableChangeAdapter.java create mode 100644 fluss-flink/fluss-flink-2.2/src/test/java/org/apache/fluss/flink/adapter/Flink22TableChangeAdapterTest.java create mode 100644 fluss-flink/fluss-flink-2.2/src/test/java/org/apache/fluss/flink/utils/Flink22FlinkConversionsTest.java create mode 100644 fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/adapter/TableChangeAdapter.java diff --git a/fluss-flink/fluss-flink-2.2/src/main/java/org/apache/fluss/flink/adapter/TableChangeAdapter.java b/fluss-flink/fluss-flink-2.2/src/main/java/org/apache/fluss/flink/adapter/TableChangeAdapter.java new file mode 100644 index 0000000000..c107063248 --- /dev/null +++ b/fluss-flink/fluss-flink-2.2/src/main/java/org/apache/fluss/flink/adapter/TableChangeAdapter.java @@ -0,0 +1,60 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.flink.adapter; + +import org.apache.flink.table.catalog.TableChange; + +import java.util.Collections; +import java.util.List; +import java.util.Optional; + +import static org.apache.fluss.flink.FlinkConnectorOptions.MATERIALIZED_TABLE_DEFINITION_QUERY; + +/** + * A Flink 2.x override of {@link TableChangeAdapter} that converts the {@link TableChange}s + * introduced since Flink 2.0. + * + *

To support a new version-specific table change, add a branch to {@link #convert}. + * + *

TODO: remove this class when no longer support all the Flink 1.x series. + */ +public class TableChangeAdapter { + + private TableChangeAdapter() {} + + /** + * Converts a version-specific Flink {@link TableChange} into a list of Fluss {@link + * org.apache.fluss.metadata.TableChange}. + * + * @return the converted Fluss table changes, or {@link Optional#empty()} if the given change is + * not a version-specific table change handled by this adapter (the caller should then treat + * it as unsupported). + */ + public static Optional> convert( + TableChange tableChange) { + if (tableChange instanceof TableChange.ModifyDefinitionQuery) { + String definitionQuery = + ((TableChange.ModifyDefinitionQuery) tableChange).getDefinitionQuery(); + return Optional.of( + Collections.singletonList( + org.apache.fluss.metadata.TableChange.set( + MATERIALIZED_TABLE_DEFINITION_QUERY.key(), definitionQuery))); + } + return Optional.empty(); + } +} diff --git a/fluss-flink/fluss-flink-2.2/src/test/java/org/apache/fluss/flink/adapter/Flink22TableChangeAdapterTest.java b/fluss-flink/fluss-flink-2.2/src/test/java/org/apache/fluss/flink/adapter/Flink22TableChangeAdapterTest.java new file mode 100644 index 0000000000..e5ca0ee665 --- /dev/null +++ b/fluss-flink/fluss-flink-2.2/src/test/java/org/apache/fluss/flink/adapter/Flink22TableChangeAdapterTest.java @@ -0,0 +1,59 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.flink.adapter; + +import org.apache.fluss.metadata.TableChange; + +import org.apache.flink.table.catalog.CatalogMaterializedTable; +import org.junit.jupiter.api.Test; + +import java.util.List; +import java.util.Optional; + +import static org.apache.fluss.flink.FlinkConnectorOptions.MATERIALIZED_TABLE_DEFINITION_QUERY; +import static org.assertj.core.api.Assertions.assertThat; + +/** Test for {@link TableChangeAdapter} in Flink 2.2. */ +public class Flink22TableChangeAdapterTest { + + @Test + void testConvertModifyDefinitionQuery() { + String definitionQuery = "SELECT * FROM new_source"; + Optional> result = + TableChangeAdapter.convert( + new org.apache.flink.table.catalog.TableChange.ModifyDefinitionQuery( + definitionQuery)); + + assertThat(result).isPresent(); + assertThat(result.get()).hasSize(1); + assertThat(result.get().get(0)).isInstanceOf(TableChange.SetOption.class); + TableChange.SetOption setOption = (TableChange.SetOption) result.get().get(0); + assertThat(setOption.getKey()).isEqualTo(MATERIALIZED_TABLE_DEFINITION_QUERY.key()); + assertThat(setOption.getValue()).isEqualTo(definitionQuery); + } + + @Test + void testConvertUnhandledChangeReturnsEmpty() { + // ModifyRefreshStatus is converted directly in FlinkConversions, not by this adapter. + Optional> result = + TableChangeAdapter.convert( + new org.apache.flink.table.catalog.TableChange.ModifyRefreshStatus( + CatalogMaterializedTable.RefreshStatus.ACTIVATED)); + assertThat(result).isEmpty(); + } +} diff --git a/fluss-flink/fluss-flink-2.2/src/test/java/org/apache/fluss/flink/catalog/Flink22MaterializedTableITCase.java b/fluss-flink/fluss-flink-2.2/src/test/java/org/apache/fluss/flink/catalog/Flink22MaterializedTableITCase.java index 36240466c4..6ed89f3490 100644 --- a/fluss-flink/fluss-flink-2.2/src/test/java/org/apache/fluss/flink/catalog/Flink22MaterializedTableITCase.java +++ b/fluss-flink/fluss-flink-2.2/src/test/java/org/apache/fluss/flink/catalog/Flink22MaterializedTableITCase.java @@ -17,5 +17,208 @@ package org.apache.fluss.flink.catalog; +import org.apache.flink.api.common.JobID; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.table.catalog.CatalogMaterializedTable; +import org.apache.flink.table.catalog.ObjectIdentifier; +import org.apache.flink.table.catalog.ResolvedCatalogMaterializedTable; +import org.apache.flink.table.gateway.api.operation.OperationHandle; +import org.apache.flink.table.refresh.ContinuousRefreshHandler; +import org.apache.flink.table.refresh.ContinuousRefreshHandlerSerializer; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.file.Path; + +import static org.apache.flink.table.gateway.service.utils.SqlGatewayServiceTestUtil.awaitOperationTermination; +import static org.apache.flink.test.util.TestUtils.waitUntilAllTasksAreRunning; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + /** IT case for materialized table in Flink 2.2. */ -public class Flink22MaterializedTableITCase extends MaterializedTableITCase {} +public class Flink22MaterializedTableITCase extends MaterializedTableITCase { + + @Test + void testAlterMaterializedTableAsQueryWithoutSchemaChangesInContinuousMode( + @TempDir Path temporaryPath) throws Exception { + String materializedTableDDL = + "CREATE MATERIALIZED TABLE shop_detail_alter\n" + + " FRESHNESS = INTERVAL '3' SECOND\n" + + " AS SELECT \n" + + " DATE_FORMAT(order_created_at, 'yyyy-MM-dd') AS ds,\n" + + " user_id,\n" + + " shop_id,\n" + + " payment_amount_cents\n" + + " FROM datagenSource"; + OperationHandle materializedTableHandle = + service.executeStatement( + sessionHandle, materializedTableDDL, -1, new Configuration()); + awaitOperationTermination(service, sessionHandle, materializedTableHandle); + + ObjectIdentifier mtIdentifier = + ObjectIdentifier.of(CATALOG_NAME, DEFAULT_DB, "shop_detail_alter"); + ResolvedCatalogMaterializedTable originMaterializedTable = + (ResolvedCatalogMaterializedTable) service.getTable(sessionHandle, mtIdentifier); + String originDefinitionQuery = originMaterializedTable.getDefinitionQuery(); + + ContinuousRefreshHandler originRefreshHandler = + ContinuousRefreshHandlerSerializer.INSTANCE.deserialize( + originMaterializedTable.getSerializedRefreshHandler(), + getClass().getClassLoader()); + waitUntilAllTasksAreRunning( + restClusterClient, JobID.fromHexString(originRefreshHandler.getJobId())); + + // altering in continuous mode stops the old job with a savepoint, so a savepoint dir is + // required. + String savepointDir = temporaryPath.toString(); + String alterJobSavepointDDL = + String.format( + "SET 'execution.checkpointing.savepoint-dir' = 'file://%s'", savepointDir); + OperationHandle alterSavepointHandle = + service.executeStatement( + sessionHandle, alterJobSavepointDDL, -1, new Configuration()); + awaitOperationTermination(service, sessionHandle, alterSavepointHandle); + + // modify the definition query, keeping the same output schema so that no column evolution + // is triggered. + String alterMaterializedTableAsQueryDDL = + "ALTER MATERIALIZED TABLE shop_detail_alter\n" + + " AS SELECT \n" + + " DATE_FORMAT(order_created_at, 'yyyy-MM-dd') AS ds,\n" + + " user_id,\n" + + " shop_id,\n" + + " payment_amount_cents\n" + + " FROM datagenSource\n" + + " WHERE order_id <> 12345"; + OperationHandle alterAsQueryHandle = + service.executeStatement( + sessionHandle, alterMaterializedTableAsQueryDDL, -1, new Configuration()); + awaitOperationTermination(service, sessionHandle, alterAsQueryHandle); + + // the definition query should have been updated in the catalog. + ResolvedCatalogMaterializedTable alteredMaterializedTable = + (ResolvedCatalogMaterializedTable) service.getTable(sessionHandle, mtIdentifier); + assertThat(alteredMaterializedTable.getDefinitionQuery()) + .isNotEqualTo(originDefinitionQuery) + .contains("12345"); + assertThat(alteredMaterializedTable.getRefreshStatus()) + .isEqualTo(CatalogMaterializedTable.RefreshStatus.ACTIVATED); + + // the new refresh job should be running. + ContinuousRefreshHandler alteredRefreshHandler = + ContinuousRefreshHandlerSerializer.INSTANCE.deserialize( + alteredMaterializedTable.getSerializedRefreshHandler(), + getClass().getClassLoader()); + waitUntilAllTasksAreRunning( + restClusterClient, JobID.fromHexString(alteredRefreshHandler.getJobId())); + + dropMaterializedTable(mtIdentifier); + } + + @Test + void testAlterMaterializedTableAsQueryWithUnsupportedSchemaChanges(@TempDir Path temporaryPath) + throws Exception { + String materializedTableDDL = + "CREATE MATERIALIZED TABLE shop_detail_unsupported\n" + + " FRESHNESS = INTERVAL '3' SECOND\n" + + " AS SELECT \n" + + " DATE_FORMAT(order_created_at, 'yyyy-MM-dd') AS ds,\n" + + " user_id,\n" + + " shop_id,\n" + + " payment_amount_cents\n" + + " FROM datagenSource"; + OperationHandle materializedTableHandle = + service.executeStatement( + sessionHandle, materializedTableDDL, -1, new Configuration()); + awaitOperationTermination(service, sessionHandle, materializedTableHandle); + + ObjectIdentifier mtIdentifier = + ObjectIdentifier.of(CATALOG_NAME, DEFAULT_DB, "shop_detail_unsupported"); + ResolvedCatalogMaterializedTable originMaterializedTable = + (ResolvedCatalogMaterializedTable) service.getTable(sessionHandle, mtIdentifier); + assertThat(originMaterializedTable.getResolvedSchema().getColumnNames()) + .containsExactly("ds", "user_id", "shop_id", "payment_amount_cents"); + + ContinuousRefreshHandler originRefreshHandler = + ContinuousRefreshHandlerSerializer.INSTANCE.deserialize( + originMaterializedTable.getSerializedRefreshHandler(), + getClass().getClassLoader()); + waitUntilAllTasksAreRunning( + restClusterClient, JobID.fromHexString(originRefreshHandler.getJobId())); + + String savepointDir = temporaryPath.toString(); + String alterJobSavepointDDL = + String.format( + "SET 'execution.checkpointing.savepoint-dir' = 'file://%s'", savepointDir); + OperationHandle alterSavepointHandle = + service.executeStatement( + sessionHandle, alterJobSavepointDDL, -1, new Configuration()); + awaitOperationTermination(service, sessionHandle, alterSavepointHandle); + + // although appending a nullable column (status) at the end of the schema is the only + // schema evolution flink supports for materialized tables, fluss rejects it because the + // table alteration can only be applied to one of the following: table properties or table + // schema + String addColumnDDL = + "ALTER MATERIALIZED TABLE shop_detail_unsupported\n" + + " AS SELECT \n" + + " DATE_FORMAT(order_created_at, 'yyyy-MM-dd') AS ds,\n" + + " user_id,\n" + + " shop_id,\n" + + " payment_amount_cents,\n" + + " status\n" + + " FROM datagenSource"; + + assertThatThrownBy( + () -> { + OperationHandle handle = + service.executeStatement( + sessionHandle, addColumnDDL, -1, new Configuration()); + awaitOperationTermination(service, sessionHandle, handle); + }) + .hasStackTraceContaining( + "Table alteration can only be applied to one of the following: " + + "table properties or table schema"); + + // dropping a column is rejected by flink at statement-compile time (before touching the + // running job), so no savepoint dir is required. + String dropColumnDDL = + "ALTER MATERIALIZED TABLE shop_detail_unsupported\n" + + " AS SELECT \n" + + " DATE_FORMAT(order_created_at, 'yyyy-MM-dd') AS ds,\n" + + " user_id,\n" + + " shop_id\n" + + " FROM datagenSource"; + assertThatThrownBy( + () -> { + OperationHandle handle = + service.executeStatement( + sessionHandle, dropColumnDDL, -1, new Configuration()); + awaitOperationTermination(service, sessionHandle, handle); + }) + .hasStackTraceContaining("drop column is unsupported"); + + // reordering columns is rejected as well. + String reorderColumnDDL = + "ALTER MATERIALIZED TABLE shop_detail_unsupported\n" + + " AS SELECT \n" + + " DATE_FORMAT(order_created_at, 'yyyy-MM-dd') AS ds,\n" + + " shop_id,\n" + + " user_id,\n" + + " payment_amount_cents\n" + + " FROM datagenSource"; + assertThatThrownBy( + () -> { + OperationHandle handle = + service.executeStatement( + sessionHandle, + reorderColumnDDL, + -1, + new Configuration()); + awaitOperationTermination(service, sessionHandle, handle); + }) + .hasStackTraceContaining("reordering columns are not supported"); + + dropMaterializedTable(mtIdentifier); + } +} diff --git a/fluss-flink/fluss-flink-2.2/src/test/java/org/apache/fluss/flink/utils/Flink22FlinkConversionsTest.java b/fluss-flink/fluss-flink-2.2/src/test/java/org/apache/fluss/flink/utils/Flink22FlinkConversionsTest.java new file mode 100644 index 0000000000..6b76512261 --- /dev/null +++ b/fluss-flink/fluss-flink-2.2/src/test/java/org/apache/fluss/flink/utils/Flink22FlinkConversionsTest.java @@ -0,0 +1,49 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.flink.utils; + +import org.apache.fluss.metadata.TableChange; + +import org.apache.flink.table.catalog.TableChange.ModifyDefinitionQuery; +import org.junit.jupiter.api.Test; + +import java.util.List; + +import static org.apache.fluss.flink.FlinkConnectorOptions.MATERIALIZED_TABLE_DEFINITION_QUERY; +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Test for {@link FlinkConversions} against the Flink 2.x-only {@link ModifyDefinitionQuery} table + * change. + */ +class Flink22FlinkConversionsTest { + + @Test + void testConvertModifyDefinitionQuery() { + String newDefinitionQuery = "SELECT order_id, orig_ts FROM new_source"; + + List flussTableChanges = + FlinkConversions.toFlussTableChanges(new ModifyDefinitionQuery(newDefinitionQuery)); + + assertThat(flussTableChanges).hasSize(1); + assertThat(flussTableChanges.get(0)).isInstanceOf(TableChange.SetOption.class); + TableChange.SetOption setOption = (TableChange.SetOption) flussTableChanges.get(0); + assertThat(setOption.getKey()).isEqualTo(MATERIALIZED_TABLE_DEFINITION_QUERY.key()); + assertThat(setOption.getValue()).isEqualTo(newDefinitionQuery); + } +} diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/adapter/TableChangeAdapter.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/adapter/TableChangeAdapter.java new file mode 100644 index 0000000000..ac8d772270 --- /dev/null +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/adapter/TableChangeAdapter.java @@ -0,0 +1,54 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.flink.adapter; + +import org.apache.flink.table.catalog.TableChange; + +import java.util.List; +import java.util.Optional; + +/** + * An adapter for the Flink {@link TableChange}s that are only available in Flink versions newer + * than the one this module is compiled against, and therefore cannot be referenced directly here + * (e.g. {@code TableChange.ModifyDefinitionQuery}, introduced in Flink 2.0). + * + *

A version-specific override provides the conversion. To support a new version-specific table + * change in the future, only the corresponding override needs a new branch in {@link #convert} - no + * extra methods or call-site changes are required. + * + *

TODO: remove this class when no longer support all the Flink 1.x series. + */ +public class TableChangeAdapter { + + private TableChangeAdapter() {} + + /** + * Converts a version-specific Flink {@link TableChange} into a list of Fluss {@link + * org.apache.fluss.metadata.TableChange}. + * + * @return the converted Fluss table changes, or {@link Optional#empty()} if the given change is + * not a version-specific table change handled by this adapter (the caller should then treat + * it as unsupported). + */ + public static Optional> convert( + TableChange tableChange) { + // The Flink version this module is compiled against has no version-specific table change to + // convert. + return Optional.empty(); + } +} diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConversions.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConversions.java index 01f9a9817e..6b4fb3f4cd 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConversions.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConversions.java @@ -22,6 +22,7 @@ import org.apache.fluss.config.MemorySize; import org.apache.fluss.config.Password; import org.apache.fluss.flink.adapter.CatalogTableAdapter; +import org.apache.fluss.flink.adapter.TableChangeAdapter; import org.apache.fluss.flink.catalog.FlinkCatalogFactory; import org.apache.fluss.metadata.AggFunction; import org.apache.fluss.metadata.DatabaseDescriptor; @@ -466,6 +467,12 @@ public static List toFlussTableChanges( // MaterializedTableChange may produce multiple fluss TableChange. return convertMaterializedTableChange(tableChange); } else { + // Some table changes (e.g. ModifyDefinitionQuery) only exist in newer Flink versions + // and are converted by a version-specific adapter. + Optional> adapted = TableChangeAdapter.convert(tableChange); + if (adapted.isPresent()) { + return adapted.get(); + } throw new UnsupportedOperationException( String.format("Unsupported flink table change: %s.", tableChange)); }