This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 5371a4f22f [iceberg] Allow publishing VARIANT columns with Iceberg
format version 3 (#9246)
5371a4f22f is described below
commit 5371a4f22fcc7875abbcacfffc913ac4a8e3964a
Author: Victor Babenko <[email protected]>
AuthorDate: Sat Aug 29 19:04:33 2026 -0700
[iceberg] Allow publishing VARIANT columns with Iceberg format version 3
(#9246)
---
.../paimon/iceberg/IcebergCommitCallback.java | 14 ++++---
.../core/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 d4c25a734e..5b0a78b828 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
@@ -735,7 +735,7 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
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<String> variantFields = new LinkedHashSet<>();
for (DataField field : rowType.getFields()) {
@@ -743,9 +743,8 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
}
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);
}
@@ -2145,8 +2144,11 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
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());
SchemaValidation.validateIcebergTimestampPrecisions(
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 55a1eacb66..b10cb345d1 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.FileSystemCatalog;
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.Map;
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 class IcebergRowLineageCompatibilityTest {
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");
