Skip to content

[iceberg] Refuse types the mirror cannot publish before the commit or the replace #9477

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
plusplusjiajia wants to merge 2 commits into
base: master
Choose a base branch
from
Draft

[iceberg] Refuse types the mirror cannot publish before the commit or the replace #9477

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
0c38c87
[iceberg] Refuse types the mirror cannot publish before the commit
plusplusjiajia Aug 29, 2026
ede51e5
[core] Validate the replacement schema before replaceTable destroys t…
plusplusjiajia Aug 30, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaChange;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.schema.SchemaValidation;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.FormatTable;
Expand Down Expand Up @@ -81,6 +82,7 @@
import static org.apache.paimon.catalog.Identifier.DEFAULT_MAIN_BRANCH;
import static org.apache.paimon.options.CatalogOptions.LOCK_ENABLED;
import static org.apache.paimon.options.CatalogOptions.LOCK_TYPE;
import static org.apache.paimon.schema.ColumnDirectiveUtils.applyDirectives;

/** Common implementation of {@link Catalog}. */
public abstract class AbstractCatalog implements Catalog {
Expand Down Expand Up @@ -530,6 +532,12 @@ public void replaceTable(Identifier identifier, Schema newSchema, boolean ignore
}

TableType targetTableType = Options.fromMap(newSchema.options()).get(TYPE);
if (targetTableType.equals(TableType.TABLE)
|| targetTableType.equals(TableType.MATERIALIZED_TABLE)) {
// both paths below drop or truncate the table before the schema they write reaches
// this validation, and neither is undone once it refuses that schema
SchemaValidation.validateTableSchema(TableSchema.create(0, applyDirectives(newSchema)));
}
if (!(existing instanceof FileStoreTable) || !targetTableType.equals(TableType.TABLE)) {
dropAndCreateTable(identifier, newSchema);
return;
Expand Down
36 changes: 36 additions & 0 deletions paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -133,6 +133,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,
Expand Down Expand Up @@ -250,6 +252,7 @@ public static void validateTableSchema(TableSchema schema, Set<String> dynamicOp
validateGeospatialTypes(schema, options, tableRowType);
validateIcebergTimestampPrecisions(tableRowType, options);
validateIcebergTimePrecisions(tableRowType, options);
validateIcebergPublishableTypes(tableRowType, options);
validateBlobFields(tableRowType, options);
Set<String> blobDescriptorFields = validateBlobDescriptorFields(tableRowType, options);
Set<String> blobViewFields =
Expand Down Expand Up @@ -638,7 +641,40 @@ public static void validateHistoricalIcebergTypes(
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. Remove them, "
+ "or disable '%s'.",
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);
}

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. */
Expand Down
46 changes: 46 additions & 0 deletions paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,10 +20,17 @@

import org.apache.paimon.CoreOptions;
import org.apache.paimon.TableType;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.fs.Path;
import org.apache.paimon.iceberg.IcebergOptions;
import org.apache.paimon.options.Options;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.sink.BatchTableCommit;
import org.apache.paimon.table.sink.BatchTableWrite;
import org.apache.paimon.table.sink.BatchWriteBuilder;
import org.apache.paimon.types.DataTypes;

import org.apache.paimon.shade.guava30.com.google.common.collect.Lists;
Expand Down Expand Up @@ -151,4 +158,43 @@ public void testPartitionsFromCatalogAreRejectedOutsideRestCatalog() throws Exce
.hasMessageContaining(CoreOptions.METASTORE_PARTITIONED_TABLE.key())
.hasMessageContaining("REST catalog");
}

@Test
public void testReplaceTableKeepsTheDataWhenTheReplacementIsRefused() throws Exception {
String database = "replace_refused_db";
catalog.createDatabase(database, false);
Identifier identifier = Identifier.create(database, "t");
catalog.createTable(
identifier,
Schema.newBuilder()
.column("id", DataTypes.INT())
.column("name", DataTypes.STRING())
.option(CoreOptions.BUCKET.key(), "1")
.option(CoreOptions.BUCKET_KEY.key(), "id")
.build(),
false);
FileStoreTable table = (FileStoreTable) catalog.getTable(identifier);
BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
try (BatchTableWrite write = writeBuilder.newWrite();
BatchTableCommit commit = writeBuilder.newCommit()) {
write.write(GenericRow.of(1, BinaryString.fromString("a")));
commit.commit(write.prepareCommit());
}
long snapshotId = table.snapshotManager().latestSnapshotId();

Schema refused =
Schema.newBuilder()
.column("id", DataTypes.INT())
.column("payload", DataTypes.VECTOR(4, DataTypes.FLOAT()))
.option(CoreOptions.BUCKET.key(), "1")
.option(CoreOptions.BUCKET_KEY.key(), "id")
.option(IcebergOptions.METADATA_ICEBERG_STORAGE.key(), "table-location")
.build();
assertThatThrownBy(() -> catalog.replaceTable(identifier, refused, false))
.hasStackTraceContaining("cannot be published as Iceberg metadata");

FileStoreTable kept = (FileStoreTable) catalog.getTable(identifier);
assertThat(kept.snapshotManager().latestSnapshotId()).isEqualTo(snapshotId);
assertThat(read(kept, null, null, null, null)).hasSize(1);
}
}
107 changes: 107 additions & 0 deletions paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
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;
Expand Down Expand Up @@ -197,6 +198,112 @@ public void testIcebergMetadataRefusesUnsupportedTimestampPrecisions(int precisi
.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 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<>();
Expand Down
16 changes: 2 additions & 14 deletions paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergRowLineageCompatibilityTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -261,25 +261,13 @@ public void testVariantPublishableWithFormatVersion3() throws Exception {
}

@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
Expand Down
Loading