fix(python, rust): check timestamp_ntz in nested fields, add check_can_write in pyarrow writer

crates/core/src/operations/create.rs

@@ -235,16 +235,12 @@ impl CreateBuilder {
             )
         };
 
-        let contains_timestampntz = &self
-            .columns
-            .iter()
-            .any(|f| f.data_type() == &DataType::TIMESTAMPNTZ);
-
+        let contains_timestampntz = PROTOCOL.contains_timestampntz(&self.columns);
         // TODO configure more permissive versions based on configuration. Also how should this ideally be handled?
         // We set the lowest protocol we can, and if subsequent writes use newer features we update metadata?
 
         let (min_reader_version, min_writer_version, writer_features, reader_features) =
-            if *contains_timestampntz {
+            if contains_timestampntz {
                 let mut converted_writer_features = self
                     .configuration
                     .keys()

crates/core/src/operations/transaction/protocol.rs

@@ -4,7 +4,9 @@ use lazy_static::lazy_static;
 use once_cell::sync::Lazy;
 
 use super::{TableReference, TransactionError};
-use crate::kernel::{Action, DataType, EagerSnapshot, ReaderFeatures, Schema, WriterFeatures};
+use crate::kernel::{
+    Action, DataType, EagerSnapshot, ReaderFeatures, Schema, StructField, WriterFeatures,
+};
 use crate::protocol::DeltaOperation;
 use crate::table::state::DeltaTableState;
 
@@ -77,17 +79,30 @@ impl ProtocolChecker {
         Ok(())
     }
 
+    /// checks if table contains timestamp_ntz in any field including nested fields.
+    pub fn contains_timestampntz(&self, fields: &Vec<StructField>) -> bool {
+        fn check_vec_fields(fields: &Vec<StructField>) -> bool {
+            fields.iter().any(|f| _check_type(f.data_type()))
+        }
+
+        fn _check_type(dtype: &DataType) -> bool {
+            match dtype {
+                &DataType::TIMESTAMPNTZ => true,
+                DataType::Array(inner) => _check_type(inner.element_type()),
+                DataType::Struct(inner) => check_vec_fields(inner.fields()),
+                _ => false,
+            }
+        }
+        check_vec_fields(fields)
+    }
+
     /// Check can write_timestamp_ntz
     pub fn check_can_write_timestamp_ntz(
         &self,
         snapshot: &DeltaTableState,
         schema: &Schema,
     ) -> Result<(), TransactionError> {
-        let contains_timestampntz = schema
-            .fields()
-            .iter()
-            .any(|f| f.data_type() == &DataType::TIMESTAMPNTZ);
-
+        let contains_timestampntz = self.contains_timestampntz(schema.fields());
         let required_features: Option<&HashSet<WriterFeatures>> =
             match snapshot.protocol().min_writer_version {
                 0..=6 => None,

python/tests/test_writer.py

@@ -1528,3 +1528,25 @@ def test_rust_decimal_cast(tmp_path: pathlib.Path):
         write_deltalake(
             tmp_path, data, mode="append", schema_mode="merge", engine="rust"
         )
+
+
+@pytest.mark.parametrize(
+    "array",
+    [
+        pa.array([[datetime(2010, 1, 1)]]),
+        pa.array([{"foo": datetime(2010, 1, 1)}]),
+        pa.array([{"foo": [[datetime(2010, 1, 1)]]}]),
+        pa.array([{"foo": [[{"foo": datetime(2010, 1, 1)}]]}]),
+    ],
+)
+def test_write_timestamp_ntz_nested(tmp_path: pathlib.Path, array: pa.array):
+    data = pa.table({"x": array})
+    write_deltalake(tmp_path, data, mode="append", engine="rust")
+
+    dt = DeltaTable(tmp_path)
+
+    protocol = dt.protocol()
+    assert protocol.min_reader_version == 3
+    assert protocol.min_writer_version == 7
+    assert protocol.reader_features == ["timestampNtz"]
+    assert protocol.writer_features == ["timestampNtz"]