From feb4e6449bcbbc6e758e9034f261781e6a509447 Mon Sep 17 00:00:00 2001 From: Victor Babenko <37556649+vbabenkoru@users.noreply.github.com> Date: Sat, 15 Aug 2026 22:01:44 -0700 Subject: [PATCH] [iceberg] Allow publishing VARIANT columns with format version 3 --- .../paimon/iceberg/IcebergCommitCallback.java | 14 +++--- .../IcebergRowLineageCompatibilityTest.java | 48 +++++++++++++++++++ 2 files changed, 56 insertions(+), 6 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 f8a41f56f28e..ea6d256a4efb 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 @@ -733,7 +733,7 @@ private List getPartitionFields( return result; } - /** VARIANT needs Iceberg row lineage, which Paimon Iceberg compatibility cannot publish. */ + /** VARIANT is an Iceberg format-version-3 type; reject publishing it into v2 metadata. */ static void checkVariantNotPublishable(RowType rowType) { Collection variantFields = new LinkedHashSet<>(); for (DataField field : rowType.getFields()) { @@ -741,9 +741,8 @@ static void checkVariantNotPublishable(RowType rowType) { } 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.", + "Columns %s use the VARIANT type, which requires Iceberg format version 3. " + + "Set 'metadata.iceberg.format-version' = '3' to publish this table.", variantFields); } @@ -2136,8 +2135,11 @@ private IcebergSchema get(long schemaId) { schemaId, id -> { TableSchema schema = schemaManager.schema(id); - // backstop: reject variant on each schema as it is emitted - checkVariantNotPublishable(schema.logicalRowType()); + if (formatVersion < IcebergMetadata.FORMAT_VERSION_V3) { + // VARIANT is an Iceberg format-version-3 type; v2 metadata cannot + // represent it + checkVariantNotPublishable(schema.logicalRowType()); + } SchemaValidation.validateIcebergGeospatialTypes( schema.logicalRowType(), table.coreOptions()); return IcebergSchema.create(schema); diff --git a/paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergRowLineageCompatibilityTest.java b/paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergRowLineageCompatibilityTest.java index 55a1eacb665d..b10cb345d19d 100644 --- a/paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergRowLineageCompatibilityTest.java +++ b/paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergRowLineageCompatibilityTest.java @@ -23,6 +23,7 @@ import org.apache.paimon.catalog.Identifier; import org.apache.paimon.data.BinaryRow; import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.variant.GenericVariant; import org.apache.paimon.disk.IOManagerImpl; import org.apache.paimon.fs.Path; import org.apache.paimon.fs.SeekableInputStream; @@ -76,6 +77,7 @@ import java.util.UUID; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Tests for Iceberg format-version 3 row-lineage metadata fields. */ public class IcebergRowLineageCompatibilityTest { @@ -234,6 +236,52 @@ public void testTagPreservesNextRowId() throws Exception { assertThat(metadata.nextRowId()).isEqualTo(3L); } + @Test + public void testVariantPublishableWithFormatVersion3() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.VARIANT()}, + new String[] {"k", "payload"}); + FileStoreTable table = createPaimonTable(rowType, formatVersionOptions(3), "parquet"); + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = + table.newWrite(commitUser) + .withIOManager(new IOManagerImpl(tempDir.toString() + "/tmp")); + TableCommitImpl commit = table.newCommit(commitUser); + + write.write(GenericRow.of(1, GenericVariant.fromJson("{\"a\": 1}"))); + commit.commit(1, write.prepareCommit(false, 1)); + write.close(); + commit.close(); + + IcebergMetadata metadata = readIcebergMetadata(table, 1); + assertThat(metadata.nextRowId()).isEqualTo(1L); + assertThat(metadata.schemas().get(metadata.currentSchemaId()).fields().get(1).type()) + .isEqualTo("variant"); + } + + @Test + public void testVariantRejectedWithFormatVersion2() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.VARIANT()}, + new String[] {"k", "payload"}); + FileStoreTable table = createPaimonTable(rowType, formatVersionOptions(2), "parquet"); + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = + table.newWrite(commitUser) + .withIOManager(new IOManagerImpl(tempDir.toString() + "/tmp")); + TableCommitImpl commit = table.newCommit(commitUser); + + write.write(GenericRow.of(1, GenericVariant.fromJson("{\"a\": 1}"))); + // hasStackTraceContaining: robust whether or not the commit path wraps the + // IllegalArgumentException from the guard + assertThatThrownBy(() -> commit.commit(1, write.prepareCommit(false, 1))) + .hasStackTraceContaining("VARIANT"); + write.close(); + commit.close(); + } + @Test public void testManifestListCarriesFirstRowIdColumn() throws Exception { FileStoreTable table = createPaimonTable(defaultRowType(), formatVersionOptions(3), "avro");