fix(spec): validate schema types by format version

Reject schema types that require a newer table format and update the DataFusion catalog registration path to request v3 only when the converted schema needs it. This keeps ordinary CREATE TABLE defaults on v2 while allowing timestamp_ns schemas to pass validation.

Co-authored-by: Codex <codex@openai.com>

Author: manuzhang

crates/iceberg/src/spec/table_metadata.rs

@@ -22,7 +22,7 @@ use std::cmp::Ordering;
 use std::collections::HashMap;
 use std::fmt::{Display, Formatter};
 use std::hash::Hash;
-use std::sync::Arc;
+use std::sync::{Arc, LazyLock};
 
 use _serde::TableMetadataEnum;
 use chrono::{DateTime, Utc};
@@ -33,9 +33,9 @@ use uuid::Uuid;
 use super::snapshot::SnapshotReference;
 pub use super::table_metadata_builder::{TableMetadataBuildResult, TableMetadataBuilder};
 use super::{
-    DEFAULT_PARTITION_SPEC_ID, PartitionSpecRef, PartitionStatisticsFile, SchemaId, SchemaRef,
-    SnapshotRef, SnapshotRetention, SortOrder, SortOrderRef, StatisticsFile, StructType,
-    TableProperties, parse_metadata_file_compression,
+    DEFAULT_PARTITION_SPEC_ID, PartitionSpecRef, PartitionStatisticsFile, PrimitiveType, Schema,
+    SchemaId, SchemaRef, SnapshotRef, SnapshotRetention, SortOrder, SortOrderRef, StatisticsFile,
+    StructType, TableProperties, parse_metadata_file_compression,
 };
 use crate::catalog::MetadataLocation;
 use crate::compression::CompressionCodec;
@@ -60,6 +60,14 @@ pub const MIN_FORMAT_VERSION_ROW_LINEAGE: FormatVersion = FormatVersion::V3;
 /// Reference to [`TableMetadata`].
 pub type TableMetadataRef = Arc<TableMetadata>;
 
+static PRIMITIVE_TYPE_MIN_FORMAT_VERSION: LazyLock<HashMap<PrimitiveType, FormatVersion>> =
+    LazyLock::new(|| {
+        HashMap::from([
+            (PrimitiveType::TimestampNs, FormatVersion::V3),
+            (PrimitiveType::TimestamptzNs, FormatVersion::V3),
+        ])
+    });
+
 #[derive(Debug, PartialEq, Deserialize, Eq, Clone)]
 #[serde(try_from = "TableMetadataEnum")]
 /// Fields for the version 2 of the table metadata.
@@ -1565,6 +1573,25 @@ impl Display for FormatVersion {
     }
 }
 
+/// Returns the minimum table format version required by any type in a schema.
+///
+/// Returns [`None`] when the schema contains no type with a specific minimum
+/// table format version requirement.
+pub fn min_format_version_for_schema(schema: &Schema) -> Option<FormatVersion> {
+    schema
+        .field_id_to_fields()
+        .values()
+        .filter_map(|field| field.field_type.as_primitive_type())
+        .filter_map(min_format_version_for_primitive_type)
+        .max()
+}
+
+pub(crate) fn min_format_version_for_primitive_type(
+    primitive: &PrimitiveType,
+) -> Option<FormatVersion> {
+    PRIMITIVE_TYPE_MIN_FORMAT_VERSION.get(primitive).copied()
+}
+
 #[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone)]
 #[serde(rename_all = "kebab-case")]
 /// Encodes changes to the previous metadata files for the table
@@ -1616,10 +1643,10 @@ mod tests {
     use crate::io::FileIO;
     use crate::spec::table_metadata::TableMetadata;
     use crate::spec::{
-        BlobMetadata, EncryptedKey, INITIAL_ROW_ID, Literal, NestedField, NullOrder, Operation,
-        PartitionSpec, PartitionStatisticsFile, PrimitiveLiteral, PrimitiveType, Schema, Snapshot,
-        SnapshotReference, SnapshotRetention, SortDirection, SortField, SortOrder, StatisticsFile,
-        Summary, TableProperties, Transform, Type, UnboundPartitionField,
+        BlobMetadata, EncryptedKey, INITIAL_ROW_ID, ListType, Literal, NestedField, NullOrder,
+        Operation, PartitionSpec, PartitionStatisticsFile, PrimitiveLiteral, PrimitiveType, Schema,
+        Snapshot, SnapshotReference, SnapshotRetention, SortDirection, SortField, SortOrder,
+        StatisticsFile, Summary, TableProperties, Transform, Type, UnboundPartitionField,
     };
     use crate::{ErrorKind, TableCreation};
 
@@ -3541,6 +3568,28 @@ mod tests {
         )
     }
 
