feat: enable "TimestampWithoutTimezone" table feature and add protocol validation for it (delta-io#988)

gotocoding-DB · zachschuermann · web-flow · commit 717fd48d30d0 · 2025-06-01T21:43:44.000-07:00
Resolves delta-io#987. And adds protocol validation: if the provided table has "TIMESTAMP_NTZ" column, it should have "timestampNtz" reader/writer features. ## What changes are proposed in this pull request? Support `TIMESTAMP_NTZ` column type for writes. ## How was this change tested? * Added test to `kernel/tests/write.rs` * Added test to new `kernel/src/table_features/timestamp_ntz.rs` --------- Co-authored-by: Zach Schuermann <zachary.zvs@gmail.com>
diff --git a/kernel/src/table_configuration.rs b/kernel/src/table_configuration.rs
@@ -15,8 +15,8 @@ use url::Url;
 use crate::actions::{ensure_supported_features, Metadata, Protocol};
 use crate::schema::{InvariantChecker, SchemaRef};
 use crate::table_features::{
-    column_mapping_mode, validate_schema_column_mapping, ColumnMappingMode, ReaderFeature,
-    WriterFeature,
+    column_mapping_mode, validate_schema_column_mapping, validate_timestamp_ntz_feature_support,
+    ColumnMappingMode, ReaderFeature, WriterFeature,
 };
 use crate::table_properties::TableProperties;
 use crate::{DeltaResult, Error, Version};
@@ -78,6 +78,9 @@ impl TableConfiguration {
 
         // validate column mapping mode -- all schema fields should be correctly (un)annotated
         validate_schema_column_mapping(&schema, column_mapping_mode)?;
+
+        validate_timestamp_ntz_feature_support(&schema, &protocol)?;
+
         Ok(Self {
             schema,
             metadata,
@@ -574,4 +577,55 @@ mod test {
         );
         assert_eq!(new_table_config.table_root(), table_config.table_root());
     }
+
+    #[test]
+    fn test_timestamp_ntz_validation_integration() {
+        // Schema with TIMESTAMP_NTZ column
+        let schema_string = r#"{"type":"struct","fields":[{"name":"ts","type":"timestamp_ntz","nullable":true,"metadata":{}}]}"#.to_string();
+        let metadata = Metadata {
+            schema_string,
+            ..Default::default()
+        };
+
+        let protocol_without_timestamp_ntz_features = Protocol::try_new(
+            3,
+            7,
+            Some::<Vec<String>>(vec![]),
+            Some::<Vec<String>>(vec![]),
+        )
+        .unwrap();
+
+        let protocol_with_timestamp_ntz_features = Protocol::try_new(
+            3,
+            7,
+            Some([ReaderFeature::TimestampWithoutTimezone]),
+            Some([WriterFeature::TimestampWithoutTimezone]),
+        )
+        .unwrap();
+
+        let table_root = Url::try_from("file:///").unwrap();
+
+        let result = TableConfiguration::try_new(
+            metadata.clone(),
+            protocol_without_timestamp_ntz_features,
+            table_root.clone(),
+            0,
+        );
+        assert!(
+            result.is_err(),
+            "Should fail when TIMESTAMP_NTZ is used without required features"
+        );
+        assert!(result.unwrap_err().to_string().contains("timestampNtz"));
+
+        let result = TableConfiguration::try_new(
+            metadata,
+            protocol_with_timestamp_ntz_features,
+            table_root,
+            0,
+        );
+        assert!(
+            result.is_ok(),
+            "Should succeed when TIMESTAMP_NTZ is used with required features"
+        );
+    }
 }
diff --git a/kernel/src/table_features/mod.rs b/kernel/src/table_features/mod.rs
@@ -8,7 +8,9 @@ use crate::schema::DataType;
 
 pub(crate) use column_mapping::column_mapping_mode;
 pub use column_mapping::{validate_schema_column_mapping, ColumnMappingMode};
+pub(crate) use timestamp_ntz::validate_timestamp_ntz_feature_support;
 mod column_mapping;
+mod timestamp_ntz;
 
 /// Reader features communicate capabilities that must be implemented in order to correctly read a
 /// given table. That is, readers must implement and respect all features listed in a table's
@@ -167,6 +169,7 @@ pub(crate) static SUPPORTED_WRITER_FEATURES: LazyLock<Vec<WriterFeature>> = Lazy
         WriterFeature::AppendOnly,
         WriterFeature::DeletionVectors,
         WriterFeature::Invariants,
+        WriterFeature::TimestampWithoutTimezone,
     ]
 });
 
