diff --git a/datafusion/datasource-parquet/src/metadata.rs b/datafusion/datasource-parquet/src/metadata.rs index 4781a309f5bcc..0e90651767d7e 100644 --- a/datafusion/datasource-parquet/src/metadata.rs +++ b/datafusion/datasource-parquet/src/metadata.rs @@ -45,8 +45,8 @@ use parquet::arrow::arrow_reader::statistics::StatisticsConverter; use parquet::arrow::{parquet_column, parquet_to_arrow_schema}; use parquet::basic::{ColumnOrder, SortOrder, Type as PhysicalType}; use parquet::file::metadata::{ - PageIndexPolicy, ParquetMetaData, ParquetMetaDataPushDecoder, ParquetMetaDataReader, - RowGroupMetaData, SortingColumn, + FileMetaData, PageIndexPolicy, ParquetMetaData, ParquetMetaDataPushDecoder, + ParquetMetaDataReader, RowGroupMetaData, SortingColumn, }; use parquet::file::statistics::Statistics as ParquetStatistics; use parquet::schema::types::{ColumnDescriptor, SchemaDescriptor}; @@ -112,6 +112,35 @@ pub(crate) fn has_untrusted_byte_array_stats<'a>( }) } +/// Whether a missing Parquet row-group `null_count` can be treated as exactly +/// zero for a file written by the writer recorded in `created_by`. +/// +/// parquet-rs before 53.1.0 (fixed in apache/arrow-rs#6490) did not record a +/// `null_count` when it was zero, so for those writers a missing count is +/// exactly zero. Files written by DataFusion < 42.1.0 could link such a +/// parquet-rs, so their missing counts are exactly zero as well. +/// +/// For every other writer a missing `null_count` is unknown. Treating it as +/// zero would let `IS NULL` / `COUNT` pruning and limit pruning return the +/// wrong results, so those files must be handled conservatively. +pub(crate) fn missing_null_counts_are_zero(file_metadata: &FileMetaData) -> bool { + let Some((writer, version)) = file_metadata + .created_by() + .and_then(|s| s.split_once(" version ")) + else { + return false; + }; + let mut parts = version.split(['.', ' ', '-']).map(str::parse::); + let (Some(Ok(major)), Some(Ok(minor))) = (parts.next(), parts.next()) else { + return false; + }; + match writer { + "parquet-rs" => (major, minor) < (53, 1), + "datafusion" => (major, minor) < (42, 1), + _ => false, + } +} + /// Handles fetching Parquet file schema, metadata and statistics /// from object store. /// @@ -519,6 +548,7 @@ impl<'a> DFParquetMetadata<'a> { statistics.num_rows = Precision::Exact(num_rows); let file_metadata = metadata.file_metadata(); + let missing_null_counts_as_zero = missing_null_counts_are_zero(file_metadata); let mut physical_file_schema = parquet_to_arrow_schema( file_metadata.schema_descr(), file_metadata.key_value_metadata(), @@ -551,10 +581,23 @@ impl<'a> DFParquetMetadata<'a> { file_metadata.schema_descr(), ) { Ok(stats_converter) => { + + // A missing null_count is only exactly zero for + // writers known to omit it (i.e. old parquet-rs). + // For every other writer a missing count is + // unknown, and must not surface as an exact zero + // in file statistics used by pruning, aggregates + // and sort pushdown. + let stats_converter = stats_converter + .with_missing_null_counts_as_zero( + missing_null_counts_as_zero, + ); + // An omitted count must not become an exact zero in // file statistics used for pruning and aggregates. let stats_converter = stats_converter.with_missing_null_counts_as_zero(false); + let parquet_index = stats_converter.parquet_column_index(); if parquet_index.is_some_and(|index| { has_untrusted_min_max_order( @@ -1192,6 +1235,93 @@ mod tests { use arrow::array::Int32Array; use arrow::compute::SortOptions; use arrow::datatypes::Field; + use parquet::schema::types::Type as ParquetType; + + /// Builds `FileMetaData` for a single-column INT32 schema with the given + /// `created_by` string. + fn file_metadata_with_created_by(created_by: Option<&str>) -> FileMetaData { + let schema = Arc::new(SchemaDescriptor::new(Arc::new( + ParquetType::group_type_builder("schema") + .with_fields(vec![Arc::new( + ParquetType::primitive_type_builder("a", PhysicalType::INT32) + .build() + .unwrap(), + )]) + .build() + .unwrap(), + ))); + FileMetaData::new(1, 0, created_by.map(str::to_string), None, schema, None) + } + + #[test] + fn test_missing_null_counts_are_zero() { + // parquet-rs < 53.1.0 omitted null counts that are zero + for created_by in [ + "parquet-rs version 5.1.0", + "parquet-rs version 52.0.1", + "parquet-rs version 53.0.0", + "parquet-rs version 53.0.0 (build abc)", + ] { + assert!( + missing_null_counts_are_zero(&file_metadata_with_created_by(Some( + created_by + ))), + "expected {created_by:?} missing counts to be treated as zero" + ); + } + // parquet-rs >= 53.1.0 always records null counts + for created_by in [ + "parquet-rs version 53.1.0", + "parquet-rs version 59.3.0", + "parquet-rs version 60.0.0", + ] { + assert!( + !missing_null_counts_are_zero(&file_metadata_with_created_by(Some( + created_by + ))), + "expected {created_by:?} missing counts to stay unknown" + ); + } + // DataFusion < 42.1.0 may link a parquet-rs that omits zero null counts + for created_by in [ + "datafusion version 5.1.0", + "datafusion version 41.1.2", + "datafusion version 42.0.0", + ] { + assert!( + missing_null_counts_are_zero(&file_metadata_with_created_by(Some( + created_by + ))), + "expected {created_by:?} missing counts to be treated as zero" + ); + } + // DataFusion >= 42.1.0 always records null counts + for created_by in ["datafusion version 42.1.0", "datafusion version 55.1.0"] { + assert!( + !missing_null_counts_are_zero(&file_metadata_with_created_by(Some( + created_by + ))), + "expected {created_by:?} missing counts to stay unknown" + ); + } + // Unknown writers, unparsable versions and absent created_by are + // conservative: missing counts stay unknown + for created_by in [ + None, + Some("parquet-mr version 1.13.1"), + Some("duckdb version 1.1.0"), + Some("custom string"), + Some("parquet-rs"), + Some("parquet-rs version not-a-version"), + Some("datafusion version 42"), + ] { + let metadata = file_metadata_with_created_by(created_by); + assert!( + !missing_null_counts_are_zero(&metadata), + "expected {created_by:?} missing counts to stay unknown" + ); + } + } #[test] fn test_lex_ordering_to_sorting_columns_uses_writer_schema() -> Result<()> { diff --git a/datafusion/datasource-parquet/src/push_decoder.rs b/datafusion/datasource-parquet/src/push_decoder.rs index 77947bd8af455..6a13308150c7c 100644 --- a/datafusion/datasource-parquet/src/push_decoder.rs +++ b/datafusion/datasource-parquet/src/push_decoder.rs @@ -62,6 +62,7 @@ use datafusion_pruning::{PruningPredicate, PruningPredicateBuilder}; use crate::ParquetFileMetrics; use crate::access_plan::PreparedAccessPlan; use crate::decoder_projection::DecoderProjection; +use crate::metadata::missing_null_counts_are_zero; use crate::metrics::{ByteProgress, RowFilterSkippedFullyMatchedMetric}; use crate::row_filter::{ PrebuiltRowFilterCandidate, prebuild_row_filter_candidates, row_filter_from_prebuilt, @@ -244,15 +245,19 @@ impl RowGroupPruner { .iter() .map(|&i| self.parquet_metadata.row_group(i)) .collect::>(); + let file_metadata = self.parquet_metadata.file_metadata(); let stats = RowGroupPruningStatistics { - parquet_schema: self.parquet_metadata.file_metadata().schema_descr(), - column_orders: self - .parquet_metadata - .file_metadata() - .column_orders() - .map(Vec::as_slice), + parquet_schema: file_metadata.schema_descr(), + column_orders: file_metadata.column_orders().map(Vec::as_slice), row_group_metadatas, arrow_schema: self.arrow_schema.as_ref(), + + // Match the static row-group pruning behavior: a missing null count + // is exactly zero for old parquet-rs / DataFusion writers and + // unknown for everyone else. Runtime pruning only needs to prove a + // row group *cannot* contain matching rows, so this is sound. + missing_null_counts_as_zero: missing_null_counts_are_zero(file_metadata), + }; match pp.prune(&stats) { diff --git a/datafusion/datasource-parquet/src/row_group_filter.rs b/datafusion/datasource-parquet/src/row_group_filter.rs index ab690e345d5f5..71019b8b1905b 100644 --- a/datafusion/datasource-parquet/src/row_group_filter.rs +++ b/datafusion/datasource-parquet/src/row_group_filter.rs @@ -20,7 +20,10 @@ use std::sync::Arc; use super::{ParquetAccessPlan, ParquetFileMetrics, RowGroupAccess}; use crate::bloom_filter::BloomFilterStatistics; -use crate::metadata::{has_untrusted_byte_array_stats, has_untrusted_min_max_order}; +use crate::metadata::{ + has_untrusted_byte_array_stats, has_untrusted_min_max_order, + missing_null_counts_are_zero, +}; use crate::pruning::build_inverted_predicate; use arrow::array::{ArrayRef, BooleanArray, UInt64Array}; use arrow::compute::nullif; @@ -31,7 +34,7 @@ use datafusion_datasource::FileRange; use datafusion_pruning::PruningPredicate; use parquet::arrow::arrow_reader::statistics::StatisticsConverter; use parquet::basic::ColumnOrder; -use parquet::file::metadata::{ParquetMetaData, RowGroupMetaData}; +use parquet::file::metadata::{FileMetaData, ParquetMetaData, RowGroupMetaData}; use parquet::schema::types::SchemaDescriptor; /// Reduces the [`ParquetAccessPlan`] based on row group level metadata. @@ -278,9 +281,9 @@ impl RowGroupAccessPlanFilter { arrow_schema, parquet_schema, groups, - None, predicate, metrics, + None, ); } @@ -304,9 +307,9 @@ impl RowGroupAccessPlanFilter { arrow_schema, file_metadata.schema_descr(), metadata.row_groups(), - file_metadata.column_orders().map(Vec::as_slice), predicate, metrics, + Some(file_metadata), ); } @@ -315,9 +318,9 @@ impl RowGroupAccessPlanFilter { arrow_schema: &Schema, parquet_schema: &SchemaDescriptor, groups: &[RowGroupMetaData], - column_orders: Option<&[ColumnOrder]>, predicate: &PruningPredicate, metrics: &ParquetFileMetrics, + file_metadata: Option<&FileMetaData>, ) { // scoped timer updates on drop let _timer_guard = metrics.statistics_eval_time.timer(); @@ -330,11 +333,29 @@ impl RowGroupAccessPlanFilter { .map(|&i| &groups[i]) .collect::>(); + // A missing null count is exactly zero for old parquet-rs writers, but + // unknown for everything else. When no footer metadata is available + // (this only happens in tests), fall back to the StatisticsConverter + // default that treats a missing count as zero to preserve the original + // pruning behavior. + let missing_null_counts_as_zero = file_metadata + .map(missing_null_counts_are_zero) + .unwrap_or(true); + + // The footer also records the comparison order of string and binary + // bounds; `prune_by_statistics` has no footer and passes `None`. + let column_orders = file_metadata + .and_then(|metadata| metadata.column_orders()) + .map(Vec::as_slice); + let pruning_stats = RowGroupPruningStatistics { parquet_schema, column_orders, row_group_metadatas, arrow_schema, + + missing_null_counts_as_zero, + }; // try to prune the row groups in a single call @@ -351,10 +372,17 @@ impl RowGroupAccessPlanFilter { } } - // Check if any of the matched row groups are fully contained by the predicate + // Fully matched row groups require a stronger proof: every row + // must pass the predicate. When no footer is available, fall + // back to false (not true) so that the fully-matched claim + // stays sound for limit pruning. + let fully_matched_flag = file_metadata + .map(missing_null_counts_are_zero) + .unwrap_or(false); self.identify_fully_matched_row_groups( &fully_contained_candidates_original_idx, &pruning_stats, + fully_matched_flag, groups, predicate, metrics, @@ -380,6 +408,7 @@ impl RowGroupAccessPlanFilter { &mut self, candidate_row_group_indices: &[usize], pruning_stats: &RowGroupPruningStatistics<'_>, + fully_matched_missing_null_counts_as_zero: bool, groups: &[RowGroupMetaData], predicate: &PruningPredicate, metrics: &ParquetFileMetrics, @@ -402,6 +431,15 @@ impl RowGroupAccessPlanFilter { .map(|&i| &groups[i]) .collect::>(), arrow_schema, + + // Fully matched row groups require a stronger proof: every row + // must pass the predicate. Use the flag derived from the footer + // metadata, which for old parquet-rs writers confirms that a + // missing count is genuinely zero. Without footer metadata, use + // false so the claim stays sound for limit pruning. + missing_null_counts_as_zero: fully_matched_missing_null_counts_as_zero, + + }; let Ok(inverted_values) = inverted_predicate.prune(&inverted_pruning_stats) diff --git a/datafusion/datasource-parquet/src/statistics_order_tests.rs b/datafusion/datasource-parquet/src/statistics_order_tests.rs index 2156a92dcdf6b..24f2a1a530652 100644 --- a/datafusion/datasource-parquet/src/statistics_order_tests.rs +++ b/datafusion/datasource-parquet/src/statistics_order_tests.rs @@ -288,6 +288,147 @@ impl TestFile { }) .sum() } + + /// Builds the same `[s, n]` fixture as [`Self::new`], but with `created_by` + /// recorded in the footer as `created_by` and with the `null_count` field + /// removed from the row-group statistics exactly as parquet-rs < 53.1.0 + /// wrote it: "s" omits its null count from the row groups that truly have + /// zero nulls (groups 0 and 2, keeping the count of 3 in group 1), while + /// the non-null "n" column omits its count from every group. + fn with_missing_null_counts(created_by: &str) -> Self { + let batch = record_batch!( + ( + "s", + Utf8, + vec![ + Some("a"), + Some("b"), + Some("c"), + None, + None, + None, + Some("g"), + Some("h"), + Some("i"), + ] + ), + ("n", Int32, [1, 2, 3, 10, 11, 12, 20, 21, 22]) + ) + .unwrap(); + let schema = batch.schema(); + let properties = WriterProperties::builder() + .set_max_row_group_row_count(Some(3)) + .set_data_page_row_count_limit(3) + .set_write_batch_size(3) + .set_dictionary_enabled(false) + .set_statistics_enabled(EnabledStatistics::Page) + .set_created_by(created_by.to_string()) + .build(); + let mut original = Vec::new(); + let mut writer = + ArrowWriter::try_new(&mut original, Arc::clone(&schema), Some(properties)) + .unwrap(); + writer.write(&batch).unwrap(); + writer.close().unwrap(); + let original = Bytes::from(original); + let metadata = read_metadata(&original); + assert_eq!(metadata.num_row_groups(), 3); + + let strip_int32_null_count = |stats: &ParquetStatistics| -> Option { + stats.min_bytes_opt().and_then(|bytes| { + let bytes: &[u8; 4] = bytes.try_into().ok()?; + Some(i32::from_le_bytes(*bytes)) + }) + }; + + let mut row_groups = metadata.row_groups().to_vec(); + for (group_idx, group) in row_groups.iter_mut().enumerate() { + let mut columns = group.columns().to_vec(); + + // "s": the null count is only omitted where it is zero + if group_idx != 1 { + let stats = columns[0].statistics().unwrap(); + columns[0] = columns[0] + .clone() + .into_builder() + .set_statistics(ParquetStatistics::byte_array( + stats + .min_bytes_opt() + .map(|bytes| ByteArray::from(bytes.to_vec())), + stats + .max_bytes_opt() + .map(|bytes| ByteArray::from(bytes.to_vec())), + stats.distinct_count_opt(), + None, + stats.is_min_max_deprecated(), + )) + .build() + .unwrap(); + } + + // "n": never null, so the zero count is omitted everywhere + let stats = columns[1].statistics().unwrap(); + columns[1] = columns[1] + .clone() + .into_builder() + .set_statistics(ParquetStatistics::int32( + strip_int32_null_count(stats), + strip_int32_null_count(stats), + stats.distinct_count_opt(), + None, + stats.is_min_max_deprecated(), + )) + .build() + .unwrap(); + + *group = group + .clone() + .into_builder() + .set_column_metadata(columns) + .build() + .unwrap(); + } + let metadata = metadata.into_builder().set_row_groups(row_groups).build(); + + // Keep the real data pages and serialize the replacement row-group + // statistics at their actual file offsets. + let mut bytes = Vec::new(); + let mut tracked = TrackedWrite::new(&mut bytes); + tracked + .write_all(&original[..footer_start(&original)]) + .unwrap(); + ParquetMetaDataWriter::new_with_tracked(tracked, &metadata) + .finish() + .unwrap(); + + let bytes = Bytes::from(bytes); + let metadata = read_metadata(&bytes); + assert_eq!(metadata.file_metadata().created_by(), Some(created_by)); + // Sanity check: the counts really are missing from the metadata + for (i, group) in metadata.row_groups().iter().enumerate() { + assert_eq!( + group + .column(0) + .statistics() + .and_then(|stats| stats.null_count_opt()), + if i == 1 { Some(3) } else { None }, + "unexpected 's' null count in row group {i}" + ); + assert!( + group + .column(1) + .statistics() + .and_then(|stats| stats.null_count_opt()) + .is_none(), + "expected missing 'n' null count in row group {i}" + ); + } + Self { + bytes, + schema, + metadata: Arc::new(metadata), + } + } } fn footer_start(bytes: &[u8]) -> usize { @@ -814,3 +955,93 @@ fn undefined_logical_byte_array_order_is_not_a_bound() { ); assert!(pages.should_scan(0)); } + +/// Writers that are known to omit a `null_count` when it is zero. +const OLD_WRITERS: [&str; 2] = ["parquet-rs version 53.0.0", "datafusion version 42.0.0"]; + +/// Writers that always record null counts, or whose created_by is unparsable. +const OTHER_WRITERS: [&str; 3] = [ + "parquet-rs version 53.1.0", + "parquet-mr version 1.13.1", + "custom string", +]; + +/// A missing row-group `null_count` must be read exactly the way the writer +/// meant it: exactly zero for parquet-rs < 53.1.0 / DataFusion < 42.1.0, and +/// unknown for every other writer. +#[test] +fn missing_null_counts_are_zero_only_for_writers_that_omitted_them() { + let cases: Vec<(&str, bool)> = OLD_WRITERS + .iter() + .map(|writer| (*writer, true)) + .chain(OTHER_WRITERS.iter().map(|writer| (*writer, false))) + .collect(); + + for (created_by, is_old_writer) in cases { + let file = TestFile::with_missing_null_counts(created_by); + + // File-level statistics feed aggregate (`COUNT`), file-level filter + // pruning and sort pushdown. "s" has 3 nulls in total; "n" has none. + let statistics = file.statistics(); + assert_eq!( + statistics.column_statistics[0].null_count, + if is_old_writer { + Precision::Exact(3) + } else { + Precision::Inexact(3) + }, + "'s' null count for {created_by}" + ); + // "n" gates sort pushdown: `try_pushdown_sort` only claims Exact when + // the projected sort column's null count is Precision::Exact(0). + assert_eq!( + statistics.column_statistics[1].null_count, + if is_old_writer { + Precision::Exact(0) + } else { + Precision::Absent + }, + "'n' null count for {created_by}" + ); + + // Static row-group pruning: `s IS NULL` skips the groups whose null + // count is known to be zero and keeps the group that has nulls. + let (physical, predicate) = file.predicate(&col("s").is_null()); + let row_groups = file.row_group_plan(&predicate); + assert_eq!( + row_groups.row_group_indexes(), + if is_old_writer { + vec![1] // groups 0 and 2 have exactly zero nulls + } else { + vec![0, 1, 2] // unknown counts can't be pruned + }, + "row groups scanned for {created_by}" + ); + // Whatever is pruned away, the query must still return every null row. + assert_eq!(file.matching_rows(&physical, row_groups), 3); + + // Runtime pruning (e.g. dynamic TopK row-group pruning) must agree. + let mut runtime_pruner = RowGroupPruner::new( + Arc::clone(&physical), + Arc::clone(&file.schema), + Arc::clone(&file.metadata), + Count::new(), + Count::new(), + MAX_IN_LIST_SIZE, + ); + assert_eq!( + runtime_pruner.should_prune(&[0]), + is_old_writer, + "should_prune group 0 for {created_by}" + ); + assert!( + !runtime_pruner.should_prune(&[1]), + "row group with nulls must be scanned for {created_by}" + ); + assert_eq!( + runtime_pruner.should_prune(&[2]), + is_old_writer, + "should_prune group 2 for {created_by}" + ); + } +} 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 7658ce60eddf2..7db3c3f661ae9 100644 --- a/docs/source/library-user-guide/upgrading/56.0.0.md +++ b/docs/source/library-user-guide/upgrading/56.0.0.md @@ -314,3 +314,31 @@ let df = DataFrame::from_columns([ Most existing call sites using `Vec` require no changes. Code that relies on the exact non-generic function signature of `DataFrame::from_columns` may need to be updated to account for the new generic API. + +### Missing Parquet null counts are treated as exactly zero only for files written by parquet-rs < 53.1.0 and DataFusion < 42.1.0 + +A Parquet file's `created_by` value is now used to decide how a missing (absent) +`null_count` in a column's row-group statistics is interpreted: + +- If the file was written by `parquet-rs` before version `53.1.0`, or by + DataFusion before version `42.1.0` (a release that could link a + parquet-rs < 53.1.0), a missing `null_count` means the column has exactly + zero null values. Pruning, file statistics, `COUNT` / `IS NULL` optimization + and sort pushdown treat it as `Exact(0)`. +- For every other writer (or when `created_by` is missing or unparsable), a + missing `null_count` is treated as unknown. Row-group pruning and file + statistics no longer assume it is zero, which prevents queries from returning + incorrect results on files whose null counts are genuinely absent (for + example, files written by other tools that omit null counts). + +**Who is affected:** + +- Users reading Parquet files written by older `parquet-rs` (< 53.1.0) or + DataFusion (< 42.1.0) writers regain correct row-group pruning, exact file + statistics, and sort pushdown for free. +- Users reading Parquet files from other writers that omit `null_count` + (uncommon; e.g. hand-constructed files) may see slightly less efficient + pruning or conservative statistics until the file is rewritten with a + current writer. + +See [issue #25253](https://github.com/apache/datafusion/issues/25253).