feat!: reject non-null columns in CREATE TABLE unconditionally (#2404)

## What changes are proposed in this pull request?

Gates non-null columns (nullable: false) on the invariants writer
feature during CREATE TABLE. Delta Spark treats non-nullable schema
columns as implicit invariants, requiring the invariants writer feature
in the protocol. Tables created by kernel with non-null columns but
without the feature cause Spark read failures.

Since invariants is not in ALLOWED_DELTA_FEATURES and is not
auto-enabled by any property or schema-driven check, non-null columns
are effectively rejected unconditionally today. The invariants gate
ensures this relaxes automatically when invariants support is added in
the future.

Core changes:
1/ Added invariants_enabled parameter to validate_schema_for_create with
recursive non-null rejection in SchemaValidator
2/ Overrides transform_variant to skip protocol-mandated non-null
Variant internal fields (metadata, value)
3/ Passes the invariants feature check from validated.writer_features
into schema validation during build()
4 Converted incidental non-null fields to nullable in test schemas that
flow through create_table but are not testing nullability behavior

## How was this change tested?

1/ Added rstest unit coverage in schema/validation.rs for non-null
rejection across schema shapes: top-level, nested struct, array-nested,
map-nested, and invariants-enabled pass cases
2/ Added integration test in create_table/main.rs asserting CREATE TABLE
rejects non-null schemas with top-level and nested cases
3/ Full cargo test -p delta_kernel --all-features --locked passes (0
failures)
4/ cargo clippy and cargo fmt clean

---------

Co-authored-by: Sanuj Basu <sanujbasu@users.noreply.github.com>

Author: sanujbasu

delta-kernel-unity-catalog/src/utils/create_table.rs

@@ -171,7 +171,7 @@ mod tests {
         let table_path = "memory:///test_table/";
         let schema = Arc::new(
             StructType::try_new(vec![
-                StructField::new("id", DataType::INTEGER, false),
+                StructField::new("id", DataType::INTEGER, true),
                 StructField::new("region", DataType::STRING, true),
             ])
             .unwrap(),
@@ -232,7 +232,7 @@ mod tests {
         ]);
         let schema = Arc::new(
             StructType::try_new(vec![
-                StructField::new("id", DataType::INTEGER, false),
+                StructField::new("id", DataType::INTEGER, true),
                 StructField::new("region", DataType::STRING, true),
                 StructField::new("address", DataType::Struct(Box::new(address_struct)), true),
             ])
@@ -278,7 +278,7 @@ mod tests {
         let engine = DefaultEngineBuilder::new(storage).build();
         let table_path = "memory:///test_version_check/";
         let schema = Arc::new(
-            StructType::try_new(vec![StructField::new("id", DataType::INTEGER, false)]).unwrap(),
+            StructType::try_new(vec![StructField::new("id", DataType::INTEGER, true)]).unwrap(),
         );
 
         // Create a table (version 0) and append (version 1)

kernel/src/schema/validation.rs

@@ -20,14 +20,16 @@ const INVALID_PARQUET_CHARS: &[char] = &[' ', ',', ';', '{', '}', '(', ')', '\n'
 /// 1. Schema is non-empty
 /// 2. No duplicate column names (case-insensitive, including nested fields)
 /// 3. Column names contain only valid characters
+/// 4. Rejects non-null columns when the `invariants` writer feature is not enabled
 pub(crate) fn validate_schema_for_create(
     schema: &StructType,
     column_mapping_mode: ColumnMappingMode,
+    invariants_enabled: bool,
 ) -> DeltaResult<()> {
     if schema.num_fields() == 0 {
         return Err(Error::generic("Schema cannot be empty"));
     }
-    let mut validator = SchemaValidator::new(column_mapping_mode);
+    let mut validator = SchemaValidator::new(column_mapping_mode, invariants_enabled);
     // We reuse the SchemaTransform trait for its recursive traversal machinery.
     // The validator never transforms the schema -- it only inspects fields and
     // collects errors. The return value is intentionally discarded.
@@ -46,15 +48,17 @@ pub(crate) fn validate_schema_for_create(
 /// built with `new_unchecked`.
 struct SchemaValidator {
     cm_enabled: bool,
+    invariants_enabled: bool,
     seen_paths: HashSet<String>,
     current_path: Vec<String>,
     errors: Vec<String>,
 }
 
 impl SchemaValidator {
-    fn new(column_mapping_mode: ColumnMappingMode) -> Self {
+    fn new(column_mapping_mode: ColumnMappingMode, invariants_enabled: bool) -> Self {
         Self {
             cm_enabled: !matches!(column_mapping_mode, ColumnMappingMode::None),
+            invariants_enabled,
             seen_paths: HashSet::new(),
             current_path: Vec::new(),
             errors: Vec::new(),
@@ -74,6 +78,15 @@ impl SchemaValidator {
 }
 
 impl<'a> SchemaTransform<'a> for SchemaValidator {
+    /// The default `transform_variant` recurses into the variant's struct fields
+    /// (metadata, value) via `recurse_into_struct`. Those fields are protocol-defined
+    /// and must be non-null -- they are not user-controlled schema columns. We override
+    /// to return the variant struct unchanged, skipping recursion so the non-null check
+    /// in `transform_struct_field` does not reject these fixed internal fields.
+    fn transform_variant(&mut self, stype: &'a StructType) -> Option<Cow<'a, StructType>> {
+        Some(Cow::Borrowed(stype))
+    }
+
     fn transform_struct_field(&mut self, field: &'a StructField) -> Option<Cow<'a, StructField>> {
         if let Err(e) = validate_field_name(field.name(), self.cm_enabled) {
             self.errors.push(e.to_string());
@@ -84,6 +97,13 @@ impl<'a> SchemaTransform<'a> for SchemaValidator {
         // separator would make column "a.b" indistinguishable from nested field b in
         // struct a. Null bytes cannot appear in column names, so they are safe to use.
         self.current_path.push(field.name().to_ascii_lowercase());
+        if !self.invariants_enabled && !field.is_nullable() {
+            self.errors.push(format!(
+                "Non-null column '{}' is not supported during CREATE TABLE unless \
+                 writer feature 'invariants' is enabled",
+                self.current_path.join(".")
+            ));
+        }
         let key = self.current_path.join("\0");
         if !self.seen_paths.insert(key) {
             self.errors.push(format!(
@@ -274,6 +294,57 @@ mod tests {
         ])
     }
 
+    fn schema_all_nullable() -> StructType {
+        StructType::new_unchecked(vec![
+            StructField::new("id", DataType::INTEGER, true),
+            StructField::new("name", DataType::STRING, true),
+        ])
+    }
+
+    fn schema_top_level_non_null() -> StructType {
+        StructType::new_unchecked(vec![
+            StructField::new("id", DataType::INTEGER, false),
+            StructField::new("name", DataType::STRING, true),
+        ])
+    }
+
+    fn schema_nested_non_null() -> StructType {
+        let nested =
+            StructType::new_unchecked(vec![StructField::new("child", DataType::INTEGER, false)]);
+        StructType::new_unchecked(vec![StructField::new(
+            "parent",
+            DataType::Struct(Box::new(nested)),
+            true,
+        )])
+    }
+
+    fn schema_array_nested_non_null() -> StructType {
+        let nested =
+            StructType::new_unchecked(vec![StructField::new("child", DataType::INTEGER, false)]);
+        StructType::new_unchecked(vec![StructField::new(
+            "arr",
+            DataType::Array(Box::new(ArrayType::new(
+                DataType::Struct(Box::new(nested)),
+                true,
+            ))),
+            true,
+        )])
+    }
+
+    fn schema_map_nested_non_null() -> StructType {
+        let nested =
+            StructType::new_unchecked(vec![StructField::new("child", DataType::INTEGER, false)]);
+        StructType::new_unchecked(vec![StructField::new(
+            "map",
+            DataType::Map(Box::new(MapType::new(
+                DataType::STRING,
+                DataType::Struct(Box::new(nested)),
+                true,
+            ))),
+            true,
+        )])
+    }
+
     // === Valid schemas ===
 
     #[rstest]
@@ -283,7 +354,7 @@ mod tests {
     #[case::dot_in_name_with_cm(schema_with_dot(), ColumnMappingMode::Name)]
     #[case::different_struct_children(schema_different_struct_children(), ColumnMappingMode::None)]
     fn valid_schema_accepted(#[case] schema: StructType, #[case] cm: ColumnMappingMode) {
-        assert!(validate_schema_for_create(&schema, cm).is_ok());
+        assert!(validate_schema_for_create(&schema, cm, true).is_ok());
     }
 
     // === Invalid schemas ===
@@ -305,7 +376,7 @@ mod tests {
         #[case] cm: ColumnMappingMode,
         #[case] expected_errs: &[&str],
     ) {
-        let result = validate_schema_for_create(&schema, cm);
+        let result = validate_schema_for_create(&schema, cm, true);
         assert!(result.is_err());
         let err = result.unwrap_err().to_string();
         for expected in expected_errs {
@@ -315,4 +386,64 @@ mod tests {
             );
         }
     }
+
+    #[rstest]
+    #[case::all_nullable_invariants_disabled(schema_all_nullable(), false, true, None)]
+    #[case::top_level_non_null_invariants_disabled(
+        schema_top_level_non_null(),
+        false,
+        false,
+        Some("id")
+    )]
+    #[case::nested_non_null_invariants_disabled(
+        schema_nested_non_null(),
+        false,
+        false,
+        Some("parent.child")
+    )]
+    #[case::array_nested_non_null_invariants_disabled(
+        schema_array_nested_non_null(),
+        false,
+        false,
+        Some("arr.child")
+    )]
+    #[case::map_nested_non_null_invariants_disabled(
+        schema_map_nested_non_null(),
+        false,
+        false,
+        Some("map.child")
+    )]
+    #[case::top_level_non_null_invariants_enabled(schema_top_level_non_null(), true, true, None)]
+    #[case::nested_non_null_invariants_enabled(schema_nested_non_null(), true, true, None)]
+    fn non_null_columns_require_invariants_feature(
+        #[case] schema: StructType,
+        #[case] invariants_enabled: bool,
+        #[case] expect_ok: bool,
+        #[case] expected_path: Option<&str>,
+    ) {
+        let result =
+            validate_schema_for_create(&schema, ColumnMappingMode::None, invariants_enabled);
+
+        if expect_ok {
+            assert!(result.is_ok(), "expected success, got {result:?}");
+            return;
+        }
+
+        let err = result.expect_err("expected non-null validation error");
+        let msg = err.to_string();
+        assert!(
+            msg.contains("Non-null column"),
+            "Expected non-null validation error, got: {msg}"
+        );
+        if let Some(path) = expected_path {
+            assert!(
+                msg.contains(path),
+                "Expected path '{path}' in error message, got: {msg}"
+            );
+        }
+        assert!(
+            msg.contains("writer feature 'invariants'"),
+            "Expected invariants guidance in error message, got: {msg}"
+        );
+    }
 }

kernel/src/snapshot/mod.rs

@@ -3030,7 +3030,7 @@ mod tests {
         let engine = DefaultEngineBuilder::new(storage).build();
         let schema = Arc::new(
             crate::schema::StructType::try_new(vec![
-                crate::schema::StructField::new("id", crate::schema::DataType::INTEGER, false),
+                crate::schema::StructField::new("id", crate::schema::DataType::INTEGER, true),
                 crate::schema::StructField::new("region", crate::schema::DataType::STRING, true),
             ])
             .unwrap(),

kernel/src/transaction/builder/create_table.rs

@@ -18,6 +18,7 @@ use crate::clustering::{create_clustering_domain_metadata, validate_clustering_c
 use crate::committer::Committer;
 use crate::expressions::ColumnName;
 use crate::log_segment::LogSegment;
+use crate::schema::validation::validate_schema_for_create;
 use crate::schema::variant_utils::schema_contains_variant_type;
 use crate::schema::{normalize_column_names_to_schema_casing, DataType, SchemaRef, StructType};
 use crate::snapshot::Snapshot;
@@ -648,7 +649,7 @@ impl CreateTableTransactionBuilder {
     /// # use delta_kernel::schema::{StructType, DataType, StructField};
     /// # use std::sync::Arc;
     /// # fn example() -> delta_kernel::DeltaResult<()> {
-    /// # let schema = Arc::new(StructType::try_new(vec![StructField::new("id", DataType::INTEGER, false)])?);
+    /// # let schema = Arc::new(StructType::try_new(vec![StructField::new("id", DataType::INTEGER, true)])?);
     /// let builder = create_table("/path/to/table", schema, "MyApp/1.0")
     ///     .with_table_properties([
     ///         ("myapp.version", "1.0"),
@@ -693,8 +694,8 @@ impl CreateTableTransactionBuilder {
     /// # use std::sync::Arc;
     /// # fn example() -> delta_kernel::DeltaResult<()> {
     /// # let schema = Arc::new(StructType::try_new(vec![
-    /// #     StructField::new("id", DataType::INTEGER, false),
-    /// #     StructField::new("date", DataType::STRING, false),
+    /// #     StructField::new("id", DataType::INTEGER, true),
+    /// #     StructField::new("date", DataType::STRING, true),
     /// # ])?);
     /// // Clustered layout:
     /// let builder = create_table("/path/to/table", schema.clone(), "MyApp/1.0")
@@ -721,6 +722,7 @@ impl CreateTableTransactionBuilder {
     /// - Checks that the table path is valid
     /// - Verifies the table doesn't already exist
     /// - Validates the schema is non-empty
+    /// - Rejects non-null columns unless `invariants` writer feature is enabled
     /// - Validates the data layout is valid
     /// - Validates table properties against the allow list
     ///
@@ -735,6 +737,7 @@ impl CreateTableTransactionBuilder {
     /// - The table path is invalid
     /// - A table already exists at the given path
     /// - The schema is empty
+    /// - The schema has non-null columns without the `invariants` writer feature
     /// - The data layout is invalid
     /// - Unsupported delta properties or feature flags are specified
     pub fn build(
@@ -761,9 +764,12 @@ impl CreateTableTransactionBuilder {
             maybe_apply_column_mapping_for_table_create(&self.schema, &mut validated)?;
 
         // Validate schema (non-empty, column names, duplicates)
-        crate::schema::validation::validate_schema_for_create(
+        validate_schema_for_create(
             &effective_schema,
             column_mapping_mode,
+            validated
+                .writer_features
+                .contains(&TableFeature::Invariants),
         )?;
 
         // Validate data layout and resolve column names (physical for clustering, logical

kernel/src/transaction/create_table.rs

@@ -15,7 +15,7 @@
 //! # fn example(engine: &dyn Engine) -> delta_kernel::DeltaResult<()> {
 //!
 //! let schema = Arc::new(StructType::try_new(vec![
-//!     StructField::new("id", DataType::INTEGER, false),
+//!     StructField::new("id", DataType::INTEGER, true),
 //! ])?);
 //!
 //! let result = create_table("/path/to/table", schema, "MyApp/1.0")
@@ -72,7 +72,7 @@ use crate::DeltaResult;
 /// # fn example(engine: &dyn Engine) -> delta_kernel::DeltaResult<()> {
 ///
 /// let schema = Arc::new(StructType::try_new(vec![
-///     StructField::new("id", DataType::INTEGER, false),
+///     StructField::new("id", DataType::INTEGER, true),
 /// ])?);
 ///
 /// let result = create_table("/path/to/table", schema, "MyApp/1.0")
@@ -106,7 +106,7 @@ pub type CreateTableTransaction = Transaction<CreateTable>;
 ///
 /// # fn main() -> delta_kernel::DeltaResult<()> {
 /// let schema = Arc::new(StructType::new_unchecked(vec![
-///     StructField::new("id", DataType::INTEGER, false),
+///     StructField::new("id", DataType::INTEGER, true),
 ///     StructField::new("name", DataType::STRING, true),
 /// ]));
 ///

kernel/src/transaction/mod.rs

@@ -2469,7 +2469,7 @@ mod tests {
 
         // Create a non-catalog-managed table using a catalog committer
         let schema = Arc::new(crate::schema::StructType::new_unchecked(vec![
-            crate::schema::StructField::new("id", crate::schema::DataType::INTEGER, false),
+            crate::schema::StructField::new("id", crate::schema::DataType::INTEGER, true),
         ]));
         let committer = Box::new(MockCatalogCommitter);
         let err = create_table("memory:///", schema, "test-engine")