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 41717512f0c00ec5dd1d5a2c8b91210f244220a5 Author: Krishen <[email protected]> AuthorDate: Mon May 18 21:26:41 2026 -0700 fix(flink): enforce Parquet VARIANT annotation in Flink schema conversion for unshredded variant (#18539) * feat(flink): write/read unshredded variant to Flink parquet file writers/readers using Flink's Variant type --------- Co-authored-by: Krishen Bhan <“[email protected]”> Co-authored-by: Cursor <[email protected]> (cherry picked from commit 71aa121035cf912770f6ad4bf157a945ae3a8409) --- .../row/parquet/ParquetSchemaConverter.java | 31 ++++- .../apache/hudi/util/HoodieSchemaConverter.java | 4 + .../row/parquet/TestParquetSchemaConverter.java | 148 +++++++++++++++++++++ .../hudi/util/TestHoodieSchemaConverter.java | 34 ++++- .../ITTestVariantCrossEngineCompatibility.java | 3 +- .../org/apache/hudi/adapter/DataTypeAdapter.java | 20 ++- .../org/apache/hudi/adapter/DataTypeAdapter.java | 20 ++- .../org/apache/hudi/adapter/DataTypeAdapter.java | 20 ++- .../org/apache/hudi/adapter/DataTypeAdapter.java | 20 ++- .../org/apache/hudi/adapter/DataTypeAdapter.java | 20 ++- .../org/apache/hudi/adapter/DataTypeAdapter.java | 35 +++++ 11 files changed, 323 insertions(+), 32 deletions(-) diff --git a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/parquet/ParquetSchemaConverter.java b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/parquet/ParquetSchemaConverter.java index 668098314d48..dd19100f8ccd 100644 --- a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/parquet/ParquetSchemaConverter.java +++ b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/parquet/ParquetSchemaConverter.java @@ -49,6 +49,17 @@ import static org.apache.parquet.schema.LogicalTypeAnnotation.TimeUnit; /** * Schema converter converts Parquet schema to and from Flink internal types. * + * <p>On reads, this converter performs best-effort physical type mapping. It detects the + * Parquet {@code VARIANT} annotation and will reject shredded variants. Blob and Vector types + * cannot be distinguished from ordinary binary columns via Parquet schema alone. + * + * <p>On writes, this converter maps Flink {@code VariantType} to the canonical unshredded Parquet + * layout (group with binary metadata + value fields). The VARIANT logical type annotation is + * resolved by {@link DataTypeAdapter#variantParquetAnnotation()} — on Flink 2.1+ with + * parquet-java 1.16.0+ the annotation is attached automatically; on pre-2.1 Flink or with + * parquet < 1.16.0 the write throws {@link UnsupportedOperationException} because writing + * variant data without the annotation would produce files that no reader can identify as variant. + * * <p>Reference org.apache.flink.formats.parquet.utils.ParquetSchemaConverter to support timestamp of INT64 8 bytes. */ @Slf4j @@ -158,6 +169,9 @@ public class ParquetSchemaConverter { convertToRowField(keyValueType.getLeft()).getType().copy(true), convertToRowField(keyValueType.getRight()).getType())); } else if (hasVariantAnnotation(logicalType)) { + // Fires for files written with parquet-java that carry the VARIANT annotation. + // The reader infers the Flink RowType from the Parquet footer via convertToRowType(), + // so this annotation detection is the primary mechanism for recognizing Variant columns. if (isShreddedVariant(groupType)) { throw new UnsupportedOperationException( "Shredded Variant is not supported in Flink. " @@ -223,10 +237,25 @@ public class ParquetSchemaConverter { /** * Converts a Variant column to the canonical unshredded Parquet layout: * a group with required binary {@code metadata} and required binary {@code value}. + * + * <p>No shredded-variant guard is needed here: Flink 2.1's {@code VariantType} is a single + * atomic {@code LogicalTypeRoot.VARIANT} with no shredding representation (FLIP-521 scopes + * shredding out), so a shredded variant can never arrive as a Flink LogicalType. + * + * <p>Delegates to {@link DataTypeAdapter#variantParquetAnnotation()} for the VARIANT logical + * type annotation. On Flink < 2.1 this throws (variant writes are unsupported). On Flink 2.1+ + * with parquet-java < 1.16.0 this also throws, because writing variant data without the + * annotation would produce files that no reader can identify as variant. */ private static Type convertVariantToParquetType(String name, Type.Repetition repetition) { - // TODO: add .as(LogicalTypeAnnotation.variantType()) once parquet-java is bumped to 1.16.0 + LogicalTypeAnnotation annotation = DataTypeAdapter.variantParquetAnnotation() + .orElseThrow(() -> new UnsupportedOperationException( + "Cannot write Variant columns: parquet-java 1.16.0+ is required to emit the VARIANT " + + "logical type annotation. Without the annotation, readers cannot identify the " + + "column as Variant. Current parquet-java version does not support " + + "LogicalTypeAnnotation.variantType().")); return Types.buildGroup(repetition) + .as(annotation) .addField(Types.primitive(PrimitiveType.PrimitiveTypeName.BINARY, Type.Repetition.REQUIRED) .named(HoodieSchema.Variant.VARIANT_METADATA_FIELD)) .addField(Types.primitive(PrimitiveType.PrimitiveTypeName.BINARY, Type.Repetition.REQUIRED) diff --git a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/HoodieSchemaConverter.java b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/HoodieSchemaConverter.java index c49142800b45..c413101dde46 100644 --- a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/HoodieSchemaConverter.java +++ b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/HoodieSchemaConverter.java @@ -70,6 +70,10 @@ public class HoodieSchemaConverter { * <p>The "{rowName}." is used as the nested row type name prefix in order to generate * the right schema. Nested record types that only differ by type name are still compatible. * + * <p>On Flink 2.1+, {@code LogicalTypeRoot.VARIANT} is detected via string comparison + * (to avoid compile-time dependency) and mapped to {@link HoodieSchema#createVariant()}. + * Pre-2.1 Flink does not support Variant. + * * @param logicalType Flink logical type * @param rowName the record name * @return HoodieSchema matching this logical type diff --git a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/io/storage/row/parquet/TestParquetSchemaConverter.java b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/io/storage/row/parquet/TestParquetSchemaConverter.java index 318f2034d175..403b84068496 100644 --- a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/io/storage/row/parquet/TestParquetSchemaConverter.java +++ b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/io/storage/row/parquet/TestParquetSchemaConverter.java @@ -18,6 +18,10 @@ package org.apache.hudi.io.storage.row.parquet; +import org.apache.hudi.adapter.DataTypeAdapter; +import org.apache.hudi.common.schema.HoodieSchema; +import org.apache.hudi.common.util.Option; + import org.apache.flink.table.api.DataTypes; import org.apache.flink.table.types.DataType; import org.apache.flink.table.types.logical.ArrayType; @@ -27,13 +31,19 @@ import org.apache.flink.table.types.logical.DecimalType; import org.apache.flink.table.types.logical.DoubleType; import org.apache.flink.table.types.logical.FloatType; import org.apache.flink.table.types.logical.IntType; +import org.apache.flink.table.types.logical.LogicalType; import org.apache.flink.table.types.logical.MapType; import org.apache.flink.table.types.logical.RowType; import org.apache.flink.table.types.logical.SmallIntType; import org.apache.flink.table.types.logical.TimestampType; import org.apache.flink.table.types.logical.TinyIntType; import org.apache.flink.table.types.logical.VarCharType; +import org.apache.parquet.schema.GroupType; +import org.apache.parquet.schema.LogicalTypeAnnotation; import org.apache.parquet.schema.MessageType; +import org.apache.parquet.schema.PrimitiveType; +import org.apache.parquet.schema.Type; +import org.apache.parquet.schema.Types; import org.junit.jupiter.api.Test; import java.util.Arrays; @@ -41,6 +51,9 @@ import java.util.Collections; 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.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; /** * Test cases for {@link ParquetSchemaConverter}. @@ -216,4 +229,139 @@ public class TestParquetSchemaConverter { + "}\n"; assertThat(messageType.toString(), is(expected)); } + + /** + * A Parquet group with metadata + value binary fields but NO VARIANT annotation must be + * treated as a plain ROW. Only the Parquet {@code VARIANT} annotation triggers variant + * detection in this converter; unannotated groups are never guessed as variant. + */ + @Test + void testVariantPhysicalLayoutTreatedAsRow() { + MessageType variantParquet = new MessageType( + "test", + Types.primitive(PrimitiveType.PrimitiveTypeName.INT32, + Type.Repetition.REQUIRED).named("id"), + Types.buildGroup(Type.Repetition.REQUIRED) + .addField(Types.primitive(PrimitiveType.PrimitiveTypeName.BINARY, + Type.Repetition.REQUIRED).named("metadata")) + .addField(Types.primitive(PrimitiveType.PrimitiveTypeName.BINARY, + Type.Repetition.REQUIRED).named("value")) + .named("data")); + + RowType rowType = ParquetSchemaConverter.convertToRowType(variantParquet); + assertEquals(2, rowType.getFieldCount()); + assertEquals("ROW", rowType.getTypeAt(1).getTypeRoot().name()); + } + + /** + * Unannotated group with metadata + value + typed_value (3 fields) is treated as a generic + * ROW when no annotation or schema hint is present. + */ + @Test + void testUnannotatedShreddedGroupTreatedAsRow() { + MessageType shreddedNoAnnotation = new MessageType( + "test", + Types.primitive(PrimitiveType.PrimitiveTypeName.INT32, + Type.Repetition.REQUIRED).named("id"), + Types.buildGroup(Type.Repetition.REQUIRED) + .addField(Types.primitive(PrimitiveType.PrimitiveTypeName.BINARY, + Type.Repetition.REQUIRED).named("metadata")) + .addField(Types.primitive(PrimitiveType.PrimitiveTypeName.BINARY, + Type.Repetition.REQUIRED).named("value")) + .addField(Types.primitive(PrimitiveType.PrimitiveTypeName.INT32, + Type.Repetition.OPTIONAL).named("typed_value")) + .named("data")); + + RowType rowType = ParquetSchemaConverter.convertToRowType(shreddedNoAnnotation); + assertEquals(2, rowType.getFieldCount()); + assertEquals("ROW", rowType.getTypeAt(1).getTypeRoot().name()); + } + + /** + * On Flink 2.1+ with parquet 1.16.0+, converting a RowType containing a Variant column to a + * Parquet MessageType should produce a group with the VARIANT annotation and required binary + * {@code metadata} and {@code value} fields. + * On pre-2.1 Flink this test is skipped since VariantType does not exist. + * On parquet < 1.16.0 the write is expected to fail (annotation unavailable). + */ + @Test + void testVariantWritePathProducesCorrectLayout() { + LogicalType variantType; + try { + variantType = DataTypeAdapter.createVariantType().getLogicalType(); + } catch (UnsupportedOperationException e) { + // Pre-2.1 Flink: VariantType doesn't exist, skip + return; + } + + RowType rowType = RowType.of( + new LogicalType[]{new IntType(), variantType}, + new String[]{"id", "data"}); + + if (!DataTypeAdapter.variantParquetAnnotation().isPresent()) { + // parquet < 1.16.0: write must fail because annotation is unavailable + UnsupportedOperationException ex = org.junit.jupiter.api.Assertions.assertThrows( + UnsupportedOperationException.class, + () -> ParquetSchemaConverter.convertToParquetMessageType("test", rowType)); + assertTrue(ex.getMessage().contains("parquet-java 1.16.0+"), + "Error message should mention parquet version requirement"); + return; + } + + // parquet 1.16.0+: write succeeds with annotation + MessageType messageType = ParquetSchemaConverter.convertToParquetMessageType("test", rowType); + assertEquals(2, messageType.getFieldCount()); + + Type variantField = messageType.getType("data"); + assertTrue(variantField instanceof GroupType, "Variant column should be a Parquet group"); + GroupType variantGroup = (GroupType) variantField; + assertEquals(2, variantGroup.getFieldCount()); + assertEquals(HoodieSchema.Variant.VARIANT_METADATA_FIELD, variantGroup.getType(0).getName()); + assertEquals(HoodieSchema.Variant.VARIANT_VALUE_FIELD, variantGroup.getType(1).getName()); + assertTrue(variantGroup.getType(0).isPrimitive()); + assertTrue(variantGroup.getType(1).isPrimitive()); + assertEquals(PrimitiveType.PrimitiveTypeName.BINARY, + variantGroup.getType(0).asPrimitiveType().getPrimitiveTypeName()); + assertEquals(PrimitiveType.PrimitiveTypeName.BINARY, + variantGroup.getType(1).asPrimitiveType().getPrimitiveTypeName()); + assertNotNull(variantGroup.getLogicalTypeAnnotation(), + "Variant group must carry the VARIANT annotation"); + } + + /** + * Verifies that writing a Variant column fails with a clear error when parquet-java on the + * classpath does not support the VARIANT annotation (< 1.16.0). On pre-2.1 Flink the adapter + * throws directly; on Flink 2.1+ with parquet < 1.16.0 the write path throws. + */ + @Test + void testVariantWriteFailsWithoutAnnotation() { + Option<LogicalTypeAnnotation> annotationOpt; + try { + annotationOpt = DataTypeAdapter.variantParquetAnnotation(); + } catch (UnsupportedOperationException e) { + // Pre-2.1 Flink: expected to throw from the adapter + assertTrue(e.getMessage().contains("VARIANT type is only supported in Flink 2.1+")); + return; + } + + if (annotationOpt.isPresent()) { + // parquet 1.16.0+: annotation is available, write succeeds — nothing to test here + return; + } + + // Flink 2.1 + parquet < 1.16.0: annotation is null, write must fail + LogicalType variantType = DataTypeAdapter.createVariantType().getLogicalType(); + RowType rowType = RowType.of( + new LogicalType[]{new IntType(), variantType}, + new String[]{"id", "data"}); + + UnsupportedOperationException ex = org.junit.jupiter.api.Assertions.assertThrows( + UnsupportedOperationException.class, + () -> ParquetSchemaConverter.convertToParquetMessageType("test", rowType)); + assertTrue(ex.getMessage().contains("parquet-java 1.16.0+"), + "Error message should mention the parquet version requirement"); + assertTrue(ex.getMessage().contains("VARIANT"), + "Error message should mention VARIANT"); + } + } 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 9a4f920b859e..9f6ad8f53bf7 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 @@ -695,19 +695,16 @@ public class TestHoodieSchemaConverter { @Test public void testVariantTypeConversion() { - // Test direct Variant conversion HoodieSchema variantSchema = HoodieSchema.createVariant(); DataType dataType = HoodieSchemaConverter.convertToDataType(variantSchema); assertNotNull(dataType); - // Verify it's a Variant assertThat("the return type should be variant", dataType.getLogicalType().asSummaryString(), is("VARIANT NOT NULL")); } @Test public void testVariantInRecordConversion() { - // Test Variant field within a record HoodieSchema recordWithVariant = HoodieSchema.createRecord( "test_record", null, @@ -722,11 +719,40 @@ public class TestHoodieSchemaConverter { assertEquals(2, result.getFieldCount()); assertEquals("data", result.getFieldNames().get(1)); - // Verify variant field assertThat("the return type should be variant", result.getTypeAt(1).asSummaryString(), is("VARIANT NOT NULL")); } + @Test + public void testVariantInArrayConversion() { + HoodieSchema arrayOfVariant = HoodieSchema.createArray(HoodieSchema.createVariant()); + DataType dataType = HoodieSchemaConverter.convertToDataType(arrayOfVariant); + assertNotNull(dataType); + assertInstanceOf(ArrayType.class, dataType.getLogicalType()); + LogicalType elementType = ((ArrayType) dataType.getLogicalType()).getElementType(); + assertEquals("VARIANT", elementType.getTypeRoot().name()); + } + + @Test + public void testVariantInMapConversion() { + HoodieSchema mapOfVariant = HoodieSchema.createMap(HoodieSchema.createVariant()); + DataType dataType = HoodieSchemaConverter.convertToDataType(mapOfVariant); + assertNotNull(dataType); + assertInstanceOf(MapType.class, dataType.getLogicalType()); + LogicalType valueType = ((MapType) dataType.getLogicalType()).getValueType(); + assertEquals("VARIANT", valueType.getTypeRoot().name()); + } + + @Test + public void testShreddedVariantConversionThrows() { + HoodieSchema.Variant shredded = HoodieSchema.createVariantShredded( + HoodieSchema.create(HoodieSchemaType.STRING)); + UnsupportedOperationException ex = assertThrows( + UnsupportedOperationException.class, + () -> HoodieSchemaConverter.convertToDataType(shredded)); + assertTrue(ex.getMessage().contains("Shredded Variant is not yet supported in Flink")); + } + @Test public void testBlobStructureValidation() { // Positive case: Create ROW matching BLOB structure 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 de6e3835d120..e7f9ed3766f1 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 @@ -41,6 +41,7 @@ import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; + /** * Integration test for cross-engine compatibility - verifying that Flink can read Variant tables written by Spark 4.0. */ @@ -56,7 +57,6 @@ public class ITTestVariantCrossEngineCompatibility { 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 String createTableDdl = String.format( "CREATE TABLE variant_table (" + " id INT," @@ -73,7 +73,6 @@ public class ITTestVariantCrossEngineCompatibility { tableEnv.executeSql(createTableDdl); - // Query the table to verify Flink can read the data TableResult result = tableEnv.executeSql("SELECT id, name, v, ts FROM variant_table ORDER BY id"); List<Row> rows = CollectionUtil.iteratorToList(result.collect()); diff --git a/hudi-flink-datasource/hudi-flink1.17.x/src/main/java/org/apache/hudi/adapter/DataTypeAdapter.java b/hudi-flink-datasource/hudi-flink1.17.x/src/main/java/org/apache/hudi/adapter/DataTypeAdapter.java index 79b9e254dd61..e8e31b341a18 100644 --- a/hudi-flink-datasource/hudi-flink1.17.x/src/main/java/org/apache/hudi/adapter/DataTypeAdapter.java +++ b/hudi-flink-datasource/hudi-flink1.17.x/src/main/java/org/apache/hudi/adapter/DataTypeAdapter.java @@ -22,17 +22,27 @@ import org.apache.flink.table.data.RowData; import org.apache.flink.table.types.DataType; import org.apache.flink.table.types.logical.LogicalType; import org.apache.flink.types.variant.Variant; +import org.apache.hudi.common.util.Option; +import org.apache.parquet.schema.LogicalTypeAnnotation; /** * Adapter utils to provide {@code DataType} utilities. */ public class DataTypeAdapter { + private static final String VARIANT_UNSUPPORTED_MSG = + "VARIANT type is only supported in Flink 2.1+. " + + "Please upgrade your Flink version to use Variant columns."; + + public static Option<LogicalTypeAnnotation> variantParquetAnnotation() { + throw new UnsupportedOperationException(VARIANT_UNSUPPORTED_MSG); + } + public static Variant getVariant(RowData rowData, int pos) { - throw new UnsupportedOperationException("Variant is not supported yet."); + throw new UnsupportedOperationException(VARIANT_UNSUPPORTED_MSG); } public static Object createVariant(byte[] value, byte[] metadata) { - throw new UnsupportedOperationException("Variant is not supported yet."); + throw new UnsupportedOperationException(VARIANT_UNSUPPORTED_MSG); } public static boolean isVariantType(LogicalType logicalType) { @@ -40,14 +50,14 @@ public class DataTypeAdapter { } public static DataType createVariantType() { - throw new UnsupportedOperationException("Variant is not supported yet."); + throw new UnsupportedOperationException(VARIANT_UNSUPPORTED_MSG); } public static byte[] getVariantMetadata(Object obj) { - throw new UnsupportedOperationException("Variant is not supported yet."); + throw new UnsupportedOperationException(VARIANT_UNSUPPORTED_MSG); } public static byte[] getVariantValue(Object obj) { - throw new UnsupportedOperationException("Variant is not supported yet."); + throw new UnsupportedOperationException(VARIANT_UNSUPPORTED_MSG); } } diff --git a/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/adapter/DataTypeAdapter.java b/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/adapter/DataTypeAdapter.java index 79b9e254dd61..e8e31b341a18 100644 --- a/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/adapter/DataTypeAdapter.java +++ b/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/adapter/DataTypeAdapter.java @@ -22,17 +22,27 @@ import org.apache.flink.table.data.RowData; import org.apache.flink.table.types.DataType; import org.apache.flink.table.types.logical.LogicalType; import org.apache.flink.types.variant.Variant; +import org.apache.hudi.common.util.Option; +import org.apache.parquet.schema.LogicalTypeAnnotation; /** * Adapter utils to provide {@code DataType} utilities. */ public class DataTypeAdapter { + private static final String VARIANT_UNSUPPORTED_MSG = + "VARIANT type is only supported in Flink 2.1+. " + + "Please upgrade your Flink version to use Variant columns."; + + public static Option<LogicalTypeAnnotation> variantParquetAnnotation() { + throw new UnsupportedOperationException(VARIANT_UNSUPPORTED_MSG); + } + public static Variant getVariant(RowData rowData, int pos) { - throw new UnsupportedOperationException("Variant is not supported yet."); + throw new UnsupportedOperationException(VARIANT_UNSUPPORTED_MSG); } public static Object createVariant(byte[] value, byte[] metadata) { - throw new UnsupportedOperationException("Variant is not supported yet."); + throw new UnsupportedOperationException(VARIANT_UNSUPPORTED_MSG); } public static boolean isVariantType(LogicalType logicalType) { @@ -40,14 +50,14 @@ public class DataTypeAdapter { } public static DataType createVariantType() { - throw new UnsupportedOperationException("Variant is not supported yet."); + throw new UnsupportedOperationException(VARIANT_UNSUPPORTED_MSG); } public static byte[] getVariantMetadata(Object obj) { - throw new UnsupportedOperationException("Variant is not supported yet."); + throw new UnsupportedOperationException(VARIANT_UNSUPPORTED_MSG); } public static byte[] getVariantValue(Object obj) { - throw new UnsupportedOperationException("Variant is not supported yet."); + throw new UnsupportedOperationException(VARIANT_UNSUPPORTED_MSG); } } diff --git a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/adapter/DataTypeAdapter.java b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/adapter/DataTypeAdapter.java index 79b9e254dd61..e8e31b341a18 100644 --- a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/adapter/DataTypeAdapter.java +++ b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/adapter/DataTypeAdapter.java @@ -22,17 +22,27 @@ import org.apache.flink.table.data.RowData; import org.apache.flink.table.types.DataType; import org.apache.flink.table.types.logical.LogicalType; import org.apache.flink.types.variant.Variant; +import org.apache.hudi.common.util.Option; +import org.apache.parquet.schema.LogicalTypeAnnotation; /** * Adapter utils to provide {@code DataType} utilities. */ public class DataTypeAdapter { + private static final String VARIANT_UNSUPPORTED_MSG = + "VARIANT type is only supported in Flink 2.1+. " + + "Please upgrade your Flink version to use Variant columns."; + + public static Option<LogicalTypeAnnotation> variantParquetAnnotation() { + throw new UnsupportedOperationException(VARIANT_UNSUPPORTED_MSG); + } + public static Variant getVariant(RowData rowData, int pos) { - throw new UnsupportedOperationException("Variant is not supported yet."); + throw new UnsupportedOperationException(VARIANT_UNSUPPORTED_MSG); } public static Object createVariant(byte[] value, byte[] metadata) { - throw new UnsupportedOperationException("Variant is not supported yet."); + throw new UnsupportedOperationException(VARIANT_UNSUPPORTED_MSG); } public static boolean isVariantType(LogicalType logicalType) { @@ -40,14 +50,14 @@ public class DataTypeAdapter { } public static DataType createVariantType() { - throw new UnsupportedOperationException("Variant is not supported yet."); + throw new UnsupportedOperationException(VARIANT_UNSUPPORTED_MSG); } public static byte[] getVariantMetadata(Object obj) { - throw new UnsupportedOperationException("Variant is not supported yet."); + throw new UnsupportedOperationException(VARIANT_UNSUPPORTED_MSG); } public static byte[] getVariantValue(Object obj) { - throw new UnsupportedOperationException("Variant is not supported yet."); + throw new UnsupportedOperationException(VARIANT_UNSUPPORTED_MSG); } } diff --git a/hudi-flink-datasource/hudi-flink1.20.x/src/main/java/org/apache/hudi/adapter/DataTypeAdapter.java b/hudi-flink-datasource/hudi-flink1.20.x/src/main/java/org/apache/hudi/adapter/DataTypeAdapter.java index 79b9e254dd61..e8e31b341a18 100644 --- a/hudi-flink-datasource/hudi-flink1.20.x/src/main/java/org/apache/hudi/adapter/DataTypeAdapter.java +++ b/hudi-flink-datasource/hudi-flink1.20.x/src/main/java/org/apache/hudi/adapter/DataTypeAdapter.java @@ -22,17 +22,27 @@ import org.apache.flink.table.data.RowData; import org.apache.flink.table.types.DataType; import org.apache.flink.table.types.logical.LogicalType; import org.apache.flink.types.variant.Variant; +import org.apache.hudi.common.util.Option; +import org.apache.parquet.schema.LogicalTypeAnnotation; /** * Adapter utils to provide {@code DataType} utilities. */ public class DataTypeAdapter { + private static final String VARIANT_UNSUPPORTED_MSG = + "VARIANT type is only supported in Flink 2.1+. " + + "Please upgrade your Flink version to use Variant columns."; + + public static Option<LogicalTypeAnnotation> variantParquetAnnotation() { + throw new UnsupportedOperationException(VARIANT_UNSUPPORTED_MSG); + } + public static Variant getVariant(RowData rowData, int pos) { - throw new UnsupportedOperationException("Variant is not supported yet."); + throw new UnsupportedOperationException(VARIANT_UNSUPPORTED_MSG); } public static Object createVariant(byte[] value, byte[] metadata) { - throw new UnsupportedOperationException("Variant is not supported yet."); + throw new UnsupportedOperationException(VARIANT_UNSUPPORTED_MSG); } public static boolean isVariantType(LogicalType logicalType) { @@ -40,14 +50,14 @@ public class DataTypeAdapter { } public static DataType createVariantType() { - throw new UnsupportedOperationException("Variant is not supported yet."); + throw new UnsupportedOperationException(VARIANT_UNSUPPORTED_MSG); } public static byte[] getVariantMetadata(Object obj) { - throw new UnsupportedOperationException("Variant is not supported yet."); + throw new UnsupportedOperationException(VARIANT_UNSUPPORTED_MSG); } public static byte[] getVariantValue(Object obj) { - throw new UnsupportedOperationException("Variant is not supported yet."); + throw new UnsupportedOperationException(VARIANT_UNSUPPORTED_MSG); } } diff --git a/hudi-flink-datasource/hudi-flink2.0.x/src/main/java/org/apache/hudi/adapter/DataTypeAdapter.java b/hudi-flink-datasource/hudi-flink2.0.x/src/main/java/org/apache/hudi/adapter/DataTypeAdapter.java index 566ca723648a..18db48f0fed1 100644 --- a/hudi-flink-datasource/hudi-flink2.0.x/src/main/java/org/apache/hudi/adapter/DataTypeAdapter.java +++ b/hudi-flink-datasource/hudi-flink2.0.x/src/main/java/org/apache/hudi/adapter/DataTypeAdapter.java @@ -22,17 +22,27 @@ import org.apache.flink.table.data.RowData; import org.apache.flink.table.types.DataType; import org.apache.flink.table.types.logical.LogicalType; import org.apache.flink.types.variant.Variant; +import org.apache.hudi.common.util.Option; +import org.apache.parquet.schema.LogicalTypeAnnotation; /** * Adapter utils to provide {@code DataType} utilities. */ public class DataTypeAdapter { + private static final String VARIANT_UNSUPPORTED_MSG = + "VARIANT type is only supported in Flink 2.1+. " + + "Please upgrade your Flink version to use Variant columns."; + + public static Option<LogicalTypeAnnotation> variantParquetAnnotation() { + throw new UnsupportedOperationException(VARIANT_UNSUPPORTED_MSG); + } + public static Variant getVariant(RowData rowData, int pos) { - throw new UnsupportedOperationException("Variant is not supported yet."); + throw new UnsupportedOperationException(VARIANT_UNSUPPORTED_MSG); } public static Object createVariant(byte[] value, byte[] metadata) { - throw new UnsupportedOperationException("Variant is not supported yet."); + throw new UnsupportedOperationException(VARIANT_UNSUPPORTED_MSG); } public static boolean isVariantType(LogicalType logicalType) { @@ -40,14 +50,14 @@ public class DataTypeAdapter { } public static DataType createVariantType() { - throw new UnsupportedOperationException("Variant is not supported yet."); + throw new UnsupportedOperationException(VARIANT_UNSUPPORTED_MSG); } public static byte[] getVariantMetadata(Object obj) { - throw new UnsupportedOperationException("Variant is not supported yet."); + throw new UnsupportedOperationException(VARIANT_UNSUPPORTED_MSG); } public static byte[] getVariantValue(Object obj) { - throw new UnsupportedOperationException("Variant is not supported yet."); + throw new UnsupportedOperationException(VARIANT_UNSUPPORTED_MSG); } } \ No newline at end of file diff --git a/hudi-flink-datasource/hudi-flink2.1.x/src/main/java/org/apache/hudi/adapter/DataTypeAdapter.java b/hudi-flink-datasource/hudi-flink2.1.x/src/main/java/org/apache/hudi/adapter/DataTypeAdapter.java index 6ae4c4a6baba..77d6f708ece3 100644 --- a/hudi-flink-datasource/hudi-flink2.1.x/src/main/java/org/apache/hudi/adapter/DataTypeAdapter.java +++ b/hudi-flink-datasource/hudi-flink2.1.x/src/main/java/org/apache/hudi/adapter/DataTypeAdapter.java @@ -25,11 +25,46 @@ import org.apache.flink.table.types.logical.LogicalType; import org.apache.flink.table.types.logical.LogicalTypeRoot; import org.apache.flink.types.variant.BinaryVariant; import org.apache.flink.types.variant.Variant; +import org.apache.hudi.common.util.Option; +import org.apache.parquet.schema.LogicalTypeAnnotation; + +import java.lang.reflect.Method; /** * Adapter utils to provide {@code DataType} utilities. */ public class DataTypeAdapter { + + /** + * The Parquet Variant binary format specification version passed to + * {@code LogicalTypeAnnotation.variantType(byte)}. Version 1 is the initial spec + * defined by the Parquet Variant proposal (parquet-format 2.11.0 / parquet-java 1.16.0). + */ + private static final byte VARIANT_SPEC_VERSION = 1; + + /** + * Cached VARIANT annotation resolved via reflection. Empty if parquet-java + * on the classpath predates {@code LogicalTypeAnnotation.variantType()} (< 1.16.0). + */ + private static final Option<LogicalTypeAnnotation> VARIANT_ANNOTATION = resolveVariantAnnotation(); + + private static Option<LogicalTypeAnnotation> resolveVariantAnnotation() { + try { + Method factory = LogicalTypeAnnotation.class.getMethod("variantType", byte.class); + return Option.of((LogicalTypeAnnotation) factory.invoke(null, VARIANT_SPEC_VERSION)); + } catch (Exception e) { + return Option.empty(); + } + } + + /** + * Returns the Parquet VARIANT {@link LogicalTypeAnnotation} if parquet-java 1.16.0+ is on the + * classpath, or empty if the annotation class is unavailable. + */ + public static Option<LogicalTypeAnnotation> variantParquetAnnotation() { + return VARIANT_ANNOTATION; + } + public static Variant getVariant(RowData rowData, int pos) { return rowData.getVariant(pos); }
