fix: read Hive's null partition placeholder as NULL

datafusion/catalog-listing/src/helpers.rs

@@ -16,7 +16,7 @@ use datafusion_datasource::{FileExtensions, ListingTableUrl};
 use datafusion_expr::{BinaryExpr, Operator, lit, utils};
 
 use arrow::{
-    array::AsArray,
+    array::{Array, AsArray},
     datatypes::{DataType, Field, Metadata},
     record_batch::RecordBatch,
 };
@@ -324,23 +324,30 @@ pub fn filter_partitioned_file(
     )?;
 
     // Since we're only operating on a single file, our batch and resulting "array" holds only one
-    // value indicating if the input file matches the provided filters
+    // value indicating if the input file matches the provided filters. A NULL is not a match:
+    // a null partition value makes that reachable, as `part = 1` is NULL for a null partition.
     let matches = expr.evaluate(&batch)?.into_array(1)?;
-    if matches.as_boolean().value(0) {
+    let matches = matches.as_boolean();
+    if matches.is_valid(0) && matches.value(0) {
         return Ok(Some(pf));
     }
 
     Ok(None)
 }
 
 /// Returns `Ok(None)` when the file is not inside a valid partition path
-/// (e.g. a stale file in the table root directory). Such files are skipped
-/// because hive-style partition values are never null and there is no valid
-/// value to assign for non-partitioned files.
+/// (e.g. a stale file in the table root directory), as there is no value to
+/// assign for non-partitioned files.
+///
+/// A partition directory named `null_partition_name` is read as NULL; see
+/// [`ExecutionOptions::listing_table_hive_default_partition_name`].
+///
+/// [`ExecutionOptions::listing_table_hive_default_partition_name`]: datafusion_common::config::ExecutionOptions::listing_table_hive_default_partition_name
 fn try_into_partitioned_file(
     object_meta: ObjectMeta,
     partition_cols: &[(String, DataType)],
     table_path: &ListingTableUrl,
+    null_partition_name: &str,
 ) -> Result<Option<PartitionedFile>> {
     let cols = partition_cols.iter().map(|(name, _)| name.as_str());
     let parsed = parse_partitions_for_path(table_path, &object_meta.location, cols);
@@ -354,6 +361,9 @@ fn try_into_partitioned_file(
         .into_iter()
         .zip(partition_cols)
         .map(|(parsed, (_, datatype))| {
+            if !null_partition_name.is_empty() && parsed == null_partition_name {
+                return ScalarValue::try_new_null(datatype);
+            }
             ScalarValue::try_from_string(parsed.into_owned(), datatype)
         })
         .collect::<Result<Vec<_>>>()?;
@@ -413,12 +423,19 @@ pub async fn pruned_partition_list<'a>(
             Metadata::new(),
         )?;
 
+        let null_partition_name = ctx
+            .config_options()
+            .execution
+            .listing_table_hive_default_partition_name
+            .as_str();
+
         Ok(objects
-            .try_filter_map(|object_meta| {
+            .try_filter_map(move |object_meta| {
                 futures::future::ready(try_into_partitioned_file(
                     object_meta,
                     partition_cols,
                     table_path,
+                    null_partition_name,
                 ))
             })
             .try_filter_map(move |pf| {
@@ -495,6 +512,7 @@ pub fn describe_partition(partition: &Partition) -> (&str, usize, Vec<&str>) {
 
 #[cfg(test)]
 mod tests {
+    use datafusion_common::config::HIVE_DEFAULT_PARTITION_NAME;
     use datafusion_datasource::file_groups::FileGroup;
     use std::ops::Not;
 
@@ -659,8 +677,13 @@ mod tests {
             version: None,
         };
 
-        let result =
-            try_into_partitioned_file(meta, &partition_cols, &table_path).unwrap();
+        let result = try_into_partitioned_file(
+            meta,
+            &partition_cols,
+            &table_path,
+            HIVE_DEFAULT_PARTITION_NAME,
+        )
+        .unwrap();
         assert!(result.is_some());
         let pf = result.unwrap();
         assert_eq!(pf.partition_values.len(), 1);
@@ -685,8 +708,13 @@ mod tests {
             version: None,
         };
 
-        let result =
-            try_into_partitioned_file(meta, &partition_cols, &table_path).unwrap();
+        let result = try_into_partitioned_file(
+            meta,
+            &partition_cols,
+            &table_path,
+            HIVE_DEFAULT_PARTITION_NAME,
+        )
+        .unwrap();
         assert!(result.is_some());
         let pf = result.unwrap();
         assert_eq!(pf.partition_values.len(), 1);
@@ -696,6 +724,60 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_try_into_partitioned_file_null_partition() {
+        let table_path = ListingTableUrl::parse("file:///bucket/mytable").unwrap();
+        // A type the placeholder cannot be parsed as, which is how this used to fail.
+        let partition_cols = vec![("day".to_string(), DataType::Date32)];
+        let meta = ObjectMeta {
+            location: Path::from(
+                "bucket/mytable/day=__HIVE_DEFAULT_PARTITION__/data.parquet",
+            ),
+            last_modified: chrono::Utc::now(),
+            size: 100,
+            e_tag: None,
+            version: None,
+        };
+
+        let pf = try_into_partitioned_file(
+            meta,
+            &partition_cols,
+            &table_path,
+            HIVE_DEFAULT_PARTITION_NAME,
+        )
+        .unwrap()
+        .expect("file should be listed");
+
+        assert_eq!(pf.partition_values, vec![ScalarValue::Date32(None)]);
+    }
+
+    #[test]
+    fn test_try_into_partitioned_file_null_partition_name_not_matched() {
+        let table_path = ListingTableUrl::parse("file:///bucket/mytable").unwrap();
+        let partition_cols = vec![("category".to_string(), DataType::Utf8)];
+        let meta = ObjectMeta {
+            location: Path::from(
+                "bucket/mytable/category=__HIVE_DEFAULT_PARTITION__/data.parquet",
+            ),
+            last_modified: chrono::Utc::now(),
+            size: 100,
+            e_tag: None,
+            version: None,
+        };
+
+        // The empty name turns the mapping off, leaving the directory name as the value.
+        let pf = try_into_partitioned_file(meta, &partition_cols, &table_path, "")
+            .unwrap()
+            .expect("file should be listed");
+
+        assert_eq!(
+            pf.partition_values,
+            vec![ScalarValue::Utf8(Some(
+                "__HIVE_DEFAULT_PARTITION__".to_string()
+            ))]
+        );
+    }
+
     #[test]
     fn test_try_into_partitioned_file_root_file_skipped() {
         // File in root directory (not inside any partition path) should be
@@ -711,8 +793,13 @@ mod tests {
             version: None,
         };
 
-        let result =
-            try_into_partitioned_file(meta, &partition_cols, &table_path).unwrap();
+        let result = try_into_partitioned_file(
+            meta,
+            &partition_cols,
+            &table_path,
+            HIVE_DEFAULT_PARTITION_NAME,
+        )
+        .unwrap();
         assert!(
             result.is_none(),
             "Files outside partition structure should be skipped"
@@ -732,8 +819,13 @@ mod tests {
             version: None,
         };
 
-        let result =
-            try_into_partitioned_file(meta, &partition_cols, &table_path).unwrap();
+        let result = try_into_partitioned_file(
+            meta,
+            &partition_cols,
+            &table_path,
+            HIVE_DEFAULT_PARTITION_NAME,
+        )
+        .unwrap();
         assert!(
             result.is_none(),
             "Files with wrong partition column name should be skipped"
@@ -755,8 +847,13 @@ mod tests {
             version: None,
         };
 
-        let result =
-            try_into_partitioned_file(meta, &partition_cols, &table_path).unwrap();
+        let result = try_into_partitioned_file(
+            meta,
+            &partition_cols,
+            &table_path,
+            HIVE_DEFAULT_PARTITION_NAME,
+        )
+        .unwrap();
         assert!(result.is_some());
         let pf = result.unwrap();
         assert_eq!(pf.partition_values.len(), 2);
@@ -786,8 +883,13 @@ mod tests {
             version: None,
         };
 
-        let result =
-            try_into_partitioned_file(meta, &partition_cols, &table_path).unwrap();
+        let result = try_into_partitioned_file(
+            meta,
+            &partition_cols,
+            &table_path,
+            HIVE_DEFAULT_PARTITION_NAME,
+        )
+        .unwrap();
         // File has year=2024 but no month= directory — parse_partitions_for_path
         // returns None because the path component "data.parquet" doesn't match
         // the expected "month=..." pattern.

datafusion/catalog-listing/src/table.rs

@@ -241,10 +241,12 @@ impl ListingTable {
             file_schema
         };
 
-        // Add the partition columns to the file schema
+        // Add the partition columns to the file schema. They are nullable because a Hive
+        // compliant writer spells a null partition value with a placeholder directory name
+        // (see `ExecutionOptions::listing_table_hive_default_partition_name`).
         let mut builder = SchemaBuilder::from(file_schema.as_ref().to_owned());
         for (part_col_name, part_col_type) in &options.table_partition_cols {
-            builder.push(Field::new(part_col_name, part_col_type.clone(), false));
+            builder.push(Field::new(part_col_name, part_col_type.clone(), true));
         }
 
         let table_schema = Arc::new(
@@ -360,7 +362,8 @@ impl ListingTable {
                 self.options
                     .table_partition_cols
                     .iter()
-                    .map(|(col, field)| Arc::new(Field::new(col, field.clone(), false)))
+                    // Nullable to match the table schema built in `try_new`.
+                    .map(|(col, field)| Arc::new(Field::new(col, field.clone(), true)))
                     .collect::<Vec<_>>(),
             )
             .build();
@@ -1120,7 +1123,7 @@ impl ListingTable {
 
         // Note: Statistics already include both file columns and partition columns.
         // PartitionedFile::with_statistics automatically appends exact partition column
-        // statistics (min=max=partition_value, null_count=0, distinct_count=1) computed
+        // statistics (e.g. min=max=partition_value, null_count=0, distinct_count=1) computed
         // from partition_values.
         Ok(ListFilesResult {
             file_groups,

datafusion/common/src/config.rs

@@ -20,6 +20,7 @@ use crate::{DataFusionError, Result};
 #[cfg(feature = "parquet_encryption")]
 use hex;
 use std::any::Any;
+use std::borrow::Cow;
 use std::collections::{BTreeMap, HashMap};
 use std::error::Error;
 use std::fmt::{self, Display};
@@ -886,6 +887,10 @@ impl From<SpillCompression> for Option<CompressionType> {
     }
 }
 
+/// The directory name Hive compliant writers use in place of a null partition value, and the
+/// default for [`ExecutionOptions::listing_table_hive_default_partition_name`].
+pub const HIVE_DEFAULT_PARTITION_NAME: &str = "__HIVE_DEFAULT_PARTITION__";
+
 config_namespace! {
     /// Options related to query execution
     ///
@@ -1095,6 +1100,14 @@ config_namespace! {
         /// inferred and will be represented in the table schema).
         pub listing_table_factory_infer_partitions: bool, default = true
 
+        /// The directory name Hive compliant writers use in place of a null partition
+        /// value, e.g. `/table/date=__HIVE_DEFAULT_PARTITION__/data.parquet`. Partition
+        /// values equal to this name are read as NULL, as Hive, Spark and Impala do, and
+        /// partitioned writes reject them. Corresponds to Hive's
+        /// `hive.exec.default.partition.name`. Set this to the empty string to read and
+        /// write such values as the literal directory name instead.
+        pub listing_table_hive_default_partition_name: String, default = HIVE_DEFAULT_PARTITION_NAME.to_string()
+
         /// Should DataFusion support recursive CTEs
         pub enable_recursive_ctes: bool, default = true
 
@@ -2516,14 +2529,26 @@ impl ConfigOptions {
                 &mut docs,
                 "| {} | {} | {} |",
                 entry.key,
-                entry.value.as_deref().unwrap_or("NULL"),
+                display_default(entry.value.as_deref()),
                 entry.description
             );
         }
         docs
     }
 }
 
+/// Renders a default value for the generated documentation table.
+///
+/// Values containing `_` or `*` are shown as code, since markdown would otherwise read them
+/// as formatting: a default of `__HIVE_DEFAULT_PARTITION__` renders as bold text.
+fn display_default(value: Option<&str>) -> Cow<'_, str> {
+    match value {
+        None => Cow::Borrowed("NULL"),
+        Some(value) if value.contains(['_', '*']) => Cow::Owned(format!("`{value}`")),
+        Some(value) => Cow::Borrowed(value),
+    }
+}
+
 /// [`ConfigExtension`] provides a mechanism to store third-party configuration
 /// within DataFusion [`ConfigOptions`]
 ///