diff --git a/datafusion-examples/examples/data_io/json_shredding.rs b/datafusion-examples/examples/data_io/json_shredding.rs index 72fbb56773123..d61cbb11def63 100644 --- a/datafusion-examples/examples/data_io/json_shredding.rs +++ b/datafusion-examples/examples/data_io/json_shredding.rs @@ -23,6 +23,7 @@ use arrow::array::{RecordBatch, StringArray}; use arrow::datatypes::{DataType, Field, Schema, SchemaRef}; use datafusion::assert_batches_eq; +use datafusion::common::config::ParquetPushdownFilterMode; use datafusion::common::tree_node::{ Transformed, TransformedResult, TreeNode, TreeNodeRecursion, }; @@ -92,6 +93,11 @@ pub async fn json_shredding() -> Result<()> { // Set up query execution let mut cfg = SessionConfig::new(); cfg.options_mut().execution.parquet.pushdown_filters = true; + // This example needs the filter pushed into the scan so the JSON + // shredding rewriter can rewrite it into direct shredded-column + // access. Force pushdown regardless of the projection width. + cfg.options_mut().execution.parquet.pushdown_filter_mode = + ParquetPushdownFilterMode::Always; let ctx = SessionContext::new_with_config(cfg); ctx.runtime_env().register_object_store( ObjectStoreUrl::parse("memory://")?.as_ref(), diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs index f5742f09f9b08..deeee59156dc8 100644 --- a/datafusion/common/src/config.rs +++ b/datafusion/common/src/config.rs @@ -583,6 +583,72 @@ impl Display for SpillCompression { } } +/// Strategy for filter pushdown in Parquet scan when +/// `datafusion.execution.parquet.pushdown_filters` +/// (*[`ParquetOptions::pushdown_filters`]) is enabled +/// +/// Different strategies are better depending on how data is stored in the +/// Parquet files and what rows predicates select (e.g. their selectivity and +/// how many contiguous rows they select). +/// +/// Note: This option has no effect unless `pushdown_filters` is also enabled. +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] +pub enum ParquetPushdownFilterMode { + /// Let DataFusion pick the best available strategy. + /// + /// Note: currently identical to [`Self::Heuristic`], but as we implement + /// more sophisticated pushdown strategies (e.g. runtime-adaptive placement + /// in ), this may change. + #[default] + Auto, + /// Always push filters into the scan. + Always, + /// Use plan-time heuristics to decide which filters to push. + /// + /// The current heuristic skips pushdown when the projection contains fewer + /// than 3 non-filter columns, which avoid narrow-projection queries such as + /// `SELECT col2 FROM t WHERE col1 <> ''`, where `RowFilter` overhead + /// tends to dominate the decode it would save. + Heuristic, +} + +impl FromStr for ParquetPushdownFilterMode { + type Err = DataFusionError; + + fn from_str(s: &str) -> Result { + match s.to_ascii_lowercase().as_str() { + "auto" | "" => Ok(Self::Auto), + "always" => Ok(Self::Always), + "heuristic" => Ok(Self::Heuristic), + other => Err(DataFusionError::Configuration(format!( + "Invalid pushdown filter mode: {other}. Expected one of: auto, always, heuristic" + ))), + } + } +} + +impl ConfigField for ParquetPushdownFilterMode { + fn visit(&self, v: &mut V, key: &str, description: &'static str) { + v.some(key, self, description) + } + + fn set(&mut self, _: &str, value: &str) -> Result<()> { + *self = ParquetPushdownFilterMode::from_str(value)?; + Ok(()) + } +} + +impl Display for ParquetPushdownFilterMode { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let str = match self { + Self::Auto => "auto", + Self::Always => "always", + Self::Heuristic => "heuristic", + }; + write!(f, "{str}") + } +} + /// A `usize` configuration value that rejects zero when set from strings. /// /// Use this for options where zero is never a meaningful runtime value. @@ -1228,7 +1294,12 @@ config_namespace! { /// (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". - pub pushdown_filters: bool, default = false + pub pushdown_filters: bool, default = true + + /// (reading) When `pushdown_filters` is enabled, determines how DataFusion + /// pushes each filter into the Parquet scan. Options are `auto` (the default) + /// `always`, and `heurstic` (plan time heuristic). + pub pushdown_filter_mode: ParquetPushdownFilterMode, default = ParquetPushdownFilterMode::Auto /// (reading) If true, filter expressions evaluated during the parquet decoding operation /// will be reordered heuristically to minimize the cost of evaluation. If false, @@ -1241,6 +1312,19 @@ config_namespace! { /// pattern of selected rows. pub force_filter_selections: bool, default = false + /// (reading) Controls the I/O pattern used when `pushdown_filters` is + /// enabled. If false (the default), all data pages needed to read a row + /// group (for both filter evaluation and output projection) are fetched + /// in a single request, the same I/O pattern used when + /// `pushdown_filters` is disabled. If true, data is fetched + /// progressively: first the columns needed by each filter, then, after + /// the filters are evaluated, the remaining projected columns for the + /// rows that passed. Progressive fetching can reduce the total bytes + /// read when the file has a Parquet offset index, at the cost of + /// additional I/O requests per row group; files without an offset index + /// are always read with a single request per row group. + pub progressive_io: bool, default = false + /// (reading) If true, parquet reader will read columns of `Utf8/Utf8Large` with `Utf8View`, /// and `Binary/BinaryLarge` with `BinaryView`. pub schema_force_view_types: bool, default = true diff --git a/datafusion/common/src/file_options/parquet_writer.rs b/datafusion/common/src/file_options/parquet_writer.rs index c539245764d45..7835f722f60b9 100644 --- a/datafusion/common/src/file_options/parquet_writer.rs +++ b/datafusion/common/src/file_options/parquet_writer.rs @@ -236,8 +236,10 @@ impl ParquetOptions { skip_metadata: _, metadata_size_hint: _, pushdown_filters: _, + pushdown_filter_mode: _, // reads-only, not used for writer props reorder_filters: _, force_filter_selections: _, // not used for writer props + progressive_io: _, // not used for writer props allow_single_file_parallelism: _, maximum_parallel_row_group_writers: _, maximum_buffered_record_batches_per_stream: _, @@ -494,8 +496,10 @@ mod tests { skip_metadata: defaults.skip_metadata, metadata_size_hint: defaults.metadata_size_hint, pushdown_filters: defaults.pushdown_filters, + pushdown_filter_mode: defaults.pushdown_filter_mode, reorder_filters: defaults.reorder_filters, force_filter_selections: defaults.force_filter_selections, + progressive_io: defaults.progressive_io, allow_single_file_parallelism: defaults.allow_single_file_parallelism, maximum_parallel_row_group_writers: defaults .maximum_parallel_row_group_writers, @@ -614,8 +618,10 @@ mod tests { skip_metadata: global_options_defaults.skip_metadata, metadata_size_hint: global_options_defaults.metadata_size_hint, pushdown_filters: global_options_defaults.pushdown_filters, + pushdown_filter_mode: global_options_defaults.pushdown_filter_mode, reorder_filters: global_options_defaults.reorder_filters, force_filter_selections: global_options_defaults.force_filter_selections, + progressive_io: global_options_defaults.progressive_io, allow_single_file_parallelism: global_options_defaults .allow_single_file_parallelism, maximum_parallel_row_group_writers: global_options_defaults diff --git a/datafusion/core/tests/datasource/object_store_access.rs b/datafusion/core/tests/datasource/object_store_access.rs index 2503de862e06a..2b39cef5b74c8 100644 --- a/datafusion/core/tests/datasource/object_store_access.rs +++ b/datafusion/core/tests/datasource/object_store_access.rs @@ -920,6 +920,85 @@ async fn query_single_parquet_file_multi_row_groups_multiple_predicates() { ); } +#[tokio::test] +async fn query_single_parquet_file_pushdown_filters_one_shot_io() { + let test = Test::new().with_single_file_parquet().await; + test.set("datafusion.execution.parquet.pushdown_filters", "true") + .await; + // Opt out of the narrow-projection pushdown gate: these tests exist to + // demonstrate the pushdown I/O pattern, which requires a RowFilter to + // actually be installed. + test.set( + "datafusion.execution.parquet.pushdown_filter_mode", + "always", + ) + .await; + + // With filter pushdown enabled and `progressive_io` disabled (the + // default), all data needed for each row group (columns needed by the + // filter and the projection) is fetched with a single request per row + // group: the same I/O pattern as when pushdown_filters is disabled. + assert_snapshot!( + test.query("select min(b) from parquet_table WHERE a > 50").await, + @r" + ------- Query Output (1 rows) ------- + +----------------------+ + | min(parquet_table.b) | + +----------------------+ + | 1051 | + +----------------------+ + ------- Object Store Request Summary ------- + RequestCountingObjectStore() + Total Requests: 3 + - GET (opts) path=parquet_table.parquet head=true + - GET (ranges) path=parquet_table.parquet ranges=4-534,534-1064 + - GET (ranges) path=parquet_table.parquet ranges=1064-1594,1594-2124 + " + ); +} + +#[tokio::test] +async fn query_single_parquet_file_pushdown_filters_progressive_io() { + let test = Test::new().with_single_file_parquet().await; + test.set("datafusion.execution.parquet.pushdown_filters", "true") + .await; + // Opt out of the narrow-projection pushdown gate: these tests exist to + // demonstrate the pushdown I/O pattern, which requires a RowFilter to + // actually be installed. + test.set( + "datafusion.execution.parquet.pushdown_filter_mode", + "always", + ) + .await; + test.set("datafusion.execution.parquet.progressive_io", "true") + .await; + + // With `progressive_io` enabled (and an offset index present in the + // file), each row group is fetched progressively: first the columns + // needed to evaluate the filter (`a`), then the remaining projected + // columns (`b`) for the rows that passed, resulting in multiple + // requests per row group. + assert_snapshot!( + test.query("select min(b) from parquet_table WHERE a > 50").await, + @r" + ------- Query Output (1 rows) ------- + +----------------------+ + | min(parquet_table.b) | + +----------------------+ + | 1051 | + +----------------------+ + ------- Object Store Request Summary ------- + RequestCountingObjectStore() + Total Requests: 5 + - GET (opts) path=parquet_table.parquet head=true + - GET (ranges) path=parquet_table.parquet ranges=4-534 + - GET (ranges) path=parquet_table.parquet ranges=534-951,951-1064 + - GET (ranges) path=parquet_table.parquet ranges=1064-1594 + - GET (ranges) path=parquet_table.parquet ranges=1594-2124 + " + ); +} + /// Runs tests with a request counting object store struct Test { object_store: Arc, @@ -957,6 +1036,17 @@ impl Test { format!("{}", self.object_store) } + /// Set a session configuration value via SQL `SET` + async fn set(&self, key: &str, value: &str) { + self.session_context + .sql(&format!("set {key} = {value}")) + .await + .unwrap() + .collect() + .await + .unwrap(); + } + /// Store the specified bytes at the given path async fn with_bytes(self, path: &str, bytes: impl Into) -> Self { let path = Path::from(path); diff --git a/datafusion/core/tests/parquet/mod.rs b/datafusion/core/tests/parquet/mod.rs index 7066a4147c017..8540be9b6ac73 100644 --- a/datafusion/core/tests/parquet/mod.rs +++ b/datafusion/core/tests/parquet/mod.rs @@ -37,6 +37,7 @@ use datafusion::{ physical_plan::metrics::MetricsSet, prelude::{ParquetReadOptions, SessionConfig, SessionContext}, }; +use datafusion_common::config::ParquetPushdownFilterMode; use datafusion_expr::{Expr, LogicalPlan, LogicalPlanBuilder}; use datafusion_physical_plan::metrics::MetricValue; use parquet::arrow::ArrowWriter; @@ -323,6 +324,10 @@ impl ContextWithParquet { Unit::RowGroup(row_per_group) => { config = config.with_parquet_bloom_filter_pruning(true); config.options_mut().execution.parquet.pushdown_filters = true; + // force unconditional pushdown to test so the filters are + // applied for TopK dynamic RG pruning + config.options_mut().execution.parquet.pushdown_filter_mode = + ParquetPushdownFilterMode::Always; make_test_file_rg( scenario, row_per_group, @@ -339,6 +344,10 @@ impl ContextWithParquet { config = config.with_parquet_bloom_filter_pruning(true); config = config.with_parquet_page_index_pruning(true); config.options_mut().execution.parquet.pushdown_filters = true; + // force unconditional pushdown to test so the filters are + // applied for TopK dynamic RG pruning + config.options_mut().execution.parquet.pushdown_filter_mode = + ParquetPushdownFilterMode::Always; make_test_file_rg( scenario, row_per_group, diff --git a/datafusion/datasource-parquet/src/file_format.rs b/datafusion/datasource-parquet/src/file_format.rs index 6358201c06fa5..489d956edff35 100644 --- a/datafusion/datasource-parquet/src/file_format.rs +++ b/datafusion/datasource-parquet/src/file_format.rs @@ -716,8 +716,21 @@ impl From<&ParquetFormatFactory> for protobuf::TableParquetOptions { parquet_options::MetadataSizeHintOpt::MetadataSizeHint(size as u64) }), pushdown_filters: global_options.global.pushdown_filters, + pushdown_filter_mode: match global_options.global.pushdown_filter_mode { + datafusion_common::config::ParquetPushdownFilterMode::Auto => { + parquet_options::PushdownFilterMode::Auto + } + datafusion_common::config::ParquetPushdownFilterMode::Always => { + parquet_options::PushdownFilterMode::Always + } + datafusion_common::config::ParquetPushdownFilterMode::Heuristic => { + parquet_options::PushdownFilterMode::Heuristic + } + } + .into(), reorder_filters: global_options.global.reorder_filters, force_filter_selections: global_options.global.force_filter_selections, + progressive_io: global_options.global.progressive_io, data_pagesize_limit: global_options.global.data_pagesize_limit as u64, write_batch_size: global_options.global.write_batch_size as u64, writer_version: global_options.global.writer_version.to_string(), diff --git a/datafusion/datasource-parquet/src/opener/mod.rs b/datafusion/datasource-parquet/src/opener/mod.rs index a57f4695b55e3..b088272391b40 100644 --- a/datafusion/datasource-parquet/src/opener/mod.rs +++ b/datafusion/datasource-parquet/src/opener/mod.rs @@ -28,6 +28,7 @@ use crate::decoder_projection::DecoderProjection; use crate::page_filter::PagePruningAccessPlanFilter; use crate::push_decoder::{ DecoderBuilderConfig, PushDecoderStreamState, RgPlanEntry, RowGroupPruner, + one_shot_row_group_ranges, }; use crate::row_filter::RowFilterGenerator; use crate::row_group_filter::RowGroupAccessPlanFilter; @@ -263,6 +264,11 @@ pub(super) struct ParquetMorselizer { pub reorder_filters: bool, /// Should we force the reader to use RowSelections for filtering pub force_filter_selections: bool, + /// If true (and filters are pushed down), fetch column data + /// progressively as filters are evaluated. If false, fetch all data + /// pages needed for a row group in a single request. See + /// [`ParquetOptions::progressive_io`](datafusion_common::config::ParquetOptions::progressive_io) + pub progressive_io: bool, /// Should the page index be read from parquet files, if present, to skip /// data pages pub enable_page_index: bool, @@ -447,6 +453,7 @@ struct PreparedParquetOpen { reorder_predicates: bool, pushdown_filters: bool, force_filter_selections: bool, + progressive_io: bool, enable_page_index: bool, enable_bloom_filter: bool, enable_row_group_stats_pruning: bool, @@ -844,6 +851,7 @@ impl ParquetMorselizer { reorder_predicates: self.reorder_filters, pushdown_filters: self.pushdown_filters, force_filter_selections: self.force_filter_selections, + progressive_io: self.progressive_io, enable_page_index: self.enable_page_index, enable_bloom_filter: self.enable_bloom_filter, enable_row_group_stats_pruning: self.enable_row_group_stats_pruning, @@ -1435,7 +1443,7 @@ impl RowGroupsPrunedParquetOpen { prepared.virtual_state.as_deref(), )?; - let (decoder, rg_plan) = { + let (decoder, rg_plan, one_shot_ranges) = { let pushdown_predicate = prepared .pushdown_filters .then_some(prepared.predicate.as_ref()) @@ -1470,6 +1478,7 @@ impl RowGroupsPrunedParquetOpen { .copied() .map(|rg_index| RgPlanEntry { rg_index }) .collect(); + let row_selection = prepared_access_plan.row_selection.clone(); let mut builder = decoder_config.build(prepared_access_plan, reader_metadata.clone()); @@ -1482,7 +1491,48 @@ impl RowGroupsPrunedParquetOpen { } } - (builder.build()?, rg_plan) + // When filters are pushed down, the decoder normally fetches + // data progressively: the columns for each filter first, then, + // once the filters have been evaluated, the remaining projected + // columns for the rows that passed. Progressive fetching can only + // reduce the bytes read when the file has an offset index + // (without one, whole column chunks are fetched either way and + // the extra requests are pure overhead). + // + // When `progressive_io` is disabled (the default) or the file has + // no offset index, instead fetch each row group's data (all + // column chunks needed by the pushed down filters or the output + // projection) with a single I/O request: the same I/O pattern + // used when `pushdown_filters` is disabled. Filters are still + // evaluated progressively against the buffered bytes. + // + // When page-index pruning produced a row selection that actually + // prunes rows, keep progressive I/O: the selection prunes data + // pages from each request, which whole-column-chunk one-shot + // ranges would defeat (a selection implies the offset index is + // present, so progressive I/O can help). Note page-index pruning + // can also produce a selection that selects all rows; such a + // selection prunes nothing, so one-shot I/O is still used. + let selection_prunes_rows = row_selection.as_ref().is_some_and(|s| { + let plan_rows: usize = rg_plan + .iter() + .map(|e| file_metadata.row_group(e.rg_index).num_rows() as usize) + .sum(); + s.row_count() < plan_rows + }); + let filter_mask = row_filter_generator.filter_mask(); + let use_one_shot_io = filter_mask.is_some() + && !selection_prunes_rows + && (!prepared.progressive_io || file_metadata.offset_index().is_none()); + let one_shot_ranges = use_one_shot_io.then(|| { + let mut mask = decoder_projection.projection_mask().clone(); + if let Some(filter_mask) = filter_mask { + mask.union(filter_mask); + } + one_shot_row_group_ranges(&file_metadata, &rg_plan, &mask) + }); + + (builder.build()?, rg_plan, one_shot_ranges) }; let predicate_cache_inner_records = @@ -1539,6 +1589,7 @@ impl RowGroupsPrunedParquetOpen { baseline_metrics: prepared.baseline_metrics, row_group_pruner, row_groups_pruned_dynamic, + one_shot_ranges, } .into_stream(); @@ -1783,6 +1834,7 @@ mod test { pushdown_filters: bool, reorder_filters: bool, force_filter_selections: bool, + progressive_io: bool, enable_page_index: bool, enable_bloom_filter: bool, enable_row_group_stats_pruning: bool, @@ -1994,6 +2046,7 @@ mod test { pushdown_filters: false, reorder_filters: false, force_filter_selections: false, + progressive_io: false, enable_page_index: false, enable_bloom_filter: false, enable_row_group_stats_pruning: false, @@ -2163,6 +2216,7 @@ mod test { pushdown_filters: self.pushdown_filters, reorder_filters: self.reorder_filters, force_filter_selections: self.force_filter_selections, + progressive_io: self.progressive_io, enable_page_index: self.enable_page_index, enable_bloom_filter: self.enable_bloom_filter, enable_row_group_stats_pruning: self.enable_row_group_stats_pruning, diff --git a/datafusion/datasource-parquet/src/push_decoder.rs b/datafusion/datasource-parquet/src/push_decoder.rs index 74d8997198872..74d7d69788e96 100644 --- a/datafusion/datasource-parquet/src/push_decoder.rs +++ b/datafusion/datasource-parquet/src/push_decoder.rs @@ -35,7 +35,8 @@ //! The opener constructs both halves and hands the state off to //! [`PushDecoderStreamState::into_stream`] for consumption. -use std::collections::VecDeque; +use std::collections::{HashMap, VecDeque}; +use std::ops::Range; use std::sync::Arc; use arrow::array::RecordBatch; @@ -111,6 +112,46 @@ pub(crate) struct RgPlanEntry { pub(crate) rg_index: usize, } +/// Compute, for each row group in `rg_plan`, the byte ranges of all column +/// chunks included in `mask`. +/// +/// Fetching these ranges in a single request provides all the data needed to +/// decode the row group, including evaluating any pushed down filters whose +/// columns are included in `mask`. Used when +/// [`progressive_io`](datafusion_common::config::ParquetOptions::progressive_io) +/// is disabled to keep the I/O pattern the same as when `pushdown_filters` +/// is disabled (one request per row group). +/// +/// The ranges are whole column chunks (one range per column). This is the +/// coarsest granularity the decoder can request, so any byte range the +/// decoder asks for during filter evaluation or output decoding is +/// contained within one of these ranges — a requirement for the pushed +/// bytes to satisfy the decoder's requests, since the decoder's buffered +/// range lookup is a per-range containment check that does not coalesce +/// adjacent ranges. +pub(crate) fn one_shot_row_group_ranges( + metadata: &ParquetMetaData, + rg_plan: &VecDeque, + mask: &ProjectionMask, +) -> HashMap>> { + rg_plan + .iter() + .map(|entry| { + let rg = metadata.row_group(entry.rg_index); + let ranges = (0..rg.columns().len()) + .filter(|leaf_idx| mask.leaf_included(*leaf_idx)) + .map(|leaf_idx| { + // `byte_range` covers the dictionary page (if any) and + // all data pages of the column chunk + let (start, length) = rg.column(leaf_idx).byte_range(); + start..start + length + }) + .collect(); + (entry.rg_index, ranges) + }) + .collect() +} + /// Runtime row-group pruner driven by a dynamic predicate (e.g. the /// threshold expression a `TopK` operator pushes down). /// @@ -272,6 +313,18 @@ pub(crate) struct PushDecoderStreamState { pub(crate) row_group_pruner: Option, /// Count of row groups skipped at runtime by [`Self::row_group_pruner`]. pub(crate) row_groups_pruned_dynamic: Count, + /// When `Some`, use "one shot" I/O: the first data request for each row + /// group is expanded to the precomputed byte ranges covering every + /// column chunk that could be needed to evaluate pushed down filters + /// and decode the output projection, so each row group is fetched with + /// a single I/O request (the same I/O pattern as when + /// `pushdown_filters` is disabled). Keyed by row group index; entries + /// are removed as they are used. + /// + /// `None` means data is fetched progressively, exactly as requested by + /// the decoder. See + /// [`progressive_io`](datafusion_common::config::ParquetOptions::progressive_io). + pub(crate) one_shot_ranges: Option>>>, } impl PushDecoderStreamState { @@ -392,8 +445,25 @@ impl PushDecoderStreamState { // Step 3: drive the decoder. let decoder = self.decoder.as_mut().expect("decoder present"); + // When one-shot I/O is enabled and the decoder is about to start + // a new row group, note which row group so that its first data + // request can be expanded to cover the entire row group. This + // must be peeked *before* `try_next_reader` plans the row group: + // once planning starts, `peek_next_row_group` reports the row + // group after the in-flight one. + let starting_rg_index = if at_boundary && self.one_shot_ranges.is_some() { + match decoder.peek_next_row_group() { + Ok(rg_index) => rg_index, + Err(e) => { + return Some((Err(DataFusionError::from(e)), self)); + } + } + } else { + None + }; match decoder.try_next_reader() { Ok(DecodeResult::NeedsData(ranges)) => { + let ranges = self.expand_one_shot_ranges(starting_rg_index, ranges); let data = self .reader .get_byte_ranges(ranges.clone()) @@ -428,6 +498,33 @@ impl PushDecoderStreamState { } } + /// When one-shot I/O is enabled ([`Self::one_shot_ranges`] is `Some`), + /// expand the first data request for the row group `rg_index` to the + /// precomputed ranges covering everything the row group could need + /// (filter and projection columns), so the row group is fetched in a + /// single request. + /// + /// The ranges the decoder requested are always contained within the + /// expanded ranges (the decoder requests, at most, the same + /// selection-pruned pages for a subset of the columns in the mask the + /// expanded ranges were computed from), so the pushed data satisfies the + /// original request. Follow-up requests for the same row group (its + /// entry has already been removed, and `rg_index` is `None` because the + /// decoder is no longer at a row group boundary) are passed through + /// unchanged. + fn expand_one_shot_ranges( + &mut self, + rg_index: Option, + ranges: Vec>, + ) -> Vec> { + let (Some(one_shot_ranges), Some(rg_index)) = + (self.one_shot_ranges.as_mut(), rg_index) + else { + return ranges; + }; + one_shot_ranges.remove(&rg_index).unwrap_or(ranges) + } + /// Keep `rg_plan.front()` aligned with the row group the decoder will emit /// next. `try_next_reader` silently finishes row groups whose post-predicate /// selection is empty (no reader handed back), which would otherwise leave diff --git a/datafusion/datasource-parquet/src/row_filter.rs b/datafusion/datasource-parquet/src/row_filter.rs index c1a47c896c170..6294b8ffc7f2c 100644 --- a/datafusion/datasource-parquet/src/row_filter.rs +++ b/datafusion/datasource-parquet/src/row_filter.rs @@ -401,6 +401,28 @@ pub fn build_row_filter( reorder_predicates: bool, file_metrics: &ParquetFileMetrics, ) -> Result> { + Ok(build_row_filter_with_mask( + expr, + file_schema, + metadata, + reorder_predicates, + file_metrics, + )? + .map(|(filter, _mask)| filter)) +} + +/// Like [`build_row_filter`], but also returns the union of the +/// [`ProjectionMask`]s of all predicates in the filter (the set of parquet +/// leaf columns that will be read to evaluate the filter). Used by the +/// opener to compute the byte ranges needed for a whole row group when +/// `progressive_io` is disabled. +pub(crate) fn build_row_filter_with_mask( + expr: &Arc, + file_schema: &SchemaRef, + metadata: &ParquetMetaData, + reorder_predicates: bool, + file_metrics: &ParquetFileMetrics, +) -> Result> { let rows_pruned = &file_metrics.pushdown_rows_pruned; let rows_matched = &file_metrics.pushdown_rows_matched; let time = &file_metrics.row_pushdown_eval_time; @@ -458,10 +480,20 @@ pub fn build_row_filter( predicate_rows_matched, time.clone(), ) - .map(|pred| Box::new(pred) as _) + .map(|pred| Box::new(pred) as Box) }) .collect::, _>>() - .map(|filters| Some(RowFilter::new(filters))) + .map(|filters: Vec>| { + let mut mask_iter = filters.iter().map(|f| f.projection()); + let mut filter_mask = mask_iter + .next() + .expect("candidates is non-empty, checked above") + .clone(); + for mask in mask_iter { + filter_mask.union(mask); + } + Some((RowFilter::new(filters), filter_mask)) + }) } /// Builds row filters for a parquet decoder. @@ -476,6 +508,9 @@ pub(crate) struct RowFilterGenerator<'a> { reorder_predicates: bool, file_metrics: &'a ParquetFileMetrics, first_row_filter: Option, + /// Union of the [`ProjectionMask`]s of all predicates in the built + /// filter. `None` when no filter could be built. + filter_mask: Option, } impl<'a> RowFilterGenerator<'a> { @@ -493,6 +528,7 @@ impl<'a> RowFilterGenerator<'a> { reorder_predicates, file_metrics, first_row_filter: None, + filter_mask: None, }; generator.first_row_filter = generator.build(); generator @@ -502,16 +538,25 @@ impl<'a> RowFilterGenerator<'a> { self.first_row_filter.take().or_else(|| self.build()) } - fn build(&self) -> Option { + /// Returns the union of the [`ProjectionMask`]s of all predicates in the + /// row filter, or `None` if no filter was built. + pub(crate) fn filter_mask(&self) -> Option<&ProjectionMask> { + self.filter_mask.as_ref() + } + + fn build(&mut self) -> Option { let predicate = self.predicate?; - match build_row_filter( + match build_row_filter_with_mask( predicate, self.physical_file_schema, self.file_metadata, self.reorder_predicates, self.file_metrics, ) { - Ok(Some(filter)) => Some(filter), + Ok(Some((filter, mask))) => { + self.filter_mask = Some(mask); + Some(filter) + } Ok(None) => None, Err(e) => { log::debug!( diff --git a/datafusion/datasource-parquet/src/source.rs b/datafusion/datasource-parquet/src/source.rs index 097b4563af5df..67eb14cc9288d 100644 --- a/datafusion/datasource-parquet/src/source.rs +++ b/datafusion/datasource-parquet/src/source.rs @@ -32,6 +32,7 @@ use arrow_schema::{DataType, Field}; use datafusion_common::config::ConfigOptions; #[cfg(feature = "parquet_encryption")] use datafusion_common::config::EncryptionFactoryOptions; +use datafusion_common::config::ParquetPushdownFilterMode; use datafusion_datasource::as_file_source; use datafusion_datasource::file_stream::FileOpener; use datafusion_datasource::morsel::Morselizer; @@ -47,6 +48,7 @@ use datafusion_datasource::file_scan_config::FileScanConfig; use datafusion_functions::core::file_row_index::FileRowIndexFunc; use datafusion_physical_expr::expressions::{Column, DynamicFilterTracking}; use datafusion_physical_expr::projection::ProjectionExprs; +use datafusion_physical_expr::utils::collect_columns; use datafusion_physical_expr::{EquivalenceProperties, conjunction}; use datafusion_physical_expr_adapter::DefaultPhysicalExprAdapterFactory; use datafusion_physical_expr_adapter::rewrite::{ @@ -445,6 +447,21 @@ impl ParquetSource { self.table_parquet_options.global.force_filter_selections } + /// If true and `pushdown_filters` is enabled, fetch the data needed to + /// evaluate filters progressively (filter columns first, then the + /// remaining projected columns after filtering). If false (the default), + /// all data pages needed for a row group are fetched in a single request. + /// See [`datafusion_common::config::ParquetOptions::progressive_io`]. + pub fn with_progressive_io(mut self, progressive_io: bool) -> Self { + self.table_parquet_options.global.progressive_io = progressive_io; + self + } + + /// Return the value described in [`Self::with_progressive_io`] + pub(crate) fn progressive_io(&self) -> bool { + self.table_parquet_options.global.progressive_io + } + /// If enabled, the reader will read the page index /// This is used to optimize filter pushdown /// via `RowSelector` and `RowFilter` by @@ -645,6 +662,7 @@ impl FileSource for ParquetSource { pushdown_filters: self.pushdown_filters(), reorder_filters: self.reorder_filters(), force_filter_selections: self.force_filter_selections(), + progressive_io: self.progressive_io(), enable_page_index: self.enable_page_index(), enable_bloom_filter: self.bloom_filter_on_read(), enable_row_group_stats_pruning: self.table_parquet_options.global.pruning, @@ -838,7 +856,77 @@ impl FileSource for ParquetSource { // because even if scan pushdown is disabled we can still use the filters for stats pruning. let config_pushdown_enabled = config.execution.parquet.pushdown_filters; let table_pushdown_enabled = self.pushdown_filters(); - let pushdown_filters = table_pushdown_enabled || config_pushdown_enabled; + let mut pushdown_filters = table_pushdown_enabled || config_pushdown_enabled; + // Narrow-projection gate (issue #3463, discussion on PR #23369): + // RowFilter has a fixed per-row machinery overhead that only pays + // for itself when the wide-column decode it lets us skip is + // meaningful. When the projection is narrow and the filter columns + // already cover most of it (typical for `GROUP BY col`-style + // queries such as ClickBench Q10/Q11/Q40), the overhead dominates + // and pushdown regresses. Decline pushdown for those scans; keep + // it for wider projections where the fast-path pays. + // + // Kept intentionally simple: a plan-time column-count check, + // used when `pushdown_filter_mode` is `auto` or `heuristic`. A + // future adaptive-placement pass ([#22883]) can supersede this + // with a runtime cost model behind the same `auto` mode. + // + // When declined: the filter stays above the scan in a + // `FilterExec` (correctness preserved) and the predicate is + // still injected into `ParquetSource` for stats / bloom / + // page-index pruning. + // + // Users who want the pre-existing "always push" behavior can + // set `pushdown_filter_mode = always`. + // + // [#22883]: https://github.com/apache/datafusion/issues/22883 + const PUSHDOWN_MIN_NON_FILTER_COLS: usize = 3; + // Never gate a scan whose predicate already contains a dynamic + // filter — either in the incoming `filters` parameter, or already + // installed on `self.predicate` by an earlier optimizer rule (this + // is how `TopK`'s heap-threshold expression arrives: the sort + // pushdown / TopK rule injects the `DynamicFilterPhysicalExpr` + // into the scan's predicate before filter pushdown runs). Dynamic + // filters rely on the scan-time RowFilter/RowGroupPruner cascade + // to prune data as the threshold tightens, so declining pushdown + // here would silently disable the entire dynamic-RG-prune path + // for narrow `ORDER BY ... LIMIT` queries. + let existing_predicate_has_dynamic = self.predicate.as_ref().is_some_and(|p| { + DynamicFilterTracking::classify(p).contains_dynamic_filter() + }); + let incoming_filters_have_dynamic = filters + .iter() + .any(|f| DynamicFilterTracking::classify(f).contains_dynamic_filter()); + let has_dynamic_filter = + existing_predicate_has_dynamic || incoming_filters_have_dynamic; + if pushdown_filters + && !has_dynamic_filter + && config.execution.parquet.pushdown_filter_mode + != ParquetPushdownFilterMode::Always + { + // `TableSchema` layout is `[file, partition, virtual]`; only + // file columns are actually decoded from parquet, so partition + // and virtual columns don't count toward the "wide-decode + // saving" the RowFilter fast-path buys us. + let file_col_count = self.table_schema.file_schema().fields().len(); + let filter_col_indices: std::collections::HashSet = filters + .iter() + .flat_map(|f| collect_columns(f).into_iter().map(|c| c.index())) + .filter(|idx| *idx < file_col_count) + .collect(); + let non_filter_projected = self + .projection + .as_ref() + .iter() + .flat_map(|pe| collect_columns(&pe.expr)) + .map(|c| c.index()) + .filter(|idx| *idx < file_col_count && !filter_col_indices.contains(idx)) + .collect::>() + .len(); + if non_filter_projected < PUSHDOWN_MIN_NON_FILTER_COLS { + pushdown_filters = false; + } + } let mut source = self.clone(); let filters: Vec = filters @@ -875,7 +963,32 @@ impl FileSource for ParquetSource { None => conjunction(allowed_filters), }; source.predicate = Some(predicate); + // Always persist the effective pushdown decision on the source. + // When the gate declined, `pushdown_filters` is already `false` + // here (set above by the gate), so `source.pushdown_filters` + // becomes `false` and the scan will NOT install a RowFilter — + // avoiding a "double filter" (FilterExec above the scan + RowFilter + // inside the scan running the same conjuncts on every batch). + // + // A later `try_pushdown_filters` call (e.g. from TopK's dynamic + // filter injection) re-computes `pushdown_filters` from scratch: + // + // let pushdown_filters = + // table_pushdown_enabled || config_pushdown_enabled; + // + // Because the config default is `true`, `||` restores `true` even + // if a previous call flipped the source-level flag to `false`. + // The gate in the next call then re-decides based on the new + // filter shape (`has_dynamic_filter` is true for TopK's injected + // filter, so the gate skips the decline path and pushdown is + // re-enabled). So flipping the flag to `false` here does not + // permanently disable pushdown for future calls. source = source.with_pushdown_filters(pushdown_filters); + // `progressive_io` resolves table-or-session, the same way as + // `pushdown_filters` above + source = source.with_progressive_io( + self.progressive_io() || config.execution.parquet.progressive_io, + ); let source = Arc::new(source); // If pushdown_filters is false we tell our parents that they still have to handle the filters, // even if we updated the predicate to include the filters (they will only be used for stats pruning). @@ -2084,7 +2197,13 @@ mod tests { ) .expect("file_row_index should rewrite to the row_number virtual column"); - let config = ConfigOptions::default(); + let mut config = ConfigOptions::default(); + // Force unconditional pushdown so this test can exercise the + // virtual-column rejection path independently. The scan projection + // here has 0 non-filter file columns, which would otherwise trip + // the narrow-projection heuristic and mark every filter as + // `PushedDown::No` regardless of virtual-column content. + config.execution.parquet.pushdown_filter_mode = ParquetPushdownFilterMode::Always; let prop = source .try_pushdown_filters(vec![pushable, virtual_only, mixed, row_index], &config) .expect("try_pushdown_filters must not error"); @@ -2109,4 +2228,208 @@ mod tests { pushed down" ); } + + /// Regression test for the "double filter" bug on the narrow-projection + /// gate. + /// + /// Before the fix: when the gate decided to decline pushdown for a + /// narrow projection, the code deliberately skipped calling + /// `source.with_pushdown_filters(false)`. That left the source-level + /// `pushdown_filters` flag at whatever it was (which is `true` by + /// default now that the config default has been flipped), so the + /// scan still installed a `RowFilter` at read time even though the + /// parent `FilterExec` was told to keep the predicate. The result was + /// that the same conjuncts ran twice per batch — once inside the + /// scan's row filter and once in the surviving `FilterExec` above. + /// + /// This test locks in the fix: when the gate declines, the returned + /// source has `pushdown_filters == false`, guaranteeing the scan + /// will not install a RowFilter and no double filtering occurs. + #[test] + fn test_narrow_projection_gate_disables_source_pushdown_flag() { + use arrow::datatypes::{DataType, Field, Schema}; + use datafusion_common::config::ConfigOptions; + use datafusion_datasource::TableSchema; + use datafusion_expr::{col, lit as logical_lit}; + use datafusion_physical_expr::planner::logical2physical; + use datafusion_physical_plan::filter_pushdown::PushedDown; + + // Two-column file schema. Projection covers only column `a`. + // Filter references column `a`. So: + // projection = {a} (1 col) + // filter = {a} + // non_filter_projected = {a} - {a} = 0 + // 0 < PUSHDOWN_MIN_NON_FILTER_COLS (=3) + // → gate declines pushdown. + let file_schema = Arc::new(Schema::new(vec![ + Field::new("a", DataType::Int64, false), + Field::new("b", DataType::Int64, false), + ])); + let table_schema = TableSchema::from(file_schema); + let full_schema: arrow::datatypes::SchemaRef = + Arc::clone(table_schema.table_schema()); + + // Narrow projection: only column `a` (index 0). + let projection = ProjectionExprs::from_indices(&[0usize], &full_schema); + let mut source = ParquetSource::new(table_schema).with_pushdown_filters(true); + source.projection = projection; + + let filter = logical2physical(&col("a").eq(logical_lit(1i64)), &full_schema); + // Default config keeps `pushdown_filter_mode = Auto`, so the gate + // is active. + let config = ConfigOptions::default(); + assert!(config.execution.parquet.pushdown_filters); + + let prop = source + .try_pushdown_filters(vec![filter], &config) + .expect("try_pushdown_filters must not error"); + + // The gate declined: parent must retain the filter. + assert_eq!(prop.filters.len(), 1); + assert!( + matches!(prop.filters[0], PushedDown::No), + "gate should report PushedDown::No so the FilterExec above the \ + scan keeps the predicate" + ); + + // Key assertion: the updated source's `pushdown_filters` flag must + // be `false`. If it were `true`, the scan would install a RowFilter + // and the same predicate would run twice. + let updated = prop + .updated_node + .as_ref() + .expect("gate declined path must attach an updated source"); + let parquet_src = updated + .downcast_ref::() + .expect("updated node must be a ParquetSource"); + assert!( + !parquet_src.pushdown_filters(), + "narrow-projection gate must flip source.pushdown_filters to false, \ + otherwise the scan double-filters (RowFilter inside + FilterExec above)" + ); + } + + /// A wide projection must NOT trip the gate: pushdown stays enabled + /// end-to-end (both the return marker and the source-level flag). + #[test] + fn test_wide_projection_keeps_pushdown_enabled() { + use arrow::datatypes::{DataType, Field, Schema}; + use datafusion_common::config::ConfigOptions; + use datafusion_datasource::TableSchema; + use datafusion_expr::{col, lit as logical_lit}; + use datafusion_physical_expr::planner::logical2physical; + use datafusion_physical_plan::filter_pushdown::PushedDown; + + // Five-column file schema. Projection covers b, c, d, e (4 cols + // not in the filter set), filter references a only. + // non_filter_projected = 4 >= PUSHDOWN_MIN_NON_FILTER_COLS + // → gate does NOT decline. + let file_schema = Arc::new(Schema::new(vec![ + Field::new("a", DataType::Int64, false), + Field::new("b", DataType::Int64, false), + Field::new("c", DataType::Int64, false), + Field::new("d", DataType::Int64, false), + Field::new("e", DataType::Int64, false), + ])); + let table_schema = TableSchema::from(file_schema); + let full_schema: arrow::datatypes::SchemaRef = + Arc::clone(table_schema.table_schema()); + let projection = ProjectionExprs::from_indices(&[1usize, 2, 3, 4], &full_schema); + let mut source = ParquetSource::new(table_schema).with_pushdown_filters(true); + source.projection = projection; + + let filter = logical2physical(&col("a").eq(logical_lit(1i64)), &full_schema); + let config = ConfigOptions::default(); + + let prop = source + .try_pushdown_filters(vec![filter], &config) + .expect("try_pushdown_filters must not error"); + + assert_eq!(prop.filters.len(), 1); + assert!( + matches!(prop.filters[0], PushedDown::Yes), + "wide projection should push down" + ); + let updated = prop + .updated_node + .as_ref() + .expect("wide projection must attach an updated source too"); + let parquet_src = updated + .downcast_ref::() + .expect("updated node must be a ParquetSource"); + assert!( + parquet_src.pushdown_filters(), + "wide projection: source.pushdown_filters must stay true so the \ + scan installs the RowFilter" + ); + } + + /// After the narrow-projection gate declines a first pushdown call, + /// a subsequent call (e.g. TopK injecting a dynamic filter) must be + /// able to re-enable pushdown. This locks in that the `||` recovery + /// against the config default keeps the flag reachable for later + /// callers, so flipping the source flag to `false` in the gate path + /// is safe. + #[test] + fn test_gate_declined_does_not_permanently_disable_pushdown() { + use arrow::datatypes::{DataType, Field, Schema}; + use datafusion_common::config::ConfigOptions; + use datafusion_datasource::TableSchema; + use datafusion_expr::{col, lit as logical_lit}; + use datafusion_physical_expr::expressions::{DynamicFilterPhysicalExpr, lit}; + use datafusion_physical_expr::planner::logical2physical; + + let file_schema = Arc::new(Schema::new(vec![ + Field::new("a", DataType::Int64, false), + Field::new("b", DataType::Int64, false), + ])); + let table_schema = TableSchema::from(file_schema); + let full_schema: arrow::datatypes::SchemaRef = + Arc::clone(table_schema.table_schema()); + let projection = ProjectionExprs::from_indices(&[0usize], &full_schema); + let mut source = ParquetSource::new(table_schema).with_pushdown_filters(true); + source.projection = projection; + + let filter = logical2physical(&col("a").eq(logical_lit(1i64)), &full_schema); + let config = ConfigOptions::default(); + let prop = source + .try_pushdown_filters(vec![filter], &config) + .expect("first pushdown call must not error"); + let after_gate = prop + .updated_node + .as_ref() + .expect("gate declined path must attach an updated source") + .downcast_ref::() + .expect("updated node must be a ParquetSource") + .clone(); + assert!( + !after_gate.pushdown_filters(), + "first call: gate should have flipped source flag to false", + ); + + // Simulate TopK injecting a dynamic filter on the source it just + // received back from the gate-declined call. + let dyn_col = Arc::new(Column::new("a", 0)) as Arc; + let dynamic_filter = Arc::new(DynamicFilterPhysicalExpr::new( + vec![dyn_col], + lit(true) as Arc, + )) as Arc; + + let prop2 = after_gate + .try_pushdown_filters(vec![dynamic_filter], &config) + .expect("second pushdown call must not error"); + let after_dynamic = prop2 + .updated_node + .as_ref() + .expect("dynamic-filter path must attach an updated source") + .downcast_ref::() + .expect("updated node must be a ParquetSource"); + assert!( + after_dynamic.pushdown_filters(), + "second call with a dynamic filter must re-enable pushdown so the \ + RG-prune / RowFilter cascade can drive the dynamic threshold — \ + the `||` recovery against the config default is what makes this \ + safe after the first call flipped the flag to false" + ); + } } diff --git a/datafusion/proto-common/proto/datafusion_common.proto b/datafusion/proto-common/proto/datafusion_common.proto index 27d1101036d9b..5c34e6dfb0888 100644 --- a/datafusion/proto-common/proto/datafusion_common.proto +++ b/datafusion/proto-common/proto/datafusion_common.proto @@ -558,8 +558,18 @@ message ParquetOptions { bool pruning = 2; // default = true bool skip_metadata = 3; // default = true bool pushdown_filters = 5; // default = false + // Strategy for deciding whether to push filters into the parquet scan when + // `pushdown_filters` is enabled. Nested so its `ALWAYS` value does not + // collide with other file-level enum values. + enum PushdownFilterMode { + AUTO = 0; + ALWAYS = 1; + HEURISTIC = 2; + } + PushdownFilterMode pushdown_filter_mode = 40; bool reorder_filters = 6; // default = false bool force_filter_selections = 34; // default = false + bool progressive_io = 39; // default = false uint64 data_pagesize_limit = 7; // default = 1024 * 1024 uint64 write_batch_size = 8; // default = 1024 string writer_version = 9; // default = "1.0" diff --git a/datafusion/proto-common/src/from_proto/mod.rs b/datafusion/proto-common/src/from_proto/mod.rs index 169ff7f3d9ff2..5c55e83dd6ed1 100644 --- a/datafusion/proto-common/src/from_proto/mod.rs +++ b/datafusion/proto-common/src/from_proto/mod.rs @@ -40,7 +40,8 @@ use datafusion_common::{ arrow_datafusion_err, config::{ CsvOptions, JsonOptions, MaxRowGroupBytes, ParquetCdcOptions, - ParquetColumnOptions, ParquetOptions, TableParquetOptions, + ParquetColumnOptions, ParquetOptions, ParquetPushdownFilterMode, + TableParquetOptions, }, file_options::{csv_writer::CsvWriterOptions, json_writer::JsonWriterOptions}, parsers::CompressionTypeVariant, @@ -964,6 +965,26 @@ impl From for protobuf::CompressionTypeVariant { } } +impl From for ParquetPushdownFilterMode { + fn from(value: protobuf::parquet_options::PushdownFilterMode) -> Self { + match value { + protobuf::parquet_options::PushdownFilterMode::Auto => Self::Auto, + protobuf::parquet_options::PushdownFilterMode::Always => Self::Always, + protobuf::parquet_options::PushdownFilterMode::Heuristic => Self::Heuristic, + } + } +} + +impl From for protobuf::parquet_options::PushdownFilterMode { + fn from(value: ParquetPushdownFilterMode) -> Self { + match value { + ParquetPushdownFilterMode::Auto => Self::Auto, + ParquetPushdownFilterMode::Always => Self::Always, + ParquetPushdownFilterMode::Heuristic => Self::Heuristic, + } + } +} + impl From for datafusion_common::parsers::CsvQuoteStyle { fn from(value: protobuf::CsvQuoteStyle) -> Self { match value { @@ -1061,8 +1082,10 @@ impl TryFrom<&protobuf::ParquetOptions> for ParquetOptions { }) .unwrap_or(None), pushdown_filters: value.pushdown_filters, + pushdown_filter_mode: value.pushdown_filter_mode().into(), reorder_filters: value.reorder_filters, force_filter_selections: value.force_filter_selections, + progressive_io: value.progressive_io, data_pagesize_limit: value.data_pagesize_limit as usize, write_batch_size: value.write_batch_size as usize, writer_version: value.writer_version.parse().map_err(|e| { diff --git a/datafusion/proto-common/src/generated/pbjson.rs b/datafusion/proto-common/src/generated/pbjson.rs index c222cd1cb8687..5811d8cd999ed 100644 --- a/datafusion/proto-common/src/generated/pbjson.rs +++ b/datafusion/proto-common/src/generated/pbjson.rs @@ -6361,12 +6361,18 @@ impl serde::Serialize for ParquetOptions { if self.pushdown_filters { len += 1; } + if self.pushdown_filter_mode != 0 { + len += 1; + } if self.reorder_filters { len += 1; } if self.force_filter_selections { len += 1; } + if self.progressive_io { + len += 1; + } if self.data_pagesize_limit != 0 { len += 1; } @@ -6470,12 +6476,20 @@ impl serde::Serialize for ParquetOptions { if self.pushdown_filters { struct_ser.serialize_field("pushdownFilters", &self.pushdown_filters)?; } + if self.pushdown_filter_mode != 0 { + let v = parquet_options::PushdownFilterMode::try_from(self.pushdown_filter_mode) + .map_err(|_| serde::ser::Error::custom(format!("Invalid variant {}", self.pushdown_filter_mode)))?; + struct_ser.serialize_field("pushdownFilterMode", &v)?; + } if self.reorder_filters { struct_ser.serialize_field("reorderFilters", &self.reorder_filters)?; } if self.force_filter_selections { struct_ser.serialize_field("forceFilterSelections", &self.force_filter_selections)?; } + if self.progressive_io { + struct_ser.serialize_field("progressiveIo", &self.progressive_io)?; + } if self.data_pagesize_limit != 0 { #[allow(clippy::needless_borrow)] #[allow(clippy::needless_borrows_for_generic_args)] @@ -6663,10 +6677,14 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { "skipMetadata", "pushdown_filters", "pushdownFilters", + "pushdown_filter_mode", + "pushdownFilterMode", "reorder_filters", "reorderFilters", "force_filter_selections", "forceFilterSelections", + "progressive_io", + "progressiveIo", "data_pagesize_limit", "dataPagesizeLimit", "write_batch_size", @@ -6733,8 +6751,10 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { Pruning, SkipMetadata, PushdownFilters, + PushdownFilterMode, ReorderFilters, ForceFilterSelections, + ProgressiveIo, DataPagesizeLimit, WriteBatchSize, WriterVersion, @@ -6790,8 +6810,10 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { "pruning" => Ok(GeneratedField::Pruning), "skipMetadata" | "skip_metadata" => Ok(GeneratedField::SkipMetadata), "pushdownFilters" | "pushdown_filters" => Ok(GeneratedField::PushdownFilters), + "pushdownFilterMode" | "pushdown_filter_mode" => Ok(GeneratedField::PushdownFilterMode), "reorderFilters" | "reorder_filters" => Ok(GeneratedField::ReorderFilters), "forceFilterSelections" | "force_filter_selections" => Ok(GeneratedField::ForceFilterSelections), + "progressiveIo" | "progressive_io" => Ok(GeneratedField::ProgressiveIo), "dataPagesizeLimit" | "data_pagesize_limit" => Ok(GeneratedField::DataPagesizeLimit), "writeBatchSize" | "write_batch_size" => Ok(GeneratedField::WriteBatchSize), "writerVersion" | "writer_version" => Ok(GeneratedField::WriterVersion), @@ -6845,8 +6867,10 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { let mut pruning__ = None; let mut skip_metadata__ = None; let mut pushdown_filters__ = None; + let mut pushdown_filter_mode__ = None; let mut reorder_filters__ = None; let mut force_filter_selections__ = None; + let mut progressive_io__ = None; let mut data_pagesize_limit__ = None; let mut write_batch_size__ = None; let mut writer_version__ = None; @@ -6903,6 +6927,12 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { } pushdown_filters__ = Some(map_.next_value()?); } + GeneratedField::PushdownFilterMode => { + if pushdown_filter_mode__.is_some() { + return Err(serde::de::Error::duplicate_field("pushdownFilterMode")); + } + pushdown_filter_mode__ = Some(map_.next_value::()? as i32); + } GeneratedField::ReorderFilters => { if reorder_filters__.is_some() { return Err(serde::de::Error::duplicate_field("reorderFilters")); @@ -6915,6 +6945,12 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { } force_filter_selections__ = Some(map_.next_value()?); } + GeneratedField::ProgressiveIo => { + if progressive_io__.is_some() { + return Err(serde::de::Error::duplicate_field("progressiveIo")); + } + progressive_io__ = Some(map_.next_value()?); + } GeneratedField::DataPagesizeLimit => { if data_pagesize_limit__.is_some() { return Err(serde::de::Error::duplicate_field("dataPagesizeLimit")); @@ -7118,8 +7154,10 @@ 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(), + pushdown_filter_mode: pushdown_filter_mode__.unwrap_or_default(), reorder_filters: reorder_filters__.unwrap_or_default(), force_filter_selections: force_filter_selections__.unwrap_or_default(), + progressive_io: progressive_io__.unwrap_or_default(), data_pagesize_limit: data_pagesize_limit__.unwrap_or_default(), write_batch_size: write_batch_size__.unwrap_or_default(), writer_version: writer_version__.unwrap_or_default(), @@ -7156,6 +7194,80 @@ impl<'de> serde::Deserialize<'de> for ParquetOptions { deserializer.deserialize_struct("datafusion_common.ParquetOptions", FIELDS, GeneratedVisitor) } } +impl serde::Serialize for parquet_options::PushdownFilterMode { + #[allow(deprecated)] + fn serialize(&self, serializer: S) -> std::result::Result + where + S: serde::Serializer, + { + let variant = match self { + Self::Auto => "AUTO", + Self::Always => "ALWAYS", + Self::Heuristic => "HEURISTIC", + }; + serializer.serialize_str(variant) + } +} +impl<'de> serde::Deserialize<'de> for parquet_options::PushdownFilterMode { + #[allow(deprecated)] + fn deserialize(deserializer: D) -> std::result::Result + where + D: serde::Deserializer<'de>, + { + const FIELDS: &[&str] = &[ + "AUTO", + "ALWAYS", + "HEURISTIC", + ]; + + struct GeneratedVisitor; + + impl serde::de::Visitor<'_> for GeneratedVisitor { + type Value = parquet_options::PushdownFilterMode; + + fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(formatter, "expected one of: {:?}", &FIELDS) + } + + fn visit_i64(self, v: i64) -> std::result::Result + where + E: serde::de::Error, + { + i32::try_from(v) + .ok() + .and_then(|x| x.try_into().ok()) + .ok_or_else(|| { + serde::de::Error::invalid_value(serde::de::Unexpected::Signed(v), &self) + }) + } + + fn visit_u64(self, v: u64) -> std::result::Result + where + E: serde::de::Error, + { + i32::try_from(v) + .ok() + .and_then(|x| x.try_into().ok()) + .ok_or_else(|| { + serde::de::Error::invalid_value(serde::de::Unexpected::Unsigned(v), &self) + }) + } + + fn visit_str(self, value: &str) -> std::result::Result + where + E: serde::de::Error, + { + match value { + "AUTO" => Ok(parquet_options::PushdownFilterMode::Auto), + "ALWAYS" => Ok(parquet_options::PushdownFilterMode::Always), + "HEURISTIC" => Ok(parquet_options::PushdownFilterMode::Heuristic), + _ => Err(serde::de::Error::unknown_variant(value, FIELDS)), + } + } + } + deserializer.deserialize_any(GeneratedVisitor) + } +} impl serde::Serialize for Precision { #[allow(deprecated)] fn serialize(&self, serializer: S) -> std::result::Result diff --git a/datafusion/proto-common/src/generated/prost.rs b/datafusion/proto-common/src/generated/prost.rs index bdbe38538e1d7..0287a99f7cf93 100644 --- a/datafusion/proto-common/src/generated/prost.rs +++ b/datafusion/proto-common/src/generated/prost.rs @@ -815,12 +815,17 @@ pub struct ParquetOptions { /// default = false #[prost(bool, tag = "5")] pub pushdown_filters: bool, + #[prost(enumeration = "parquet_options::PushdownFilterMode", tag = "40")] + pub pushdown_filter_mode: i32, /// default = false #[prost(bool, tag = "6")] pub reorder_filters: bool, /// default = false #[prost(bool, tag = "34")] pub force_filter_selections: bool, + /// default = false + #[prost(bool, tag = "39")] + pub progressive_io: bool, /// default = 1024 * 1024 #[prost(uint64, tag = "7")] pub data_pagesize_limit: u64, @@ -915,6 +920,48 @@ pub struct ParquetOptions { } /// Nested message and enum types in `ParquetOptions`. pub mod parquet_options { + /// Strategy for deciding whether to push filters into the parquet scan when + /// `pushdown_filters` is enabled. Nested so its `ALWAYS` value does not + /// collide with other file-level enum values. + #[derive( + Clone, + Copy, + Debug, + PartialEq, + Eq, + Hash, + PartialOrd, + Ord, + ::prost::Enumeration + )] + #[repr(i32)] + pub enum PushdownFilterMode { + Auto = 0, + Always = 1, + Heuristic = 2, + } + impl PushdownFilterMode { + /// String value of the enum field names used in the ProtoBuf definition. + /// + /// The values are not transformed in any way and thus are considered stable + /// (if the ProtoBuf definition does not change) and safe for programmatic use. + pub fn as_str_name(&self) -> &'static str { + match self { + Self::Auto => "AUTO", + Self::Always => "ALWAYS", + Self::Heuristic => "HEURISTIC", + } + } + /// Creates an enum from field names used in the ProtoBuf definition. + pub fn from_str_name(value: &str) -> ::core::option::Option { + match value { + "AUTO" => Some(Self::Auto), + "ALWAYS" => Some(Self::Always), + "HEURISTIC" => Some(Self::Heuristic), + _ => None, + } + } + } #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Oneof)] pub enum MetadataSizeHintOpt { #[prost(uint64, tag = "4")] diff --git a/datafusion/proto-common/src/to_proto/mod.rs b/datafusion/proto-common/src/to_proto/mod.rs index 360981746585b..42b943de22feb 100644 --- a/datafusion/proto-common/src/to_proto/mod.rs +++ b/datafusion/proto-common/src/to_proto/mod.rs @@ -902,8 +902,12 @@ 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, + pushdown_filter_mode: protobuf::parquet_options::PushdownFilterMode::from( + value.pushdown_filter_mode, + ) as i32, reorder_filters: value.reorder_filters, force_filter_selections: value.force_filter_selections, + progressive_io: value.progressive_io, data_pagesize_limit: value.data_pagesize_limit as u64, write_batch_size: value.write_batch_size as u64, writer_version: value.writer_version.to_string(), diff --git a/datafusion/proto-models/src/from_proto.rs b/datafusion/proto-models/src/from_proto.rs index 74ead8c52049b..2b5185b859861 100644 --- a/datafusion/proto-models/src/from_proto.rs +++ b/datafusion/proto-models/src/from_proto.rs @@ -27,7 +27,7 @@ use std::sync::Arc; use datafusion_common::config::{ CsvOptions, JsonOptions, MaxRowGroupBytes, ParquetCdcOptions, ParquetColumnOptions, - ParquetOptions, TableParquetOptions, + ParquetOptions, ParquetPushdownFilterMode, TableParquetOptions, }; use datafusion_common::display::{PlanType, StringifiedPlan}; use datafusion_common::parsers::{CompressionTypeVariant, CsvQuoteStyle}; @@ -348,8 +348,20 @@ impl TryFrom<&ParquetOptionsProto> for ParquetOptions { } }), pushdown_filters: proto.pushdown_filters, + pushdown_filter_mode: match proto.pushdown_filter_mode() { + parquet_options::PushdownFilterMode::Auto => { + ParquetPushdownFilterMode::Auto + } + parquet_options::PushdownFilterMode::Always => { + ParquetPushdownFilterMode::Always + } + parquet_options::PushdownFilterMode::Heuristic => { + ParquetPushdownFilterMode::Heuristic + } + }, reorder_filters: proto.reorder_filters, force_filter_selections: proto.force_filter_selections, + progressive_io: proto.progressive_io, data_pagesize_limit: proto.data_pagesize_limit as usize, write_batch_size: proto.write_batch_size as usize, writer_version, diff --git a/datafusion/proto-models/src/generated/datafusion_proto_common.rs b/datafusion/proto-models/src/generated/datafusion_proto_common.rs index bdbe38538e1d7..0287a99f7cf93 100644 --- a/datafusion/proto-models/src/generated/datafusion_proto_common.rs +++ b/datafusion/proto-models/src/generated/datafusion_proto_common.rs @@ -815,12 +815,17 @@ pub struct ParquetOptions { /// default = false #[prost(bool, tag = "5")] pub pushdown_filters: bool, + #[prost(enumeration = "parquet_options::PushdownFilterMode", tag = "40")] + pub pushdown_filter_mode: i32, /// default = false #[prost(bool, tag = "6")] pub reorder_filters: bool, /// default = false #[prost(bool, tag = "34")] pub force_filter_selections: bool, + /// default = false + #[prost(bool, tag = "39")] + pub progressive_io: bool, /// default = 1024 * 1024 #[prost(uint64, tag = "7")] pub data_pagesize_limit: u64, @@ -915,6 +920,48 @@ pub struct ParquetOptions { } /// Nested message and enum types in `ParquetOptions`. pub mod parquet_options { + /// Strategy for deciding whether to push filters into the parquet scan when + /// `pushdown_filters` is enabled. Nested so its `ALWAYS` value does not + /// collide with other file-level enum values. + #[derive( + Clone, + Copy, + Debug, + PartialEq, + Eq, + Hash, + PartialOrd, + Ord, + ::prost::Enumeration + )] + #[repr(i32)] + pub enum PushdownFilterMode { + Auto = 0, + Always = 1, + Heuristic = 2, + } + impl PushdownFilterMode { + /// String value of the enum field names used in the ProtoBuf definition. + /// + /// The values are not transformed in any way and thus are considered stable + /// (if the ProtoBuf definition does not change) and safe for programmatic use. + pub fn as_str_name(&self) -> &'static str { + match self { + Self::Auto => "AUTO", + Self::Always => "ALWAYS", + Self::Heuristic => "HEURISTIC", + } + } + /// Creates an enum from field names used in the ProtoBuf definition. + pub fn from_str_name(value: &str) -> ::core::option::Option { + match value { + "AUTO" => Some(Self::Auto), + "ALWAYS" => Some(Self::Always), + "HEURISTIC" => Some(Self::Heuristic), + _ => None, + } + } + } #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Oneof)] pub enum MetadataSizeHintOpt { #[prost(uint64, tag = "4")] diff --git a/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt b/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt index eec6e5ae179bc..496be094b3456 100644 --- a/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt +++ b/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt @@ -337,8 +337,7 @@ physical_plan 01)HashJoinExec: mode=CollectLeft, join_type=RightSemi, on=[(id@0, id@0)] 02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id], file_type=parquet 03)--SortExec: expr=[data@1 DESC], preserve_partitioning=[false] -04)----FilterExec: DynamicFilter [ empty ] -05)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[data@1 DESC], reverse_row_groups=true, dynamic_rg_pruning=eligible +04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[data@1 DESC], reverse_row_groups=true, dynamic_rg_pruning=eligible statement count 0 SET datafusion.execution.parquet.pushdown_filters = true; @@ -606,6 +605,12 @@ LOCATION 'test_files/scratch/dynamic_filter_pushdown_config/agg_data.parquet'; statement ok SET datafusion.execution.parquet.pushdown_filters = true; +# This section exercises the pushdown path directly. Force unconditional +# pushdown (pushdown_filter_mode = always) so the expected plans (FilterExec absorbed +# into DataSourceExec) are reproduced regardless of projection width. +statement ok +SET datafusion.execution.parquet.pushdown_filter_mode = always; + # Aggregate dynamic filter should be pushed into the scan when enabled # Expecting a `DynamicFilter` inside parquet scanner's predicate query TT @@ -903,3 +908,9 @@ SET datafusion.optimizer.enable_dynamic_filter_pushdown = true; statement ok RESET datafusion.execution.parquet.max_row_group_size; + +statement ok +RESET datafusion.execution.parquet.pushdown_filter_mode; + +statement ok +RESET datafusion.execution.parquet.pushdown_filters; diff --git a/datafusion/sqllogictest/test_files/explain_analyze.slt b/datafusion/sqllogictest/test_files/explain_analyze.slt index d64efe80ccae5..bc95e3f9b162c 100644 --- a/datafusion/sqllogictest/test_files/explain_analyze.slt +++ b/datafusion/sqllogictest/test_files/explain_analyze.slt @@ -189,6 +189,12 @@ reset datafusion.explain.analyze_level; statement ok set datafusion.execution.parquet.pushdown_filters = true; +# Section exercises the pushdown path directly. Force unconditional +# pushdown (pushdown_filter_mode = always) so the expected plans (FilterExec absorbed +# into DataSourceExec) are reproduced regardless of projection width. +statement ok +set datafusion.execution.parquet.pushdown_filter_mode = always; + statement ok CREATE TABLE _cat_data AS VALUES ('Anow Vole', 7), @@ -713,3 +719,6 @@ drop table cat_tracking; statement ok reset datafusion.execution.parquet.pushdown_filters; + +statement ok +reset datafusion.execution.parquet.pushdown_filter_mode; diff --git a/datafusion/sqllogictest/test_files/information_schema.slt b/datafusion/sqllogictest/test_files/information_schema.slt index 573fb04b3451b..06009e57a58d4 100644 --- a/datafusion/sqllogictest/test_files/information_schema.slt +++ b/datafusion/sqllogictest/test_files/information_schema.slt @@ -260,8 +260,10 @@ 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 false datafusion.execution.parquet.pruning true -datafusion.execution.parquet.pushdown_filters false +datafusion.execution.parquet.pushdown_filter_mode auto +datafusion.execution.parquet.pushdown_filters true datafusion.execution.parquet.reorder_filters false datafusion.execution.parquet.schema_force_view_types true datafusion.execution.parquet.skip_arrow_metadata false @@ -420,8 +422,10 @@ 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 false (reading) Controls the I/O pattern used when `pushdown_filters` is enabled. If false (the default), all data pages needed to read a row group (for both filter evaluation and output projection) are fetched in a single request, the same I/O pattern used when `pushdown_filters` is disabled. If true, data is fetched progressively: first the columns needed by each filter, then, after the filters are evaluated, the remaining projected columns for the rows that passed. Progressive fetching can reduce the total bytes read when the file has a Parquet offset index, at the cost of additional I/O requests per row group; files without an offset index are always read with a single request per row group. 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.pushdown_filter_mode auto (reading) When `pushdown_filters` is enabled, determines how DataFusion pushes each filter into the Parquet scan. Options are `auto` (the default) `always`, and `heurstic` (plan time heuristic). +datafusion.execution.parquet.pushdown_filters true (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 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`. datafusion.execution.parquet.skip_arrow_metadata false (writing) Skip encoding the embedded arrow metadata in the KV_meta This is analogous to the `ArrowWriterOptions::with_skip_arrow_metadata`. Refer to diff --git a/datafusion/sqllogictest/test_files/limit_pruning.slt b/datafusion/sqllogictest/test_files/limit_pruning.slt index 4ef0b5c74f3e7..09fe5cd879430 100644 --- a/datafusion/sqllogictest/test_files/limit_pruning.slt +++ b/datafusion/sqllogictest/test_files/limit_pruning.slt @@ -18,6 +18,12 @@ statement ok set datafusion.execution.parquet.pushdown_filters = true; +# Test file exercises the pushdown path directly. Force unconditional +# pushdown (pushdown_filter_mode = always) so the expected plans (FilterExec absorbed +# into DataSourceExec) are reproduced. +statement ok +set datafusion.execution.parquet.pushdown_filter_mode = always; + statement ok CREATE TABLE tracking_data AS VALUES @@ -131,3 +137,6 @@ reset datafusion.explain.analyze_level; # Config reset statement ok RESET datafusion.execution.parquet.pushdown_filters; + +statement ok +RESET datafusion.execution.parquet.pushdown_filter_mode; diff --git a/datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt b/datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt index cb3be93191fb2..dfc3afb913e1b 100644 --- a/datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt +++ b/datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt @@ -21,6 +21,12 @@ # scan not just the metadata) ########## +# Test file exercises the pushdown path directly. Force unconditional +# pushdown (pushdown_filter_mode = always) so the expected plans (FilterExec absorbed +# into DataSourceExec) are reproduced regardless of projection width. +statement ok +set datafusion.execution.parquet.pushdown_filter_mode = always; + # File1 has only columns a and b statement ok COPY ( @@ -933,10 +939,13 @@ A 78 # Cleanup statement ok -set datafusion.execution.parquet.pushdown_filters = false; +RESET datafusion.execution.parquet.pushdown_filters; statement ok set datafusion.execution.parquet.reorder_filters = false; +statement ok +RESET datafusion.execution.parquet.pushdown_filter_mode; + statement ok DROP TABLE dict_filter_bug; diff --git a/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt b/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt index 72d034067663e..933ea8f629a76 100644 --- a/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt +++ b/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt @@ -24,6 +24,12 @@ set datafusion.explain.physical_plan_only = true; statement ok set datafusion.execution.parquet.pushdown_filters = true; +# Test file exercises the pushdown path directly. Force unconditional +# pushdown (pushdown_filter_mode = always) so the expected plans (FilterExec absorbed +# into DataSourceExec) are reproduced regardless of projection width. +statement ok +set datafusion.execution.parquet.pushdown_filter_mode = always; + # this one is also required to make DF skip second file due to "sufficient" amount of rows statement ok set datafusion.execution.collect_statistics = true; @@ -1334,5 +1340,8 @@ RESET datafusion.explain.physical_plan_only; statement ok RESET datafusion.execution.parquet.pushdown_filters; +statement ok +RESET datafusion.execution.parquet.pushdown_filter_mode; + statement ok drop table t; diff --git a/datafusion/sqllogictest/test_files/push_down_filter_regression.slt b/datafusion/sqllogictest/test_files/push_down_filter_regression.slt index 57509fd0395b9..5c19a58fc0d38 100644 --- a/datafusion/sqllogictest/test_files/push_down_filter_regression.slt +++ b/datafusion/sqllogictest/test_files/push_down_filter_regression.slt @@ -17,6 +17,12 @@ # Test push down filter +# Test file exercises the pushdown path directly. Force unconditional +# pushdown (pushdown_filter_mode = always) so the expected plans (FilterExec absorbed +# into DataSourceExec) are reproduced regardless of projection width. +statement ok +set datafusion.execution.parquet.pushdown_filter_mode = always; + # Regression test for https://github.com/apache/datafusion/issues/17188 query I COPY (select i as k, i as v from generate_series(1, 10000000) as t(i)) @@ -625,3 +631,6 @@ drop table t1; statement ok drop table t2; + +statement ok +RESET datafusion.execution.parquet.pushdown_filter_mode; diff --git a/datafusion/sqllogictest/test_files/sort_pushdown.slt b/datafusion/sqllogictest/test_files/sort_pushdown.slt index f2442762f3fd2..93b5f75f97fd9 100644 --- a/datafusion/sqllogictest/test_files/sort_pushdown.slt +++ b/datafusion/sqllogictest/test_files/sort_pushdown.slt @@ -5,6 +5,12 @@ SET datafusion.execution.parquet.pushdown_filters = true; statement ok SET datafusion.optimizer.enable_sort_pushdown = true; +# Test file exercises the pushdown path directly. Force unconditional +# pushdown (pushdown_filter_mode = always) so the expected plans (FilterExec absorbed +# into DataSourceExec) are reproduced regardless of projection width. +statement ok +SET datafusion.execution.parquet.pushdown_filter_mode = always; + # Test 1: Sort Pushdown for ordered Parquet files # Create a sorted dataset statement ok @@ -2959,3 +2965,6 @@ DROP TABLE tp_multifile; # Restore settings statement ok SET datafusion.execution.target_partitions = 4; + +statement ok +RESET datafusion.execution.parquet.pushdown_filter_mode; diff --git a/docs/source/user-guide/configs.md b/docs/source/user-guide/configs.md index a66ad3edf5c14..b2eb2ee97a26a 100644 --- a/docs/source/user-guide/configs.md +++ b/docs/source/user-guide/configs.md @@ -84,9 +84,11 @@ The following configuration settings are available: | 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.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.pushdown_filters | true | (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.pushdown_filter_mode | auto | (reading) When `pushdown_filters` is enabled, determines how DataFusion pushes each filter into the Parquet scan. Options are `auto` (the default) `always`, and `heurstic` (plan time heuristic). | | 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.progressive_io | false | (reading) Controls the I/O pattern used when `pushdown_filters` is enabled. If false (the default), all data pages needed to read a row group (for both filter evaluation and output projection) are fetched in a single request, the same I/O pattern used when `pushdown_filters` is disabled. If true, data is fetched progressively: first the columns needed by each filter, then, after the filters are evaluated, the remaining projected columns for the rows that passed. Progressive fetching can reduce the total bytes read when the file has a Parquet offset index, at the cost of additional I/O requests per row group; files without an offset index are always read with a single request per row group. | | 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`. | | datafusion.execution.parquet.binary_as_string | false | (reading) If true, parquet reader will read columns of `Binary/LargeBinary` with `Utf8`, and `BinaryView` with `Utf8View`. Parquet files generated by some legacy writers do not correctly set the UTF8 flag for strings, causing string columns to be loaded as BLOB instead. | | datafusion.execution.parquet.coerce_int96 | NULL | (reading) If true, parquet reader will read columns of physical type int96 as originating from a different resolution than nanosecond. This is useful for reading data from systems like Spark which stores microsecond resolution timestamps in an int96 allowing it to write values with a larger date range than 64-bit timestamps with nanosecond resolution. |