You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "raminqaf (via GitHub)" <gi...@apache.org> on 2026/10/16 09:55:11 UTC
Re: [PR] [flink] Support ALTER MATERIALIZED TABLE ... AS in FlinkCatalog [fluss]
raminqaf commented on code in PR #3596:
URL: https://github.com/apache/fluss/pull/3596#discussion_r4293236523
##########
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)
Review Comment:
dropping a column would be another
##########
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);
Review Comment:
Should we do an assertion that the schema hasn't changed? Can be also a helper method call
##########
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)
Review Comment:
Do we support re-ordering of columns? We should also have a test for that case
##########
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
Review Comment:
Isn't appending a nullable column actually changing the "table schema"? The last part of this comment is confusing
```suggestion
// table alteration can only be applied to one of the following: table options
```
##########
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).
+ *
+ * <p>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.
+ *
+ * <p>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<List<org.apache.fluss.metadata.TableChange>> convert(
Review Comment:
Why Optional here? How about a nullable return type?
##########
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)
Review Comment:
How about computed/metadata columns?
##########
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);
Review Comment:
Should we move the creation of the MT to a helper method to reduce the noise?
--
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-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org