+    fn schema_with_primitive_field(field_type: PrimitiveType) -> Schema {
+        Schema::builder()
+            .with_fields(vec![
+                NestedField::required(1, "ts", Type::Primitive(field_type)).into(),
+            ])
+            .build()
+            .unwrap()
+    }
+
+    fn table_creation_with_format_version(
+        schema: Schema,
+        format_version: FormatVersion,
+    ) -> TableCreation {
+        TableCreation::builder()
+            .location("s3://db/table".to_string())
+            .name("table".to_string())
+            .properties(HashMap::new())
+            .schema(schema)
+            .format_version(format_version)
+            .build()
+    }
+
     #[test]
     fn test_table_metadata_builder_from_table_creation() {
         let table_creation = TableCreation::builder()
@@ -3591,6 +3640,91 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_table_metadata_builder_rejects_v1_v2_nanosecond_timestamp_tables() {
+        for (format_version, primitive_type) in [
+            (FormatVersion::V1, PrimitiveType::TimestampNs),
+            (FormatVersion::V1, PrimitiveType::TimestamptzNs),
+            (FormatVersion::V2, PrimitiveType::TimestampNs),
+            (FormatVersion::V2, PrimitiveType::TimestamptzNs),
+        ] {
+            let table_creation = table_creation_with_format_version(
+                schema_with_primitive_field(primitive_type),
+                format_version,
+            );
+
+            let err = TableMetadataBuilder::from_table_creation(table_creation).unwrap_err();
+
+            assert_eq!(err.kind(), ErrorKind::DataInvalid);
+            assert!(
+                err.message().contains("Invalid type for ts:"),
+                "expected error message to name the invalid column, got {}",
+                err.message()
+            );
+            assert!(
+                err.message().contains("is not supported until v3"),
+                "expected error message to explain v3 requirement, got {}",
+                err.message()
+            );
+        }
+    }
+
+    #[test]
+    fn test_table_metadata_builder_rejects_v2_list_element_requiring_v3() {
+        let schema = Schema::builder()
+            .with_fields(vec![
+                NestedField::required(
+                    1,
+                    "ts_values",
+                    Type::List(ListType::new(
+                        NestedField::list_element(
+                            2,
+                            Type::Primitive(PrimitiveType::TimestampNs),
+                            false,
+                        )
+                        .into(),
+                    )),
+                )
+                .into(),
+            ])
+            .build()
+            .unwrap();
+        let table_creation = table_creation_with_format_version(schema, FormatVersion::V2);
+
+        let err = TableMetadataBuilder::from_table_creation(table_creation).unwrap_err();
+
+        assert_eq!(err.kind(), ErrorKind::DataInvalid);
+        assert!(
+            err.message().contains(
+                "Invalid type for ts_values.element: timestamp_ns is not supported until v3"
+            ),
+            "expected error message to explain nested v3 requirement with column name, got {}",
+            err.message()
+        );
+    }
+
+    #[test]
+    fn test_table_metadata_builder_allows_v3_nanosecond_timestamp_tables() {
+        let schema = Schema::builder()
+            .with_fields(vec![
+                NestedField::required(1, "ts_ns", Type::Primitive(PrimitiveType::TimestampNs))
+                    .into(),
+                NestedField::required(2, "tstz_ns", Type::Primitive(PrimitiveType::TimestamptzNs))
+                    .into(),
+            ])
+            .build()
+            .unwrap();
+        let table_creation = table_creation_with_format_version(schema, FormatVersion::V3);
+
+        let table_metadata = TableMetadataBuilder::from_table_creation(table_creation)
+            .unwrap()
+            .build()
+            .unwrap()
+            .metadata;
+
+        assert_eq!(table_metadata.format_version, FormatVersion::V3);
+    }
+
     #[tokio::test]
     async fn test_table_metadata_read_write() {
         // Create a temporary directory for our test

crates/iceberg/src/spec/table_metadata_builder.rs

@@ -15,7 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::collections::{HashMap, HashSet};
+use std::collections::{BTreeMap, HashMap, HashSet};
 use std::sync::Arc;
 
 use uuid::Uuid;
@@ -28,7 +28,10 @@ use super::{
     UnboundPartitionSpec,
 };
 use crate::error::{Error, ErrorKind, Result};
-use crate::spec::{EncryptedKey, INITIAL_ROW_ID, MIN_FORMAT_VERSION_ROW_LINEAGE};
+use crate::spec::{
+    EncryptedKey, INITIAL_ROW_ID, MIN_FORMAT_VERSION_ROW_LINEAGE,
+    min_format_version_for_primitive_type,
+};
 use crate::{TableCreation, TableUpdate};
 
 pub(crate) const FIRST_FIELD_ID: i32 = 1;
@@ -196,6 +199,41 @@ impl TableMetadataBuilder {
         )
     }
 
+    fn validate_schema_compatible_with_format_version(
+        format_version: FormatVersion,
+        schema: &Schema,
+    ) -> Result<()> {
+        let problems = schema
+            .field_id_to_fields()
+            .values()
+            .filter_map(|field| {
+                let field_type = field.field_type.as_ref();
+                let primitive = field_type.as_primitive_type()?;
+                let min_format_version = min_format_version_for_primitive_type(primitive)?;
+                (format_version < min_format_version).then(|| {
+                    let column_name = schema
+                        .name_by_field_id(field.id)
+                        .unwrap_or(field.name.as_str());
+                    (
+                        field.id,
+                        format!(
+                            "Invalid type for {column_name}: {field_type} is not supported until {min_format_version}"
+                        ),
+                    )
+                })
+            })
+            .collect::<BTreeMap<_, _>>();
+
+        if !problems.is_empty() {
+            return Err(Error::new(
+                ErrorKind::DataInvalid,
+                problems.into_values().collect::<Vec<_>>().join("; "),
+            ));
+        }
+
+        Ok(())
+    }
+
     /// Changes uuid of table metadata.
     pub fn assign_uuid(mut self, uuid: Uuid) -> Self {
         if self.metadata.table_uuid != uuid {
@@ -638,6 +676,11 @@ impl TableMetadataBuilder {
     /// Important: Use this method with caution. The builder does not check
     /// if the added schema is compatible with the current schema.
     pub fn add_schema(mut self, schema: Schema) -> Result<Self> {
+        Self::validate_schema_compatible_with_format_version(
+            self.metadata.format_version,
+            &schema,
+        )?;
+
         // Validate that new schema fields don't conflict with existing partition field names
         self.validate_schema_field_names(&schema)?;
 

crates/integrations/datafusion/src/schema.rs

@@ -29,6 +29,7 @@ use futures::StreamExt;
 use futures::future::try_join_all;
 use iceberg::arrow::arrow_schema_to_schema_auto_assign_ids;
 use iceberg::inspect::MetadataTableType;
+use iceberg::spec::min_format_version_for_schema;
 use iceberg::{Catalog, Error, ErrorKind, NamespaceIdent, Result, TableCreation, TableIdent};
 
 use crate::table::IcebergTableProvider;
@@ -162,12 +163,20 @@ impl SchemaProvider for IcebergSchemaProvider {
         let df_schema = table.schema();
         let iceberg_schema = arrow_schema_to_schema_auto_assign_ids(df_schema.as_ref())
             .map_err(to_datafusion_error)?;
+        let format_version = min_format_version_for_schema(&iceberg_schema);
 
         // Create the table in the Iceberg catalog
-        let table_creation = TableCreation::builder()
-            .name(name.clone())
-            .schema(iceberg_schema)
-            .build();
+        let table_creation = match format_version {
+            Some(format_version) => TableCreation::builder()
+                .name(name.clone())
+                .format_version(format_version)
+                .schema(iceberg_schema)
+                .build(),
+            None => TableCreation::builder()
+                .name(name.clone())
+                .schema(iceberg_schema)
+                .build(),
+        };
 
         let catalog = self.catalog.clone();
         let namespace = self.namespace.clone();
@@ -288,10 +297,11 @@ mod tests {
     use std::sync::Arc;
 
     use datafusion::arrow::array::{Int32Array, StringArray};
-    use datafusion::arrow::datatypes::{DataType, Field, Schema as ArrowSchema};
+    use datafusion::arrow::datatypes::{DataType, Field, Schema as ArrowSchema, TimeUnit};
     use datafusion::arrow::record_batch::RecordBatch;
     use datafusion::datasource::MemTable;
     use iceberg::memory::{MEMORY_CATALOG_WAREHOUSE, MemoryCatalogBuilder};
+    use iceberg::spec::FormatVersion;
     use iceberg::{Catalog, CatalogBuilder, NamespaceIdent};
     use tempfile::TempDir;
 
@@ -375,6 +385,37 @@ mod tests {
         assert!(schema_provider.table_exist("empty_table"));
     }
 
+    #[tokio::test]
+    async fn test_register_timestamp_ns_table_uses_v3() {
+        let (schema_provider, _temp_dir) = create_test_schema_provider().await;
+
+        let arrow_schema = Arc::new(ArrowSchema::new(vec![Field::new(
+            "ts",
+            DataType::Timestamp(TimeUnit::Nanosecond, None),
+            true,
+        )]));
+
+        let empty_batch = RecordBatch::new_empty(arrow_schema.clone());
+        let mem_table = MemTable::try_new(arrow_schema, vec![vec![empty_batch]]).unwrap();
+
+        let result =
+            schema_provider.register_table("timestamp_ns_table".to_string(), Arc::new(mem_table));
+
+        assert!(result.is_ok(), "Expected success, got: {result:?}");
+
+        let table_ident = TableIdent::new(
+            schema_provider.namespace.clone(),
+            "timestamp_ns_table".to_string(),
+        );
+        let table = schema_provider
+            .catalog
+            .load_table(&table_ident)
+            .await
+            .unwrap();
+
+        assert_eq!(FormatVersion::V3, table.metadata().format_version());
+    }
+
     #[tokio::test]
     async fn test_register_duplicate_table_fails() {
         let (schema_provider, _temp_dir) = create_test_schema_provider().await;