[iceberg] Leave the nanosecond refusal to the validator that knows the table

paimon-core/src/main/java/org/apache/paimon/iceberg/metadata/IcebergDataField.java

@@ -186,20 +186,22 @@ private static Object toTypeObject(DataType dataType, int fieldId, int depth) {
                 return String.format(
                         "decimal(%d, %d)", decimalType.getPrecision(), decimalType.getScale());
             case TIMESTAMP_WITHOUT_TIME_ZONE:
+                // Nanoseconds name the Iceberg v3 type. Whether a table may publish one is decided
+                // by SchemaValidation#validateIcebergTimestampPrecisions, which knows the mirror is
+                // enabled and writes INT96.
                 int timestampPrecision = ((TimestampType) dataType).getPrecision();
-                // Paimon writes these as Parquet INT96, which Iceberg reads as microseconds
                 Preconditions.checkArgument(
-                        timestampPrecision >= 3 && timestampPrecision <= 6,
+                        timestampPrecision >= 3 && timestampPrecision <= 9,
                         "Paimon Iceberg compatibility only supports timestamp types with a "
-                                + "precision from 3 to 6.");
-                return "timestamp";
+                                + "precision from 3 to 9.");
+                return timestampPrecision >= 7 ? "timestamp_ns" : "timestamp";
             case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
                 int timestampLtzPrecision = ((LocalZonedTimestampType) dataType).getPrecision();
                 Preconditions.checkArgument(
-                        timestampLtzPrecision >= 3 && timestampLtzPrecision <= 6,
+                        timestampLtzPrecision >= 3 && timestampLtzPrecision <= 9,
                         "Paimon Iceberg compatibility only supports timestamp types with a "
-                                + "precision from 3 to 6.");
-                return "timestamptz";
+                                + "precision from 3 to 9.");
+                return timestampLtzPrecision >= 7 ? "timestamptz_ns" : "timestamptz";
             case VARIANT:
                 return "variant";
             case GEOMETRY:

paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java

@@ -128,7 +128,10 @@ public class SchemaValidation {
     /** The ceiling {@code IcebergDataField} converts. */
     private static final int MAX_ICEBERG_TIME_PRECISION = 3;
 
-    /** The precisions {@code IcebergDataField} maps to the Iceberg timestamp types. */
+    /**
+     * The timestamp precisions the mirror can publish, narrower than the 3 to 9 {@code
+     * IcebergDataField} names a type for.
+     */
     private static final int MIN_ICEBERG_TIMESTAMP_PRECISION = 3;
 
     private static final int MAX_ICEBERG_TIMESTAMP_PRECISION = 6;
@@ -564,10 +567,10 @@ private static void validateGeospatialTypes(
     }
 
     /**
-     * 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.
+     * Refuses the timestamp precisions the Iceberg mirror cannot publish. 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. The refusal belongs
+     * here rather than in the type mapping, which does not know who writes the files.
      */
     public static void validateIcebergTimestampPrecisions(DataType dataType, CoreOptions options) {
         if (options.toConfiguration().get(IcebergOptions.METADATA_ICEBERG_STORAGE)

paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java

@@ -82,6 +82,7 @@
 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.MethodSource;
 import org.junit.jupiter.params.provider.ValueSource;
 
 import java.io.File;
@@ -106,6 +107,7 @@
 import java.util.function.BiFunction;
 import java.util.function.Function;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -1705,19 +1707,26 @@ public void testDynamicallyEnablingIcebergRefusesHistoricalTimestampPrecisions(i
                 .hasMessageContaining("precision from 3 to 6");
     }
 
+    /** Below the floor, and above the ceiling in both timestamp families. */
+    static Stream<DataType> unpublishableTimestampTypes() {
+        return Stream.of(
+                DataTypes.TIMESTAMP(2),
+                DataTypes.TIMESTAMP(9),
+                DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(9));
+    }
+
     @ParameterizedTest
-    @ValueSource(ints = {2, 9})
-    public void testExistingTableWithUnpublishableHistoricalTimestampsRefusesToCommit(int precision)
-            throws Exception {
+    @MethodSource("unpublishableTimestampTypes")
+    public void testExistingTableWithUnpublishableHistoricalTimestampsRefusesToCommit(
+            DataType timestampType) 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"});
+                        new DataType[] {DataTypes.INT(), timestampType}, new String[] {"k", "ts"});
         Schema schema =
                 new Schema(
                         rowType.getFields(),

paimon-core/src/test/java/org/apache/paimon/iceberg/metadata/IcebergDataFieldTest.java

@@ -226,19 +226,16 @@ void testTimestampTypeConversions() {
         IcebergDataField icebergTimestampLtz = new IcebergDataField(timestampLtzField);
         assertThat(icebergTimestampLtz.type()).isEqualTo("timestamptz");
 
+        // Nanoseconds name the Iceberg v3 type; SchemaValidation decides who may publish one.
         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");
+            assertThat(new IcebergDataField(nanosField).type()).isEqualTo("timestamp_ns");
 
             DataField nanosLtzField =
                     new DataField(
                             4, "timestamptz_ns", new LocalZonedTimestampType(false, precision));
-            assertThatThrownBy(() -> new IcebergDataField(nanosLtzField))
-                    .isInstanceOf(IllegalArgumentException.class)
-                    .hasMessageContaining("precision from 3 to 6");
+            assertThat(new IcebergDataField(nanosLtzField).type()).isEqualTo("timestamptz_ns");
         }
     }
 
@@ -250,21 +247,21 @@ void testTimestampPrecisionValidation() {
                 new DataField(1, "timestamp", new TimestampType(false, 2));
         assertThatThrownBy(() -> new IcebergDataField(invalidTimestampField))
                 .isInstanceOf(IllegalArgumentException.class)
-                .hasMessageContaining("precision from 3 to 6");
+                .hasMessageContaining("precision from 3 to 9");
 
         // Test invalid precision (<= 3)
         DataField invalidTimestampField2 =
                 new DataField(2, "timestamp", new TimestampType(false, 2));
         assertThatThrownBy(() -> new IcebergDataField(invalidTimestampField2))
                 .isInstanceOf(IllegalArgumentException.class)
-                .hasMessageContaining("precision from 3 to 6");
+                .hasMessageContaining("precision from 3 to 9");
 
         // 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("precision from 3 to 6");
+                .hasMessageContaining("precision from 3 to 9");
 
         // Test valid precision boundaries
         DataField validTimestamp4 = new DataField(4, "timestamp", new TimestampType(false, 4));