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 f5c20ce29a [iceberg] Refuse types the mirror cannot publish before the
commit (#9477)
f5c20ce29a is described below
commit f5c20ce29ad965170eb0a8b0317dc3eb609502cf
Author: Jiajia Li <[email protected]>
AuthorDate: Mon Oct 5 22:47:50 2026 +0800
[iceberg] Refuse types the mirror cannot publish before the commit (#9477)
---
.../org/apache/paimon/schema/SchemaValidation.java | 45 ++++++++
.../apache/paimon/schema/SchemaManagerTest.java | 125 +++++++++++++++++++++
.../core/IcebergRowLineageCompatibilityTest.java | 16 +--
3 files changed, 172 insertions(+), 14 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
index 2da251e5ea..cd5a3fafed 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
@@ -136,6 +136,8 @@ public class SchemaValidation {
private static final int MAX_ICEBERG_TIMESTAMP_PRECISION = 6;
+ private static final int MIN_ICEBERG_VARIANT_FORMAT_VERSION = 3;
+
public static final List<Class<? extends DataType>>
PRIMARY_KEY_UNSUPPORTED_LOGICAL_TYPES =
Arrays.asList(
MapType.class,
@@ -254,6 +256,7 @@ public class SchemaValidation {
validateVariantTypes(schema, options);
validateIcebergTimestampPrecisions(tableRowType, options);
validateIcebergTimePrecisions(tableRowType, options);
+ validateIcebergPublishableTypes(tableRowType, options);
validateBlobFields(tableRowType, options);
Set<String> blobDescriptorFields =
validateBlobDescriptorFields(tableRowType, options);
Set<String> blobViewFields =
@@ -687,9 +690,51 @@ public class SchemaValidation {
validateIcebergGeospatialTypes(schema.logicalRowType(), options);
validateIcebergTimestampPrecisions(schema.logicalRowType(),
options);
validateIcebergTimePrecisions(schema.logicalRowType(), options);
+ validateIcebergPublishableTypes(schema.logicalRowType(), options);
}
}
+ /**
+ * Refuses the types the mirror cannot publish: the conversion that would
catch them only runs
+ * once the snapshot is durable.
+ */
+ public static void validateIcebergPublishableTypes(DataType dataType,
CoreOptions options) {
+ if
(options.toConfiguration().get(IcebergOptions.METADATA_ICEBERG_STORAGE)
+ == IcebergOptions.StorageType.DISABLED) {
+ return;
+ }
+ checkArgument(
+ !containsType(dataType, SchemaValidation::isUnpublishableType),
+ "Columns of type %s or %s cannot be published as Iceberg
metadata. The schema "
+ + "history counts too, so dropping such a column does
not make an existing "
+ + "table publishable; disable '%s' instead.",
+ DataTypeRoot.BLOB,
+ DataTypeRoot.VECTOR,
+ IcebergOptions.METADATA_ICEBERG_STORAGE.key());
+ checkArgument(
+ options.toConfiguration().get(IcebergOptions.FORMAT_VERSION)
+ >= MIN_ICEBERG_VARIANT_FORMAT_VERSION
+ || !containsType(dataType, type ->
type.is(DataTypeRoot.VARIANT)),
+ "Columns of type %s require Iceberg format version %s. Set
'%s' = '%s', or remove "
+ + "them.",
+ DataTypeRoot.VARIANT,
+ MIN_ICEBERG_VARIANT_FORMAT_VERSION,
+ IcebergOptions.FORMAT_VERSION.key(),
+ MIN_ICEBERG_VARIANT_FORMAT_VERSION);
+ checkArgument(
+
options.toConfiguration().get(IcebergOptions.METADATA_ICEBERG_STORAGE)
+ != IcebergOptions.StorageType.REST_CATALOG
+ || !containsType(dataType, type ->
type.is(DataTypeRoot.VARIANT)),
+ "Variant columns do not support '%s'='%s' because the bundled
Iceberg REST client "
+ + "cannot parse the Iceberg variant type.",
+ IcebergOptions.METADATA_ICEBERG_STORAGE.key(),
+ IcebergOptions.StorageType.REST_CATALOG);
+ }
+
+ private static boolean isUnpublishableType(DataType dataType) {
+ return dataType.isAnyOf(DataTypeRoot.BLOB, DataTypeRoot.VECTOR);
+ }
+
/** Validate geospatial types in a schema that will be published as
Iceberg metadata. */
public static void validateIcebergGeospatialTypes(DataType dataType,
CoreOptions options) {
boolean hasGeospatial =
diff --git
a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerTest.java
b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerTest.java
index ea2d4d8c9d..f6fceda8db 100644
--- a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerTest.java
@@ -36,6 +36,7 @@ import org.apache.paimon.table.sink.TableWriteImpl;
import org.apache.paimon.types.ArrayType;
import org.apache.paimon.types.BigIntType;
import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.DoubleType;
import org.apache.paimon.types.IntType;
@@ -197,6 +198,130 @@ public class SchemaManagerTest {
.hasStackTraceContaining("precision from 3 to 6");
}
+ @Test
+ public void testIcebergMetadataRefusesUnpublishableTypes() throws
Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.BUCKET.key(), "-1");
+ options.put(IcebergOptions.METADATA_ICEBERG_STORAGE.key(),
"table-location");
+
+ DataType vector = DataTypes.VECTOR(4, DataTypes.FLOAT());
+ assertThatThrownBy(
+ () ->
+ retryArtificialException(
+ () ->
+ manager.createTable(
+
unpublishableSchema(options, vector))))
+ .hasStackTraceContaining("cannot be published as Iceberg
metadata");
+
+ Map<String, String> disabled = new HashMap<>(options);
+ disabled.remove(IcebergOptions.METADATA_ICEBERG_STORAGE.key());
+ assertThatCode(
+ () ->
+ retryArtificialException(
+ () ->
+ manager.createTable(
+
unpublishableSchema(disabled, vector))))
+ .doesNotThrowAnyException();
+ }
+
+ @Test
+ public void testIcebergMetadataRefusesVariantBelowFormatVersionThree()
throws Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.BUCKET.key(), "-1");
+ options.put(IcebergOptions.METADATA_ICEBERG_STORAGE.key(),
"table-location");
+
+ assertThatThrownBy(
+ () ->
+ retryArtificialException(
+ () ->
+ manager.createTable(
+ unpublishableSchema(
+ options,
DataTypes.VARIANT()))))
+ .hasStackTraceContaining("require Iceberg format version 3");
+
+ Map<String, String> formatVersionThree = new HashMap<>(options);
+ formatVersionThree.put(IcebergOptions.FORMAT_VERSION.key(), "3");
+ assertThatCode(
+ () ->
+ retryArtificialException(
+ () ->
+ manager.createTable(
+ unpublishableSchema(
+
formatVersionThree,
+
DataTypes.VARIANT()))))
+ .doesNotThrowAnyException();
+ }
+
+ @Test
+ public void testIcebergMetadataRefusesVariantThroughRestCatalog() throws
Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.BUCKET.key(), "-1");
+ options.put(IcebergOptions.METADATA_ICEBERG_STORAGE.key(),
"rest-catalog");
+ options.put(IcebergOptions.FORMAT_VERSION.key(), "3");
+
+ assertThatThrownBy(
+ () ->
+ retryArtificialException(
+ () ->
+ manager.createTable(
+ unpublishableSchema(
+ options,
DataTypes.VARIANT()))))
+ .hasStackTraceContaining(
+ "do not support
'metadata.iceberg.storage'='rest-catalog'");
+ }
+
+ @Test
+ public void testIcebergMetadataRefusesBlobColumns() throws Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.BUCKET.key(), "-1");
+ options.put(CoreOptions.ROW_TRACKING_ENABLED.key(), "true");
+ options.put(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true");
+ options.put(IcebergOptions.METADATA_ICEBERG_STORAGE.key(),
"table-location");
+
+ assertThatThrownBy(
+ () ->
+ retryArtificialException(
+ () ->
+ manager.createTable(
+ unpublishableSchema(
+ options,
DataTypes.BLOB()))))
+ .hasStackTraceContaining("cannot be published as Iceberg
metadata");
+ }
+
+ @Test
+ public void
testEnableIcebergMetadataValidatesHistoricalUnpublishableTypes() throws
Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.BUCKET.key(), "-1");
+ retryArtificialException(
+ () ->
+ manager.createTable(
+ unpublishableSchema(
+ options, DataTypes.VECTOR(4,
DataTypes.FLOAT()))));
+ retryArtificialException(() ->
manager.commitChanges(SchemaChange.dropColumn("payload")));
+
+ assertThatThrownBy(
+ () ->
+ retryArtificialException(
+ () ->
+ manager.commitChanges(
+ SchemaChange.setOption(
+ IcebergOptions
+
.METADATA_ICEBERG_STORAGE
+ .key(),
+
"table-location"))))
+ .hasStackTraceContaining("cannot be published as Iceberg
metadata");
+ }
+
+ private Schema unpublishableSchema(Map<String, String> options, DataType
type) {
+ return new Schema(
+ Arrays.asList(
+ new DataField(0, "id", DataTypes.INT()), new
DataField(1, "payload", type)),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ options,
+ "");
+ }
+
@Test
public void testIcebergMetadataAllowsMicrosecondTimestamps() throws
Exception {
Map<String, String> options = new HashMap<>();
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 b10cb345d1..24ae1633a3 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
@@ -261,25 +261,13 @@ public class IcebergRowLineageCompatibilityTest {
}
@Test
- public void testVariantRejectedWithFormatVersion2() throws Exception {
+ public void testVariantRejectedWithFormatVersion2() {
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)))
+ assertThatThrownBy(() -> createPaimonTable(rowType,
formatVersionOptions(2), "parquet"))
.hasStackTraceContaining("VARIANT");
- write.close();
- commit.close();
}
@Test
