Timm0 commented on code in PR #29659: URL: https://github.com/apache/flink/pull/29659#discussion_r4456584018
########## 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: I tried it, but `@Nested` classes need the registering of the `MiniClusterExtension` per nested class, but this breaks the `ITCASE_USE_MINICLUSTER` ArchUnit rule. I kept the class flat, with the tests grouped by statement under section comments now to make it easier to add tests here for checking new functionalities. Let me know if this works for. There is also the alternative of accepting the violation as it has been done for `CsvFilesystemBatchITCase`. -- 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]
