diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java index 5784d56b0ed4..ed59fb01b60e 100644 --- a/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java +++ b/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java @@ -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; @@ -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 { @@ -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; 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 fe1239a20846..f41adde231dd 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 @@ -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> PRIMARY_KEY_UNSUPPORTED_LOGICAL_TYPES = Arrays.asList( MapType.class, @@ -250,6 +252,7 @@ public static void validateTableSchema(TableSchema schema, Set dynamicOp validateGeospatialTypes(schema, options, tableRowType); validateIcebergTimestampPrecisions(tableRowType, options); validateIcebergTimePrecisions(tableRowType, options); + validateIcebergPublishableTypes(tableRowType, options); validateBlobFields(tableRowType, options); Set blobDescriptorFields = validateBlobDescriptorFields(tableRowType, options); Set blobViewFields = @@ -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. */ diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java b/paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java index 5e186cdf10df..aa2af62e4c14 100644 --- a/paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java @@ -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; @@ -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); + } } 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 78396006a7fe..f16c6ef74fc3 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.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,112 @@ public void testIcebergMetadataRefusesUnsupportedTimestampPrecisions(int precisi .hasStackTraceContaining("precision from 3 to 6"); } + @Test + public void testIcebergMetadataRefusesUnpublishableTypes() throws Exception { + Map 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 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 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 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 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 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 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 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 b10cb345d19d..24ae1633a314 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 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