From de786b767e0ec1830653653bfa109ebf5679af65 Mon Sep 17 00:00:00 2001 From: Jiajia Li Date: Fri, 28 Aug 2026 14:00:02 -0400 Subject: [PATCH 1/5] [iceberg] Refuse nanosecond timestamps while Iceberg metadata is enabled --- docs/docs/iceberg/index.md | 8 +- .../apache/paimon/schema/SchemaManager.java | 6 +- .../paimon/schema/SchemaValidation.java | 34 +++++++ .../paimon/schema/SchemaManagerTest.java | 96 +++++++++++++++++++ 4 files changed, 139 insertions(+), 5 deletions(-) diff --git a/docs/docs/iceberg/index.md b/docs/docs/iceberg/index.md index f2878bcd6869..641cf030d669 100644 --- a/docs/docs/iceberg/index.md +++ b/docs/docs/iceberg/index.md @@ -96,8 +96,8 @@ Paimon Iceberg compatibility currently supports the following data types. | `DATE` | `date` | | `TIMESTAMP` (precision 3-6) | `timestamp` | | `TIMESTAMP_LTZ` (precision 3-6) | `timestamptz` | -| `TIMESTAMP` (precision 7-9) | `timestamp_ns` | -| `TIMESTAMP_LTZ` (precision 7-9) | `timestamptz_ns` | +| `TIMESTAMP` (precision 7-9) | not supported | +| `TIMESTAMP_LTZ` (precision 7-9) | not supported | | `GEOMETRY(crs)` | `geometry(crs)` | | `GEOGRAPHY(crs, algorithm)` | `geography(crs, algorithm)` | | `ARRAY` | `list` | @@ -108,7 +108,9 @@ Paimon Iceberg compatibility currently supports the following data types. **Note on Timestamp Types:** - `TIMESTAMP` and `TIMESTAMP_LTZ` types with precision from 3 to 6 are mapped to standard Iceberg timestamp types -- `TIMESTAMP` and `TIMESTAMP_LTZ` types with precision from 7 to 9 use nanosecond precision and require Iceberg v3 format +- `TIMESTAMP` and `TIMESTAMP_LTZ` types with a precision above 6 are rejected while Iceberg metadata is + enabled: Paimon writes them as Parquet INT96, which Iceberg reads as a microsecond zoned timestamp + rather than the `timestamp_ns` the metadata would declare. Use a precision of 6 or less. **Note on Geospatial Types:** - `GEOMETRY` and `GEOGRAPHY` values use OGC Well-Known Binary (WKB). The default CRS is `OGC:CRS84`, and the default geography edge algorithm is `spherical`. diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java index d730f332aeb0..6fe7866f7afc 100644 --- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java +++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java @@ -1217,13 +1217,13 @@ protected void updateLastColumn(int depth, List newFields, String fie @VisibleForTesting public boolean commit(TableSchema newSchema) throws Exception { SchemaValidation.validateTableSchema(newSchema); - validateHistoricalIcebergGeospatialTypes(newSchema); + validateHistoricalIcebergTypes(newSchema); SchemaValidation.validateFallbackBranch(this, newSchema); Path schemaPath = toSchemaPath(newSchema.id()); return fileIO.tryToWriteAtomic(schemaPath, newSchema.toString()); } - private void validateHistoricalIcebergGeospatialTypes(TableSchema newSchema) { + private void validateHistoricalIcebergTypes(TableSchema newSchema) { CoreOptions options = new CoreOptions(newSchema.options()); IcebergOptions.StorageType storage = options.toConfiguration().get(IcebergOptions.METADATA_ICEBERG_STORAGE); @@ -1231,8 +1231,10 @@ private void validateHistoricalIcebergGeospatialTypes(TableSchema newSchema) { return; } + // the mirror emits historical schemas too, so enabling it has to judge all of them for (TableSchema schema : listAll()) { SchemaValidation.validateIcebergGeospatialTypes(schema.logicalRowType(), options); + SchemaValidation.validateIcebergNanosecondTimestamps(schema.logicalRowType(), options); } } 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 cc0ee88ad702..8266154373c8 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 @@ -123,6 +123,9 @@ /** Validation utilities for {@link TableSchema}. */ public class SchemaValidation { + /** Above this precision Iceberg would need timestamp_ns, which Paimon does not write. */ + private static final int MAX_ICEBERG_TIMESTAMP_PRECISION = 6; + public static final List> PRIMARY_KEY_UNSUPPORTED_LOGICAL_TYPES = Arrays.asList( MapType.class, @@ -234,6 +237,7 @@ public static void validateTableSchema(TableSchema schema, Set dynamicOp FileFormat.fromIdentifier(options.formatType(), new Options(schema.options())); RowType tableRowType = new RowType(schema.fields()); validateGeospatialTypes(schema, options, tableRowType); + validateIcebergNanosecondTimestamps(tableRowType, options); validateBlobFields(tableRowType, options); Set blobDescriptorFields = validateBlobDescriptorFields(tableRowType, options); Set blobViewFields = @@ -532,6 +536,36 @@ private static void validateGeospatialTypes( } /** Validate geospatial types in a schema that will be published as Iceberg metadata. */ + /** + * Refuses nanosecond-precision timestamps while Iceberg metadata is enabled: Paimon writes them + * as Parquet INT96, which Iceberg reads as a microsecond zoned timestamp rather than the {@code + * timestamp_ns} the emitted metadata declares, so the two disagree about the data. + */ + public static void validateIcebergNanosecondTimestamps(DataType dataType, CoreOptions options) { + if (options.toConfiguration().get(IcebergOptions.METADATA_ICEBERG_STORAGE) + == IcebergOptions.StorageType.DISABLED) { + return; + } + checkArgument( + !containsType(dataType, SchemaValidation::isNanosecondTimestamp), + "Timestamp columns with a precision above %s are not supported when Iceberg metadata " + + "is enabled: Paimon writes them as Parquet INT96, which Iceberg cannot read " + + "back as the 'timestamp_ns' the metadata declares. Use a precision of %s or " + + "less, or disable '%s'.", + MAX_ICEBERG_TIMESTAMP_PRECISION, + MAX_ICEBERG_TIMESTAMP_PRECISION, + IcebergOptions.METADATA_ICEBERG_STORAGE.key()); + } + + private static boolean isNanosecondTimestamp(DataType dataType) { + if (dataType instanceof TimestampType) { + return ((TimestampType) dataType).getPrecision() > MAX_ICEBERG_TIMESTAMP_PRECISION; + } + return dataType instanceof LocalZonedTimestampType + && ((LocalZonedTimestampType) dataType).getPrecision() + > MAX_ICEBERG_TIMESTAMP_PRECISION; + } + public static void validateIcebergGeospatialTypes(DataType dataType, CoreOptions options) { boolean hasGeospatial = containsType( 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 cb39ea112dc5..518ba2933e7c 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 @@ -176,6 +176,102 @@ public void testUpdateOptions() throws Exception { assertThat(latest.get().options()).containsEntry("new_k", "new_v"); } + @Test + public void testIcebergMetadataRefusesNanosecondTimestamps() throws Exception { + Map options = new HashMap<>(); + options.put(CoreOptions.BUCKET.key(), "-1"); + options.put(IcebergOptions.METADATA_ICEBERG_STORAGE.key(), "table-location"); + Schema nanos = + new Schema( + Arrays.asList( + new DataField(0, "id", DataTypes.INT()), + new DataField(1, "ts", DataTypes.TIMESTAMP(9))), + Collections.emptyList(), + Collections.emptyList(), + options, + ""); + + assertThatThrownBy(() -> retryArtificialException(() -> manager.createTable(nanos))) + .hasStackTraceContaining("Timestamp columns with a precision above 6"); + } + + @Test + public void testIcebergMetadataAllowsMicrosecondTimestamps() throws Exception { + Map options = new HashMap<>(); + options.put(CoreOptions.BUCKET.key(), "-1"); + options.put(IcebergOptions.METADATA_ICEBERG_STORAGE.key(), "table-location"); + Schema micros = + new Schema( + Arrays.asList( + new DataField(0, "id", DataTypes.INT()), + new DataField(1, "ts", DataTypes.TIMESTAMP(6))), + Collections.emptyList(), + Collections.emptyList(), + options, + ""); + + assertThatCode(() -> retryArtificialException(() -> manager.createTable(micros))) + .doesNotThrowAnyException(); + } + + @Test + public void testEnablingIcebergMetadataRefusesNanosecondTimestamps() throws Exception { + Map options = new HashMap<>(); + options.put(CoreOptions.BUCKET.key(), "-1"); + Schema nanos = + new Schema( + Arrays.asList( + new DataField(0, "id", DataTypes.INT()), + new DataField(1, "ts", DataTypes.TIMESTAMP(9))), + Collections.emptyList(), + Collections.emptyList(), + options, + ""); + retryArtificialException(() -> manager.createTable(nanos)); + + assertThatThrownBy( + () -> + retryArtificialException( + () -> + manager.commitChanges( + SchemaChange.setOption( + IcebergOptions + .METADATA_ICEBERG_STORAGE + .key(), + "table-location")))) + .hasStackTraceContaining("Timestamp columns with a precision above 6"); + } + + @Test + public void testEnableIcebergMetadataValidatesHistoricalNanosecondSchemas() throws Exception { + Map options = new HashMap<>(); + options.put(CoreOptions.BUCKET.key(), "-1"); + Schema nanos = + new Schema( + Arrays.asList( + new DataField(0, "id", DataTypes.INT()), + new DataField(1, "ts", DataTypes.TIMESTAMP(9))), + Collections.emptyList(), + Collections.emptyList(), + options, + ""); + + retryArtificialException(() -> manager.createTable(nanos)); + retryArtificialException(() -> manager.commitChanges(SchemaChange.dropColumn("ts"))); + + assertThatThrownBy( + () -> + retryArtificialException( + () -> + manager.commitChanges( + SchemaChange.setOption( + IcebergOptions + .METADATA_ICEBERG_STORAGE + .key(), + "table-location")))) + .hasStackTraceContaining("Timestamp columns with a precision above 6"); + } + @Test public void testEnableIcebergMetadataValidatesHistoricalGeospatialSchemas() throws Exception { Map geospatialOptions = new HashMap<>(); From 21e2660d6c79709da487454f62a12462b967b642 Mon Sep 17 00:00:00 2001 From: Jiajia Li Date: Fri, 28 Aug 2026 21:41:20 -0400 Subject: [PATCH 2/5] [iceberg] Validate historical schemas when Iceberg is enabled by a dynamic option --- .../apache/paimon/schema/SchemaManager.java | 19 +------ .../paimon/schema/SchemaValidation.java | 19 ++++++- .../paimon/table/AbstractFileStoreTable.java | 10 ++++ .../iceberg/IcebergCompatibilityTest.java | 49 +++++++++++++++++++ 4 files changed, 79 insertions(+), 18 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java index 6fe7866f7afc..1a76668448b7 100644 --- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java +++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java @@ -25,7 +25,6 @@ import org.apache.paimon.catalog.Identifier; import org.apache.paimon.fs.FileIO; import org.apache.paimon.fs.Path; -import org.apache.paimon.iceberg.IcebergOptions; import org.apache.paimon.schema.ColumnDirectiveUtils.ConvertedColumn; import org.apache.paimon.schema.SchemaChange.AddColumn; import org.apache.paimon.schema.SchemaChange.DropColumn; @@ -1217,27 +1216,13 @@ protected void updateLastColumn(int depth, List newFields, String fie @VisibleForTesting public boolean commit(TableSchema newSchema) throws Exception { SchemaValidation.validateTableSchema(newSchema); - validateHistoricalIcebergTypes(newSchema); + SchemaValidation.validateHistoricalIcebergTypes( + this::listAll, new CoreOptions(newSchema.options())); SchemaValidation.validateFallbackBranch(this, newSchema); Path schemaPath = toSchemaPath(newSchema.id()); return fileIO.tryToWriteAtomic(schemaPath, newSchema.toString()); } - private void validateHistoricalIcebergTypes(TableSchema newSchema) { - CoreOptions options = new CoreOptions(newSchema.options()); - IcebergOptions.StorageType storage = - options.toConfiguration().get(IcebergOptions.METADATA_ICEBERG_STORAGE); - if (storage == IcebergOptions.StorageType.DISABLED) { - return; - } - - // the mirror emits historical schemas too, so enabling it has to judge all of them - for (TableSchema schema : listAll()) { - SchemaValidation.validateIcebergGeospatialTypes(schema.logicalRowType(), options); - SchemaValidation.validateIcebergNanosecondTimestamps(schema.logicalRowType(), options); - } - } - /** Read schema for schema id. */ public TableSchema schema(long id) { return fromPath(fileIO, toSchemaPath(id)); 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 8266154373c8..179cf04953a4 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 @@ -70,6 +70,7 @@ import java.util.Optional; import java.util.Set; import java.util.function.Predicate; +import java.util.function.Supplier; import java.util.stream.Collectors; import static org.apache.paimon.CoreOptions.BUCKET_KEY; @@ -535,7 +536,6 @@ private static void validateGeospatialTypes( geospatialSequenceFields); } - /** Validate geospatial types in a schema that will be published as Iceberg metadata. */ /** * Refuses nanosecond-precision timestamps while Iceberg metadata is enabled: Paimon writes them * as Parquet INT96, which Iceberg reads as a microsecond zoned timestamp rather than the {@code @@ -566,6 +566,23 @@ private static boolean isNanosecondTimestamp(DataType dataType) { > MAX_ICEBERG_TIMESTAMP_PRECISION; } + /** + * The mirror emits historical schemas too, so enabling it has to judge all of them. The history + * is read lazily, so a disabled mirror costs no listing. + */ + public static void validateHistoricalIcebergTypes( + Supplier> history, CoreOptions options) { + if (options.toConfiguration().get(IcebergOptions.METADATA_ICEBERG_STORAGE) + == IcebergOptions.StorageType.DISABLED) { + return; + } + for (TableSchema schema : history.get()) { + validateIcebergGeospatialTypes(schema.logicalRowType(), options); + validateIcebergNanosecondTimestamps(schema.logicalRowType(), options); + } + } + + /** Validate geospatial types in a schema that will be published as Iceberg metadata. */ public static void validateIcebergGeospatialTypes(DataType dataType, CoreOptions options) { boolean hasGeospatial = containsType( diff --git a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java index 97058f2b8e03..ef21981661dd 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java @@ -25,6 +25,7 @@ import org.apache.paimon.fs.FileIO; import org.apache.paimon.fs.Path; import org.apache.paimon.iceberg.IcebergCommitCallback; +import org.apache.paimon.iceberg.IcebergOptions; import org.apache.paimon.manifest.IndexManifestEntry; import org.apache.paimon.manifest.ManifestEntry; import org.apache.paimon.manifest.ManifestFileMeta; @@ -367,6 +368,15 @@ protected FileStoreTable copyInternal( // validate schema with new options SchemaValidation.validateTableSchema(newTableSchema, dynamicOptions.keySet()); + if (new CoreOptions(tableSchema.options()) + .toConfiguration() + .get(IcebergOptions.METADATA_ICEBERG_STORAGE) + == IcebergOptions.StorageType.DISABLED) { + // turning the mirror on here publishes the schemas already on disk, which no commit + // has judged under these options + SchemaValidation.validateHistoricalIcebergTypes( + () -> schemaManager().listAll(), new CoreOptions(newTableSchema.options())); + } return copy(newTableSchema); } diff --git a/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java b/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java index ca90e7a60850..2075b8d8585d 100644 --- a/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java @@ -1650,6 +1650,55 @@ public void testWithIncorrectBase() throws Exception { commit.close(); } + @Test + public void testDynamicallyEnablingIcebergRefusesHistoricalNanosecondTimestamps() + throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path path = new Path(tempDir.toString()); + Options options = new Options(); + options.set(CoreOptions.BUCKET, 1); + options.set(CoreOptions.FILE_FORMAT, "parquet"); + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.TIMESTAMP(9)}, + new String[] {"k", "ts"}); + Schema schema = + new Schema( + rowType.getFields(), + Collections.emptyList(), + Collections.singletonList("k"), + options.toMap(), + ""); + + FileStoreTable table; + Identifier identifier = Identifier.create("mydb", "t"); + try (FileSystemCatalog paimonCatalog = new FileSystemCatalog(fileIO, path)) { + paimonCatalog.createDatabase("mydb", false); + paimonCatalog.createTable(identifier, schema, false); + table = (FileStoreTable) paimonCatalog.getTable(identifier); + + String commitUser = UUID.randomUUID().toString(); + try (TableWriteImpl write = table.newWrite(commitUser); + TableCommitImpl commit = table.newCommit(commitUser)) { + write.write(GenericRow.of(1, Timestamp.fromEpochMillis(0))); + commit.commit(1, write.prepareCommit(false, 1)); + } + + paimonCatalog.alterTable(identifier, SchemaChange.dropColumn("ts"), false); + table = (FileStoreTable) paimonCatalog.getTable(identifier); + } + + FileStoreTable currentTable = table; + assertThatThrownBy( + () -> + currentTable.copy( + Collections.singletonMap( + IcebergOptions.METADATA_ICEBERG_STORAGE.key(), + IcebergOptions.StorageType.TABLE_LOCATION + .toString()))) + .hasMessageContaining("Timestamp columns with a precision above 6"); + } + /* Create snapshots Create tags From 379289c9105a8de0e7959af87b4c66b35e5f6dd6 Mon Sep 17 00:00:00 2001 From: Jiajia Li Date: Fri, 28 Aug 2026 22:18:32 -0400 Subject: [PATCH 3/5] [iceberg] Refuse nanosecond timestamps before the commit and where metadata is built --- .../org/apache/paimon/AbstractFileStore.java | 5 ++ .../paimon/iceberg/IcebergCommitCallback.java | 2 + .../iceberg/IcebergPreCommitValidation.java | 70 +++++++++++++++++++ .../iceberg/metadata/IcebergDataField.java | 15 ++-- .../iceberg/IcebergCompatibilityTest.java | 61 ++++++++++++++++ .../metadata/IcebergDataFieldTest.java | 69 +++++++----------- 6 files changed, 171 insertions(+), 51 deletions(-) create mode 100644 paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergPreCommitValidation.java diff --git a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java index 25c64029adf7..b88798b9246d 100644 --- a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java +++ b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java @@ -29,6 +29,7 @@ import org.apache.paimon.fs.Path; import org.apache.paimon.iceberg.IcebergCommitCallback; import org.apache.paimon.iceberg.IcebergOptions; +import org.apache.paimon.iceberg.IcebergPreCommitValidation; import org.apache.paimon.index.IndexFileHandler; import org.apache.paimon.manifest.IndexManifestFile; import org.apache.paimon.manifest.ManifestFile; @@ -397,6 +398,10 @@ private List createCommitPreCallbacks(FileStoreTable table) { if (options.isChainTable()) { callbacks.add(new ChainTableCommitPreCallback(table)); } + if (options.toConfiguration().get(IcebergOptions.METADATA_ICEBERG_STORAGE) + != IcebergOptions.StorageType.DISABLED) { + callbacks.add(new IcebergPreCommitValidation(table)); + } return callbacks; } 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 bb4490562509..a26db67c42d3 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 @@ -2149,6 +2149,8 @@ private IcebergSchema get(long schemaId) { checkVariantNotPublishable(schema.logicalRowType()); SchemaValidation.validateIcebergGeospatialTypes( schema.logicalRowType(), table.coreOptions()); + SchemaValidation.validateIcebergNanosecondTimestamps( + schema.logicalRowType(), table.coreOptions()); return IcebergSchema.create(schema); }); } diff --git a/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergPreCommitValidation.java b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergPreCommitValidation.java new file mode 100644 index 000000000000..956134b40955 --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergPreCommitValidation.java @@ -0,0 +1,70 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.iceberg; + +import org.apache.paimon.Snapshot; +import org.apache.paimon.manifest.IndexManifestEntry; +import org.apache.paimon.manifest.ManifestEntry; +import org.apache.paimon.manifest.SimpleFileEntry; +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.sink.CommitPreCallback; + +import java.util.List; + +/** + * Vetoes a commit whose schemas the Iceberg mirror cannot publish, before the snapshot becomes + * visible. The mirror emits every schema from 0 to the latest, so a table that was already enabled + * when an unsupported type entered its history keeps failing on the metadata it produces rather + * than on the commit that produced it. + * + *

Memoized on the latest schema, so steady-state commits only read the latest schema file. The + * comparison is by content: a rollback lets a later alteration reuse an id. + */ +public class IcebergPreCommitValidation implements CommitPreCallback { + + private final FileStoreTable table; + + private TableSchema validatedSchema; + + public IcebergPreCommitValidation(FileStoreTable table) { + this.table = table; + } + + @Override + public void call( + List baseFiles, + List deltaFiles, + List indexFiles, + Snapshot snapshot) { + SchemaManager schemaManager = table.schemaManager(); + TableSchema latest = schemaManager.latest().get(); + if (latest.equals(validatedSchema)) { + return; + } + SchemaValidation.validateHistoricalIcebergTypes( + schemaManager::listAll, table.coreOptions()); + validatedSchema = latest; + } + + @Override + public void close() {} +} diff --git a/paimon-core/src/main/java/org/apache/paimon/iceberg/metadata/IcebergDataField.java b/paimon-core/src/main/java/org/apache/paimon/iceberg/metadata/IcebergDataField.java index d69fef3fb3ef..2f2dce35c5eb 100644 --- a/paimon-core/src/main/java/org/apache/paimon/iceberg/metadata/IcebergDataField.java +++ b/paimon-core/src/main/java/org/apache/paimon/iceberg/metadata/IcebergDataField.java @@ -187,16 +187,19 @@ private static Object toTypeObject(DataType dataType, int fieldId, int depth) { "decimal(%d, %d)", decimalType.getPrecision(), decimalType.getScale()); case TIMESTAMP_WITHOUT_TIME_ZONE: int timestampPrecision = ((TimestampType) dataType).getPrecision(); + // Paimon writes these as Parquet INT96, which Iceberg reads as microseconds Preconditions.checkArgument( - timestampPrecision >= 3 && timestampPrecision <= 9, - "Paimon Iceberg compatibility only support timestamp type with precision from 3 to 9."); - return timestampPrecision >= 7 ? "timestamp_ns" : "timestamp"; + timestampPrecision >= 3 && timestampPrecision <= 6, + "Paimon Iceberg compatibility cannot publish a nanosecond-precision " + + "timestamp; use a timestamp precision from 3 to 6."); + return "timestamp"; case TIMESTAMP_WITH_LOCAL_TIME_ZONE: int timestampLtzPrecision = ((LocalZonedTimestampType) dataType).getPrecision(); Preconditions.checkArgument( - timestampLtzPrecision >= 3 && timestampLtzPrecision <= 9, - "Paimon Iceberg compatibility only support timestamp type with precision from 3 to 9."); - return timestampLtzPrecision >= 7 ? "timestamptz_ns" : "timestamptz"; + timestampLtzPrecision >= 3 && timestampLtzPrecision <= 6, + "Paimon Iceberg compatibility cannot publish a nanosecond-precision " + + "timestamp; use a timestamp precision from 3 to 6."); + return "timestamptz"; case VARIANT: return "variant"; case GEOMETRY: diff --git a/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java b/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java index 2075b8d8585d..a9efda69236e 100644 --- a/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java @@ -49,7 +49,9 @@ import org.apache.paimon.schema.Schema; import org.apache.paimon.schema.SchemaChange; import org.apache.paimon.schema.SchemaManager; +import org.apache.paimon.schema.TableSchema; import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.FileStoreTableFactory; import org.apache.paimon.table.sink.CommitMessage; import org.apache.paimon.table.sink.TableCommitImpl; import org.apache.paimon.table.sink.TableWriteImpl; @@ -1699,6 +1701,65 @@ public void testDynamicallyEnablingIcebergRefusesHistoricalNanosecondTimestamps( .hasMessageContaining("Timestamp columns with a precision above 6"); } + @Test + public void testExistingTableWithHistoricalNanosecondTimestampsRefusesToCommit() + throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path warehouse = new Path(tempDir.toString()); + Options options = new Options(); + options.set(CoreOptions.BUCKET, 1); + options.set(CoreOptions.FILE_FORMAT, "parquet"); + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.TIMESTAMP(9)}, + new String[] {"k", "ts"}); + Schema schema = + new Schema( + rowType.getFields(), + Collections.emptyList(), + Collections.singletonList("k"), + options.toMap(), + ""); + + Identifier identifier = Identifier.create("mydb", "t"); + try (FileSystemCatalog paimonCatalog = new FileSystemCatalog(fileIO, warehouse)) { + paimonCatalog.createDatabase("mydb", false); + paimonCatalog.createTable(identifier, schema, false); + FileStoreTable table = (FileStoreTable) paimonCatalog.getTable(identifier); + + String commitUser = UUID.randomUUID().toString(); + try (TableWriteImpl write = table.newWrite(commitUser); + TableCommitImpl commit = table.newCommit(commitUser)) { + write.write(GenericRow.of(1, Timestamp.fromEpochMillis(0))); + commit.commit(1, write.prepareCommit(false, 1)); + } + + paimonCatalog.alterTable(identifier, SchemaChange.dropColumn("ts"), false); + } + + Path tablePath = new Path(warehouse, "mydb.db/t"); + TableSchema latest = new SchemaManager(fileIO, tablePath).latest().get(); + Map upgraded = new HashMap<>(latest.options()); + upgraded.put( + IcebergOptions.METADATA_ICEBERG_STORAGE.key(), + IcebergOptions.StorageType.TABLE_LOCATION.toString()); + FileStoreTable table = + FileStoreTableFactory.create(fileIO, tablePath, latest.copy(upgraded)); + + String commitUser = UUID.randomUUID().toString(); + assertThatThrownBy( + () -> { + try (TableWriteImpl write = table.newWrite(commitUser); + TableCommitImpl commit = table.newCommit(commitUser)) { + write.write(GenericRow.of(2)); + commit.commit(2, write.prepareCommit(false, 2)); + } + }) + .hasMessageContaining("Timestamp columns with a precision above 6"); + assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(1L); + assertThat(fileIO.exists(new Path(tablePath, "metadata"))).isFalse(); + } + /* Create snapshots Create tags diff --git a/paimon-core/src/test/java/org/apache/paimon/iceberg/metadata/IcebergDataFieldTest.java b/paimon-core/src/test/java/org/apache/paimon/iceberg/metadata/IcebergDataFieldTest.java index f1daf206f421..c4838c263df9 100644 --- a/paimon-core/src/test/java/org/apache/paimon/iceberg/metadata/IcebergDataFieldTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/iceberg/metadata/IcebergDataFieldTest.java @@ -226,38 +226,20 @@ void testTimestampTypeConversions() { IcebergDataField icebergTimestampLtz = new IcebergDataField(timestampLtzField); assertThat(icebergTimestampLtz.type()).isEqualTo("timestamptz"); - // Test timestamp_ns (precision 7) - DataField timestampNs7Field = new DataField(3, "timestamp_ns", new TimestampType(false, 7)); - IcebergDataField icebergTimestampNs7 = new IcebergDataField(timestampNs7Field); - assertThat(icebergTimestampNs7.type()).isEqualTo("timestamp_ns"); - - // Test timestamp_ns (precision 8) - DataField timestampNs8Field = new DataField(4, "timestamp_ns", new TimestampType(false, 8)); - IcebergDataField icebergTimestampNs8 = new IcebergDataField(timestampNs8Field); - assertThat(icebergTimestampNs8.type()).isEqualTo("timestamp_ns"); - - // Test timestamp_ns (precision 9) - DataField timestampNs9Field = new DataField(5, "timestamp_ns", new TimestampType(false, 9)); - IcebergDataField icebergTimestampNs9 = new IcebergDataField(timestampNs9Field); - assertThat(icebergTimestampNs9.type()).isEqualTo("timestamp_ns"); - - // Test timestamptz_ns (precision 7) - DataField timestampLtzNs7Field = - new DataField(6, "timestamptz_ns", new LocalZonedTimestampType(false, 7)); - IcebergDataField icebergTimestampLtzNs7 = new IcebergDataField(timestampLtzNs7Field); - assertThat(icebergTimestampLtzNs7.type()).isEqualTo("timestamptz_ns"); - - // Test timestamptz_ns (precision 8) - DataField timestampLtzNs8Field = - new DataField(7, "timestamptz_ns", new LocalZonedTimestampType(false, 8)); - IcebergDataField icebergTimestampLtzNs8 = new IcebergDataField(timestampLtzNs8Field); - assertThat(icebergTimestampLtzNs8.type()).isEqualTo("timestamptz_ns"); - - // Test timestamptz_ns (precision 9) - DataField timestampLtzNs9Field = - new DataField(8, "timestamptz_ns", new LocalZonedTimestampType(false, 9)); - IcebergDataField icebergTimestampLtzNs9 = new IcebergDataField(timestampLtzNs9Field); - assertThat(icebergTimestampLtzNs9.type()).isEqualTo("timestamptz_ns"); + for (int precision = 7; precision <= 9; precision++) { + DataField nanosField = + new DataField(3, "timestamp_ns", new TimestampType(false, precision)); + assertThatThrownBy(() -> new IcebergDataField(nanosField)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("nanosecond-precision"); + + DataField nanosLtzField = + new DataField( + 4, "timestamptz_ns", new LocalZonedTimestampType(false, precision)); + assertThatThrownBy(() -> new IcebergDataField(nanosLtzField)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("nanosecond-precision"); + } } @Test @@ -268,43 +250,40 @@ void testTimestampPrecisionValidation() { new DataField(1, "timestamp", new TimestampType(false, 2)); assertThatThrownBy(() -> new IcebergDataField(invalidTimestampField)) .isInstanceOf(IllegalArgumentException.class) - .hasMessageContaining( - "Paimon Iceberg compatibility only support timestamp type with precision from 3 to 9"); + .hasMessageContaining("precision from 3 to 6"); // Test invalid precision (<= 3) DataField invalidTimestampField2 = new DataField(2, "timestamp", new TimestampType(false, 2)); assertThatThrownBy(() -> new IcebergDataField(invalidTimestampField2)) .isInstanceOf(IllegalArgumentException.class) - .hasMessageContaining( - "Paimon Iceberg compatibility only support timestamp type with precision from 3 to 9"); + .hasMessageContaining("precision from 3 to 6"); // Test invalid local timezone timestamp precision (<= 3) DataField invalidTimestampLtzField = new DataField(3, "timestamptz", new LocalZonedTimestampType(false, 2)); assertThatThrownBy(() -> new IcebergDataField(invalidTimestampLtzField)) .isInstanceOf(IllegalArgumentException.class) - .hasMessageContaining( - "Paimon Iceberg compatibility only support timestamp type with precision from 3 to 9"); + .hasMessageContaining("precision from 3 to 6"); // Test valid precision boundaries DataField validTimestamp4 = new DataField(4, "timestamp", new TimestampType(false, 4)); IcebergDataField icebergTimestamp4 = new IcebergDataField(validTimestamp4); assertThat(icebergTimestamp4.type()).isEqualTo("timestamp"); - DataField validTimestamp9 = new DataField(5, "timestamp", new TimestampType(false, 9)); - IcebergDataField icebergTimestamp9 = new IcebergDataField(validTimestamp9); - assertThat(icebergTimestamp9.type()).isEqualTo("timestamp_ns"); + DataField validTimestamp6 = new DataField(5, "timestamp", new TimestampType(false, 6)); + IcebergDataField icebergTimestamp6 = new IcebergDataField(validTimestamp6); + assertThat(icebergTimestamp6.type()).isEqualTo("timestamp"); DataField validTimestampLtz4 = new DataField(6, "timestamptz", new LocalZonedTimestampType(false, 4)); IcebergDataField icebergTimestampLtz4 = new IcebergDataField(validTimestampLtz4); assertThat(icebergTimestampLtz4.type()).isEqualTo("timestamptz"); - DataField validTimestampLtz9 = - new DataField(7, "timestamptz", new LocalZonedTimestampType(false, 9)); - IcebergDataField icebergTimestampLtz9 = new IcebergDataField(validTimestampLtz9); - assertThat(icebergTimestampLtz9.type()).isEqualTo("timestamptz_ns"); + DataField validTimestampLtz6 = + new DataField(7, "timestamptz", new LocalZonedTimestampType(false, 6)); + IcebergDataField icebergTimestampLtz6 = new IcebergDataField(validTimestampLtz6); + assertThat(icebergTimestampLtz6.type()).isEqualTo("timestamptz"); } @Test From 26b34410b25bcb571702cf1e8e46f24bb4a48c61 Mon Sep 17 00:00:00 2001 From: Jiajia Li Date: Fri, 28 Aug 2026 22:34:56 -0400 Subject: [PATCH 4/5] [iceberg] Align the timestamp precision guard with the range the mirror converts --- docs/docs/iceberg/index.md | 10 ++--- .../paimon/iceberg/IcebergCommitCallback.java | 2 +- .../paimon/schema/SchemaValidation.java | 40 +++++++++++-------- .../iceberg/IcebergCompatibilityTest.java | 20 ++++++---- .../paimon/schema/SchemaManagerTest.java | 30 ++++++++------ 5 files changed, 60 insertions(+), 42 deletions(-) diff --git a/docs/docs/iceberg/index.md b/docs/docs/iceberg/index.md index 641cf030d669..a0bc183b7320 100644 --- a/docs/docs/iceberg/index.md +++ b/docs/docs/iceberg/index.md @@ -96,8 +96,8 @@ Paimon Iceberg compatibility currently supports the following data types. | `DATE` | `date` | | `TIMESTAMP` (precision 3-6) | `timestamp` | | `TIMESTAMP_LTZ` (precision 3-6) | `timestamptz` | -| `TIMESTAMP` (precision 7-9) | not supported | -| `TIMESTAMP_LTZ` (precision 7-9) | not supported | +| `TIMESTAMP` (other precisions) | not supported | +| `TIMESTAMP_LTZ` (other precisions) | not supported | | `GEOMETRY(crs)` | `geometry(crs)` | | `GEOGRAPHY(crs, algorithm)` | `geography(crs, algorithm)` | | `ARRAY` | `list` | @@ -108,9 +108,9 @@ Paimon Iceberg compatibility currently supports the following data types. **Note on Timestamp Types:** - `TIMESTAMP` and `TIMESTAMP_LTZ` types with precision from 3 to 6 are mapped to standard Iceberg timestamp types -- `TIMESTAMP` and `TIMESTAMP_LTZ` types with a precision above 6 are rejected while Iceberg metadata is - enabled: Paimon writes them as Parquet INT96, which Iceberg reads as a microsecond zoned timestamp - rather than the `timestamp_ns` the metadata would declare. Use a precision of 6 or less. +- Any other precision is rejected while Iceberg metadata is enabled. A precision above 6 is written + as Parquet INT96, which Iceberg reads as a microsecond zoned timestamp rather than the + nanoseconds the column declares. Use a precision from 3 to 6. **Note on Geospatial Types:** - `GEOMETRY` and `GEOGRAPHY` values use OGC Well-Known Binary (WKB). The default CRS is `OGC:CRS84`, and the default geography edge algorithm is `spherical`. 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 a26db67c42d3..b3b250474b27 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 @@ -2149,7 +2149,7 @@ private IcebergSchema get(long schemaId) { checkVariantNotPublishable(schema.logicalRowType()); SchemaValidation.validateIcebergGeospatialTypes( schema.logicalRowType(), table.coreOptions()); - SchemaValidation.validateIcebergNanosecondTimestamps( + SchemaValidation.validateIcebergTimestampPrecisions( schema.logicalRowType(), table.coreOptions()); return IcebergSchema.create(schema); }); 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 179cf04953a4..f90f5a84c9d9 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 @@ -124,7 +124,9 @@ /** Validation utilities for {@link TableSchema}. */ public class SchemaValidation { - /** Above this precision Iceberg would need timestamp_ns, which Paimon does not write. */ + /** The precisions {@code IcebergDataField} maps to the Iceberg timestamp types. */ + private static final int MIN_ICEBERG_TIMESTAMP_PRECISION = 3; + private static final int MAX_ICEBERG_TIMESTAMP_PRECISION = 6; public static final List> PRIMARY_KEY_UNSUPPORTED_LOGICAL_TYPES = @@ -238,7 +240,7 @@ public static void validateTableSchema(TableSchema schema, Set dynamicOp FileFormat.fromIdentifier(options.formatType(), new Options(schema.options())); RowType tableRowType = new RowType(schema.fields()); validateGeospatialTypes(schema, options, tableRowType); - validateIcebergNanosecondTimestamps(tableRowType, options); + validateIcebergTimestampPrecisions(tableRowType, options); validateBlobFields(tableRowType, options); Set blobDescriptorFields = validateBlobDescriptorFields(tableRowType, options); Set blobViewFields = @@ -537,33 +539,39 @@ private static void validateGeospatialTypes( } /** - * Refuses nanosecond-precision timestamps while Iceberg metadata is enabled: Paimon writes them - * as Parquet INT96, which Iceberg reads as a microsecond zoned timestamp rather than the {@code - * timestamp_ns} the emitted metadata declares, so the two disagree about the data. + * Refuses the timestamp precisions the Iceberg mirror cannot publish, matching the range {@link + * org.apache.paimon.iceberg.metadata.IcebergDataField} converts. A higher precision is written + * as Parquet INT96, which Iceberg reads as a microsecond zoned timestamp rather than the + * nanoseconds the column declares, so the two disagree about the data. */ - public static void validateIcebergNanosecondTimestamps(DataType dataType, CoreOptions options) { + public static void validateIcebergTimestampPrecisions(DataType dataType, CoreOptions options) { if (options.toConfiguration().get(IcebergOptions.METADATA_ICEBERG_STORAGE) == IcebergOptions.StorageType.DISABLED) { return; } checkArgument( - !containsType(dataType, SchemaValidation::isNanosecondTimestamp), - "Timestamp columns with a precision above %s are not supported when Iceberg metadata " - + "is enabled: Paimon writes them as Parquet INT96, which Iceberg cannot read " - + "back as the 'timestamp_ns' the metadata declares. Use a precision of %s or " - + "less, or disable '%s'.", + !containsType(dataType, SchemaValidation::isUnpublishableTimestamp), + "Timestamp columns must have a precision from %s to %s when Iceberg metadata is " + + "enabled, the only precisions Iceberg compatibility can publish. Use a " + + "precision from %s to %s, or disable '%s'.", + MIN_ICEBERG_TIMESTAMP_PRECISION, MAX_ICEBERG_TIMESTAMP_PRECISION, + MIN_ICEBERG_TIMESTAMP_PRECISION, MAX_ICEBERG_TIMESTAMP_PRECISION, IcebergOptions.METADATA_ICEBERG_STORAGE.key()); } - private static boolean isNanosecondTimestamp(DataType dataType) { + private static boolean isUnpublishableTimestamp(DataType dataType) { if (dataType instanceof TimestampType) { - return ((TimestampType) dataType).getPrecision() > MAX_ICEBERG_TIMESTAMP_PRECISION; + return isUnpublishablePrecision(((TimestampType) dataType).getPrecision()); } return dataType instanceof LocalZonedTimestampType - && ((LocalZonedTimestampType) dataType).getPrecision() - > MAX_ICEBERG_TIMESTAMP_PRECISION; + && isUnpublishablePrecision(((LocalZonedTimestampType) dataType).getPrecision()); + } + + private static boolean isUnpublishablePrecision(int precision) { + return precision < MIN_ICEBERG_TIMESTAMP_PRECISION + || precision > MAX_ICEBERG_TIMESTAMP_PRECISION; } /** @@ -578,7 +586,7 @@ public static void validateHistoricalIcebergTypes( } for (TableSchema schema : history.get()) { validateIcebergGeospatialTypes(schema.logicalRowType(), options); - validateIcebergNanosecondTimestamps(schema.logicalRowType(), options); + validateIcebergTimestampPrecisions(schema.logicalRowType(), options); } } diff --git a/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java b/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java index a9efda69236e..60c7be92f7fe 100644 --- a/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java @@ -80,6 +80,8 @@ import org.apache.iceberg.io.CloseableIterable; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; import java.io.File; import java.math.BigDecimal; @@ -1652,8 +1654,9 @@ public void testWithIncorrectBase() throws Exception { commit.close(); } - @Test - public void testDynamicallyEnablingIcebergRefusesHistoricalNanosecondTimestamps() + @ParameterizedTest + @ValueSource(ints = {2, 9}) + public void testDynamicallyEnablingIcebergRefusesHistoricalTimestampPrecisions(int precision) throws Exception { LocalFileIO fileIO = LocalFileIO.create(); Path path = new Path(tempDir.toString()); @@ -1662,7 +1665,7 @@ public void testDynamicallyEnablingIcebergRefusesHistoricalNanosecondTimestamps( options.set(CoreOptions.FILE_FORMAT, "parquet"); RowType rowType = RowType.of( - new DataType[] {DataTypes.INT(), DataTypes.TIMESTAMP(9)}, + new DataType[] {DataTypes.INT(), DataTypes.TIMESTAMP(precision)}, new String[] {"k", "ts"}); Schema schema = new Schema( @@ -1698,11 +1701,12 @@ public void testDynamicallyEnablingIcebergRefusesHistoricalNanosecondTimestamps( IcebergOptions.METADATA_ICEBERG_STORAGE.key(), IcebergOptions.StorageType.TABLE_LOCATION .toString()))) - .hasMessageContaining("Timestamp columns with a precision above 6"); + .hasMessageContaining("precision from 3 to 6"); } - @Test - public void testExistingTableWithHistoricalNanosecondTimestampsRefusesToCommit() + @ParameterizedTest + @ValueSource(ints = {2, 9}) + public void testExistingTableWithUnpublishableHistoricalTimestampsRefusesToCommit(int precision) throws Exception { LocalFileIO fileIO = LocalFileIO.create(); Path warehouse = new Path(tempDir.toString()); @@ -1711,7 +1715,7 @@ public void testExistingTableWithHistoricalNanosecondTimestampsRefusesToCommit() options.set(CoreOptions.FILE_FORMAT, "parquet"); RowType rowType = RowType.of( - new DataType[] {DataTypes.INT(), DataTypes.TIMESTAMP(9)}, + new DataType[] {DataTypes.INT(), DataTypes.TIMESTAMP(precision)}, new String[] {"k", "ts"}); Schema schema = new Schema( @@ -1755,7 +1759,7 @@ public void testExistingTableWithHistoricalNanosecondTimestampsRefusesToCommit() commit.commit(2, write.prepareCommit(false, 2)); } }) - .hasMessageContaining("Timestamp columns with a precision above 6"); + .hasMessageContaining("precision from 3 to 6"); assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(1L); assertThat(fileIO.exists(new Path(tablePath, "metadata"))).isFalse(); } 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 518ba2933e7c..b3094e7fc88c 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 @@ -176,8 +176,10 @@ public void testUpdateOptions() throws Exception { assertThat(latest.get().options()).containsEntry("new_k", "new_v"); } - @Test - public void testIcebergMetadataRefusesNanosecondTimestamps() throws Exception { + @ParameterizedTest + @ValueSource(ints = {2, 9}) + public void testIcebergMetadataRefusesUnsupportedTimestampPrecisions(int precision) + throws Exception { Map options = new HashMap<>(); options.put(CoreOptions.BUCKET.key(), "-1"); options.put(IcebergOptions.METADATA_ICEBERG_STORAGE.key(), "table-location"); @@ -185,14 +187,14 @@ public void testIcebergMetadataRefusesNanosecondTimestamps() throws Exception { new Schema( Arrays.asList( new DataField(0, "id", DataTypes.INT()), - new DataField(1, "ts", DataTypes.TIMESTAMP(9))), + new DataField(1, "ts", DataTypes.TIMESTAMP(precision))), Collections.emptyList(), Collections.emptyList(), options, ""); assertThatThrownBy(() -> retryArtificialException(() -> manager.createTable(nanos))) - .hasStackTraceContaining("Timestamp columns with a precision above 6"); + .hasStackTraceContaining("precision from 3 to 6"); } @Test @@ -214,15 +216,17 @@ public void testIcebergMetadataAllowsMicrosecondTimestamps() throws Exception { .doesNotThrowAnyException(); } - @Test - public void testEnablingIcebergMetadataRefusesNanosecondTimestamps() throws Exception { + @ParameterizedTest + @ValueSource(ints = {2, 9}) + public void testEnablingIcebergMetadataRefusesUnsupportedTimestampPrecisions(int precision) + throws Exception { Map options = new HashMap<>(); options.put(CoreOptions.BUCKET.key(), "-1"); Schema nanos = new Schema( Arrays.asList( new DataField(0, "id", DataTypes.INT()), - new DataField(1, "ts", DataTypes.TIMESTAMP(9))), + new DataField(1, "ts", DataTypes.TIMESTAMP(precision))), Collections.emptyList(), Collections.emptyList(), options, @@ -239,18 +243,20 @@ public void testEnablingIcebergMetadataRefusesNanosecondTimestamps() throws Exce .METADATA_ICEBERG_STORAGE .key(), "table-location")))) - .hasStackTraceContaining("Timestamp columns with a precision above 6"); + .hasStackTraceContaining("precision from 3 to 6"); } - @Test - public void testEnableIcebergMetadataValidatesHistoricalNanosecondSchemas() throws Exception { + @ParameterizedTest + @ValueSource(ints = {2, 9}) + public void testEnableIcebergMetadataValidatesHistoricalTimestampPrecisions(int precision) + throws Exception { Map options = new HashMap<>(); options.put(CoreOptions.BUCKET.key(), "-1"); Schema nanos = new Schema( Arrays.asList( new DataField(0, "id", DataTypes.INT()), - new DataField(1, "ts", DataTypes.TIMESTAMP(9))), + new DataField(1, "ts", DataTypes.TIMESTAMP(precision))), Collections.emptyList(), Collections.emptyList(), options, @@ -269,7 +275,7 @@ public void testEnableIcebergMetadataValidatesHistoricalNanosecondSchemas() thro .METADATA_ICEBERG_STORAGE .key(), "table-location")))) - .hasStackTraceContaining("Timestamp columns with a precision above 6"); + .hasStackTraceContaining("precision from 3 to 6"); } @Test From a62aa3c3a81c6dd6360414ed6c11ce8087ab61f4 Mon Sep 17 00:00:00 2001 From: Jiajia Li Date: Fri, 28 Aug 2026 23:28:47 -0400 Subject: [PATCH 5/5] [iceberg] Read the committing branch's schemas when building metadata --- .../paimon/iceberg/IcebergCommitCallback.java | 2 +- .../iceberg/metadata/IcebergDataField.java | 8 ++-- .../iceberg/IcebergCompatibilityTest.java | 42 +++++++++++++++++++ .../metadata/IcebergDataFieldTest.java | 4 +- 4 files changed, 49 insertions(+), 7 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 b3b250474b27..d4c25a734e5c 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 @@ -2137,7 +2137,7 @@ private static List materializeFirstRowIds( private class SchemaCache { - SchemaManager schemaManager = new SchemaManager(table.fileIO(), table.location()); + SchemaManager schemaManager = table.schemaManager(); Map schemas = new HashMap<>(); private IcebergSchema get(long schemaId) { diff --git a/paimon-core/src/main/java/org/apache/paimon/iceberg/metadata/IcebergDataField.java b/paimon-core/src/main/java/org/apache/paimon/iceberg/metadata/IcebergDataField.java index 2f2dce35c5eb..890954aea3b7 100644 --- a/paimon-core/src/main/java/org/apache/paimon/iceberg/metadata/IcebergDataField.java +++ b/paimon-core/src/main/java/org/apache/paimon/iceberg/metadata/IcebergDataField.java @@ -190,15 +190,15 @@ private static Object toTypeObject(DataType dataType, int fieldId, int depth) { // Paimon writes these as Parquet INT96, which Iceberg reads as microseconds Preconditions.checkArgument( timestampPrecision >= 3 && timestampPrecision <= 6, - "Paimon Iceberg compatibility cannot publish a nanosecond-precision " - + "timestamp; use a timestamp precision from 3 to 6."); + "Paimon Iceberg compatibility only supports timestamp types with a " + + "precision from 3 to 6."); return "timestamp"; case TIMESTAMP_WITH_LOCAL_TIME_ZONE: int timestampLtzPrecision = ((LocalZonedTimestampType) dataType).getPrecision(); Preconditions.checkArgument( timestampLtzPrecision >= 3 && timestampLtzPrecision <= 6, - "Paimon Iceberg compatibility cannot publish a nanosecond-precision " - + "timestamp; use a timestamp precision from 3 to 6."); + "Paimon Iceberg compatibility only supports timestamp types with a " + + "precision from 3 to 6."); return "timestamptz"; case VARIANT: return "variant"; diff --git a/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java b/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java index 60c7be92f7fe..acaf154edcf5 100644 --- a/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java @@ -38,6 +38,7 @@ import org.apache.paimon.iceberg.manifest.IcebergManifestFile; import org.apache.paimon.iceberg.manifest.IcebergManifestFileMeta; import org.apache.paimon.iceberg.manifest.IcebergManifestList; +import org.apache.paimon.iceberg.metadata.IcebergDataField; import org.apache.paimon.iceberg.metadata.IcebergMetadata; import org.apache.paimon.iceberg.metadata.IcebergRef; import org.apache.paimon.iceberg.metadata.IcebergSchema; @@ -1764,6 +1765,47 @@ public void testExistingTableWithUnpublishableHistoricalTimestampsRefusesToCommi assertThat(fileIO.exists(new Path(tablePath, "metadata"))).isFalse(); } + @Test + public void testCommitOnBranchMirrorsTheBranchSchema() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}); + FileStoreTable table = + createPaimonTable( + rowType, Collections.emptyList(), Collections.singletonList("k"), 1); + + String commitUser = UUID.randomUUID().toString(); + try (TableWriteImpl write = table.newWrite(commitUser); + TableCommitImpl commit = table.newCommit(commitUser)) { + write.write(GenericRow.of(1, 10)); + commit.commit(1, write.prepareCommit(false, 1)); + } + + table.branchManager().createBranch("b1"); + new SchemaManager(table.fileIO(), table.location(), "b1") + .commitChanges(SchemaChange.addColumn("branch_only", DataTypes.INT())); + + FileStoreTable branchTable = table.switchToBranch("b1"); + try (TableWriteImpl write = branchTable.newWrite(commitUser); + TableCommitImpl commit = branchTable.newCommit(commitUser)) { + write.write(GenericRow.of(2, 20, 200)); + commit.commit(2, write.prepareCommit(false, 2)); + } + + IcebergMetadata metadata = + IcebergMetadata.fromPath( + branchTable.fileIO(), + new Path( + branchTable.location(), + "metadata/v" + + branchTable.snapshotManager().latestSnapshotId() + + ".metadata.json")); + assertThat( + metadata.schemas().get(metadata.currentSchemaId()).fields().stream() + .map(IcebergDataField::name)) + .containsExactly("k", "v", "branch_only"); + } + /* Create snapshots Create tags diff --git a/paimon-core/src/test/java/org/apache/paimon/iceberg/metadata/IcebergDataFieldTest.java b/paimon-core/src/test/java/org/apache/paimon/iceberg/metadata/IcebergDataFieldTest.java index c4838c263df9..eafa732d2019 100644 --- a/paimon-core/src/test/java/org/apache/paimon/iceberg/metadata/IcebergDataFieldTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/iceberg/metadata/IcebergDataFieldTest.java @@ -231,14 +231,14 @@ void testTimestampTypeConversions() { new DataField(3, "timestamp_ns", new TimestampType(false, precision)); assertThatThrownBy(() -> new IcebergDataField(nanosField)) .isInstanceOf(IllegalArgumentException.class) - .hasMessageContaining("nanosecond-precision"); + .hasMessageContaining("precision from 3 to 6"); DataField nanosLtzField = new DataField( 4, "timestamptz_ns", new LocalZonedTimestampType(false, precision)); assertThatThrownBy(() -> new IcebergDataField(nanosLtzField)) .isInstanceOf(IllegalArgumentException.class) - .hasMessageContaining("nanosecond-precision"); + .hasMessageContaining("precision from 3 to 6"); } }