From b8c7ba40874d57953620d44fe9b14d1011f50940 Mon Sep 17 00:00:00 2001 From: peterxcli Date: Wed, 16 Sep 2026 02:11:57 +0800 Subject: [PATCH] feat: separate Parquet I/O policy from filter pushdown --- datafusion/common/src/config.rs | 7 + .../common/src/file_options/parquet_writer.rs | 3 + .../core/tests/parquet/filter_pushdown.rs | 177 +++++++++ .../datasource-parquet/src/file_format.rs | 1 + .../datasource-parquet/src/opener/mod.rs | 82 ++-- .../datasource-parquet/src/push_decoder.rs | 375 +++++++++++++++++- .../datasource-parquet/src/row_filter.rs | 2 +- datafusion/datasource-parquet/src/source.rs | 12 + .../proto/datafusion_common.proto | 3 +- datafusion/proto-common/src/from_proto/mod.rs | 16 + .../proto-common/src/generated/pbjson.rs | 18 + .../proto-common/src/generated/prost.rs | 3 + datafusion/proto-common/src/to_proto/mod.rs | 1 + datafusion/proto-models/src/from_proto.rs | 1 + .../src/generated/datafusion_proto_common.rs | 3 + datafusion/proto/tests/cases/plans/sources.rs | 1 + .../tests/cases/roundtrip_logical_plan.rs | 2 + datafusion/pruning/src/pruning_predicate.rs | 36 +- .../dynamic_filter_pushdown_config.slt | 2 +- .../test_files/explain_analyze.slt | 20 +- .../test_files/information_schema.slt | 2 + .../sqllogictest/test_files/limit_pruning.slt | 2 +- docs/source/user-guide/configs.md | 1 + 23 files changed, 709 insertions(+), 61 deletions(-) diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs index 360586b0e9bae..5459b025c5355 100644 --- a/datafusion/common/src/config.rs +++ b/datafusion/common/src/config.rs @@ -1336,6 +1336,13 @@ config_namespace! { /// reduce the number of rows decoded. This optimization is sometimes called "late materialization". pub pushdown_filters: bool, default = false + /// (reading) Fetch columns progressively as row filtering and decoding need + /// them. If false, fetch output and predicate pages together before evaluating + /// row filters, preserving page-index pruning performed at file open. This + /// can reduce dependent I/O rounds but may fetch pages that row filtering + /// would otherwise avoid. Does not enable filter pushdown or prefetch. + pub progressive_io: bool, default = true + /// (reading) If true, filter expressions evaluated during the parquet decoding operation /// will be reordered heuristically to minimize the cost of evaluation. If false, /// the filters are applied in the same order as written in the query diff --git a/datafusion/common/src/file_options/parquet_writer.rs b/datafusion/common/src/file_options/parquet_writer.rs index c50bb42a38ef7..abb602d0a227a 100644 --- a/datafusion/common/src/file_options/parquet_writer.rs +++ b/datafusion/common/src/file_options/parquet_writer.rs @@ -236,6 +236,7 @@ impl ParquetOptions { skip_metadata: _, metadata_size_hint: _, pushdown_filters: _, + progressive_io: _, reorder_filters: _, force_filter_selections: _, // not used for writer props allow_single_file_parallelism: _, @@ -493,6 +494,7 @@ mod tests { skip_metadata: defaults.skip_metadata, metadata_size_hint: defaults.metadata_size_hint, pushdown_filters: defaults.pushdown_filters, + progressive_io: defaults.progressive_io, reorder_filters: defaults.reorder_filters, force_filter_selections: defaults.force_filter_selections, allow_single_file_parallelism: defaults.allow_single_file_parallelism, @@ -579,6 +581,7 @@ mod tests { TableParquetOptions { global: ParquetOptions { + progressive_io: true, // global options data_pagesize_limit: props.data_page_size_limit(), write_batch_size: props.write_batch_size(), diff --git a/datafusion/core/tests/parquet/filter_pushdown.rs b/datafusion/core/tests/parquet/filter_pushdown.rs index d337979e5fd00..eaca57d7e963c 100644 --- a/datafusion/core/tests/parquet/filter_pushdown.rs +++ b/datafusion/core/tests/parquet/filter_pushdown.rs @@ -809,3 +809,180 @@ async fn pushed_down_predicate_reports_the_original_error() { "expected the original cast error, got {root:?}" ); } + +#[tokio::test] +async fn upfront_io_preserves_page_pruning_from_session_configuration() { + use arrow::array::{Int64Array, StringArray, StructArray}; + use arrow::datatypes::{DataType, Field, Schema}; + use datafusion_common::tree_node::{TreeNode, TreeNodeRecursion}; + use datafusion_datasource::source::DataSourceExec; + use datafusion_datasource_parquet::source::ParquetSource; + use parquet::arrow::ArrowWriter; + use parquet::file::properties::EnabledStatistics; + use std::sync::Arc; + + let nested_fields = vec![Arc::new(Field::new("value", DataType::Int64, false))]; + let schema = Arc::new(Schema::new(vec![ + Field::new("key", DataType::Int64, false), + Field::new("payload", DataType::Utf8, false), + Field::new( + "nested", + DataType::Struct(nested_fields.clone().into()), + false, + ), + ])); + let temp = TempDir::new().unwrap(); + for clustered in [false, true] { + let expected: Vec = (0..3) + .flat_map(|group| { + (0..4096) + .filter(move |row| { + if clustered { + *row < 100 + } else { + row % 128 < 100 + } + }) + .map(move |row| group * 4096 + row) + }) + .collect(); + for indexed in [false, true] { + let path = temp + .path() + .join(format!("pages-{indexed}-{clustered}.parquet")); + let props = WriterProperties::builder() + .set_max_row_group_row_count(Some(4096)) + .set_data_page_row_count_limit(128) + .set_write_batch_size(128) + .set_dictionary_enabled(true) + .set_statistics_enabled(if indexed { + EnabledStatistics::Page + } else { + EnabledStatistics::Chunk + }) + .set_offset_index_disabled(!indexed) + .build(); + let mut writer = ArrowWriter::try_new( + File::create(&path).unwrap(), + Arc::clone(&schema), + Some(props), + ) + .unwrap(); + for group in 0..3i64 { + let nested = StructArray::new( + nested_fields.clone().into(), + vec![Arc::new(Int64Array::from_iter_values( + (0..4096).map(|row| group * 4096 + row), + ))], + None, + ); + let batch = RecordBatch::try_new( + Arc::clone(&schema), + vec![ + Arc::new(Int64Array::from_iter_values( + (0..4096).map(|row| if clustered { row } else { row % 128 }), + )), + Arc::new(StringArray::from_iter_values( + (0..4096).map(|row| format!("value-{}", row % 11)), + )), + Arc::new(nested), + ], + ) + .unwrap(); + writer.write(&batch).unwrap(); + writer.flush().unwrap(); + } + let metadata = writer.close().unwrap(); + assert_eq!(metadata.num_row_groups(), 3); + for group in metadata.row_groups() { + assert!(group.column(1).dictionary_page_offset().is_some()); + assert_eq!(group.column(0).offset_index_offset().is_some(), indexed); + } + for predicate_in_output in [false, true] { + let mut expected_batch = None; + let mut baseline_bytes = None; + for (pushdown, progressive) in + [(false, true), (true, true), (true, false)] + { + let config = SessionConfig::new() + .with_target_partitions(1) + .with_batch_size(127); + let ctx = SessionContext::new_with_config(config); + for (key, value) in [ + ("pushdown_filters", pushdown), + ("progressive_io", progressive), + ] { + ctx.sql(&format!( + "SET datafusion.execution.parquet.{key} = {value}" + )) + .await + .unwrap() + .collect() + .await + .unwrap(); + } + ctx.register_parquet( + "t", + path.to_str().unwrap(), + ParquetReadOptions::default(), + ) + .await + .unwrap(); + let key = if predicate_in_output { "key," } else { "" }; + let plan = ctx.sql(&format!("SELECT {key} payload, nested.value AS value FROM t WHERE key < 100 ORDER BY value")) + .await.unwrap().create_physical_plan().await.unwrap(); + let mut scans = 0; + plan.apply(|plan| { + if let Some(exec) = plan.downcast_ref::() + && let Some((_, source)) = + exec.downcast_to_file_source::() + { + scans += 1; + assert_eq!( + source.table_parquet_options().global.progressive_io, + progressive + ); + } + Ok(TreeNodeRecursion::Continue) + }) + .unwrap(); + assert_eq!(scans, 1); + let batches = + collect(Arc::clone(&plan), ctx.task_ctx()).await.unwrap(); + let batch = concat_batches(&plan.schema(), &batches).unwrap(); + let values = batch + .column(batch.num_columns() - 1) + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!(values.values().as_ref(), expected.as_slice()); + if let Some(expected) = &expected_batch { + assert_eq!(&batch, expected); + } else { + expected_batch = Some(batch); + } + let metrics = TestParquetFile::parquet_metrics(&plan).unwrap(); + let bytes = get_value(&metrics, "bytes_scanned"); + if indexed && clustered { + assert!( + get_pruning_metrics(&metrics, "page_index_rows_pruned").0 > 0 + ); + } + if !predicate_in_output { + if let Some(expected) = baseline_bytes { + assert_eq!( + bytes, expected, + "clustered={clustered}, indexed={indexed}, pushdown={pushdown}, progressive={progressive}" + ); + } else { + baseline_bytes = Some(bytes); + } + } + if pushdown { + assert!(get_value(&metrics, "pushdown_rows_pruned") > 0); + } + } + } + } + } +} diff --git a/datafusion/datasource-parquet/src/file_format.rs b/datafusion/datasource-parquet/src/file_format.rs index 18f2b5a650c8d..a426a024c2897 100644 --- a/datafusion/datasource-parquet/src/file_format.rs +++ b/datafusion/datasource-parquet/src/file_format.rs @@ -728,6 +728,7 @@ impl From<&ParquetFormatFactory> for protobuf::TableParquetOptions { parquet_options::MetadataSizeHintOpt::MetadataSizeHint(size as u64) }), pushdown_filters: global_options.global.pushdown_filters, + progressive_io: Some(global_options.global.progressive_io), reorder_filters: global_options.global.reorder_filters, force_filter_selections: global_options.global.force_filter_selections, data_pagesize_limit: global_options.global.data_pagesize_limit as u64, diff --git a/datafusion/datasource-parquet/src/opener/mod.rs b/datafusion/datasource-parquet/src/opener/mod.rs index ec24462db0564..23f9a64ed2ad7 100644 --- a/datafusion/datasource-parquet/src/opener/mod.rs +++ b/datafusion/datasource-parquet/src/opener/mod.rs @@ -236,6 +236,7 @@ fn validate_predicate_does_not_reference_virtual_columns( /// as an explicit state machine -- see [`ParquetOpenState`] for details. #[derive(Clone)] pub(super) struct ParquetMorselizer { + pub(crate) progressive_io: bool, /// Execution partition index pub(crate) partition_index: usize, /// Projection to apply on top of the table schema (i.e. can reference partition columns). @@ -421,6 +422,7 @@ impl fmt::Debug for ParquetOpenState { } struct PreparedParquetOpen { + progressive_io: bool, partition_index: usize, partitioned_file: PartitionedFile, /// Tracks how much of this file range the scan has finished with. @@ -854,6 +856,7 @@ impl ParquetMorselizer { metrics: self.metrics.clone(), parquet_file_reader_factory: Arc::clone(&self.parquet_file_reader_factory), async_file_reader, + progressive_io: self.progressive_io, batch_size: self.batch_size, logical_file_schema: Arc::clone(&logical_file_schema), physical_file_schema: logical_file_schema, @@ -1471,6 +1474,7 @@ impl RowGroupsPrunedParquetOpen { prepared.virtual_state.as_deref(), )?; + let mut fetch_projection = decoder_projection.projection_mask().clone(); // Lazily-registered suppression counter shared by the open-time first-RG // skip below and the stream's per-RG toggle (registered on first use so // scans that never suppress don't carry a zero-valued counter). @@ -1522,6 +1526,20 @@ impl RowGroupsPrunedParquetOpen { decoder_limit: prepared.limit, }; + let mut fetch_selections = if !prepared.progressive_io { + access_plan + .inner() + .iter() + .map(|access| match access { + crate::RowGroupAccess::Selection(selection) => { + Some(selection.clone()) + } + _ => None, + }) + .collect::>() + } else { + vec![] + }; let prepared_access_plan = prepare_access_plan(access_plan)?; // #24355: a row selection (from page-index pruning, or an externally // supplied `ParquetRowSelection`) is carried by the decoder as one @@ -1548,6 +1566,9 @@ impl RowGroupsPrunedParquetOpen { .zip(prepared_access_plan.fully_matched.iter().copied()) .map(|(rg_index, fully_matched)| RgPlanEntry { rg_index, + row_selection: fetch_selections + .get_mut(rg_index) + .and_then(Option::take), fully_matched, bytes: row_group_bytes(&rg_metadata[rg_index]), }) @@ -1571,6 +1592,7 @@ impl RowGroupsPrunedParquetOpen { decoder_config.build(prepared_access_plan, reader_metadata.clone()); let mut filter_installed = false; if let Some(ctx) = row_filter_context.as_ref() { + ctx.extend_projection(&mut fetch_projection); if first_rg_fully_matched { // The first RG is fully matched: install an empty filter // and count the suppression, exactly as the per-RG toggle @@ -1679,6 +1701,10 @@ impl RowGroupsPrunedParquetOpen { active_reader: None, rg_plan, reader: prepared.async_file_reader, + progressive_io: prepared.progressive_io, + fetch_projection, + upfront_row_group: None, + parquet_metadata: Arc::clone(reader_metadata.metadata()), decoder_projection, arrow_reader_metrics, predicate_cache_inner_records, @@ -2311,6 +2337,7 @@ mod test { )?; Ok(ParquetMorselizer { + progressive_io: true, partition_index: self.partition_index, projection, batch_size: self.batch_size, @@ -3367,6 +3394,8 @@ mod test { let props = WriterProperties::builder() .set_max_row_group_row_count(Some(4)) + .set_data_page_row_count_limit(1) + .set_write_batch_size(1) .build(); let data_len = write_parquet_batches( @@ -3401,43 +3430,22 @@ mod test { ) .with_extension(access_plan); - let make_opener = |reverse_scan: bool| { - ParquetMorselizerBuilder::new() - .with_store(Arc::clone(&store)) - .with_schema(Arc::clone(&schema)) - .with_projection_indices(&[0]) - .with_reverse_row_groups(reverse_scan) - .build() - }; - - // Forward scan: RG0(3,4), RG1(5,6,7,8), RG2(9,10) - let opener = make_opener(false); - let stream = open_file(&opener, file.clone()).await.unwrap(); - let forward_values = collect_int32_values(stream).await; - - // Forward scan should produce: RG0(3,4), RG1(5,6,7,8), RG2(9,10) - assert_eq!( - forward_values, - vec![3, 4, 5, 6, 7, 8, 9, 10], - "Forward scan should select correct rows based on RowSelection" - ); - - // Reverse scan - // CORRECT behavior: reverse row groups AND their corresponding selections - // - RG2 is read first, WITH RG2's selection (select 2, skip 2) -> 9, 10 - // - RG1 is read second, WITH RG1's selection (select all) -> 5, 6, 7, 8 - // - RG0 is read third, WITH RG0's selection (skip 2, select 2) -> 3, 4 - let opener = make_opener(true); - let stream = open_file(&opener, file).await.unwrap(); - let reverse_values = collect_int32_values(stream).await; - - // Correct expected result: row groups reversed but each keeps its own selection - // RG2 with its selection (9,10), RG1 with its selection (5,6,7,8), RG0 with its selection (3,4) - assert_eq!( - reverse_values, - vec![9, 10, 5, 6, 7, 8, 3, 4], - "Reverse scan should reverse row group order while maintaining correct RowSelection for each group" - ); + for progressive_io in [false, true] { + for (reverse, expected) in [ + (false, vec![3, 4, 5, 6, 7, 8, 9, 10]), + (true, vec![9, 10, 5, 6, 7, 8, 3, 4]), + ] { + let mut opener = ParquetMorselizerBuilder::new() + .with_store(Arc::clone(&store)) + .with_schema(Arc::clone(&schema)) + .with_projection_indices(&[0]) + .with_reverse_row_groups(reverse) + .build(); + opener.progressive_io = progressive_io; + let stream = open_file(&opener, file.clone()).await.unwrap(); + assert_eq!(collect_int32_values(stream).await, expected); + } + } } #[tokio::test] diff --git a/datafusion/datasource-parquet/src/push_decoder.rs b/datafusion/datasource-parquet/src/push_decoder.rs index 77947bd8af455..cd1c6925cda39 100644 --- a/datafusion/datasource-parquet/src/push_decoder.rs +++ b/datafusion/datasource-parquet/src/push_decoder.rs @@ -36,6 +36,7 @@ //! [`PushDecoderStreamState::into_stream`] for consumption. use std::collections::VecDeque; +use std::ops::Range; use std::sync::Arc; use arrow::array::RecordBatch; @@ -47,7 +48,8 @@ use parquet::DecodeResult; use parquet::arrow::ProjectionMask; use parquet::arrow::arrow_reader::metrics::ArrowReaderMetrics; use parquet::arrow::arrow_reader::{ - ArrowReaderMetadata, ParquetRecordBatchReader, RowFilter, RowSelectionPolicy, + ArrowReaderMetadata, ParquetRecordBatchReader, RowFilter, RowSelection, + RowSelectionPolicy, }; use parquet::arrow::async_reader::AsyncFileReader; use parquet::arrow::push_decoder::{ParquetPushDecoder, ParquetPushDecoderBuilder}; @@ -114,6 +116,8 @@ impl DecoderBuilderConfig<'_> { #[derive(Debug, Clone)] pub(crate) struct RgPlanEntry { pub(crate) rg_index: usize, + /// Selection made before row-filter evaluation, in this row group's coordinates. + pub(crate) row_selection: Option, /// `true` when static pruning proved every row of this RG satisfies the /// predicate, so the per-row `RowFilter` can be skipped as a no-op. pub(crate) fully_matched: bool, @@ -275,6 +279,64 @@ impl RowGroupPruner { } } +/// Preserve the access plan's page pruning when fetching output and predicate +/// columns together. Without a selection or offset index, fetch complete chunks. +fn column_ranges( + metadata: &ParquetMetaData, + entry: &RgPlanEntry, + projection: &ProjectionMask, +) -> Option>> { + let offset_index = metadata + .offset_index() + .and_then(|index| index.get(entry.rg_index)); + let mut ranges = Vec::new(); + for (i, column) in metadata + .row_group(entry.rg_index) + .columns() + .iter() + .enumerate() + .filter(|(i, _)| projection.leaf_included(*i)) + { + let (start, len) = column.byte_range(); + let end = start.checked_add(len)?; + if let Some((selection, index)) = entry.row_selection.as_ref().zip(offset_index) { + if !selection.selects_any() { + continue; + } + let pages = &index.get(i)?.page_locations; + // The prefix before the first data page contains the dictionary. + if let Some(first) = pages.first() { + let first_offset = u64::try_from(first.offset).ok()?; + if first_offset < start || first_offset > end { + return None; + } + if first_offset > start { + ranges.push(start..first_offset); + } + } + ranges.extend(selection.scan_ranges(pages)); + } else { + ranges.push(start..end); + } + } + Some(ranges) +} + +/// Make adjacent pages usable for decoder requests that span a whole chunk, +/// without fetching across gaps left by page pruning. +fn merge_ranges(mut ranges: Vec>) -> Vec> { + ranges.sort_unstable_by_key(|range| range.start); + ranges.dedup_by(|next, previous| { + if next.start <= previous.end { + previous.end = previous.end.max(next.end); + true + } else { + false + } + }); + ranges +} + /// State for a stream that decodes a single Parquet file using a push-based decoder. /// /// The [`transition`](Self::transition) method drives the decoder in a loop: it requests @@ -286,6 +348,10 @@ pub(crate) struct PushDecoderStreamState { pub(crate) active_reader: Option, pub(crate) rg_plan: VecDeque, pub(crate) reader: Box, + pub(crate) progressive_io: bool, + pub(crate) fetch_projection: ProjectionMask, + pub(crate) upfront_row_group: Option, + pub(crate) parquet_metadata: Arc, /// Per-file projection: the mask installed on every decoder and the /// per-batch transform applied by [`Self::project_batch`]. pub(crate) decoder_projection: DecoderProjection, @@ -360,6 +426,12 @@ pub(crate) struct RowFilterContext { } impl RowFilterContext { + pub(crate) fn extend_projection(&self, projection: &mut ProjectionMask) { + for candidate in self.prebuilt.as_slice() { + projection.union(&candidate.projection_mask); + } + } + /// Precompute the candidate list from the raw predicate + file schema + /// metadata. Returns `None` when the predicate has no push-downable /// conjuncts (mirrors the file-open path behaviour). @@ -419,10 +491,10 @@ impl PushDecoderStreamState { /// Advances the decoder state machine until the next [`RecordBatch`] is /// produced, the file is fully consumed, or an error occurs. /// - /// On each iteration the decoder is polled via [`ParquetPushDecoder::try_decode`]: + /// At a row-group boundary the decoder is polled via [`ParquetPushDecoder::try_next_reader`]: /// - [`NeedsData`](DecodeResult::NeedsData) – the requested byte ranges are /// fetched from the [`AsyncFileReader`] and fed back into the decoder. - /// - [`Data`](DecodeResult::Data) – a decoded batch is projected and returned. + /// - [`Data`](DecodeResult::Data) – a reader is retained for subsequent batch decoding. /// - [`Finished`](DecodeResult::Finished) – signals end-of-stream (`None`). /// /// Takes `self` by value (rather than `&mut self`) so the generated future @@ -500,7 +572,65 @@ impl PushDecoderStreamState { // Step 3: drive the decoder. let decoder = self.decoder.as_mut().expect("decoder present"); match decoder.try_next_reader() { - Ok(DecodeResult::NeedsData(ranges)) => { + Ok(DecodeResult::NeedsData(mut ranges)) => { + if !self.progressive_io { + // The decoder can skip fully filtered groups internally. + // Locate its current group from the demand range rather + // than assuming the previous group yielded a reader. + let row_group = self + .rg_plan + .iter() + .find(|entry| { + self.parquet_metadata + .row_group(entry.rg_index) + .columns() + .iter() + .any(|column| { + let (start, len) = column.byte_range(); + ranges.iter().any(|r| { + r.start >= start + && r.end <= start.saturating_add(len) + }) + }) + }) + .map(|entry| entry.rg_index); + if let Some(row_group) = row_group { + if let Err(e) = Self::advance_rg_plan_to( + &mut self.rg_plan, + row_group, + &mut self.byte_progress, + ) { + return Some((Err(e), self)); + } + if self.upfront_row_group != Some(row_group) { + decoder.clear_all_ranges(); + let entry = + self.rg_plan.front().expect("current row group"); + let Some(mut upfront_ranges) = column_ranges( + &self.parquet_metadata, + entry, + if entry.fully_matched { + self.decoder_projection.projection_mask() + } else { + &self.fetch_projection + }, + ) else { + return Some(( + internal_err!( + "Invalid Parquet column or page byte range" + ), + self, + )); + }; + // Predicate caching may expand its selection to + // batch boundaries. Keep the decoder's request + // as well as the page-pruned output ranges. + upfront_ranges.extend(ranges); + ranges = merge_ranges(upfront_ranges); + self.upfront_row_group = Some(row_group); + } + } + } let data = self .reader .get_byte_ranges(ranges.clone()) @@ -534,6 +664,11 @@ impl PushDecoderStreamState { self.byte_progress.credit(entry.bytes); } self.active_reader = Some(reader); + // The extracted reader now owns required bytes. Release any + // unused speculation (e.g. pages removed by a row filter). + if !self.progressive_io { + decoder.clear_all_ranges(); + } } Ok(DecodeResult::Finished) => return None, Err(e) => { @@ -717,7 +852,15 @@ mod tests { /// column statistics are disjoint: RG0 → 0..1000, RG1 → 1000..2000, /// RG2 → 2000..3000. Returns (metadata, schema). fn build_three_rg_file() -> (Arc, SchemaRef) { - let schema = Arc::new(Schema::new(vec![Field::new("v", DataType::Int64, false)])); + let (_, metadata, schema) = build_three_rg_file_data(); + (metadata, schema) + } + + fn build_three_rg_file_data() -> (Bytes, Arc, SchemaRef) { + let schema = Arc::new(Schema::new(vec![ + Field::new("v", DataType::Int64, false), + Field::new("w", DataType::Int64, false), + ])); let mut buf = Vec::new(); let props = WriterProperties::builder() .set_max_row_group_row_count(Some(1000)) @@ -729,7 +872,10 @@ mod tests { let vals: Vec = (base..base + 1000).collect(); let batch = RecordBatch::try_new( Arc::clone(&schema), - vec![Arc::new(Int64Array::from(vals))], + vec![ + Arc::new(Int64Array::from(vals.clone())), + Arc::new(Int64Array::from(vals)), + ], ) .unwrap(); writer.write(&batch).unwrap(); @@ -747,12 +893,224 @@ mod tests { reason = "we want a single range covering the whole file" )] let ranges = vec![0..len]; - md.push_ranges(ranges, vec![file]).unwrap(); + md.push_ranges(ranges, vec![file.clone()]).unwrap(); let DecodeResult::Data(meta) = md.try_decode().unwrap() else { panic!("decoding metadata"); }; assert_eq!(meta.num_row_groups(), 3, "test fixture must have 3 RGs"); - (Arc::new(meta), schema) + (file, Arc::new(meta), schema) + } + + #[derive(Debug, Default)] + struct ReadControl { + calls: std::sync::atomic::AtomicUsize, + bytes: std::sync::atomic::AtomicUsize, + } + + #[derive(Debug, Clone)] + struct TestReader { + data: Bytes, + metadata: Arc, + control: Arc, + } + + impl AsyncFileReader for TestReader { + fn get_bytes( + &mut self, + range: Range, + ) -> futures::future::BoxFuture<'_, parquet::errors::Result> { + use futures::FutureExt; + async move { Ok(self.data.slice(range.start as usize..range.end as usize)) } + .boxed() + } + + fn get_byte_ranges( + &mut self, + ranges: Vec>, + ) -> futures::future::BoxFuture<'_, parquet::errors::Result>> { + use futures::FutureExt; + use std::sync::atomic::Ordering; + async move { + self.control.bytes.fetch_add( + ranges.iter().map(|r| (r.end - r.start) as usize).sum(), + Ordering::SeqCst, + ); + self.control.calls.fetch_add(1, Ordering::SeqCst); + Ok(ranges + .into_iter() + .map(|range| { + self.data.slice(range.start as usize..range.end as usize) + }) + .collect()) + } + .boxed() + } + + fn get_metadata<'a>( + &'a mut self, + _options: Option<&'a parquet::arrow::arrow_reader::ArrowReaderOptions>, + ) -> futures::future::BoxFuture<'a, parquet::errors::Result>> + { + use futures::FutureExt; + async move { Ok(Arc::clone(&self.metadata)) }.boxed() + } + } + + impl crate::ParquetFileReaderFactory for TestReader { + fn create_reader( + &self, + _partition: usize, + _file: datafusion_datasource::PartitionedFile, + _hint: Option, + _metrics: &ExecutionPlanMetricsSet, + ) -> Result> { + Ok(Box::new(self.clone())) + } + } + + fn io_test_stream( + control: Arc, + limit: Option, + predicate: Option>, + progressive: bool, + ) -> datafusion_execution::SendableRecordBatchStream { + use datafusion_datasource::file_scan_config::FileScanConfigBuilder; + use datafusion_datasource::source::DataSourceExec; + use datafusion_datasource::{PartitionedFile, file_groups::FileGroup}; + use datafusion_execution::{ + TaskContext, config::SessionConfig, object_store::ObjectStoreUrl, + }; + use datafusion_physical_plan::ExecutionPlan; + let (data, metadata, schema) = build_three_rg_file_data(); + let file = PartitionedFile::new("prefetch.parquet", data.len() as u64); + let mut source = crate::source::ParquetSource::new(schema) + .with_progressive_io(progressive) + // Page selections disable runtime pruning (#24355). Page-pruned + // I/O is covered by the session-configuration integration test. + .with_enable_page_index(false) + .with_pushdown_filters(true) + .with_parquet_file_reader_factory(Arc::new(TestReader { + data, + metadata, + control, + })); + if let Some(predicate) = predicate { + source = source.with_predicate(predicate); + } + let config = FileScanConfigBuilder::new( + ObjectStoreUrl::local_filesystem(), + Arc::new(source), + ) + .with_file_group(FileGroup::new(vec![file])) + .with_limit(limit) + .with_projection_indices(Some(vec![0])) + .unwrap() + .build(); + let task = TaskContext::default() + .with_session_config(SessionConfig::new().with_batch_size(100)); + DataSourceExec::new(Arc::new(config)) + .execute(0, Arc::new(task)) + .unwrap() + } + + #[tokio::test] + async fn upfront_reads_include_predicate_only_columns_and_skip_empty_groups() { + use std::sync::atomic::Ordering; + // A predicate-only column empties the first, middle, or last group + // without row-group statistics pruning. Compare every returned value. + for empty_group in 0..3i64 { + for progressive in [false, true] { + let predicate = Arc::new(BinaryExpr::new( + Arc::new(BinaryExpr::new( + Arc::new(Column::new("w", 1)), + Operator::Divide, + lit(1000i64), + )), + Operator::NotEq, + lit(empty_group), + )) as Arc; + + let control = Arc::new(ReadControl::default()); + let mut stream = io_test_stream( + Arc::clone(&control), + None, + Some(predicate), + progressive, + ); + let mut values = Vec::new(); + while let Some(batch) = stream.next().await { + let batch = batch.unwrap(); + assert_eq!(batch.num_columns(), 1); + values.extend_from_slice( + batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap() + .values(), + ); + } + assert_eq!( + values, + (0..3000) + .filter(|v| v / 1000 != empty_group) + .collect::>() + ); + let expected_calls = if !progressive { 3 } else { 5 }; + assert_eq!(control.calls.load(Ordering::SeqCst), expected_calls); + } + } + } + + #[tokio::test] + async fn io_policies_preserve_fully_matched_filter_suppression() { + use std::sync::atomic::Ordering; + + let (_, metadata, _) = build_three_rg_file_data(); + // Exercise both directions: fully matched -> filtered, and the reverse. + // The predicate column is absent from the output in both cases. + for (operator, expected, row_groups) in [ + (Operator::Lt, (0..1500).collect::>(), [0, 1]), + (Operator::GtEq, (1500..3000).collect::>(), [1, 2]), + ] { + let expected_bytes = row_groups + .iter() + .map(|&rg| metadata.row_group(rg).column(0).byte_range().1) + .sum::() + + metadata.row_group(1).column(1).byte_range().1; + for progressive in [true, false] { + let control = Arc::new(ReadControl::default()); + let predicate = Arc::new(BinaryExpr::new( + Arc::new(Column::new("w", 1)), + operator, + lit(1500i64), + )); + let mut stream = io_test_stream( + Arc::clone(&control), + None, + Some(predicate), + progressive, + ); + let mut values = Vec::new(); + while let Some(batch) = stream.next().await { + values.extend_from_slice( + batch + .unwrap() + .column(0) + .as_any() + .downcast_ref::() + .unwrap() + .values(), + ); + } + assert_eq!(values, expected); + assert_eq!( + control.bytes.load(Ordering::SeqCst) as u64, + expected_bytes, + "predicate-only chunks in fully matched groups must stay unread" + ); + } + } } /// Create a fresh `(creation_errors, evaluation_errors)` counter pair @@ -879,6 +1237,7 @@ mod tests { .into_iter() .map(|rg_index| RgPlanEntry { rg_index, + row_selection: None, fully_matched: false, bytes: 100 * (rg_index as u64 + 1), }) diff --git a/datafusion/datasource-parquet/src/row_filter.rs b/datafusion/datasource-parquet/src/row_filter.rs index 833764ba0809f..d788e2fc35b63 100644 --- a/datafusion/datasource-parquet/src/row_filter.rs +++ b/datafusion/datasource-parquet/src/row_filter.rs @@ -460,7 +460,7 @@ pub(crate) struct PrebuiltRowFilterCandidate { physical_expr: Arc, /// Projection mask over the parquet leaf columns needed to evaluate this /// predicate. - projection_mask: ProjectionMask, + pub(crate) projection_mask: ProjectionMask, /// Precomputed sum-of-compressed-bytes for the referenced columns across /// all row groups in the file. Used to sort predicates when /// `reorder_predicates` is enabled. Stable across row groups within a diff --git a/datafusion/datasource-parquet/src/source.rs b/datafusion/datasource-parquet/src/source.rs index 4872db9fd3329..f045bcda53c4b 100644 --- a/datafusion/datasource-parquet/src/source.rs +++ b/datafusion/datasource-parquet/src/source.rs @@ -347,6 +347,17 @@ impl ParquetSource { } } + /// Fetch pages progressively as decoding and row filtering require + /// them (the default). When false, the first demand read for each row group + /// fetches the output and predicate pages selected at file open together. This reduces + /// dependent I/O rounds but can read pages that filtering would skip. + /// Controls demand reads independently of next-row-group prefetch. + /// Also configurable as `datafusion.execution.parquet.progressive_io`. + pub fn with_progressive_io(mut self, progressive_io: bool) -> Self { + self.table_parquet_options.global.progressive_io = progressive_io; + self + } + /// Set the `TableParquetOptions` for this ParquetSource. pub fn with_table_parquet_options( mut self, @@ -634,6 +645,7 @@ impl FileSource for ParquetSource { Ok(Box::new(ParquetMorselizer { partition_index: partition, + progressive_io: self.table_parquet_options.global.progressive_io, projection: self.projection.clone(), batch_size: self .batch_size diff --git a/datafusion/proto-common/proto/datafusion_common.proto b/datafusion/proto-common/proto/datafusion_common.proto index 712212f6b6ae5..f6c3fbd848308 100644 --- a/datafusion/proto-common/proto/datafusion_common.proto +++ b/datafusion/proto-common/proto/datafusion_common.proto @@ -559,6 +559,7 @@ message ParquetOptions { bool pruning = 2; // default = true bool skip_metadata = 3; // default = true bool pushdown_filters = 5; // default = false + optional bool progressive_io = 39; // absent defaults to true for older plans bool reorder_filters = 6; // default = false bool force_filter_selections = 34; // default = false uint64 data_pagesize_limit = 7; // default = 1024 * 1024 @@ -718,4 +719,4 @@ enum MetricCategory { message ExplainAnalyzeCategoriesNode { bool all = 1; repeated MetricCategory only = 2; -} \ No newline at end of file +} diff --git a/datafusion/proto-common/src/from_proto/mod.rs b/datafusion/proto-common/src/from_proto/mod.rs index 92506ad92bad0..c50a7007505a2 100644 --- a/datafusion/proto-common/src/from_proto/mod.rs +++ b/datafusion/proto-common/src/from_proto/mod.rs @@ -1083,6 +1083,7 @@ impl TryFrom<&protobuf::ParquetOptions> for ParquetOptions { }) .transpose()?, pushdown_filters: value.pushdown_filters, + progressive_io: value.progressive_io.unwrap_or(true), reorder_filters: value.reorder_filters, force_filter_selections: value.force_filter_selections, data_pagesize_limit: to_usize( @@ -1479,6 +1480,21 @@ mod tests { assert_eq!(opts, recovered); } + #[test] + fn test_parquet_progressive_io_round_trip_and_older_plan_default() { + for progressive_io in [false, true] { + let opts = ParquetOptions { + progressive_io, + ..Default::default() + }; + assert_eq!(parquet_options_proto_round_trip(opts.clone()), opts); + let mut proto: crate::protobuf_common::ParquetOptions = + (&opts).try_into().unwrap(); + proto.progressive_io = None; + assert!(ParquetOptions::try_from(&proto).unwrap().progressive_io); + } + } + #[test] fn test_parquet_options_coerce_int96_tz_unset_round_trip() { let opts = ParquetOptions::default(); diff --git a/datafusion/proto-common/src/generated/pbjson.rs b/datafusion/proto-common/src/generated/pbjson.rs index 2e94368bfd01e..4e60fc729d56a 100644 --- a/datafusion/proto-common/src/generated/pbjson.rs +++ b/datafusion/proto-common/src/generated/pbjson.rs @@ -6381,6 +6381,9 @@ impl serde::Serialize for ParquetOptions { if self.pushdown_filters { len += 1; } + if self.progressive_io.is_some() { + len += 1; + } if self.reorder_filters { len += 1; } @@ -6490,6 +6493,9 @@ impl serde::Serialize for ParquetOptions { if self.pushdown_filters { struct_ser.serialize_field("pushdownFilters", &self.pushdown_filters)?; } + if let Some(v) = self.progressive_io.as_ref() { + struct_ser.serialize_field("progressiveIo", v)?; + } if self.reorder_filters { struct_ser.serialize_field("reorderFilters", &self.reorder_filters)?; } @@ -6683,6 +6689,8 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { "skipMetadata", "pushdown_filters", "pushdownFilters", + "progressive_io", + "progressiveIo", "reorder_filters", "reorderFilters", "force_filter_selections", @@ -6753,6 +6761,7 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { Pruning, SkipMetadata, PushdownFilters, + ProgressiveIo, ReorderFilters, ForceFilterSelections, DataPagesizeLimit, @@ -6810,6 +6819,7 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { "pruning" => Ok(GeneratedField::Pruning), "skipMetadata" | "skip_metadata" => Ok(GeneratedField::SkipMetadata), "pushdownFilters" | "pushdown_filters" => Ok(GeneratedField::PushdownFilters), + "progressiveIo" | "progressive_io" => Ok(GeneratedField::ProgressiveIo), "reorderFilters" | "reorder_filters" => Ok(GeneratedField::ReorderFilters), "forceFilterSelections" | "force_filter_selections" => Ok(GeneratedField::ForceFilterSelections), "dataPagesizeLimit" | "data_pagesize_limit" => Ok(GeneratedField::DataPagesizeLimit), @@ -6865,6 +6875,7 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { let mut pruning__ = None; let mut skip_metadata__ = None; let mut pushdown_filters__ = None; + let mut progressive_io__ = None; let mut reorder_filters__ = None; let mut force_filter_selections__ = None; let mut data_pagesize_limit__ = None; @@ -6923,6 +6934,12 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { } pushdown_filters__ = Some(map_.next_value()?); } + GeneratedField::ProgressiveIo => { + if progressive_io__.is_some() { + return Err(serde::de::Error::duplicate_field("progressiveIo")); + } + progressive_io__ = map_.next_value()?; + } GeneratedField::ReorderFilters => { if reorder_filters__.is_some() { return Err(serde::de::Error::duplicate_field("reorderFilters")); @@ -7138,6 +7155,7 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { pruning: pruning__.unwrap_or_default(), skip_metadata: skip_metadata__.unwrap_or_default(), pushdown_filters: pushdown_filters__.unwrap_or_default(), + progressive_io: progressive_io__, reorder_filters: reorder_filters__.unwrap_or_default(), force_filter_selections: force_filter_selections__.unwrap_or_default(), data_pagesize_limit: data_pagesize_limit__.unwrap_or_default(), diff --git a/datafusion/proto-common/src/generated/prost.rs b/datafusion/proto-common/src/generated/prost.rs index c0e79aec6d873..42c18b3f35120 100644 --- a/datafusion/proto-common/src/generated/prost.rs +++ b/datafusion/proto-common/src/generated/prost.rs @@ -817,6 +817,9 @@ pub struct ParquetOptions { /// default = false #[prost(bool, tag = "5")] pub pushdown_filters: bool, + /// absent defaults to true for older plans + #[prost(bool, optional, tag = "39")] + pub progressive_io: ::core::option::Option, /// default = false #[prost(bool, tag = "6")] pub reorder_filters: bool, diff --git a/datafusion/proto-common/src/to_proto/mod.rs b/datafusion/proto-common/src/to_proto/mod.rs index 7f95e03f41db4..be249885718bd 100644 --- a/datafusion/proto-common/src/to_proto/mod.rs +++ b/datafusion/proto-common/src/to_proto/mod.rs @@ -909,6 +909,7 @@ impl TryFrom<&ParquetOptions> for protobuf::ParquetOptions { skip_metadata: value.skip_metadata, metadata_size_hint_opt: value.metadata_size_hint.map(|v| protobuf::parquet_options::MetadataSizeHintOpt::MetadataSizeHint(v as u64)), pushdown_filters: value.pushdown_filters, + progressive_io: Some(value.progressive_io), reorder_filters: value.reorder_filters, force_filter_selections: value.force_filter_selections, data_pagesize_limit: value.data_pagesize_limit as u64, diff --git a/datafusion/proto-models/src/from_proto.rs b/datafusion/proto-models/src/from_proto.rs index b2340a09678ac..b8c8665ed0115 100644 --- a/datafusion/proto-models/src/from_proto.rs +++ b/datafusion/proto-models/src/from_proto.rs @@ -372,6 +372,7 @@ impl TryFrom<&ParquetOptionsProto> for ParquetOptions { }) .transpose()?, pushdown_filters: proto.pushdown_filters, + progressive_io: proto.progressive_io.unwrap_or(true), reorder_filters: proto.reorder_filters, force_filter_selections: proto.force_filter_selections, data_pagesize_limit: to_usize( diff --git a/datafusion/proto-models/src/generated/datafusion_proto_common.rs b/datafusion/proto-models/src/generated/datafusion_proto_common.rs index c0e79aec6d873..42c18b3f35120 100644 --- a/datafusion/proto-models/src/generated/datafusion_proto_common.rs +++ b/datafusion/proto-models/src/generated/datafusion_proto_common.rs @@ -817,6 +817,9 @@ pub struct ParquetOptions { /// default = false #[prost(bool, tag = "5")] pub pushdown_filters: bool, + /// absent defaults to true for older plans + #[prost(bool, optional, tag = "39")] + pub progressive_io: ::core::option::Option, /// default = false #[prost(bool, tag = "6")] pub reorder_filters: bool, diff --git a/datafusion/proto/tests/cases/plans/sources.rs b/datafusion/proto/tests/cases/plans/sources.rs index 7aff89eb3fdf7..053fb2d0622c5 100644 --- a/datafusion/proto/tests/cases/plans/sources.rs +++ b/datafusion/proto/tests/cases/plans/sources.rs @@ -87,6 +87,7 @@ fn roundtrip_parquet_exec_with_pruning_predicate() -> Result<()> { let mut options = TableParquetOptions::new(); options.global.pushdown_filters = true; + options.global.progressive_io = false; let file_source = Arc::new( ParquetSource::new(Arc::clone(&file_schema)) diff --git a/datafusion/proto/tests/cases/roundtrip_logical_plan.rs b/datafusion/proto/tests/cases/roundtrip_logical_plan.rs index b4c121ef30714..ca9063bae50ac 100644 --- a/datafusion/proto/tests/cases/roundtrip_logical_plan.rs +++ b/datafusion/proto/tests/cases/roundtrip_logical_plan.rs @@ -1091,6 +1091,7 @@ async fn roundtrip_logical_plan_copy_to_parquet() -> Result<()> { .clone_from(&key_value_metadata); parquet_format.global.allow_single_file_parallelism = false; + parquet_format.global.progressive_io = false; parquet_format.global.created_by = "test".to_string(); let file_type = format_as_file_type(Arc::new( @@ -1133,6 +1134,7 @@ async fn roundtrip_logical_plan_copy_to_parquet() -> Result<()> { let parquet_config = parquet_factory.options.as_ref().unwrap(); assert_eq!(parquet_config.key_value_metadata, key_value_metadata); assert!(!parquet_config.global.allow_single_file_parallelism); + assert!(!parquet_config.global.progressive_io); assert_eq!(parquet_config.global.created_by, "test".to_string()); } _ => panic!(), diff --git a/datafusion/pruning/src/pruning_predicate.rs b/datafusion/pruning/src/pruning_predicate.rs index b49b72058e0cd..9539a8892702e 100644 --- a/datafusion/pruning/src/pruning_predicate.rs +++ b/datafusion/pruning/src/pruning_predicate.rs @@ -525,6 +525,13 @@ impl<'a> PruningPredicateBuilder<'a> { ) })?; + // A changing filter can tighten after this snapshot. Its current inverse + // cannot prove that every row will still match when the data is read. + let stable_predicate = !matches!( + phys_expr::DynamicFilterTracking::classify(&predicate), + phys_expr::DynamicFilterTracking::Watching(_) + ); + // Get a (simpler) snapshot of the physical expr here to use with `PruningPredicate`. // In particular this unravels any `DynamicFilterPhysicalExpr`s by snapshotting them // so that PruningPredicate can work with a static expression. @@ -567,7 +574,8 @@ impl<'a> PruningPredicateBuilder<'a> { orig_expr: predicate, literal_guarantees, max_in_list_size: self.max_in_list_size, - can_be_inverted_for_full_match: !properties.has_filter_semantics_only, + can_be_inverted_for_full_match: stable_predicate + && !properties.has_filter_semantics_only, }) } } @@ -746,7 +754,8 @@ impl PruningPredicate { } /// Returns whether pruning the logical inverse can safely prove that every - /// row in a container satisfies the original predicate. + /// row in a container satisfies the original predicate. Snapshots of dynamic + /// filters that can still change cannot provide this guarantee. pub fn can_be_inverted_for_full_match(&self) -> bool { self.can_be_inverted_for_full_match } @@ -3398,6 +3407,29 @@ mod tests { ); } + #[test] + fn full_match_requires_a_stable_dynamic_filter() -> Result<()> { + let schema = + Arc::new(Schema::new(vec![Field::new("c1", DataType::Int32, false)])); + let expression = logical2physical(&col("c1").gt(lit(5)), &schema); + let dynamic = Arc::new(DynamicFilterPhysicalExpr::new( + vec![Arc::new(phys_expr::Column::new("c1", 0))], + Arc::clone(&expression), + )); + let build = |expression| { + PruningPredicateBuilder::new() + .with_file_schema(Arc::clone(&schema)) + .try_build(expression) + }; + assert!(build(expression)?.can_be_inverted_for_full_match()); + let snapshot = build(Arc::clone(&dynamic) as _)?; + assert!(!snapshot.can_be_inverted_for_full_match()); + dynamic.mark_complete(); + assert!(build(dynamic)?.can_be_inverted_for_full_match()); + assert!(!snapshot.can_be_inverted_for_full_match()); + Ok(()) + } + /// Integration test demonstrating that a dynamic filter with replaced children as literals will be snapshotted, simplified and then pruned correctly. #[test] fn row_group_predicate_dynamic_filter_with_literals() { diff --git a/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt b/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt index 514ae81095a49..8e6ac297eb665 100644 --- a/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt +++ b/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt @@ -104,7 +104,7 @@ Plan with Metrics 03)----ProjectionExec: expr=[id@0 as id, value@1 as v, value@1 + id@0 as name], metrics=[output_rows=10, ] 04)------FilterExec: value@1 > 3, metrics=[output_rows=10, , selectivity=100% (10/10)] 05)--------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1, metrics=[output_rows=10, ] -06)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/test_data.parquet]]}, projection=[id, value], file_type=parquet, predicate=value@1 > 3 AND DynamicFilter [ value@1 IS NULL OR value@1 > 800 ], dynamic_rg_pruning=eligible, pruning_predicate=value_null_count@1 != row_count@2 AND value_max@0 > 3 AND (value_null_count@1 > 0 OR value_null_count@1 != row_count@2 AND value_max@0 > 800), required_guarantees=[], metrics=[output_rows=10, elapsed_compute=, output_bytes=80.0 B, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched -> 1 fully matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_pages_skipped_by_fully_matched=1, limit_pruned_row_groups=0 total → 0 matched, bytes_processed=1147.0 B, bytes_scanned=210.0 B, page_index_load_skipped=1, row_groups_pruned_dynamic_filter=0, metadata_load_time=, scan_efficiency_ratio=18.31% (210/1.15 K)] +06)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/test_data.parquet]]}, projection=[id, value], file_type=parquet, predicate=value@1 > 3 AND DynamicFilter [ value@1 IS NULL OR value@1 > 800 ], dynamic_rg_pruning=eligible, pruning_predicate=value_null_count@1 != row_count@2 AND value_max@0 > 3 AND (value_null_count@1 > 0 OR value_null_count@1 != row_count@2 AND value_max@0 > 800), required_guarantees=[], metrics=[output_rows=10, elapsed_compute=, output_bytes=80.0 B, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=1 total → 1 matched, limit_pruned_row_groups=0 total → 0 matched, bytes_processed=1147.0 B, bytes_scanned=210.0 B, row_groups_pruned_dynamic_filter=0, metadata_load_time=, scan_efficiency_ratio=18.31% (210/1.15 K)] statement ok set datafusion.explain.analyze_level = dev; diff --git a/datafusion/sqllogictest/test_files/explain_analyze.slt b/datafusion/sqllogictest/test_files/explain_analyze.slt index 81b274d6c1377..cdf7bd9c54f84 100644 --- a/datafusion/sqllogictest/test_files/explain_analyze.slt +++ b/datafusion/sqllogictest/test_files/explain_analyze.slt @@ -247,7 +247,7 @@ explain analyze select * from cat_tracking where species > 'M' AND s >= 50 order ---- Plan with Metrics 01)SortExec: TopK(fetch=3), expr=[species@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[species@0 < Nlpine Sheep], metrics=[output_rows=3] -02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND species_min@5 < Nlpine Sheep, required_guarantees=[], metrics=[output_rows=3, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched -> 1 fully matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=2 total → 2 matched, page_index_pages_skipped_by_fully_matched=1, limit_pruned_row_groups=0 total → 0 matched, scan_efficiency_ratio=21.75% (485/2.23 K)] +02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND species_min@5 < Nlpine Sheep, required_guarantees=[], metrics=[output_rows=3, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=3 total → 3 matched, limit_pruned_row_groups=0 total → 0 matched, scan_efficiency_ratio=21.75% (485/2.23 K)] statement ok reset datafusion.explain.analyze_categories; @@ -262,7 +262,7 @@ explain analyze select * from cat_tracking where species > 'M' AND s >= 50 order ---- Plan with Metrics 01)SortExec: TopK(fetch=3), expr=[species@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[species@0 < Nlpine Sheep], metrics=[output_rows=3, output_bytes=] -02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND species_min@5 < Nlpine Sheep, required_guarantees=[], metrics=[output_rows=3, output_bytes=, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched -> 1 fully matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=2 total → 2 matched, page_index_pages_skipped_by_fully_matched=1, limit_pruned_row_groups=0 total → 0 matched, bytes_processed=, bytes_scanned=, scan_efficiency_ratio=] +02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND species_min@5 < Nlpine Sheep, required_guarantees=[], metrics=[output_rows=3, output_bytes=, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=3 total → 3 matched, limit_pruned_row_groups=0 total → 0 matched, bytes_processed=, bytes_scanned=, scan_efficiency_ratio=] statement ok reset datafusion.explain.analyze_categories; @@ -277,7 +277,7 @@ explain analyze select * from cat_tracking where species > 'M' AND s >= 50 order ---- Plan with Metrics 01)SortExec: TopK(fetch=3), expr=[species@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[species@0 < Nlpine Sheep], metrics=[output_rows=3, output_bytes=] -02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND species_min@5 < Nlpine Sheep, required_guarantees=[], metrics=[output_rows=3, output_bytes=, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched -> 1 fully matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=2 total → 2 matched, page_index_pages_skipped_by_fully_matched=1, limit_pruned_row_groups=0 total → 0 matched, bytes_processed=, bytes_scanned=, scan_efficiency_ratio=] +02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND species_min@5 < Nlpine Sheep, required_guarantees=[], metrics=[output_rows=3, output_bytes=, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=3 total → 3 matched, limit_pruned_row_groups=0 total → 0 matched, bytes_processed=, bytes_scanned=, scan_efficiency_ratio=] statement ok reset datafusion.explain.analyze_categories; @@ -559,7 +559,7 @@ EXPLAIN (ANALYZE, METRICS 'rows', LEVEL summary) select * from cat_tracking wher ---- Plan with Metrics 01)SortExec: TopK(fetch=3), expr=[species@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[species@0 < Nlpine Sheep], metrics=[output_rows=3] -02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND species_min@5 < Nlpine Sheep, required_guarantees=[], metrics=[output_rows=3, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched -> 1 fully matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=2 total → 2 matched, page_index_pages_skipped_by_fully_matched=1, limit_pruned_row_groups=0 total → 0 matched, scan_efficiency_ratio=21.75% (485/2.23 K)] +02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND species_min@5 < Nlpine Sheep, required_guarantees=[], metrics=[output_rows=3, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=3 total → 3 matched, limit_pruned_row_groups=0 total → 0 matched, scan_efficiency_ratio=21.75% (485/2.23 K)] # ---- Quoted-string METRICS with multiple categories ---- @@ -568,7 +568,7 @@ EXPLAIN (ANALYZE, METRICS 'rows,bytes', LEVEL summary) select * from cat_trackin ---- Plan with Metrics 01)SortExec: TopK(fetch=3), expr=[species@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[species@0 < Nlpine Sheep], metrics=[output_rows=3, output_bytes=] -02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND species_min@5 < Nlpine Sheep, required_guarantees=[], metrics=[output_rows=3, output_bytes=, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched -> 1 fully matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=2 total → 2 matched, page_index_pages_skipped_by_fully_matched=1, limit_pruned_row_groups=0 total → 0 matched, bytes_processed=, bytes_scanned=, scan_efficiency_ratio=] +02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND species_min@5 < Nlpine Sheep, required_guarantees=[], metrics=[output_rows=3, output_bytes=, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=3 total → 3 matched, limit_pruned_row_groups=0 total → 0 matched, bytes_processed=, bytes_scanned=, scan_efficiency_ratio=] # ---- (METRICS 'timing', LEVEL summary) — timing metrics only ---- @@ -588,7 +588,7 @@ EXPLAIN (ANALYZE, METRICS 'rows,bytes', TIMING off, LEVEL summary) select * from ---- Plan with Metrics 01)SortExec: TopK(fetch=3), expr=[species@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[species@0 < Nlpine Sheep], metrics=[output_rows=3, output_bytes=] -02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND species_min@5 < Nlpine Sheep, required_guarantees=[], metrics=[output_rows=3, output_bytes=, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched -> 1 fully matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=2 total → 2 matched, page_index_pages_skipped_by_fully_matched=1, limit_pruned_row_groups=0 total → 0 matched, bytes_processed=, bytes_scanned=, scan_efficiency_ratio=] +02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND species_min@5 < Nlpine Sheep, required_guarantees=[], metrics=[output_rows=3, output_bytes=, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=3 total → 3 matched, limit_pruned_row_groups=0 total → 0 matched, bytes_processed=, bytes_scanned=, scan_efficiency_ratio=] # ---- TIMING sugar: `METRICS 'rows', TIMING on` ↔ rows + timing ---- @@ -597,7 +597,7 @@ EXPLAIN (ANALYZE, METRICS 'rows', TIMING on, LEVEL summary) select * from cat_tr ---- Plan with Metrics 01)SortExec: TopK(fetch=3), expr=[species@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[species@0 < Nlpine Sheep], metrics=[output_rows=3, elapsed_compute=] -02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND species_min@5 < Nlpine Sheep, required_guarantees=[], metrics=[output_rows=3, elapsed_compute=, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched -> 1 fully matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=2 total → 2 matched, page_index_pages_skipped_by_fully_matched=1, limit_pruned_row_groups=0 total → 0 matched, metadata_load_time=, scan_efficiency_ratio=21.75% (485/2.23 K)] +02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND species_min@5 < Nlpine Sheep, required_guarantees=[], metrics=[output_rows=3, elapsed_compute=, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=3 total → 3 matched, limit_pruned_row_groups=0 total → 0 matched, metadata_load_time=, scan_efficiency_ratio=21.75% (485/2.23 K)] # ---- SUMMARY sugar: `SUMMARY on` ↔ `LEVEL summary` ---- # Equivalent to METRICS 'rows', LEVEL summary above. @@ -607,7 +607,7 @@ EXPLAIN (ANALYZE, METRICS 'rows', SUMMARY on) select * from cat_tracking where s ---- Plan with Metrics 01)SortExec: TopK(fetch=3), expr=[species@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[species@0 < Nlpine Sheep], metrics=[output_rows=3] -02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND species_min@5 < Nlpine Sheep, required_guarantees=[], metrics=[output_rows=3, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched -> 1 fully matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=2 total → 2 matched, page_index_pages_skipped_by_fully_matched=1, limit_pruned_row_groups=0 total → 0 matched, scan_efficiency_ratio=21.75% (485/2.23 K)] +02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND species_min@5 < Nlpine Sheep, required_guarantees=[], metrics=[output_rows=3, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=3 total → 3 matched, limit_pruned_row_groups=0 total → 0 matched, scan_efficiency_ratio=21.75% (485/2.23 K)] # ---- Statement option overrides session config ---- # Session says 'timing' but statement-level `METRICS 'rows'` wins. @@ -620,7 +620,7 @@ EXPLAIN (ANALYZE, METRICS 'rows', LEVEL summary) select * from cat_tracking wher ---- Plan with Metrics 01)SortExec: TopK(fetch=3), expr=[species@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[species@0 < Nlpine Sheep], metrics=[output_rows=3] -02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND species_min@5 < Nlpine Sheep, required_guarantees=[], metrics=[output_rows=3, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched -> 1 fully matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=2 total → 2 matched, page_index_pages_skipped_by_fully_matched=1, limit_pruned_row_groups=0 total → 0 matched, scan_efficiency_ratio=21.75% (485/2.23 K)] +02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND species_min@5 < Nlpine Sheep, required_guarantees=[], metrics=[output_rows=3, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=3 total → 3 matched, limit_pruned_row_groups=0 total → 0 matched, scan_efficiency_ratio=21.75% (485/2.23 K)] # ---- pgjson format: structural golden with no metrics ---- @@ -682,7 +682,7 @@ EXPLAIN (ANALYZE, METRICS rows, LEVEL summary) select * from cat_tracking where ---- Plan with Metrics 01)SortExec: TopK(fetch=3), expr=[species@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[species@0 < Nlpine Sheep], metrics=[output_rows=3] -02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND species_min@5 < Nlpine Sheep, required_guarantees=[], metrics=[output_rows=3, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched -> 1 fully matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=2 total → 2 matched, page_index_pages_skipped_by_fully_matched=1, limit_pruned_row_groups=0 total → 0 matched, scan_efficiency_ratio=21.75% (485/2.23 K)] +02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND species_min@5 < Nlpine Sheep, required_guarantees=[], metrics=[output_rows=3, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=3 total → 3 matched, limit_pruned_row_groups=0 total → 0 matched, scan_efficiency_ratio=21.75% (485/2.23 K)] statement ok reset datafusion.sql_parser.dialect; diff --git a/datafusion/sqllogictest/test_files/information_schema.slt b/datafusion/sqllogictest/test_files/information_schema.slt index b270eba99d7b0..b0c8075c890f9 100644 --- a/datafusion/sqllogictest/test_files/information_schema.slt +++ b/datafusion/sqllogictest/test_files/information_schema.slt @@ -260,6 +260,7 @@ datafusion.execution.parquet.max_row_group_size 1048576 datafusion.execution.parquet.maximum_buffered_record_batches_per_stream 2 datafusion.execution.parquet.maximum_parallel_row_group_writers 1 datafusion.execution.parquet.metadata_size_hint 524288 +datafusion.execution.parquet.progressive_io true datafusion.execution.parquet.pruning true datafusion.execution.parquet.pushdown_filters false datafusion.execution.parquet.reorder_filters false @@ -421,6 +422,7 @@ datafusion.execution.parquet.max_row_group_size 1048576 (writing) Target maximum datafusion.execution.parquet.maximum_buffered_record_batches_per_stream 2 (writing) By default parallel parquet writer is tuned for minimum memory usage in a streaming execution plan. You may see a performance benefit when writing large parquet files by increasing maximum_parallel_row_group_writers and maximum_buffered_record_batches_per_stream if your system has idle cores and can tolerate additional memory usage. Boosting these values is likely worthwhile when writing out already in-memory data, such as from a cached data frame. datafusion.execution.parquet.maximum_parallel_row_group_writers 1 (writing) By default parallel parquet writer is tuned for minimum memory usage in a streaming execution plan. You may see a performance benefit when writing large parquet files by increasing maximum_parallel_row_group_writers and maximum_buffered_record_batches_per_stream if your system has idle cores and can tolerate additional memory usage. Boosting these values is likely worthwhile when writing out already in-memory data, such as from a cached data frame. datafusion.execution.parquet.metadata_size_hint 524288 (reading) If specified, the parquet reader will try and fetch the last `size_hint` bytes of the parquet file optimistically. If not specified, two reads are required: One read to fetch the 8-byte parquet footer and another to fetch the metadata length encoded in the footer Default setting to 512 KiB, which should be sufficient for most parquet files, it can reduce one I/O operation per parquet file. If the metadata is larger than the hint, two reads will still be performed. +datafusion.execution.parquet.progressive_io true (reading) Fetch columns progressively as row filtering and decoding need them. If false, fetch output and predicate pages together before evaluating row filters, preserving page-index pruning performed at file open. This can reduce dependent I/O rounds but may fetch pages that row filtering would otherwise avoid. Does not enable filter pushdown or prefetch. datafusion.execution.parquet.pruning true (reading) If true, the parquet reader attempts to skip entire row groups based on the predicate in the query and the metadata (min/max values) stored in the parquet file datafusion.execution.parquet.pushdown_filters false (reading) If true, filter expressions are be applied during the parquet decoding operation to reduce the number of rows decoded. This optimization is sometimes called "late materialization". datafusion.execution.parquet.reorder_filters false (reading) If true, filter expressions evaluated during the parquet decoding operation will be reordered heuristically to minimize the cost of evaluation. If false, the filters are applied in the same order as written in the query diff --git a/datafusion/sqllogictest/test_files/limit_pruning.slt b/datafusion/sqllogictest/test_files/limit_pruning.slt index dd04506ed3e54..3bb8e182535db 100644 --- a/datafusion/sqllogictest/test_files/limit_pruning.slt +++ b/datafusion/sqllogictest/test_files/limit_pruning.slt @@ -120,7 +120,7 @@ explain analyze select * from tracking_data where species > 'M' AND s >= 50 orde ---- Plan with Metrics 01)SortExec: TopK(fetch=3), expr=[species@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[species@0 < Nlpine Sheep], metrics=[output_rows=3, elapsed_compute=, output_bytes=] -02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/limit_pruning/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND species_min@5 < Nlpine Sheep, required_guarantees=[], metrics=[output_rows=3, elapsed_compute=, output_bytes=, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched -> 1 fully matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=2 total → 2 matched, page_index_pages_skipped_by_fully_matched=1, limit_pruned_row_groups=0 total → 0 matched, bytes_processed=, bytes_scanned=, row_groups_pruned_dynamic_filter=0, metadata_load_time=, scan_efficiency_ratio= (/)] +02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/limit_pruning/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND species_min@5 < Nlpine Sheep, required_guarantees=[], metrics=[output_rows=3, elapsed_compute=, output_bytes=, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=3 total → 3 matched, limit_pruned_row_groups=0 total → 0 matched, bytes_processed=, bytes_scanned=, row_groups_pruned_dynamic_filter=0, metadata_load_time=, scan_efficiency_ratio= (/)] statement ok drop table tracking_data; diff --git a/docs/source/user-guide/configs.md b/docs/source/user-guide/configs.md index 0085d4ac7c1fa..3b37283f4cddd 100644 --- a/docs/source/user-guide/configs.md +++ b/docs/source/user-guide/configs.md @@ -85,6 +85,7 @@ The following configuration settings are available: | datafusion.execution.parquet.skip_metadata | true | (reading) If true, the parquet reader skip the optional embedded metadata that may be in the file Schema. This setting can help avoid schema conflicts when querying multiple parquet files with schemas containing compatible types but different metadata | | datafusion.execution.parquet.metadata_size_hint | 524288 | (reading) If specified, the parquet reader will try and fetch the last `size_hint` bytes of the parquet file optimistically. If not specified, two reads are required: One read to fetch the 8-byte parquet footer and another to fetch the metadata length encoded in the footer Default setting to 512 KiB, which should be sufficient for most parquet files, it can reduce one I/O operation per parquet file. If the metadata is larger than the hint, two reads will still be performed. | | datafusion.execution.parquet.pushdown_filters | false | (reading) If true, filter expressions are be applied during the parquet decoding operation to reduce the number of rows decoded. This optimization is sometimes called "late materialization". | +| datafusion.execution.parquet.progressive_io | true | (reading) Fetch columns progressively as row filtering and decoding need them. If false, fetch output and predicate pages together before evaluating row filters, preserving page-index pruning performed at file open. This can reduce dependent I/O rounds but may fetch pages that row filtering would otherwise avoid. Does not enable filter pushdown or prefetch. | | datafusion.execution.parquet.reorder_filters | false | (reading) If true, filter expressions evaluated during the parquet decoding operation will be reordered heuristically to minimize the cost of evaluation. If false, the filters are applied in the same order as written in the query | | datafusion.execution.parquet.force_filter_selections | false | (reading) Force the use of RowSelections for filter results, when pushdown_filters is enabled. If false, the reader will automatically choose between a RowSelection and a Bitmap based on the number and pattern of selected rows. | | datafusion.execution.parquet.schema_force_view_types | true | (reading) If true, parquet reader will read columns of `Utf8/Utf8Large` with `Utf8View`, and `Binary/BinaryLarge` with `BinaryView`. The parquet reader is optimized for reading `Utf8View` and `BinaryView`, so such queries are significantly faster than reading `Utf8`/`Binary` and then casting to the view types. |