xudong963 commented on code in PR #22698: URL: https://github.com/apache/datafusion/pull/22698#discussion_r4022745053
########## 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 +//! (<https://github.com/apache/datafusion/pull/22343>). 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 <https://github.com/apache/datafusion/pull/22698>. + +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<f64> { + (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<f64> { + (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 + /// (<https://www.vldb.org/pvldb/vol15/p3372-pedreira.pdf>), so maximising it + /// minimises time per discarded row. `None` when unmeasured, so such + /// conjuncts sort last. + fn effectiveness(&self) -> Option<f64> { + (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<SharedInner>, +} + +#[derive(Debug, Default)] +struct SharedInner { + /// Pooled per-conjunct counts, sized by the first measured batch. + stats: Vec<ConjunctStats>, + /// Measured batches contributed by all streams so far. + measured_batches: u64, + /// The settled decision, once made; `None` while learning. + settled: Option<Settled>, +} + +/// The settled outcome of the warm-up. +#[derive(Debug, Clone)] +struct Settled { + /// The settled order as a right-nested `AND` chain. + predicate: Arc<dyn PhysicalExpr>, + /// 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<Settled> { + 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<dyn PhysicalExpr>, + /// 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<dyn PhysicalExpr>) -> 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<H: std::hash::Hasher>(&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<DataType> { + self.inner.data_type(input_schema) + } + + fn nullable(&self, input_schema: &Schema) -> Result<bool> { + self.inner.nullable(input_schema) + } + + fn evaluate(&self, batch: &RecordBatch) -> Result<ColumnarValue> { + 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<dyn PhysicalExpr>> { + vec![&self.inner] + } + + fn with_new_children( + self: Arc<Self>, + children: Vec<Arc<dyn PhysicalExpr>>, + ) -> Result<Arc<dyn PhysicalExpr>> { + 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<Arc<dyn PhysicalExpr>>, + /// Measurements and the settled decision, shared by every partition stream. + shared: Arc<AdaptiveFilterShared>, + /// The written order as a right-nested `AND` chain over the wrappers. + warmup_predicate: Arc<dyn PhysicalExpr>, + /// The wrappers inside `warmup_predicate`, in written order. + measured: Vec<Arc<MeasuredConjunct>>, + /// The settled order as a right-nested `AND` chain; the warm-up chain until then. + settled_predicate: Arc<dyn PhysicalExpr>, + /// 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<Count>, +} + +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<dyn PhysicalExpr>) -> 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<dyn PhysicalExpr>, + shared: Arc<AdaptiveFilterShared>, + adaptive_reorders: Option<Count>, + ) -> Option<Self> { + if !Self::applies(predicate) { + return None; + } + let conjuncts: Vec<Arc<dyn PhysicalExpr>> = split_conjunction(predicate) + .into_iter() + .map(Arc::clone) + .collect(); + let order: Vec<usize> = (0..conjuncts.len()).collect(); + let measured: Vec<Arc<MeasuredConjunct>> = conjuncts + .iter() + .map(|c| Arc::new(MeasuredConjunct::new(Arc::clone(c)))) + .collect(); + let wrapped: Vec<Arc<dyn PhysicalExpr>> = measured + .iter() + .map(|m| Arc::clone(m) as Arc<dyn PhysicalExpr>) + .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<ArrayRef> { + 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<ArrayRef> { + self.warmup_predicate + .evaluate(batch)? + .into_array(batch.num_rows()) + } + + /// Drain the wrappers, indexed by written position. + fn take_measurements(&self) -> Vec<ConjunctStats> { + self.measured.iter().map(|m| m.take()).collect() + } + + /// Evaluate the settled chain, uninstrumented. + fn evaluate_settled(&self, batch: &RecordBatch) -> Result<ArrayRef> { + 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<dyn PhysicalExpr>]) -> Settled { + let identity: Vec<usize> = (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<dyn PhysicalExpr>], + order: &[usize], +) -> Arc<dyn PhysicalExpr> { + 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<usize> { + let mut ids: Vec<usize> = (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 { Review Comment: I do not think the current cost model matches how `BinaryExpr` evaluates `AND`. `ConjunctStats` records rows, true values, and time, while `expected_cost_per_row` assumes every pass rate reduces the work of subsequent predicates. In practice, `check_short_circuit` only pre-selects when there are no nulls and the true rate is at most 20%. For example, predicates keeping 30% and 90% of rows both leave the next predicate evaluating the full batch, but the current model discounts both downstream costs. Nullable predicates are even more problematic: nulls make the predicate look selective here, while any null disables pre-selection. Could we record the null count and model the actual strategy? - all false: stop evaluating - all true: continue with the full batch - no nulls and pass rate <= the pre-selection threshold: reduce the downstream row weight - otherwise: continue with the full batch It would also be good to add regression tests just below, exactly at, and just above the 20% threshold, plus a nullable case. Ideally this rule should be shared with `BinaryExpr` so the two implementations cannot drift apart. ########## 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 +//! (<https://github.com/apache/datafusion/pull/22343>). 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 <https://github.com/apache/datafusion/pull/22698>. + +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<f64> { + (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<f64> { + (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 + /// (<https://www.vldb.org/pvldb/vol15/p3372-pedreira.pdf>), so maximising it + /// minimises time per discarded row. `None` when unmeasured, so such + /// conjuncts sort last. + fn effectiveness(&self) -> Option<f64> { + (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<SharedInner>, +} + +#[derive(Debug, Default)] +struct SharedInner { + /// Pooled per-conjunct counts, sized by the first measured batch. + stats: Vec<ConjunctStats>, + /// Measured batches contributed by all streams so far. + measured_batches: u64, + /// The settled decision, once made; `None` while learning. + settled: Option<Settled>, +} + +/// The settled outcome of the warm-up. +#[derive(Debug, Clone)] +struct Settled { + /// The settled order as a right-nested `AND` chain. + predicate: Arc<dyn PhysicalExpr>, + /// 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<Settled> { + 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<dyn PhysicalExpr>, + /// 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<dyn PhysicalExpr>) -> 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<H: std::hash::Hasher>(&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<DataType> { + self.inner.data_type(input_schema) + } + + fn nullable(&self, input_schema: &Schema) -> Result<bool> { + self.inner.nullable(input_schema) + } + + fn evaluate(&self, batch: &RecordBatch) -> Result<ColumnarValue> { + 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<dyn PhysicalExpr>> { + vec![&self.inner] + } + + fn with_new_children( + self: Arc<Self>, + children: Vec<Arc<dyn PhysicalExpr>>, + ) -> Result<Arc<dyn PhysicalExpr>> { + 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<Arc<dyn PhysicalExpr>>, + /// Measurements and the settled decision, shared by every partition stream. + shared: Arc<AdaptiveFilterShared>, + /// The written order as a right-nested `AND` chain over the wrappers. + warmup_predicate: Arc<dyn PhysicalExpr>, + /// The wrappers inside `warmup_predicate`, in written order. + measured: Vec<Arc<MeasuredConjunct>>, + /// The settled order as a right-nested `AND` chain; the warm-up chain until then. + settled_predicate: Arc<dyn PhysicalExpr>, + /// 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<Count>, +} + +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<dyn PhysicalExpr>) -> 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<dyn PhysicalExpr>, + shared: Arc<AdaptiveFilterShared>, + adaptive_reorders: Option<Count>, + ) -> Option<Self> { + if !Self::applies(predicate) { + return None; + } + let conjuncts: Vec<Arc<dyn PhysicalExpr>> = split_conjunction(predicate) + .into_iter() + .map(Arc::clone) + .collect(); + let order: Vec<usize> = (0..conjuncts.len()).collect(); + let measured: Vec<Arc<MeasuredConjunct>> = conjuncts + .iter() + .map(|c| Arc::new(MeasuredConjunct::new(Arc::clone(c)))) + .collect(); + let wrapped: Vec<Arc<dyn PhysicalExpr>> = measured + .iter() + .map(|m| Arc::clone(m) as Arc<dyn PhysicalExpr>) + .collect(); + let warmup_predicate = right_nested_conjunction(&wrapped, &order); Review Comment: I think we should preserve the original expression tree unless a real reorder is adopted. The warm-up predicate is always rebuilt as right-nested here, and the no-reorder path also returns a right-nested identity order. Reassociation can change short-circuit behavior even though the conjunct sequence is unchanged. For example, a left-nested prefix may compact the batch before a fallible predicate, while the equivalent right-nested tree can evaluate that predicate on the full batch. This means enabling the flag can change observable behavior before any reorder is selected, and `adaptive_reorders` can remain `0` even though evaluation behavior changed. Could we wrap the conjunct leaves during warm-up without changing the original tree shape, and return the original predicate `Arc` when the candidate reorder is rejected? If right-nesting the identity order is valuable independently, I think it should be treated as a separate optimization with its own tests and metrics. ########## datafusion/physical-plan/src/filter.rs: ########## @@ -98,6 +99,10 @@ pub struct FilterExec { batch_size: usize, /// Number of rows to fetch fetch: Option<usize>, + /// Measurements and trial verdicts shared by all partition streams, + /// used by adaptive conjunct reordering (see [`AdaptiveConjunction`]) so + /// the streams learn as one. Fresh per plan node; never affects the plan. + adaptive_stats: Arc<AdaptiveFilterShared>, Review Comment: Thanks. I do not think `reset_state()` fully addresses this. `FilterExec::execute()` still clones the pool stored on the plan node, so ordinary repeated or concurrent executions of the same physical plan share measurements and the settled decision unless the caller explicitly invokes `reset_state()`. The `ExecutionPlan` contract does not guarantee that `reset_state()` is called before every `execute()`. Could we scope this state to one logical query execution while still sharing it across that execution's partitions? If the current execution API cannot provide that lifetime, then I think this needs to be documented explicitly as plan-lifetime caching, with tests covering repeated and concurrent executions. In its current form, describing the state as "per-execution" is inaccurate. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