diff --git a/kernel/src/table_features/timestamp_ntz.rs b/kernel/src/table_features/timestamp_ntz.rs
@@ -0,0 +1,133 @@
+//! Validation for TIMESTAMP_NTZ feature support
+
+use super::{ReaderFeature, WriterFeature};
+use crate::actions::Protocol;
+use crate::schema::{PrimitiveType, Schema, SchemaTransform};
+use crate::utils::require;
+use crate::{DeltaResult, Error};
+
+use std::borrow::Cow;
+
+/// Validates that if a table schema contains TIMESTAMP_NTZ columns, the table must have the
+/// TimestampWithoutTimezone feature in both reader and writer features.
+pub(crate) fn validate_timestamp_ntz_feature_support(
+    schema: &Schema,
+    protocol: &Protocol,
+) -> DeltaResult<()> {
+    if !protocol.has_reader_feature(&ReaderFeature::TimestampWithoutTimezone)
+        || !protocol.has_writer_feature(&WriterFeature::TimestampWithoutTimezone)
+    {
+        let mut uses_timestamp_ntz = UsesTimestampNtz(false);
+        let _ = uses_timestamp_ntz.transform_struct(schema);
+        require!(
+            !uses_timestamp_ntz.0,
+            Error::unsupported(
+                "Table contains TIMESTAMP_NTZ columns but does not have the required 'timestampNtz' feature in reader and writer features"
+            )
+        );
+    }
+    Ok(())
+}
+
+/// Schema visitor that checks if any column in the schema uses TIMESTAMP_NTZ type
+struct UsesTimestampNtz(bool);
+
+impl<'a> SchemaTransform<'a> for UsesTimestampNtz {
+    fn transform_primitive(&mut self, ptype: &'a PrimitiveType) -> Option<Cow<'a, PrimitiveType>> {
+        if *ptype == PrimitiveType::TimestampNtz {
+            self.0 = true;
+        }
+        None
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::actions::Protocol;
+    use crate::schema::{DataType, PrimitiveType, StructField, StructType};
+    use crate::table_features::{ReaderFeature, WriterFeature};
+
+    #[test]
+    fn test_timestamp_ntz_feature_validation() {
+        let schema_with_timestamp_ntz = StructType::new([
+            StructField::new("id", DataType::INTEGER, false),
+            StructField::new("ts", DataType::Primitive(PrimitiveType::TimestampNtz), true),
+        ]);
+
+        let schema_without_timestamp_ntz = StructType::new([
+            StructField::new("id", DataType::INTEGER, false),
+            StructField::new("name", DataType::STRING, true),
+        ]);
+
+        // Protocol with TimestampWithoutTimezone features
+        let protocol_with_features = Protocol::try_new(
+            3,
+            7,
+            Some([ReaderFeature::TimestampWithoutTimezone]),
+            Some([WriterFeature::TimestampWithoutTimezone]),
+        )
+        .unwrap();
+
+        // Protocol without TimestampWithoutTimezone features
+        let protocol_without_features = Protocol::try_new(
+            3,
+            7,
+            Some::<Vec<String>>(vec![]),
+            Some::<Vec<String>>(vec![]),
+        )
+        .unwrap();
+
+        // Schema with TIMESTAMP_NTZ + Protocol with features = OK
+        validate_timestamp_ntz_feature_support(&schema_with_timestamp_ntz, &protocol_with_features)
+            .expect("Should succeed when features are present");
+
+        // Schema without TIMESTAMP_NTZ + Protocol without features = OK
+        validate_timestamp_ntz_feature_support(
+            &schema_without_timestamp_ntz,
+            &protocol_without_features,
+        )
+        .expect("Should succeed when no TIMESTAMP_NTZ columns are present");
+
+        // Schema without TIMESTAMP_NTZ + Protocol with features = OK
+        validate_timestamp_ntz_feature_support(
+            &schema_without_timestamp_ntz,
+            &protocol_with_features,
+        )
+        .expect("Should succeed when no TIMESTAMP_NTZ columns are present, even with features");
+
+        // Schema with TIMESTAMP_NTZ + Protocol without features = ERROR
+        let result = validate_timestamp_ntz_feature_support(
+            &schema_with_timestamp_ntz,
+            &protocol_without_features,
+        );
+        assert!(
+            result.is_err(),
+            "Should fail when TIMESTAMP_NTZ columns are present but features are missing"
+        );
+        assert!(result.unwrap_err().to_string().contains("timestampNtz"));
+
+        // Nested schema with TIMESTAMP_NTZ
+        let nested_schema_with_timestamp_ntz = StructType::new([
+            StructField::new("id", DataType::INTEGER, false),
+            StructField::new(
+                "nested",
+                DataType::Struct(Box::new(StructType::new([StructField::new(
+                    "inner_ts",
+                    DataType::Primitive(PrimitiveType::TimestampNtz),
+                    true,
+                )]))),
+                true,
+            ),
+        ]);
+
+        let result = validate_timestamp_ntz_feature_support(
+            &nested_schema_with_timestamp_ntz,
+            &protocol_without_features,
+        );
+        assert!(
+            result.is_err(),
+            "Should fail for nested TIMESTAMP_NTZ columns when features are missing"
+        );
+    }
+}
diff --git a/kernel/tests/write.rs b/kernel/tests/write.rs
@@ -2,7 +2,7 @@ use std::collections::HashMap;
 use std::sync::Arc;
 
 use delta_kernel::arrow::array::{
-    Int32Array, MapBuilder, MapFieldNames, StringArray, StringBuilder,
+    Int32Array, MapBuilder, MapFieldNames, StringArray, StringBuilder, TimestampMicrosecondArray,
 };
 use delta_kernel::arrow::datatypes::{DataType as ArrowDataType, Field, Schema as ArrowSchema};
 use delta_kernel::arrow::error::ArrowError;
@@ -63,17 +63,28 @@ async fn create_table(
     schema: SchemaRef,
     partition_columns: &[&str],
     use_37_protocol: bool,
+    enable_timestamp_without_timezone: bool,
 ) -> Result<Table, Box<dyn std::error::Error>> {
     let table_id = "test_id";
     let schema = serde_json::to_string(&schema)?;
 
+    let (reader_features, writer_features) = {
+        let mut reader_features = vec![];
+        let mut writer_features = vec![];
+        if enable_timestamp_without_timezone {
+            reader_features.push("timestampNtz");
+            writer_features.push("timestampNtz");
+        }
+        (reader_features, writer_features)
+    };
+
     let protocol = if use_37_protocol {
         json!({
             "protocol": {
                 "minReaderVersion": 3,
                 "minWriterVersion": 7,
-                "readerFeatures": [],
-                "writerFeatures": []
+                "readerFeatures": reader_features,
+                "writerFeatures": writer_features,
             }
         })
     } else {
@@ -175,6 +186,7 @@ async fn setup_tables(
                 schema.clone(),
                 partition_columns,
                 true,
+                false,
             )
             .await?,
             engine_37,
@@ -188,6 +200,7 @@ async fn setup_tables(
                 schema,
                 partition_columns,
                 false,
+                false,
             )
             .await?,
             engine_11,
@@ -837,3 +850,88 @@ async fn test_write_txn_actions() -> Result<(), Box<dyn std::error::Error>> {
     }
     Ok(())
 }
+
+#[tokio::test]
+async fn test_append_timestamp_ntz() -> Result<(), Box<dyn std::error::Error>> {
+    // setup tracing
+    let _ = tracing_subscriber::fmt::try_init();
+
+    // create a table with TIMESTAMP_NTZ column
+    let schema = Arc::new(StructType::new(vec![StructField::nullable(
+        "ts_ntz",
+        DataType::TIMESTAMP_NTZ,
+    )]));
+
+    let (store, engine, table_location) = setup("test_table_timestamp_ntz", true);
+    let table = create_table(
+        store.clone(),
+        table_location,
+        schema.clone(),
+        &[],
+        true,
+        true, // enable "timestamp without timezone" feature
+    )
+    .await?;
+
+    let commit_info = new_commit_info()?;
+
+    let mut txn = table
+        .new_transaction(&engine)?
+        .with_commit_info(commit_info);
+
+    // Create Arrow data with TIMESTAMP_NTZ values including edge cases
+    // These are microseconds since Unix epoch
+    let timestamp_values = vec![
+        0i64,                  // Unix epoch (1970-01-01T00:00:00.000000)
+        1634567890123456i64,   // 2021-10-18T12:31:30.123456
+        1634567950654321i64,   // 2021-10-18T12:32:30.654321
+        1672531200000000i64,   // 2023-01-01T00:00:00.000000
+        253402300799999999i64, // 9999-12-31T23:59:59.999999 (near max valid timestamp)
+        -62135596800000000i64, // 0001-01-01T00:00:00.000000 (near min valid timestamp)
+    ];
+
+    let data = RecordBatch::try_new(
+        Arc::new(schema.as_ref().try_into_arrow()?),
+        vec![Arc::new(TimestampMicrosecondArray::from(timestamp_values))],
+    )?;
+
+    // Write data
+    let engine = Arc::new(engine);
+    let write_context = Arc::new(txn.get_write_context());
+
+    let write_metadata = engine
+        .write_parquet(
+            &ArrowEngineData::new(data.clone()),
+            write_context.as_ref(),
+            HashMap::new(),
+            true,
+        )
+        .await?;
+
+    txn.add_write_metadata(write_metadata);
+
+    // Commit the transaction
+    txn.commit(engine.as_ref())?;
+
+    // Verify the commit was written correctly
+    let commit1 = store
+        .get(&Path::from(
+            "/test_table_timestamp_ntz/_delta_log/00000000000000000001.json",
+        ))
+        .await?;
+
+    let parsed_commits: Vec<_> = Deserializer::from_slice(&commit1.bytes().await?)
+        .into_iter::<serde_json::Value>()
+        .try_collect()?;
+
+    // Check that we have the expected number of commits (commitInfo + add)
+    assert_eq!(parsed_commits.len(), 2);
+
+    // Check that the add action exists
+    assert!(parsed_commits[1].get("add").is_some());
+
+    // Verify the data can be read back correctly
+    test_read(&ArrowEngineData::new(data), &table, engine)?;
+
+    Ok(())
+}
