[iceberg] Allow publishing VARIANT columns with Iceberg format version 3

paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java

@@ -733,17 +733,16 @@ private List<IcebergPartitionField> 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<String> 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.",
+                "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);

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");