[iceberg] Refuse nanosecond timestamps while Iceberg metadata is enabled

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`.

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<CommitPreCallback> 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;
     }
 

paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java

@@ -2137,7 +2137,7 @@ private static List<IcebergManifestEntry> materializeFirstRowIds(
 
     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 @@ private IcebergSchema get(long schemaId) {
                         checkVariantNotPublishable(schema.logicalRowType());
                         SchemaValidation.validateIcebergGeospatialTypes(
                                 schema.logicalRowType(), table.coreOptions());
+                        SchemaValidation.validateIcebergTimestampPrecisions(
+                                schema.logicalRowType(), table.coreOptions());
                         return IcebergSchema.create(schema);
                     });
         }

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() {}
+}

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:

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<DataField> 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));

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<Class<? extends DataType>> PRIMARY_KEY_UNSUPPORTED_LOGICAL_TYPES =
             Arrays.asList(
                     MapType.class,
@@ -234,6 +240,7 @@ public static void validateTableSchema(TableSchema schema, Set<String> 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<String> blobDescriptorFields = validateBlobDescriptorFields(tableRowType, options);
         Set<String> 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<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 =

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);
     }