[iceberg] Refuse time precisions the mirror cannot publish

docs/docs/iceberg/index.md

@@ -94,6 +94,8 @@ Paimon Iceberg compatibility currently supports the following data types.
 | `BINARY`       | `binary`          |
 | `VARBINARY`    | `binary`          |
 | `DATE`         | `date`            |
+| `TIME` (precision 0-3) | `time`            |
+| `TIME` (other precisions) | not supported  |
 | `TIMESTAMP` (precision 3-6)   | `timestamp`       |
 | `TIMESTAMP_LTZ` (precision 3-6) | `timestamptz`     |
 | `TIMESTAMP` (other precisions)  | not supported     |
@@ -112,6 +114,10 @@ Paimon Iceberg compatibility currently supports the following data types.
   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 Time Types:**
+`TIME` types with a precision above 3 are rejected while Iceberg metadata is enabled: Iceberg
+compatibility publishes only millisecond time values.
+
 **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`.
 - Geospatial columns require Parquet for data, per-level, and changelog files. When Iceberg metadata is enabled, set `metadata.iceberg.format-version` to `3`.

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

@@ -53,6 +53,7 @@
 import org.apache.paimon.types.MapType;
 import org.apache.paimon.types.MultisetType;
 import org.apache.paimon.types.RowType;
+import org.apache.paimon.types.TimeType;
 import org.apache.paimon.types.TimestampType;
 import org.apache.paimon.types.VariantType;
 import org.apache.paimon.types.VectorType;
@@ -124,6 +125,9 @@
 /** Validation utilities for {@link TableSchema}. */
 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. */
     private static final int MIN_ICEBERG_TIMESTAMP_PRECISION = 3;
 
@@ -241,6 +245,7 @@ public static void validateTableSchema(TableSchema schema, Set<String> dynamicOp
         RowType tableRowType = new RowType(schema.fields());
         validateGeospatialTypes(schema, options, tableRowType);
         validateIcebergTimestampPrecisions(tableRowType, options);
+        validateIcebergTimePrecisions(tableRowType, options);
         validateBlobFields(tableRowType, options);
         Set<String> blobDescriptorFields = validateBlobDescriptorFields(tableRowType, options);
         Set<String> blobViewFields =
@@ -561,6 +566,31 @@ public static void validateIcebergTimestampPrecisions(DataType dataType, CoreOpt
                 IcebergOptions.METADATA_ICEBERG_STORAGE.key());
     }
 
+    /**
+     * Refuses the time precisions the mirror cannot publish: it writes whole milliseconds into
+     * Iceberg's microsecond time values, and the conversion enforcing that would only fail once the
+     * snapshot is durable.
+     */
+    public static void validateIcebergTimePrecisions(DataType dataType, CoreOptions options) {
+        if (options.toConfiguration().get(IcebergOptions.METADATA_ICEBERG_STORAGE)
+                == IcebergOptions.StorageType.DISABLED) {
+            return;
+        }
+        checkArgument(
+                !containsType(dataType, SchemaValidation::isUnpublishableTime),
+                "Time columns must have a precision of %s or less when Iceberg metadata is "
+                        + "enabled, the only precisions Iceberg compatibility can publish. Use a "
+                        + "precision of %s or less, or disable '%s'.",
+                MAX_ICEBERG_TIME_PRECISION,
+                MAX_ICEBERG_TIME_PRECISION,
+                IcebergOptions.METADATA_ICEBERG_STORAGE.key());
+    }
+
+    private static boolean isUnpublishableTime(DataType dataType) {
+        return dataType instanceof TimeType
+                && ((TimeType) dataType).getPrecision() > MAX_ICEBERG_TIME_PRECISION;
+    }
+
     private static boolean isUnpublishableTimestamp(DataType dataType) {
         if (dataType instanceof TimestampType) {
             return isUnpublishablePrecision(((TimestampType) dataType).getPrecision());
@@ -587,6 +617,7 @@ public static void validateHistoricalIcebergTypes(
         for (TableSchema schema : history.get()) {
             validateIcebergGeospatialTypes(schema.logicalRowType(), options);
             validateIcebergTimestampPrecisions(schema.logicalRowType(), options);
+            validateIcebergTimePrecisions(schema.logicalRowType(), options);
         }
     }
 

paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerTest.java

@@ -246,6 +246,68 @@ public void testEnablingIcebergMetadataRefusesUnsupportedTimestampPrecisions(int
                 .hasStackTraceContaining("precision from 3 to 6");
     }
 
+    @ParameterizedTest
+    @ValueSource(ints = {4, 6, 9})
+    public void testIcebergMetadataRefusesUnsupportedTimePrecisions(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");
+
+        assertThatThrownBy(
+                        () ->
+                                retryArtificialException(
+                                        () -> manager.createTable(timeSchema(options, precision))))
+                .hasStackTraceContaining("precision of 3 or less");
+
+        assertThatCode(
+                        () ->
+                                retryArtificialException(
+                                        () -> manager.createTable(timeSchema(options, 3))))
+                .doesNotThrowAnyException();
+    }
+
+    private Schema timeSchema(Map<String, String> options, int precision) {
+        return new Schema(
+                Arrays.asList(
+                        new DataField(0, "id", DataTypes.INT()),
+                        new DataField(1, "t", DataTypes.TIME(precision))),
+                Collections.emptyList(),
+                Collections.emptyList(),
+                options,
+                "");
+    }
+
+    @Test
+    public void testEnableIcebergMetadataValidatesHistoricalTimePrecisions() throws Exception {
+        Map<String, String> options = new HashMap<>();
+        options.put(CoreOptions.BUCKET.key(), "-1");
+        Schema micros =
+                new Schema(
+                        Arrays.asList(
+                                new DataField(0, "id", DataTypes.INT()),
+                                new DataField(1, "t", DataTypes.TIME(6))),
+                        Collections.emptyList(),
+                        Collections.emptyList(),
+                        options,
+                        "");
+
+        retryArtificialException(() -> manager.createTable(micros));
+        retryArtificialException(() -> manager.commitChanges(SchemaChange.dropColumn("t")));
+
+        assertThatThrownBy(
+                        () ->
+                                retryArtificialException(
+                                        () ->
+                                                manager.commitChanges(
+                                                        SchemaChange.setOption(
+                                                                IcebergOptions
+                                                                        .METADATA_ICEBERG_STORAGE
+                                                                        .key(),
+                                                                "table-location"))))
+                .hasStackTraceContaining("precision of 3 or less");
+    }
+
     @ParameterizedTest
     @ValueSource(ints = {2, 9})
     public void testEnableIcebergMetadataValidatesHistoricalTimestampPrecisions(int precision)