This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 22daae7378 [iceberg] Refuse nanosecond timestamps while Iceberg
metadata is enabled (#9455)
22daae7378 is described below
commit 22daae737895063be2ec50538f9cf725876dcdf6
Author: Jiajia Li <[email protected]>
AuthorDate: Sat Aug 29 20:54:37 2026 +0800
[iceberg] Refuse nanosecond timestamps while Iceberg metadata is enabled
(#9455)
---
docs/docs/iceberg/index.md | 8 +-
.../java/org/apache/paimon/AbstractFileStore.java | 5 +
.../paimon/iceberg/IcebergCommitCallback.java | 4 +-
.../paimon/iceberg/IcebergPreCommitValidation.java | 70 +++++++++
.../paimon/iceberg/metadata/IcebergDataField.java | 15 +-
.../org/apache/paimon/schema/SchemaManager.java | 17 +--
.../org/apache/paimon/schema/SchemaValidation.java | 59 ++++++++
.../paimon/table/AbstractFileStoreTable.java | 10 ++
.../paimon/iceberg/IcebergCompatibilityTest.java | 156 +++++++++++++++++++++
.../iceberg/metadata/IcebergDataFieldTest.java | 69 ++++-----
.../apache/paimon/schema/SchemaManagerTest.java | 102 ++++++++++++++
11 files changed, 445 insertions(+), 70 deletions(-)
diff --git a/docs/docs/iceberg/index.md b/docs/docs/iceberg/index.md
index f2878bcd68..a0bc183b73 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 25c64029ad..b88798b924 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.FileIO;
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 @@ abstract class AbstractFileStore<T> implements
FileStore<T> {
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 bb44905625..d4c25a734e 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 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
private class SchemaCache {
- SchemaManager schemaManager = new SchemaManager(table.fileIO(),
table.location());
+ SchemaManager schemaManager = table.schemaManager();
Map<Long, IcebergSchema> schemas = new HashMap<>();
private IcebergSchema get(long schemaId) {
@@ -2149,6 +2149,8 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
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 0000000000..956134b409
--- /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.
+ *
+ * <p>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<SimpleFileEntry> baseFiles,
+ List<ManifestEntry> deltaFiles,
+ List<IndexManifestEntry> 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 d69fef3fb3..890954aea3 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 @@ public class IcebergDataField {
"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 d730f332ae..1a76668448 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.Catalog;
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 @@ public class SchemaManager implements Serializable {
@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 cc0ee88ad7..f90f5a84c9 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.Map;
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 @@ import static
org.apache.paimon.utils.Preconditions.checkState;
/** 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<Class<? extends DataType>>
PRIMARY_KEY_UNSUPPORTED_LOGICAL_TYPES =
Arrays.asList(
MapType.class,
@@ -234,6 +240,7 @@ public class SchemaValidation {
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<String> blobDescriptorFields =
validateBlobDescriptorFields(tableRowType, options);
Set<String> blobViewFields =
@@ -531,6 +538,58 @@ public class SchemaValidation {
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<List<TableSchema>> 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 97058f2b8e..ef21981661 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.consumer.ConsumerManager;
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 @@ abstract class AbstractFileStoreTable implements
FileStoreTable {
// 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 ca90e7a608..acaf154edc 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.IcebergManifestEntry;
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.options.Options;
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.hadoop.HadoopCatalog;
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 class IcebergCompatibilityTest {
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<String, String> 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 f1daf206f4..eafa732d20 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 @@ class IcebergDataFieldTest {
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 @@ class IcebergDataFieldTest {
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 cb39ea112d..b3094e7fc8 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 class SchemaManagerTest {
assertThat(latest.get().options()).containsEntry("new_k", "new_v");
}
+ @ParameterizedTest
+ @ValueSource(ints = {2, 9})
+ public void testIcebergMetadataRefusesUnsupportedTimestampPrecisions(int
precision)
+ throws Exception {
+ Map<String, String> 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<String, String> 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<String, String> 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<String, String> 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<String, String> geospatialOptions = new HashMap<>();
