This is an automated email from the ASF dual-hosted git repository. voonhous pushed a commit to branch release-1.2.1 in repository https://gitbox.apache.org/repos/asf/hudi.git
commit a3272bd3293a7b01a66bb1efd602d6d1ed430004 Author: Danny Chan <[email protected]> AuthorDate: Tue May 12 17:09:39 2026 +0800 chore: migrate the flink ITs run to flink2.1 (#18717) * chore: migrate the flink ITs to run with flink2.1 * fix compile errors * fix test failures * fix test failure * address review comments (cherry picked from commit b6bc1658e8f2862583c22425eaa0665ae100f5a3) --- .github/workflows/bot.yml | 30 ++++---- azure-pipelines-20230430.yml | 4 +- .../hudi/util/TestHoodieSchemaConverter.java | 23 +++--- .../hudi/sink/bootstrap/RLIBootstrapOperator.java | 27 ++++--- .../sink/bootstrap/TestRLIBootstrapOperator.java | 80 +++++++++++++++++++++ .../ITTestVariantCrossEngineCompatibility.java | 31 ++++---- .../hudi/adapter/DataTypeAdapterTestUtils.java | 28 ++++++++ .../hudi/adapter/DataTypeAdapterTestUtils.java | 28 ++++++++ .../hudi/adapter/DataTypeAdapterTestUtils.java | 28 ++++++++ .../hudi/adapter/DataTypeAdapterTestUtils.java | 28 ++++++++ .../runtime/kryo/KryoSerializerSnapshot.java | 3 + .../hudi/adapter/DataTypeAdapterTestUtils.java | 28 ++++++++ .../table/format/cow/ParquetSplitReaderUtil.java | 83 ++++++++++++++++++++++ .../runtime/kryo/KryoSerializerSnapshot.java | 3 + .../hudi/adapter/DataTypeAdapterTestUtils.java | 32 +++++++++ pom.xml | 14 ++-- 16 files changed, 404 insertions(+), 66 deletions(-) diff --git a/.github/workflows/bot.yml b/.github/workflows/bot.yml index 1e50c64121cb..06468d0c4fa2 100644 --- a/.github/workflows/bot.yml +++ b/.github/workflows/bot.yml @@ -67,7 +67,7 @@ jobs: include: - scalaProfile: "scala-2.12" sparkProfile: "spark3.5" - flinkProfile: "flink1.18" + flinkProfile: "flink2.1" steps: - uses: actions/checkout@v5 @@ -126,7 +126,7 @@ jobs: include: - scalaProfile: "scala-2.12" sparkProfile: "spark3.5" - flinkProfile: "flink1.18" + flinkProfile: "flink2.1" steps: - uses: actions/checkout@v5 @@ -178,7 +178,7 @@ jobs: include: - scalaProfile: "scala-2.12" sparkProfile: "spark3.5" - flinkProfile: "flink1.18" + flinkProfile: "flink2.1" env: UT_MODULES: >- @@ -563,7 +563,7 @@ jobs: include: - scalaProfile: "scala-2.12" sparkProfile: "spark3.5" - flinkProfile: "flink1.20" + flinkProfile: "flink2.1" steps: - uses: actions/checkout@v5 @@ -938,7 +938,7 @@ jobs: FLINK_PROFILE: ${{ matrix.flinkProfile }} FLINK_AVRO_VERSION: ${{ matrix.flinkAvroVersion }} FLINK_PARQUET_VERSION: ${{ matrix.flinkParquetVersion }} - if: ${{ endsWith(env.FLINK_PROFILE, '1.20') }} + if: ${{ endsWith(env.FLINK_PROFILE, '2.1') }} run: | mvn clean install -T 2 -Pintegration-tests -D"$SCALA_PROFILE" -D"$FLINK_PROFILE" -pl hudi-flink-datasource/hudi-flink -am -Davro.version="$FLINK_AVRO_VERSION" -Dparquet.version="$FLINK_PARQUET_VERSION" -DskipTests=true $MVN_ARGS mvn verify -Pintegration-tests -D"$SCALA_PROFILE" -D"$FLINK_PROFILE" -Davro.version="$FLINK_AVRO_VERSION" -Dparquet.version="$FLINK_PARQUET_VERSION" $FLINK_IT_FILTER1 -pl hudi-flink-datasource/hudi-flink $MVN_ARGS @@ -948,9 +948,9 @@ jobs: strategy: matrix: include: - - flinkProfile: "flink1.20" + - flinkProfile: "flink2.1" flinkAvroVersion: "1.11.4" - flinkParquetVersion: '1.13.1' + flinkParquetVersion: '1.15.2' steps: - uses: actions/checkout@v5 - name: Set up JDK 11 @@ -974,7 +974,7 @@ jobs: FLINK_PROFILE: ${{ matrix.flinkProfile }} FLINK_AVRO_VERSION: ${{ matrix.flinkAvroVersion }} FLINK_PARQUET_VERSION: ${{ matrix.flinkParquetVersion }} - if: ${{ endsWith(env.FLINK_PROFILE, '1.20') }} + if: ${{ endsWith(env.FLINK_PROFILE, '2.1') }} run: | mvn clean install -T 2 -Pintegration-tests -D"$SCALA_PROFILE" -D"$FLINK_PROFILE" -pl hudi-flink-datasource/hudi-flink -am -Davro.version="$FLINK_AVRO_VERSION" -Dparquet.version="$FLINK_PARQUET_VERSION" -DskipTests=true $MVN_ARGS mvn verify -Pintegration-tests -D"$SCALA_PROFILE" -D"$FLINK_PROFILE" -Davro.version="$FLINK_AVRO_VERSION" -Dparquet.version="$FLINK_PARQUET_VERSION" $FLINK_IT_FILTER2 -pl hudi-flink-datasource/hudi-flink $MVN_ARGS @@ -985,19 +985,19 @@ jobs: matrix: include: - scalaProfile: 'scala-2.13' - flinkProfile: 'flink1.20' + flinkProfile: 'flink2.1' sparkProfile: 'spark3.5' sparkRuntime: 'spark3.5.0' - scalaProfile: 'scala-2.12' - flinkProfile: 'flink1.20' + flinkProfile: 'flink2.1' sparkProfile: 'spark3.5' sparkRuntime: 'spark3.5.0' - scalaProfile: 'scala-2.13' - flinkProfile: 'flink1.20' + flinkProfile: 'flink2.1' sparkProfile: 'spark4.0' sparkRuntime: 'spark4.0.0' - scalaProfile: 'scala-2.13' - flinkProfile: 'flink1.20' + flinkProfile: 'flink2.1' sparkProfile: 'spark4.1' sparkRuntime: 'spark4.1.1' @@ -1201,7 +1201,7 @@ jobs: matrix: include: - sparkProfile: 'spark3.5' - flinkProfile: 'flink1.20' + flinkProfile: 'flink2.1' sparkArchive: 'spark-3.5.3/spark-3.5.3-bin-hadoop3.tgz' steps: - uses: actions/checkout@v5 @@ -1297,9 +1297,9 @@ jobs: matrix: include: - scalaProfile: "scala-2.12" - flinkProfile: "flink1.20" + flinkProfile: "flink2.1" flinkAvroVersion: '1.11.4' - flinkParquetVersion: '1.13.1' + flinkParquetVersion: '1.15.2' steps: - uses: actions/checkout@v5 - name: Set up JDK 17 diff --git a/azure-pipelines-20230430.yml b/azure-pipelines-20230430.yml index 10fab9f34992..3edac6046b1a 100644 --- a/azure-pipelines-20230430.yml +++ b/azure-pipelines-20230430.yml @@ -14,7 +14,7 @@ # limitations under the License. # NOTE: -# This config file defines how Azure CI runs tests with Spark 3.5 and Flink 1.18 profiles. +# This config file defines how Azure CI runs tests with Spark 3.5 and Flink 2.1 profiles. # PRs will need to keep in sync with master's version to trigger the CI runs. # See scripts/jacoco/README.md for how aggregated code coverage report works # across multiple modules. @@ -101,7 +101,7 @@ parameters: - '!packaging/hudi-utilities-slim-bundle' variables: - BUILD_PROFILES: '-Dscala-2.12 -Dspark3.5 -Dflink1.18' + BUILD_PROFILES: '-Dscala-2.12 -Dspark3.5 -Dflink2.1' PLUGIN_OPTS: '-Dcheckstyle.skip=true -Drat.skip=true -ntp -B -V -Pwarn-log -Dorg.slf4j.simpleLogger.log.org.apache.maven.plugins.shade=warn -Dorg.slf4j.simpleLogger.log.org.apache.maven.plugins.dependency=warn' MVN_OPTS_INSTALL: '-T 3 -Phudi-platform-service -DskipTests $(BUILD_PROFILES) $(PLUGIN_OPTS) -Dmaven.wagon.httpconnectionManager.ttlSeconds=25 -Dmaven.wagon.http.retryHandler.count=5 -Dorg.eclipse.jetty.LEVEL=WARN' MVN_OPTS_TEST: '-fae -Pwarn-log $(BUILD_PROFILES) $(PLUGIN_OPTS)' diff --git a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/util/TestHoodieSchemaConverter.java b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/util/TestHoodieSchemaConverter.java index 51b8a15f4c7b..9a4f920b859e 100644 --- a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/util/TestHoodieSchemaConverter.java +++ b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/util/TestHoodieSchemaConverter.java @@ -36,11 +36,12 @@ import org.apache.flink.table.types.logical.RowType; import org.apache.flink.table.types.logical.TimestampType; import org.apache.flink.table.types.logical.VarCharType; import org.apache.flink.table.types.logical.VarBinaryType; -import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; import java.util.Arrays; +import static org.hamcrest.CoreMatchers.is; +import static org.hamcrest.MatcherAssert.assertThat; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertInstanceOf; @@ -693,24 +694,18 @@ public class TestHoodieSchemaConverter { } @Test - @Disabled("disabled and reopen the tests for 1.3") public void testVariantTypeConversion() { // Test direct Variant conversion HoodieSchema variantSchema = HoodieSchema.createVariant(); DataType dataType = HoodieSchemaConverter.convertToDataType(variantSchema); assertNotNull(dataType); - // Verify it's a ROW with metadata and value binary fields - RowType rowType = (RowType) dataType.getLogicalType(); - assertEquals(2, rowType.getFieldCount()); - assertEquals("metadata", rowType.getFieldNames().get(0)); - assertEquals("value", rowType.getFieldNames().get(1)); - assertInstanceOf(VarBinaryType.class, rowType.getTypeAt(0)); - assertInstanceOf(VarBinaryType.class, rowType.getTypeAt(1)); + // Verify it's a Variant + assertThat("the return type should be variant", + dataType.getLogicalType().asSummaryString(), is("VARIANT NOT NULL")); } @Test - @Disabled("disabled and reopen the tests for 1.3") public void testVariantInRecordConversion() { // Test Variant field within a record HoodieSchema recordWithVariant = HoodieSchema.createRecord( @@ -727,11 +722,9 @@ public class TestHoodieSchemaConverter { assertEquals(2, result.getFieldCount()); assertEquals("data", result.getFieldNames().get(1)); - // Verify variant field is a ROW<metadata BYTES, value BYTES> - RowType variantRowType = (RowType) result.getTypeAt(1); - assertEquals(2, variantRowType.getFieldCount()); - assertEquals("metadata", variantRowType.getFieldNames().get(0)); - assertEquals("value", variantRowType.getFieldNames().get(1)); + // Verify variant field + assertThat("the return type should be variant", + result.getTypeAt(1).asSummaryString(), is("VARIANT NOT NULL")); } @Test diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bootstrap/RLIBootstrapOperator.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bootstrap/RLIBootstrapOperator.java index 515b180d3cad..d0cb16a789e5 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bootstrap/RLIBootstrapOperator.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bootstrap/RLIBootstrapOperator.java @@ -24,6 +24,7 @@ import org.apache.hudi.common.data.HoodiePairData; import org.apache.hudi.common.function.SerializableFunctionUnchecked; import org.apache.hudi.common.model.FileSlice; import org.apache.hudi.common.model.HoodieRecordGlobalLocation; +import org.apache.hudi.common.table.HoodieTableConfig; import org.apache.hudi.common.table.HoodieTableMetaClient; import org.apache.hudi.configuration.FlinkOptions; import org.apache.hudi.metadata.HoodieBackedTableMetadata; @@ -52,7 +53,7 @@ import java.util.stream.Collectors; public class RLIBootstrapOperator extends AbstractBootstrapOperator { - private transient HoodieBackedTableMetadata metadataTable; + private transient HoodieBackedTableMetadata tableMetadata; private transient long loadedCnt; public RLIBootstrapOperator(Configuration conf) { @@ -63,13 +64,13 @@ public class RLIBootstrapOperator public void initializeState(StateInitializationContext context) throws Exception { loadedCnt = 0; HoodieTableMetaClient metaClient = StreamerUtil.createMetaClient(conf); - this.metadataTable = (HoodieBackedTableMetadata) metaClient.getTableFormat().getMetadataFactory().create( + this.tableMetadata = new HoodieBackedTableMetadata( HoodieFlinkEngineContext.DEFAULT, metaClient.getStorage(), StreamerUtil.metadataConfig(conf), conf.get(FlinkOptions.PATH)); // Load RLI records - preLoadRLIRecords(); + preLoadRLIRecords(metaClient.getTableConfig()); } @Override @@ -82,10 +83,20 @@ public class RLIBootstrapOperator // Utilities // ------------------------------------------------------------------------- - private void preLoadRLIRecords() { + private void preLoadRLIRecords(HoodieTableConfig tableConfig) { int taskID = RuntimeContextUtils.getIndexOfThisSubtask(getRuntimeContext()); int parallelism = RuntimeContextUtils.getNumberOfParallelSubtasks(getRuntimeContext()); + if (!tableMetadata.enabled()) { + if (tableConfig.isMetadataTableAvailable()) { + throw new RuntimeException("Can not initialize the table metadata"); + } + log.info("Skip loading RLI records because table metadata is not initialized, taskId = {}", taskID); + waitForBootstrapReady(taskID); + closeMetadataTable(); + return; + } + log.info("Start loading RLI records from metadata table, taskId = {}, parallelism = {}", taskID, parallelism); SerializableFunctionUnchecked<List<FileSlice>, List<FileSlice>> fileSlicesFilter = fileSlices -> { @@ -102,7 +113,7 @@ public class RLIBootstrapOperator // Each subtask loads buckets assigned to it long startTime = System.currentTimeMillis(); - HoodiePairData<String, HoodieRecordGlobalLocation> rliData = metadataTable.readRecordIndexLocations(fileSlicesFilter); + HoodiePairData<String, HoodieRecordGlobalLocation> rliData = tableMetadata.readRecordIndexLocations(fileSlicesFilter); rliData.forEach(locationPair -> emitIndexRecord(locationPair.getLeft(), locationPair.getRight())); long costMs = System.currentTimeMillis() - startTime; log.info("Finish loading RLI records, total records: {}, cost: {} ms, taskId = {}", loadedCnt, costMs, taskID); @@ -133,13 +144,13 @@ public class RLIBootstrapOperator } private void closeMetadataTable() { - if (metadataTable != null) { + if (tableMetadata != null) { try { - metadataTable.close(); + tableMetadata.close(); } catch (Exception e) { log.warn("Failed to close metadata table", e); } - metadataTable = null; + tableMetadata = null; } } } diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/bootstrap/TestRLIBootstrapOperator.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/bootstrap/TestRLIBootstrapOperator.java new file mode 100644 index 000000000000..2e684d0652ae --- /dev/null +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/bootstrap/TestRLIBootstrapOperator.java @@ -0,0 +1,80 @@ +/* + * 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.hudi.sink.bootstrap; + +import org.apache.hudi.client.model.HoodieFlinkInternalRow; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.configuration.FlinkOptions; +import org.apache.hudi.index.HoodieIndex; +import org.apache.hudi.metadata.MetadataPartitionType; +import org.apache.hudi.util.StreamerUtil; +import org.apache.hudi.utils.TestConfigurations; + +import org.apache.flink.configuration.Configuration; +import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.File; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +/** + * Tests for {@link RLIBootstrapOperator}. + */ +public class TestRLIBootstrapOperator { + + @TempDir + File tempFile; + + @Test + void testSkipPreloadForFreshTableWithoutMetadataTable() throws Exception { + Configuration conf = getRLIConf(); + StreamerUtil.initTableIfNotExists(conf); + + try (OneInputStreamOperatorTestHarness<HoodieFlinkInternalRow, HoodieFlinkInternalRow> harness = + new OneInputStreamOperatorTestHarness<>(new RLIBootstrapOperator(conf), 1, 1, 0)) { + harness.open(); + + assertEquals(0, harness.getOutput().size()); + } + } + + @Test + void testFailFastWhenMetadataTableIsMarkedAvailableButCannotBeLoaded() throws Exception { + Configuration conf = getRLIConf(); + HoodieTableMetaClient metaClient = StreamerUtil.initTableIfNotExists(conf); + metaClient.getTableConfig().setMetadataPartitionState(metaClient, MetadataPartitionType.FILES.getPartitionPath(), true); + + try (OneInputStreamOperatorTestHarness<HoodieFlinkInternalRow, HoodieFlinkInternalRow> harness = + new OneInputStreamOperatorTestHarness<>(new RLIBootstrapOperator(conf), 1, 1, 0)) { + RuntimeException error = assertThrows(RuntimeException.class, harness::open); + + assertEquals("Can not initialize the table metadata", error.getMessage()); + } + } + + private Configuration getRLIConf() { + Configuration conf = TestConfigurations.getDefaultConf(tempFile.getAbsolutePath()); + conf.set(FlinkOptions.METADATA_ENABLED, true); + conf.set(FlinkOptions.INDEX_TYPE, HoodieIndex.IndexType.GLOBAL_RECORD_LEVEL_INDEX.name()); + return conf; + } +} diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestVariantCrossEngineCompatibility.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestVariantCrossEngineCompatibility.java index 03305d478ecd..de6e3835d120 100644 --- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestVariantCrossEngineCompatibility.java +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestVariantCrossEngineCompatibility.java @@ -18,6 +18,8 @@ package org.apache.hudi.table; +import org.apache.hudi.adapter.DataTypeAdapter; +import org.apache.hudi.adapter.DataTypeAdapterTestUtils; import org.apache.hudi.common.testutils.HoodieTestUtils; import org.apache.hudi.common.util.StringUtils; import org.apache.hudi.utils.FlinkMiniCluster; @@ -27,7 +29,6 @@ import org.apache.flink.table.api.TableEnvironment; import org.apache.flink.table.api.TableResult; import org.apache.flink.types.Row; import org.apache.flink.util.CollectionUtil; -import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; import org.junit.jupiter.api.io.TempDir; @@ -51,19 +52,16 @@ public class ITTestVariantCrossEngineCompatibility { /** * Helper method to verify that Flink can read Spark 4.0 Variant tables. - * Variant data is represented as ROW<metadata BYTES, value BYTES> in Flink. */ private void verifyFlinkCanReadSparkVariantTable(String tablePath, String tableType, String testDescription) throws Exception { TableEnvironment tableEnv = TestTableEnvs.getBatchTableEnv(); // Create a Hudi table pointing to the Spark-written data - // In Flink, Variant is represented as ROW<metadata BYTES, value BYTES> - // NOTE: value is a reserved keyword String createTableDdl = String.format( "CREATE TABLE variant_table (" + " id INT," + " name STRING," - + " v ROW<metadata BYTES, `value` BYTES>," + + " v VARIANT," + " ts BIGINT," + " PRIMARY KEY (id) NOT ENFORCED" + ") WITH (" @@ -87,35 +85,32 @@ public class ITTestVariantCrossEngineCompatibility { assertEquals("row1", row.getField(1), "Second column should be name=row1"); assertEquals(1000L, row.getField(3), "Fourth column should be ts=1000"); - // Verify the variant column is readable as a ROW with binary fields - Row variantRow = (Row) row.getField(2); - assertNotNull(variantRow, "Variant column should not be null"); - - byte[] metadataBytes = (byte[]) variantRow.getField(0); - byte[] valueBytes = (byte[]) variantRow.getField(1); + // Verify the variant column is readable as a native Flink Variant. + Object variantObject = row.getField(2); + assertNotNull(variantObject, "Variant column should not be null"); + DataTypeAdapterTestUtils.assertAsBinaryVariant(variantObject); // Expected byte values from Spark 4.0 Variant representation: {"updated": true, "new_field": 123} byte[] expectedValueBytes = new byte[]{0x02, 0x02, 0x01, 0x00, 0x01, 0x00, 0x03, 0x04, 0x0C, 0x7B}; byte[] expectedMetadataBytes = new byte[]{0x01, 0x02, 0x00, 0x07, 0x10, 0x75, 0x70, 0x64, 0x61, 0x74, 0x65, 0x64, 0x6E, 0x65, 0x77, 0x5F, 0x66, 0x69, 0x65, 0x6C, 0x64}; - assertArrayEquals(expectedValueBytes, valueBytes, + assertArrayEquals(expectedValueBytes, DataTypeAdapter.getVariantValue(variantObject), String.format("Variant value bytes mismatch (%s). Expected: %s, Got: %s", testDescription, Arrays.toString(StringUtils.encodeHex(expectedValueBytes)), - Arrays.toString(StringUtils.encodeHex(valueBytes)))); + Arrays.toString(StringUtils.encodeHex(DataTypeAdapter.getVariantValue(variantObject))))); - assertArrayEquals(expectedMetadataBytes, metadataBytes, + assertArrayEquals(expectedMetadataBytes, DataTypeAdapter.getVariantMetadata(variantObject), String.format("Variant metadata bytes mismatch (%s). Expected: %s, Got: %s", testDescription, Arrays.toString(StringUtils.encodeHex(expectedMetadataBytes)), - Arrays.toString(StringUtils.encodeHex(metadataBytes)))); + Arrays.toString(StringUtils.encodeHex(DataTypeAdapter.getVariantMetadata(variantObject))))); tableEnv.executeSql("DROP TABLE variant_table"); } @Test - @Disabled("disabled and reopen the tests for 1.3") public void testFlinkReadSparkVariantCOWTable() throws Exception { // Test that Flink can read a COW table with Variant data written by Spark 4.0 Path cowTargetDir = tempDir.resolve("cow"); @@ -125,7 +120,6 @@ public class ITTestVariantCrossEngineCompatibility { } @Test - @Disabled("disabled and reopen the tests for 1.3") public void testFlinkReadSparkVariantMORTableWithAvro() throws Exception { // Test that Flink can read a MOR table with AVRO record type and Variant data written by Spark 4.0 Path morAvroTargetDir = tempDir.resolve("mor_avro"); @@ -135,7 +129,6 @@ public class ITTestVariantCrossEngineCompatibility { } @Test - @Disabled("disabled and reopen the tests for 1.3") public void testFlinkReadSparkVariantMORTableWithSpark() throws Exception { // Test that Flink can read a MOR table with SPARK record type and Variant data written by Spark 4.0 Path morSparkTargetDir = tempDir.resolve("mor_spark"); @@ -143,4 +136,4 @@ public class ITTestVariantCrossEngineCompatibility { String morSparkPath = morSparkTargetDir.resolve("variant_mor_spark").toString(); verifyFlinkCanReadSparkVariantTable(morSparkPath, "MERGE_ON_READ", "MOR table with SPARK record type"); } -} \ No newline at end of file +} diff --git a/hudi-flink-datasource/hudi-flink1.17.x/src/test/java/org/apache/hudi/adapter/DataTypeAdapterTestUtils.java b/hudi-flink-datasource/hudi-flink1.17.x/src/test/java/org/apache/hudi/adapter/DataTypeAdapterTestUtils.java new file mode 100644 index 000000000000..ae2e4107d6ea --- /dev/null +++ b/hudi-flink-datasource/hudi-flink1.17.x/src/test/java/org/apache/hudi/adapter/DataTypeAdapterTestUtils.java @@ -0,0 +1,28 @@ +/* + * 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.hudi.adapter; + +/** + * Adapter utils. + */ +public class DataTypeAdapterTestUtils { + public static void assertAsBinaryVariant(Object variantObject) { + throw new UnsupportedOperationException("Variant is not supported yet."); + } +} diff --git a/hudi-flink-datasource/hudi-flink1.18.x/src/test/java/org/apache/hudi/adapter/DataTypeAdapterTestUtils.java b/hudi-flink-datasource/hudi-flink1.18.x/src/test/java/org/apache/hudi/adapter/DataTypeAdapterTestUtils.java new file mode 100644 index 000000000000..ae2e4107d6ea --- /dev/null +++ b/hudi-flink-datasource/hudi-flink1.18.x/src/test/java/org/apache/hudi/adapter/DataTypeAdapterTestUtils.java @@ -0,0 +1,28 @@ +/* + * 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.hudi.adapter; + +/** + * Adapter utils. + */ +public class DataTypeAdapterTestUtils { + public static void assertAsBinaryVariant(Object variantObject) { + throw new UnsupportedOperationException("Variant is not supported yet."); + } +} diff --git a/hudi-flink-datasource/hudi-flink1.19.x/src/test/java/org/apache/hudi/adapter/DataTypeAdapterTestUtils.java b/hudi-flink-datasource/hudi-flink1.19.x/src/test/java/org/apache/hudi/adapter/DataTypeAdapterTestUtils.java new file mode 100644 index 000000000000..ae2e4107d6ea --- /dev/null +++ b/hudi-flink-datasource/hudi-flink1.19.x/src/test/java/org/apache/hudi/adapter/DataTypeAdapterTestUtils.java @@ -0,0 +1,28 @@ +/* + * 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.hudi.adapter; + +/** + * Adapter utils. + */ +public class DataTypeAdapterTestUtils { + public static void assertAsBinaryVariant(Object variantObject) { + throw new UnsupportedOperationException("Variant is not supported yet."); + } +} diff --git a/hudi-flink-datasource/hudi-flink1.20.x/src/test/java/org/apache/hudi/adapter/DataTypeAdapterTestUtils.java b/hudi-flink-datasource/hudi-flink1.20.x/src/test/java/org/apache/hudi/adapter/DataTypeAdapterTestUtils.java new file mode 100644 index 000000000000..ae2e4107d6ea --- /dev/null +++ b/hudi-flink-datasource/hudi-flink1.20.x/src/test/java/org/apache/hudi/adapter/DataTypeAdapterTestUtils.java @@ -0,0 +1,28 @@ +/* + * 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.hudi.adapter; + +/** + * Adapter utils. + */ +public class DataTypeAdapterTestUtils { + public static void assertAsBinaryVariant(Object variantObject) { + throw new UnsupportedOperationException("Variant is not supported yet."); + } +} diff --git a/hudi-flink-datasource/hudi-flink2.0.x/src/test/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializerSnapshot.java b/hudi-flink-datasource/hudi-flink2.0.x/src/test/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializerSnapshot.java index 1a256ffe7bba..aac9a4651aff 100644 --- a/hudi-flink-datasource/hudi-flink2.0.x/src/test/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializerSnapshot.java +++ b/hudi-flink-datasource/hudi-flink2.0.x/src/test/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializerSnapshot.java @@ -34,6 +34,9 @@ public class KryoSerializerSnapshot<T> implements TypeSerializerSnapshot<T> { private Class<T> type; + public KryoSerializerSnapshot() { + } + public KryoSerializerSnapshot(Class<T> type) { this.type = type; } diff --git a/hudi-flink-datasource/hudi-flink2.0.x/src/test/java/org/apache/hudi/adapter/DataTypeAdapterTestUtils.java b/hudi-flink-datasource/hudi-flink2.0.x/src/test/java/org/apache/hudi/adapter/DataTypeAdapterTestUtils.java new file mode 100644 index 000000000000..ae2e4107d6ea --- /dev/null +++ b/hudi-flink-datasource/hudi-flink2.0.x/src/test/java/org/apache/hudi/adapter/DataTypeAdapterTestUtils.java @@ -0,0 +1,28 @@ +/* + * 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.hudi.adapter; + +/** + * Adapter utils. + */ +public class DataTypeAdapterTestUtils { + public static void assertAsBinaryVariant(Object variantObject) { + throw new UnsupportedOperationException("Variant is not supported yet."); + } +} diff --git a/hudi-flink-datasource/hudi-flink2.1.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java b/hudi-flink-datasource/hudi-flink2.1.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java index bb5d0c55b81d..3e16417ba541 100644 --- a/hudi-flink-datasource/hudi-flink2.1.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java +++ b/hudi-flink-datasource/hudi-flink2.1.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java @@ -18,6 +18,7 @@ package org.apache.hudi.table.format.cow; +import org.apache.hudi.common.schema.HoodieSchema; import org.apache.hudi.common.util.ValidationUtils; import org.apache.hudi.table.format.cow.vector.HeapArrayGroupColumnVector; import org.apache.hudi.table.format.cow.vector.HeapArrayVector; @@ -320,6 +321,17 @@ public class ParquetSplitReaderUtil { } else { throw new UnsupportedOperationException("Unsupported create row with default value."); } + case VARIANT: + HeapRowColumnVector variantVector = new HeapRowColumnVector( + batchSize, + new HeapBytesVector(batchSize), + new HeapBytesVector(batchSize)); + if (value == null) { + variantVector.fillWithNulls(); + return variantVector; + } else { + throw new UnsupportedOperationException("Unsupported create variant with default value."); + } default: throw new UnsupportedOperationException("Unsupported type: " + type); } @@ -498,6 +510,20 @@ public class ParquetSplitReaderUtil { } } return new RowColumnReader(fieldReaders); + case VARIANT: + ColumnDescriptor valueDescriptor = getVariantColumnDescriptor( + physicalType, + descriptors, + depth, + HoodieSchema.Variant.VARIANT_VALUE_FIELD); + ColumnDescriptor metadataDescriptor = getVariantColumnDescriptor( + physicalType, + descriptors, + depth, + HoodieSchema.Variant.VARIANT_METADATA_FIELD); + return new RowColumnReader(Arrays.asList( + new BytesColumnReader(valueDescriptor, pages.getPageReader(valueDescriptor)), + new BytesColumnReader(metadataDescriptor, pages.getPageReader(metadataDescriptor)))); default: throw new UnsupportedOperationException(fieldType + " is not supported now."); } @@ -675,11 +701,68 @@ public class ParquetSplitReaderUtil { } } return new HeapRowColumnVector(batchSize, columnVectors); + case VARIANT: + validateVariantType(physicalType); + return new HeapRowColumnVector( + batchSize, + new HeapBytesVector(batchSize), + new HeapBytesVector(batchSize)); default: throw new UnsupportedOperationException(fieldType + " is not supported now."); } } + private static ColumnDescriptor getVariantColumnDescriptor( + Type physicalType, + List<ColumnDescriptor> descriptors, + int depth, + String fieldName) { + return descriptors.stream() + .filter(descriptor -> descriptor.getPath().length > depth + 1 + && fieldName.equals(descriptor.getPath()[depth + 1])) + .findFirst() + .orElseThrow(() -> new IllegalArgumentException( + "Invalid Variant Parquet schema: missing binary field '" + fieldName + "'.")); + } + + private static void validateVariantType(Type physicalType) { + if (!physicalType.isPrimitive()) { + GroupType groupType = physicalType.asGroupType(); + if (isShreddedVariant(groupType)) { + throw new UnsupportedOperationException( + "Shredded Variant is not supported in Flink. " + + "The Parquet group '" + groupType.getName() + "' contains a '" + + HoodieSchema.Variant.VARIANT_TYPED_VALUE_FIELD + + "' field indicating a shredded layout."); + } + validateVariantField(groupType, HoodieSchema.Variant.VARIANT_VALUE_FIELD); + validateVariantField(groupType, HoodieSchema.Variant.VARIANT_METADATA_FIELD); + } else { + throw new IllegalArgumentException( + "Type mismatch, expected Variant but got '" + physicalType + "'."); + } + } + + /** + * Checks whether a variant group contains a {@code typed_value} field, indicating a shredded + * layout. + */ + private static boolean isShreddedVariant(GroupType groupType) { + return groupType.containsField(HoodieSchema.Variant.VARIANT_TYPED_VALUE_FIELD); + } + + private static void validateVariantField(GroupType groupType, String fieldName) { + if (groupType.containsField(fieldName)) { + Type fieldType = groupType.getType(fieldName); + if (fieldType.isPrimitive() + && fieldType.asPrimitiveType().getPrimitiveTypeName() == PrimitiveType.PrimitiveTypeName.BINARY) { + return; + } + } + throw new IllegalArgumentException( + "Invalid Variant Parquet schema: missing binary field '" + fieldName + "'."); + } + /** * Returns the field index with given physical row type {@code groupType} and field name {@code fieldName}. * diff --git a/hudi-flink-datasource/hudi-flink2.1.x/src/test/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializerSnapshot.java b/hudi-flink-datasource/hudi-flink2.1.x/src/test/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializerSnapshot.java index 1a256ffe7bba..aac9a4651aff 100644 --- a/hudi-flink-datasource/hudi-flink2.1.x/src/test/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializerSnapshot.java +++ b/hudi-flink-datasource/hudi-flink2.1.x/src/test/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializerSnapshot.java @@ -34,6 +34,9 @@ public class KryoSerializerSnapshot<T> implements TypeSerializerSnapshot<T> { private Class<T> type; + public KryoSerializerSnapshot() { + } + public KryoSerializerSnapshot(Class<T> type) { this.type = type; } diff --git a/hudi-flink-datasource/hudi-flink2.1.x/src/test/java/org/apache/hudi/adapter/DataTypeAdapterTestUtils.java b/hudi-flink-datasource/hudi-flink2.1.x/src/test/java/org/apache/hudi/adapter/DataTypeAdapterTestUtils.java new file mode 100644 index 000000000000..2727b40b1393 --- /dev/null +++ b/hudi-flink-datasource/hudi-flink2.1.x/src/test/java/org/apache/hudi/adapter/DataTypeAdapterTestUtils.java @@ -0,0 +1,32 @@ +/* + * 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.hudi.adapter; + +import org.apache.flink.types.variant.BinaryVariant; + +import static org.junit.jupiter.api.Assertions.assertInstanceOf; + +/** + * Adapter utils. + */ +public class DataTypeAdapterTestUtils { + public static void assertAsBinaryVariant(Object variantObject) { + assertInstanceOf(BinaryVariant.class, variantObject, "Variant column should be a BinaryVariant"); + } +} diff --git a/pom.xml b/pom.xml index cd26cc509f36..0e556cc76c1c 100644 --- a/pom.xml +++ b/pom.xml @@ -153,14 +153,14 @@ <flink1.19.version>1.19.2</flink1.19.version> <flink1.18.version>1.18.1</flink1.18.version> <flink1.17.version>1.17.1</flink1.17.version> - <flink.version>${flink1.20.version}</flink.version> - <hudi.flink.module>hudi-flink1.20.x</hudi.flink.module> - <flink.bundle.version>1.20</flink.bundle.version> + <flink.version>${flink2.1.version}</flink.version> + <hudi.flink.module>hudi-flink2.1.x</hudi.flink.module> + <flink.bundle.version>2.1</flink.bundle.version> <!-- This is fixed to match with version from flink-avro --> <flink.avro.version>1.11.4</flink.avro.version> - <flink.format.parquet.version>1.13.1</flink.format.parquet.version> + <flink.format.parquet.version>1.15.2</flink.format.parquet.version> <!-- check kafka version --> - <flink.connector.kafka.version>3.3.0-1.20</flink.connector.kafka.version> + <flink.connector.kafka.version>4.0.1-2.0</flink.connector.kafka.version> <flink.runtime.artifactId>flink-runtime</flink.runtime.artifactId> <flink.table.runtime.artifactId>flink-table-runtime</flink.table.runtime.artifactId> <flink.table.planner.artifactId>flink-table-planner_2.12</flink.table.planner.artifactId> @@ -170,7 +170,7 @@ <flink.streaming.java.artifactId>flink-streaming-java</flink.streaming.java.artifactId> <flink.clients.artifactId>flink-clients</flink.clients.artifactId> <flink.connector.kafka.artifactId>flink-connector-kafka</flink.connector.kafka.artifactId> - <flink.hadoop.compatibility.artifactId>flink-hadoop-compatibility_2.12</flink.hadoop.compatibility.artifactId> + <flink.hadoop.compatibility.artifactId>flink-hadoop-compatibility</flink.hadoop.compatibility.artifactId> <rocksdbjni.version>7.5.3</rocksdbjni.version> <spark33.version>3.3.4</spark33.version> <spark34.version>3.4.3</spark34.version> @@ -2899,6 +2899,7 @@ <module>hudi-flink-datasource/hudi-flink2.1.x</module> </modules> <activation> + <activeByDefault>true</activeByDefault> <property> <name>flink2.1</name> </property> @@ -2942,7 +2943,6 @@ <module>hudi-flink-datasource/hudi-flink1.20.x</module> </modules> <activation> - <activeByDefault>true</activeByDefault> <property> <name>flink1.20</name> </property>
