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 53559242345bead1a1dff3aac60d832605b0ce97 Author: Danny Chan <[email protected]> AuthorDate: Mon May 11 13:44:04 2026 +0800 feat: add variant type adapter for Flink (#18702) * Add variant type adapter for Flink * address the review comments (cherry picked from commit 5e72c96fae6117135e8b9d3a405c7d9c06003a65) --- .../hudi/client/model/AbstractHoodieRowData.java | 3 +- .../apache/hudi/client/model/BootstrapRowData.java | 6 ++- .../row/parquet/ParquetSchemaConverter.java | 47 ++++++++++++++++++ .../apache/hudi/util/AvroToRowDataConverters.java | 17 +++++++ .../apache/hudi/util/HoodieSchemaConverter.java | 24 ++++++---- .../apache/hudi/util/RowDataToAvroConverters.java | 28 +++++++++++ .../hudi/util/TestHoodieSchemaConverter.java | 3 ++ .../ITTestVariantCrossEngineCompatibility.java | 4 ++ .../org/apache/hudi/adapter/DataTypeAdapter.java | 53 ++++++++++++++++++++ .../org/apache/hudi/adapter/DataTypeAdapter.java | 53 ++++++++++++++++++++ .../org/apache/hudi/adapter/DataTypeAdapter.java | 53 ++++++++++++++++++++ .../org/apache/hudi/adapter/DataTypeAdapter.java | 53 ++++++++++++++++++++ .../org/apache/hudi/adapter/DataTypeAdapter.java | 53 ++++++++++++++++++++ .../org/apache/hudi/adapter/DataTypeAdapter.java | 56 ++++++++++++++++++++++ 14 files changed, 441 insertions(+), 12 deletions(-) diff --git a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/model/AbstractHoodieRowData.java b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/model/AbstractHoodieRowData.java index 94b13ea32a8a..d01bf3e34d76 100644 --- a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/model/AbstractHoodieRowData.java +++ b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/model/AbstractHoodieRowData.java @@ -18,6 +18,7 @@ package org.apache.hudi.client.model; +import org.apache.hudi.adapter.DataTypeAdapter; import org.apache.hudi.common.model.HoodieOperation; import org.apache.hudi.common.util.ValidationUtils; @@ -169,6 +170,6 @@ public abstract class AbstractHoodieRowData implements RowData { protected abstract int rebaseOrdinal(int ordinal); public Variant getVariant(int i) { - throw new UnsupportedOperationException("Variant is not supported yet."); + return DataTypeAdapter.getVariant(row, rebaseOrdinal(i)); } } diff --git a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/model/BootstrapRowData.java b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/model/BootstrapRowData.java index 92c0f35a4232..23c4c82f9668 100644 --- a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/model/BootstrapRowData.java +++ b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/model/BootstrapRowData.java @@ -18,6 +18,8 @@ package org.apache.hudi.client.model; +import org.apache.hudi.adapter.DataTypeAdapter; + import org.apache.flink.table.data.ArrayData; import org.apache.flink.table.data.DecimalData; import org.apache.flink.table.data.MapData; @@ -149,7 +151,7 @@ public class BootstrapRowData implements RowData { return getter.apply(pos); } - public Variant getVariant(int i) { - throw new UnsupportedOperationException("Variant is not supported yet."); + public Variant getVariant(int pos) { + return DataTypeAdapter.getVariant(row, pos); } } 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 d98497fdc86c..668098314d48 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 @@ -18,6 +18,8 @@ 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.collection.Pair; import lombok.extern.slf4j.Slf4j; @@ -155,6 +157,15 @@ public class ParquetSchemaConverter { new MapType( convertToRowField(keyValueType.getLeft()).getType().copy(true), convertToRowField(keyValueType.getRight()).getType())); + } else if (hasVariantAnnotation(logicalType)) { + 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."); + } + dataType = DataTypeAdapter.createVariantType(); } else { dataType = DataTypes.of(new RowType( @@ -190,6 +201,39 @@ public class ParquetSchemaConverter { return new MessageType(name, types); } + /** + * Checks whether the group carries the Parquet {@code VARIANT} logical type annotation. + * Uses class-name matching so this compiles against parquet-java versions that predate the + * {@code VariantLogicalTypeAnnotation} class (< 1.15.2). + */ + private static boolean hasVariantAnnotation(LogicalTypeAnnotation logicalType) { + // needs to ensure the writer attach the variant annotation in 1.3. + return logicalType != null + && logicalType.getClass().getSimpleName().equals("VariantLogicalTypeAnnotation"); + } + + /** + * Checks whether a variant group contains a {@code typed_value} field, indicating a shredded + * layout. Called only after {@link #hasVariantAnnotation} returns true. + */ + private static boolean isShreddedVariant(GroupType groupType) { + return groupType.containsField(HoodieSchema.Variant.VARIANT_TYPED_VALUE_FIELD); + } + + /** + * Converts a Variant column to the canonical unshredded Parquet layout: + * a group with required binary {@code metadata} and required binary {@code value}. + */ + private static Type convertVariantToParquetType(String name, Type.Repetition repetition) { + // TODO: add .as(LogicalTypeAnnotation.variantType()) once parquet-java is bumped to 1.16.0 + return Types.buildGroup(repetition) + .addField(Types.primitive(PrimitiveType.PrimitiveTypeName.BINARY, Type.Repetition.REQUIRED) + .named(HoodieSchema.Variant.VARIANT_METADATA_FIELD)) + .addField(Types.primitive(PrimitiveType.PrimitiveTypeName.BINARY, Type.Repetition.REQUIRED) + .named(HoodieSchema.Variant.VARIANT_VALUE_FIELD)) + .named(name); + } + private static Type convertToParquetType( String name, LogicalType type, Type.Repetition repetition) { switch (type.getTypeRoot()) { @@ -304,6 +348,9 @@ public class ParquetSchemaConverter { .addField(convertToParquetType(field.getName(), field.getType(), field.getType().isNullable() ? Type.Repetition.OPTIONAL : Type.Repetition.REQUIRED))); return builder.named(name); default: + if (DataTypeAdapter.isVariantType(type)) { + return convertVariantToParquetType(name, repetition); + } throw new UnsupportedOperationException("Unsupported type: " + type); } } diff --git a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/AvroToRowDataConverters.java b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/AvroToRowDataConverters.java index fee24e520382..d63ce350b487 100644 --- a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/AvroToRowDataConverters.java +++ b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/AvroToRowDataConverters.java @@ -18,6 +18,8 @@ package org.apache.hudi.util; +import org.apache.hudi.adapter.DataTypeAdapter; + import lombok.AccessLevel; import lombok.NoArgsConstructor; import org.apache.avro.generic.GenericFixed; @@ -155,6 +157,9 @@ public class AvroToRowDataConverters { case MULTISET: return createMapConverter(type, utcTimezone); default: + if (DataTypeAdapter.isVariantType(type)) { + return createVariantConverter(); + } throw new UnsupportedOperationException("Unsupported type: " + type); } } @@ -212,6 +217,18 @@ public class AvroToRowDataConverters { }; } + /** + * Creates a converter for Flink 2.1+ VARIANT LogicalType. The converter receives an Avro + * GenericRecord carrying metadata/value binary fields and produces a Flink + * {@code BinaryVariant}. + */ + private static AvroToRowDataConverter createVariantConverter() { + return avroObject -> { + IndexedRecord record = (IndexedRecord) avroObject; + return DataTypeAdapter.createVariant(convertToBytes(record.get(1)), convertToBytes(record.get(0))); + }; + } + private static AvroToRowDataConverter createTimestampConverter(int precision, boolean utcTimezone) { final ChronoUnit chronoUnit; if (precision <= 3) { 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 0b3a1af7982b..c49142800b45 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 @@ -19,6 +19,7 @@ package org.apache.hudi.util; +import org.apache.hudi.adapter.DataTypeAdapter; import org.apache.hudi.common.schema.HoodieSchema; import org.apache.hudi.common.schema.HoodieSchemaField; import org.apache.hudi.common.schema.HoodieSchemaType; @@ -222,6 +223,10 @@ public class HoodieSchemaConverter { case RAW: default: + if (DataTypeAdapter.isVariantType(logicalType)) { + schema = HoodieSchema.createVariant(); + break; + } throw new UnsupportedOperationException( "Unsupported type for HoodieSchema conversion: " + logicalType); } @@ -576,23 +581,24 @@ public class HoodieSchemaConverter { } /** - * Converts a Variant schema to Flink's ROW type. - * Variant is represented as ROW<`metadata` BYTES, `value` BYTES> in Flink. + * Converts a Variant HoodieSchema to the native Flink {@code VariantType} DataType. + * Requires Flink 2.1+ at runtime; throws {@link UnsupportedOperationException} on older versions. * * @param schema HoodieSchema to convert (must be a VARIANT type) - * @return DataType representing the Variant as a ROW with binary fields + * @return native VariantType DataType + * @throws UnsupportedOperationException if Flink runtime is pre-2.1 or variant is shredded */ private static DataType convertVariant(HoodieSchema schema) { if (schema.getType() != HoodieSchemaType.VARIANT) { throw new IllegalStateException("Expected HoodieSchema.Variant but got: " + schema.getClass()); } - // Variant is stored as a struct with two binary fields: metadata and value. - // Field order follows the Parquet spec and Iceberg convention (metadata first, value second). - return DataTypes.ROW( - DataTypes.FIELD("metadata", DataTypes.BYTES().notNull()), - DataTypes.FIELD("value", DataTypes.BYTES().notNull()) - ).notNull(); + if (((HoodieSchema.Variant) schema).isShredded()) { + throw new UnsupportedOperationException( + "Shredded Variant is not yet supported in Flink. Use unshredded Variant instead."); + } + + return DataTypeAdapter.createVariantType().notNull(); } /** diff --git a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/RowDataToAvroConverters.java b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/RowDataToAvroConverters.java index 9eb2979e19aa..6bf94b527074 100644 --- a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/RowDataToAvroConverters.java +++ b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/RowDataToAvroConverters.java @@ -18,6 +18,7 @@ package org.apache.hudi.util; +import org.apache.hudi.adapter.DataTypeAdapter; import org.apache.hudi.common.schema.HoodieSchema; import org.apache.hudi.common.schema.HoodieSchemaField; import org.apache.hudi.common.schema.HoodieSchemaType; @@ -239,6 +240,10 @@ public class RowDataToAvroConverters { break; case RAW: default: + if (DataTypeAdapter.isVariantType(type)) { + converter = createVariantConverter(); + break; + } throw new UnsupportedOperationException("Unsupported type: " + type); } @@ -357,5 +362,28 @@ public class RowDataToAvroConverters { } }; } + + /** + * Creates a converter for Flink 2.1+ VARIANT LogicalType. The converter receives a Flink + * {@code Variant} object at runtime and extracts the raw metadata/value byte arrays, + * then packs them into an Avro GenericRecord with the Variant schema. + * + * <p>No shredded-variant check is needed here: {@code HoodieSchemaConverter.convertVariant()} + * already rejects shredded variants before a Flink type or converter is ever constructed, + * and Flink 2.1 itself only supports unshredded variants (FLIP-521). + */ + private static RowDataToAvroConverter createVariantConverter() { + return new RowDataToAvroConverter() { + private static final long serialVersionUID = 1L; + + @Override + public Object convert(HoodieSchema schema, Object object) { + final GenericRecord record = new GenericData.Record(schema.toAvroSchema()); + record.put(HoodieSchema.Variant.VARIANT_METADATA_FIELD, ByteBuffer.wrap(DataTypeAdapter.getVariantMetadata(object))); + record.put(HoodieSchema.Variant.VARIANT_VALUE_FIELD, ByteBuffer.wrap(DataTypeAdapter.getVariantValue(object))); + return record; + } + }; + } } 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 e2fd16d7112b..51b8a15f4c7b 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,6 +36,7 @@ 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; @@ -692,6 +693,7 @@ public class TestHoodieSchemaConverter { } @Test + @Disabled("disabled and reopen the tests for 1.3") public void testVariantTypeConversion() { // Test direct Variant conversion HoodieSchema variantSchema = HoodieSchema.createVariant(); @@ -708,6 +710,7 @@ public class TestHoodieSchemaConverter { } @Test + @Disabled("disabled and reopen the tests for 1.3") public void testVariantInRecordConversion() { // Test Variant field within a record HoodieSchema recordWithVariant = HoodieSchema.createRecord( 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 9aade5503b4c..03305d478ecd 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 @@ -27,6 +27,7 @@ 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; @@ -114,6 +115,7 @@ public class ITTestVariantCrossEngineCompatibility { } @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"); @@ -123,6 +125,7 @@ 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"); @@ -132,6 +135,7 @@ 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"); 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 new file mode 100644 index 000000000000..79b9e254dd61 --- /dev/null +++ b/hudi-flink-datasource/hudi-flink1.17.x/src/main/java/org/apache/hudi/adapter/DataTypeAdapter.java @@ -0,0 +1,53 @@ +/* + * 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.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; + +/** + * Adapter utils to provide {@code DataType} utilities. + */ +public class DataTypeAdapter { + public static Variant getVariant(RowData rowData, int pos) { + throw new UnsupportedOperationException("Variant is not supported yet."); + } + + public static Object createVariant(byte[] value, byte[] metadata) { + throw new UnsupportedOperationException("Variant is not supported yet."); + } + + public static boolean isVariantType(LogicalType logicalType) { + return false; + } + + public static DataType createVariantType() { + throw new UnsupportedOperationException("Variant is not supported yet."); + } + + public static byte[] getVariantMetadata(Object obj) { + throw new UnsupportedOperationException("Variant is not supported yet."); + } + + public static byte[] getVariantValue(Object obj) { + throw new UnsupportedOperationException("Variant is not supported yet."); + } +} 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 new file mode 100644 index 000000000000..79b9e254dd61 --- /dev/null +++ b/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/adapter/DataTypeAdapter.java @@ -0,0 +1,53 @@ +/* + * 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.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; + +/** + * Adapter utils to provide {@code DataType} utilities. + */ +public class DataTypeAdapter { + public static Variant getVariant(RowData rowData, int pos) { + throw new UnsupportedOperationException("Variant is not supported yet."); + } + + public static Object createVariant(byte[] value, byte[] metadata) { + throw new UnsupportedOperationException("Variant is not supported yet."); + } + + public static boolean isVariantType(LogicalType logicalType) { + return false; + } + + public static DataType createVariantType() { + throw new UnsupportedOperationException("Variant is not supported yet."); + } + + public static byte[] getVariantMetadata(Object obj) { + throw new UnsupportedOperationException("Variant is not supported yet."); + } + + public static byte[] getVariantValue(Object obj) { + throw new UnsupportedOperationException("Variant is not supported yet."); + } +} 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 new file mode 100644 index 000000000000..79b9e254dd61 --- /dev/null +++ b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/adapter/DataTypeAdapter.java @@ -0,0 +1,53 @@ +/* + * 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.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; + +/** + * Adapter utils to provide {@code DataType} utilities. + */ +public class DataTypeAdapter { + public static Variant getVariant(RowData rowData, int pos) { + throw new UnsupportedOperationException("Variant is not supported yet."); + } + + public static Object createVariant(byte[] value, byte[] metadata) { + throw new UnsupportedOperationException("Variant is not supported yet."); + } + + public static boolean isVariantType(LogicalType logicalType) { + return false; + } + + public static DataType createVariantType() { + throw new UnsupportedOperationException("Variant is not supported yet."); + } + + public static byte[] getVariantMetadata(Object obj) { + throw new UnsupportedOperationException("Variant is not supported yet."); + } + + public static byte[] getVariantValue(Object obj) { + throw new UnsupportedOperationException("Variant is not supported yet."); + } +} 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 new file mode 100644 index 000000000000..79b9e254dd61 --- /dev/null +++ b/hudi-flink-datasource/hudi-flink1.20.x/src/main/java/org/apache/hudi/adapter/DataTypeAdapter.java @@ -0,0 +1,53 @@ +/* + * 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.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; + +/** + * Adapter utils to provide {@code DataType} utilities. + */ +public class DataTypeAdapter { + public static Variant getVariant(RowData rowData, int pos) { + throw new UnsupportedOperationException("Variant is not supported yet."); + } + + public static Object createVariant(byte[] value, byte[] metadata) { + throw new UnsupportedOperationException("Variant is not supported yet."); + } + + public static boolean isVariantType(LogicalType logicalType) { + return false; + } + + public static DataType createVariantType() { + throw new UnsupportedOperationException("Variant is not supported yet."); + } + + public static byte[] getVariantMetadata(Object obj) { + throw new UnsupportedOperationException("Variant is not supported yet."); + } + + public static byte[] getVariantValue(Object obj) { + throw new UnsupportedOperationException("Variant is not supported yet."); + } +} 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 new file mode 100644 index 000000000000..566ca723648a --- /dev/null +++ b/hudi-flink-datasource/hudi-flink2.0.x/src/main/java/org/apache/hudi/adapter/DataTypeAdapter.java @@ -0,0 +1,53 @@ +/* + * 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.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; + +/** + * Adapter utils to provide {@code DataType} utilities. + */ +public class DataTypeAdapter { + public static Variant getVariant(RowData rowData, int pos) { + throw new UnsupportedOperationException("Variant is not supported yet."); + } + + public static Object createVariant(byte[] value, byte[] metadata) { + throw new UnsupportedOperationException("Variant is not supported yet."); + } + + public static boolean isVariantType(LogicalType logicalType) { + return false; + } + + public static DataType createVariantType() { + throw new UnsupportedOperationException("Variant is not supported yet."); + } + + public static byte[] getVariantMetadata(Object obj) { + throw new UnsupportedOperationException("Variant is not supported yet."); + } + + public static byte[] getVariantValue(Object obj) { + throw new UnsupportedOperationException("Variant is not supported yet."); + } +} \ 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 new file mode 100644 index 000000000000..6ae4c4a6baba --- /dev/null +++ b/hudi-flink-datasource/hudi-flink2.1.x/src/main/java/org/apache/hudi/adapter/DataTypeAdapter.java @@ -0,0 +1,56 @@ +/* + * 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.table.api.DataTypes; +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.table.types.logical.LogicalTypeRoot; +import org.apache.flink.types.variant.BinaryVariant; +import org.apache.flink.types.variant.Variant; + +/** + * Adapter utils to provide {@code DataType} utilities. + */ +public class DataTypeAdapter { + public static Variant getVariant(RowData rowData, int pos) { + return rowData.getVariant(pos); + } + + public static Object createVariant(byte[] value, byte[] metadata) { + return new BinaryVariant(value, metadata); + } + + public static boolean isVariantType(LogicalType logicalType) { + return logicalType.getTypeRoot() == LogicalTypeRoot.VARIANT; + } + + public static DataType createVariantType() { + return DataTypes.VARIANT(); + } + + public static byte[] getVariantMetadata(Object obj) { + return ((BinaryVariant) obj).getMetadata(); + } + + public static byte[] getVariantValue(Object obj) { + return ((BinaryVariant) obj).getValue(); + } +}
