diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs index e6487f8be5f23..36b066701355a 100644 --- a/datafusion/common/src/config.rs +++ b/datafusion/common/src/config.rs @@ -1057,6 +1057,27 @@ config_namespace! { /// file only after the current one is fully consumed. pub file_stream_open_ahead: bool, default = false + /// When `true`, a Parquet scan whose predicate holds a dynamic filter + /// that can still change (for example the threshold pushed down by an + /// `ORDER BY ... LIMIT` TopK) re-checks the file's statistics against + /// the current filter right before reading the file's first data + /// pages. A file range that was opened early (by + /// `file_stream_open_ahead`, or while the filter was still empty) is + /// then skipped without fetching or decoding its row groups if the + /// filter has since been updated to exclude it. The check reuses the + /// scan's per-batch early-stop pruner, so the pruning predicate is + /// only rebuilt when the filter changed. Skipped ranges are counted by + /// the `files_ranges_pruned_before_read` metric, which is registered + /// only for scans with a dynamic filter while this option is enabled. + /// The check uses the file's `PartitionedFile` statistics, so it does + /// nothing for files without them, and it applies to every dynamic + /// filter that can still change (TopK thresholds and aggregate MIN/MAX + /// bounds; hash-join filters are complete before the probe side runs). + /// When `false` (the default) a single-row-group file is re-checked only + /// after its first batch; a file with several row groups is already + /// re-checked before each row group. + pub recheck_dynamic_filter_before_read: bool, default = false + /// Aggregation ratio (number of distinct groups / number of input rows) /// threshold for skipping partial aggregation. If the value is greater /// then partial aggregation will skip aggregation for further input diff --git a/datafusion/core/tests/fuzz_cases/mod.rs b/datafusion/core/tests/fuzz_cases/mod.rs index edb53df382c62..2865106253d43 100644 --- a/datafusion/core/tests/fuzz_cases/mod.rs +++ b/datafusion/core/tests/fuzz_cases/mod.rs @@ -26,6 +26,7 @@ mod sort_fuzz; #[expect(clippy::needless_pass_by_value)] mod sort_query_fuzz; mod topk_filter_pushdown; +mod topk_recheck_before_read_fuzz; mod aggregation_fuzzer; #[expect(clippy::needless_pass_by_value)] diff --git a/datafusion/core/tests/fuzz_cases/topk_recheck_before_read_fuzz.rs b/datafusion/core/tests/fuzz_cases/topk_recheck_before_read_fuzz.rs new file mode 100644 index 0000000000000..0c88248b2c252 --- /dev/null +++ b/datafusion/core/tests/fuzz_cases/topk_recheck_before_read_fuzz.rs @@ -0,0 +1,883 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Randomized differential test for +//! `datafusion.execution.recheck_dynamic_filter_before_read`. +//! +//! Each case writes a random set of Parquet files to an in-memory object +//! store that delays metadata and data reads differently, then runs TopK +//! queries (`ORDER BY ... LIMIT ... [OFFSET ...]`) with the option off and on +//! on a multi-threaded runtime. Both runs are compared against an oracle that +//! sorts every input row with `lexsort_to_indices`: +//! +//! * with `rid` appended as the final sort key the result must equal the +//! oracle exactly, in every column; +//! * without the tiebreaker the sort-key columns must equal the oracle row by +//! row, every returned row must equal the input row with the same `rid` in +//! every column, and no `rid` may repeat. +//! +//! The test asserts that the re-check actually skipped files in some cases +//! (`files_ranges_pruned_before_read` > 0), so it cannot pass vacuously. +//! +//! Float64 keys include -0.0, +0.0 and infinities in every file, but NaN only +//! in files written without statistics. Parquet min/max statistics exclude +//! NaN, and the scan currently trusts them in ways that are wrong for NaN +//! regardless of this option: a file whose non-NaN values are all equal is +//! treated as constant (its NaN rows are read back as that value), and a +//! `k > x` filter can prune a file whose only rows above `x` are NaN. +//! +//! Set `TOPK_RECHECK_FUZZ_SEED` to replay a seed and +//! `TOPK_RECHECK_FUZZ_CASES` to change the number of cases. Timing is not +//! replayed by a seed; a failing seed is best looped. + +use std::collections::{HashMap, HashSet}; +use std::fmt::{Display, Formatter}; +use std::ops::Range; +use std::sync::Arc; +use std::time::Duration; + +use arrow::array::{ + Array, ArrayRef, Date32Array, Decimal128Array, Float64Array, Int64Array, RecordBatch, + StringArray, +}; +use arrow::compute::{SortColumn, SortOptions, cast, concat_batches, lexsort_to_indices}; +use arrow::datatypes::{DataType, Field, Schema, SchemaRef}; +use async_trait::async_trait; +use bytes::Bytes; +use datafusion::datasource::source::DataSourceExec; +use datafusion::physical_plan::{ExecutionPlan, collect}; +use datafusion::prelude::{ParquetReadOptions, SessionConfig, SessionContext}; +use datafusion_common::tree_node::{TreeNode, TreeNodeRecursion}; +use futures::stream::BoxStream; +use object_store::memory::InMemory; +use object_store::path::Path; +use object_store::{ + CopyOptions, GetOptions, GetRange, GetResult, ListResult, MultipartUpload, + ObjectMeta, ObjectStore, ObjectStoreExt, PutMultipartOptions, PutOptions, PutPayload, + PutResult, +}; +use parking_lot::Mutex; +use parquet::arrow::ArrowWriter; +use parquet::file::metadata::ParquetMetaDataReader; +use parquet::file::properties::{EnabledStatistics, WriterProperties}; +use rand::prelude::IndexedRandom; +use rand::rngs::StdRng; +use rand::{Rng, SeedableRng}; +use url::Url; + +const DEFAULT_CASES: usize = 64; + +/// [`ObjectStore`] over [`InMemory`] that delays metadata reads (footer, +/// page index) and row-group data reads by different random amounts. +#[derive(Debug)] +struct DelayedStore { + inner: InMemory, + /// End offset of the column chunks of each file; reads ending at or + /// before it fetch row-group data. + data_end: HashMap, + metadata_delay_us: Range, + data_delay_us: Range, + rng: Mutex, +} + +impl Display for DelayedStore { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!(f, "DelayedStore") + } +} + +impl DelayedStore { + async fn delay<'a>( + &self, + location: &Path, + mut ranges: impl Iterator>, + ) { + let is_data = self + .data_end + .get(location) + .is_some_and(|end| ranges.all(|range| range.end <= *end)); + let range = if is_data { + &self.data_delay_us + } else { + &self.metadata_delay_us + }; + let micros = if range.is_empty() { + 0 + } else { + self.rng.lock().random_range(range.clone()) + }; + if micros == 0 { + tokio::task::yield_now().await; + } else { + tokio::time::sleep(Duration::from_micros(micros)).await; + } + } +} + +#[async_trait] +impl ObjectStore for DelayedStore { + async fn put_opts( + &self, + location: &Path, + payload: PutPayload, + opts: PutOptions, + ) -> object_store::Result { + self.inner.put_opts(location, payload, opts).await + } + + async fn put_multipart_opts( + &self, + location: &Path, + opts: PutMultipartOptions, + ) -> object_store::Result> { + self.inner.put_multipart_opts(location, opts).await + } + + async fn get_opts( + &self, + location: &Path, + options: GetOptions, + ) -> object_store::Result { + match &options.range { + Some(GetRange::Bounded(range)) => { + self.delay(location, std::iter::once(range)).await + } + _ => self.delay(location, std::iter::empty()).await, + } + self.inner.get_opts(location, options).await + } + + async fn get_ranges( + &self, + location: &Path, + ranges: &[Range], + ) -> object_store::Result> { + self.delay(location, ranges.iter()).await; + self.inner.get_ranges(location, ranges).await + } + + fn list( + &self, + prefix: Option<&Path>, + ) -> BoxStream<'static, object_store::Result> { + self.inner.list(prefix) + } + + fn list_with_offset( + &self, + prefix: Option<&Path>, + offset: &Path, + ) -> BoxStream<'static, object_store::Result> { + self.inner.list_with_offset(prefix, offset) + } + + async fn list_with_delimiter( + &self, + prefix: Option<&Path>, + ) -> object_store::Result { + self.inner.list_with_delimiter(prefix).await + } + + fn delete_stream( + &self, + locations: BoxStream<'static, object_store::Result>, + ) -> BoxStream<'static, object_store::Result> { + self.inner.delete_stream(locations) + } + + async fn copy_opts( + &self, + from: &Path, + to: &Path, + options: CopyOptions, + ) -> object_store::Result<()> { + self.inner.copy_opts(from, to, options).await + } +} + +#[derive(Clone, Copy, Debug)] +enum KeyType { + Int64, + /// Includes -0.0, +0.0, infinities and (only without statistics) NaN. + Float64, + Utf8, + Decimal128, + Date32, +} + +impl KeyType { + fn data_type(self) -> DataType { + match self { + KeyType::Int64 => DataType::Int64, + KeyType::Float64 => DataType::Float64, + KeyType::Utf8 => DataType::Utf8, + KeyType::Decimal128 => DataType::Decimal128(18, 3), + KeyType::Date32 => DataType::Date32, + } + } + + /// Build a key column from order-preserving ordinals (`None` = NULL). + /// + /// `with_stats` is whether the file is written with statistics; NaN is + /// only generated without them (see the module documentation). + fn build( + self, + ordinals: &[Option], + with_stats: bool, + rng: &mut StdRng, + ) -> ArrayRef { + match self { + KeyType::Int64 => Arc::new(Int64Array::from(ordinals.to_vec())), + KeyType::Float64 => { + let specials: &[f64] = if with_stats { + &[-0.0, 0.0, f64::INFINITY, f64::NEG_INFINITY] + } else { + &[f64::NAN, -0.0, 0.0, f64::INFINITY, f64::NEG_INFINITY] + }; + Arc::new(Float64Array::from_iter(ordinals.iter().map(|o| { + o.map(|o| { + if rng.random_bool(0.04) { + *specials.choose(rng).unwrap() + } else { + o as f64 / 4.0 + } + }) + }))) + } + KeyType::Utf8 => Arc::new(StringArray::from_iter( + ordinals + .iter() + .map(|o| o.map(|o| format!("s{:07}", o + 1_000_000))), + )), + KeyType::Decimal128 => Arc::new( + Decimal128Array::from_iter( + ordinals.iter().map(|o| o.map(|o| o as i128 * 1000 + 7)), + ) + .with_precision_and_scale(18, 3) + .unwrap(), + ), + KeyType::Date32 => Arc::new(Date32Array::from_iter( + ordinals.iter().map(|o| o.map(|o| o as i32)), + )), + } + } +} + +/// How key ordinals are spread across files. +#[derive(Clone, Copy, Debug)] +enum Layout { + /// File `i` holds ascending keys from its own disjoint range. + SortedDisjoint, + /// Neighbouring files' key ranges overlap. + Overlapping, + /// Uniformly random keys over a wide domain. + Random, + /// Few distinct keys, lots of ties. + DuplicateHeavy, +} + +fn ordinals( + layout: Layout, + file_index: usize, + rows: usize, + constant: bool, + null_fraction: f64, + rng: &mut StdRng, +) -> Vec> { + let file = file_index as i64; + let constant_value = rng.random_range(-500..500); + (0..rows as i64) + .map(|row| { + if rng.random_bool(null_fraction) { + return None; + } + if constant { + return Some(constant_value); + } + Some(match layout { + Layout::SortedDisjoint => file * 1000 + row - 2000, + Layout::Overlapping => file * 60 + rng.random_range(0..200) - 300, + Layout::Random => rng.random_range(-5000..5000), + Layout::DuplicateHeavy => rng.random_range(-4..4), + }) + }) + .collect() +} + +/// One generated data set: the files in the store and every row in `rid` +/// order (`rid` equals the row's index in `all_rows`). +struct DataSet { + store: Arc, + schema: SchemaRef, + all_rows: RecordBatch, + /// Row count of each file, in generation order. + file_rows: Vec, + description: String, +} + +const KEY_TYPES: [KeyType; 5] = [ + KeyType::Int64, + KeyType::Float64, + KeyType::Utf8, + KeyType::Decimal128, + KeyType::Date32, +]; +const LAYOUTS: [Layout; 4] = [ + Layout::SortedDisjoint, + Layout::Overlapping, + Layout::Random, + Layout::DuplicateHeavy, +]; + +/// Generate a data set. A `race_prone` case uses sorted, disjoint `k1` +/// ranges, files with statistics and data reads much slower than metadata +/// reads, which is the shape where files opened ahead are skipped by the +/// re-check. +async fn generate_data(rng: &mut StdRng, race_prone: bool) -> DataSet { + let k1_type = *KEY_TYPES.choose(rng).unwrap(); + let k2_type = *KEY_TYPES.choose(rng).unwrap(); + let k1_layout = if race_prone { + Layout::SortedDisjoint + } else { + *LAYOUTS.choose(rng).unwrap() + }; + let k2_layout = *LAYOUTS.choose(rng).unwrap(); + let num_files = rng.random_range(2..=12); + let schema = Arc::new(Schema::new(vec![ + Field::new("rid", DataType::Int64, false), + Field::new("k1", k1_type.data_type(), true), + Field::new("k2", k2_type.data_type(), true), + Field::new("payload", DataType::Utf8, false), + ])); + + // Write files in a shuffled path order so that the scan's statistics + // based file reordering has work to do. + let mut path_order: Vec = (0..num_files).collect(); + for i in (1..path_order.len()).rev() { + path_order.swap(i, rng.random_range(0..=i)); + } + + let inner = InMemory::new(); + let mut data_end = HashMap::new(); + let mut batches = vec![]; + let mut file_rows = vec![]; + let mut file_notes = vec![]; + let mut next_rid = 0i64; + for (file_index, path_index) in path_order.iter().copied().enumerate() { + let rows = if rng.random_bool(0.1) { + 0 + } else { + rng.random_range(1..=300) + }; + let constant = rng.random_bool(0.15); + let with_stats = race_prone || rng.random_bool(0.75); + let null_fraction = *[0.0, 0.0, 0.1, 0.5].choose(rng).unwrap(); + let row_group_rows = *[None, None, Some(64), Some(100)].choose(rng).unwrap(); + + let k1 = ordinals(k1_layout, file_index, rows, constant, null_fraction, rng); + let k2 = ordinals(k2_layout, file_index, rows, false, null_fraction, rng); + let rid = next_rid..next_rid + rows as i64; + next_rid += rows as i64; + let batch = RecordBatch::try_new( + Arc::clone(&schema), + vec![ + Arc::new(Int64Array::from_iter_values(rid.clone())), + k1_type.build(&k1, with_stats, rng), + k2_type.build(&k2, with_stats, rng), + Arc::new(StringArray::from_iter_values( + rid.map(|r| format!("payload-{r}")), + )), + ], + ) + .unwrap(); + + let mut props = WriterProperties::builder(); + if !with_stats { + props = props.set_statistics_enabled(EnabledStatistics::None); + } + if let Some(row_group_rows) = row_group_rows { + props = props.set_max_row_group_row_count(Some(row_group_rows)); + } + let mut buf = vec![]; + let mut writer = + ArrowWriter::try_new(&mut buf, Arc::clone(&schema), Some(props.build())) + .unwrap(); + writer.write(&batch).unwrap(); + writer.close().unwrap(); + let bytes = Bytes::from(buf); + let metadata = ParquetMetaDataReader::new() + .parse_and_finish(&bytes) + .unwrap(); + let end = metadata + .row_groups() + .iter() + .flat_map(|rg| rg.columns()) + .map(|c| { + let (start, len) = c.byte_range(); + start + len + }) + .max() + .unwrap_or(0); + let path = Path::from(format!("data/file_{path_index:02}.parquet")); + inner.put(&path, bytes.into()).await.unwrap(); + data_end.insert(path, end); + + file_notes.push(format!( + "{rows}r{}{}{}", + if constant { ",const" } else { "" }, + if with_stats { "" } else { ",nostats" }, + row_group_rows + .map(|r| format!(",rg{r}")) + .unwrap_or_default(), + )); + file_rows.push(rows); + batches.push(batch); + } + + // Different, random delays for metadata and data reads. Data reads are + // usually slower than footers so files opened ahead are planned before + // the TopK has seen any batch. + let delays = if race_prone { + 2 + } else { + rng.random_range(0..4) + }; + let (metadata_delay_us, data_delay_us) = match delays { + 0 => (0..1, 0..1), + 1 => (0..200, 500..4000), + 2 => (0..1, 1000..3000), + _ => (100..2000, 0..500), + }; + let description = format!( + "k1={k1_type:?}/{k1_layout:?} k2={k2_type:?}/{k2_layout:?} \ + files=[{}] metadata_delay_us={metadata_delay_us:?} \ + data_delay_us={data_delay_us:?}", + file_notes.join(" "), + ); + let store = Arc::new(DelayedStore { + inner, + data_end, + metadata_delay_us, + data_delay_us, + rng: Mutex::new(StdRng::seed_from_u64(rng.random())), + }); + let all_rows = concat_batches(&schema, &batches).unwrap(); + DataSet { + store, + schema, + all_rows, + file_rows, + description, + } +} + +#[derive(Clone, Debug)] +struct OrderKey { + column: &'static str, + descending: bool, + nulls_first: bool, +} + +impl OrderKey { + fn sql(&self) -> String { + format!( + "{} {} {}", + self.column, + if self.descending { "DESC" } else { "ASC" }, + if self.nulls_first { + "NULLS FIRST" + } else { + "NULLS LAST" + } + ) + } +} + +#[derive(Debug)] +struct QueryCase { + keys: Vec, + limit: usize, + offset: usize, +} + +impl QueryCase { + fn sql(&self, tiebreak: bool) -> String { + let mut order: Vec = self.keys.iter().map(OrderKey::sql).collect(); + if tiebreak { + order.push("rid ASC".to_string()); + } + let offset = if self.offset > 0 { + format!(" OFFSET {}", self.offset) + } else { + String::new() + }; + format!( + "SELECT * FROM t ORDER BY {} LIMIT {}{offset}", + order.join(", "), + self.limit + ) + } +} + +/// Generate a query; a `race_prone` one orders by `k1 ASC` with a small +/// LIMIT (see [`generate_data`]). +fn generate_query(rng: &mut StdRng, file_rows: &[usize], race_prone: bool) -> QueryCase { + let mut keys = vec![]; + let num_keys = if race_prone { + 1 + } else { + rng.random_range(1..=2) + }; + for column in ["k1", "k2"].into_iter().take(num_keys) { + keys.push(OrderKey { + column, + descending: !race_prone && rng.random_bool(0.5), + nulls_first: rng.random_bool(0.5), + }); + } + if race_prone { + return QueryCase { + keys, + limit: *[1, 2, 3, 10].choose(rng).unwrap(), + offset: 0, + }; + } + // LIMIT values around file boundaries (cumulative row counts), plus small + // and whole-table limits. + let total: usize = file_rows.iter().sum(); + let mut candidates = vec![1, 2, 3, 10, total.saturating_sub(1), total, total + 1]; + let mut cumulative = 0; + for rows in file_rows { + cumulative += rows; + candidates.extend([cumulative.saturating_sub(1), cumulative, cumulative + 1]); + } + candidates.extend(file_rows.iter().map(|r| r + 1)); + let limit = (*candidates.choose(rng).unwrap()).max(1); + let offset = if rng.random_bool(0.2) { + rng.random_range(1..=5) + } else { + 0 + }; + QueryCase { + keys, + limit, + offset, + } +} + +#[derive(Debug)] +struct ScanConfig { + target_partitions: usize, + open_ahead: bool, + pushdown_filters: bool, + split_files: bool, + work_stealing: bool, + page_index: bool, + batch_size: usize, +} + +fn generate_config(rng: &mut StdRng, race_prone: bool) -> ScanConfig { + ScanConfig { + target_partitions: *[1, 2, 3, 5, 8].choose(rng).unwrap(), + open_ahead: race_prone || rng.random_bool(0.75), + pushdown_filters: rng.random_bool(0.3), + split_files: rng.random_bool(0.6), + work_stealing: rng.random_bool(0.8), + page_index: rng.random_bool(0.5), + batch_size: if rng.random_bool(0.3) { + *[1, 2, 3, 7, 64, 8192].choose(rng).unwrap() + } else { + rng.random_range(1..=8192) + }, + } +} + +impl ScanConfig { + fn session_config(&self, recheck: bool) -> SessionConfig { + let mut config = SessionConfig::new() + .with_target_partitions(self.target_partitions) + .with_batch_size(self.batch_size); + let options = config.options_mut(); + options.execution.file_stream_open_ahead = self.open_ahead; + options.execution.recheck_dynamic_filter_before_read = recheck; + options.execution.enable_file_stream_work_stealing = self.work_stealing; + options.execution.parquet.pushdown_filters = self.pushdown_filters; + options.execution.parquet.enable_page_index = self.page_index; + if self.split_files { + options.optimizer.repartition_file_min_size = 0; + } + config + } +} + +/// Sum a counter over every Parquet scan in `plan`. +fn sum_scan_metric(plan: &Arc, name: &str) -> Option { + let mut total = None; + plan.apply(|node| { + if node.downcast_ref::().is_some() + && let Some(value) = node + .metrics() + .and_then(|m| m.sum_by_name(name)) + .map(|v| v.as_usize()) + { + *total.get_or_insert(0) += value; + } + Ok(TreeNodeRecursion::Continue) + }) + .unwrap(); + total +} + +async fn run_query( + data: &DataSet, + config: &ScanConfig, + recheck: bool, + sql: &str, +) -> (RecordBatch, Option) { + let ctx = SessionContext::new_with_config(config.session_config(recheck)); + let url = Url::parse("delayed://bucket").unwrap(); + ctx.register_object_store(&url, Arc::clone(&data.store) as _); + ctx.register_parquet("t", "delayed://bucket/data/", ParquetReadOptions::default()) + .await + .unwrap(); + let plan = ctx + .sql(sql) + .await + .unwrap() + .create_physical_plan() + .await + .unwrap(); + let batches = collect(Arc::clone(&plan), ctx.task_ctx()).await.unwrap(); + let pruned_before_read = sum_scan_metric(&plan, "files_ranges_pruned_before_read"); + + // Cast back to the input types (strings are read as views). + let columns = data + .schema + .fields() + .iter() + .map(|field| { + let parts = batches + .iter() + .map(|b| { + let column = b.column_by_name(field.name()).unwrap(); + cast(column, field.data_type()).unwrap() + }) + .collect::>(); + let parts = parts.iter().map(|a| a.as_ref()).collect::>(); + if parts.is_empty() { + arrow::array::new_empty_array(field.data_type()) + } else { + arrow::compute::concat(&parts).unwrap() + } + }) + .collect(); + let result = RecordBatch::try_new(Arc::clone(&data.schema), columns).unwrap(); + (result, pruned_before_read) +} + +/// Oracle: every input row sorted by `query`'s keys (plus `rid` when +/// `tiebreak`), then the LIMIT/OFFSET window applied. +fn oracle(data: &DataSet, query: &QueryCase, tiebreak: bool) -> RecordBatch { + let rows = &data.all_rows; + let mut sort_columns: Vec = query + .keys + .iter() + .map(|key| SortColumn { + values: Arc::clone(rows.column_by_name(key.column).unwrap()), + options: Some(SortOptions { + descending: key.descending, + nulls_first: key.nulls_first, + }), + }) + .collect(); + if tiebreak { + sort_columns.push(SortColumn { + values: Arc::clone(rows.column_by_name("rid").unwrap()), + options: Some(SortOptions::default()), + }); + } + let indices = lexsort_to_indices(&sort_columns, None).unwrap(); + let sorted = arrow::compute::take_record_batch(rows, &indices).unwrap(); + let start = query.offset.min(sorted.num_rows()); + let len = query.limit.min(sorted.num_rows() - start); + sorted.slice(start, len) +} + +/// Compare `actual` with `expected` exactly (bitwise for floats, so NaN and +/// the sign of zero matter). +fn check_exact(expected: &RecordBatch, actual: &RecordBatch) -> Result<(), String> { + if expected.num_rows() != actual.num_rows() { + return Err(format!( + "row count {} != expected {}", + actual.num_rows(), + expected.num_rows() + )); + } + for (i, field) in expected.schema().fields().iter().enumerate() { + if expected.column(i).to_data() != actual.column(i).to_data() { + return Err(format!( + "column {} differs:\nexpected {:?}\nactual {:?}", + field.name(), + expected.column(i), + actual.column(i) + )); + } + } + Ok(()) +} + +/// Check a result without a tiebreaker: sort keys equal the oracle row by row, +/// each row equals the input row with the same `rid`, and `rid`s are unique. +fn check_without_tiebreak( + data: &DataSet, + query: &QueryCase, + expected: &RecordBatch, + actual: &RecordBatch, +) -> Result<(), String> { + if expected.num_rows() != actual.num_rows() { + return Err(format!( + "row count {} != expected {}", + actual.num_rows(), + expected.num_rows() + )); + } + for key in &query.keys { + let e = expected.column_by_name(key.column).unwrap(); + let a = actual.column_by_name(key.column).unwrap(); + if e.to_data() != a.to_data() { + return Err(format!( + "sort key {} differs:\nexpected {e:?}\nactual {a:?}", + key.column + )); + } + } + let rids = actual + .column_by_name("rid") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + let mut seen = HashSet::new(); + for row in 0..actual.num_rows() { + let rid = rids.value(row); + if !seen.insert(rid) { + return Err(format!("rid {rid} returned twice")); + } + let Ok(input_row) = usize::try_from(rid) else { + return Err(format!("unknown rid {rid}")); + }; + if input_row >= data.all_rows.num_rows() { + return Err(format!("unknown rid {rid}")); + } + for (i, field) in data.schema.fields().iter().enumerate() { + let input = data.all_rows.column(i).slice(input_row, 1); + let output = actual.column(i).slice(row, 1); + if input.to_data() != output.to_data() { + return Err(format!( + "row with rid {rid} has wrong {}: input {input:?} output {output:?}", + field.name() + )); + } + } + } + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn test_fuzz_topk_recheck_before_read() { + let seed = std::env::var("TOPK_RECHECK_FUZZ_SEED") + .ok() + .filter(|s| !s.is_empty()) + .map(|s| { + s.parse::() + .expect("TOPK_RECHECK_FUZZ_SEED must be a u64") + }) + .unwrap_or_else(|| rand::rng().random()); + let cases = std::env::var("TOPK_RECHECK_FUZZ_CASES") + .ok() + .filter(|s| !s.is_empty()) + .map(|s| { + s.parse::() + .expect("TOPK_RECHECK_FUZZ_CASES must be a usize") + }) + .unwrap_or(DEFAULT_CASES); + println!("topk recheck fuzz: TOPK_RECHECK_FUZZ_SEED={seed} cases={cases}"); + let mut rng = StdRng::seed_from_u64(seed); + + let mut engaged_cases = 0; + let mut files_pruned_before_read = 0; + let mut failures = vec![]; + for case in 0..cases { + // Every fourth case is biased towards the open-ahead race so that the + // re-check is reliably exercised; the others are fully random. + let race_prone = case % 4 == 0; + let data = generate_data(&mut rng, race_prone).await; + let config = generate_config(&mut rng, race_prone); + let query = generate_query(&mut rng, &data.file_rows, race_prone); + let context = format!( + "seed={seed} case={case} race_prone={race_prone} {}\n{config:?}\n{query:?}", + data.description + ); + println!("{context}"); + + let mut case_pruned = 0; + for tiebreak in [true, false] { + let sql = query.sql(tiebreak); + let expected = oracle(&data, &query, tiebreak); + for recheck in [false, true] { + let (actual, pruned) = run_query(&data, &config, recheck, &sql).await; + if !recheck && pruned.is_some() { + failures.push(format!( + "{context}\n{sql}: metric registered with recheck off" + )); + } + case_pruned += pruned.unwrap_or(0); + let check = if tiebreak { + check_exact(&expected, &actual) + } else { + check_without_tiebreak(&data, &query, &expected, &actual) + }; + if let Err(e) = check { + failures.push(format!("{context}\n{sql}\nrecheck={recheck}: {e}")); + } + } + } + if case_pruned > 0 { + engaged_cases += 1; + files_pruned_before_read += case_pruned; + } + } + + println!( + "topk recheck fuzz: {engaged_cases}/{cases} cases skipped \ + {files_pruned_before_read} file ranges before their first read" + ); + if !failures.is_empty() { + for failure in &failures { + println!("FAILURE:\n{failure}\n"); + } + panic!( + "{} TopK recheck fuzz failures (TOPK_RECHECK_FUZZ_SEED={seed})", + failures.len() + ); + } + assert!( + engaged_cases > 0, + "the pre-read re-check never skipped a file (TOPK_RECHECK_FUZZ_SEED={seed}); \ + the test would pass without exercising it" + ); +} diff --git a/datafusion/core/tests/parquet/mod.rs b/datafusion/core/tests/parquet/mod.rs index 7066a4147c017..d7d118c7007ec 100644 --- a/datafusion/core/tests/parquet/mod.rs +++ b/datafusion/core/tests/parquet/mod.rs @@ -58,6 +58,7 @@ mod page_pruning; mod row_group_pruning; mod schema; mod schema_coercion; +mod topk_recheck_before_read; mod utils; #[cfg(test)] diff --git a/datafusion/core/tests/parquet/topk_recheck_before_read.rs b/datafusion/core/tests/parquet/topk_recheck_before_read.rs new file mode 100644 index 0000000000000..500dbacce513a --- /dev/null +++ b/datafusion/core/tests/parquet/topk_recheck_before_read.rs @@ -0,0 +1,537 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! End-to-end tests for +//! `datafusion.execution.recheck_dynamic_filter_before_read`. +//! +//! A TopK (`ORDER BY k LIMIT n`) pushes a dynamic filter into the Parquet +//! scan that starts as `true` and tightens once the TopK heap is full. A file +//! claimed before that first update (opened ahead by +//! `datafusion.execution.file_stream_open_ahead`, or claimed by a partition +//! whose first byte range was an empty sliver) passes file pruning and +//! row-group pruning against the empty filter. Without the option, such a +//! single-row-group file is only dropped by the per-batch early stop, after +//! its row group has been fetched. With the option, the scan re-checks the +//! filter right before the first data read and skips the file without +//! fetching it. +//! +//! The race is made deterministic with [`GatedStore`]: metadata reads +//! complete immediately while every data read first yields to the runtime. +//! `FileStream` polls the open-ahead file's pending metadata I/O before it +//! polls the active reader, so the open-ahead file is always planned (and +//! keeps its row group) before the first batch reaches the TopK. + +use std::collections::{BTreeSet, HashMap}; +use std::fmt::{Display, Formatter}; +use std::ops::Range; +use std::sync::Arc; + +use arrow::array::{ArrayRef, Int64Array, RecordBatch, StringArray}; +use arrow::datatypes::{DataType, Field, Schema}; +use async_trait::async_trait; +use bytes::Bytes; +use datafusion::datasource::physical_plan::FileScanConfig; +use datafusion::datasource::source::DataSourceExec; +use datafusion::physical_plan::{ExecutionPlan, collect}; +use datafusion::prelude::{ParquetReadOptions, SessionConfig, SessionContext}; +use datafusion_common::assert_batches_eq; +use datafusion_common::tree_node::{TreeNode, TreeNodeRecursion}; +use datafusion_physical_plan::metrics::{MetricValue, MetricsSet}; +use futures::stream::BoxStream; +use object_store::memory::InMemory; +use object_store::path::Path; +use object_store::{ + CopyOptions, GetOptions, GetRange, GetResult, ListResult, MultipartUpload, + ObjectMeta, ObjectStore, ObjectStoreExt, PutMultipartOptions, PutOptions, PutPayload, + PutResult, +}; +use parking_lot::Mutex; +use parquet::arrow::ArrowWriter; +use parquet::file::metadata::ParquetMetaDataReader; +use url::Url; + +use crate::parquet::utils::MetricsFinder; + +/// How many times each data read yields to the runtime before it completes. +/// +/// One yield is enough to let `FileStream` plan its open-ahead file before the +/// active file's first batch; more yields let every partition of a +/// multi-partition scan finish its initial claims first, which reproduces the +/// "everything was claimed before the first TopK update" shape seen on remote +/// object stores. +const DATA_READ_YIELDS: usize = 8; + +/// [`ObjectStore`] that answers metadata reads (footer, page index) +/// immediately and holds each data read for [`DATA_READ_YIELDS`] yields, +/// recording which file every data read was for. +#[derive(Debug)] +struct GatedStore { + inner: InMemory, + /// End offset of the column chunks of each file. A read that ends at or + /// before it fetches row-group data; any other read is metadata. + data_end: HashMap, + data_reads: Mutex>, +} + +impl Display for GatedStore { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!(f, "GatedStore") + } +} + +impl GatedStore { + /// Returns `true` (and records the read) when `ranges` fetch row-group + /// data of `location`. + fn record_if_data_read<'a>( + &self, + location: &Path, + mut ranges: impl Iterator>, + ) -> bool { + let Some(data_end) = self.data_end.get(location) else { + return false; + }; + let is_data = ranges.all(|range| range.end <= *data_end); + if is_data { + self.data_reads.lock().push(location.clone()); + } + is_data + } + + async fn hold_data_read() { + for _ in 0..DATA_READ_YIELDS { + tokio::task::yield_now().await; + } + } + + fn take_data_reads(&self) -> Vec { + std::mem::take(&mut *self.data_reads.lock()) + } +} + +#[async_trait] +impl ObjectStore for GatedStore { + async fn put_opts( + &self, + location: &Path, + payload: PutPayload, + opts: PutOptions, + ) -> object_store::Result { + self.inner.put_opts(location, payload, opts).await + } + + async fn put_multipart_opts( + &self, + location: &Path, + opts: PutMultipartOptions, + ) -> object_store::Result> { + self.inner.put_multipart_opts(location, opts).await + } + + async fn get_opts( + &self, + location: &Path, + options: GetOptions, + ) -> object_store::Result { + if let Some(GetRange::Bounded(range)) = &options.range + && self.record_if_data_read(location, std::iter::once(range)) + { + Self::hold_data_read().await; + } + self.inner.get_opts(location, options).await + } + + async fn get_ranges( + &self, + location: &Path, + ranges: &[Range], + ) -> object_store::Result> { + if self.record_if_data_read(location, ranges.iter()) { + Self::hold_data_read().await; + } + self.inner.get_ranges(location, ranges).await + } + + fn list( + &self, + prefix: Option<&Path>, + ) -> BoxStream<'static, object_store::Result> { + self.inner.list(prefix) + } + + fn list_with_offset( + &self, + prefix: Option<&Path>, + offset: &Path, + ) -> BoxStream<'static, object_store::Result> { + self.inner.list_with_offset(prefix, offset) + } + + async fn list_with_delimiter( + &self, + prefix: Option<&Path>, + ) -> object_store::Result { + self.inner.list_with_delimiter(prefix).await + } + + fn delete_stream( + &self, + locations: BoxStream<'static, object_store::Result>, + ) -> BoxStream<'static, object_store::Result> { + self.inner.delete_stream(locations) + } + + async fn copy_opts( + &self, + from: &Path, + to: &Path, + options: CopyOptions, + ) -> object_store::Result<()> { + self.inner.copy_opts(from, to, options).await + } +} + +/// A set of single-row-group files with disjoint, ascending `k` ranges: +/// file `i` holds `k` in `[i * 1000, i * 1000 + rows[i])`. +struct TestFiles { + store: Arc, + /// Path and row-group data bytes of each file, in key order. + files: Vec<(Path, u64)>, +} + +impl TestFiles { + async fn new(rows_per_file: &[i64]) -> Self { + let schema = Arc::new(Schema::new(vec![ + Field::new("k", DataType::Int64, false), + Field::new("v", DataType::Utf8, false), + ])); + let inner = InMemory::new(); + let mut data_end = HashMap::new(); + let mut files = vec![]; + for (i, rows) in rows_per_file.iter().enumerate() { + let base = i as i64 * 1000; + let k: ArrayRef = Arc::new(Int64Array::from_iter_values(base..base + rows)); + let v: ArrayRef = Arc::new(StringArray::from_iter_values( + (base..base + rows).map(|k| format!("payload-{k:08}")), + )); + let batch = RecordBatch::try_new(Arc::clone(&schema), vec![k, v]).unwrap(); + let mut buf = vec![]; + let mut writer = + ArrowWriter::try_new(&mut buf, Arc::clone(&schema), None).unwrap(); + writer.write(&batch).unwrap(); + writer.close().unwrap(); + let bytes = Bytes::from(buf); + + let metadata = ParquetMetaDataReader::new() + .parse_and_finish(&bytes) + .unwrap(); + assert_eq!(metadata.num_row_groups(), 1); + let (end, data_bytes) = metadata.row_group(0).columns().iter().fold( + (0, 0), + |(end, total), column| { + let (start, len) = column.byte_range(); + (end.max(start + len), total + len) + }, + ); + + let path = Path::from(format!("data/file_{i:02}.parquet")); + inner.put(&path, bytes.into()).await.unwrap(); + data_end.insert(path.clone(), end); + files.push((path, data_bytes)); + } + let store = Arc::new(GatedStore { + inner, + data_end, + data_reads: Mutex::new(vec![]), + }); + Self { store, files } + } + + fn data_bytes(&self, path: &Path) -> u64 { + self.files + .iter() + .find_map(|(p, bytes)| (p == path).then_some(*bytes)) + .unwrap() + } +} + +/// What one TopK run read and recorded. +#[derive(Debug)] +struct RunOutcome { + batches: Vec, + /// Files whose row group was fetched (each file has a single row group). + fetched_files: BTreeSet, + /// Number of data reads issued (one per fetched row group). + data_reads: usize, + bytes_scanned: usize, + /// `(pruned, matched)` of `files_ranges_pruned_statistics`. + files_ranges: (usize, usize), + /// `(pruned, matched)` of `row_groups_pruned_statistics`. + row_groups: (usize, usize), + /// `files_ranges_pruned_before_read`; `None` when not registered. + pruned_before_read: Option, + /// `(ranges, empty slivers)` of the scan's file groups. + ranges: (usize, usize), +} + +fn pruning(metrics: &MetricsSet, name: &str) -> (usize, usize) { + match metrics.sum_by_name(name) { + Some(MetricValue::PruningMetrics { + pruning_metrics, .. + }) => (pruning_metrics.pruned(), pruning_metrics.matched()), + other => panic!("unexpected {name}: {other:?}"), + } +} + +/// Count the byte ranges of the scan and how many of them are empty slivers +/// (the tail of a split single-row-group file, which holds no row group). +fn file_ranges(plan: &Arc) -> (usize, usize) { + let mut found = None; + plan.apply(|node| { + if let Some(exec) = node.downcast_ref::() + && let Some(config) = exec.data_source().downcast_ref::() + { + let files = config.file_groups.iter().flat_map(|g| g.iter()); + let (mut ranges, mut slivers) = (0, 0); + for file in files { + ranges += 1; + // Each file's only row group starts right after the 4-byte + // magic, so every range that starts later is empty. + if file.range.as_ref().is_some_and(|r| r.start > 4) { + slivers += 1; + } + } + found = Some((ranges, slivers)); + } + Ok(TreeNodeRecursion::Continue) + }) + .unwrap(); + found.expect("plan has a file scan") +} + +/// Run `sql` with open-ahead on, on `target_partitions` partitions. +/// +/// File scans are only split into byte ranges when the table has more rows +/// than `batch_size`, so tests that need empty slivers pass a small one. +async fn run_topk( + files: &TestFiles, + target_partitions: usize, + batch_size: usize, + recheck: bool, + sql: &str, +) -> RunOutcome { + let mut config = SessionConfig::new() + .with_target_partitions(target_partitions) + .with_batch_size(batch_size); + let options = config.options_mut(); + options.execution.file_stream_open_ahead = true; + options.execution.recheck_dynamic_filter_before_read = recheck; + // Split files into byte ranges even though they are small. + options.optimizer.repartition_file_min_size = 0; + let ctx = SessionContext::new_with_config(config); + let url = Url::parse("gated://bucket").unwrap(); + ctx.register_object_store(&url, Arc::clone(&files.store) as _); + ctx.register_parquet("t", "gated://bucket/data/", ParquetReadOptions::default()) + .await + .unwrap(); + + let plan = ctx + .sql(sql) + .await + .unwrap() + .create_physical_plan() + .await + .unwrap(); + let ranges = file_ranges(&plan); + assert!( + files.store.take_data_reads().is_empty(), + "planning must not read row-group data" + ); + let batches = collect(Arc::clone(&plan), ctx.task_ctx()).await.unwrap(); + let data_reads = files.store.take_data_reads(); + + let metrics = MetricsFinder::find_metrics(plan.as_ref()) + .unwrap() + .aggregate_by_name(); + RunOutcome { + batches, + fetched_files: data_reads.iter().cloned().collect(), + data_reads: data_reads.len(), + bytes_scanned: metrics.sum_by_name("bytes_scanned").unwrap().as_usize(), + files_ranges: pruning(&metrics, "files_ranges_pruned_statistics"), + row_groups: pruning(&metrics, "row_groups_pruned_statistics"), + pruned_before_read: metrics + .sum_by_name("files_ranges_pruned_before_read") + .map(|m| m.as_usize()), + ranges, + } +} + +const TOPK_SQL: &str = "SELECT * FROM t ORDER BY k LIMIT 3"; + +fn assert_topk_result(batches: &[RecordBatch]) { + assert_batches_eq!( + [ + "+---+------------------+", + "| k | v |", + "+---+------------------+", + "| 0 | payload-00000000 |", + "| 1 | payload-00000001 |", + "| 2 | payload-00000002 |", + "+---+------------------+", + ], + batches + ); +} + +/// One partition with open-ahead: while `file_00` is read, `file_01` is +/// claimed and planned against the still-empty filter. The TopK fills from +/// `file_00`'s first batch, and the tightened filter excludes every other +/// file. Without the re-check `file_01`'s row group is fetched before the +/// early stop drops it; with the re-check it is never fetched. The remaining +/// files are claimed after the update and pruned at open in both runs. +#[tokio::test] +async fn recheck_skips_open_ahead_file_single_partition() { + let files = TestFiles::new(&[100; 6]).await; + let file = |i: usize| files.files[i].0.clone(); + + let off = run_topk(&files, 1, 8192, false, TOPK_SQL).await; + let on = run_topk(&files, 1, 8192, true, TOPK_SQL).await; + assert_topk_result(&off.batches); + assert_topk_result(&on.batches); + assert_eq!(off.ranges, (6, 0)); + assert_eq!(on.ranges, (6, 0)); + + // Without the re-check: two row groups fetched, the open-ahead file is + // stopped after its first batch. + assert_eq!( + off.fetched_files, + BTreeSet::from([file(0), file(1)]), + "{off:?}" + ); + assert_eq!(off.data_reads, 2, "{off:?}"); + assert_eq!( + off.bytes_scanned as u64, + files.data_bytes(&file(0)) + files.data_bytes(&file(1)) + ); + assert_eq!( + off.pruned_before_read, None, + "metric must not be registered" + ); + + // With the re-check: only `file_00` is fetched. + assert_eq!(on.fetched_files, BTreeSet::from([file(0)]), "{on:?}"); + assert_eq!(on.data_reads, 1, "{on:?}"); + assert_eq!(on.bytes_scanned as u64, files.data_bytes(&file(0))); + assert_eq!(on.pruned_before_read, Some(1), "{on:?}"); + + // Both runs end with only `file_00` matched: a file stopped before its + // first read is counted like one stopped after its first batch. + assert_eq!(off.files_ranges, (5, 1), "{off:?}"); + assert_eq!(on.files_ranges, (5, 1), "{on:?}"); + // Both open-ahead candidates were planned against the empty filter. + assert_eq!(off.row_groups, (0, 2), "{off:?}"); + assert_eq!(on.row_groups, (0, 2), "{on:?}"); +} + +/// Five partitions over ten files of slightly different sizes: byte-range +/// splitting turns them into fourteen ranges, four of which are empty slivers +/// (the tails of split files). Every partition claims its first range, opens +/// a second one ahead, and partitions whose range was a sliver claim again, +/// all before the first TopK update. Without the re-check every claimed file +/// whose row group survived that empty filter is fetched; with it, only the +/// file each partition was already reading is fetched, and the files opened +/// ahead are skipped before their first read. +/// +/// On the single-threaded test runtime this currently fetches 9 of the 10 +/// row groups without the re-check and 5 with it; the assertions below check +/// the properties that hold for any interleaving. +#[tokio::test] +async fn recheck_skips_open_ahead_files_with_empty_slivers() { + let rows: Vec = (0..10).map(|i| 100 + 13 * i).collect(); + let files = TestFiles::new(&rows).await; + + let off = run_topk(&files, 5, 100, false, TOPK_SQL).await; + let on = run_topk(&files, 5, 100, true, TOPK_SQL).await; + assert_topk_result(&off.batches); + assert_topk_result(&on.batches); + assert_eq!(off.ranges, (14, 4)); + assert_eq!(on.ranges, (14, 4)); + + // Planning before the first TopK update is the same in both runs. + assert_eq!(off.row_groups, on.row_groups); + // Without the re-check every row group that survived the empty filter is + // fetched. + assert_eq!(off.fetched_files.len(), off.row_groups.1, "{off:?}"); + + // Each fetched file is fetched once, and `bytes_scanned` is exactly the + // row-group data of the fetched files. + for outcome in [&off, &on] { + assert_eq!( + outcome.data_reads, + outcome.fetched_files.len(), + "{outcome:?}" + ); + let fetched_bytes: u64 = outcome + .fetched_files + .iter() + .map(|p| files.data_bytes(p)) + .sum(); + assert_eq!(outcome.bytes_scanned as u64, fetched_bytes, "{outcome:?}"); + let (pruned, matched) = outcome.files_ranges; + assert_eq!(pruned + matched, 14, "{outcome:?}"); + } + + // With the re-check at most one file per partition is fetched: the one it + // was reading when the filter was first updated. + assert!(on.fetched_files.len() <= 5, "{on:?}"); + assert!( + on.fetched_files.len() < off.fetched_files.len(), + "re-check must skip files opened ahead\noff: {off:?}\non: {on:?}" + ); + assert!(on.fetched_files.is_subset(&off.fetched_files)); + assert!(on.bytes_scanned < off.bytes_scanned); + + // Every file fetched without the re-check but not with it was skipped by + // the re-check; empty slivers opened ahead may be skipped too. + let skipped_real = off.fetched_files.len() - on.fetched_files.len(); + let pruned_before_read = on.pruned_before_read.expect("metric registered"); + assert!(pruned_before_read >= skipped_real, "{on:?}"); + assert!(pruned_before_read <= skipped_real + 4, "{on:?}"); + assert_eq!(off.pruned_before_read, None); + assert!(on.files_ranges.1 <= off.files_ranges.1); +} + +/// With the option on, scans without a dynamic filter keep their metrics and +/// I/O unchanged: the new counter is not registered and nothing is skipped. +#[tokio::test] +async fn recheck_does_not_touch_scans_without_dynamic_filter() { + let files = TestFiles::new(&[100; 4]).await; + for sql in [ + "SELECT * FROM t WHERE k >= 2000", + "SELECT * FROM t", + "SELECT * FROM t ORDER BY k", + ] { + let off = run_topk(&files, 1, 8192, false, sql).await; + let on = run_topk(&files, 1, 8192, true, sql).await; + assert_eq!(on.pruned_before_read, None, "{sql}"); + assert_eq!(off.pruned_before_read, None, "{sql}"); + assert_eq!(on.fetched_files, off.fetched_files, "{sql}"); + assert_eq!(on.bytes_scanned, off.bytes_scanned, "{sql}"); + assert_eq!(on.files_ranges, off.files_ranges, "{sql}"); + assert_eq!(on.batches, off.batches, "{sql}"); + } +} diff --git a/datafusion/datasource-parquet/src/opener/early_stop.rs b/datafusion/datasource-parquet/src/opener/early_stop.rs index 75749d284068b..3fafb8e6932e9 100644 --- a/datafusion/datasource-parquet/src/opener/early_stop.rs +++ b/datafusion/datasource-parquet/src/opener/early_stop.rs @@ -23,7 +23,7 @@ use std::task::{Context, Poll}; use arrow::array::RecordBatch; use datafusion_common::Result; -use datafusion_physical_plan::metrics::PruningMetrics; +use datafusion_physical_plan::metrics::{Count, PruningMetrics}; use datafusion_pruning::FilePruner; use futures::{Stream, StreamExt, ready}; @@ -32,12 +32,23 @@ use futures::{Stream, StreamExt, ready}; /// This can terminate the scan early when some dynamic filters is updated after /// the scan starts, so we discover after the scan starts that the file can be /// pruned (can't have matching rows). +/// +/// The pruner is consulted after every batch. With +/// [`Self::with_recheck_before_first_read`] it is also consulted once before +/// the inner stream is first polled, i.e. before the file's first data read: +/// a file that was opened while the filter was still loose (for example ahead +/// of time, by `datafusion.execution.file_stream_open_ahead`) is then skipped +/// without fetching or decoding any row group when the filter has tightened +/// since the file was opened. pub(super) struct EarlyStoppingStream { /// Has the stream finished processing? All subsequent polls will return /// None done: bool, file_pruner: FilePruner, files_ranges_pruned_statistics: PruningMetrics, + /// `Some` until the pre-read check has run, when that check is enabled. + /// Counts the file ranges it skipped. + recheck_before_first_read: Option, /// The inner stream inner: S, } @@ -53,8 +64,31 @@ impl EarlyStoppingStream { inner: stream, file_pruner, files_ranges_pruned_statistics, + recheck_before_first_read: None, } } + + /// Also check the pruner once before the first poll of the inner stream, + /// counting the file ranges skipped this way in `files_pruned`. + /// + /// The pruner was already evaluated when the file was opened, so this + /// check only rebuilds the pruning predicate when a watched dynamic filter + /// has been updated since; otherwise it costs one atomic load per filter. + pub(super) fn with_recheck_before_first_read( + mut self, + files_pruned: Option, + ) -> Self { + self.recheck_before_first_read = files_pruned; + self + } + + /// Stop the scan because the file can no longer contain matching rows. + fn stop_pruned(&mut self) { + self.files_ranges_pruned_statistics.add_pruned(1); + // Previously this file range has been counted as matched + self.files_ranges_pruned_statistics.subtract_matched(1); + self.done = true; + } } impl EarlyStoppingStream @@ -67,16 +101,29 @@ where // Since dynamic filters may have been updated, see if we can stop // reading this stream entirely. if self.file_pruner.should_prune()? { - self.files_ranges_pruned_statistics.add_pruned(1); - // Previously this file range has been counted as matched - self.files_ranges_pruned_statistics.subtract_matched(1); - self.done = true; + self.stop_pruned(); Ok(None) } else { // Return the adapted batch Ok(Some(batch)) } } + + /// Runs the optional check before the inner stream is first polled. + /// Returns `true` when the file was pruned and the stream is done. + fn check_prune_before_first_read(&mut self) -> Result { + let Some(files_pruned) = self.recheck_before_first_read.take() else { + return Ok(false); + }; + if self.file_pruner.should_prune()? { + // Counted exactly like a file stopped after its first batch, so + // `files_ranges_pruned_statistics` is the same either way. + self.stop_pruned(); + files_pruned.add(1); + return Ok(true); + } + Ok(false) + } } impl Stream for EarlyStoppingStream @@ -92,6 +139,11 @@ where if self.done { return Poll::Ready(None); } + match self.check_prune_before_first_read() { + Ok(true) => return Poll::Ready(None), + Ok(false) => {} + Err(e) => return Poll::Ready(Some(Err(e))), + } match ready!(self.inner.poll_next_unpin(cx)) { None => { // input done diff --git a/datafusion/datasource-parquet/src/opener/mod.rs b/datafusion/datasource-parquet/src/opener/mod.rs index 693e9bd2cbf31..a9f4662f23c94 100644 --- a/datafusion/datasource-parquet/src/opener/mod.rs +++ b/datafusion/datasource-parquet/src/opener/mod.rs @@ -301,6 +301,12 @@ pub(super) struct ParquetMorselizer { /// Per-scan virtual-column state (validation already performed). `None` /// when no virtual columns are requested — the common path. pub(crate) virtual_state: Option>, + /// Counter for file ranges skipped by the dynamic-filter re-check that + /// runs right before a file's first data read. `Some` enables that + /// re-check (`datafusion.execution.recheck_dynamic_filter_before_read` + /// with a dynamic filter in the predicate); `None` keeps the default + /// behaviour of re-checking only after the first batch. + pub(crate) files_ranges_pruned_before_read: Option, } impl fmt::Debug for ParquetMorselizer { @@ -460,6 +466,8 @@ struct PreparedParquetOpen { reverse_row_groups: bool, sort_order_for_reorder: Option, preserve_order: bool, + /// See [`ParquetMorselizer::files_ranges_pruned_before_read`]. + files_ranges_pruned_before_read: Option, #[cfg(feature = "parquet_encryption")] file_decryption_properties: Option>, } @@ -857,6 +865,7 @@ impl ParquetMorselizer { reverse_row_groups: self.reverse_row_groups, sort_order_for_reorder: self.sort_order_for_reorder.clone(), preserve_order: self.preserve_order, + files_ranges_pruned_before_read: self.files_ranges_pruned_before_read.clone(), #[cfg(feature = "parquet_encryption")] file_decryption_properties: None, }) @@ -1544,6 +1553,16 @@ impl RowGroupsPrunedParquetOpen { .file_metrics .row_groups_pruned_dynamic_filter .clone(); + // The pre-read re-check only saves I/O for a range that has a row group + // left to fetch and no `RowGroupPruner` (which already checks before every + // row group): an empty byte-range sliver or a fully pruned range would + // otherwise be counted as pruned before read without saving anything. + let files_ranges_pruned_before_read = + if !rg_plan.is_empty() && row_group_pruner.is_none() { + prepared.files_ranges_pruned_before_read + } else { + None + }; let stream = PushDecoderStreamState { decoder: Some(decoder), @@ -1565,6 +1584,11 @@ impl RowGroupsPrunedParquetOpen { // mid-scan. For a static (or already-complete) predicate the up-front // `prune_file` check already captured everything that can be pruned, so // per-batch re-checking would only add overhead. + // + // With `files_ranges_pruned_before_read` set, the wrapper also re-checks + // once before the first data read, so a file opened ahead of time is + // skipped without any row-group I/O when the filter has tightened since + // `prune_file` ran. match prepared.file_pruner { Some(file_pruner) if file_pruner.is_watching() => { Ok(EarlyStoppingStream::new( @@ -1572,6 +1596,7 @@ impl RowGroupsPrunedParquetOpen { file_pruner, files_ranges_pruned_statistics, ) + .with_recheck_before_first_read(files_ranges_pruned_before_read) .boxed()) } _ => Ok(stream), @@ -1809,6 +1834,7 @@ mod test { max_in_list_size: usize, reverse_row_groups: bool, preserve_order: bool, + files_ranges_pruned_before_read: Option, } #[test] @@ -2020,6 +2046,7 @@ mod test { max_in_list_size: MAX_IN_LIST_SIZE, reverse_row_groups: false, preserve_order: false, + files_ranges_pruned_before_read: None, } } @@ -2118,6 +2145,13 @@ mod test { self } + /// Enable the dynamic-filter re-check before a file's first data + /// read, counting skipped files in `counter`. + fn with_recheck_before_first_read(mut self, counter: Count) -> Self { + self.files_ranges_pruned_before_read = Some(counter); + self + } + /// Build the ParquetMorselizer instance, unwrapping validation errors. /// /// # Panics @@ -2199,6 +2233,7 @@ mod test { reverse_row_groups: self.reverse_row_groups, sort_order_for_reorder: None, virtual_state, + files_ranges_pruned_before_read: self.files_ranges_pruned_before_read, }) } } @@ -3440,6 +3475,225 @@ mod test { assert_eq!(values, vec![7, 4, 5, 6, 3]); } + /// [`ParquetFileReaderFactory`] that counts data reads (`get_bytes` and + /// `get_byte_ranges`) separately from metadata reads (`get_metadata`). + #[derive(Debug)] + struct DataReadCountingReaderFactory { + inner: DefaultParquetFileReaderFactory, + data_reads: Arc, + } + + struct DataReadCountingReader { + inner: Box, + data_reads: Arc, + } + + impl AsyncFileReader for DataReadCountingReader { + fn get_bytes( + &mut self, + range: std::ops::Range, + ) -> BoxFuture<'_, parquet::errors::Result> { + self.data_reads + .fetch_add(1, std::sync::atomic::Ordering::SeqCst); + self.inner.get_bytes(range) + } + + fn get_byte_ranges( + &mut self, + ranges: Vec>, + ) -> BoxFuture<'_, parquet::errors::Result>> { + self.data_reads + .fetch_add(1, std::sync::atomic::Ordering::SeqCst); + self.inner.get_byte_ranges(ranges) + } + + fn get_metadata<'a>( + &'a mut self, + options: Option<&'a ArrowReaderOptions>, + ) -> BoxFuture<'a, parquet::errors::Result>> { + self.inner.get_metadata(options) + } + } + + impl ParquetFileReaderFactory for DataReadCountingReaderFactory { + fn create_reader( + &self, + partition_index: usize, + partitioned_file: PartitionedFile, + metadata_size_hint: Option, + metrics: &ExecutionPlanMetricsSet, + ) -> Result> { + let inner = self.inner.create_reader( + partition_index, + partitioned_file, + metadata_size_hint, + metrics, + )?; + Ok(Box::new(DataReadCountingReader { + inner, + data_reads: Arc::clone(&self.data_reads), + })) + } + } + + fn pruning_metric( + metrics: &ExecutionPlanMetricsSet, + name: &str, + ) -> Option<(usize, usize)> { + use datafusion_physical_plan::metrics::MetricValue; + metrics + .clone_inner() + .sum_by_name(name) + .and_then(|metric| match metric { + MetricValue::PruningMetrics { + pruning_metrics, .. + } => Some((pruning_metrics.pruned(), pruning_metrics.matched())), + _ => None, + }) + } + + /// Outcome of [`open_then_tighten_dynamic_filter`]. + struct TightenedOpen { + rows: usize, + data_reads_at_open: usize, + data_reads: usize, + /// `(pruned, matched)` of `files_ranges_pruned_statistics`. + files_ranges: Option<(usize, usize)>, + /// `files_ranges_pruned_before_read`, `None` when not registered. + pruned_before_read: Option, + } + + /// Open a one-row-group file (`a` in `1..=100`) under a TopK-style dynamic + /// filter that is still `true`, then update the filter to `new_filter` + /// before reading the opened stream. This is the shape of a file opened + /// ahead of time (`file_stream_open_ahead`) and only read after the TopK + /// published a threshold. + async fn open_then_tighten_dynamic_filter( + recheck_before_first_read: bool, + new_filter: Option, + ) -> TightenedOpen { + use datafusion_physical_plan::metrics::MetricBuilder; + + let store = Arc::new(InMemory::new()) as Arc; + let values: Vec = (1..=100).collect(); + let batch = record_batch!(( + "a", + Int32, + values.iter().map(|v| Some(*v)).collect::>() + )) + .unwrap(); + let schema = batch.schema(); + let data_len = + write_parquet(Arc::clone(&store), "test.parquet", batch.clone()).await; + let file = PartitionedFile::new( + "test.parquet".to_string(), + u64::try_from(data_len).unwrap(), + ) + .with_statistics(Arc::new(Statistics { + num_rows: Precision::Exact(100), + total_byte_size: Precision::Absent, + column_statistics: vec![ + ColumnStatistics::new_unknown() + .with_min_value(Precision::Exact(ScalarValue::Int32(Some(1)))) + .with_max_value(Precision::Exact(ScalarValue::Int32(Some(100)))) + .with_null_count(Precision::Exact(0)), + ], + })); + + let column = Arc::new(Column::new("a", 0)) as Arc; + let filter = Arc::new(DynamicFilterPhysicalExpr::new( + vec![Arc::clone(&column)], + datafusion_physical_expr::expressions::lit(true), + )); + let predicate = Arc::clone(&filter) as Arc; + + let metrics = ExecutionPlanMetricsSet::new(); + let data_reads = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let mut builder = ParquetMorselizerBuilder::new() + .with_store(Arc::clone(&store)) + .with_schema(Arc::clone(&schema)) + .with_predicate(predicate) + .with_row_group_stats_pruning(true) + .with_metrics(metrics.clone()) + .with_parquet_file_reader_factory(Arc::new(DataReadCountingReaderFactory { + inner: DefaultParquetFileReaderFactory::new(Arc::clone(&store)), + data_reads: Arc::clone(&data_reads), + })); + if recheck_before_first_read { + builder = builder.with_recheck_before_first_read( + MetricBuilder::new(&metrics) + .counter("files_ranges_pruned_before_read", 0), + ); + } + let morselizer = builder.build(); + + // Opening reads only metadata: the row group is fetched by the stream. + let stream = open_file(&morselizer, file).await.unwrap(); + let data_reads_at_open = data_reads.load(std::sync::atomic::Ordering::SeqCst); + + if let Some(new_filter) = new_filter { + filter + .update(logical2physical(&new_filter, &schema)) + .unwrap(); + } + let (_, rows) = count_batches_and_rows(stream).await; + + TightenedOpen { + rows, + data_reads_at_open, + data_reads: data_reads.load(std::sync::atomic::Ordering::SeqCst), + files_ranges: pruning_metric(&metrics, "files_ranges_pruned_statistics"), + pruned_before_read: metrics + .clone_inner() + .sum_by_name("files_ranges_pruned_before_read") + .map(|m| m.as_usize()), + } + } + + /// A file opened before the dynamic filter tightened, and excluded by the + /// tightened filter, must be skipped without any data read when the + /// pre-read re-check is enabled. + #[tokio::test] + async fn test_recheck_before_first_read_skips_data_reads() { + let result = + open_then_tighten_dynamic_filter(true, Some(col("a").lt(lit(0i32)))).await; + assert_eq!(result.rows, 0); + assert_eq!(result.data_reads_at_open, 0); + assert_eq!(result.data_reads, 0, "stopped file must not read data"); + // Counted like an early-stopped file: matched at open, then pruned. + assert_eq!(result.files_ranges, Some((1, 0))); + assert_eq!(result.pruned_before_read, Some(1)); + } + + /// Without the pre-read re-check the same file is only stopped after its + /// first batch, which requires fetching its row group. + #[tokio::test] + async fn test_without_recheck_stopped_file_reads_data_first() { + let result = + open_then_tighten_dynamic_filter(false, Some(col("a").lt(lit(0i32)))).await; + assert_eq!(result.rows, 0, "early stop drops the first batch"); + assert_eq!(result.data_reads_at_open, 0); + assert!( + result.data_reads > 0, + "the early stop fires only after the row group was fetched" + ); + assert_eq!(result.files_ranges, Some((1, 0))); + assert_eq!(result.pruned_before_read, None, "metric not registered"); + } + + /// The re-check must not skip a file that the current filter still admits, + /// whether or not the filter changed since the file was opened. + #[tokio::test] + async fn test_recheck_before_first_read_keeps_matching_files() { + for new_filter in [None, Some(col("a").lt(lit(50i32)))] { + let result = open_then_tighten_dynamic_filter(true, new_filter.clone()).await; + assert_eq!(result.rows, 100, "filter {new_filter:?}"); + assert!(result.data_reads > 0, "filter {new_filter:?}"); + assert_eq!(result.files_ranges, Some((0, 1)), "filter {new_filter:?}"); + assert_eq!(result.pruned_before_read, Some(0), "filter {new_filter:?}"); + } + } + /// Helpers for tests that exercise parquet virtual columns /// (e.g. `row_number`) plumbed through `TableSchema`/`ParquetOpener`. mod virtual_columns { diff --git a/datafusion/datasource-parquet/src/source.rs b/datafusion/datasource-parquet/src/source.rs index 097b4563af5df..88cefc07d5eba 100644 --- a/datafusion/datasource-parquet/src/source.rs +++ b/datafusion/datasource-parquet/src/source.rs @@ -29,9 +29,9 @@ use crate::row_filter::can_expr_be_pushed_down_with_schemas; use arrow_schema::Fields; use arrow_schema::extension::ExtensionType; use arrow_schema::{DataType, Field}; -use datafusion_common::config::ConfigOptions; #[cfg(feature = "parquet_encryption")] use datafusion_common::config::EncryptionFactoryOptions; +use datafusion_common::config::{ConfigOptions, ExecutionOptions}; use datafusion_datasource::as_file_source; use datafusion_datasource::file_stream::FileOpener; use datafusion_datasource::morsel::Morselizer; @@ -62,6 +62,7 @@ use datafusion_physical_plan::filter_pushdown::{ }; use datafusion_physical_plan::metrics::Count; use datafusion_physical_plan::metrics::ExecutionPlanMetricsSet; +use datafusion_physical_plan::metrics::{MetricBuilder, MetricType}; use log::warn; #[cfg(feature = "parquet_encryption")] @@ -314,6 +315,11 @@ pub struct ParquetSource { /// Sort order driving `PreparedAccessPlan::reorder_by_statistics` /// in the opener. sort_order_for_reorder: Option, + /// Re-check a still-changing dynamic filter against each file's + /// statistics right before the file's first data read. Set at open time + /// from `datafusion.execution.recheck_dynamic_filter_before_read` via + /// [`FileSource::with_execution_options`]. + recheck_dynamic_filter_before_read: bool, } impl ParquetSource { @@ -340,6 +346,7 @@ impl ParquetSource { encryption_factory: None, reverse_row_groups: false, sort_order_for_reorder: None, + recheck_dynamic_filter_before_read: false, } } @@ -629,6 +636,19 @@ impl FileSource for ParquetSource { self.pushdown_filters(), )?; + // Registered only when the pre-read re-check is enabled and the + // predicate carries a dynamic filter, so EXPLAIN ANALYZE output of + // every other scan is unchanged. `Some` also switches the re-check on. + let files_ranges_pruned_before_read = (self.recheck_dynamic_filter_before_read + && self.predicate.as_ref().is_some_and(|predicate| { + DynamicFilterTracking::classify(predicate).contains_dynamic_filter() + })) + .then(|| { + MetricBuilder::new(self.metrics()) + .with_type(MetricType::Summary) + .counter("files_ranges_pruned_before_read", partition) + }); + Ok(Box::new(ParquetMorselizer { partition_index: partition, projection: self.projection.clone(), @@ -660,6 +680,7 @@ impl FileSource for ParquetSource { reverse_row_groups: self.reverse_row_groups, sort_order_for_reorder: self.sort_order_for_reorder.clone(), virtual_state, + files_ranges_pruned_before_read, })) } @@ -689,6 +710,19 @@ impl FileSource for ParquetSource { Arc::new(conf) } + fn with_execution_options( + &self, + options: &ExecutionOptions, + ) -> Option> { + let recheck = options.recheck_dynamic_filter_before_read; + if self.recheck_dynamic_filter_before_read == recheck { + return None; + } + let mut conf = self.clone(); + conf.recheck_dynamic_filter_before_read = recheck; + Some(Arc::new(conf)) + } + fn try_pushdown_projection( &self, projection: &ProjectionExprs, diff --git a/datafusion/datasource/src/file.rs b/datafusion/datasource/src/file.rs index f1a94f2e12363..495220e30fc55 100644 --- a/datafusion/datasource/src/file.rs +++ b/datafusion/datasource/src/file.rs @@ -28,7 +28,7 @@ use crate::file_stream::FileOpener; use crate::morsel::{FileOpenerMorselizer, Morselizer}; #[expect(deprecated)] use crate::schema_adapter::SchemaAdapterFactory; -use datafusion_common::config::ConfigOptions; +use datafusion_common::config::{ConfigOptions, ExecutionOptions}; use datafusion_common::tree_node::TreeNodeRecursion; use datafusion_common::{Result, not_impl_err}; use datafusion_physical_expr::projection::ProjectionExprs; @@ -107,6 +107,27 @@ pub trait FileSource: Any + Send + Sync { /// Initialize new type with batch size configuration fn with_batch_size(&self, batch_size: usize) -> Arc; + /// Returns a copy of this source that applies the execution-time settings + /// in `options`, or `None` when the source reads none of them (or already + /// has the same values), in which case it is used unchanged. + /// + /// [`FileScanConfig`] calls this each time it opens a scan partition, + /// passing the session's [`ExecutionOptions`] from the `TaskContext`. + /// Settings that only matter while a scan runs therefore reach the source + /// even when it was built without the session configuration, for example + /// by a custom `TableProvider` that constructs it with default options. + /// Implementations should only read settings that are not already part of + /// their own format options, so per-table configuration is not + /// overridden. + /// + /// The default implementation returns `None`. + fn with_execution_options( + &self, + _options: &ExecutionOptions, + ) -> Option> { + None + } + /// Returns the filter expression that will be applied *during* the file scan. /// /// These expressions are in terms of the unprojected [`Self::table_schema`]. diff --git a/datafusion/datasource/src/file_scan_config/mod.rs b/datafusion/datasource/src/file_scan_config/mod.rs index 8a4687e7450d1..f2c9c1368aff9 100644 --- a/datafusion/datasource/src/file_scan_config/mod.rs +++ b/datafusion/datasource/src/file_scan_config/mod.rs @@ -709,6 +709,11 @@ impl DataSource for FileScanConfig { .unwrap_or_else(|| context.session_config().batch_size()); let source = self.file_source.with_batch_size(batch_size); + // Apply execution-time settings from the session. Reading them here, + // rather than when the source is built, also covers sources that a + // custom table provider constructed with default options. + let execution = &context.session_config().options().execution; + let source = source.with_execution_options(execution).unwrap_or(source); let morselizer = source.create_morselizer(object_store, self, partition)?; @@ -720,11 +725,7 @@ impl DataSource for FileScanConfig { .and_then(|state| state.downcast_ref::()) .cloned(); - let open_ahead = context - .session_config() - .options() - .execution - .file_stream_open_ahead; + let open_ahead = execution.file_stream_open_ahead; let stream = FileStreamBuilder::new(self) .with_partition(partition) diff --git a/datafusion/sqllogictest/test_files/information_schema.slt b/datafusion/sqllogictest/test_files/information_schema.slt index 27a3fb00a35c3..4af466ae3970c 100644 --- a/datafusion/sqllogictest/test_files/information_schema.slt +++ b/datafusion/sqllogictest/test_files/information_schema.slt @@ -278,6 +278,7 @@ datafusion.execution.parquet.writer_version 1.0 datafusion.execution.perfect_hash_join_min_key_density 0.15 datafusion.execution.perfect_hash_join_small_build_threshold 1024 datafusion.execution.planning_concurrency 13 +datafusion.execution.recheck_dynamic_filter_before_read false datafusion.execution.skip_partial_aggregation_probe_ratio_threshold 0.8 datafusion.execution.skip_partial_aggregation_probe_rows_threshold 100000 datafusion.execution.skip_physical_aggregate_schema_check false @@ -445,6 +446,7 @@ datafusion.execution.parquet.writer_version 1.0 (writing) Sets parquet writer ve datafusion.execution.perfect_hash_join_min_key_density 0.15 The minimum required density of join keys on the build side to consider a perfect hash join (see `HashJoinExec` for more details). Density is calculated as: `(number of rows) / (max_key - min_key + 1)`. A perfect hash join may be used if the actual key density > this value. Currently only supports cases where build_side.num_rows() < u32::MAX. Support for build_side.num_rows() >= u32::MAX will be added in the future. datafusion.execution.perfect_hash_join_small_build_threshold 1024 A perfect hash join (see `HashJoinExec` for more details) will be considered if the range of keys (max - min) on the build side is < this threshold. This provides a fast path for joins with very small key ranges, bypassing the density check. Currently only supports cases where build_side.num_rows() < u32::MAX. Support for build_side.num_rows() >= u32::MAX will be added in the future. datafusion.execution.planning_concurrency 13 Fan-out during initial physical planning. This is mostly use to plan `UNION` children in parallel. Defaults to the number of CPU cores on the system +datafusion.execution.recheck_dynamic_filter_before_read false When `true`, a Parquet scan whose predicate holds a dynamic filter that can still change (for example the threshold pushed down by an `ORDER BY ... LIMIT` TopK) re-checks the file's statistics against the current filter right before reading the file's first data pages. A file range that was opened early (by `file_stream_open_ahead`, or while the filter was still empty) is then skipped without fetching or decoding its row groups if the filter has since been updated to exclude it. The check reuses the scan's per-batch early-stop pruner, so the pruning predicate is only rebuilt when the filter changed. Skipped ranges are counted by the `files_ranges_pruned_before_read` metric, which is registered only for scans with a dynamic filter while this option is enabled. The check uses the file's `PartitionedFile` statistics, so it does nothing for files without them, and it applies to every dynamic filter that can still change (TopK thresholds and aggregate MIN/MAX bounds; hash-join filters are complete before the probe side runs). When `false` (the default) a single-row-group file is re-checked only after its first batch; a file with several row groups is already re-checked before each row group. datafusion.execution.skip_partial_aggregation_probe_ratio_threshold 0.8 Aggregation ratio (number of distinct groups / number of input rows) threshold for skipping partial aggregation. If the value is greater then partial aggregation will skip aggregation for further input datafusion.execution.skip_partial_aggregation_probe_rows_threshold 100000 Number of input rows partial aggregation partition should process, before aggregation ratio check and trying to switch to skipping aggregation mode datafusion.execution.skip_physical_aggregate_schema_check false When set to true, skips verifying that the schema produced by planning the input of `LogicalPlan::Aggregate` exactly matches the schema of the input plan. When set to false, if the schema does not match exactly (including nullability and metadata), a planning error will be raised. This is used to workaround bugs in the planner that are now caught by the new schema verification step. diff --git a/docs/source/user-guide/configs.md b/docs/source/user-guide/configs.md index 96f6972c1051e..fd5cd74232489 100644 --- a/docs/source/user-guide/configs.md +++ b/docs/source/user-guide/configs.md @@ -142,6 +142,7 @@ The following configuration settings are available: | datafusion.execution.keep_partition_by_columns | false | Should DataFusion keep the columns used for partition_by in the output RecordBatches | | datafusion.execution.enable_file_stream_work_stealing | true | When `true` (the default), DataFusion's built-in file scans dynamically rebalance files across partitions at query execution time: a partition that goes idle reads files (or byte-range morsels) originally assigned to a sibling partition, which keeps all partitions busy in a single process. Executors that depend on the plan-time partition assignment — such as Ballista and datafusion-distributed, which run each partition as an isolated task and never poll the siblings — should set this to `false` so each partition reads only its own file group and no runtime reassignment occurs. | | datafusion.execution.file_stream_open_ahead | false | When `true`, each file-scan partition starts opening its next file (footer, page index and bloom filter I/O) while the current file is still being scanned, keeping at most one file in flight ahead of the active reader. This overlaps per-file open latency with data reads, which matters for scans over many files on high-latency object stores. The cost is holding one extra file's metadata per partition. When `false` (the default) a partition opens the next file only after the current one is fully consumed. | +| datafusion.execution.recheck_dynamic_filter_before_read | false | When `true`, a Parquet scan whose predicate holds a dynamic filter that can still change (for example the threshold pushed down by an `ORDER BY ... LIMIT` TopK) re-checks the file's statistics against the current filter right before reading the file's first data pages. A file range that was opened early (by `file_stream_open_ahead`, or while the filter was still empty) is then skipped without fetching or decoding its row groups if the filter has since been updated to exclude it. The check reuses the scan's per-batch early-stop pruner, so the pruning predicate is only rebuilt when the filter changed. Skipped ranges are counted by the `files_ranges_pruned_before_read` metric, which is registered only for scans with a dynamic filter while this option is enabled. The check uses the file's `PartitionedFile` statistics, so it does nothing for files without them, and it applies to every dynamic filter that can still change (TopK thresholds and aggregate MIN/MAX bounds; hash-join filters are complete before the probe side runs). When `false` (the default) a single-row-group file is re-checked only after its first batch; a file with several row groups is already re-checked before each row group. | | datafusion.execution.skip_partial_aggregation_probe_ratio_threshold | 0.8 | Aggregation ratio (number of distinct groups / number of input rows) threshold for skipping partial aggregation. If the value is greater then partial aggregation will skip aggregation for further input | | datafusion.execution.skip_partial_aggregation_probe_rows_threshold | 100000 | Number of input rows partial aggregation partition should process, before aggregation ratio check and trying to switch to skipping aggregation mode | | datafusion.execution.use_row_number_estimates_to_optimize_partitioning | false | Should DataFusion use row number estimates at the input to decide whether increasing parallelism is beneficial or not. By default, only exact row numbers (not estimates) are used for this decision. Setting this flag to `true` will likely produce better plans. if the source of statistics is accurate. We plan to make this the default in the future. |