feat: Make tags and remove partition values allow null values in map (delta-io#2281)

## What changes are proposed in this pull request?

- `remove.tags`, `remove.partitionValues`, `checkpointMetadata.tags`,
`cdc.tags`, and `sidecar.tags` all allow null values in map. These null
values will be ignored when read by kernel.
- Extend the `#[allow_null_container_values]` derive macro attribute to
support `Option<HashMap<...>>`
  fields

The Delta protocol allows null values in partitionValues maps (a null
partition value means the
partition column is null for that file) and in tags maps. The Add action
schema already handled this
correctly, but Remove and other actions did not, causing Arrow
validation errors when reading actions
  with null map values.


## How was this change tested?
- New parameterized test read_actions_with_null_map_values covers null
map values across all action
  types (remove, add, cdc, sidecar, checkpointMetadata). 
- NOTE: commitInfo.operationParameters and metadata.configuration are
left as future work.
- verified `#[allow_null_container_values]` works on
Option<HashMap<...>> and fails on other non-hashmap Optionals.
  - Existing schema tests updated to expect nullable map values

Author: OussamaSaoudi

derive-macros/src/lib.rs

@@ -82,6 +82,26 @@ fn get_schema_name(name: &Ident) -> Ident {
     Ident::new(&ret, name.span())
 }
 
+/// Check if a path segment is `Option<HashMap<K, V>>`.
+fn is_option_of_hashmap(seg: &syn::PathSegment) -> bool {
+    if seg.ident != "Option" {
+        return false;
+    }
+    let PathArguments::AngleBracketed(angle_args) = &seg.arguments else {
+        return false;
+    };
+    // Option has exactly one type argument
+    let Some(syn::GenericArgument::Type(Type::Path(inner_type))) = angle_args.args.first() else {
+        return false;
+    };
+    // Check if the inner type's last segment is HashMap
+    inner_type
+        .path
+        .segments
+        .last()
+        .is_some_and(|seg| seg.ident == "HashMap")
+}
+
 fn gen_schema_fields(data: &Data) -> TokenStream {
     let fields = match data {
         Data::Struct(DataStruct {
@@ -119,11 +139,13 @@ fn gen_schema_fields(data: &Data) -> TokenStream {
                     }
                 });
                 if have_schema_null {
-                    if let Some(last_ident) = type_path.path.segments.last().map(|seg| &seg.ident) {
-                        if last_ident != "HashMap" {
-                           return Error::new(
-                                last_ident.span(),
-                                format!("Can only use allow_null_container_values on HashMap fields, not {last_ident}")
+                    if let Some(last_seg) = type_path.path.segments.last() {
+                        let is_valid =
+                            last_seg.ident == "HashMap" || is_option_of_hashmap(last_seg);
+                        if !is_valid {
+                            return Error::new(
+                                last_seg.ident.span(),
+                                format!("Can only use allow_null_container_values on HashMap or Option<HashMap> fields, not {}", last_seg.ident)
                             ).to_compile_error()
                         }
                     }

kernel/src/actions/mod.rs

@@ -786,7 +786,14 @@ pub(crate) struct Remove {
     #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
     pub(crate) extended_file_metadata: Option<bool>,
 
-    /// A map from partition column to value for this logical file.
+    /// A map from partition column to value for this logical file. This map can contain null in
+    /// the values meaning a partition is null. We drop those values from this map, due to the
+    /// `allow_null_container_values` annotation allowing them and because [`materialize`] drops
+    /// null values. This means an engine can assume that if a partition is found in
+    /// [`Metadata::partition_columns`] but not in this map, its value is null.
+    ///
+    /// [`materialize`]: crate::engine_data::EngineMap::materialize
+    #[allow_null_container_values]
     #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
     pub(crate) partition_values: Option<HashMap<String, String>>,
 
@@ -800,7 +807,8 @@ pub(crate) struct Remove {
     #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
     pub stats: Option<String>,
 
-    /// Map containing metadata about this logical file.
+    /// Map containing metadata about this logical file. Values can be null.
+    #[allow_null_container_values]
     #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
     pub(crate) tags: Option<HashMap<String, String>>,
 
@@ -850,7 +858,8 @@ pub(crate) struct Cdc {
     /// data of the table
     pub data_change: bool,
 
-    /// Map containing metadata about this logical file.
+    /// Map containing metadata about this logical file. Values can be null.
+    #[allow_null_container_values]
     pub tags: Option<HashMap<String, String>>,
 }
 
@@ -900,7 +909,8 @@ pub(crate) struct Sidecar {
     /// The time this logical file was created, as milliseconds since the epoch.
     pub modification_time: i64,
 
-    /// A map containing any additional metadata about the logicial file.
+    /// A map containing any additional metadata about the logical file. Values can be null.
+    #[allow_null_container_values]
     pub tags: Option<HashMap<String, String>>,
 }
 
@@ -937,7 +947,8 @@ pub(crate) struct CheckpointMetadata {
     /// See issue #786 for tracking progress.
     pub(crate) version: i64,
 
-    /// Map containing any additional metadata about the V2 spec checkpoint.
+    /// Map containing any additional metadata about the V2 spec checkpoint. Values can be null.
+    #[allow_null_container_values]
     pub(crate) tags: Option<HashMap<String, String>>,
 }
 
@@ -1153,14 +1164,14 @@ mod tests {
     fn tags_field() -> StructField {
         StructField::nullable(
             "tags",
-            MapType::new(DataType::STRING, DataType::STRING, false),
+            MapType::new(DataType::STRING, DataType::STRING, true),
         )
     }
 
     fn partition_values_field() -> StructField {
         StructField::nullable(
             "partitionValues",
-            MapType::new(DataType::STRING, DataType::STRING, false),
+            MapType::new(DataType::STRING, DataType::STRING, true),
         )
     }
 

kernel/src/doctests/to_schema.rs

@@ -52,6 +52,33 @@ pub struct MacroTestStructWithAttributedField;
 #[cfg(doctest)]
 pub struct MacroTestStructWithInvalidAttributeTarget;
 
+/// Verify that `#[allow_null_container_values]` works on `Option<HashMap<_, _>>` fields.
+/// This is needed for optional map fields like `Remove.partition_values` that can contain
+/// null values.
+/// ```
+/// # use delta_kernel_derive::ToSchema;
+/// # use std::collections::HashMap;
+/// #[derive(ToSchema)]
+/// pub struct WithOptionalAttributedField {
+///     #[allow_null_container_values]
+///     map_field: Option<HashMap<String, String>>,
+/// }
+/// ```
+#[cfg(doctest)]
+pub struct MacroTestStructWithOptionalAttributedField;
+
+/// Verify that `#[allow_null_container_values]` fails on `Option<_>` fields that are not maps.
+/// ```compile_fail
+/// # use delta_kernel_derive::ToSchema;
+/// #[derive(ToSchema)]
+/// pub struct WithInvalidOptionalAttributeTarget {
+///     #[allow_null_container_values]
+///     some_name: Option<String>,
+/// }
+/// ```
+#[cfg(doctest)]
+pub struct MacroTestStructWithInvalidOptionalAttributeTarget;
+
 /// ```compile_fail
 /// # use delta_kernel_derive::ToSchema;
 /// # use syn::Token;

kernel/src/log_segment/tests.rs

@@ -4233,3 +4233,155 @@ fn test_schema_to_is_not_null_predicate(
 ) {
     assert_eq!(schema_to_is_not_null_predicate(&schema), expected);
 }
+
+/// Verify that `read_actions` correctly handles null values in map fields across all
+/// action types. The Delta protocol allows null values in `partitionValues` maps (a null
+/// partition value means the partition column is null for that file) and in `tags` maps.
+///
+/// Spark defaults all `Map[String, String]` types to `valueContainsNull = true`, and
+/// checkpoint writing calls `schema.asNullable` which forces all maps nullable. The
+/// schema must match this behavior.
+///
+/// This test reads JSON actions through `DefaultEngine` + `InMemory` store +
+/// `log_segment.read_actions()`, then re-validates the resulting Arrow `StructArray` with
+/// `StructArray::try_new`. Without the fix, non-nullable map value fields cause:
+///   "Found unmasked nulls for non-nullable StructArray field 'value'"
+#[rstest]
+// remove.partitionValues.month: null
+#[case::remove_partition_values(
+    "remove",
+    "partitionValues",
+    r#"{"remove":{"path":"file.parquet","deletionTimestamp":1000,"dataChange":true,"extendedFileMetadata":true,"partitionValues":{"year":"2024","month":null},"size":100}}"#
+)]
+// remove.tags.key2: null
+#[case::remove_tags(
+    "remove",
+    "tags",
+    r#"{"remove":{"path":"file.parquet","deletionTimestamp":1000,"dataChange":true,"tags":{"key1":"val1","key2":null}}}"#
+)]
+// add.partitionValues.month: null
+#[case::add_partition_values(
+    "add",
+    "partitionValues",
+    r#"{"add":{"path":"file.parquet","partitionValues":{"year":"2024","month":null},"size":100,"modificationTime":1000,"dataChange":true}}"#
+)]
+// add.tags.key2: null
+#[case::add_tags(
+    "add",
+    "tags",
+    r#"{"add":{"path":"file.parquet","partitionValues":{},"size":100,"modificationTime":1000,"dataChange":true,"tags":{"key1":"val1","key2":null}}}"#
+)]
+// cdc.partitionValues.month: null
+#[case::cdc_partition_values(
+    "cdc",
+    "partitionValues",
+    r#"{"cdc":{"path":"file.parquet","partitionValues":{"year":"2024","month":null},"size":100,"dataChange":false}}"#
+)]
+// cdc.tags.key2: null
+#[case::cdc_tags(
+    "cdc",
+    "tags",
+    r#"{"cdc":{"path":"file.parquet","partitionValues":{},"size":100,"dataChange":false,"tags":{"key1":"val1","key2":null}}}"#
+)]
+// sidecar.tags.key2: null
+#[case::sidecar_tags(
+    "sidecar",
+    "tags",
+    r#"{"sidecar":{"path":"sidecar.parquet","sizeInBytes":100,"modificationTime":1000,"tags":{"key1":"val1","key2":null}}}"#
+)]
+// checkpointMetadata.tags.key2: null
+#[case::checkpoint_metadata_tags(
+    "checkpointMetadata",
+    "tags",
+    r#"{"checkpointMetadata":{"version":0,"tags":{"key1":"val1","key2":null}}}"#
+)]
+// Known issues: these map fields don't yet have #[allow_null_container_values].
+// commitInfo.operationParameters.description: null
+#[should_panic(expected = "StructArray re-validation failed")]
+#[case::commit_info_operation_parameters_known_issue(
+    "commitInfo",
+    "operationParameters",
+    r#"{"commitInfo":{"timestamp":1000,"operation":"WRITE","operationParameters":{"mode":"ErrorIfExists","description":null}}}"#
+)]
+// metaData.configuration.key2: null
+#[should_panic(expected = "StructArray re-validation failed")]
+#[case::metadata_configuration_known_issue(
+    "metaData",
+    "configuration",
+    r#"{"metaData":{"id":"test","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[]}","partitionColumns":[],"configuration":{"key1":"val1","key2":null},"createdTime":1000}}"#
+)]
+#[tokio::test]
+async fn read_actions_with_null_map_values(
+    #[case] action_name: &str,
+    #[case] map_field: &str,
+    #[case] json_action: &str,
+) {
+    use crate::arrow::array::{Array, AsArray, MapArray, StructArray};
+
+    let store = Arc::new(InMemory::new());
+    let log_root = Url::parse("memory:///_delta_log/").unwrap();
+
+    // Write a single commit file with the action containing null map values.
+    store
+        .put(
+            &delta_path_for_version(0, "json"),
+            json_action.to_string().into(),
+        )
+        .await
+        .unwrap();
+
+    // Build engine and read actions -- same as DeltaActionExtractor::get_actions.
+    let engine = DefaultEngineBuilder::new(store).build();
+    let log_segment =
+        LogSegment::for_table_changes(engine.storage_handler().as_ref(), log_root, 0, Some(0))
+            .unwrap();
+
+    // Use all_actions_schema to cover sidecar and checkpointMetadata (checkpoint-only actions).
+    let action_schema = get_all_actions_schema().clone();
+    let action_batches = log_segment
+        .read_actions(&engine, action_schema)
+        .expect("read_actions should succeed");
+
+    // Iterate batches and verify the map value field is nullable.
+    let mut found = false;
+    for batch_result in action_batches {
+        let actions_batch = batch_result.expect("Iterating action batches should succeed");
+
+        let data_any = actions_batch.actions.into_any();
+        let arrow_data = data_any
+            .downcast_ref::<ArrowEngineData>()
+            .expect("ArrowEngineData");
+        let rb = arrow_data.record_batch();
+
+        let Some(action_col) = rb.column_by_name(action_name) else {
+            continue;
+        };
+        let action_struct = action_col
+            .as_struct_opt()
+            .unwrap_or_else(|| panic!("{action_name} column should be a struct"));
+        let map_col = action_struct
+            .column_by_name(map_field)
+            .unwrap_or_else(|| panic!("{action_name}.{map_field} not found"));
+        let map_array = map_col
+            .as_any()
+            .downcast_ref::<MapArray>()
+            .unwrap_or_else(|| panic!("{action_name}.{map_field} should be a MapArray"));
+        // Re-validate the entries StructArray with its own schema, same as what Arrow's
+        // IPC deserializer does. Without the fix, this fails with:
+        // "Found unmasked nulls for non-nullable StructArray field 'value'"
+        let entries = map_array.entries();
+        StructArray::try_new(
+            entries.fields().clone(),
+            entries.columns().to_vec(),
+            entries.nulls().cloned(),
+        )
+        .unwrap_or_else(|e| {
+            panic!(
+                "{action_name}.{map_field} entries StructArray re-validation failed: {e}. \
+                 This means the schema has non-nullable value field but the data has nulls."
+            )
+        });
+        found = true;
+    }
+    assert!(found, "Should have found a {action_name} action batch");
+}

kernel/src/schema/derive_macro_utils.rs

@@ -123,3 +123,10 @@ impl<T: ToNullableContainerType> GetNullableContainerStructField for T {
         StructField::not_null(name, T::to_nullable_container_type())
     }
 }
+
+// Optional container types produce nullable fields with nullable values.
+impl<T: ToNullableContainerType> GetNullableContainerStructField for Option<T> {
+    fn get_nullable_container_struct_field(name: impl Into<String>) -> StructField {
+        StructField::nullable(name, T::to_nullable_container_type())
+    }
+}