From e1e82d93d2ee398ee3a259e0d35a2e7037a7c194 Mon Sep 17 00:00:00 2001 From: Jiajia Li Date: Thu, 23 Jul 2026 08:23:24 -0400 Subject: [PATCH] [iceberg] Support variant type in schema conversion; reject it in Paimon Iceberg compatibility --- .../paimon/iceberg/IcebergCommitCallback.java | 57 ++++++++++++++++++- .../iceberg/metadata/IcebergDataField.java | 7 ++- .../iceberg/IcebergCompatibilityTest.java | 13 +++++ .../metadata/IcebergDataFieldTest.java | 51 +++++++++++++++++ .../apache/paimon/rest/RESTCatalogTest.java | 5 +- 5 files changed, 130 insertions(+), 3 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java index df82c7dccccb..d8fa8457db27 100644 --- a/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java +++ b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java @@ -51,6 +51,7 @@ import org.apache.paimon.options.Options; import org.apache.paimon.partition.PartitionPredicate; import org.apache.paimon.schema.SchemaManager; +import org.apache.paimon.schema.TableSchema; import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.table.sink.CommitCallback; import org.apache.paimon.table.sink.TagCallback; @@ -60,7 +61,11 @@ import org.apache.paimon.table.source.ScanMode; import org.apache.paimon.table.source.snapshot.SnapshotReader; import org.apache.paimon.tag.Tag; +import org.apache.paimon.types.ArrayType; +import org.apache.paimon.types.DataField; import org.apache.paimon.types.DataType; +import org.apache.paimon.types.MapType; +import org.apache.paimon.types.MultisetType; import org.apache.paimon.types.RowType; import org.apache.paimon.utils.DataFilePathFactories; import org.apache.paimon.utils.FileStorePathFactory; @@ -77,6 +82,7 @@ import java.io.IOException; import java.io.UncheckedIOException; import java.util.ArrayList; +import java.util.Collection; import java.util.Collections; import java.util.Comparator; import java.util.HashMap; @@ -547,6 +553,49 @@ private List getPartitionFields( return result; } + /** VARIANT needs Iceberg row lineage, which Paimon Iceberg compatibility cannot publish. */ + static void checkVariantNotPublishable(RowType rowType) { + Collection variantFields = new LinkedHashSet<>(); + for (DataField field : rowType.getFields()) { + collectVariantFields(field.name(), field.type(), variantFields); + } + Preconditions.checkArgument( + variantFields.isEmpty(), + "Columns %s use the VARIANT type, which Paimon Iceberg compatibility cannot " + + "publish: it is an Iceberg format-version-3 type that requires row " + + "lineage.", + variantFields); + } + + private static void collectVariantFields( + String path, DataType type, Collection variantFields) { + switch (type.getTypeRoot()) { + case VARIANT: + variantFields.add(path + ": " + type.asSQLString()); + break; + case ARRAY: + collectVariantFields( + path + ".element", ((ArrayType) type).getElementType(), variantFields); + break; + case MULTISET: + collectVariantFields( + path + ".element", ((MultisetType) type).getElementType(), variantFields); + break; + case MAP: + collectVariantFields(path + ".key", ((MapType) type).getKeyType(), variantFields); + collectVariantFields( + path + ".value", ((MapType) type).getValueType(), variantFields); + break; + case ROW: + for (DataField field : ((RowType) type).getFields()) { + collectVariantFields(path + "." + field.name(), field.type(), variantFields); + } + break; + default: + break; + } + } + // ------------------------------------------------------------------------------------- // Create metadata based on old ones // ------------------------------------------------------------------------------------- @@ -1512,7 +1561,13 @@ private class SchemaCache { private IcebergSchema get(long schemaId) { return schemas.computeIfAbsent( - schemaId, id -> IcebergSchema.create(schemaManager.schema(id))); + schemaId, + id -> { + TableSchema schema = schemaManager.schema(id); + // backstop: reject variant on each schema as it is emitted + checkVariantNotPublishable(schema.logicalRowType()); + return IcebergSchema.create(schema); + }); } private long getLatestSchemaId() { diff --git a/paimon-core/src/main/java/org/apache/paimon/iceberg/metadata/IcebergDataField.java b/paimon-core/src/main/java/org/apache/paimon/iceberg/metadata/IcebergDataField.java index 95bbed9982aa..9862ff7f90c4 100644 --- a/paimon-core/src/main/java/org/apache/paimon/iceberg/metadata/IcebergDataField.java +++ b/paimon-core/src/main/java/org/apache/paimon/iceberg/metadata/IcebergDataField.java @@ -38,6 +38,7 @@ import org.apache.paimon.types.TimestampType; import org.apache.paimon.types.VarBinaryType; import org.apache.paimon.types.VarCharType; +import org.apache.paimon.types.VariantType; import org.apache.paimon.utils.Preconditions; import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator; @@ -193,11 +194,13 @@ private static Object toTypeObject(DataType dataType, int fieldId, int depth) { timestampLtzPrecision >= 3 && timestampLtzPrecision <= 9, "Paimon Iceberg compatibility only support timestamp type with precision from 3 to 9."); return timestampLtzPrecision >= 7 ? "timestamptz_ns" : "timestamptz"; + case VARIANT: + return "variant"; case ARRAY: ArrayType arrayType = (ArrayType) dataType; return new IcebergListType( SpecialFields.getArrayElementFieldId(fieldId, depth + 1), - !dataType.isNullable(), + !arrayType.getElementType().isNullable(), toTypeObject(arrayType.getElementType(), fieldId, depth + 1)); case MAP: MapType mapType = (MapType) dataType; @@ -285,6 +288,8 @@ private DataType getDataTypeFromType(Object icebergType, boolean isRequired) { return new TimestampType(!isRequired, 9); case "timestamptz_ns": // iceberg v3 format return new LocalZonedTimestampType(!isRequired, 9); + case "variant": // iceberg v3 format + return new VariantType(!isRequired); default: throw new UnsupportedOperationException( "Unsupported primitive data type: " + icebergType); diff --git a/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java b/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java index ceb0b906957c..c8a1b3fa4c99 100644 --- a/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java @@ -1790,4 +1790,17 @@ private void parseAvroFields( } } } + + @Test + public void testVariantIsNotPublishableToIceberg() { + RowType withVariant = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.VARIANT()}, + new String[] {"k", "payload"}); + assertThatThrownBy(() -> IcebergCommitCallback.checkVariantNotPublishable(withVariant)) + .hasMessageContaining("VARIANT type"); + + IcebergCommitCallback.checkVariantNotPublishable( + RowType.of(new DataType[] {DataTypes.INT()}, new String[] {"k"})); + } } diff --git a/paimon-core/src/test/java/org/apache/paimon/iceberg/metadata/IcebergDataFieldTest.java b/paimon-core/src/test/java/org/apache/paimon/iceberg/metadata/IcebergDataFieldTest.java index 926a83b24e4a..0cfd5f0fe5fc 100644 --- a/paimon-core/src/test/java/org/apache/paimon/iceberg/metadata/IcebergDataFieldTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/iceberg/metadata/IcebergDataFieldTest.java @@ -37,6 +37,7 @@ import org.apache.paimon.types.TimestampType; import org.apache.paimon.types.VarBinaryType; import org.apache.paimon.types.VarCharType; +import org.apache.paimon.types.VariantType; import org.apache.paimon.utils.JsonSerdeUtil; import org.junit.jupiter.api.DisplayName; @@ -269,6 +270,19 @@ void testArrayTypeConversion() { assertThat(listType.elementRequired()).isTrue(); } + @Test + @DisplayName("Test array element-required follows the element, not the array") + void testArrayElementNullabilityIndependentOfArray() { + DataField arrayField = + new DataField(1, "array", new ArrayType(true, new VariantType(false))); + IcebergDataField icebergArray = new IcebergDataField(arrayField); + + IcebergListType listType = (IcebergListType) icebergArray.type(); + assertThat(listType.element()).isEqualTo("variant"); + assertThat(listType.elementRequired()).isTrue(); + assertThat(icebergArray.dataType()).isEqualTo(new ArrayType(true, new VariantType(false))); + } + @Test @DisplayName("Test map type conversion") void testMapTypeConversion() { @@ -461,6 +475,43 @@ void testTimestampTypeParsing() { assertThat(timestamptzNsField.dataType()).isEqualTo(new LocalZonedTimestampType(true, 9)); } + @Test + @DisplayName("Test variant type conversion") + void testVariantTypeConversion() { + DataField variantField = new DataField(1, "variant", new VariantType(true)); + IcebergDataField icebergVariant = new IcebergDataField(variantField); + + assertThat(icebergVariant.type()).isEqualTo("variant"); + assertThat(icebergVariant.required()).isFalse(); + assertThat(icebergVariant.dataType()).isEqualTo(new VariantType(true)); + } + + @Test + @DisplayName("Test variant type parsing") + void testVariantTypeParsing() { + IcebergDataField optionalVariant = + new IcebergDataField(1, "variant", false, "variant", null, "doc"); + assertThat(optionalVariant.dataType()).isEqualTo(new VariantType(true)); + + IcebergDataField requiredVariant = + new IcebergDataField(2, "variant", true, "variant", null, "doc"); + assertThat(requiredVariant.dataType()).isEqualTo(new VariantType(false)); + } + + @Test + @DisplayName("Test variant type serialization round trip") + void testVariantTypeSerializationRoundTrip() { + DataField variantField = new DataField(1, "variant", new VariantType(true)); + IcebergDataField originalField = new IcebergDataField(variantField); + + String json = JsonSerdeUtil.toJson(originalField); + IcebergDataField deserializedField = JsonSerdeUtil.fromJson(json, IcebergDataField.class); + + assertThat(deserializedField.type()).isEqualTo("variant"); + assertThat(deserializedField.dataType()).isEqualTo(new VariantType(true)); + assertThat(deserializedField.toDatafield()).isEqualTo(variantField); + } + @Test @DisplayName("Test unsupported primitive type parsing") void testUnsupportedPrimitiveTypeParsing() { diff --git a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java index 05376d3585b5..1438b49f17fe 100644 --- a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java @@ -3294,7 +3294,8 @@ public void testCreateIcebergTable() throws Exception { Lists.newArrayList( new DataField(0, "pt", DataTypes.INT()), new DataField(1, "col1", DataTypes.STRING()), - new DataField(2, "col2", DataTypes.STRING())), + new DataField(2, "col2", DataTypes.STRING()), + new DataField(3, "payload", DataTypes.VARIANT())), Collections.singletonList("pt"), Collections.emptyList(), options, @@ -3310,6 +3311,8 @@ public void testCreateIcebergTable() throws Exception { assertThat(tables).containsExactlyInAnyOrder("table1"); assertThat(table.uuid()).isNotEmpty(); assertThat(table.uuid()).isNotEqualTo(table.fullName()); + assertThat(table.rowType().getField("payload").type()) + .isInstanceOf(org.apache.paimon.types.VariantType.class); } @Test