raminqaf commented on code in PR #29659: URL: https://github.com/apache/flink/pull/29659#discussion_r4453890796
########## flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/delegation/materializedtable/RefreshJobResult.java: ########## @@ -0,0 +1,64 @@ +/* + * 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.flink.table.delegation.materializedtable; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.core.execution.JobClient; + +import javax.annotation.Nullable; + +import java.util.Optional; + +import static org.apache.flink.util.Preconditions.checkArgument; +import static org.apache.flink.util.Preconditions.checkNotNull; + +/** + * The result of submitting a materialized table refresh job, describing where the job was deployed Review Comment: " describing where the job was deployed" is being done with the `RefreshJobTarget`? " how to identify it" How can I identify it? ########## flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/materializedtable/MaterializedTableContext.java: ########## @@ -0,0 +1,85 @@ +/* + * 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.flink.table.gateway.service.materializedtable; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.table.factories.WorkflowSchedulerFactoryUtil; +import org.apache.flink.table.gateway.rest.SqlGatewayRestEndpointFactory; +import org.apache.flink.table.gateway.rest.util.SqlGatewayRestOptions; +import org.apache.flink.table.refresh.RefreshHandler; +import org.apache.flink.table.workflow.WorkflowScheduler; + +import javax.annotation.Nullable; + +import static org.apache.flink.table.gateway.api.endpoint.SqlGatewayEndpointFactoryUtils.getEndpointConfig; + +/** + * Session-scoped holder for the SQL Gateway capabilities a {@link + * GatewayMaterializedTableJobSubmitter} needs to run materialized table jobs. It owns the lifecycle + * of the workflow scheduler and is created once per session in {@code SessionContext}. + */ +@Internal +public class MaterializedTableContext { Review Comment: Can this implement `AutoCloseable`? ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ExplainTest.java: ########## @@ -121,17 +122,16 @@ void testExplainAlterMaterializedTable() { @Test void testExplainFullAlterMaterializedTable() { - util.getTableEnv() - .executeSql( - "CREATE OR ALTER MATERIALIZED TABLE MyMTTable\n" - + " WITH (\n" - + " 'connector' = 'values'\n" - + ") AS\n" - + " SELECT\n" - + " `a`,\n" - + " `b`\n" - + " FROM\n" - + " MyTable"); + createMaterializedTableInCatalogOnly( Review Comment: Do we need the "Only" suffix here? ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/delegation/materializedtable/TestingJobClient.java: ########## @@ -0,0 +1,108 @@ +/* + * 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.flink.table.planner.delegation.materializedtable; + +import org.apache.flink.api.common.JobExecutionResult; +import org.apache.flink.api.common.JobID; +import org.apache.flink.api.common.JobStatus; +import org.apache.flink.core.execution.JobClient; +import org.apache.flink.core.execution.SavepointFormatType; + +import javax.annotation.Nullable; + +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.function.Function; +import java.util.function.Supplier; + +/** A {@link JobClient} with a controllable status. */ +class TestingJobClient implements JobClient { Review Comment: nit: How about this? ```suggestion class JobClientMock implements JobClient { ``` ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/MaterializedTableTableApiITCase.java: ########## @@ -0,0 +1,851 @@ +/* + * 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.flink.table.planner.runtime.stream.sql; + +import org.apache.flink.api.common.JobID; +import org.apache.flink.api.common.JobStatus; +import org.apache.flink.client.program.rest.RestClusterClient; +import org.apache.flink.configuration.CheckpointingOptions; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.configuration.ConfigOptions; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.DeploymentOptions; +import org.apache.flink.configuration.RestOptions; +import org.apache.flink.core.testutils.CommonTestUtils; +import org.apache.flink.runtime.client.JobStatusMessage; +import org.apache.flink.runtime.jobgraph.JobType; +import org.apache.flink.runtime.rest.messages.EmptyRequestBody; +import org.apache.flink.runtime.rest.messages.JobMessageParameters; +import org.apache.flink.runtime.rest.messages.MessageHeaders; +import org.apache.flink.runtime.rest.messages.ResponseBody; +import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointConfigHeaders; +import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointConfigInfo; +import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointingStatistics.RestoredCheckpointStatistics; +import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointingStatisticsHeaders; +import org.apache.flink.runtime.rest.util.RestMapperUtils; +import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.streaming.util.TestStreamEnvironment; +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.api.EnvironmentSettings; +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.table.api.TableException; +import org.apache.flink.table.api.ValidationException; +import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; +import org.apache.flink.table.api.internal.TableEnvironmentInternal; +import org.apache.flink.table.catalog.CatalogBaseTable.TableKind; +import org.apache.flink.table.catalog.CatalogMaterializedTable.LogicalRefreshMode; +import org.apache.flink.table.catalog.CatalogMaterializedTable.RefreshMode; +import org.apache.flink.table.catalog.CatalogMaterializedTable.RefreshStatus; +import org.apache.flink.table.catalog.Column; +import org.apache.flink.table.catalog.GenericInMemoryCatalog; +import org.apache.flink.table.catalog.IntervalFreshness; +import org.apache.flink.table.catalog.ObjectIdentifier; +import org.apache.flink.table.catalog.ResolvedCatalogMaterializedTable; +import org.apache.flink.table.catalog.ResolvedSchema; +import org.apache.flink.table.planner.factories.TestValuesTableFactory; +import org.apache.flink.table.refresh.ContinuousRefreshHandler; +import org.apache.flink.table.refresh.ContinuousRefreshHandlerSerializer; +import org.apache.flink.test.junit5.InjectClusterClient; +import org.apache.flink.test.junit5.MiniClusterExtension; +import org.apache.flink.types.Row; +import org.apache.flink.util.CollectionUtil; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; + +import javax.annotation.Nullable; + +import java.nio.file.Path; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +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 related statement via {@link TableEnvironment}. */ +class MaterializedTableTableApiITCase { + + private static final String CATALOG = "test_catalog"; + private static final String DATABASE = "test_db"; + private static final ObjectIdentifier USERS_SHOPS = + ObjectIdentifier.of(CATALOG, DATABASE, "users_shops"); + + private static final Duration TIMEOUT = Duration.ofSeconds(60); + private static final Duration PAUSE = Duration.ofMillis(200); + + private static final ConfigOption<String> KUBERNETES_CLUSTER_ID = + ConfigOptions.key("kubernetes.cluster-id").stringType().noDefaultValue(); + + private static final String UPDATING_VALUES_SINK = + " WITH(\n" + + " 'connector' = 'values',\n" + + " 'sink-insert-only' = 'false'\n" + + " )\n"; + + private static final String USERS_SHOPS_QUERY = + " AS SELECT \n" + + " user_id,\n" + + " shop_id,\n" + + " ds,\n" + + " SUM (payment_amount_cents) AS payed_buy_fee_sum,\n" + + " SUM (1) AS pv\n" + + " FROM (\n" + + " SELECT user_id, shop_id, DATE_FORMAT(order_created_at, 'yyyy-MM-dd') AS ds, payment_amount_cents FROM datagenSource" + + " ) AS tmp\n" + + " GROUP BY (user_id, shop_id, ds)"; + + private static final String USERS_SHOPS_DDL = + "CREATE MATERIALIZED TABLE users_shops" + + " PARTITIONED BY (ds)\n" + + UPDATING_VALUES_SINK + + " FRESHNESS = INTERVAL '30' SECOND\n" + + USERS_SHOPS_QUERY; + + private static final ResolvedSchema USERS_SHOPS_SCHEMA = + ResolvedSchema.of( + Arrays.asList( + Column.physical("user_id", DataTypes.BIGINT()), + Column.physical("shop_id", DataTypes.BIGINT()), + Column.physical("ds", DataTypes.STRING()), + Column.physical("payed_buy_fee_sum", DataTypes.BIGINT()), + Column.physical("pv", DataTypes.INT().notNull()))); + + @RegisterExtension + private static final MiniClusterExtension MINI_CLUSTER = + new MiniClusterExtension( + new MiniClusterResourceConfiguration.Builder() + .setNumberTaskManagers(2) + .setNumberSlotsPerTaskManager(4) + .build()); + + private StreamTableEnvironment tEnv; + private GenericInMemoryCatalog catalog; + private RestClusterClient<?> clusterClient; + + @BeforeEach + void before(@InjectClusterClient RestClusterClient<?> injectedClusterClient) { + TestStreamEnvironment.unsetAsContext(); + clusterClient = injectedClusterClient; + + final StreamExecutionEnvironment executionEnvironment = + StreamExecutionEnvironment.getExecutionEnvironment( + new Configuration(MINI_CLUSTER.getClientConfiguration())); + executionEnvironment.setParallelism(1); + tEnv = StreamTableEnvironment.create(executionEnvironment); + tEnv.getConfig().set(DeploymentOptions.TARGET, "remote"); + tEnv.getConfig() + .set( + RestOptions.ADDRESS, + MINI_CLUSTER.getClientConfiguration().get(RestOptions.ADDRESS)); + + catalog = new GenericInMemoryCatalog(CATALOG, DATABASE); + tEnv.registerCatalog(CATALOG, catalog); + tEnv.useCatalog(CATALOG); + tEnv.useDatabase(DATABASE); + + tEnv.executeSql( + "CREATE TABLE datagenSource (\n" + + " order_id BIGINT,\n" + + " order_number VARCHAR(20),\n" + + " user_id BIGINT,\n" + + " shop_id BIGINT,\n" + + " product_id BIGINT,\n" + + " status BIGINT,\n" + + " order_type BIGINT,\n" + + " order_created_at TIMESTAMP,\n" + + " payment_amount_cents BIGINT\n" + + ")\n" + + "WITH (\n" + + " 'connector' = 'datagen',\n" + + " 'rows-per-second' = '10'\n" + + ")"); + } + + @AfterEach + void after() throws Exception { + for (JobStatusMessage job : + clusterClient.listJobs().get(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS)) { + if (!job.getJobState().isTerminalState()) { + cancelUnlessEnded(job.getJobId()); + } + } + TestValuesTableFactory.clearAllData(); + } + + private void cancelUnlessEnded(JobID jobId) throws Exception { + try { + clusterClient.cancel(jobId).get(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS); + } catch (Exception cancelFailure) { + if (!getJobStatus(jobId).isGloballyTerminalState()) { + throw cancelFailure; + } + } + } + + @Test + void testCreateMaterializedTableInContinuousMode() throws Exception { Review Comment: Should we break this test down into `@nested` test classes? One for create, one for alter, etc. ########## flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/delegation/materializedtable/RefreshJobTarget.java: ########## @@ -0,0 +1,196 @@ +/* + * 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.flink.table.delegation.materializedtable; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.configuration.ConfigOptions; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.DeploymentOptions; +import org.apache.flink.configuration.ReadableConfig; +import org.apache.flink.table.api.ValidationException; +import org.apache.flink.table.catalog.ObjectIdentifier; +import org.apache.flink.table.refresh.ContinuousRefreshHandler; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.Optional; + +import static org.apache.flink.util.Preconditions.checkArgument; +import static org.apache.flink.util.Preconditions.checkNotNull; + +/** The execution target and cluster id a materialized table refresh job is submitted to. */ +@Internal +public final class RefreshJobTarget { Review Comment: We can add some cheap parameterized test for this class and keep one ITCase. WDYT? ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/delegation/materializedtable/RefreshJobControllerTest.java: ########## @@ -0,0 +1,159 @@ +/* + * 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.flink.table.planner.delegation.materializedtable; + +import org.apache.flink.api.common.JobID; +import org.apache.flink.api.common.JobStatus; +import org.apache.flink.api.dag.Pipeline; +import org.apache.flink.client.deployment.StandaloneClusterId; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.core.execution.JobClient; +import org.apache.flink.core.execution.JobRetrievalSupportedPipelineExecutor; +import org.apache.flink.table.api.TableConfig; +import org.apache.flink.table.api.TableException; +import org.apache.flink.table.refresh.ContinuousRefreshHandler; + +import org.junit.jupiter.api.Test; + +import java.util.Optional; +import java.util.concurrent.CompletableFuture; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for {@link RefreshJobController}. */ +class RefreshJobControllerTest { Review Comment: Is this possible to add some tests for the TimeoutException handling you do? ########## docs/content/docs/sql/materialized-table/deployment.md: ########## @@ -49,7 +49,7 @@ Materialized Table refresh jobs currently support execution in these cluster env ## Flink SQL Gateway Deployment -Materialized Tables must be created through SQL Gateway, which requires specific configurations for metadata persistence and job scheduling. +Materialized Tables in full refresh mode must be created and operated through SQL Gateway, which schedules their refresh workflows. SQL Gateway requires specific configurations for metadata persistence and job scheduling. Materialized Tables in continuous refresh mode can also be created from any `TableEnvironment` (see [Using Table API](#using-table-api)). Review Comment: We need to add this to the Chinese docs too. Can you sync them together? ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/MaterializedTableTableApiITCase.java: ########## @@ -0,0 +1,851 @@ +/* + * 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.flink.table.planner.runtime.stream.sql; + +import org.apache.flink.api.common.JobID; +import org.apache.flink.api.common.JobStatus; +import org.apache.flink.client.program.rest.RestClusterClient; +import org.apache.flink.configuration.CheckpointingOptions; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.configuration.ConfigOptions; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.DeploymentOptions; +import org.apache.flink.configuration.RestOptions; +import org.apache.flink.core.testutils.CommonTestUtils; +import org.apache.flink.runtime.client.JobStatusMessage; +import org.apache.flink.runtime.jobgraph.JobType; +import org.apache.flink.runtime.rest.messages.EmptyRequestBody; +import org.apache.flink.runtime.rest.messages.JobMessageParameters; +import org.apache.flink.runtime.rest.messages.MessageHeaders; +import org.apache.flink.runtime.rest.messages.ResponseBody; +import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointConfigHeaders; +import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointConfigInfo; +import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointingStatistics.RestoredCheckpointStatistics; +import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointingStatisticsHeaders; +import org.apache.flink.runtime.rest.util.RestMapperUtils; +import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.streaming.util.TestStreamEnvironment; +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.api.EnvironmentSettings; +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.table.api.TableException; +import org.apache.flink.table.api.ValidationException; +import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; +import org.apache.flink.table.api.internal.TableEnvironmentInternal; +import org.apache.flink.table.catalog.CatalogBaseTable.TableKind; +import org.apache.flink.table.catalog.CatalogMaterializedTable.LogicalRefreshMode; +import org.apache.flink.table.catalog.CatalogMaterializedTable.RefreshMode; +import org.apache.flink.table.catalog.CatalogMaterializedTable.RefreshStatus; +import org.apache.flink.table.catalog.Column; +import org.apache.flink.table.catalog.GenericInMemoryCatalog; +import org.apache.flink.table.catalog.IntervalFreshness; +import org.apache.flink.table.catalog.ObjectIdentifier; +import org.apache.flink.table.catalog.ResolvedCatalogMaterializedTable; +import org.apache.flink.table.catalog.ResolvedSchema; +import org.apache.flink.table.planner.factories.TestValuesTableFactory; +import org.apache.flink.table.refresh.ContinuousRefreshHandler; +import org.apache.flink.table.refresh.ContinuousRefreshHandlerSerializer; +import org.apache.flink.test.junit5.InjectClusterClient; +import org.apache.flink.test.junit5.MiniClusterExtension; +import org.apache.flink.types.Row; +import org.apache.flink.util.CollectionUtil; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; + +import javax.annotation.Nullable; + +import java.nio.file.Path; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +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 related statement via {@link TableEnvironment}. */ +class MaterializedTableTableApiITCase { + + private static final String CATALOG = "test_catalog"; + private static final String DATABASE = "test_db"; + private static final ObjectIdentifier USERS_SHOPS = + ObjectIdentifier.of(CATALOG, DATABASE, "users_shops"); + + private static final Duration TIMEOUT = Duration.ofSeconds(60); + private static final Duration PAUSE = Duration.ofMillis(200); + + private static final ConfigOption<String> KUBERNETES_CLUSTER_ID = + ConfigOptions.key("kubernetes.cluster-id").stringType().noDefaultValue(); + + private static final String UPDATING_VALUES_SINK = + " WITH(\n" + + " 'connector' = 'values',\n" + + " 'sink-insert-only' = 'false'\n" + + " )\n"; + + private static final String USERS_SHOPS_QUERY = + " AS SELECT \n" + + " user_id,\n" + + " shop_id,\n" + + " ds,\n" + + " SUM (payment_amount_cents) AS payed_buy_fee_sum,\n" + + " SUM (1) AS pv\n" + + " FROM (\n" + + " SELECT user_id, shop_id, DATE_FORMAT(order_created_at, 'yyyy-MM-dd') AS ds, payment_amount_cents FROM datagenSource" + + " ) AS tmp\n" + + " GROUP BY (user_id, shop_id, ds)"; + + private static final String USERS_SHOPS_DDL = + "CREATE MATERIALIZED TABLE users_shops" + + " PARTITIONED BY (ds)\n" + + UPDATING_VALUES_SINK + + " FRESHNESS = INTERVAL '30' SECOND\n" + + USERS_SHOPS_QUERY; + + private static final ResolvedSchema USERS_SHOPS_SCHEMA = + ResolvedSchema.of( + Arrays.asList( + Column.physical("user_id", DataTypes.BIGINT()), + Column.physical("shop_id", DataTypes.BIGINT()), + Column.physical("ds", DataTypes.STRING()), + Column.physical("payed_buy_fee_sum", DataTypes.BIGINT()), + Column.physical("pv", DataTypes.INT().notNull()))); + + @RegisterExtension + private static final MiniClusterExtension MINI_CLUSTER = + new MiniClusterExtension( + new MiniClusterResourceConfiguration.Builder() + .setNumberTaskManagers(2) + .setNumberSlotsPerTaskManager(4) + .build()); + + private StreamTableEnvironment tEnv; + private GenericInMemoryCatalog catalog; + private RestClusterClient<?> clusterClient; + + @BeforeEach + void before(@InjectClusterClient RestClusterClient<?> injectedClusterClient) { + TestStreamEnvironment.unsetAsContext(); + clusterClient = injectedClusterClient; + + final StreamExecutionEnvironment executionEnvironment = + StreamExecutionEnvironment.getExecutionEnvironment( + new Configuration(MINI_CLUSTER.getClientConfiguration())); + executionEnvironment.setParallelism(1); + tEnv = StreamTableEnvironment.create(executionEnvironment); + tEnv.getConfig().set(DeploymentOptions.TARGET, "remote"); + tEnv.getConfig() + .set( + RestOptions.ADDRESS, + MINI_CLUSTER.getClientConfiguration().get(RestOptions.ADDRESS)); + + catalog = new GenericInMemoryCatalog(CATALOG, DATABASE); + tEnv.registerCatalog(CATALOG, catalog); + tEnv.useCatalog(CATALOG); + tEnv.useDatabase(DATABASE); + + tEnv.executeSql( + "CREATE TABLE datagenSource (\n" + + " order_id BIGINT,\n" + + " order_number VARCHAR(20),\n" + + " user_id BIGINT,\n" + + " shop_id BIGINT,\n" + + " product_id BIGINT,\n" + + " status BIGINT,\n" + + " order_type BIGINT,\n" + + " order_created_at TIMESTAMP,\n" + + " payment_amount_cents BIGINT\n" + + ")\n" + + "WITH (\n" + + " 'connector' = 'datagen',\n" + + " 'rows-per-second' = '10'\n" + + ")"); + } + + @AfterEach + void after() throws Exception { + for (JobStatusMessage job : + clusterClient.listJobs().get(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS)) { + if (!job.getJobState().isTerminalState()) { + cancelUnlessEnded(job.getJobId()); + } + } + TestValuesTableFactory.clearAllData(); + } + + private void cancelUnlessEnded(JobID jobId) throws Exception { + try { + clusterClient.cancel(jobId).get(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS); + } catch (Exception cancelFailure) { + if (!getJobStatus(jobId).isGloballyTerminalState()) { + throw cancelFailure; + } + } + } + + @Test + void testCreateMaterializedTableInContinuousMode() throws Exception { + tEnv.executeSql(USERS_SHOPS_DDL); + + final ResolvedCatalogMaterializedTable actualMaterializedTable = getUsersShops(); + assertThat(actualMaterializedTable.getResolvedSchema()).isEqualTo(USERS_SHOPS_SCHEMA); + assertThat(actualMaterializedTable.getLogicalRefreshMode()) + .isSameAs(LogicalRefreshMode.AUTOMATIC); + assertThat(actualMaterializedTable.getRefreshMode()).isSameAs(RefreshMode.CONTINUOUS); + assertThat(actualMaterializedTable.getRefreshStatus()).isSameAs(RefreshStatus.ACTIVATED); + assertThat(actualMaterializedTable.getRefreshHandlerDescription()).isNotEmpty(); + assertThat(actualMaterializedTable.getSerializedRefreshHandler()).isNotEmpty(); + + final JobID refreshJobId = getRefreshJobId(); + waitUntilAllTasksAreRunning(clusterClient, refreshJobId); + assertThat(getJobStatus(refreshJobId)).isEqualTo(JobStatus.RUNNING); + assertThat(getCheckpointInterval(refreshJobId)).isEqualTo(30 * 1000); + } + + @Test + void testCreateMaterializedTableInContinuousModeWithoutFreshnessAndRefreshMode() + throws Exception { + tEnv.executeSql( + "CREATE MATERIALIZED TABLE users_shops" + + " PARTITIONED BY (ds)\n" + + UPDATING_VALUES_SINK + + USERS_SHOPS_QUERY); + + final ResolvedCatalogMaterializedTable actualMaterializedTable = getUsersShops(); + assertThat(actualMaterializedTable.getResolvedSchema()).isEqualTo(USERS_SHOPS_SCHEMA); + assertThat(actualMaterializedTable.getDefinitionFreshness()) + .isEqualTo(IntervalFreshness.ofMinute(3)); + assertThat(actualMaterializedTable.getLogicalRefreshMode()) + .isSameAs(LogicalRefreshMode.AUTOMATIC); + assertThat(actualMaterializedTable.getRefreshMode()).isSameAs(RefreshMode.CONTINUOUS); + assertThat(actualMaterializedTable.getRefreshStatus()).isSameAs(RefreshStatus.ACTIVATED); + assertThat(actualMaterializedTable.getRefreshHandlerDescription()).isNotEmpty(); + assertThat(actualMaterializedTable.getSerializedRefreshHandler()).isNotEmpty(); + + final JobID refreshJobId = getRefreshJobId(); + waitUntilAllTasksAreRunning(clusterClient, refreshJobId); + assertThat(getJobStatus(refreshJobId)).isEqualTo(JobStatus.RUNNING); + // default checkpoint interval is 3 minutes + assertThat(getCheckpointInterval(refreshJobId)).isEqualTo(60 * 3 * 1000); + } + + @Test + void testCreateMaterializedTableInContinuousModeWithCustomCheckpointInterval() + throws Exception { + final long checkpointInterval = 60 * 1000; + tEnv.getConfig() + .set( + CheckpointingOptions.CHECKPOINTING_INTERVAL, + Duration.ofMillis(checkpointInterval)); + + tEnv.executeSql(USERS_SHOPS_DDL); + + final JobID refreshJobId = getRefreshJobId(); + waitUntilAllTasksAreRunning(clusterClient, refreshJobId); + assertThat(getCheckpointInterval(refreshJobId)).isEqualTo(checkpointInterval); + } + + @Test + void testCreateMaterializedTableFailedInContinuousMode() { + // an insert-only sink cannot consume the updates of the aggregation + assertThatThrownBy( + () -> + tEnv.executeSql( + "CREATE MATERIALIZED TABLE users_shops" + + " PARTITIONED BY (ds)\n" + + " WITH(\n" + + " 'connector' = 'values',\n" + + " 'sink-insert-only' = 'true'\n" + + " )\n" + + " FRESHNESS = INTERVAL '30' SECOND\n" + + USERS_SHOPS_QUERY)) + .isInstanceOf(TableException.class) + .hasMessageContaining( + String.format( + "Failed to submit continuous refresh job for materialized table %s.", + USERS_SHOPS.asSerializableString())); + + assertThat(catalog.tableExists(USERS_SHOPS.toObjectPath())).isFalse(); + } + + @Test + void testAlterMaterializedTableRefresh() throws Exception { + final List<Row> data = new ArrayList<>(); + data.add(Row.of(1L, 1L, 1L, "2024-01-01")); + data.add(Row.of(2L, 2L, 2L, "2024-01-02")); + data.add(Row.of(3L, 3L, 3L, "2024-01-02")); + createBoundedValuesSource(data); + + tEnv.executeSql( + "CREATE MATERIALIZED TABLE users_shops" + + " PARTITIONED BY (ds)\n" + + UPDATING_VALUES_SINK + + " FRESHNESS = INTERVAL '30' SECOND\n" + + " REFRESH_MODE = CONTINUOUS\n" + + " AS SELECT \n" + + " user_id,\n" + + " shop_id,\n" + + " ds,\n" + + " COUNT(order_id) AS order_cnt\n" + + " FROM (\n" + + " SELECT user_id, shop_id, order_created_at AS ds, order_id FROM my_source" + + " ) AS tmp\n" + + " GROUP BY (user_id, shop_id, ds)"); + CommonTestUtils.waitUtil( + () -> getUsersShopsResults().size() == data.size(), + TIMEOUT, + PAUSE, + "Failed to verify the data in materialized table."); + + data.remove(2); + + final long currentTime = System.currentTimeMillis(); + final List<Row> result = + CollectionUtil.iteratorToList( + tEnv.executeSql( + "ALTER MATERIALIZED TABLE users_shops REFRESH PARTITION (ds = '2024-01-02')") + .collect()); + assertThat(result).hasSize(1); + final JobID refreshJobId = JobID.fromHexString(result.get(0).getFieldAs(0)); + + verifyRefreshJobCreated(refreshJobId, currentTime); + + assertThat(getUsersShopsResults()).containsExactly("+I[2, 2, 2024-01-02, 1]"); + } + + @Test + void testAlterMaterializedTableRefreshWithInvalidPartitionSpec() { + tEnv.executeSql( + "CREATE MATERIALIZED TABLE users_shops" + + " PARTITIONED BY (ds1, ds2)\n" + + UPDATING_VALUES_SINK + + " FRESHNESS = INTERVAL '30' SECOND\n" + + " AS SELECT \n" + + " user_id,\n" + + " shop_id,\n" + + " ds1,\n" + + " ds2,\n" + + " SUM (payment_amount_cents) AS payed_buy_fee_sum,\n" + + " SUM (1) AS pv\n" + + " FROM (\n" + + " SELECT user_id, shop_id, DATE_FORMAT(order_created_at, 'yyyy-MM-dd') AS ds1, user_id % 10 as ds2, payment_amount_cents FROM datagenSource" + + " ) AS tmp\n" + + " GROUP BY (user_id, shop_id, ds1, ds2)"); + + // CASE 1: check unknown partition keys + assertThatThrownBy( + () -> + tEnv.executeSql( + "ALTER MATERIALIZED TABLE users_shops REFRESH PARTITION (ds3 = '2024-01-01')")) + .isInstanceOf(ValidationException.class) + .hasMessage( + "The partition spec contains unknown partition keys:\n" + + "\n" + + "ds3\n" + + "\n" + + "All known partition keys are:\n" + + "\n" + + "ds2\n" + + "ds1"); + + // CASE 2: check specific non-string partition keys as partition spec to refresh + assertThatThrownBy( + () -> + tEnv.executeSql( + "ALTER MATERIALIZED TABLE users_shops REFRESH PARTITION (ds2 = 5)")) + .isInstanceOf(ValidationException.class) + .hasMessage( + "Currently, refreshing materialized table only supports referring to char, varchar and string type partition keys. All specified partition keys in partition specs with unsupported types are:\n" + + "\n" + + "ds2"); + } + + @Test + void testAlterMaterializedTableSuspendAndResumeInContinuousMode(@TempDir Path temporaryPath) + throws Exception { + tEnv.executeSql(USERS_SHOPS_DDL); + + assertThat(getUsersShops().getRefreshStatus()).isSameAs(RefreshStatus.ACTIVATED); + waitUntilAllTasksAreRunning(clusterClient, getRefreshJobId()); + + setupSavepointDir(temporaryPath); + + tEnv.executeSql("ALTER MATERIALIZED TABLE users_shops SUSPEND"); + + assertThat(getUsersShops().getRefreshStatus()).isSameAs(RefreshStatus.SUSPENDED); + final ContinuousRefreshHandler suspendRefreshHandler = getRefreshHandler(); + assertThat(getJobStatus(JobID.fromHexString(suspendRefreshHandler.getJobId()))) + .isEqualTo(JobStatus.FINISHED); + assertThat(suspendRefreshHandler.getRestorePath()).isNotEmpty(); + final String actualSavepointPath = suspendRefreshHandler.getRestorePath().get(); + + tEnv.executeSql( + "ALTER MATERIALIZED TABLE users_shops RESUME WITH ('sink.drop-late-event' = 'false')"); + + final ResolvedCatalogMaterializedTable resumedMaterializedTable = getUsersShops(); + assertThat(resumedMaterializedTable.getOptions()).doesNotContainKey("sink.drop-late-event"); + assertThat(resumedMaterializedTable.getRefreshStatus()).isSameAs(RefreshStatus.ACTIVATED); + + final JobID resumeJobId = getRefreshJobId(); + waitUntilAllTasksAreRunning(clusterClient, resumeJobId); + assertThat(getJobStatus(resumeJobId)).isEqualTo(JobStatus.RUNNING); + assertThat(getJobRestoreSavepointPath(resumeJobId)).hasValue(actualSavepointPath); + } + + @Test + void testAlterMaterializedTableWithoutSavepointDirConfiguredInContinuousMode() + throws Exception { + tEnv.executeSql(USERS_SHOPS_DDL); + waitUntilAllTasksAreRunning(clusterClient, getRefreshJobId()); + + assertThatThrownBy(() -> tEnv.executeSql("ALTER MATERIALIZED TABLE users_shops SUSPEND")) + .rootCause() + .isInstanceOf(ValidationException.class) + .hasMessageContaining( + "Savepoint directory is not configured, can't stop job with savepoint."); + } + + @Test + void testAlterMaterializedTableWithRepeatedSuspendAndResumeInContinuousMode( + @TempDir Path temporaryPath) throws Exception { + tEnv.executeSql(USERS_SHOPS_DDL); + waitUntilAllTasksAreRunning(clusterClient, getRefreshJobId()); + + setupSavepointDir(temporaryPath); + + tEnv.executeSql("ALTER MATERIALIZED TABLE users_shops SUSPEND"); + + assertThatThrownBy(() -> tEnv.executeSql("ALTER MATERIALIZED TABLE users_shops SUSPEND")) + .rootCause() + .isInstanceOf(TableException.class) + .hasMessageContaining( + String.format( + "Materialized table %s continuous refresh job has been suspended", + USERS_SHOPS)); + + tEnv.executeSql("ALTER MATERIALIZED TABLE users_shops RESUME"); + + assertThatThrownBy(() -> tEnv.executeSql("ALTER MATERIALIZED TABLE users_shops RESUME")) + .isInstanceOf(TableException.class) + .hasMessageContaining( + String.format( + "Materialized table %s continuous refresh job has been resumed", + USERS_SHOPS)); + } + + @Test + void testCreateOrAlterMaterializedTableColumnListReordersSchemaOnCreatePath() { + createBoundedValuesSource(List.of()); + + tEnv.executeSql( + "CREATE OR ALTER MATERIALIZED TABLE users_shops (shop_id, user_id, oca, order_id)" + + UPDATING_VALUES_SINK + + " AS SELECT \n" + + " user_id,\n" + + " shop_id,\n" + + " order_created_at AS oca,\n" + + " order_id" + + " FROM my_source"); + + assertThat(getUsersShops().getResolvedSchema().getColumnNames()) + .containsExactly("shop_id", "user_id", "oca", "order_id"); + } + + @Test + void testCreateMaterializedTableWithDistribution() { + assertThatThrownBy( + () -> + tEnv.executeSql( + "CREATE MATERIALIZED TABLE users_shops (" + + " PRIMARY KEY (shop_id, user_id) not enforced)" + + " DISTRIBUTED BY HASH (user_id) INTO 7 BUCKETS\n" + + " WITH(\n" + + " 'connector' = 'blackhole'\n" + + " )\n" + + " FRESHNESS = INTERVAL '30' SECOND\n" + + " AS SELECT 1 as shop_id, 2 as user_id")) + .isInstanceOf(TableException.class) + .hasRootCauseMessage( + String.format( + "Table '%s' is distributed into buckets," + + " but the underlying DynamicTableSink doesn't implement the SupportsBucketing interface.", + USERS_SHOPS.asSummaryString())); + } + + @Test + void testAlterMaterializedTableAsQueryInContinuousMode(@TempDir Path temporaryPath) + throws Exception { + tEnv.executeSql( + "CREATE MATERIALIZED TABLE users_shops (" + + " PRIMARY KEY (ds, user_id) not enforced)" + + " PARTITIONED BY (ds)\n" + + " WITH(\n" + + " 'connector' = 'values',\n" + + " 'sink-insert-only' = 'false',\n" + + " 'sink-changelog-mode-enforced' = 'I,UB,UA,D'\n" + + " )\n" + + " FRESHNESS = INTERVAL '30' SECOND\n" + + " AS SELECT \n" + + " coalesce(user_id, 0) as user_id,\n" + + " shop_id,\n" + + " coalesce(ds, '') as ds,\n" + + " SUM (payment_amount_cents) AS payed_buy_fee_sum\n" + + " FROM (\n" + + " SELECT user_id, shop_id, DATE_FORMAT(order_created_at, 'yyyy-MM-dd') AS ds, payment_amount_cents FROM datagenSource" + + " ) AS tmp\n" + + " GROUP BY (user_id, shop_id, ds)"); + final ResolvedCatalogMaterializedTable oldTable = getUsersShops(); + waitUntilAllTasksAreRunning(clusterClient, getRefreshJobId()); + + setupSavepointDir(temporaryPath); + + tEnv.executeSql( + "ALTER MATERIALIZED TABLE users_shops" + + " AS SELECT \n" + + " coalesce(user_id, 0) as user_id,\n" + + " shop_id,\n" + + " coalesce(ds, '') as ds,\n" + + " SUM (payment_amount_cents) AS payed_buy_fee_sum,\n" + + " SUM (1) AS pv\n" + + " FROM (\n" + + " SELECT user_id, shop_id, DATE_FORMAT(order_created_at, 'yyyy-MM-dd') AS ds, payment_amount_cents FROM datagenSource" + + " ) AS tmp\n" + + " GROUP BY (user_id, shop_id, ds)"); + final ResolvedCatalogMaterializedTable newTable = getUsersShops(); + + assertThat(getAddedColumns(newTable.getResolvedSchema(), oldTable.getResolvedSchema())) + .containsExactly(Column.physical("pv", DataTypes.INT())); + assertThat(newTable.getResolvedSchema().getPrimaryKey()) + .isEqualTo(oldTable.getResolvedSchema().getPrimaryKey()); + assertThat(newTable.getResolvedSchema().getWatermarkSpecs()) + .isEqualTo(oldTable.getResolvedSchema().getWatermarkSpecs()); + assertThat(newTable.getExpandedQuery()) + .hasToString( + String.format( + "SELECT COALESCE(`tmp`.`user_id`, 0) AS `user_id`, `tmp`.`shop_id`, COALESCE(`tmp`.`ds`, '') AS `ds`, SUM(`tmp`.`payment_amount_cents`) AS `payed_buy_fee_sum`, SUM(1) AS `pv`\n" + + "FROM (SELECT `datagenSource`.`user_id`, `datagenSource`.`shop_id`, DATE_FORMAT(`datagenSource`.`order_created_at`, 'yyyy-MM-dd') AS `ds`, `datagenSource`.`payment_amount_cents`\n" + + "FROM `%s`.`%s`.`datagenSource` AS `datagenSource`) AS `tmp`\n" + + "GROUP BY ROW(`tmp`.`user_id`, `tmp`.`shop_id`, `tmp`.`ds`)", + CATALOG, DATABASE)); + assertThat(oldTable.getSerializedRefreshHandler()) + .isNotEqualTo(newTable.getSerializedRefreshHandler()); + assertThat(getJobRestoreSavepointPath(getRefreshJobId())).isEmpty(); + } + + @Test + void testAlterMaterializedTableAsQueryInContinuousModeWithSuspendStatus( + @TempDir Path temporaryPath) throws Exception { + tEnv.executeSql( + "CREATE MATERIALIZED TABLE users_shops" + + " PARTITIONED BY (ds)\n" + + UPDATING_VALUES_SINK + + " FRESHNESS = INTERVAL '30' SECOND\n" + + " AS SELECT \n" + + " user_id,\n" + + " shop_id,\n" + + " ds,\n" + + " SUM (payment_amount_cents) AS payed_buy_fee_sum\n" + + " FROM (\n" + + " SELECT user_id, shop_id, DATE_FORMAT(order_created_at, 'yyyy-MM-dd') AS ds, payment_amount_cents FROM datagenSource" + + " ) AS tmp\n" + + " GROUP BY (user_id, shop_id, ds)"); + final ResolvedCatalogMaterializedTable oldTable = getUsersShops(); + waitUntilAllTasksAreRunning(clusterClient, getRefreshJobId()); + + setupSavepointDir(temporaryPath); + + tEnv.executeSql("ALTER MATERIALIZED TABLE users_shops SUSPEND"); + + tEnv.executeSql( + "ALTER MATERIALIZED TABLE users_shops" + + " AS SELECT \n" + + " user_id,\n" + + " shop_id,\n" + + " ds,\n" + + " SUM (payment_amount_cents) AS payed_buy_fee_sum,\n" + + " SUM (1) AS pv\n" + + " FROM (\n" + + " SELECT user_id, shop_id, DATE_FORMAT(order_created_at, 'yyyy-MM-dd') AS ds, payment_amount_cents FROM datagenSource" + + " ) AS tmp\n" + + " GROUP BY (user_id, shop_id, ds)"); + final ResolvedCatalogMaterializedTable newTable = getUsersShops(); + + assertThat(getAddedColumns(newTable.getResolvedSchema(), oldTable.getResolvedSchema())) + .containsExactly(Column.physical("pv", DataTypes.INT())); + assertThat(newTable.getExpandedQuery()) + .hasToString( + String.format( + "SELECT `tmp`.`user_id`, `tmp`.`shop_id`, `tmp`.`ds`, SUM(`tmp`.`payment_amount_cents`) AS `payed_buy_fee_sum`, SUM(1) AS `pv`\n" + + "FROM (SELECT `datagenSource`.`user_id`, `datagenSource`.`shop_id`, DATE_FORMAT(`datagenSource`.`order_created_at`, 'yyyy-MM-dd') AS `ds`, `datagenSource`.`payment_amount_cents`\n" + + "FROM `%s`.`%s`.`datagenSource` AS `datagenSource`) AS `tmp`\n" + + "GROUP BY ROW(`tmp`.`user_id`, `tmp`.`shop_id`, `tmp`.`ds`)", + CATALOG, DATABASE)); + assertThat(oldTable.getSerializedRefreshHandler()) + .isEqualTo(newTable.getSerializedRefreshHandler()); + assertThat(oldTable.getDefinitionFreshness()).isEqualTo(newTable.getDefinitionFreshness()); + + // verify the restore path is empty + assertThat(getRefreshHandler().getRestorePath()).isEmpty(); + } + + @Test + void testDropMaterializedTableInContinuousMode() throws Exception { + tEnv.executeSql(USERS_SHOPS_DDL); + + assertThat(catalog.getTable(USERS_SHOPS.toObjectPath()).getTableKind()) + .isEqualTo(TableKind.MATERIALIZED_TABLE); + final JobID refreshJobId = getRefreshJobId(); + waitUntilAllTasksAreRunning(clusterClient, refreshJobId); + assertThat(getJobStatus(refreshJobId)).isEqualTo(JobStatus.RUNNING); + + // Drop materialized table using drop table statement + assertThatThrownBy(() -> tEnv.executeSql("DROP TABLE users_shops")) + .isInstanceOf(ValidationException.class) + .hasMessage( + String.format( + "Table with identifier '%s' does not exist.", + USERS_SHOPS.asSummaryString())); + + tEnv.executeSql("DROP MATERIALIZED TABLE IF EXISTS users_shops"); + + assertThat(catalog.tableExists(USERS_SHOPS.toObjectPath())).isFalse(); + awaitJobStatus(refreshJobId, JobStatus.CANCELED); + + // verify drop materialized table that doesn't exist + assertThatThrownBy(() -> tEnv.executeSql("DROP MATERIALIZED TABLE users_shops")) + .isInstanceOf(ValidationException.class) + .hasMessage( + String.format( + "Materialized table with identifier %s does not exist.", + USERS_SHOPS.asSerializableString())); + + tEnv.executeSql("DROP MATERIALIZED TABLE IF EXISTS users_shops"); + + // Drop a table using drop materialized table statement + assertThatThrownBy(() -> tEnv.executeSql("DROP MATERIALIZED TABLE IF EXISTS datagenSource")) + .isInstanceOf(ValidationException.class) + .hasMessage( + String.format( + "Table %s is not a materialized table, does not support materialized table related operation.", + ObjectIdentifier.of(CATALOG, DATABASE, "datagenSource") + .asSerializableString())); + } + + @ParameterizedTest(name = "{0}") + @MethodSource("unsupportedTargets") + void unsupportedTargetIsRejectedBeforeAnyCatalogChange( Review Comment: This test is very hard to read ########## flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/operation/OperationExecutor.java: ########## @@ -224,6 +227,8 @@ public ResultFetcher configureSession(OperationHandle handle, String statement) return callSetOperation(tableEnv, handle, (SetOperation) op); } else if (op instanceof ResetOperation) { return callResetOperation(handle, (ResetOperation) op); + } else if (op instanceof MaterializedTableOperation) { + return callExecutableOperation(handle, (ExecutableOperation) op); Review Comment: Do we have a test to that covers this branch? ########## docs/content/docs/sql/materialized-table/deployment.md: ########## @@ -114,7 +114,7 @@ For session modes, pre-create session cluster as documented in [yarn-session]({ **Kubernetes session mode:** ```sql -Flink SQL> SET 'execution.mode' = 'kubernetes-session'; +Flink SQL> SET 'execution.target' = 'kubernetes-session'; Review Comment: Was this out dated or is the change part of this PR? ########## flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/delegation/materializedtable/DefaultMaterializedTableExecutor.java: ########## @@ -100,148 +90,118 @@ import static org.apache.flink.api.common.RuntimeExecutionMode.BATCH; import static org.apache.flink.api.common.RuntimeExecutionMode.STREAMING; -import static org.apache.flink.configuration.CheckpointingOptions.SAVEPOINT_DIRECTORY; -import static org.apache.flink.configuration.DeploymentOptions.TARGET; import static org.apache.flink.configuration.ExecutionOptions.RUNTIME_MODE; import static org.apache.flink.configuration.PipelineOptions.NAME; -import static org.apache.flink.configuration.PipelineOptionsInternal.PIPELINE_FIXED_JOB_ID; import static org.apache.flink.configuration.StateRecoveryOptions.SAVEPOINT_PATH; import static org.apache.flink.table.api.config.MaterializedTableConfigOptions.DATE_FORMATTER; import static org.apache.flink.table.api.config.MaterializedTableConfigOptions.PARTITION_FIELDS; import static org.apache.flink.table.api.config.MaterializedTableConfigOptions.SCHEDULE_TIME_DATE_FORMATTER_DEFAULT; import static org.apache.flink.table.api.internal.TableResultInternal.TABLE_RESULT_OK; import static org.apache.flink.table.catalog.CatalogBaseTable.TableKind.MATERIALIZED_TABLE; import static org.apache.flink.table.catalog.IntervalFreshness.convertFreshnessToCron; -import static org.apache.flink.table.factories.WorkflowSchedulerFactoryUtil.WORKFLOW_SCHEDULER_PREFIX; -import static org.apache.flink.table.gateway.api.endpoint.SqlGatewayEndpointFactoryUtils.getEndpointConfig; -import static org.apache.flink.table.gateway.service.utils.Constants.CLUSTER_INFO; -import static org.apache.flink.table.gateway.service.utils.Constants.JOB_ID; import static org.apache.flink.table.utils.DateTimeUtils.formatTimestampStringWithOffset; -/** Manager is responsible for execute the {@link MaterializedTableOperation}. */ +/** Default executor for {@link MaterializedTableOperation}s. */ @Internal -public class MaterializedTableManager { - - private static final Logger LOG = LoggerFactory.getLogger(MaterializedTableManager.class); - - private final URLClassLoader userCodeClassLoader; - - private final @Nullable WorkflowScheduler<? extends RefreshHandler> workflowScheduler; - - private final String restEndpointUrl; - - public MaterializedTableManager( - Configuration configuration, URLClassLoader userCodeClassLoader) { - this.userCodeClassLoader = userCodeClassLoader; - this.restEndpointUrl = buildRestEndpointUrl(configuration); - this.workflowScheduler = buildWorkflowScheduler(configuration, userCodeClassLoader); - } - - private String buildRestEndpointUrl(Configuration configuration) { - Configuration restEndpointConfig = - Configuration.fromMap( - getEndpointConfig(configuration, SqlGatewayRestEndpointFactory.IDENTIFIER)); - String address = restEndpointConfig.get(SqlGatewayRestOptions.ADDRESS); - int port = restEndpointConfig.get(SqlGatewayRestOptions.PORT); - - return String.format("http://%s:%s";, address, port); - } - - private WorkflowScheduler<? extends RefreshHandler> buildWorkflowScheduler( - Configuration configuration, URLClassLoader userCodeClassLoader) { - return WorkflowSchedulerFactoryUtil.createWorkflowScheduler( - configuration, userCodeClassLoader); - } - - public void open() throws Exception { - if (workflowScheduler != null) { - workflowScheduler.open(); - } +public class DefaultMaterializedTableExecutor implements MaterializedTableExecutor { Review Comment: This should be only a simple move of the old MaterializedTableManager, but it adds new behavior to the old logic too. Can the new failure handling go to its own commit? -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
