From 6becfd321b83cd28012fdf6ba9c294ef7f6e1c3f Mon Sep 17 00:00:00 2001
From: Arnav Balyan <arnavbalyan1@gmail.com>
Date: Mon, 18 May 2026 17:12:17 +0530
Subject: [PATCH] update

---
 .../paimon/schema/SchemaValidation.java       | 29 +++++++++++++
 .../paimon/schema/SchemaValidationTest.java   | 41 +++++++++++++++++++
 2 files changed, 70 insertions(+)

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 13ee86cdd0cb..4ffc3ec0259e 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
@@ -186,6 +186,35 @@ public static void validateTableSchema(TableSchema schema) {
         }
         fileFormat.validateDataFields(new RowType(fieldsInNormalFile));
 
+        for (Map.Entry<Integer, String> entry : options.fileFormatPerLevel().entrySet()) {
+            if (!"avro".equalsIgnoreCase(entry.getValue())) {
+                continue;
+            }
+            for (DataField field : fieldsInNormalFile) {
+                DataType type = field.type();
+                int precision = -1;
+                if (type instanceof TimestampType) {
+                    precision = ((TimestampType) type).getPrecision();
+                } else if (type instanceof LocalZonedTimestampType) {
+                    precision = ((LocalZonedTimestampType) type).getPrecision();
+                }
+                if (precision > 6) {
+                    throw new IllegalArgumentException(
+                            String.format(
+                                    "'%s' entry '%d:avro' is incompatible with column '%s' of type %s: "
+                                            + "Avro supports timestamp precision up to 6, got %d. "
+                                            + "Either lower the column precision, drop the per-level mapping for level %d, "
+                                            + "or use a different format (parquet or orc) for that level.",
+                                    CoreOptions.FILE_FORMAT_PER_LEVEL.key(),
+                                    entry.getKey(),
+                                    field.name(),
+                                    type,
+                                    precision,
+                                    entry.getKey()));
+                }
+            }
+        }
+
         // Check column names in schema
         schema.fieldNames()
                 .forEach(
diff --git a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
index cec075724f47..d518f79a20f5 100644
--- a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
@@ -444,4 +444,45 @@ private void validateTableSchemaWithMapField(Map<String, String> options) {
         validateTableSchema(
                 new TableSchema(1, fields, 10, emptyList(), singletonList("f1"), options, ""));
     }
+
+    @Test
+    public void testFileFormatPerLevelRejectsIncompatibleSchema() {
+        List<DataField> fields =
+                Arrays.asList(
+                        new DataField(0, "k", DataTypes.INT()),
+                        new DataField(1, "v", DataTypes.TIMESTAMP(9)));
+        Map<String, String> options = new HashMap<>();
+        options.put(BUCKET.key(), String.valueOf(-1));
+        options.put(CoreOptions.FILE_FORMAT_PER_LEVEL.key(), "0:avro");
+
+        assertThatThrownBy(
+                        () ->
+                                validateTableSchema(
+                                        new TableSchema(
+                                                1,
+                                                fields,
+                                                10,
+                                                emptyList(),
+                                                singletonList("k"),
+                                                options,
+                                                "")))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("file.format.per.level")
+                .hasMessageContaining("0:avro")
+                .hasMessageContaining("TIMESTAMP");
+    }
+
+    @Test
+    public void testFileFormatPerLevelAcceptsCompatibleSchema() {
+        List<DataField> fields =
+                Arrays.asList(
+                        new DataField(0, "k", DataTypes.INT()),
+                        new DataField(1, "v", DataTypes.TIMESTAMP(9)));
+        Map<String, String> options = new HashMap<>();
+        options.put(BUCKET.key(), String.valueOf(-1));
+        options.put(CoreOptions.FILE_FORMAT_PER_LEVEL.key(), "0:parquet");
+
+        validateTableSchema(
+                new TableSchema(1, fields, 10, emptyList(), singletonList("k"), options, ""));
+    }
 }