You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "talatuyarer (via GitHub)" <gi...@apache.org> on 2025/10/05 12:06:14 UTC
[PR] Adding Variant Support to Flink [iceberg]
talatuyarer opened a new pull request, #14259:
URL: https://github.com/apache/iceberg/pull/14259
Added full variant read/write support for Flink, including type conversion, schema integration, and binary serialization. The implementation uses Flink's BinaryVariant class with getValue() and getMetadata() methods.
**Supported Types:**
- All primitive types: BOOLEAN, INT8/16/32/64, FLOAT, DOUBLE, STRING, BINARY, UUID
- Date/time types: DATE, TIME, TIMESTAMPTZ, TIMESTAMPNTZ
- Decimal types: DECIMAL4/8/16 with proper precision handling
- Complex types: Arrays, objects, nested structures
**Future Work**
- Missing features for future PRs:
- Filter pushdown on variant field contents
- Variant field metrics and statistics
- Variant field aggregations
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Adding Variant Support to Flink 2.1 [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14259:
URL: https://github.com/apache/iceberg/pull/14259#discussion_r2404946222
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetWriters.java:
##########
@@ -605,4 +621,53 @@ protected Object get(RowData struct, int index) {
return fieldGetter[index].getFieldOrNull(struct);
}
}
+
+ /** Variant writer converts from Flink Variant to Iceberg Variant */
+ public static class VariantWriter implements ParquetValueWriter<Variant> {
+ private final ParquetValueWriter<org.apache.iceberg.variants.Variant> writer;
+
+ private VariantWriter(ParquetValueWriter<?> writer) {
+ this.writer = (ParquetValueWriter<org.apache.iceberg.variants.Variant>) writer;
+ }
+
+ @Override
+ public void write(int repetitionLevel, Variant variant) {
+ // Flink's Variant implementation uses BinaryVariant as the standard
+ // Based on FLIP-521
+ if (!(variant instanceof org.apache.flink.types.variant.BinaryVariant)) {
+ throw new IllegalArgumentException(
+ "Expected BinaryVariant but got: " + variant.getClass().getSimpleName());
+ }
+
+ org.apache.flink.types.variant.BinaryVariant binaryVariant =
+ (org.apache.flink.types.variant.BinaryVariant) variant;
+
+ byte[] metadataBytes = binaryVariant.getMetadata();
+ byte[] valueBytes = binaryVariant.getValue();
+
+ org.apache.iceberg.variants.VariantMetadata metadata =
Review Comment:
Could we import the clas
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetWriters.java:
##########
@@ -605,4 +621,53 @@ protected Object get(RowData struct, int index) {
return fieldGetter[index].getFieldOrNull(struct);
}
}
+
+ /** Variant writer converts from Flink Variant to Iceberg Variant */
+ public static class VariantWriter implements ParquetValueWriter<Variant> {
+ private final ParquetValueWriter<org.apache.iceberg.variants.Variant> writer;
+
+ private VariantWriter(ParquetValueWriter<?> writer) {
+ this.writer = (ParquetValueWriter<org.apache.iceberg.variants.Variant>) writer;
+ }
+
+ @Override
+ public void write(int repetitionLevel, Variant variant) {
+ // Flink's Variant implementation uses BinaryVariant as the standard
+ // Based on FLIP-521
+ if (!(variant instanceof org.apache.flink.types.variant.BinaryVariant)) {
+ throw new IllegalArgumentException(
+ "Expected BinaryVariant but got: " + variant.getClass().getSimpleName());
+ }
+
+ org.apache.flink.types.variant.BinaryVariant binaryVariant =
+ (org.apache.flink.types.variant.BinaryVariant) variant;
+
+ byte[] metadataBytes = binaryVariant.getMetadata();
+ byte[] valueBytes = binaryVariant.getValue();
+
+ org.apache.iceberg.variants.VariantMetadata metadata =
Review Comment:
Could we import the class
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Adding Variant Support to Flink 2.1 [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14259:
URL: https://github.com/apache/iceberg/pull/14259#discussion_r2404947045
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetWriters.java:
##########
@@ -605,4 +621,53 @@ protected Object get(RowData struct, int index) {
return fieldGetter[index].getFieldOrNull(struct);
}
}
+
+ /** Variant writer converts from Flink Variant to Iceberg Variant */
+ public static class VariantWriter implements ParquetValueWriter<Variant> {
+ private final ParquetValueWriter<org.apache.iceberg.variants.Variant> writer;
+
+ private VariantWriter(ParquetValueWriter<?> writer) {
+ this.writer = (ParquetValueWriter<org.apache.iceberg.variants.Variant>) writer;
+ }
+
+ @Override
+ public void write(int repetitionLevel, Variant variant) {
+ // Flink's Variant implementation uses BinaryVariant as the standard
+ // Based on FLIP-521
+ if (!(variant instanceof org.apache.flink.types.variant.BinaryVariant)) {
+ throw new IllegalArgumentException(
+ "Expected BinaryVariant but got: " + variant.getClass().getSimpleName());
+ }
+
+ org.apache.flink.types.variant.BinaryVariant binaryVariant =
+ (org.apache.flink.types.variant.BinaryVariant) variant;
+
+ byte[] metadataBytes = binaryVariant.getMetadata();
+ byte[] valueBytes = binaryVariant.getValue();
+
+ org.apache.iceberg.variants.VariantMetadata metadata =
+ org.apache.iceberg.variants.VariantMetadata.from(
+ ByteBuffer.wrap(metadataBytes).order(java.nio.ByteOrder.LITTLE_ENDIAN));
Review Comment:
Could we import `ByteOrder`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Adding Variant Support to Flink 2.1 [iceberg]
Posted by "aihuaxu (via GitHub)" <gi...@apache.org>.
aihuaxu commented on PR #14259:
URL: https://github.com/apache/iceberg/pull/14259#issuecomment-3492428854
> @aihuaxu When you have chance can you also review conceptually. Am I covering all cases of Iceberg's Variant ?
Sorry that I missed this message. I will take a look. Thanks for working on the support.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Adding Variant Support to Flink 2.1 [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14259:
URL: https://github.com/apache/iceberg/pull/14259#discussion_r2404945701
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetWriters.java:
##########
@@ -605,4 +621,53 @@ protected Object get(RowData struct, int index) {
return fieldGetter[index].getFieldOrNull(struct);
}
}
+
+ /** Variant writer converts from Flink Variant to Iceberg Variant */
+ public static class VariantWriter implements ParquetValueWriter<Variant> {
+ private final ParquetValueWriter<org.apache.iceberg.variants.Variant> writer;
+
+ private VariantWriter(ParquetValueWriter<?> writer) {
+ this.writer = (ParquetValueWriter<org.apache.iceberg.variants.Variant>) writer;
+ }
+
+ @Override
+ public void write(int repetitionLevel, Variant variant) {
+ // Flink's Variant implementation uses BinaryVariant as the standard
+ // Based on FLIP-521
+ if (!(variant instanceof org.apache.flink.types.variant.BinaryVariant)) {
Review Comment:
Could we import the class?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Adding Variant Support to Flink 2.1 [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #14259:
URL: https://github.com/apache/iceberg/pull/14259#issuecomment-3369964158
Thanks @talatuyarer for the PR.
I will need some time to understand the Variants in depth in both Iceberg and in Flink. Maybe @mxm, @stevenzwu or @Guosmilesmile can help too.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Adding Variant Support to Flink 2.1 [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #14259: Flink: Adding Variant Support to Flink 2.1
URL: https://github.com/apache/iceberg/pull/14259
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Adding Variant Support to Flink 2.1 [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on code in PR #14259:
URL: https://github.com/apache/iceberg/pull/14259#discussion_r2511476214
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/data/ParquetWithFlinkSchemaVisitor.java:
##########
@@ -147,6 +149,18 @@ public static <T> T visit(
} finally {
visitor.fieldNames.pop();
}
+ } else if (LogicalTypeAnnotation.variantType(Variant.VARIANT_SPEC_VERSION).equals(annotation)
+ || sType instanceof VariantType) {
Review Comment:
Flink can probably have a different/cleaner logic here, as Flink doesn't produce variant type without the Parquet annotation. Spark may produce variant type without the Parquet annotation, as the Spark integration was added before Parquet logical type release.
Hence I am thinking maybe sth like this
```
} else if (LogicalTypeAnnotation.variantType(Variant.VARIANT_SPEC_VERSION).equals(annotation)) {
...
}
```
@Fokko @aihuaxu what do you think?
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetWriters.java:
##########
@@ -605,4 +627,50 @@ protected Object get(RowData struct, int index) {
return fieldGetter[index].getFieldOrNull(struct);
}
}
+
+ /** Variant writer converts from Flink Variant to Iceberg Variant */
+ public static class VariantWriter implements ParquetValueWriter<Variant> {
+ private final ParquetValueWriter<org.apache.iceberg.variants.Variant> writer;
+
+ private VariantWriter(ParquetValueWriter<?> writer) {
+ this.writer = (ParquetValueWriter<org.apache.iceberg.variants.Variant>) writer;
+ }
+
+ @Override
+ public void write(int repetitionLevel, Variant variant) {
+ // Flink's Variant implementation uses BinaryVariant as the standard
+ // Based on FLIP-521
+ if (!(variant instanceof BinaryVariant)) {
Review Comment:
nit: use `Preconditions.checkArguement`
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/TestHelpers.java:
##########
@@ -378,11 +378,55 @@ private static void assertEquals(
.isInstanceOf(byte[].class)
.isEqualTo(expected);
break;
+ case VARIANT:
+ assertThat(actual)
+ .as("Should expect a Variant")
+ .isInstanceOf(org.apache.flink.types.variant.Variant.class);
+ assertThat(expected)
+ .as("Should expect a Variant")
+ .isInstanceOf(org.apache.iceberg.variants.Variant.class);
+
+ org.apache.flink.types.variant.Variant flinkVariant =
+ (org.apache.flink.types.variant.Variant) actual;
+ org.apache.iceberg.variants.Variant icebergVariant =
+ (org.apache.iceberg.variants.Variant) expected;
+
+ compareVariants(icebergVariant, flinkVariant);
+ break;
default:
throw new IllegalArgumentException("Not a supported type: " + type);
}
}
+ /**
+ * Compare two variants by their underlying data structures rather than string representations.
+ * This method handles the differences between Iceberg and Flink variant implementations.
+ */
+ private static void compareVariants(
+ org.apache.iceberg.variants.Variant icebergVariant,
+ org.apache.flink.types.variant.Variant flinkVariant) {
+
+ // Extract the underlying data from both variants
+ org.apache.iceberg.variants.VariantValue icebergValue = icebergVariant.value();
+ org.apache.iceberg.variants.VariantMetadata icebergMetadata = icebergVariant.metadata();
+
+ // For Flink variant, we expect BinaryVariant which is the standard implementation
+ assertThat(flinkVariant)
+ .as("Flink variant should be a BinaryVariant")
+ .isInstanceOf(org.apache.flink.types.variant.BinaryVariant.class);
+
+ org.apache.flink.types.variant.BinaryVariant binaryVariant =
+ (org.apache.flink.types.variant.BinaryVariant) flinkVariant;
+
+ // Compare the binary data directly
+ byte[] flinkValueBytes = binaryVariant.getValue();
+ byte[] flinkMetadataBytes = binaryVariant.getMetadata();
+
+ // Verify that both variants contain data
+ assertThat(flinkValueBytes).as("Flink variant should contain value bytes").isNotNull();
+ assertThat(flinkMetadataBytes).as("Flink variant should contain metadata bytes").isNotNull();
+ }
Review Comment:
it seems that the comparison is not completed. there is no actual comparison of the byte[]
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkVariants.java:
##########
@@ -0,0 +1,263 @@
+/*
+ * 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.iceberg.flink.data;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.file.Path;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.inmemory.InMemoryOutputFile;
+import org.apache.iceberg.io.FileAppender;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.parquet.Parquet;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.ByteBuffers;
+import org.apache.iceberg.variants.ShreddedObject;
+import org.apache.iceberg.variants.ValueArray;
+import org.apache.iceberg.variants.VariantMetadata;
+import org.apache.iceberg.variants.VariantPrimitive;
+import org.apache.iceberg.variants.VariantTestUtil;
+import org.apache.iceberg.variants.VariantValue;
+import org.apache.iceberg.variants.Variants;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.FieldSource;
+
+public class TestFlinkVariants {
Review Comment:
can we extract a base test class to share the common code btw Flink and Spark variants test?
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetWriters.java:
##########
@@ -605,4 +627,50 @@ protected Object get(RowData struct, int index) {
return fieldGetter[index].getFieldOrNull(struct);
}
}
+
+ /** Variant writer converts from Flink Variant to Iceberg Variant */
+ public static class VariantWriter implements ParquetValueWriter<Variant> {
+ private final ParquetValueWriter<org.apache.iceberg.variants.Variant> writer;
+
+ private VariantWriter(ParquetValueWriter<?> writer) {
+ this.writer = (ParquetValueWriter<org.apache.iceberg.variants.Variant>) writer;
+ }
+
+ @Override
+ public void write(int repetitionLevel, Variant variant) {
+ // Flink's Variant implementation uses BinaryVariant as the standard
+ // Based on FLIP-521
+ if (!(variant instanceof BinaryVariant)) {
+ throw new IllegalArgumentException(
+ "Expected BinaryVariant but got: " + variant.getClass().getSimpleName());
+ }
+
+ BinaryVariant binaryVariant = (BinaryVariant) variant;
+
+ byte[] metadataBytes = binaryVariant.getMetadata();
Review Comment:
nit: no need for these two local variables
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetReaders.java:
##########
@@ -297,6 +301,21 @@ public Optional<ParquetValueReader<?>> visit(
}
}
+ @Override
+ public ParquetVariantVisitor<ParquetValueReader<?>> variantVisitor() {
+ return new VariantReaderBuilder(type, Arrays.asList(currentPath()));
+ }
+
+ @Override
+ public ParquetValueReader<?> variant(
+ Types.VariantType expected, GroupType group, ParquetValueReader<?> result) {
+ if (expected == null) {
Review Comment:
I didn't see similar null check in other places. it this required?
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/TypeToFlinkType.java:
##########
@@ -83,6 +84,11 @@ public LogicalType map(Types.MapType map, LogicalType keyResult, LogicalType val
return new MapType(keyResult.copy(false), valueResult.copy(map.isValueOptional()));
}
+ @Override
+ public LogicalType variant(Types.VariantType variant) {
+ return new VariantType(false); // VariantType is always non-nullable in Flink
Review Comment:
> VariantType is always non-nullable in Flink
Is this correct? I saw plenty of places in Flink where nullable variant type is used.
https://github.com/search?q=repo%3Aapache%2Fflink%20%22new%20VariantType()%22&type=code
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkVariants.java:
##########
@@ -0,0 +1,263 @@
+/*
+ * 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.iceberg.flink.data;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.file.Path;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.inmemory.InMemoryOutputFile;
+import org.apache.iceberg.io.FileAppender;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.parquet.Parquet;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.ByteBuffers;
+import org.apache.iceberg.variants.ShreddedObject;
+import org.apache.iceberg.variants.ValueArray;
+import org.apache.iceberg.variants.VariantMetadata;
+import org.apache.iceberg.variants.VariantPrimitive;
+import org.apache.iceberg.variants.VariantTestUtil;
+import org.apache.iceberg.variants.VariantValue;
+import org.apache.iceberg.variants.Variants;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.FieldSource;
+
+public class TestFlinkVariants {
+
+ @TempDir private Path temp;
+
+ private static final VariantPrimitive<?>[] PRIMITIVES =
+ new VariantPrimitive[] {
+ Variants.ofNull(),
+ Variants.of(true),
+ Variants.of(false),
+ Variants.of((byte) 34),
+ Variants.of((byte) -34),
+ Variants.of((short) 1234),
+ Variants.of((short) -1234),
+ Variants.of(12345),
+ Variants.of(-12345),
+ Variants.of(9876543210L),
+ Variants.of(-9876543210L),
+ Variants.of(10.11F),
+ Variants.of(-10.11F),
+ Variants.of(14.3D),
+ Variants.of(-14.3D),
+ Variants.ofIsoDate("2024-11-07"),
+ Variants.ofIsoDate("1957-11-07"),
+ Variants.ofIsoTimestamptz("2024-11-07T12:33:54.123456+00:00"),
+ Variants.ofIsoTimestamptz("1957-11-07T12:33:54.123456+00:00"),
+ Variants.ofIsoTimestampntz("2024-11-07T12:33:54.123456"),
+ Variants.ofIsoTimestampntz("1957-11-07T12:33:54.123456"),
+ Variants.of(new BigDecimal("12345.6789")), // decimal4
+ Variants.of(new BigDecimal("-12345.6789")), // decimal4
+ Variants.of(new BigDecimal("123456789.987654321")), // decimal8
+ Variants.of(new BigDecimal("-123456789.987654321")), // decimal8
+ Variants.of(new BigDecimal("9876543210.123456789")), // decimal16
+ Variants.of(new BigDecimal("-9876543210.123456789")), // decimal16
+ Variants.of(ByteBuffer.wrap(new byte[] {0x0a, 0x0b, 0x0c, 0x0d})),
+ Variants.of("iceberg"),
+ Variants.ofUUID("f24f9b64-81fa-49d1-b74e-8c09a6e31c56"),
+ };
+
+ private static final VariantPrimitive<?>[] UNSUPPORTED_PRIMITIVES =
+ new VariantPrimitive[] {
+ Variants.ofIsoTime("12:33:54.123456"),
+ Variants.ofIsoTimestamptzNanos("2024-11-07T12:33:54.123456789+00:00"),
+ Variants.ofIsoTimestampntzNanos("2024-11-07T12:33:54.123456789"),
+ };
+
+ @Test
+ public void testIcebergVariantTypeToFlinkVariantType() {
+ // Test that Iceberg's VariantType converts to Flink's VariantType
+ Types.VariantType icebergVariantType = Types.VariantType.get();
+ LogicalType flinkVariantType = FlinkSchemaUtil.convert(icebergVariantType);
+
+ assertThat(flinkVariantType)
+ .isInstanceOf(org.apache.flink.table.types.logical.VariantType.class);
+ }
+
+ @Test
+ public void testFlinkVariantTypeToIcebergVariantType() {
+ org.apache.flink.table.types.logical.VariantType flinkVariantType =
+ new org.apache.flink.table.types.logical.VariantType(false);
+ Types.VariantType icebergVariantType =
+ (Types.VariantType) FlinkSchemaUtil.convert(flinkVariantType);
+
+ assertThat(icebergVariantType).isEqualTo(Types.VariantType.get());
Review Comment:
the line above already did type cast. this assertion is not useful. maybe the line above should just use the base `Type`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Adding Variant Support to Flink 2.1 [iceberg]
Posted by "aihuaxu (via GitHub)" <gi...@apache.org>.
aihuaxu commented on PR #14259:
URL: https://github.com/apache/iceberg/pull/14259#issuecomment-3492502943
@talatuyarer Overall the logic makes sense to me in which we are relying on the existing Parquet variant reader/writer to handle most of the work including shredding and what we need is mostly to convert between Flink Variant and Iceberg Variant.
As I read, shredding should work automatically. Can we add some basic validation for shredding coverage to make sure the Parquet schema is as expected?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Adding Variant Support to Flink 2.1 [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #14259:
URL: https://github.com/apache/iceberg/pull/14259#issuecomment-3672841535
This pull request has been closed due to lack of activity. This is not a judgement on the merit of the PR in any way. It is just a way of keeping the PR queue manageable. If you think that is incorrect, or the pull request requires review, you can revive the PR at any time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Adding Variant Support to Flink 2.1 [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #14259:
URL: https://github.com/apache/iceberg/pull/14259#issuecomment-3639473873
This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Adding Variant Support to Flink 2.1 [iceberg]
Posted by "talatuyarer (via GitHub)" <gi...@apache.org>.
talatuyarer commented on PR #14259:
URL: https://github.com/apache/iceberg/pull/14259#issuecomment-3370008132
Thank you @pvary
This was useful for me: https://cwiki.apache.org/confluence/display/FLINK/FLIP-521%3A+Integrating+Variant+Type+into+Flink%3A+Enabling+Efficient+Semi-Structured+Data+Processing
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Adding Variant Support to Flink 2.1 [iceberg]
Posted by "talatuyarer (via GitHub)" <gi...@apache.org>.
talatuyarer commented on PR #14259:
URL: https://github.com/apache/iceberg/pull/14259#issuecomment-3370068039
@aihuaxu When you have chance can you also review conceptually. Am I covering all cases of Iceberg's Variant ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org