leekeiabstraction commented on code in PR #2040: URL: https://github.com/apache/fluss/pull/2040#discussion_r2568356271
########## fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/FlinkArrayTypeITCase.java: ########## @@ -0,0 +1,314 @@ +/* + * 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.sink; + +import org.apache.fluss.client.Connection; +import org.apache.fluss.client.ConnectionFactory; +import org.apache.fluss.client.table.Table; +import org.apache.fluss.client.table.scanner.ScanRecord; +import org.apache.fluss.client.table.scanner.log.LogScanner; +import org.apache.fluss.client.table.scanner.log.ScanRecords; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.metadata.TableBucket; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.row.InternalRow; +import org.apache.fluss.server.testutils.FlussClusterExtension; + +import org.apache.flink.api.common.RuntimeExecutionMode; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.table.api.EnvironmentSettings; +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; +import org.apache.flink.table.api.config.ExecutionConfigOptions; +import org.apache.flink.test.util.AbstractTestBase; +import org.apache.flink.types.Row; +import org.apache.flink.util.CloseableIterator; +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 java.time.Duration; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.apache.fluss.flink.FlinkConnectorOptions.BOOTSTRAP_SERVERS; +import static org.apache.fluss.flink.source.testutils.FlinkRowAssertionsUtils.assertResultsIgnoreOrder; +import static org.apache.fluss.flink.source.testutils.FlinkRowAssertionsUtils.collectRowsWithTimeout; +import static org.apache.fluss.server.testutils.FlussClusterExtension.BUILTIN_DATABASE; +import static org.assertj.core.api.Assertions.assertThat; + +/** Integration tests for Array type support in Flink connector. */ +abstract class FlinkArrayTypeITCase extends AbstractTestBase { + + @RegisterExtension + public static final FlussClusterExtension FLUSS_CLUSTER_EXTENSION = + FlussClusterExtension.builder().setNumOfTabletServers(3).build(); + + static final String CATALOG_NAME = "testcatalog"; + static final String DEFAULT_DB = "defaultdb"; + + protected StreamExecutionEnvironment env; + protected StreamTableEnvironment tEnv; + protected TableEnvironment tBatchEnv; + + @BeforeEach + void before() { + String bootstrapServers = FLUSS_CLUSTER_EXTENSION.getBootstrapServers(); + + env = StreamExecutionEnvironment.getExecutionEnvironment(); + env.setRuntimeMode(RuntimeExecutionMode.STREAMING); + + tEnv = StreamTableEnvironment.create(env); + tEnv.executeSql( + String.format( + "create catalog %s with ('type' = 'fluss', '%s' = '%s')", + CATALOG_NAME, BOOTSTRAP_SERVERS.key(), bootstrapServers)); + tEnv.executeSql("use catalog " + CATALOG_NAME); + tEnv.getConfig().set(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM, 2); + + tBatchEnv = + TableEnvironment.create(EnvironmentSettings.newInstance().inBatchMode().build()); + tBatchEnv.executeSql( + String.format( + "create catalog %s with ('type' = 'fluss', '%s' = '%s')", + CATALOG_NAME, BOOTSTRAP_SERVERS.key(), bootstrapServers)); + tBatchEnv.executeSql("use catalog " + CATALOG_NAME); + tBatchEnv + .getConfig() + .set(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM, 2); + + tEnv.executeSql("create database " + DEFAULT_DB); + tEnv.useDatabase(DEFAULT_DB); + tBatchEnv.useDatabase(DEFAULT_DB); + } + + @AfterEach + void after() { + tEnv.useDatabase(BUILTIN_DATABASE); + tEnv.executeSql(String.format("drop database %s cascade", DEFAULT_DB)); + } + + @Test + void testArrayOfPrimitiveTypesInLogTable() throws Exception { + tEnv.executeSql( + "create table array_test (" + + "id int, " + + "int_array array<int>, " + + "bigint_array array<bigint>, " + + "string_array array<string>, " + + "boolean_array array<boolean>, " + + "double_array array<double>" + + ") with ('bucket.num' = '3')"); + + tEnv.executeSql( + "INSERT INTO array_test VALUES " + + "(1, ARRAY[1, 2, 3], ARRAY[100, 200, 300], ARRAY['a', 'b', 'c'], ARRAY[true, false], ARRAY[1.1, 2.2]), " + + "(2, ARRAY[4, 5], ARRAY[400, 500], ARRAY['d', 'e'], ARRAY[true, true], ARRAY[3.3, 4.4]), " + + "(3, ARRAY[6], ARRAY[600], ARRAY['f'], ARRAY[false], ARRAY[5.5])") + .await(); + + CloseableIterator<Row> rowIter = tEnv.executeSql("select * from array_test").collect(); + List<String> expectedRows = + Arrays.asList( + "+I[1, [1, 2, 3], [100, 200, 300], [a, b, c], [true, false], [1.1, 2.2]]", + "+I[2, [4, 5], [400, 500], [d, e], [true, true], [3.3, 4.4]]", + "+I[3, [6], [600], [f], [false], [5.5]]"); + assertResultsIgnoreOrder(rowIter, expectedRows, true); + } + + @Test + void testArrayWithNullElements() throws Exception { + tEnv.executeSql( + "create table array_null_test (" + + "id int, " + + "int_array array<int>, " + + "string_array array<string>" + + ") with ('bucket.num' = '3')"); + + tEnv.executeSql( + "INSERT INTO array_null_test VALUES " + + "(1, ARRAY[1, CAST(NULL AS INT), 3], ARRAY['a', CAST(NULL AS STRING), 'c']), " + + "(2, ARRAY[CAST(NULL AS INT), CAST(NULL AS INT)], ARRAY[CAST(NULL AS STRING), 'b']), " + + "(3, CAST(NULL AS ARRAY<INT>), CAST(NULL AS ARRAY<STRING>))") + .await(); + + CloseableIterator<Row> rowIter = tEnv.executeSql("select * from array_null_test").collect(); + List<String> expectedRows = + Arrays.asList( + "+I[1, [1, null, 3], [a, null, c]]", + "+I[2, [null, null], [null, b]]", + "+I[3, null, null]"); + assertResultsIgnoreOrder(rowIter, expectedRows, true); + } + + @Test + void testNestedArrays() throws Exception { + tEnv.executeSql( + "create table nested_array_test (" + + "id int, " + + "nested_int_array array<array<int>>, " + + "nested_string_array array<array<string>>" + + ") with ('bucket.num' = '3')"); + + tEnv.executeSql( + "INSERT INTO nested_array_test VALUES " + + "(1, ARRAY[ARRAY[1, 2], ARRAY[3, 4]], ARRAY[ARRAY['a', 'b'], ARRAY['c']]), " + + "(2, ARRAY[ARRAY[5]], ARRAY[ARRAY['d', 'e', 'f']]), " + + "(3, ARRAY[ARRAY[6, 7, 8], ARRAY[9]], ARRAY[ARRAY['g']])") + .await(); + + CloseableIterator<Row> rowIter = + tEnv.executeSql("select * from nested_array_test").collect(); + List<String> expectedRows = + Arrays.asList( + "+I[1, [[1, 2], [3, 4]], [[a, b], [c]]]", + "+I[2, [[5]], [[d, e, f]]]", + "+I[3, [[6, 7, 8], [9]], [[g]]]"); + assertResultsIgnoreOrder(rowIter, expectedRows, true); + } + + @Test + void testArrayInPrimaryKeyTable() throws Exception { + tEnv.executeSql( + "create table array_pk_test (" + + "id int, " + + "int_array array<int>, " + + "string_array array<string>, " + + "primary key(id) not enforced" + + ") with ('bucket.num' = '3')"); + + tEnv.executeSql( + "INSERT INTO array_pk_test VALUES " + + "(1, ARRAY[1, 2, 3], ARRAY['a', 'b']), " + + "(2, ARRAY[4, 5], ARRAY['c', 'd', 'e'])") + .await(); + + CloseableIterator<Row> rowIter = tEnv.executeSql("select * from array_pk_test").collect(); + List<String> expectedRows = + Arrays.asList("+I[1, [1, 2, 3], [a, b]]", "+I[2, [4, 5], [c, d, e]]"); + assertResultsIgnoreOrder(rowIter, expectedRows, true); + } + + @Test + void testArrayWithAllDataTypes() throws Exception { + tEnv.executeSql( + "create table array_all_types_test (" + + "id int, " + + "int_array array<int>, " + + "bigint_array array<bigint>, " + + "double_array array<double>, " + + "boolean_array array<boolean>, " + + "string_array array<string>" Review Comment: Are the types here exhaustive? E.g. float, smallint, tinyint, timestamp etc. isn't tested. https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/types/ https://fluss.apache.org/docs/next/table-design/data-types/ -- 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]
