diff --git a/docs/docs/iceberg/index.md b/docs/docs/iceberg/index.md index f2878bcd6869..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) | `timestamp_ns` | -| `TIMESTAMP_LTZ` (precision 7-9) | `timestamptz_ns` | +| `TIMESTAMP` (other precisions) | not supported | +| `TIMESTAMP_LTZ` (other precisions) | 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 +- 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/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..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) { @@ -2149,6 +2149,8 @@ private IcebergSchema get(long schemaId) { checkVariantNotPublishable(schema.logicalRowType()); SchemaValidation.validateIcebergGeospatialTypes( schema.logicalRowType(), table.coreOptions()); + SchemaValidation.validateIcebergTimestampPrecisions( + 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..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 @@ -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 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 <= 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 only supports timestamp types with a " + + "precision from 3 to 6."); + return "timestamptz"; case VARIANT: return "variant"; case GEOMETRY: 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..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,25 +1216,13 @@ protected void updateLastColumn(int depth, List newFields, String fie @VisibleForTesting public boolean commit(TableSchema newSchema) throws Exception { SchemaValidation.validateTableSchema(newSchema); - validateHistoricalIcebergGeospatialTypes(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 validateHistoricalIcebergGeospatialTypes(TableSchema newSchema) { - CoreOptions options = new CoreOptions(newSchema.options()); - IcebergOptions.StorageType storage = - options.toConfiguration().get(IcebergOptions.METADATA_ICEBERG_STORAGE); - if (storage == IcebergOptions.StorageType.DISABLED) { - return; - } - - for (TableSchema schema : listAll()) { - SchemaValidation.validateIcebergGeospatialTypes(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 cc0ee88ad702..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 @@ -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; @@ -123,6 +124,11 @@ /** Validation utilities for {@link TableSchema}. */ public class SchemaValidation { + /** 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 = Arrays.asList( MapType.class, @@ -234,6 +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); + validateIcebergTimestampPrecisions(tableRowType, options); validateBlobFields(tableRowType, options); Set blobDescriptorFields = validateBlobDescriptorFields(tableRowType, options); Set blobViewFields = @@ -531,6 +538,58 @@ private static void validateGeospatialTypes( geospatialSequenceFields); } + /** + * 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 validateIcebergTimestampPrecisions(DataType dataType, CoreOptions options) { + if (options.toConfiguration().get(IcebergOptions.METADATA_ICEBERG_STORAGE) + == IcebergOptions.StorageType.DISABLED) { + return; + } + checkArgument( + !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 isUnpublishableTimestamp(DataType dataType) { + if (dataType instanceof TimestampType) { + return isUnpublishablePrecision(((TimestampType) dataType).getPrecision()); + } + return dataType instanceof LocalZonedTimestampType + && isUnpublishablePrecision(((LocalZonedTimestampType) dataType).getPrecision()); + } + + private static boolean isUnpublishablePrecision(int precision) { + return precision < MIN_ICEBERG_TIMESTAMP_PRECISION + || precision > 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); + validateIcebergTimestampPrecisions(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 = 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..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; @@ -49,7 +50,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; @@ -78,6 +81,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; @@ -1650,6 +1655,157 @@ public void testWithIncorrectBase() throws Exception { commit.close(); } + @ParameterizedTest + @ValueSource(ints = {2, 9}) + public void testDynamicallyEnablingIcebergRefusesHistoricalTimestampPrecisions(int precision) + 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(precision)}, + 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("precision from 3 to 6"); + } + + @ParameterizedTest + @ValueSource(ints = {2, 9}) + public void testExistingTableWithUnpublishableHistoricalTimestampsRefusesToCommit(int precision) + 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(precision)}, + 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("precision from 3 to 6"); + assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(1L); + 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 f1daf206f421..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 @@ -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("precision from 3 to 6"); + + DataField nanosLtzField = + new DataField( + 4, "timestamptz_ns", new LocalZonedTimestampType(false, precision)); + assertThatThrownBy(() -> new IcebergDataField(nanosLtzField)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("precision from 3 to 6"); + } } @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 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..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,6 +176,108 @@ public void testUpdateOptions() throws Exception { assertThat(latest.get().options()).containsEntry("new_k", "new_v"); } + @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"); + Schema nanos = + new Schema( + Arrays.asList( + new DataField(0, "id", DataTypes.INT()), + new DataField(1, "ts", DataTypes.TIMESTAMP(precision))), + Collections.emptyList(), + Collections.emptyList(), + options, + ""); + + assertThatThrownBy(() -> retryArtificialException(() -> manager.createTable(nanos))) + .hasStackTraceContaining("precision from 3 to 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(); + } + + @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(precision))), + Collections.emptyList(), + Collections.emptyList(), + options, + ""); + retryArtificialException(() -> manager.createTable(nanos)); + + assertThatThrownBy( + () -> + retryArtificialException( + () -> + manager.commitChanges( + SchemaChange.setOption( + IcebergOptions + .METADATA_ICEBERG_STORAGE + .key(), + "table-location")))) + .hasStackTraceContaining("precision from 3 to 6"); + } + + @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(precision))), + 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("precision from 3 to 6"); + } + @Test public void testEnableIcebergMetadataValidatesHistoricalGeospatialSchemas() throws Exception { Map geospatialOptions = new HashMap<>();