diff --git a/datafusion/catalog-listing/src/helpers.rs b/datafusion/catalog-listing/src/helpers.rs index df81d36d93104..1907483c256fb 100644 --- a/datafusion/catalog-listing/src/helpers.rs +++ b/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,9 +324,11 @@ 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)); } @@ -334,13 +336,18 @@ pub fn filter_partitioned_file( } /// 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> { 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::>>()?; @@ -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. diff --git a/datafusion/catalog-listing/src/table.rs b/datafusion/catalog-listing/src/table.rs index 9555ec873d81c..a1c3f16d4e66c 100644 --- a/datafusion/catalog-listing/src/table.rs +++ b/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::>(), ) .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, diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs index 3253ce404d322..825536e165ddd 100644 --- a/datafusion/common/src/config.rs +++ b/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 for Option { } } +/// 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,7 +2529,7 @@ impl ConfigOptions { &mut docs, "| {} | {} | {} |", entry.key, - entry.value.as_deref().unwrap_or("NULL"), + display_default(entry.value.as_deref()), entry.description ); } @@ -2524,6 +2537,18 @@ impl ConfigOptions { } } +/// 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`] /// diff --git a/datafusion/core/tests/catalog_listing/pruned_partition_list.rs b/datafusion/core/tests/catalog_listing/pruned_partition_list.rs index 098acc35d7119..06803854d0526 100644 --- a/datafusion/core/tests/catalog_listing/pruned_partition_list.rs +++ b/datafusion/core/tests/catalog_listing/pruned_partition_list.rs @@ -9,6 +9,7 @@ use futures::{FutureExt, StreamExt as _, TryStreamExt as _}; use object_store::{ObjectStoreExt, memory::InMemory, path::Path}; use datafusion::execution::SessionStateBuilder; +use datafusion::prelude::SessionConfig; use datafusion_catalog_listing::helpers::{ describe_partition, list_partitions, pruned_partition_list, }; @@ -220,8 +221,146 @@ async fn test_list_partition() { ); } +/// A partition directory named `__HIVE_DEFAULT_PARTITION__` holds the rows whose partition +/// value is null, so it is read as NULL rather than parsed as the column's type. +#[tokio::test] +async fn test_pruned_partition_list_null_partition() { + let (store, state) = make_test_store_and_state(&[ + ("tablepath/mypartition=3/file.parquet", 100), + ( + "tablepath/mypartition=__HIVE_DEFAULT_PARTITION__/file.parquet", + 100, + ), + ]); + + let pruned = pruned_partition_list( + state.as_ref(), + store.as_ref(), + &ListingTableUrl::parse("file:///tablepath/").unwrap(), + &[], + ".parquet", + &[(String::from("mypartition"), DataType::Int32)], + ) + .await + .expect("partition pruning failed") + .try_collect::>() + .await + .unwrap(); + + assert_eq!(pruned.len(), 2); + assert_eq!(&pruned[0].partition_values, &[ScalarValue::Int32(Some(3))]); + assert_eq!(&pruned[1].partition_values, &[ScalarValue::Int32(None)]); +} + +#[tokio::test] +async fn test_pruned_partition_list_prunes_null_partition() { + let files: &[(&str, u64)] = &[ + ("tablepath/mypartition=3/file.parquet", 100), + ( + "tablepath/mypartition=__HIVE_DEFAULT_PARTITION__/file.parquet", + 100, + ), + ]; + let partition_cols = &[(String::from("mypartition"), DataType::Int32)]; + + let pruned_with = |filter: Expr| async move { + let (store, state) = make_test_store_and_state(files); + pruned_partition_list( + state.as_ref(), + store.as_ref(), + &ListingTableUrl::parse("file:///tablepath/").unwrap(), + &[filter], + ".parquet", + partition_cols, + ) + .await + .expect("partition pruning failed") + .try_collect::>() + .await + .unwrap() + }; + + // The null partition does not match an equality filter: `mypartition = 3` is NULL there. + let pruned = pruned_with(Expr::eq(col("mypartition"), lit(3i32))).await; + assert_eq!(pruned.len(), 1); + assert_eq!( + pruned[0].object_meta.location.as_ref(), + "tablepath/mypartition=3/file.parquet" + ); + + // ... and only the null partition matches `IS NULL`. + let pruned = pruned_with(col("mypartition").is_null()).await; + assert_eq!(pruned.len(), 1); + assert_eq!( + pruned[0].object_meta.location.as_ref(), + "tablepath/mypartition=__HIVE_DEFAULT_PARTITION__/file.parquet" + ); +} + +#[tokio::test] +async fn test_pruned_partition_list_null_partition_name_is_configurable() { + let files: &[(&str, u64)] = &[ + ("tablepath/mypartition=MY_NULL/file.parquet", 100), + ( + "tablepath/mypartition=__HIVE_DEFAULT_PARTITION__/file.parquet", + 100, + ), + ]; + + let pruned_with_name = |name: &str| { + let config = SessionConfig::new().set_str( + "datafusion.execution.listing_table_hive_default_partition_name", + name, + ); + let (store, state) = make_test_store_and_state_with_config(files, config); + async move { + pruned_partition_list( + state.as_ref(), + store.as_ref(), + &ListingTableUrl::parse("file:///tablepath/").unwrap(), + &[], + ".parquet", + &[(String::from("mypartition"), DataType::Utf8)], + ) + .await + .expect("partition pruning failed") + .try_collect::>() + .await + .unwrap() + } + }; + + // A writer configured with a different placeholder, e.g. Hive's + // `hive.exec.default.partition.name`: only that name is read as NULL. + let pruned = pruned_with_name("MY_NULL").await; + assert_eq!( + &pruned[0].partition_values, + &[ScalarValue::Utf8(None)], + "MY_NULL should be read as NULL" + ); + assert_eq!( + &pruned[1].partition_values, + &[ScalarValue::from("__HIVE_DEFAULT_PARTITION__")] + ); + + // The empty name turns the mapping off, leaving the directory name as the value. + let pruned = pruned_with_name("").await; + assert_eq!(&pruned[0].partition_values, &[ScalarValue::from("MY_NULL")]); + assert_eq!( + &pruned[1].partition_values, + &[ScalarValue::from("__HIVE_DEFAULT_PARTITION__")] + ); +} + pub fn make_test_store_and_state( files: &[(&str, u64)], +) -> (Arc, Arc) { + make_test_store_and_state_with_config(files, SessionConfig::new()) +} + +pub fn make_test_store_and_state_with_config( + files: &[(&str, u64)], + config: SessionConfig, ) -> (Arc, Arc) { let memory = InMemory::new(); @@ -233,6 +372,6 @@ pub fn make_test_store_and_state( .unwrap(); } - let state = SessionStateBuilder::new().build(); + let state = SessionStateBuilder::new().with_config(config).build(); (Arc::new(memory), Arc::new(state)) } diff --git a/datafusion/datasource/src/file_groups.rs b/datafusion/datasource/src/file_groups.rs index 7fc8c29b22602..561aaf0872582 100644 --- a/datafusion/datasource/src/file_groups.rs +++ b/datafusion/datasource/src/file_groups.rs @@ -360,7 +360,7 @@ impl FileGroupPartitioner { /// Partition column statistics are derived from the individual file partition values: /// - `min` = minimum partition value across all files in the group /// - `max` = maximum partition value across all files in the group -/// - `null_count` = 0 (partition values are never null) +/// - `null_count` = the number of rows in files with a null partition value /// /// This allows query optimizers to prune entire file groups based on partition bounds. #[derive(Debug, Clone)] diff --git a/datafusion/datasource/src/mod.rs b/datafusion/datasource/src/mod.rs index bc7c181bb70a8..1847aab6fa2c6 100644 --- a/datafusion/datasource/src/mod.rs +++ b/datafusion/datasource/src/mod.rs @@ -107,9 +107,12 @@ impl FileRange { /// computed from [`Self::partition_values`]: /// /// - `min = max = partition_value` (all rows in a file share the same partition value) -/// - `null_count = 0` (partition values extracted from paths are never null) +/// - `null_count = 0` (every row has the file's partition value) /// - `distinct_count = 1` (single distinct value per file for each partition column) /// +/// A null partition value instead gives `null_count = num_rows`, `distinct_count = 0` +/// and absent `min`/`max`. +/// /// This enables query optimizers to use partition column bounds for pruning and planning. pub struct PartitionedFile { /// Path for the file (e.g. URL, filesystem path, etc) @@ -328,8 +331,11 @@ impl PartitionedFile { /// This method will automatically append exact statistics for partition columns /// based on `partition_values`: /// - `min = max = partition_value` (all rows have the same value) - /// - `null_count = 0` (partition values from paths are never null) + /// - `null_count = 0` (all rows have a non-null value) /// - `distinct_count = 1` (all rows have the same partition value) + /// + /// A null partition value (e.g. a Hive default partition) instead gives + /// `null_count = num_rows`, `distinct_count = 0` and absent `min`/`max`. pub fn with_statistics(mut self, file_statistics: Arc) -> Self { if self.partition_values.is_empty() { // No partition columns, use stats as-is @@ -338,11 +344,20 @@ impl PartitionedFile { // Extend stats with exact partition column statistics let mut stats = Arc::unwrap_or_clone(file_statistics); for partition_value in &self.partition_values { + let (null_count, min_max, distinct_count) = if partition_value.is_null() { + (stats.num_rows, Precision::Absent, Precision::Exact(0)) + } else { + ( + Precision::Exact(0), + Precision::Exact(partition_value.clone()), + Precision::Exact(1), + ) + }; let col_stats = ColumnStatistics { - null_count: Precision::Exact(0), - max_value: Precision::Exact(partition_value.clone()), - min_value: Precision::Exact(partition_value.clone()), - distinct_count: Precision::Exact(1), + null_count, + max_value: min_max.clone(), + min_value: min_max, + distinct_count, sum_value: Precision::Absent, byte_size: partition_value .data_type() @@ -684,6 +699,29 @@ mod tests { ); } + #[test] + fn test_with_statistics_null_partition_value() { + use crate::PartitionedFile; + use datafusion_common::stats::Precision; + use datafusion_common::{ScalarValue, Statistics}; + + let mut pf = PartitionedFile::new("test.parquet", 100); + pf.partition_values = vec![ScalarValue::Date32(None)]; + + let file_stats = Arc::new(Statistics { + num_rows: Precision::Exact(2), + total_byte_size: Precision::Exact(16), + column_statistics: vec![], + }); + + let stats = pf.with_statistics(file_stats).statistics.unwrap(); + let partition_col_stats = &stats.column_statistics[0]; + assert_eq!(partition_col_stats.null_count, Precision::Exact(2)); + assert_eq!(partition_col_stats.min_value, Precision::Absent); + assert_eq!(partition_col_stats.max_value, Precision::Absent); + assert_eq!(partition_col_stats.distinct_count, Precision::Exact(0)); + } + #[test] fn test_url_contains() { let url = ListingTableUrl::parse("file:///var/data/mytable/").unwrap(); diff --git a/datafusion/datasource/src/statistics.rs b/datafusion/datasource/src/statistics.rs index 1987f36714427..903c9fafafddc 100644 --- a/datafusion/datasource/src/statistics.rs +++ b/datafusion/datasource/src/statistics.rs @@ -405,7 +405,6 @@ mod tests { use crate::PartitionedFile; use crate::file_groups::FileGroup; use arrow::datatypes::{DataType, Field, Schema}; - use datafusion_common::stats::Precision; use datafusion_physical_plan::ColumnStatistics; fn test_schema() -> SchemaRef { @@ -522,4 +521,56 @@ mod tests { "unexpected error: {err:?}" ); } + + #[test] + fn min_max_statistics_null_partition_value() -> Result<()> { + let schema = Arc::new(Schema::new(vec![ + Field::new("a", DataType::Int64, true), + Field::new("d", DataType::Date32, true), + ])); + let sort_order = + [PhysicalSortExpr::new_default(Arc::new(Column::new("d", 1)))].into(); + let file = |path: &str, d: Option| { + let mut pf = PartitionedFile::new(path, 1); + pf.partition_values = vec![ScalarValue::Date32(d)]; + pf.with_statistics(Arc::new( + Statistics::new_unknown(&test_schema()) + .with_num_rows(Precision::Exact(1)), + )) + }; + let files = [file("d=1/f.parquet", Some(1)), file("null/f.parquet", None)]; + + let stats = + MinMaxStatistics::new_from_files(&sort_order, &schema, None, files.iter())?; + + let sorted = stats.min_values_sorted(); + assert_eq!(sorted.iter().map(|(i, _)| *i).collect::>(), [1, 0]); + assert!(!stats.is_sorted()); + + Ok(()) + } + + #[test] + fn min_max_statistics_null_partition_value_unknown_num_rows() { + let schema = Arc::new(Schema::new(vec![ + Field::new("a", DataType::Int64, true), + Field::new("d", DataType::Date32, true), + ])); + let sort_order = + [PhysicalSortExpr::new_default(Arc::new(Column::new("d", 1)))].into(); + let mut pf = PartitionedFile::new("null/f.parquet", 1); + pf.partition_values = vec![ScalarValue::Date32(None)]; + let files = + [pf.with_statistics(Arc::new(Statistics::new_unknown(&test_schema())))]; + + let Err(err) = + MinMaxStatistics::new_from_files(&sort_order, &schema, None, files.iter()) + else { + panic!("expected missing statistics error") + }; + assert!( + err.to_string().contains("statistics not found"), + "unexpected error: {err:?}" + ); + } } diff --git a/datafusion/datasource/src/write/demux.rs b/datafusion/datasource/src/write/demux.rs index 829cc85b216bb..17fde3eb9e29b 100644 --- a/datafusion/datasource/src/write/demux.rs +++ b/datafusion/datasource/src/write/demux.rs @@ -326,7 +326,11 @@ async fn hive_style_partitions_demuxer( while let Some(rb) = input.next().await.transpose()? { // First compute partition key for each row of batch, e.g. (col1=val1, col2=val2, ...) - let all_partition_values = compute_partition_keys_by_row(&rb, &partition_by)?; + let all_partition_values = compute_partition_keys_by_row( + &rb, + &partition_by, + &exec_options.listing_table_hive_default_partition_name, + )?; // Next compute how the batch should be split up to take each distinct key to its own batch let take_map = compute_take_arrays(&rb, &all_partition_values); @@ -394,6 +398,7 @@ pub struct FileMetadata { fn compute_partition_keys_by_row<'a>( rb: &'a RecordBatch, partition_by: &'a [(String, DataType)], + null_partition_name: &str, ) -> Result>>> { let mut all_partition_values = Vec::with_capacity(partition_by.len()); @@ -560,6 +565,15 @@ fn compute_partition_keys_by_row<'a>( } } + if !null_partition_name.is_empty() + && partition_values.iter().any(|v| v == null_partition_name) + { + return exec_err!( + "Cannot write value '{null_partition_name}' for partition column '{col}': \ + it is the null partition name and would be read back as NULL" + ); + } + all_partition_values.push(partition_values); } @@ -631,6 +645,7 @@ mod tests { use super::*; use arrow::array::{Array, ArrayRef, DictionaryArray, Int32Array}; use arrow::datatypes::Int32Type; + use datafusion_common::config::HIVE_DEFAULT_PARTITION_NAME; fn partition_batch(column: ArrayRef) -> RecordBatch { RecordBatch::try_from_iter_with_nullable([("p", column, true)]).unwrap() @@ -645,7 +660,12 @@ mod tests { for column in columns { let partition_by = vec![("p".to_string(), column.data_type().clone())]; let batch = partition_batch(column); - let err = compute_partition_keys_by_row(&batch, &partition_by).unwrap_err(); + let err = compute_partition_keys_by_row( + &batch, + &partition_by, + HIVE_DEFAULT_PARTITION_NAME, + ) + .unwrap_err(); assert_eq!( err.strip_backtrace(), "Execution error: NULL values are not supported for partition column 'p'" @@ -658,7 +678,12 @@ mod tests { let column = Int32Array::from(vec![None, Some(0), Some(1)]).slice(1, 2); let batch = partition_batch(Arc::new(column)); let partition_by = vec![("p".to_string(), DataType::Int32)]; - let keys = compute_partition_keys_by_row(&batch, &partition_by).unwrap(); + let keys = compute_partition_keys_by_row( + &batch, + &partition_by, + HIVE_DEFAULT_PARTITION_NAME, + ) + .unwrap(); assert_eq!(keys, vec![vec![Cow::Borrowed("0"), Cow::Borrowed("1")]]); } @@ -675,7 +700,12 @@ mod tests { let column = dictionary_with_null_value(); let partition_by = vec![("p".to_string(), column.data_type().clone())]; let batch = partition_batch(Arc::new(column)); - let err = compute_partition_keys_by_row(&batch, &partition_by).unwrap_err(); + let err = compute_partition_keys_by_row( + &batch, + &partition_by, + HIVE_DEFAULT_PARTITION_NAME, + ) + .unwrap_err(); assert_eq!( err.strip_backtrace(), "Execution error: NULL values are not supported for partition column 'p'" @@ -691,7 +721,12 @@ mod tests { .unwrap(); let partition_by = vec![("p".to_string(), column.data_type().clone())]; let batch = partition_batch(Arc::new(column)); - let err = compute_partition_keys_by_row(&batch, &partition_by).unwrap_err(); + let err = compute_partition_keys_by_row( + &batch, + &partition_by, + HIVE_DEFAULT_PARTITION_NAME, + ) + .unwrap_err(); assert_eq!( err.strip_backtrace(), "Execution error: NULL values are not supported for partition column 'p'" @@ -703,7 +738,46 @@ mod tests { let column = dictionary_with_null_value().slice(0, 1); let partition_by = vec![("p".to_string(), column.data_type().clone())]; let batch = partition_batch(Arc::new(column)); - let keys = compute_partition_keys_by_row(&batch, &partition_by).unwrap(); + let keys = compute_partition_keys_by_row( + &batch, + &partition_by, + HIVE_DEFAULT_PARTITION_NAME, + ) + .unwrap(); assert_eq!(keys, vec![vec![Cow::Borrowed("a")]]); } + + #[test] + fn partition_keys_reject_null_partition_name() { + let columns: Vec = vec![ + Arc::new(StringArray::from(vec!["a", HIVE_DEFAULT_PARTITION_NAME])), + Arc::new(DictionaryArray::::new( + Int32Array::from(vec![0, 1]), + Arc::new(StringArray::from(vec!["a", HIVE_DEFAULT_PARTITION_NAME])), + )), + ]; + for column in columns { + let partition_by = vec![("p".to_string(), column.data_type().clone())]; + let batch = partition_batch(column); + let err = compute_partition_keys_by_row( + &batch, + &partition_by, + HIVE_DEFAULT_PARTITION_NAME, + ) + .unwrap_err(); + assert_eq!( + err.strip_backtrace(), + "Execution error: Cannot write value '__HIVE_DEFAULT_PARTITION__' for partition column 'p': it is the null partition name and would be read back as NULL" + ); + } + } + + #[test] + fn partition_keys_accept_null_partition_name_when_disabled() { + let column = StringArray::from(vec![HIVE_DEFAULT_PARTITION_NAME]); + let batch = partition_batch(Arc::new(column)); + let partition_by = vec![("p".to_string(), DataType::Utf8)]; + let keys = compute_partition_keys_by_row(&batch, &partition_by, "").unwrap(); + assert_eq!(keys, vec![vec![Cow::Borrowed(HIVE_DEFAULT_PARTITION_NAME)]]); + } } diff --git a/datafusion/sqllogictest/test_files/information_schema.slt b/datafusion/sqllogictest/test_files/information_schema.slt index e7b0e08d5975c..443f1f585f2e8 100644 --- a/datafusion/sqllogictest/test_files/information_schema.slt +++ b/datafusion/sqllogictest/test_files/information_schema.slt @@ -217,6 +217,7 @@ datafusion.execution.enforce_batch_size_in_joins false datafusion.execution.hash_join_buffering_capacity 0 datafusion.execution.keep_partition_by_columns false datafusion.execution.listing_table_factory_infer_partitions true +datafusion.execution.listing_table_hive_default_partition_name __HIVE_DEFAULT_PARTITION__ datafusion.execution.listing_table_ignore_subdirectory true datafusion.execution.max_buffered_batches_per_output_file 2 datafusion.execution.max_spill_file_size_bytes 134217728 @@ -383,6 +384,7 @@ datafusion.execution.enforce_batch_size_in_joins false Should DataFusion enforce datafusion.execution.hash_join_buffering_capacity 0 How many bytes to buffer in the probe side of hash joins while the build side is concurrently being built. Without this, hash joins will wait until the full materialization of the build side before polling the probe side. This is useful in scenarios where the query is not completely CPU bounded, allowing to do some early work concurrently and reducing the latency of the query. Note that when hash join buffering is enabled, the probe side will start eagerly polling data, not giving time for the producer side of dynamic filters to produce any meaningful predicate. Queries with dynamic filters might see performance degradation. Disabled by default, set to a number greater than 0 for enabling it. datafusion.execution.keep_partition_by_columns false Should DataFusion keep the columns used for partition_by in the output RecordBatches datafusion.execution.listing_table_factory_infer_partitions true Should a `ListingTable` created through the `ListingTableFactory` infer table partitions from Hive compliant directories. Defaults to true (partition columns are inferred and will be represented in the table schema). +datafusion.execution.listing_table_hive_default_partition_name __HIVE_DEFAULT_PARTITION__ 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. datafusion.execution.listing_table_ignore_subdirectory true Should sub directories be ignored when scanning directories for data files. Defaults to true (ignores subdirectories), consistent with Hive. Note that this setting does not affect reading partitioned tables (e.g. `/table/year=2021/month=01/data.parquet`). datafusion.execution.max_buffered_batches_per_output_file 2 This is the maximum number of RecordBatches buffered for each output file being worked. Higher values can potentially give faster write performance at the cost of higher peak memory consumption. This budget is split evenly between two independent points in the write pipeline (see the demuxer diagram in #7791): how many files can be in flight from the demuxer to a writer task, and how many RecordBatches are buffered for a single file's writer. Must be at least 2 so each half gets at least 1 unit of buffering - 0 or 1 would leave one side with a zero-capacity channel and panic at write time. datafusion.execution.max_spill_file_size_bytes 134217728 Maximum size in bytes for individual spill files before rotating to a new file. When operators spill data to disk (e.g., RepartitionExec), they write multiple batches to the same file until this size limit is reached, then rotate to a new file. This reduces syscall overhead compared to one-file-per-batch while preventing files from growing too large. A larger value reduces file creation overhead but may hold more disk space. A smaller value creates more files but allows finer-grained space reclamation as files can be deleted once fully consumed. Now only `RepartitionExec` supports this spill file rotation feature, other spilling operators may create spill files larger than the limit. Default: 128 MB diff --git a/datafusion/sqllogictest/test_files/insert_to_external.slt b/datafusion/sqllogictest/test_files/insert_to_external.slt index a689dfc5bbdce..80dcf355eb9c7 100644 --- a/datafusion/sqllogictest/test_files/insert_to_external.slt +++ b/datafusion/sqllogictest/test_files/insert_to_external.slt @@ -172,8 +172,8 @@ query TTT describe partitioned_insert_test_readback; ---- c Int64 YES -a Dictionary(UInt16, Utf8) NO -b Dictionary(UInt16, Utf8) NO +a Dictionary(UInt16, Utf8) YES +b Dictionary(UInt16, Utf8) YES query ITT select * from partitioned_insert_test_readback order by a,b,c; diff --git a/datafusion/sqllogictest/test_files/listing_table_partitions.slt b/datafusion/sqllogictest/test_files/listing_table_partitions.slt index c3e903ac7fa44..d5954caae09c1 100644 --- a/datafusion/sqllogictest/test_files/listing_table_partitions.slt +++ b/datafusion/sqllogictest/test_files/listing_table_partitions.slt @@ -127,3 +127,118 @@ select year, count(*) from root_file_test group by year order by year; ---- 2024 2 2025 1 + +# Test: a null partition value, which Hive compliant writers spell with a +# `__HIVE_DEFAULT_PARTITION__` directory, is read as NULL. + +query I +copy (values(1, 'alice')) +to 'test_files/scratch/listing_table_partitions/null_part/d=2024-01-01/data.parquet'; +---- +1 + +query I +copy (values(2, 'bob')) +to 'test_files/scratch/listing_table_partitions/null_part/d=__HIVE_DEFAULT_PARTITION__/data.parquet'; +---- +1 + +statement count 0 +create external table null_part (column1 int, column2 varchar, d date) +stored as parquet +partitioned by (d) +location 'test_files/scratch/listing_table_partitions/null_part/'; + +query ITD +select column1, column2, d from null_part order by column1; +---- +1 alice 2024-01-01 +2 bob NULL + +query ITD +select column1, column2, d from null_part where d is null; +---- +2 bob NULL + +# The null partition does not match an equality filter. +query ITD +select column1, column2, d from null_part where d = '2024-01-01'; +---- +1 alice 2024-01-01 + +# ... nor an inequality filter. +query ITD +select column1, column2, d from null_part where d <> '2024-01-01'; +---- + +query ITD +select column1, column2, d from null_part where d not in ('2024-01-01'); +---- + +# Statistics for the null partition count its rows as nulls. +query I +select count(d) from null_part; +---- +1 + +query DD +select min(d), max(d) from null_part; +---- +2024-01-01 2024-01-01 + +# A null partition still lets files be ordered by their partition value. +statement ok +set datafusion.execution.target_partitions = 1; + +statement count 0 +create external table null_part_ordered (column1 int, column2 varchar, d date) +stored as parquet +partitioned by (d) +with order (d) +location 'test_files/scratch/listing_table_partitions/null_part/'; + +query TT +explain select column1, d from null_part_ordered order by d; +---- +logical_plan +01)Sort: null_part_ordered.d ASC NULLS LAST +02)--TableScan: null_part_ordered projection=[column1, d] +physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/listing_table_partitions/null_part/d=2024-01-01/data.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/listing_table_partitions/null_part/d=__HIVE_DEFAULT_PARTITION__/data.parquet]]}, projection=[column1, d], output_ordering=[d@1 ASC NULLS LAST], file_type=parquet + +query ID +select column1, d from null_part_ordered order by d; +---- +1 2024-01-01 +2 NULL + +statement ok +set datafusion.execution.target_partitions = 4; + +# Writing the null partition name as a real value is rejected, since it would +# be read back as NULL. +statement count 0 +create external table null_part_write (a int, p varchar) +stored as parquet +partitioned by (p) +location 'test_files/scratch/listing_table_partitions/null_part_write/'; + +statement error Cannot write value '__HIVE_DEFAULT_PARTITION__' for partition column 'p': it is the null partition name and would be read back as NULL +insert into null_part_write values (1, 'x'), (2, '__HIVE_DEFAULT_PARTITION__'); + +# With the null partition name disabled, the value round trips. +statement ok +set datafusion.execution.listing_table_hive_default_partition_name = ''; + +query I +insert into null_part_write values (1, 'x'), (2, '__HIVE_DEFAULT_PARTITION__'); +---- +2 + +query IT +select a, p from null_part_write order by a; +---- +1 x +2 __HIVE_DEFAULT_PARTITION__ + +statement ok +set datafusion.execution.listing_table_hive_default_partition_name = '__HIVE_DEFAULT_PARTITION__'; diff --git a/docs/source/library-user-guide/upgrading/56.0.0.md b/docs/source/library-user-guide/upgrading/56.0.0.md index 2b11283bd09a9..413933018a496 100644 --- a/docs/source/library-user-guide/upgrading/56.0.0.md +++ b/docs/source/library-user-guide/upgrading/56.0.0.md @@ -227,6 +227,67 @@ Rewriting affected files with a current DataFusion version, for example using restores optimizations that depend on them. Preserve the required data ordering and ordering metadata when rewriting sorted files. +### Hive null partition directories are read as NULL + +`ListingTable` now reads a partition directory named +`__HIVE_DEFAULT_PARTITION__`, for example `/table/d=__HIVE_DEFAULT_PARTITION__/`, +as a NULL partition value. This is the placeholder Hive compliant writers use for +a null partition value, and matches how Hive, Spark and Impala read it. +Previously the directory name was parsed as the partition column's type, which +failed for non-string columns and returned the literal string +`__HIVE_DEFAULT_PARTITION__` for string columns. + +The placeholder is set by the new +`datafusion.execution.listing_table_hive_default_partition_name` option, the +equivalent of Hive's `hive.exec.default.partition.name`. Setting it to the empty +string reads the directory name as the value, as before. + +Partitioned writes (`INSERT INTO` and `COPY ... PARTITIONED BY`) now reject a +partition value equal to this placeholder, since it would be read back as NULL. + +To support this, `ListingTable` now declares all partition columns as nullable, +so `DESCRIBE` and the table schema report them as nullable. + +**Who is affected:** + +- Tables with string partition columns that contain a + `__HIVE_DEFAULT_PARTITION__` directory. Its rows now have a NULL partition + value, so `WHERE col = '__HIVE_DEFAULT_PARTITION__'` no longer matches them. +- Partitioned writes of the value `__HIVE_DEFAULT_PARTITION__`, which now + return an error instead of writing a directory that reads back as NULL. +- Code that relies on `ListingTable` partition columns being non-nullable. +- Callers of `pruned_partition_list`, which can now return files with NULL + partition values. Partition fields built from its output must be nullable. +- Users constructing `ExecutionOptions` (or `ConfigOptions`) with an exhaustive + struct literal must provide the new + `listing_table_hive_default_partition_name: String` field. + +**Migration guide:** + +Select the null partition with `IS NULL`: + +```sql +SELECT * FROM t WHERE col IS NULL; +``` + +To keep reading and writing `__HIVE_DEFAULT_PARTITION__` as a literal value, +disable the mapping: + +```sql +SET datafusion.execution.listing_table_hive_default_partition_name = ''; +``` + +If your writer uses a different `hive.exec.default.partition.name`, set the +option to that name instead: + +```rust,ignore +let mut config = SessionConfig::new(); +config + .options_mut() + .execution + .listing_table_hive_default_partition_name = "MY_NULL".to_string(); +``` + ### `Timestamp(u, Some(tz)) - Timestamp(u, None)` reads the naive operand in `tz` Subtracting a timezone-naive timestamp from a timezone-aware one _at the same diff --git a/docs/source/user-guide/configs.md b/docs/source/user-guide/configs.md index ec2977c81c459..8925fffc21f95 100644 --- a/docs/source/user-guide/configs.md +++ b/docs/source/user-guide/configs.md @@ -96,7 +96,7 @@ The following configuration settings are available: | datafusion.execution.parquet.max_predicate_cache_size | NULL | (reading) The maximum predicate cache size, in bytes. When `pushdown_filters` is enabled, sets the maximum memory used to cache the results of predicate evaluation between filter evaluation and output generation. Decreasing this value will reduce memory usage, but may increase IO and CPU usage. None means use the default parquet reader setting. 0 means no caching. | | datafusion.execution.parquet.max_in_list_size | 20 | Maximum number of input values in an `IN (...)` list eligible for min/max pruning. Lists above this cap, or a cap of 0, skip this rewrite; other predicates and Bloom-filter pruning remain available. Within the cap, nonempty lists of at most 20 values use the existing per-value rewrite. Larger literal lists use a compact representation when the column type is string, variable-length binary, integer, decimal, date, time, timestamp, or duration. This applies to both `IN` and `NOT IN`, including lists with NULL members. `NOT IN` with NULL and all-NULL `IN` lists cannot match any rows. Compact lists containing NULL do not use the fully-matched-row-group optimization. Floating-point and other lists retain the existing per-value rewrite, so raising the cap can make those predicates expensive to build and evaluate. Defaults to 20. | | datafusion.execution.parquet.enable_rle_to_dictionary | false | (reading) If true, top-level string and binary Parquet columns with dictionary pages are inferred and scanned as `Dictionary` / `Dictionary` instead of their plain value type. This applies only when DataFusion infers the table schema. Tables with a user-supplied schema are not promoted because the Parquet footer is not read at DDL time, so dictionary pages cannot be detected per column. See | -| datafusion.execution.parquet.row_group_range_assignment | start_offset | (reading) Which byte range of a split file reads each row group. `start_offset` picks the range containing the row group's start. `midpoint` picks the range containing its midpoint, as Spark does, which spreads large row groups more evenly across ranges. | +| datafusion.execution.parquet.row_group_range_assignment | `start_offset` | (reading) Which byte range of a split file reads each row group. `start_offset` picks the range containing the row group's start. `midpoint` picks the range containing its midpoint, as Spark does, which spreads large row groups more evenly across ranges. | | datafusion.execution.parquet.data_pagesize_limit | 1048576 | (writing) Sets best effort maximum size of data page in bytes | | datafusion.execution.parquet.write_batch_size | 1024 | (writing) Sets write_batch_size in rows | | datafusion.execution.parquet.writer_version | 1.0 | (writing) Sets parquet writer version valid values are "1.0" and "2.0" | @@ -139,6 +139,7 @@ The following configuration settings are available: | datafusion.execution.max_buffered_batches_per_output_file | 2 | This is the maximum number of RecordBatches buffered for each output file being worked. Higher values can potentially give faster write performance at the cost of higher peak memory consumption. This budget is split evenly between two independent points in the write pipeline (see the demuxer diagram in #7791): how many files can be in flight from the demuxer to a writer task, and how many RecordBatches are buffered for a single file's writer. Must be at least 2 so each half gets at least 1 unit of buffering - 0 or 1 would leave one side with a zero-capacity channel and panic at write time. | | datafusion.execution.listing_table_ignore_subdirectory | true | Should sub directories be ignored when scanning directories for data files. Defaults to true (ignores subdirectories), consistent with Hive. Note that this setting does not affect reading partitioned tables (e.g. `/table/year=2021/month=01/data.parquet`). | | datafusion.execution.listing_table_factory_infer_partitions | true | Should a `ListingTable` created through the `ListingTableFactory` infer table partitions from Hive compliant directories. Defaults to true (partition columns are inferred and will be represented in the table schema). | +| datafusion.execution.listing_table_hive_default_partition_name | `__HIVE_DEFAULT_PARTITION__` | 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. | | datafusion.execution.enable_recursive_ctes | true | Should DataFusion support recursive CTEs | | datafusion.execution.split_file_groups_by_statistics | false | Attempt to eliminate sorts by packing & sorting files with non-overlapping statistics into the same file groups. Currently experimental | | datafusion.execution.keep_partition_by_columns | false | Should DataFusion keep the columns used for partition_by in the output RecordBatches | @@ -206,7 +207,7 @@ The following configuration settings are available: | datafusion.sql_parser.map_string_types_to_utf8view | true | If true, string types (VARCHAR, CHAR, Text, and String) are mapped to `Utf8View` during SQL planning. If false, they are mapped to `Utf8`. Default is true. | | datafusion.sql_parser.collect_spans | false | When set to true, the source locations relative to the original SQL query (i.e. [`Span`](https://docs.rs/sqlparser/latest/sqlparser/tokenizer/struct.Span.html)) will be collected and recorded in the logical plan nodes. | | datafusion.sql_parser.recursion_limit | 51 | Specifies the recursion depth limit when parsing complex SQL Queries | -| datafusion.sql_parser.default_null_ordering | nulls_max | Specifies the default null ordering for query results. There are 4 options: - `nulls_max`: Nulls appear last in ascending order. - `nulls_min`: Nulls appear first in ascending order. - `nulls_first`: Nulls always be first in any order. - `nulls_last`: Nulls always be last in any order. By default, `nulls_max` is used to follow Postgres's behavior. postgres rule: | +| datafusion.sql_parser.default_null_ordering | `nulls_max` | Specifies the default null ordering for query results. There are 4 options: - `nulls_max`: Nulls appear last in ascending order. - `nulls_min`: Nulls appear first in ascending order. - `nulls_first`: Nulls always be first in any order. - `nulls_last`: Nulls always be last in any order. By default, `nulls_max` is used to follow Postgres's behavior. postgres rule: | | datafusion.sql_parser.enable_subquery_sort_elimination | true | When set to true, DataFusion may remove `ORDER BY` clauses from subqueries or CTEs during SQL planning when their ordering cannot affect the result, such as when no `LIMIT` or other order-sensitive operator depends on them. Disable this option to preserve explicit subquery ordering in the planned query. | | datafusion.format.safe | true | If set to `true` any formatting errors will be written to the output instead of being converted into a [`std::fmt::Error`] | | datafusion.format.null | | Format string for nulls |