Prevent creating table metadata with nanosecond timestamps before v3

api/src/main/java/org/apache/iceberg/Schema.java

@@ -54,6 +54,8 @@ public class Schema implements Serializable {
   private static final Joiner NEWLINE = Joiner.on('\n');
   private static final String ALL_COLUMNS = "*";
   private static final int DEFAULT_SCHEMA_ID = 0;
+  private static final Map<Type.TypeID, Integer> MIN_FORMAT_VERSIONS =
+      ImmutableMap.of(Type.TypeID.TIMESTAMP_NANO, 3);
 
   private final StructType struct;
   private final int schemaId;
@@ -573,4 +575,27 @@ private List<NestedField> reassignIds(List<NestedField> columns, TypeUtil.GetID
             });
     return res.asStructType().fields();
   }
+
+  /**
+   * Check the compatibility of the schema with a format version.
+   *
+   * <p>This validates that the schema does not contain types that were released in later format
+   * versions.
+   *
+   * @param schema a Schema
+   * @param formatVersion table format version
+   */
+  public static void checkCompatibility(Schema schema, int formatVersion) {
+    // check the type in each field
+    for (NestedField field : schema.lazyIdToField().values()) {
+      Integer minFormatVersion = MIN_FORMAT_VERSIONS.get(field.type().typeId());
+      Preconditions.checkState(
+          minFormatVersion == null || formatVersion >= minFormatVersion,
+          "Invalid type in v%s schema: %s %s is not supported until v%s",
+          formatVersion,
+          schema.findColumnName(field.fieldId()),
+          field.type(),
+          minFormatVersion);
+    }
+  }
 }

core/src/main/java/org/apache/iceberg/TableMetadata.java

@@ -51,7 +51,7 @@ public class TableMetadata implements Serializable {
   static final long INITIAL_SEQUENCE_NUMBER = 0;
   static final long INVALID_SEQUENCE_NUMBER = -1;
   static final int DEFAULT_TABLE_FORMAT_VERSION = 2;
-  static final int SUPPORTED_TABLE_FORMAT_VERSION = 2;
+  static final int SUPPORTED_TABLE_FORMAT_VERSION = 3;
   static final int INITIAL_SPEC_ID = 0;
   static final int INITIAL_SORT_ORDER_ID = 1;
   static final int INITIAL_SCHEMA_ID = 0;
@@ -1489,6 +1489,8 @@ private int addSchemaInternal(Schema schema, int newLastColumnId) {
           newLastColumnId,
           lastColumnId);
 
+      Schema.checkCompatibility(schema, formatVersion);
+
       int newSchemaId = reuseOrCreateNewSchemaId(schema);
       boolean schemaFound = schemasById.containsKey(newSchemaId);
       if (schemaFound && newLastColumnId == lastColumnId) {

core/src/test/java/org/apache/iceberg/TestTableMetadata.java

@@ -61,6 +61,7 @@
 import org.apache.iceberg.transforms.Transforms;
 import org.apache.iceberg.types.Types;
 import org.apache.iceberg.util.JsonUtil;
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
 
@@ -1627,4 +1628,54 @@ public void buildReplacementKeepsSnapshotLog() throws Exception {
         .hasSize(2)
         .containsExactlyElementsOf(metadata.snapshotLog());
   }
+
+  @Test
+  public void testConstructV3Metadata() {
+    TableMetadata.newTableMetadata(
+        TEST_SCHEMA,
+        PartitionSpec.unpartitioned(),
+        SortOrder.unsorted(),
+        TEST_LOCATION,
+        ImmutableMap.of(),
+        3);
+  }
+
+  @Test
+  public void testV3TimestampNanoTypeSupport() {
+    Schema v3Schema =
+        new Schema(
+            Types.NestedField.required(3, "id", Types.LongType.get()),
+            Types.NestedField.required(4, "data", Types.StringType.get()),
+            Types.NestedField.required(
+                5,
+                "struct",
+                Types.StructType.of(
+                    Types.NestedField.optional(
+                        6, "ts_nanos", Types.TimestampNanoType.withZone()))));
+
+    for (int unsupportedFormatVersion : ImmutableList.of(1, 2)) {
+      Assertions.assertThrows(
+          IllegalStateException.class,
+          () ->
+              TableMetadata.newTableMetadata(
+                  v3Schema,
+                  PartitionSpec.unpartitioned(),
+                  SortOrder.unsorted(),
+                  TEST_LOCATION,
+                  ImmutableMap.of(),
+                  unsupportedFormatVersion),
+          String.format(
+              "Invalid type in v%s schema: struct.ts_nanos timestamptz_ns is not supported until v3",
+              unsupportedFormatVersion));
+    }
+
+    // should be allowed in v3
+    TableMetadata.newTableMetadata(
+        v3Schema,
+        PartitionSpec.unpartitioned(),
+        SortOrder.unsorted(),
+        TEST_LOCATION,
+        ImmutableMap.of(),
+        3);
+  }
 }

core/src/test/resources/TableMetadataUnsupportedVersion.json

@@ -1,5 +1,5 @@
 {
-  "format-version": 3,
+  "format-version": 10,
   "table-uuid": "d20125c8-7284-442c-9aea-15fee620737c",
   "location": "s3://bucket/test/location",
   "last-updated-ms": 1602638573874,
@@ -33,4 +33,4 @@
   "properties": {},
   "current-snapshot-id": -1,
   "snapshots": []
-}
+}