This is an automated email from the ASF dual-hosted git repository.
mxm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new f74aea1e68 Flink: Support Variant type in DynamicIcebergSink (#17900)
f74aea1e68 is described below
commit f74aea1e68fc8161905a748f069c055f47ca64b5
Author: fmorillo7694 <[email protected]>
AuthorDate: Tue Oct 6 16:46:50 2026 +0200
Flink: Support Variant type in DynamicIcebergSink (#17900)
---
.../flink/sink/dynamic/CompareSchemasVisitor.java | 14 ++
.../iceberg/flink/sink/dynamic/DataConverter.java | 1 +
.../flink/sink/dynamic/EvolveSchemaVisitor.java | 13 ++
.../sink/dynamic/TestCompareSchemasVisitor.java | 45 ++++++
.../flink/sink/dynamic/TestDynamicIcebergSink.java | 180 +++++++++++++++++++++
.../sink/dynamic/TestEvolveSchemaVisitor.java | 41 +++++
.../flink/sink/dynamic/TestRowDataConverter.java | 27 ++++
7 files changed, 321 insertions(+)
diff --git
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java
index 6c1b2e5673..a975b9d371 100644
---
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java
+++
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java
@@ -181,6 +181,20 @@ public class CompareSchemasVisitor
}
}
+ @Override
+ public Result variant(Types.VariantType variant, Integer tableSchemaId) {
+ if (tableSchemaId == null) {
+ return Result.SCHEMA_UPDATE_NEEDED;
+ }
+
+ Type tableSchemaType = tableSchema.findField(tableSchemaId).type();
+ if (tableSchemaType.isVariantType()) {
+ return Result.SAME;
+ } else {
+ return Result.SCHEMA_UPDATE_NEEDED;
+ }
+ }
+
/**
* Whether {@link DataConverter} can convert input data of type {@code
dataType} into the table's
* {@code tableType}. Must stay in sync with the conversions {@link
DataConverter#get} performs.
diff --git
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DataConverter.java
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DataConverter.java
index b3fe35476d..c3aa5031c5 100644
---
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DataConverter.java
+++
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DataConverter.java
@@ -80,6 +80,7 @@ interface DataConverter {
case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
case BINARY:
case VARBINARY:
+ case VARIANT:
return object -> object;
case DOUBLE:
return object -> {
diff --git
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java
index 3bead03d51..15dd11b7ff 100644
---
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java
+++
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java
@@ -194,6 +194,11 @@ public class EvolveSchemaVisitor extends
SchemaWithPartnerVisitor<Integer, Boole
return partnerId == null;
}
+ @Override
+ public Boolean variant(Types.VariantType variant, Integer partnerId) {
+ return partnerId == null;
+ }
+
private Type findFieldType(int fieldId) {
if (fieldId == -1) {
return existingSchema.asStruct();
@@ -210,6 +215,14 @@ public class EvolveSchemaVisitor extends
SchemaWithPartnerVisitor<Integer, Boole
private void updateColumn(Types.NestedField existingField, Types.NestedField
targetField) {
String existingColumnName =
this.existingSchema.findColumnName(existingField.fieldId());
+ // A column cannot be promoted to variant; fail instead of silently
skipping the update
+ Preconditions.checkArgument(
+ !targetField.type().isVariantType() ||
existingField.type().isVariantType(),
+ "Cannot change column type: %s: %s -> %s",
+ existingColumnName,
+ existingField.type(),
+ targetField.type());
+
boolean needsOptionalUpdate = targetField.isOptional() &&
existingField.isRequired();
boolean handledByDataConversion =
CompareSchemasVisitor.isDataConversionPossible(targetField.type(),
existingField.type());
diff --git
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestCompareSchemasVisitor.java
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestCompareSchemasVisitor.java
index 9e4d600f93..1cdcf8fd03 100644
---
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestCompareSchemasVisitor.java
+++
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestCompareSchemasVisitor.java
@@ -29,6 +29,7 @@ import org.apache.iceberg.types.Types.LongType;
import org.apache.iceberg.types.Types.MapType;
import org.apache.iceberg.types.Types.StringType;
import org.apache.iceberg.types.Types.StructType;
+import org.apache.iceberg.types.Types.VariantType;
import org.junit.jupiter.api.Test;
class TestCompareSchemasVisitor {
@@ -56,6 +57,50 @@ class TestCompareSchemasVisitor {
.isEqualTo(CompareSchemasVisitor.Result.SAME);
}
+ @Test
+ void testSchemaWithVariantSame() {
+ assertThat(
+ CompareSchemasVisitor.visit(
+ new Schema(
+ optional(1, "id", IntegerType.get()),
+ optional(2, "data", StringType.get()),
+ optional(3, "payload", VariantType.get())),
+ new Schema(
+ optional(1, "id", IntegerType.get()),
+ optional(2, "data", StringType.get()),
+ optional(3, "payload", VariantType.get())),
+ CASE_SENSITIVE,
+ PRESERVE_COLUMNS))
+ .isEqualTo(CompareSchemasVisitor.Result.SAME);
+ }
+
+ @Test
+ void testVariantAgainstNonVariantTableField() {
+ assertThat(
+ CompareSchemasVisitor.visit(
+ new Schema(
+ optional(1, "id", IntegerType.get()),
+ optional(2, "payload", VariantType.get())),
+ new Schema(
+ optional(1, "id", IntegerType.get()), optional(2,
"payload", StringType.get())),
+ CASE_SENSITIVE,
+ PRESERVE_COLUMNS))
+ .isEqualTo(CompareSchemasVisitor.Result.SCHEMA_UPDATE_NEEDED);
+ }
+
+ @Test
+ void testVariantFieldMissingFromTable() {
+ assertThat(
+ CompareSchemasVisitor.visit(
+ new Schema(
+ optional(1, "id", IntegerType.get()),
+ optional(2, "payload", VariantType.get())),
+ new Schema(optional(1, "id", IntegerType.get())),
+ CASE_SENSITIVE,
+ PRESERVE_COLUMNS))
+ .isEqualTo(CompareSchemasVisitor.Result.SCHEMA_UPDATE_NEEDED);
+ }
+
@Test
void testSchemaDifferentId() {
assertThat(
diff --git
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
index a426529d32..2abcf9ce01 100644
---
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
+++
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
@@ -100,7 +100,9 @@ import
org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.Types;
+import org.apache.iceberg.variants.Variant;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
@@ -692,6 +694,184 @@ class TestDynamicIcebergSink extends
TestFlinkIcebergSinkBase {
}
}
+ @Test
+ void testWriteVariant() throws Exception {
+ Schema schema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "payload", Types.VariantType.get()));
+ createV3Table("t1", schema);
+
+ executeVariantSink(
+ Lists.newArrayList(new VariantInput(schema, 1), new
VariantInput(schema, 2)), this.env);
+
+ assertVariantPayloads(ImmutableMap.of(1, 1L, 2, 2L));
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testSchemaEvolutionAddVariantField(boolean immediateUpdate) throws
Exception {
+ Schema variantSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "data", Types.StringType.get()),
+ Types.NestedField.optional(3, "payload", Types.VariantType.get()));
+ createV3Table("t1", SimpleDataUtil.SCHEMA);
+
+ executeVariantSink(
+ Lists.newArrayList(
+ new VariantInput(SimpleDataUtil.SCHEMA, 1), new
VariantInput(variantSchema, 2)),
+ this.env,
+ immediateUpdate);
+
+ Table table =
CATALOG_EXTENSION.catalog().loadTable(TableIdentifier.of(DATABASE, "t1"));
+
assertThat(table.schema().findField("payload").type()).isEqualTo(Types.VariantType.get());
+ Map<Integer, Long> expected = Maps.newHashMap();
+ expected.put(1, null);
+ expected.put(2, 2L);
+ assertVariantPayloads(expected);
+ }
+
+ @Test
+ void testSchemaEvolutionStringToVariantFails() throws Exception {
+ assertIncompatibleVariantEvolution(
+ Types.StringType.get(),
+ Types.VariantType.get(),
+ "Cannot change column type: payload: string -> variant");
+ }
+
+ @Test
+ void testSchemaEvolutionVariantToStringFails() throws Exception {
+ assertIncompatibleVariantEvolution(
+ Types.VariantType.get(),
+ Types.StringType.get(),
+ "Cannot change column type: payload: variant -> string");
+ }
+
+ private void assertIncompatibleVariantEvolution(
+ Type tableType, Type writeType, String expectedMessage) throws Exception
{
+ Schema tableSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "payload", tableType));
+ Schema writeSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "payload", writeType));
+ createV3Table("t1", tableSchema);
+
+ try {
+ executeVariantSink(
+ Lists.newArrayList(new VariantInput(writeSchema, 1)),
+ StreamExecutionEnvironment.getExecutionEnvironment());
+ fail();
+ } catch (JobExecutionException e) {
+ assertThat(
+ ExceptionUtils.findThrowable(
+ e, t -> t.getMessage() != null &&
t.getMessage().contains(expectedMessage)))
+ .isNotEmpty();
+ }
+ }
+
+ private static void createV3Table(String name, Schema schema) {
+ CATALOG_EXTENSION
+ .catalog()
+ .createTable(
+ TableIdentifier.of(DATABASE, name),
+ schema,
+ PartitionSpec.unpartitioned(),
+ null,
+ ImmutableMap.of(TableProperties.FORMAT_VERSION, "3"));
+ }
+
+ private static void executeVariantSink(List<VariantInput> inputs,
StreamExecutionEnvironment env)
+ throws Exception {
+ executeVariantSink(inputs, env, true);
+ }
+
+ private static void executeVariantSink(
+ List<VariantInput> inputs, StreamExecutionEnvironment env, boolean
immediateUpdate)
+ throws Exception {
+ DataStream<VariantInput> dataStream =
+ env.fromData(inputs, TypeInformation.of(new TypeHint<>() {}));
+ env.setParallelism(1);
+ DynamicIcebergSink.forInput(dataStream)
+ .generator(new VariantGenerator())
+ .catalogLoader(CATALOG_EXTENSION.catalogLoader())
+ .writeParallelism(1)
+ .immediateTableUpdate(immediateUpdate)
+ .append();
+ env.execute("Test Iceberg Variant DataStream");
+ }
+
+ /** Reads table t1 and checks the "k" field of each row's variant payload,
keyed by row id. */
+ private static void assertVariantPayloads(Map<Integer, Long> expectedById)
throws IOException {
+ Map<Integer, Long> actualById = Maps.newHashMap();
+ try (CloseableIterable<Record> records =
+ IcebergGenerics.read(
+
CATALOG_EXTENSION.catalog().loadTable(TableIdentifier.of(DATABASE, "t1")))
+ .build()) {
+ for (Record record : records) {
+ Variant payload = (Variant) record.getField("payload");
+ actualById.put(
+ (Integer) record.getField("id"),
+ payload == null
+ ? null
+ : (Long)
payload.value().asObject().get("k").asPrimitive().get());
+ }
+ }
+
+ assertThat(actualById).isEqualTo(expectedById);
+ }
+
+ /** Input for the variant tests; the row is built in {@link
VariantGenerator}. */
+ private static class VariantInput implements Serializable {
+ private final Schema schema;
+ private final int id;
+
+ VariantInput(Schema schema, int id) {
+ this.schema = schema;
+ this.id = id;
+ }
+ }
+
+ /**
+ * Builds RowData directly, because the legacy DataFormatConverters used by
{@link Generator} do
+ * not support VARIANT. Variant payloads are {"k": id}.
+ */
+ private static class VariantGenerator implements
DynamicRecordGenerator<VariantInput> {
+ @Override
+ public void generate(VariantInput input, Collector<DynamicRecord> out) {
+ List<Types.NestedField> columns = input.schema.columns();
+ GenericRowData row = new GenericRowData(columns.size());
+ for (int i = 0; i < columns.size(); i++) {
+ Types.NestedField column = columns.get(i);
+ if (column.type().isVariantType()) {
+ row.setField(
+ i,
+ org.apache.flink.types.variant.Variant.newBuilder()
+ .object()
+ .add("k",
org.apache.flink.types.variant.Variant.newBuilder().of((long) input.id))
+ .build());
+ } else if (column.type().typeId() == Type.TypeID.INTEGER) {
+ row.setField(i, input.id);
+ } else {
+ row.setField(i, StringData.fromString(column.name() + "-" +
input.id));
+ }
+ }
+
+ out.collect(
+ new DynamicRecord(
+ TableIdentifier.of(DATABASE, "t1"),
+ SnapshotRef.MAIN_BRANCH,
+ input.schema,
+ row,
+ PartitionSpec.unpartitioned(),
+ DistributionMode.NONE,
+ 1));
+ }
+ }
+
@Test
void testPartitionSpecEvolution() throws Exception {
PartitionSpec spec1 =
PartitionSpec.builderFor(SimpleDataUtil.SCHEMA).bucket("id", 10).build();
diff --git
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestEvolveSchemaVisitor.java
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestEvolveSchemaVisitor.java
index 55e52751fe..f08f745768 100644
---
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestEvolveSchemaVisitor.java
+++
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestEvolveSchemaVisitor.java
@@ -103,6 +103,47 @@ public class TestEvolveSchemaVisitor {
assertThat(targetSchema.asStruct()).isEqualTo(updateApi.apply().asStruct());
}
+ @Test
+ public void testAddTopLevelVariant() {
+ Schema targetSchema =
+ new Schema(
+ optional(1, "id", Types.IntegerType.get()),
+ optional(2, "payload", Types.VariantType.get()));
+ UpdateSchema updateApi = loadUpdateApi(new Schema());
+ EvolveSchemaVisitor.visit(
+ TABLE, updateApi, new Schema(), targetSchema, CASE_SENSITIVE,
PRESERVE_COLUMNS);
+
assertThat(targetSchema.asStruct()).isEqualTo(updateApi.apply().asStruct());
+ }
+
+ @Test
+ public void testExistingVariantFieldIsUnchanged() {
+ Schema existingSchema =
+ new Schema(
+ optional(1, "id", Types.IntegerType.get()),
+ optional(2, "payload", Types.VariantType.get()));
+ UpdateSchema updateApi = loadUpdateApi(existingSchema);
+ EvolveSchemaVisitor.visit(
+ TABLE, updateApi, existingSchema, existingSchema, CASE_SENSITIVE,
PRESERVE_COLUMNS);
+
assertThat(existingSchema.asStruct()).isEqualTo(updateApi.apply().asStruct());
+ }
+
+ @Test
+ public void testPrimitiveToVariantFails() {
+ Schema currentSchema = new Schema(optional(1, "payload",
StringType.get()));
+ Schema targetSchema = new Schema(optional(1, "payload",
Types.VariantType.get()));
+ assertThatThrownBy(
+ () ->
+ EvolveSchemaVisitor.visit(
+ TABLE,
+ loadUpdateApi(currentSchema),
+ currentSchema,
+ targetSchema,
+ CASE_SENSITIVE,
+ PRESERVE_COLUMNS))
+ .hasMessage("Cannot change column type: payload: string -> variant")
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
@Test
public void testMakeTopLevelPrimitivesOptional() {
Schema existingSchema = new Schema(primitiveFields(0, primitiveTypes(),
false));
diff --git
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestRowDataConverter.java
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestRowDataConverter.java
index 19e08652fb..e7efa13b44 100644
---
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestRowDataConverter.java
+++
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestRowDataConverter.java
@@ -34,6 +34,7 @@ import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
import org.apache.flink.table.data.TimestampData;
import org.apache.flink.types.RowKind;
+import org.apache.flink.types.variant.Variant;
import org.apache.iceberg.Schema;
import org.apache.iceberg.flink.DataGenerator;
import org.apache.iceberg.flink.DataGenerators;
@@ -85,6 +86,32 @@ class TestRowDataConverter {
assertThat(convert(deleteRow, SCHEMA,
SCHEMA2).getRowKind()).isEqualTo(RowKind.DELETE);
}
+ @Test
+ void testVariantWithDataConversion() {
+ Schema sourceSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "payload", Types.VariantType.get()));
+ Schema targetSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "payload", Types.VariantType.get()));
+ assertThat(
+ CompareSchemasVisitor.visit(
+ sourceSchema,
+ targetSchema,
+ true /* caseSensitive */,
+ false /* dropUnusedColumns */))
+ .isEqualTo(CompareSchemasVisitor.Result.DATA_CONVERSION_NEEDED);
+
+ Variant variant = Variant.newBuilder().object().add("k",
Variant.newBuilder().of(1L)).build();
+ RowData row = GenericRowData.of(42, variant);
+
+ RowData converted = convert(row, sourceSchema, targetSchema);
+ assertThat(converted.getLong(0)).isEqualTo(42L);
+ assertThat(converted.getVariant(1)).isSameAs(variant);
+ }
+
@Test
void testAddRequiredColumn() {
Schema currentSchema = new Schema(Types.NestedField.optional(1, "id",
Types.IntegerType.get()));
