diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs index 347721c43d7cf..2a0e55953fe6e 100644 --- a/datafusion/common/src/config.rs +++ b/datafusion/common/src/config.rs @@ -1137,6 +1137,17 @@ config_namespace! { /// tables with a highly-selective join filter, but is also slightly slower. pub enforce_batch_size_in_joins: bool, default = false + /// (experimental) When enabled, `FilterExec` measures the selectivity + /// and evaluation cost of each conjunct of an `AND` predicate at + /// runtime and reorders them to run the ones that discard the most + /// rows per unit of CPU time first. Query results never change, but + /// the observable side effects of a fallible predicate can, in either + /// direction: reordering `b <> 0 AND 1/b > 2` can make a + /// divide-by-zero error appear or disappear, since each conjunct is + /// evaluated only on the rows the conjuncts before it kept. Predicates + /// containing volatile expressions are never reordered. + pub adaptive_filter_reordering: bool, default = false + /// Size (bytes) of data buffer DataFusion uses when writing output files. /// This affects the size of the data chunks that are uploaded to remote /// object stores (e.g. AWS S3). If very large (>= 100 GiB) output files are being diff --git a/datafusion/physical-plan/src/adaptive_filter.rs b/datafusion/physical-plan/src/adaptive_filter.rs new file mode 100644 index 0000000000000..359c65b8a764e --- /dev/null +++ b/datafusion/physical-plan/src/adaptive_filter.rs @@ -0,0 +1,1216 @@ +// 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. + +//! Runtime-adaptive evaluation of a conjunctive (`AND`) predicate in +//! [`FilterExec`](crate::filter::FilterExec). +//! +//! Evaluation order matters: a selective conjunct run first gates the work of +//! the conjuncts after it. Two mechanisms already order and gate conjuncts +//! before this module sees them, and both decide statically: +//! +//! - the logical optimizer's `reorder_predicates` pass sorts the conjuncts +//! cheap-before-expensive by a static cost class +//! (). It is blind to +//! selectivity, so a cheap-but-unselective conjunct still sorts ahead of an +//! expensive-but-very-selective one, and conjuncts in the same cost class +//! keep the order they were written in. That is the order this module calls +//! the *written order* and measures against. +//! - [`BinaryExpr`]'s `AND` pre-selects: when the conjuncts evaluated so far +//! keep at most 20% of the rows and produce no nulls, it filters the batch +//! down to those rows before evaluating the next conjunct. It can only gate +//! a conjunct on the conjuncts written *before* it, never on a more +//! selective one written after it. +//! +//! This module measures each conjunct's selectivity and cost at runtime and +//! reorders them accordingly, so that pre-selection fires on the conjunct that +//! discards the most rows. Whether it runs at all is controlled by +//! `datafusion.execution.adaptive_filter_reordering`. For example: +//! +//! ```sql +//! WHERE regexp_like(s,'a') AND regexp_like(s,'b') AND regexp_like(s,'rare') +//! ``` +//! +//! All three conjuncts are equally expensive to the static cost class, so they +//! reach `FilterExec` as written, and the first two each keep most rows, so +//! `AND` pre-selection never fires. Once the warm-up has measured the three, +//! the selective one is promoted and the batch is compacted behind it: +//! +//! ```text +//! before: regexp_like(s,'a') evaluated on every row +//! regexp_like(s,'b') evaluated on every row +//! regexp_like(s,'rare') evaluated on every row +//! +//! after: regexp_like(s,'rare') every row, keeps ~1% -> batch compacted +//! regexp_like(s,'a') evaluated on those survivors only +//! regexp_like(s,'b') evaluated on those survivors only +//! ``` +//! +//! This module contains no evaluation logic of its own. While the order is +//! being learned, the written order is handed to [`BinaryExpr`] with every +//! conjunct wrapped in a [`MeasuredConjunct`]; `BinaryExpr` evaluates and +//! pre-selects as it would for the plain predicate, so each conjunct is +//! measured on the population it would really see in that position. +//! +//! Once the order settles the wrappers are gone: the settled order — the +//! written one if the warm-up found nothing materially better, otherwise the +//! learned one — is materialised once as a right-nested `AND` chain, +//! `(c_first AND (c_second AND (... AND c_last)))`. Right-nesting is what makes +//! it pay: pre-selection filters the batch an `AND` is handed before evaluating +//! its right-hand side, so the survivors of the first conjunct stay compacted +//! for the rest of the chain, where a left-nested chain — what +//! [`conjunction`](datafusion_physical_expr::utils::conjunction) builds — would +//! re-filter the original batch at every level. +//! +//! The ranking key is rows discarded per nanosecond +//! ([`effectiveness`](ConjunctStats::effectiveness)), and the ranking is +//! adopted only if it is materially cheaper than the written order +//! ([`TIE_COST_FRACTION`]), so a conjunction that does not benefit carries none +//! of this machinery past the warm-up. The decision then stays fixed. +//! +//! A `FilterExec` is split across many partition streams, each seeing only a +//! slice of the data, so measurements are pooled into a shared +//! [`AdaptiveFilterShared`] and the streams learn as one: the first stream with +//! enough samples settles the order for all of them, and the rest adopt it on +//! their next batch instead of each re-paying the warm-up. Only unsettled +//! streams take the shared lock. +//! +//! ## Known limitations +//! +//! - Results never change (a conjunction's value does not depend on evaluation +//! order), but the side effects of fallible predicates can, in either +//! direction: a conjunct evaluated after a pre-selection sees only the rows +//! that survived, so an error the written order raises can disappear and one +//! it avoided can appear. Volatile predicates are never reordered. +//! - Measurements are conditional on the written order and, after a +//! pre-selection, taken on small batches whose per-row cost is inflated by +//! fixed overheads. Correlated conjuncts can be misjudged; the material-win +//! guard only makes adoption conservative. +//! - The decision is one-shot: a misjudged reorder, or drifting data, is kept +//! for the rest of the query. +//! +//! See . + +use std::fmt; +use std::fmt::Formatter; +use std::sync::Arc; +use std::sync::Mutex; +use std::sync::atomic::{AtomicU64, Ordering::Relaxed}; + +use crate::metrics::Count; +use arrow::array::ArrayRef; +use arrow::datatypes::{DataType, Schema}; +use arrow::record_batch::RecordBatch; +use datafusion_common::Result; +use datafusion_common::cast::as_boolean_array; +use datafusion_common::instant::Instant; +use datafusion_expr::{ColumnarValue, Operator}; +use datafusion_physical_expr::PhysicalExpr; +use datafusion_physical_expr::expressions::BinaryExpr; +use datafusion_physical_expr::utils::split_conjunction; +use datafusion_physical_expr_common::physical_expr::is_volatile; + +/// Batches measured before the order is settled. +const WARMUP_BATCHES: u64 = 8; + +/// A candidate order is adopted only if its expected cost is below +/// `(1 - TIE_COST_FRACTION)` of the written order's. +const TIE_COST_FRACTION: f64 = 0.05; + +/// Per-conjunct counts over the warm-up, on exactly the rows that reached it. +#[derive(Debug, Default, Clone)] +struct ConjunctStats { + /// Total rows the conjunct was evaluated on. + rows: u64, + /// Rows that passed (non-null `true`, matching SQL filter semantics). + matched: u64, + /// Total evaluation time, nanoseconds. + nanos: u64, +} + +impl ConjunctStats { + /// Pool another stream's counts into this one. + fn merge(&mut self, other: &Self) { + self.rows += other.rows; + self.matched += other.matched; + self.nanos += other.nanos; + } + + /// Fraction of rows that pass, or `None` if never evaluated on any row. + fn pass_rate(&self) -> Option { + (self.rows > 0).then(|| self.matched as f64 / self.rows as f64) + } + + /// Per-row cost in nanoseconds, or `None` if never evaluated. Time is + /// clamped to 1ns so "too cheap to measure" ranks as very cheap. + fn cost_per_row(&self) -> Option { + (self.rows > 0).then(|| self.nanos.max(1) as f64 / self.rows as f64) + } + + /// Ranking key: rows discarded per nanosecond, `(1 + rows_in - rows_out) / + /// time` — the reciprocal of the score Velox sorts its filters by + /// (), so maximising it + /// minimises time per discarded row. `None` when unmeasured, so such + /// conjuncts sort last. + fn effectiveness(&self) -> Option { + (self.rows > 0) + .then(|| (1 + self.rows - self.matched) as f64 / self.nanos.max(1) as f64) + } +} + +/// Measurements pooled across the partition streams of one `FilterExec`, and +/// the decision the first stream to fill the warm-up makes for all of them. +#[derive(Debug, Default)] +pub(crate) struct AdaptiveFilterShared { + inner: Mutex, +} + +#[derive(Debug, Default)] +struct SharedInner { + /// Pooled per-conjunct counts, sized by the first measured batch. + stats: Vec, + /// Measured batches contributed by all streams so far. + measured_batches: u64, + /// The settled decision, once made; `None` while learning. + settled: Option, +} + +/// The settled outcome of the warm-up. +#[derive(Debug, Clone)] +struct Settled { + /// The settled order as a right-nested `AND` chain. + predicate: Arc, + /// Whether that order reorders the written conjuncts. + reordered: bool, +} + +impl AdaptiveFilterShared { + /// The settled decision, or `None` if the streams are still learning. + #[cfg(test)] + fn settled(&self) -> Option { + self.inner.lock().expect("poisoned").settled.clone() + } + + /// Seed `(rows, matched, nanos)` per conjunct one batch short of the + /// warm-up, so the next measured batch settles on the seeded decision + /// regardless of real timings. + #[cfg(test)] + pub(crate) fn seed_one_batch_short_of_warmup( + &self, + per_conjunct: &[(u64, u64, u64)], + ) { + let mut inner = self.inner.lock().expect("poisoned"); + inner.stats = per_conjunct + .iter() + .map(|&(rows, matched, nanos)| ConjunctStats { + rows, + matched, + nanos, + }) + .collect(); + inner.measured_batches = WARMUP_BATCHES - 1; + } +} + +/// A conjunct that records the rows it was handed, the rows it kept and the +/// time it took, returning its result unchanged (nulls included). Everything +/// else delegates to the wrapped conjunct. +#[derive(Debug)] +struct MeasuredConjunct { + inner: Arc, + /// Rows handed to the conjunct since the last [`take`](Self::take). + rows: AtomicU64, + /// Of those, the non-null `true`s. + matched: AtomicU64, + /// Time spent inside the conjunct over those rows, in nanoseconds. + nanos: AtomicU64, +} + +impl MeasuredConjunct { + fn new(inner: Arc) -> Self { + Self { + inner, + rows: AtomicU64::new(0), + matched: AtomicU64::new(0), + nanos: AtomicU64::new(0), + } + } + + /// Drain the counters (per stream and uncontended, hence `Relaxed`). + fn take(&self) -> ConjunctStats { + ConjunctStats { + rows: self.rows.swap(0, Relaxed), + matched: self.matched.swap(0, Relaxed), + nanos: self.nanos.swap(0, Relaxed), + } + } +} + +impl PartialEq for MeasuredConjunct { + fn eq(&self, other: &Self) -> bool { + self.inner.eq(&other.inner) + } +} + +impl Eq for MeasuredConjunct {} + +impl std::hash::Hash for MeasuredConjunct { + fn hash(&self, state: &mut H) { + self.inner.hash(state); + } +} + +impl fmt::Display for MeasuredConjunct { + fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { + write!(f, "{}", self.inner) + } +} + +impl PhysicalExpr for MeasuredConjunct { + fn data_type(&self, input_schema: &Schema) -> Result { + self.inner.data_type(input_schema) + } + + fn nullable(&self, input_schema: &Schema) -> Result { + self.inner.nullable(input_schema) + } + + fn evaluate(&self, batch: &RecordBatch) -> Result { + let rows = batch.num_rows(); + let timer = Instant::now(); + let array = self.inner.evaluate(batch)?.into_array(rows)?; + let nanos = timer.elapsed().as_nanos() as u64; + let matched = as_boolean_array(&array)?.true_count() as u64; + + self.rows.fetch_add(rows as u64, Relaxed); + self.matched.fetch_add(matched, Relaxed); + self.nanos.fetch_add(nanos, Relaxed); + + Ok(ColumnarValue::Array(array)) + } + + fn children(&self) -> Vec<&Arc> { + vec![&self.inner] + } + + fn with_new_children( + self: Arc, + children: Vec>, + ) -> Result> { + Ok(Arc::new(Self::new(Arc::clone(&children[0])))) + } + + fn fmt_sql(&self, f: &mut Formatter<'_>) -> fmt::Result { + self.inner.fmt_sql(f) + } +} + +/// Adaptive evaluator for a single conjunctive predicate, owned per partition +/// stream. Measurements are pooled into the shared [`AdaptiveFilterShared`]; +/// the per-stream state is just the chain this stream currently evaluates. +#[derive(Debug)] +pub(crate) struct AdaptiveConjunction { + /// The split conjuncts, in written order. + conjuncts: Vec>, + /// Measurements and the settled decision, shared by every partition stream. + shared: Arc, + /// The written order as a right-nested `AND` chain over the wrappers. + warmup_predicate: Arc, + /// The wrappers inside `warmup_predicate`, in written order. + measured: Vec>, + /// The settled order as a right-nested `AND` chain; the warm-up chain until then. + settled_predicate: Arc, + /// Whether the settled decision reordered the conjuncts. + reordered: bool, + /// Whether the order is settled: this stream no longer measures. + settled: bool, + /// Incremented once, if and when this stream adopts a *reordered* decision. + adaptive_reorders: Option, +} + +impl AdaptiveConjunction { + /// Whether `predicate` has at least two `AND` conjuncts, none volatile. + /// (Whether the feature is enabled is the caller's business.) + pub(crate) fn applies(predicate: &Arc) -> bool { + let conjuncts = split_conjunction(predicate); + conjuncts.len() >= 2 && !conjuncts.iter().any(|c| is_volatile(c)) + } + + /// `None` if adaptive reordering does not [apply](Self::applies). + /// `adaptive_reorders` is bumped if this stream adopts a reorder. + pub(crate) fn try_new( + predicate: &Arc, + shared: Arc, + adaptive_reorders: Option, + ) -> Option { + if !Self::applies(predicate) { + return None; + } + let conjuncts: Vec> = split_conjunction(predicate) + .into_iter() + .map(Arc::clone) + .collect(); + let order: Vec = (0..conjuncts.len()).collect(); + let measured: Vec> = conjuncts + .iter() + .map(|c| Arc::new(MeasuredConjunct::new(Arc::clone(c)))) + .collect(); + let wrapped: Vec> = measured + .iter() + .map(|m| Arc::clone(m) as Arc) + .collect(); + let warmup_predicate = right_nested_conjunction(&wrapped, &order); + Some(Self { + conjuncts, + shared, + settled_predicate: Arc::clone(&warmup_predicate), + warmup_predicate, + measured, + reordered: false, + settled: false, + adaptive_reorders, + }) + } + + /// The boolean mask of rows passing every conjunct. Until the order + /// settles, each batch is measured and its counts pooled. + pub(crate) fn evaluate(&mut self, batch: &RecordBatch) -> Result { + if self.settled { + return self.evaluate_settled(batch); + } + + // Empty batches measure nothing and must not consume the warm-up. + if batch.num_rows() == 0 { + let mask = self.evaluate_warmup(batch)?; + self.take_measurements(); + return Ok(mask); + } + + let result = self.evaluate_warmup(batch)?; + let local = self.take_measurements(); + self.pool_and_maybe_settle(&local); + Ok(result) + } + + /// Evaluate the written order through the wrappers. + fn evaluate_warmup(&self, batch: &RecordBatch) -> Result { + self.warmup_predicate + .evaluate(batch)? + .into_array(batch.num_rows()) + } + + /// Drain the wrappers, indexed by written position. + fn take_measurements(&self) -> Vec { + self.measured.iter().map(|m| m.take()).collect() + } + + /// Evaluate the settled chain, uninstrumented. + fn evaluate_settled(&self, batch: &RecordBatch) -> Result { + self.settled_predicate + .evaluate(batch)? + .into_array(batch.num_rows()) + } + + fn adopt(&mut self, decision: Settled) { + self.settled_predicate = decision.predicate; + self.reordered = decision.reordered; + self.settled = true; + if self.reordered + && let Some(count) = &self.adaptive_reorders + { + count.add(1); + } + } + + /// Pool this batch's counts and settle once the warm-up is full. + fn pool_and_maybe_settle(&mut self, local: &[ConjunctStats]) { + let mut inner = self.shared.inner.lock().expect("poisoned"); + // Another stream settled meanwhile: take its decision and drop this + // batch's counts. Checking here rather than before evaluating keeps the + // lock off the path until there is something to pool. + if let Some(decision) = inner.settled.clone() { + drop(inner); + self.adopt(decision); + return; + } + if inner.stats.is_empty() { + inner.stats = vec![ConjunctStats::default(); local.len()]; + } + // One `AdaptiveFilterShared` only ever backs one predicate. + debug_assert_eq!(inner.stats.len(), local.len()); + for (s, l) in inner.stats.iter_mut().zip(local) { + s.merge(l); + } + inner.measured_batches += 1; + if inner.measured_batches < WARMUP_BATCHES { + return; + } + let decision = settle(&inner.stats, &self.conjuncts); + inner.settled = Some(decision.clone()); + drop(inner); + self.adopt(decision); + } +} + +/// Rank by effectiveness and adopt the ranking only if it is materially +/// cheaper than the written order; either way, build the result as a +/// right-nested `AND` chain. +fn settle(stats: &[ConjunctStats], conjuncts: &[Arc]) -> Settled { + let identity: Vec = (0..stats.len()).collect(); + let candidate = rank_by_effectiveness(stats); + if candidate != identity + && expected_cost_per_row(stats, &candidate) + < (1.0 - TIE_COST_FRACTION) * expected_cost_per_row(stats, &identity) + { + Settled { + predicate: right_nested_conjunction(conjuncts, &candidate), + reordered: true, + } + } else { + Settled { + predicate: right_nested_conjunction(conjuncts, &identity), + reordered: false, + } + } +} + +/// `conjuncts` in `order` as `(c_first AND (c_second AND (... AND c_last)))`. +/// Right-nesting lets [`BinaryExpr`]'s pre-selection keep the first conjunct's +/// survivors compacted for the rest of the chain. `order` must be non-empty. +fn right_nested_conjunction( + conjuncts: &[Arc], + order: &[usize], +) -> Arc { + let (&last, rest) = order.split_last().expect("a non-empty order"); + rest.iter() + .rev() + .fold(Arc::clone(&conjuncts[last]), |acc, &id| { + Arc::new(BinaryExpr::new( + Arc::clone(&conjuncts[id]), + Operator::And, + acc, + )) as _ + }) +} + +/// Rank conjunct ids by effectiveness (discards per nanosecond) descending; +/// ids without measurements sort last. Stable, so equal ids keep their order. +fn rank_by_effectiveness(stats: &[ConjunctStats]) -> Vec { + let mut ids: Vec = (0..stats.len()).collect(); + ids.sort_by( + |&a, &b| match (stats[a].effectiveness(), stats[b].effectiveness()) { + (Some(x), Some(y)) => y.partial_cmp(&x).unwrap_or(std::cmp::Ordering::Equal), + (Some(_), None) => std::cmp::Ordering::Less, + (None, Some(_)) => std::cmp::Ordering::Greater, + (None, None) => std::cmp::Ordering::Equal, + }, + ); + ids +} + +/// Expected nanoseconds per input row for `order`: each conjunct's per-row +/// cost weighted by the product of the pass rates before it (assumed +/// independent). Unmeasured conjuncts contribute nothing. +fn expected_cost_per_row(stats: &[ConjunctStats], order: &[usize]) -> f64 { + let mut weight = 1.0_f64; + let mut total = 0.0_f64; + for &id in order { + let (Some(cost), Some(pass)) = (stats[id].cost_per_row(), stats[id].pass_rate()) + else { + continue; + }; + total += weight * cost; + weight *= pass; + } + total +} + +#[cfg(test)] +mod tests { + use super::*; + + use arrow::array::{Array, Int32Array, Int64Array}; + use arrow::datatypes::{DataType, Field, Schema}; + use datafusion_physical_expr::expressions::{binary, col, lit}; + + fn schema() -> Arc { + Arc::new(Schema::new(vec![ + Field::new("a", DataType::Int32, false), + Field::new("b", DataType::Int32, false), + ])) + } + + fn batch(schema: &Arc, a: Vec, b: Vec) -> RecordBatch { + RecordBatch::try_new( + Arc::clone(schema), + vec![Arc::new(Int32Array::from(a)), Arc::new(Int32Array::from(b))], + ) + .unwrap() + } + + /// `a > 2 AND b < 5` + fn predicate(schema: &Arc) -> Arc { + let left = + binary(col("a", schema).unwrap(), Operator::Gt, lit(2i32), schema).unwrap(); + let right = + binary(col("b", schema).unwrap(), Operator::Lt, lit(5i32), schema).unwrap(); + binary(left, Operator::And, right, schema).unwrap() + } + + /// The conjuncts of `predicate`, pointer-equal to the ones inside it. + fn split(predicate: &Arc) -> Vec> { + split_conjunction(predicate) + .into_iter() + .map(Arc::clone) + .collect() + } + + /// Assert `chain` is the right-nested `AND` of `conjuncts` in `order`. + fn assert_chain( + chain: &Arc, + conjuncts: &[Arc], + order: &[usize], + ) { + let (&last, rest) = order.split_last().expect("a non-empty order"); + let mut node = Arc::clone(chain); + for (depth, &id) in rest.iter().enumerate() { + let and = node + .downcast_ref::() + .unwrap_or_else(|| panic!("an AND at depth {depth}")); + assert_eq!(*and.op(), Operator::And); + assert!( + Arc::ptr_eq(and.left(), &conjuncts[id]), + "conjunct {id} at depth {depth}" + ); + node = Arc::clone(and.right()); + } + assert!(Arc::ptr_eq(&node, &conjuncts[last]), "last conjunct {last}"); + } + + fn passing_rows(mask: &ArrayRef) -> Vec { + let mask = as_boolean_array(mask).unwrap(); + (0..mask.len()) + .filter(|&i| !mask.is_null(i) && mask.value(i)) + .collect() + } + + fn stats(rows: u64, matched: u64, nanos: u64) -> ConjunctStats { + ConjunctStats { + rows, + matched, + nanos, + } + } + + /// `try_new` with a fresh, unshared registry and no metric. + fn try_new(predicate: &Arc) -> Option { + AdaptiveConjunction::try_new( + predicate, + Arc::new(AdaptiveFilterShared::default()), + None, + ) + } + + #[test] + fn single_conjunct_is_not_adaptive() { + let schema = schema(); + let p = + binary(col("a", &schema).unwrap(), Operator::Gt, lit(2i32), &schema).unwrap(); + assert!(try_new(&p).is_none()); + } + + #[test] + fn two_conjuncts_are_adaptive() { + let schema = schema(); + let adaptive = try_new(&predicate(&schema)).unwrap(); + assert_eq!(adaptive.conjuncts.len(), 2); + assert!(!adaptive.settled); + assert!(!adaptive.reordered); + } + + #[test] + fn ranks_by_discards_per_nanosecond() { + // id 0: cheap (1ns/row), unselective (pass 0.9): eff = 0.1 / 1 = 0.1 + // id 1: expensive (5ns/row), selective (pass 0.01): eff = 0.99 / 5 = 0.198 (first) + // id 2: cheap (1ns/row), very unselective (pass 0.95): eff = 0.05 / 1 = 0.05 (last) + let s = vec![ + stats(1000, 900, 1000), + stats(1000, 10, 5000), + stats(1000, 950, 1000), + ]; + assert_eq!(rank_by_effectiveness(&s), vec![1, 0, 2]); + } + + #[test] + fn unmeasured_conjuncts_sort_last() { + let s = vec![ + stats(0, 0, 0), // unmeasured -> last + stats(1000, 10, 1000), // selective + stats(1000, 900, 1000), // unselective + ]; + assert_eq!(rank_by_effectiveness(&s), vec![1, 2, 0]); + } + + #[test] + fn zero_nanos_ranks_as_very_cheap() { + // A conjunct evaluated faster than the timer's resolution must rank as + // very cheap (its cost clamps to 1ns total), not drop out of the + // ranking as unmeasured (which would sort it last — backwards). + let s = vec![ + stats(1000, 500, 0), // immeasurably cheap, somewhat selective + stats(1000, 500, 1000), // same selectivity, 1ns/row + ]; + assert_eq!(rank_by_effectiveness(&s), vec![0, 1]); + } + + /// The wrapper counts rows in and non-null trues, returns the array + /// untouched, and drains to zero. Elapsed time is not asserted: five rows + /// can legitimately measure 0ns on a coarse timer. + #[test] + fn measured_conjunct_counts_rows_and_matches() { + let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, true)])); + let rb = RecordBatch::try_new( + Arc::clone(&schema), + vec![Arc::new(Int32Array::from(vec![ + Some(1), + None, + Some(5), + Some(7), + None, + ]))], + ) + .unwrap(); + let inner = + binary(col("a", &schema).unwrap(), Operator::Gt, lit(2i32), &schema).unwrap(); + let measured = Arc::new(MeasuredConjunct::new(Arc::clone(&inner))); + + let got = measured + .evaluate(&rb) + .unwrap() + .into_array(rb.num_rows()) + .unwrap(); + let want = inner + .evaluate(&rb) + .unwrap() + .into_array(rb.num_rows()) + .unwrap(); + assert_eq!(&got, &want, "the conjunct's own result, unchanged"); + assert_eq!( + as_boolean_array(&got).unwrap().null_count(), + 2, + "nulls are left for `BinaryExpr` to interpret" + ); + + let s = measured.take(); + assert_eq!((s.rows, s.matched), (5, 2)); + let s = measured.take(); + assert_eq!((s.rows, s.matched, s.nanos), (0, 0, 0)); + } + + /// Empty batches must not consume the warm-up. + #[test] + fn empty_batches_do_not_consume_warmup() { + let schema = schema(); + let mut adaptive = try_new(&predicate(&schema)).unwrap(); + + for _ in 0..(2 * WARMUP_BATCHES) { + let rb = batch(&schema, vec![], vec![]); + let got = adaptive.evaluate(&rb).unwrap(); + assert_eq!(got.len(), 0); + } + assert!(!adaptive.settled); + assert_eq!(adaptive.shared.inner.lock().unwrap().measured_batches, 0); + } + + #[test] + fn expected_cost_weights_by_upstream_pass_rate() { + // a: cost 1, pass 0.5 ; b: cost 10, pass 0.5 + let s = vec![stats(1000, 500, 1000), stats(1000, 500, 10_000)]; + // order [0,1]: 1 + 0.5*10 = 6 + assert!((expected_cost_per_row(&s, &[0, 1]) - 6.0).abs() < 1e-9); + // order [1,0]: 10 + 0.5*1 = 10.5 + assert!((expected_cost_per_row(&s, &[1, 0]) - 10.5).abs() < 1e-9); + } + + /// The mask equals the written predicate's before and after settling. + #[test] + fn evaluate_matches_predicate_across_warmup() { + let schema = schema(); + let p = predicate(&schema); + let mut adaptive = try_new(&p).unwrap(); + + for round in 0..(WARMUP_BATCHES as i32 + 4) { + let base = round * 10; + let a: Vec = (base..base + 10).collect(); + let b: Vec = (base..base + 10).map(|x| x.rem_euclid(9)).collect(); + let rb = batch(&schema, a, b); + + let got = adaptive.evaluate(&rb).unwrap(); + let want = p.evaluate(&rb).unwrap().into_array(rb.num_rows()).unwrap(); + assert_eq!( + passing_rows(&got), + passing_rows(&want), + "mismatch on round {round}" + ); + } + assert!(adaptive.settled); + } + + /// Null-producing conjuncts match the written predicate on every batch. + #[test] + fn nullable_conjuncts_match_the_plain_predicate_across_warmup() { + let schema = Arc::new(Schema::new(vec![ + Field::new("a", DataType::Int32, true), + Field::new("b", DataType::Int32, true), + ])); + let p = predicate(&schema); + let mut adaptive = try_new(&p).unwrap(); + + for round in 0..(WARMUP_BATCHES as i32 + 4) { + let base = round * 10; + let a: Vec> = (base..base + 10) + .map(|x| (x.rem_euclid(3) != 0).then_some(x)) + .collect(); + let b: Vec> = (base..base + 10) + .map(|x| (x.rem_euclid(4) != 0).then_some(x.rem_euclid(9))) + .collect(); + let rb = RecordBatch::try_new( + Arc::clone(&schema), + vec![Arc::new(Int32Array::from(a)), Arc::new(Int32Array::from(b))], + ) + .unwrap(); + + let got = adaptive.evaluate(&rb).unwrap(); + let want = p.evaluate(&rb).unwrap().into_array(rb.num_rows()).unwrap(); + assert_eq!( + passing_rows(&got), + passing_rows(&want), + "mismatch on round {round}" + ); + } + assert!(adaptive.settled); + } + + /// An already-good order is kept, as a right-nested chain. + #[test] + fn settle_keeps_order_when_not_materially_better() { + let schema = schema(); + let p = predicate(&schema); + // Equally cheap and selective: swapping cannot help. + let s = vec![stats(1000, 500, 1000), stats(1000, 500, 1000)]; + let cs = split(&p); + let d = settle(&s, &cs); + assert!(!d.reordered); + assert_chain(&d.predicate, &cs, &[0, 1]); + } + + #[test] + fn settle_adopts_materially_cheaper_order() { + let schema = schema(); + let p = predicate(&schema); + let cs = split(&p); + // id 1 is far more selective at equal cost: it moves first. + let s = vec![stats(1000, 900, 1000), stats(1000, 10, 1000)]; + let d = settle(&s, &cs); + assert!(d.reordered); + assert_chain(&d.predicate, &cs, &[1, 0]); + } + + /// The adopted order is a right-nested chain. + #[test] + fn adopted_order_is_a_right_nested_and_chain() { + let schema = schema(); + // Three conjuncts: `a > 2 AND b < 5 AND a < 90`. + let p = binary( + predicate(&schema), + Operator::And, + binary( + col("a", &schema).unwrap(), + Operator::Lt, + lit(90i32), + &schema, + ) + .unwrap(), + &schema, + ) + .unwrap(); + let cs = split(&p); + assert_eq!(cs.len(), 3); + + // Equal cost, decreasing pass rate: the ranking reverses the order. + let s = vec![ + stats(1000, 900, 1000), + stats(1000, 500, 1000), + stats(1000, 10, 1000), + ]; + let d = settle(&s, &cs); + assert!(d.reordered); + + // `(cs[2] AND (cs[1] AND cs[0]))`. + let outer = d.predicate.downcast_ref::().expect("an AND"); + assert_eq!(*outer.op(), Operator::And); + assert!( + Arc::ptr_eq(outer.left(), &cs[2]), + "first conjunct is outermost" + ); + let inner = outer + .right() + .downcast_ref::() + .expect("the tail is itself an AND"); + assert_eq!(*inner.op(), Operator::And); + assert!(Arc::ptr_eq(inner.left(), &cs[1])); + assert!(Arc::ptr_eq(inner.right(), &cs[0])); + } + + /// A kept written order runs as a right-nested chain. Real timings are + /// used on purpose: with both pass rates above `1 - TIE_COST_FRACTION` the + /// guard `c1 + p*c0 < 0.95 * (c0 + p*c1)` cannot hold for any costs. + #[test] + fn no_reorder_evaluates_plain_predicate() { + let schema = schema(); + let left = + binary(col("a", &schema).unwrap(), Operator::Gt, lit(2i32), &schema).unwrap(); + let right = + binary(col("b", &schema).unwrap(), Operator::Gt, lit(2i32), &schema).unwrap(); + let p = binary(left, Operator::And, right, &schema).unwrap(); + let mut adaptive = try_new(&p).unwrap(); + + for round in 0..(WARMUP_BATCHES as i32 + 2) { + let base = round * 10; + let a: Vec = (base..base + 10).collect(); + let b: Vec = (base..base + 10).collect(); + let rb = batch(&schema, a, b); + let got = adaptive.evaluate(&rb).unwrap(); + let want = p.evaluate(&rb).unwrap().into_array(rb.num_rows()).unwrap(); + assert_eq!(passing_rows(&got), passing_rows(&want)); + } + assert!(adaptive.settled); + assert!( + !adaptive.reordered, + "interchangeable conjuncts keep the written order" + ); + assert_chain(&adaptive.settled_predicate, &split(&p), &[0, 1]); + } + + /// Two streams share one pool: the warm-up is pooled across both, and the + /// second adopts the first's decision on its next batch. + #[test] + fn streams_pool_measurements_and_share_settled_order() { + let schema = schema(); + let p = predicate(&schema); // `a > 2 AND b < 5`, written order [0, 1] + let cs = split(&p); + let shared = Arc::new(AdaptiveFilterShared::default()); + shared.seed_one_batch_short_of_warmup(&[ + (70_000_000, 63_000_000, 70_000_000), // pass 0.9, ~1ns/row + (70_000_000, 700_000, 350_000_000), // pass 0.01, ~5ns/row + ]); + let mut s1 = AdaptiveConjunction::try_new(&p, Arc::clone(&shared), None).unwrap(); + let mut s2 = AdaptiveConjunction::try_new(&p, Arc::clone(&shared), None).unwrap(); + + let mk = |round: i32| { + let base = round * 100; + let a: Vec = (base..base + 100).collect(); + let b: Vec = (base..base + 100).map(|x| x.rem_euclid(25)).collect(); + batch(&schema, a, b) + }; + + for round in 0..(WARMUP_BATCHES as i32) { + let rb = mk(round); + for s in [&mut s1, &mut s2] { + let got = s.evaluate(&rb).unwrap(); + let want = p.evaluate(&rb).unwrap().into_array(rb.num_rows()).unwrap(); + assert_eq!(passing_rows(&got), passing_rows(&want)); + } + } + + assert!(shared.settled().is_some()); + assert!(s1.settled && s2.settled); + assert!(s1.reordered && s2.reordered); + assert_chain(&s1.settled_predicate, &cs, &[1, 0]); + assert_chain(&s2.settled_predicate, &cs, &[1, 0]); + } + + /// `adaptive_reorders` is bumped once per stream: by the settler and by a + /// stream that later takes up its decision. + #[test] + fn adopted_reorder_signals_once_per_stream() { + let schema = schema(); + let p = predicate(&schema); // `a > 2 AND b < 5`, written order [0, 1] + let shared = Arc::new(AdaptiveFilterShared::default()); + shared.seed_one_batch_short_of_warmup(&[ + (70_000_000, 63_000_000, 70_000_000), // pass 0.9, ~1ns/row + (70_000_000, 700_000, 350_000_000), // pass 0.01, ~5ns/row + ]); + let settler_count = Count::new(); + let adopter_count = Count::new(); + let mut settler = AdaptiveConjunction::try_new( + &p, + Arc::clone(&shared), + Some(settler_count.clone()), + ) + .unwrap(); + let mut adopter = AdaptiveConjunction::try_new( + &p, + Arc::clone(&shared), + Some(adopter_count.clone()), + ) + .unwrap(); + + let a: Vec = (0..100).collect(); + let b: Vec = a.iter().map(|x| x.rem_euclid(25)).collect(); + let rb = batch(&schema, a, b); + + assert_eq!(settler_count.value(), 0); + + // This batch completes the warm-up. + settler.evaluate(&rb).unwrap(); + assert!(settler.reordered); + assert_eq!(settler_count.value(), 1); + settler.evaluate(&rb).unwrap(); + assert_eq!(settler_count.value(), 1); + + adopter.evaluate(&rb).unwrap(); + assert!(adopter.reordered); + assert_eq!(adopter_count.value(), 1); + adopter.evaluate(&rb).unwrap(); + assert_eq!(adopter_count.value(), 1); + } + + /// Keeping the written order is not a reorder. + #[test] + fn settling_without_reorder_signals_nothing() { + let schema = schema(); + let p = predicate(&schema); + let shared = Arc::new(AdaptiveFilterShared::default()); + // Identical cost and selectivity: no order can be materially cheaper. + shared.seed_one_batch_short_of_warmup(&[ + (70_000_000, 35_000_000, 70_000_000), + (70_000_000, 35_000_000, 70_000_000), + ]); + let count = Count::new(); + let mut adaptive = + AdaptiveConjunction::try_new(&p, Arc::clone(&shared), Some(count.clone())) + .unwrap(); + + let a: Vec = (0..100).collect(); + let rb = batch(&schema, a.clone(), a); + adaptive.evaluate(&rb).unwrap(); + assert!(adaptive.settled && !adaptive.reordered); + assert_eq!(count.value(), 0); + } + + /// Feed batches, settle on a reorder, stay there. + #[test] + fn scenario_measure_batches_then_settle_on_reorder() { + let schema = schema(); + let p = predicate(&schema); // `a > 2 AND b < 5`, written order [0, 1] + let cs = split(&p); + let shared = Arc::new(AdaptiveFilterShared::default()); + shared.seed_one_batch_short_of_warmup(&[ + (70_000_000, 63_000_000, 70_000_000), // pass 0.9, ~1ns/row + (70_000_000, 700_000, 350_000_000), // pass 0.01, ~5ns/row + ]); + let mut adaptive = + AdaptiveConjunction::try_new(&p, Arc::clone(&shared), None).unwrap(); + assert!(!adaptive.settled, "the first batch is still measured"); + + for round in 0..3 { + let base = round * 100; + let a: Vec = (base..base + 100).collect(); + let b: Vec = (base..base + 100).map(|x| x.rem_euclid(25)).collect(); + let rb = batch(&schema, a, b); + let got = adaptive.evaluate(&rb).unwrap(); + let want = p.evaluate(&rb).unwrap().into_array(rb.num_rows()).unwrap(); + assert_eq!(passing_rows(&got), passing_rows(&want), "round {round}"); + assert!(adaptive.settled, "settled after round {round}"); + assert!(adaptive.reordered, "reordered after round {round}"); + assert_chain(&adaptive.settled_predicate, &cs, &[1, 0]); + } + } + + /// Interchangeable conjuncts settle on the written order. + #[test] + fn scenario_measure_batches_then_settle_on_written_order() { + let schema = schema(); + let p = predicate(&schema); + let cs = split(&p); + let shared = Arc::new(AdaptiveFilterShared::default()); + // Identical cost and selectivity: no order can be materially cheaper. + shared.seed_one_batch_short_of_warmup(&[ + (70_000_000, 35_000_000, 70_000_000), + (70_000_000, 35_000_000, 70_000_000), + ]); + let mut adaptive = + AdaptiveConjunction::try_new(&p, Arc::clone(&shared), None).unwrap(); + assert!(!adaptive.settled, "the first batch is still measured"); + + for round in 0..3 { + let base = round * 100; + let a: Vec = (base..base + 100).collect(); + let b: Vec = (base..base + 100).collect(); + let rb = batch(&schema, a, b); + let got = adaptive.evaluate(&rb).unwrap(); + let want = p.evaluate(&rb).unwrap().into_array(rb.num_rows()).unwrap(); + assert_eq!(passing_rows(&got), passing_rows(&want), "round {round}"); + assert!(adaptive.settled, "settled after round {round}"); + assert!(!adaptive.reordered, "not reordered after round {round}"); + assert_chain(&adaptive.settled_predicate, &cs, &[0, 1]); + } + } + + /// The pool is sized by the first measured batch and receives its counts. + #[test] + fn first_measured_batch_initialises_the_shared_pool() { + let schema = schema(); + let p = predicate(&schema); // `a > 2 AND b < 5` + let shared = Arc::new(AdaptiveFilterShared::default()); + assert!(shared.inner.lock().unwrap().stats.is_empty()); + let mut adaptive = + AdaptiveConjunction::try_new(&p, Arc::clone(&shared), None).unwrap(); + adaptive.evaluate(&batch(&schema, vec![], vec![])).unwrap(); + assert!(shared.inner.lock().unwrap().stats.is_empty()); + + let a: Vec = (0..10).collect(); + adaptive.evaluate(&batch(&schema, a.clone(), a)).unwrap(); + + let inner = shared.inner.lock().unwrap(); + assert_eq!(inner.stats.len(), 2, "sized to the conjunct count"); + assert_eq!(inner.measured_batches, 1); + // `a > 2` keeps 7 of 10 (no pre-selection), so `b < 5` sees all 10. + assert_eq!((inner.stats[0].rows, inner.stats[0].matched), (10, 7)); + assert_eq!((inner.stats[1].rows, inner.stats[1].matched), (10, 5)); + } + + /// `Int64` schema for the divide-by-zero side-effect tests below. + fn int64_schema() -> Arc { + Arc::new(Schema::new(vec![ + Field::new("a", DataType::Int64, false), + Field::new("b", DataType::Int64, false), + ])) + } + + fn int64_batch(schema: &Arc, a: Vec, b: Vec) -> RecordBatch { + RecordBatch::try_new( + Arc::clone(schema), + vec![Arc::new(Int64Array::from(a)), Arc::new(Int64Array::from(b))], + ) + .unwrap() + } + + /// The two conjuncts `b <> 0` and `1 / b > 2`. + fn divide_by_zero_conjuncts( + schema: &Arc, + ) -> (Arc, Arc) { + let non_zero = binary( + col("b", schema).unwrap(), + Operator::NotEq, + lit(0i64), + schema, + ) + .unwrap(); + let divide = binary( + binary( + lit(1i64), + Operator::Divide, + col("b", schema).unwrap(), + schema, + ) + .unwrap(), + Operator::Gt, + lit(2i64), + schema, + ) + .unwrap(); + (non_zero, divide) + } + + /// The side effect the config doc warns about: a reorder can introduce an + /// error the written order avoided. `b <> 0` holds on 15% of rows, so the + /// written `AND` pre-selects and `1 / b` never sees a zero; reordered, + /// `1 / b > 2` runs first on every row. + #[test] + fn adopted_reorder_can_introduce_a_divide_by_zero() { + let schema = int64_schema(); + let (non_zero, divide) = divide_by_zero_conjuncts(&schema); + let p = binary(non_zero, Operator::And, divide, &schema).unwrap(); + let cs = split(&p); + let a: Vec = (0..100).collect(); + let b: Vec = (0..100).map(|i| i64::from(i < 15)).collect(); + let rb = int64_batch(&schema, a, b); + + // Flag off: succeeds. + assert!(p.evaluate(&rb).is_ok(), "flag-off evaluation must succeed"); + + let shared = Arc::new(AdaptiveFilterShared::default()); + shared.seed_one_batch_short_of_warmup(&[ + (70_000_000, 63_000_000, 70_000_000), // pass 0.9, ~1ns/row + (70_000_000, 700_000, 70_000_000), // pass 0.01, ~1ns/row + ]); + let mut adaptive = + AdaptiveConjunction::try_new(&p, Arc::clone(&shared), None).unwrap(); + + // The settling batch still runs in the written order. + adaptive.evaluate(&rb).unwrap(); + assert!(adaptive.reordered); + assert_chain(&adaptive.settled_predicate, &cs, &[1, 0]); + + // The next batch runs the reorder. + let err = adaptive.evaluate(&rb).unwrap_err().to_string(); + assert!(err.contains("Divide by zero"), "unexpected error: {err}"); + } + + /// The mirror: `1 / b > 2 AND a < 10` with `b = 0` exactly where `a < 10` + /// discards. The written order divides by zero; reordered, `a < 10` keeps + /// 10% of rows, so pre-selection keeps `1 / b` away from the zeros. + #[test] + fn adopted_reorder_can_avoid_a_divide_by_zero_the_written_order_raises() { + let schema = int64_schema(); + let (_, divide) = divide_by_zero_conjuncts(&schema); + let selective = binary( + col("a", &schema).unwrap(), + Operator::Lt, + lit(10i64), + &schema, + ) + .unwrap(); + let p = binary(divide, Operator::And, selective, &schema).unwrap(); + let cs = split(&p); + + let shared = Arc::new(AdaptiveFilterShared::default()); + shared.seed_one_batch_short_of_warmup(&[ + (70_000_000, 63_000_000, 350_000_000), // pass 0.9, ~5ns/row + (70_000_000, 700_000, 70_000_000), // pass 0.01, ~1ns/row + ]); + let mut adaptive = + AdaptiveConjunction::try_new(&p, Arc::clone(&shared), None).unwrap(); + + // Settle on a batch with no zeros, so the warm-up cannot error. + let a: Vec = (0..100).collect(); + adaptive + .evaluate(&int64_batch(&schema, a.clone(), vec![1; 100])) + .unwrap(); + assert!(adaptive.reordered); + assert_chain(&adaptive.settled_predicate, &cs, &[1, 0]); + + let b: Vec = (0..100).map(|i| i64::from(i < 10)).collect(); + let rb = int64_batch(&schema, a, b); + + let err = p.evaluate(&rb).unwrap_err().to_string(); + assert!(err.contains("Divide by zero"), "unexpected error: {err}"); + let got = adaptive.evaluate(&rb).unwrap(); + assert!(passing_rows(&got).is_empty(), "1 / 1 > 2 is false"); + } +} diff --git a/datafusion/physical-plan/src/filter.rs b/datafusion/physical-plan/src/filter.rs index 12771eec78470..bbf3f0ff9079c 100644 --- a/datafusion/physical-plan/src/filter.rs +++ b/datafusion/physical-plan/src/filter.rs @@ -28,6 +28,7 @@ use super::{ ColumnStatistics, DisplayAs, ExecutionPlanProperties, PlanProperties, RecordBatchStream, SendableRecordBatchStream, Statistics, }; +use crate::adaptive_filter::{AdaptiveConjunction, AdaptiveFilterShared}; use crate::coalesce::{LimitedBatchCoalescer, PushBatchStatus}; use crate::common::can_project; use crate::execution_plan::{CardinalityEffect, replace_children_if_necessary}; @@ -36,7 +37,7 @@ use crate::filter_pushdown::{ FilterPushdownPropagation, PushedDown, }; use crate::limit::LocalLimitExec; -use crate::metrics::{MetricBuilder, MetricType}; +use crate::metrics::{Count, MetricBuilder, MetricCategory, MetricType}; use crate::projection::{ EmbeddedProjection, ProjectionExec, ProjectionExpr, make_with_child, try_embed_projection, update_expr, @@ -99,6 +100,10 @@ pub struct FilterExec { batch_size: usize, /// Number of rows to fetch fetch: Option, + /// Adaptive conjunct reordering state (see [`AdaptiveConjunction`]), + /// pooled across partition streams. Shared by `Clone`; replaced by + /// [`reset_state`](ExecutionPlan::reset_state) and predicate rewrites. + adaptive_stats: Arc, } /// Builder for [`FilterExec`] to set optional parameters @@ -222,6 +227,7 @@ impl FilterExecBuilder { projection: self.projection, batch_size: self.batch_size, fetch: self.fetch, + adaptive_stats: Arc::new(AdaptiveFilterShared::default()), }) } } @@ -298,6 +304,7 @@ impl FilterExec { projection: self.projection.clone(), batch_size, fetch: self.fetch, + adaptive_stats: Arc::clone(&self.adaptive_stats), }) } @@ -575,6 +582,7 @@ impl ExecutionPlan for FilterExec { ) -> Result> { validate_child_count!(self, children); match options.children_properties { + // `adaptive_stats` is kept: the predicate is unchanged. ChildrenPropertiesMode::Keep => Ok(Arc::new(Self { input: children.swap_remove(0), metrics: ExecutionPlanMetricsSet::new(), @@ -610,6 +618,15 @@ impl ExecutionPlan for FilterExec { ) } + /// Fresh adaptive-reordering state and metrics for a re-execution; the + /// predicate, input and cached properties are still valid and kept. + fn reset_state(self: Arc) -> Result> { + let mut new = (*self).clone(); + new.adaptive_stats = Arc::new(AdaptiveFilterShared::default()); + new.metrics = ExecutionPlanMetricsSet::new(); + Ok(Arc::new(new)) + } + fn execute( &self, partition: usize, @@ -621,10 +638,30 @@ impl ExecutionPlan for FilterExec { context.session_id(), context.task_id() ); - let metrics = FilterExecMetrics::new(&self.metrics, partition); + let enabled = context + .session_config() + .options() + .execution + .adaptive_filter_reordering; + // Register the counter exactly when the adaptive path is active. + let adaptive_applies = enabled && AdaptiveConjunction::applies(&self.predicate); + let mut metrics = FilterExecMetrics::new(&self.metrics, partition); + if adaptive_applies { + metrics = metrics.with_adaptive_reorder_metrics(&self.metrics, partition); + } + let adaptive = adaptive_applies + .then(|| { + AdaptiveConjunction::try_new( + &self.predicate, + Arc::clone(&self.adaptive_stats), + metrics.adaptive_reorders.clone(), + ) + }) + .flatten(); Ok(Box::pin(FilterExecStream { schema: self.schema(), predicate: Arc::clone(&self.predicate), + adaptive, input: self.input.execute(partition, context)?, metrics, projection: self.projection.clone(), @@ -821,6 +858,8 @@ impl ExecutionPlan for FilterExec { projection: self.projection.clone(), batch_size: self.batch_size, fetch: self.fetch, + // The predicate changed. + adaptive_stats: Arc::new(AdaptiveFilterShared::default()), }; Some(Arc::new(new) as _) }; @@ -845,6 +884,7 @@ impl ExecutionPlan for FilterExec { projection: self.projection.clone(), batch_size: self.batch_size, fetch, + adaptive_stats: Arc::clone(&self.adaptive_stats), })) } @@ -881,6 +921,8 @@ impl ExecutionPlan for FilterExec { projection, batch_size, fetch, + // Per-execution adaptive measurements, not part of the plan shape. + adaptive_stats: _, } = self; let input_node = ctx.encode_child(input)?; let expr = ctx.encode_expr(predicate)?; @@ -1322,6 +1364,8 @@ struct FilterExecStream { schema: SchemaRef, /// The expression to filter on. This expression must evaluate to a boolean value. predicate: Arc, + /// Evaluates `predicate` adaptively when set. + adaptive: Option, /// The input partition to filter. input: SendableRecordBatchStream, /// Runtime metrics recording @@ -1338,6 +1382,9 @@ struct FilterExecMetrics { baseline_metrics: BaselineMetrics, /// Selectivity of the filter, calculated as output_rows / input_rows selectivity: RatioMetrics, + /// Partition streams that adopted an adaptively reordered conjunct order. + /// Registered only when adaptive reordering applies. + adaptive_reorders: Option, // Remember to update `docs/source/user-guide/metrics.md` when adding new metrics, // or modifying metrics comments } @@ -1349,8 +1396,24 @@ impl FilterExecMetrics { selectivity: MetricBuilder::new(metrics) .with_type(MetricType::Summary) .ratio_metrics("selectivity", partition), + adaptive_reorders: None, } } + + /// Also register the `adaptive_reorders` counter; see + /// [`Self::adaptive_reorders`]. + fn with_adaptive_reorder_metrics( + mut self, + metrics: &ExecutionPlanMetricsSet, + partition: usize, + ) -> Self { + self.adaptive_reorders = Some( + MetricBuilder::new(metrics) + .with_category(MetricCategory::Rows) + .counter("adaptive_reorders", partition), + ); + self + } } pub fn batch_filter( @@ -1417,9 +1480,15 @@ impl Stream for FilterExecStream { } Some(Ok(batch)) => { let timer = elapsed_compute.timer(); - let status = self.predicate.as_ref() - .evaluate(&batch) - .and_then(|v| v.into_array(batch.num_rows())) + let array = match self.adaptive.as_mut() { + Some(adaptive) => adaptive.evaluate(&batch), + None => self + .predicate + .as_ref() + .evaluate(&batch) + .and_then(|v| v.into_array(batch.num_rows())), + }; + let status = array .and_then(|array| { Ok(match self.projection.as_ref() { Some(projection) => { @@ -1541,8 +1610,11 @@ mod tests { use crate::expressions::*; use crate::statistics::{StatisticsArgs, StatisticsContext}; use crate::test; + use crate::test::TestMemoryExec; use crate::test::exec::StatisticsExec; + use arrow::array::{Array, Int64Array}; use arrow::datatypes::{Field, Schema, UnionFields, UnionMode}; + use datafusion_execution::config::SessionConfig; #[test] fn filter_rejects_zero_batch_size() -> Result<()> { @@ -2408,6 +2480,172 @@ mod tests { Ok(()) } + #[tokio::test] + async fn test_reset_state_gives_fresh_adaptive_stats() -> Result<()> { + let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]); + let input = Arc::new(StatisticsExec::new( + Statistics::new_unknown(&schema), + schema, + )); + let predicate = Arc::new(BinaryExpr::new( + Arc::new(Column::new("a", 0)), + Operator::Eq, + Arc::new(Literal::new(ScalarValue::Int32(Some(10)))), + )); + let filter = Arc::new(FilterExec::try_new(predicate, input)?); + + let reset = Arc::clone(&filter).reset_state()?; + let reset = reset + .as_ref() + .downcast_ref::() + .expect("reset_state returns a FilterExec"); + + assert!(!Arc::ptr_eq(&filter.adaptive_stats, &reset.adaptive_stats)); + assert!(Arc::ptr_eq(&filter.predicate, &reset.predicate)); + Ok(()) + } + + /// `FilterExec` with the flag on: four partitions, a nullable column, a + /// conjunction written selective-last. The pool is seeded so the reorder + /// is adopted regardless of timings; everything else is the real path. + #[tokio::test] + async fn adaptive_filter_reordering_end_to_end() -> Result<()> { + const PARTITIONS: usize = 4; + const BATCHES: usize = 16; + const ROWS: i64 = 64; + // `b` is NULL on every 37th row, including inside the selected range. + const NULL_EVERY: i64 = 37; + + let schema = Arc::new(Schema::new(vec![ + Field::new("a", DataType::Int64, false), + Field::new("b", DataType::Int64, true), + ])); + + let mut partitions = Vec::with_capacity(PARTITIONS); + for p in 0..PARTITIONS { + let mut batches = Vec::with_capacity(BATCHES); + for batch in 0..BATCHES { + let base = (p * BATCHES + batch) as i64 * ROWS; + let a: Vec = (base..base + ROWS).collect(); + let b: Vec> = a + .iter() + .map(|&v| (v % NULL_EVERY != 0).then_some(v)) + .collect(); + batches.push(RecordBatch::try_new( + Arc::clone(&schema), + vec![Arc::new(Int64Array::from(a)), Arc::new(Int64Array::from(b))], + )?); + } + partitions.push(batches); + } + let total_rows = (PARTITIONS * BATCHES) as i64 * ROWS; + let input: Arc = + TestMemoryExec::try_new_exec(&partitions, Arc::clone(&schema), None)?; + + // `(a % 97 + a % 89 >= 0) AND (b > 3990)`: the always-true conjunct is + // written first; both are cheap, so the static heuristic leaves them. + let threshold = total_rows - 106; // 3990 + let cheap = binary( + binary( + binary(col("a", &schema)?, Operator::Modulo, lit(97i64), &schema)?, + Operator::Plus, + binary(col("a", &schema)?, Operator::Modulo, lit(89i64), &schema)?, + &schema, + )?, + Operator::GtEq, + lit(0i64), + &schema, + )?; + let selective = + binary(col("b", &schema)?, Operator::Gt, lit(threshold), &schema)?; + let predicate = binary(cheap, Operator::And, selective, &schema)?; + let filter = Arc::new(FilterExec::try_new(predicate, input)?); + + // Output rows across all partitions, sorted. + async fn run( + filter: &Arc, + adaptive: bool, + ) -> Result> { + let mut config = SessionConfig::new(); + config.options_mut().execution.adaptive_filter_reordering = adaptive; + let ctx = Arc::new(TaskContext::default().with_session_config(config)); + let mut rows = vec![]; + for partition in 0..PARTITIONS { + let stream = filter.execute(partition, Arc::clone(&ctx))?; + for batch in crate::common::collect(stream).await? { + let a = batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + let b = batch + .column(1) + .as_any() + .downcast_ref::() + .unwrap(); + for i in 0..batch.num_rows() { + assert!(!b.is_null(i), "a NULL row survived the filter"); + rows.push((a.value(i), b.value(i))); + } + } + } + rows.sort_unstable(); + Ok(rows) + } + + // Expected rows: 105 values of `b > 3990`, three of them NULL. + let candidates: Vec = (threshold + 1..total_rows).collect(); + assert_eq!(candidates.len(), 105); + let expected: Vec<(i64, i64)> = candidates + .into_iter() + .filter(|v| v % NULL_EVERY != 0) + .map(|v| (v, v)) + .collect(); + assert_eq!(expected.len(), 102, "three NULL `b` rows are dropped"); + + let flag_off = run(&filter, false).await?; + assert_eq!(flag_off, expected); + assert!( + filter + .metrics() + .unwrap() + .sum_by_name("adaptive_reorders") + .is_none(), + "the flag-off path must not register the adaptive metric" + ); + + // Conjunct 0 keeps every row at ~5x the cost of conjunct 1 (keeps 1%). + filter.adaptive_stats.seed_one_batch_short_of_warmup(&[ + (70_000_000, 70_000_000, 350_000_000), + (70_000_000, 700_000, 70_000_000), + ]); + + let flag_on = run(&filter, true).await?; + assert_eq!(flag_on, flag_off, "reordering must not change results"); + let reorders = filter + .metrics() + .unwrap() + .sum_by_name("adaptive_reorders") + .map(|m| m.as_usize()) + .unwrap_or(0); + assert!(reorders >= 1, "expected an adopted reorder, got {reorders}"); + + // Re-executing the same node keeps the learned state; same rows. + assert_eq!(run(&filter, true).await?, flag_off, "state persists"); + + // A reset node learns from scratch; same rows. + let reset = Arc::clone(&filter).reset_state()?; + let reset: Arc = Arc::new( + reset + .as_ref() + .downcast_ref::() + .expect("reset_state returns a FilterExec") + .clone(), + ); + assert_eq!(run(&reset, true).await?, flag_off, "after reset_state"); + Ok(()) + } + #[test] fn test_equivalence_properties_union_type() -> Result<()> { let union_type = DataType::Union( diff --git a/datafusion/physical-plan/src/lib.rs b/datafusion/physical-plan/src/lib.rs index 5ff6cec374ee1..fb87e9c118c2f 100644 --- a/datafusion/physical-plan/src/lib.rs +++ b/datafusion/physical-plan/src/lib.rs @@ -67,6 +67,7 @@ mod render_tree; mod topk; mod visitor; +mod adaptive_filter; pub mod aggregates; pub mod analyze; pub mod async_func; diff --git a/datafusion/sqllogictest/test_files/adaptive_filter.slt b/datafusion/sqllogictest/test_files/adaptive_filter.slt new file mode 100644 index 0000000000000..e3f0876eace32 --- /dev/null +++ b/datafusion/sqllogictest/test_files/adaptive_filter.slt @@ -0,0 +1,131 @@ +# 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. + +# Tests for execution.adaptive_filter_reordering. Runtime reordering of a +# conjunction must never change query results, only evaluation order. + +# Small batches, so the 8-batch warm-up completes and the settled path runs. +statement ok +SET datafusion.execution.batch_size = 64; + +statement ok +CREATE TABLE t AS +SELECT + i AS a, + i % 7 AS b, + arrow_cast(i, 'Utf8') AS s, + CASE WHEN i % 5 = 0 THEN NULL ELSE i END AS n +FROM generate_series(1, 4000) AS tbl(i); + +# Baseline (flag off): multi-conjunct filter mixing a cheap comparison with an +# expensive LIKE. +query I +SELECT count(*) FROM t WHERE b = 3 AND s LIKE '1%'; +---- +160 + +# Baseline plan (flag off); the flag-on plan below must be identical. +query TT +EXPLAIN SELECT count(*) FROM t WHERE b = 3 AND s LIKE '1%'; +---- +logical_plan +01)Projection: count(Int64(1)) AS count(*) +02)--Aggregate: groupBy=[[]], aggr=[[count(Int64(1))]] +03)----Projection: +04)------Filter: t.b = Int64(3) AND t.s LIKE Utf8("1%") +05)--------TableScan: t projection=[b, s] +physical_plan +01)ProjectionExec: expr=[count(Int64(1))@0 as count(*)] +02)--AggregateExec: mode=Final, gby=[], aggr=[count(Int64(1))] +03)----CoalescePartitionsExec +04)------AggregateExec: mode=Partial, gby=[], aggr=[count(Int64(1))] +05)--------FilterExec: b@0 = 3 AND s@1 LIKE 1%, projection=[] +06)----------DataSourceExec: partitions=4, partition_sizes=[16, 16, 16, 15] + +statement ok +SET datafusion.execution.adaptive_filter_reordering = true; + +# Same query with adaptive reordering on must return the same result. +query I +SELECT count(*) FROM t WHERE b = 3 AND s LIKE '1%'; +---- +160 + +# A three-conjunct predicate, including an expensive-but-selective LIKE. +query I +SELECT count(*) FROM t WHERE a > 100 AND s LIKE '5%' AND b <> 0; +---- +86 + +# A conjunct that produces NULLs: SQL filter semantics treat NULL as false, +# with or without reordering. +query I +SELECT count(*) FROM t WHERE n % 2 = 0 AND s LIKE '2%'; +---- +445 + +# Full row materialization (not just count) is unchanged by reordering. +query IIT +SELECT a, b, s FROM t WHERE b = 1 AND a > 3990 AND s LIKE '39%' ORDER BY a; +---- +3991 1 3991 +3998 1 3998 + +# EXPLAIN: runtime reordering is invisible to the plan — the FilterExec +# predicate is unchanged whether the flag is on or off. +query TT +EXPLAIN SELECT count(*) FROM t WHERE b = 3 AND s LIKE '1%'; +---- +logical_plan +01)Projection: count(Int64(1)) AS count(*) +02)--Aggregate: groupBy=[[]], aggr=[[count(Int64(1))]] +03)----Projection: +04)------Filter: t.b = Int64(3) AND t.s LIKE Utf8("1%") +05)--------TableScan: t projection=[b, s] +physical_plan +01)ProjectionExec: expr=[count(Int64(1))@0 as count(*)] +02)--AggregateExec: mode=Final, gby=[], aggr=[count(Int64(1))] +03)----CoalescePartitionsExec +04)------AggregateExec: mode=Partial, gby=[], aggr=[count(Int64(1))] +05)--------FilterExec: b@0 = 3 AND s@1 LIKE 1%, projection=[] +06)----------DataSourceExec: partitions=4, partition_sizes=[16, 16, 16, 15] + +# EXPLAIN ANALYZE shows the reorder via `adaptive_reorders`. The predicate is +# written selective-last: the arithmetic chain matches every row and costs +# more than `a > 3990`, which keeps 10 rows in 4000. Both are cheap to the +# static `reorder_predicates` rule, so the written order reaches `FilterExec`. +# Each of the 4 partition streams adopts the reorder once, hence 4. +query I +SELECT count(*) FROM t WHERE a % 97 + a % 89 + a % 83 + b % 13 >= 0 AND a > 3990; +---- +10 + +query TT +EXPLAIN ANALYZE +SELECT count(*) FROM t WHERE a % 97 + a % 89 + a % 83 + b % 13 >= 0 AND a > 3990; +---- +Plan with Metrics +adaptive_reorders=4 + +statement ok +SET datafusion.execution.adaptive_filter_reordering = false; + +statement ok +SET datafusion.execution.batch_size = 8192; + +statement ok +DROP TABLE t; diff --git a/datafusion/sqllogictest/test_files/information_schema.slt b/datafusion/sqllogictest/test_files/information_schema.slt index f2a587ed72ae0..8742ebbe8689f 100644 --- a/datafusion/sqllogictest/test_files/information_schema.slt +++ b/datafusion/sqllogictest/test_files/information_schema.slt @@ -214,6 +214,7 @@ datafusion.catalog.has_header true datafusion.catalog.information_schema true datafusion.catalog.location NULL datafusion.catalog.newlines_in_values false +datafusion.execution.adaptive_filter_reordering false datafusion.execution.batch_size 8192 datafusion.execution.coalesce_batches true datafusion.execution.collect_statistics true @@ -375,6 +376,7 @@ datafusion.catalog.has_header true Default value for `format.has_header` for `CR datafusion.catalog.information_schema true Should DataFusion provide access to `information_schema` virtual tables for displaying schema information datafusion.catalog.location NULL Location scanned to load tables for `default` schema datafusion.catalog.newlines_in_values false Specifies whether newlines in (quoted) CSV values are supported. This is the default value for `format.newlines_in_values` for `CREATE EXTERNAL TABLE` if not specified explicitly in the statement. Parsing newlines in quoted values may be affected by execution behaviour such as parallel file scanning. Setting this to `true` ensures that newlines in values are parsed successfully, which may reduce performance. +datafusion.execution.adaptive_filter_reordering false (experimental) When enabled, `FilterExec` measures the selectivity and evaluation cost of each conjunct of an `AND` predicate at runtime and reorders them to run the ones that discard the most rows per unit of CPU time first. Query results never change, but the observable side effects of a fallible predicate can, in either direction: reordering `b <> 0 AND 1/b > 2` can make a divide-by-zero error appear or disappear, since each conjunct is evaluated only on the rows the conjuncts before it kept. Predicates containing volatile expressions are never reordered. datafusion.execution.batch_size 8192 Default batch size while creating new batches, it's especially useful for buffer-in-memory batches since creating tiny batches would result in too much metadata memory consumption datafusion.execution.coalesce_batches true When set to true, record batches will be examined between each operator and small batches will be coalesced into larger batches. This is helpful when there are highly selective filters or joins that could produce tiny output batches. The target batch size is determined by the configuration setting datafusion.execution.collect_statistics true Should DataFusion collect statistics when first creating a table. Has no effect after the table is created. Defaults to true. diff --git a/docs/source/user-guide/configs.md b/docs/source/user-guide/configs.md index ab344028cdab4..401ee9e528117 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.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. | | datafusion.execution.enforce_batch_size_in_joins | false | Should DataFusion enforce batch size in joins or not. By default, DataFusion will not enforce batch size in joins. Enforcing batch size in joins can reduce memory usage when joining large tables with a highly-selective join filter, but is also slightly slower. | +| datafusion.execution.adaptive_filter_reordering | false | (experimental) When enabled, `FilterExec` measures the selectivity and evaluation cost of each conjunct of an `AND` predicate at runtime and reorders them to run the ones that discard the most rows per unit of CPU time first. Query results never change, but the observable side effects of a fallible predicate can, in either direction: reordering `b <> 0 AND 1/b > 2` can make a divide-by-zero error appear or disappear, since each conjunct is evaluated only on the rows the conjuncts before it kept. Predicates containing volatile expressions are never reordered. | | datafusion.execution.objectstore_writer_buffer_size | 10485760 | Size (bytes) of data buffer DataFusion uses when writing output files. This affects the size of the data chunks that are uploaded to remote object stores (e.g. AWS S3). If very large (>= 100 GiB) output files are being written, it may be necessary to increase this size to avoid errors from the remote end point. | | datafusion.execution.enable_ansi_mode | false | Whether to enable ANSI SQL mode. The flag is experimental and relevant only for DataFusion Spark built-in functions When `enable_ansi_mode` is set to `true`, the query engine follows ANSI SQL semantics for expressions, casting, and error handling. This means: - **Strict type coercion rules:** implicit casts between incompatible types are disallowed. - **Standard SQL arithmetic behavior:** operations such as division by zero, numeric overflow, or invalid casts raise runtime errors rather than returning `NULL` or adjusted values. - **Consistent ANSI behavior** for string concatenation, comparisons, and `NULL` handling. When `enable_ansi_mode` is `false` (the default), the engine uses a more permissive, non-ANSI mode designed for user convenience and backward compatibility. In this mode: - Implicit casts between types are allowed (e.g., string to integer when possible). - Arithmetic operations are more lenient — for example, `abs()` on the minimum representable integer value returns the input value instead of raising overflow. - Division by zero or invalid casts may return `NULL` instead of failing. # Default `false` — ANSI SQL mode is disabled by default. | | datafusion.execution.hash_join_buffering_capacity | 0 | How many bytes to buffer in the probe side of hash joins while the build side is concurrently being built. Without this, hash joins will wait until the full materialization of the build side before polling the probe side. This is useful in scenarios where the query is not completely CPU bounded, allowing to do some early work concurrently and reducing the latency of the query. Note that when hash join buffering is enabled, the probe side will start eagerly polling data, not giving time for the producer side of dynamic filters to produce any meaningful predicate. Queries with dynamic filters might see performance degradation. Disabled by default, set to a number greater than 0 for enabling it. | diff --git a/docs/source/user-guide/metrics.md b/docs/source/user-guide/metrics.md index 306094c438eb7..98a7c6c80fa31 100644 --- a/docs/source/user-guide/metrics.md +++ b/docs/source/user-guide/metrics.md @@ -38,9 +38,10 @@ DataFusion operators expose runtime metrics so you can understand where time is ### FilterExec -| Metric | Description | -| ----------- | ----------------------------------------------------------------- | -| selectivity | Selectivity of the filter, calculated as output_rows / input_rows | +| Metric | Description | +| ----------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| selectivity | Selectivity of the filter, calculated as output_rows / input_rows | +| adaptive_reorders | Partition streams that adopted an adaptively reordered conjunct order. Present only when `datafusion.execution.adaptive_filter_reordering` applies; `0` means the written order was kept. | ### HashJoinExec