alamb commented on code in PR #22698: URL: https://github.com/apache/datafusion/pull/22698#discussion_r3570452108
########## datafusion/physical-plan/src/adaptive_filter.rs: ########## @@ -0,0 +1,956 @@ +// 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). +//! +//! Predicate evaluation order matters: a selective predicate run first gates +//! the work of the predicates after it. DataFusion's `BinaryExpr` `AND` +//! short-circuit only gates on the *leftmost* conjunct, so a conjunction whose +//! selective member is written last (e.g. +//! `regexp_like(s,'a') AND … AND regexp_like(s,'rare')`) evaluates every +//! predicate against ~every row. +//! +//! ## How it evaluates: the compact-once loop +//! +//! The conjuncts are evaluated sequentially, combining their boolean results +//! with `AND`. The working batch is physically compacted to the surviving rows +//! once the accumulated mask becomes selective enough — so a run of +//! non-selective conjuncts costs only cheap bitwise `AND`s, while a selective +//! conjunct shrinks the batch the conjuncts after it must decode. This +//! compaction is what makes ordering pay off (and is itself a win even without +//! reordering): a left-deep fused `BinaryExpr` `AND` does *not* compact between Review Comment: I don't understand what this `left-deep fused BinaryExpr` is -- what is fused? I kind of understand the compaction comment. ########## datafusion/physical-plan/src/adaptive_filter.rs: ########## @@ -0,0 +1,956 @@ +// 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). +//! +//! Predicate evaluation order matters: a selective predicate run first gates +//! the work of the predicates after it. DataFusion's `BinaryExpr` `AND` +//! short-circuit only gates on the *leftmost* conjunct, so a conjunction whose Review Comment: nit: it may help here to add a doc link to the `And` implementation ########## datafusion/physical-plan/src/adaptive_filter.rs: ########## @@ -0,0 +1,956 @@ +// 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). +//! +//! Predicate evaluation order matters: a selective predicate run first gates +//! the work of the predicates after it. DataFusion's `BinaryExpr` `AND` +//! short-circuit only gates on the *leftmost* conjunct, so a conjunction whose +//! selective member is written last (e.g. +//! `regexp_like(s,'a') AND … AND regexp_like(s,'rare')`) evaluates every +//! predicate against ~every row. +//! +//! ## How it evaluates: the compact-once loop +//! +//! The conjuncts are evaluated sequentially, combining their boolean results +//! with `AND`. The working batch is physically compacted to the surviving rows +//! once the accumulated mask becomes selective enough — so a run of +//! non-selective conjuncts costs only cheap bitwise `AND`s, while a selective +//! conjunct shrinks the batch the conjuncts after it must decode. This +//! compaction is what makes ordering pay off (and is itself a win even without +//! reordering): a left-deep fused `BinaryExpr` `AND` does *not* compact between +//! conjuncts, so it evaluates ~every conjunct on ~every row regardless of order. +//! +//! ## How it orders +//! +//! Each conjunct is timed and counted on exactly the rows it evaluated, giving +//! its marginal selectivity and per-row cost. After a short warm-up the +//! conjuncts are ranked by rows discarded per nanosecond +//! (`(1 - pass_rate) / cost_per_row`, the classic optimal ordering key for +//! independent conjuncts), and if the ranked order is *materially* cheaper than +//! the written one it is adopted. The order then stays fixed. +//! +//! Compact-once is used **only in service of a reorder**: if the warm-up does +//! not reorder the conjuncts (e.g. they are interchangeable), the written +//! predicate is evaluated as-is, so once settled a conjunction that does not +//! benefit from reordering pays no compact-once overhead and evaluates exactly +//! as it would with the feature off. During the warm-up itself the conjuncts +//! are necessarily evaluated individually (with compaction) so they can be +//! measured — see the side-effects caveat below. +//! +//! ## How it shares +//! +//! A `FilterExec` is split across many partition streams, each seeing only a +//! slice of the data. Measurements are pooled into a shared +//! [`AdaptiveFilterShared`] so the streams learn as one: the first stream to +//! accumulate enough samples settles the order and publishes it, and the others +//! adopt it (one relaxed atomic load per batch) without each re-paying the +//! warm-up — which is what makes the win materialise when each stream is only a +//! handful of batches long. +//! +//! It is **off by default** Review Comment: I recommend just mentioning this behavior is controlled by the flag and leave the documentation of the default value on the flag itself (otherwise this can get out of date if we change the deafult value) ########## datafusion/common/src/config.rs: ########## @@ -939,6 +939,21 @@ 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` adaptively reorders the + /// conjuncts of a conjunctive predicate at runtime. It measures each + /// conjunct's selectivity and evaluation cost on the rows that reach it + /// and runs the conjuncts that discard the most rows per unit of CPU + /// time first, so cheap-and-selective predicates gate expensive ones. + /// This never changes query results, but it is off by default because + /// it can change observable side effects of fallible predicates (even + /// when no reorder is adopted): while measuring, and after a reorder, + /// conjuncts are evaluated only on the rows that survived the conjuncts + /// before them, so an error the fused predicate would have raised on an + /// already-filtered row (e.g. `b <> 0 AND 1/b > 2` evaluating `1/b` on + /// every row) may not occur. Predicates containing volatile expressions + /// are never reordered. Review Comment: The comment says "off by default because it can change observable side effects of fallible predicates" Does this mean we can **never** turn it on by default? What do you think about using it by default but disabling it for fallable predicates -- perhaps we can reuse the 'short_circuits' logic in https://docs.rs/datafusion/latest/datafusion/logical_expr/trait.ScalarUDFImpl.html#method.short_circuits I think we can make this description significantly more concise by omitting some internal details. Something like ```suggestion /// (experimental) When enabled, `FilterExec` adaptively reorders the /// conjuncts of a conjunctive predicate at runtime by measuring each /// conjunct's selectivity and evaluation cost. Off by default because /// it can change observable side effects of fallible predicates /// (e.g. reordering `b <> 0 AND 1/b > 2` may change if/when `1/b` results in a divide by zero /// error. Predicates containing volatile expressions /// are never reordered. ``` ########## datafusion/physical-plan/src/adaptive_filter.rs: ########## @@ -0,0 +1,956 @@ +// 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). +//! +//! Predicate evaluation order matters: a selective predicate run first gates +//! the work of the predicates after it. DataFusion's `BinaryExpr` `AND` +//! short-circuit only gates on the *leftmost* conjunct, so a conjunction whose +//! selective member is written last (e.g. +//! `regexp_like(s,'a') AND … AND regexp_like(s,'rare')`) evaluates every +//! predicate against ~every row. +//! +//! ## How it evaluates: the compact-once loop +//! +//! The conjuncts are evaluated sequentially, combining their boolean results +//! with `AND`. The working batch is physically compacted to the surviving rows +//! once the accumulated mask becomes selective enough — so a run of +//! non-selective conjuncts costs only cheap bitwise `AND`s, while a selective +//! conjunct shrinks the batch the conjuncts after it must decode. This +//! compaction is what makes ordering pay off (and is itself a win even without +//! reordering): a left-deep fused `BinaryExpr` `AND` does *not* compact between +//! conjuncts, so it evaluates ~every conjunct on ~every row regardless of order. +//! +//! ## How it orders +//! +//! Each conjunct is timed and counted on exactly the rows it evaluated, giving +//! its marginal selectivity and per-row cost. After a short warm-up the +//! conjuncts are ranked by rows discarded per nanosecond +//! (`(1 - pass_rate) / cost_per_row`, the classic optimal ordering key for +//! independent conjuncts), and if the ranked order is *materially* cheaper than +//! the written one it is adopted. The order then stays fixed. +//! +//! Compact-once is used **only in service of a reorder**: if the warm-up does +//! not reorder the conjuncts (e.g. they are interchangeable), the written +//! predicate is evaluated as-is, so once settled a conjunction that does not +//! benefit from reordering pays no compact-once overhead and evaluates exactly +//! as it would with the feature off. During the warm-up itself the conjuncts +//! are necessarily evaluated individually (with compaction) so they can be +//! measured — see the side-effects caveat below. +//! +//! ## How it shares +//! +//! A `FilterExec` is split across many partition streams, each seeing only a +//! slice of the data. Measurements are pooled into a shared +//! [`AdaptiveFilterShared`] so the streams learn as one: the first stream to +//! accumulate enough samples settles the order and publishes it, and the others Review Comment: I think @andygrove and @milenkovicm hit issues with ballista when the execution plans are run on multiple nodes (so in memory sharing doesn't work as well). We should probably make sure this mechanism will work with that system ########## datafusion/physical-plan/src/filter.rs: ########## @@ -1146,9 +1196,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() { Review Comment: Would it be simpler if the adaptive conjunction also implemented `PhysicalExpr` ? That way there would be no change to the runtime path of FilterExec, and we could potentially reuse AdaptiveConjunction anywhere physical exprs were used ########## datafusion/physical-plan/src/adaptive_filter.rs: ########## @@ -0,0 +1,956 @@ +// 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). +//! +//! Predicate evaluation order matters: a selective predicate run first gates +//! the work of the predicates after it. DataFusion's `BinaryExpr` `AND` +//! short-circuit only gates on the *leftmost* conjunct, so a conjunction whose +//! selective member is written last (e.g. +//! `regexp_like(s,'a') AND … AND regexp_like(s,'rare')`) evaluates every +//! predicate against ~every row. +//! +//! ## How it evaluates: the compact-once loop +//! +//! The conjuncts are evaluated sequentially, combining their boolean results +//! with `AND`. The working batch is physically compacted to the surviving rows +//! once the accumulated mask becomes selective enough — so a run of +//! non-selective conjuncts costs only cheap bitwise `AND`s, while a selective +//! conjunct shrinks the batch the conjuncts after it must decode. This +//! compaction is what makes ordering pay off (and is itself a win even without +//! reordering): a left-deep fused `BinaryExpr` `AND` does *not* compact between +//! conjuncts, so it evaluates ~every conjunct on ~every row regardless of order. +//! +//! ## How it orders +//! +//! Each conjunct is timed and counted on exactly the rows it evaluated, giving +//! its marginal selectivity and per-row cost. After a short warm-up the +//! conjuncts are ranked by rows discarded per nanosecond +//! (`(1 - pass_rate) / cost_per_row`, the classic optimal ordering key for +//! independent conjuncts), and if the ranked order is *materially* cheaper than +//! the written one it is adopted. The order then stays fixed. +//! +//! Compact-once is used **only in service of a reorder**: if the warm-up does +//! not reorder the conjuncts (e.g. they are interchangeable), the written +//! predicate is evaluated as-is, so once settled a conjunction that does not +//! benefit from reordering pays no compact-once overhead and evaluates exactly +//! as it would with the feature off. During the warm-up itself the conjuncts +//! are necessarily evaluated individually (with compaction) so they can be +//! measured — see the side-effects caveat below. +//! +//! ## How it shares +//! +//! A `FilterExec` is split across many partition streams, each seeing only a +//! slice of the data. Measurements are pooled into a shared +//! [`AdaptiveFilterShared`] so the streams learn as one: the first stream to +//! accumulate enough samples settles the order and publishes it, and the others +//! adopt it (one relaxed atomic load per batch) without each re-paying the +//! warm-up — which is what makes the win materialise when each stream is only a +//! handful of batches long. +//! +//! It is **off by default** +//! (`datafusion.execution.adaptive_filter_reordering`) and never changes query +//! results: a conjunction's value is independent of evaluation order. Predicates +//! containing volatile expressions are never reordered (their observable side +//! effects depend on order). +//! +//! Observable *side effects* of fallible predicates can change even when no +//! reorder is adopted: whenever conjuncts are evaluated individually (during +//! warm-up, or settled with a reorder), a conjunct after a compaction sees only +//! the surviving rows, so an error a fused evaluation would have raised on an +//! already-filtered row (e.g. `b <> 0 AND 1/b > 2` with the fused `AND` +//! evaluating `1/b` on every row) may not occur. +//! +//! ## Known limitations +//! +//! The statistics are *conditional*: each conjunct is measured on the rows that +//! survived the conjuncts before it in written order, and (after a compaction) +//! on small survivor batches whose per-row cost is inflated by fixed overheads. +//! Correlated conjuncts can therefore look more selective in a late position +//! than they would be up front, and the settle decision is one-shot: once +//! adopted, the order is never re-measured, so a misjudged reorder (or drifting +//! data) is kept for the stream's lifetime. The material-win guard +//! ([`TIE_COST_FRACTION`]) makes adoption conservative but cannot detect +//! correlation. Further policies (drift re-measurement, confidence-interval +//! statistics, A/B-validated adoption for the cases a cost model cannot +//! separate) can build on top of this core. + +use std::sync::Arc; +use std::sync::Mutex; +use std::sync::atomic::{AtomicU64, Ordering}; + +use arrow::array::{Array, ArrayRef, BooleanArray, BooleanBufferBuilder, UInt32Array}; +use arrow::buffer::BooleanBuffer; +use arrow::compute::kernels::boolean::and; +use arrow::compute::{filter, filter_record_batch, prep_null_mask_filter}; +use arrow::record_batch::RecordBatch; +use datafusion_common::cast::as_boolean_array; +use datafusion_common::instant::Instant; +use datafusion_common::{Result, internal_err}; +use datafusion_physical_expr::PhysicalExpr; +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; + +/// Fraction of the conjunction's expected per-row cost below which a reorder is +/// immaterial. A candidate order is adopted only if it is expected to cost less +/// than `(1 - TIE_COST_FRACTION)` of the written order, so interchangeable +/// conjuncts never trigger a reorder. +const TIE_COST_FRACTION: f64 = 0.05; + +/// Physically compact the working batch to the surviving rows only when the +/// accumulated mask keeps at most this fraction of them. Above this, the cost +/// of materializing a barely-smaller batch is not repaid, so we keep evaluating +/// against the full working batch and just `AND` the boolean masks. +const COMPACTION_SELECTIVITY_THRESHOLD: f64 = 0.2; + +/// Per-conjunct measurement: marginal pass rate and per-row evaluation cost, +/// accumulated over the warm-up window on exactly the rows that reached the +/// conjunct. +#[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 { + fn record(&mut self, matched: u64, rows: u64, nanos: u64) { + self.rows += rows; + self.matched += matched; + self.nanos += nanos; + } + + /// Fold another accumulator's counts into this one (they are plain sums, so + /// merging is addition). Used to pool measurements across partition streams. + 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 evaluation cost in nanoseconds, or `None` if the conjunct was + /// never evaluated on any row. An evaluation faster than the timer's + /// resolution is clamped to one nanosecond total: "too cheap to measure" + /// must rank as very cheap, not drop out of the ranking as unmeasured. + 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 of evaluation + /// (`(1 - pass_rate) / cost_per_row`). Maximising this is exactly + /// minimising `cost_per_row / (1 - pass_rate)`, the classic optimal + /// ordering key for independent conjuncts — so a selective-but-expensive + /// predicate correctly sorts ahead of a cheap-but-unselective one. + /// `None` when unmeasured, so such conjuncts sort last. + fn effectiveness(&self) -> Option<f64> { + let cost = self.cost_per_row()?; + let pass = self.pass_rate()?; + Some((1.0 - pass) / cost) + } +} + +/// State shared by every partition stream of one `FilterExec`, so the streams +/// learn as one: per-conjunct measurements are pooled across streams and the +/// first stream to accumulate enough samples settles the order for all of them. +/// +/// This matters because a `FilterExec` is split across many partition streams, +/// each seeing only a slice of the data. Without sharing, every stream pays its +/// own warm-up — and when each stream is only a handful of batches long, that +/// warm-up is most of its work, so the reordering win never materialises. With +/// sharing the warm-up is paid roughly once per query, not once per stream. +#[derive(Debug, Default)] +pub(crate) struct AdaptiveFilterShared { + /// `0` until an order is published; bumped once when the first stream + /// settles. Streams poll it with one relaxed atomic load per batch. + epoch: AtomicU64, + inner: Mutex<SharedInner>, +} + +#[derive(Debug, Default)] +struct SharedInner { + /// Per-conjunct counts pooled across all streams (indexed by conjunct + /// position). Empty until the first measured batch sizes it. + 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>, +} + +/// How one batch was evaluated. Reported by +/// [`AdaptiveConjunction::evaluate_traced`] so the evaluator's behaviour is +/// observable batch by batch (its input/output contract is exercised by the +/// `scenario_*` tests). Borrows the adopted order rather than cloning it, so +/// reporting costs nothing on the per-batch path. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum BatchStrategy<'a> { + /// Still learning: the written order through the compact-once loop, each + /// conjunct instrumented and its counts pooled. (Empty batches are + /// evaluated but measure nothing and do not consume the warm-up.) + Measure, + /// Settled without a reorder: the written fused predicate, exactly as if + /// the feature were off. + Fused, + /// Settled on an adopted reorder, evaluated through the compact-once + /// loop. The payload is the adopted evaluation order: positions in the + /// written conjunct list, first-evaluated first. + Reordered(&'a [usize]), +} + +/// The settled outcome of the warm-up: the evaluation order, and whether to run +/// it through the compact-once loop or the plain predicate. +#[derive(Debug, Clone)] +struct Settled { + /// Evaluation order: indices into the conjunct list. + order: Vec<usize>, + /// `true` to run `order` through the compact-once loop; `false` to evaluate + /// the written predicate as-is (see [`settle`]). + compact: bool, +} + +impl AdaptiveFilterShared { + pub(crate) fn new() -> Self { + Self::default() + } + + /// The published settled decision, or `None` if streams are still learning. + fn settled(&self) -> Option<Settled> { + self.inner.lock().expect("poisoned").settled.clone() + } +} + +/// 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 current order and how far it has caught up. +#[derive(Debug)] +pub(crate) struct AdaptiveConjunction { + /// The split conjuncts. `order` indices refer to positions here. + conjuncts: Vec<Arc<dyn PhysicalExpr>>, + /// The written predicate, evaluated as-is when the settled order does not + /// reorder it (so a settled non-reorder costs exactly what the flag-off path + /// costs — no compact-once overhead). + predicate: Arc<dyn PhysicalExpr>, + /// Measurements and the settled decision, shared by every partition stream. + shared: Arc<AdaptiveFilterShared>, + /// Evaluation order: indices into `conjuncts`. The written order until a + /// settled order is adopted. + order: Vec<usize>, + /// Whether the settled order runs through the compact-once loop; `false` + /// means evaluate [`predicate`](Self::predicate) directly. + compact: bool, + /// Shared epoch this stream has caught up to. + epoch_seen: u64, + /// Whether the order is settled (frozen): this stream no longer measures. + settled: bool, +} + +impl AdaptiveConjunction { + /// Build an adaptive evaluator for `predicate`, or `None` if adaptive + /// reordering does not structurally apply: + /// + /// - the predicate has fewer than two `AND` conjuncts (nothing to reorder); + /// - any conjunct is volatile (reordering could change side effects). + /// + /// Whether adaptive reordering is *enabled* is the caller's policy (the + /// config flag lives with `FilterExec`); this constructor only answers + /// whether the predicate is a reorderable conjunction. `shared` is the + /// state common to all partition streams of the owning `FilterExec`. + pub(crate) fn try_new( + predicate: &Arc<dyn PhysicalExpr>, + shared: Arc<AdaptiveFilterShared>, + ) -> Option<Self> { + let conjuncts: Vec<Arc<dyn PhysicalExpr>> = split_conjunction(predicate) + .into_iter() + .map(Arc::clone) + .collect(); + if conjuncts.len() < 2 || conjuncts.iter().any(is_volatile) { + return None; + } + let order = (0..conjuncts.len()).collect(); + Some(Self { + conjuncts, + predicate: Arc::clone(predicate), + shared, + order, + compact: false, + epoch_seen: 0, + settled: false, + }) + } + + /// Evaluate the conjunction against `batch`, returning the boolean mask + /// (over the batch's rows) of rows that passed every conjunct. + /// + /// Until the order settles, each batch is measured and its counts pooled + /// into the shared registry; a stream adopts the settled order another + /// stream published as soon as it sees the epoch advance. + pub(crate) fn evaluate(&mut self, batch: &RecordBatch) -> Result<ArrayRef> { + self.evaluate_traced(batch).map(|(mask, _)| mask) + } + + /// [`evaluate`](Self::evaluate), additionally reporting the + /// [`BatchStrategy`] used for this batch, so the evaluator's behaviour is + /// observable batch by batch (see the scenario tests). + fn evaluate_traced( + &mut self, + batch: &RecordBatch, + ) -> Result<(ArrayRef, BatchStrategy<'_>)> { + // Adopt a settled order another stream published since we last looked: + // one relaxed atomic load per batch, a lock only on the transition. + if !self.settled { + let epoch = self.shared.epoch.load(Ordering::Acquire); + if epoch != self.epoch_seen { + self.epoch_seen = epoch; + if let Some(decision) = self.shared.settled() { + self.adopt(decision); + } + } + } + if self.settled { + let mask = self.evaluate_settled(batch)?; + let strategy = if self.compact { + BatchStrategy::Reordered(&self.order) + } else { + BatchStrategy::Fused + }; + return Ok((mask, strategy)); + } + + // An empty batch measures nothing; evaluating it must not consume the + // warm-up (a run of empty batches would otherwise settle the written + // order on no evidence, permanently). + if batch.num_rows() == 0 { + let mask = eval_conjuncts(&self.conjuncts, &self.order, batch, None)?; + return Ok((mask, BatchStrategy::Measure)); + } + + // Measure this batch into a local accumulator, then pool it. + let mut local = vec![ConjunctStats::default(); self.conjuncts.len()]; + let result = + eval_conjuncts(&self.conjuncts, &self.order, batch, Some(&mut local))?; + self.pool_and_maybe_settle(&local); + Ok((result, BatchStrategy::Measure)) + } + + /// Evaluate the settled arrangement with no instrumentation: the + /// compact-once loop when the order was reordered, or the written predicate + /// directly otherwise (identical to the feature being off). + fn evaluate_settled(&self, batch: &RecordBatch) -> Result<ArrayRef> { + if self.compact { + eval_conjuncts(&self.conjuncts, &self.order, batch, None) + } else { + self.predicate.evaluate(batch)?.into_array(batch.num_rows()) + } + } + + fn adopt(&mut self, decision: Settled) { + self.order = decision.order; + self.compact = decision.compact; + self.settled = true; + } + + /// Merge this batch's measurements into the shared pool and, once enough + /// batches have accrued across all streams, decide and publish the order. + fn pool_and_maybe_settle(&mut self, local: &[ConjunctStats]) { + let mut inner = self.shared.inner.lock().expect("poisoned"); + if inner.stats.len() != local.len() { + inner.stats = vec![ConjunctStats::default(); local.len()]; + } + for (s, l) in inner.stats.iter_mut().zip(local) { + s.merge(l); + } + inner.measured_batches += 1; + if inner.settled.is_some() || inner.measured_batches < WARMUP_BATCHES { + return; + } + let decision = settle(&inner.stats); + inner.settled = Some(decision.clone()); + drop(inner); + self.adopt(decision); + self.shared.epoch.fetch_add(1, Ordering::Release); + } +} + +/// Decide the settled arrangement from the pooled measurements. +/// +/// Rank the conjuncts by effectiveness and adopt the ranking only if it is +/// materially cheaper than the written order. A genuine reorder runs through the +/// compact-once loop (the source of the win); otherwise the written predicate is +/// kept and evaluated as-is. This is the guard: compact-once is only ever used +/// in service of a reorder, so a conjunction that does not benefit from +/// reordering (interchangeable conjuncts, e.g. several equally expensive +/// unselective predicates) pays no compact-once overhead and behaves exactly as +/// it would with the feature off. +fn settle(stats: &[ConjunctStats]) -> 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 { + order: candidate, + compact: true, + } + } else { + Settled { + order: identity, + compact: false, + } + } +} + +/// Evaluate `conjuncts` in `order` against `batch` via the compact-once loop, +/// returning the boolean mask (over the batch's original rows) of rows that +/// passed every conjunct. With `stats`, each conjunct is additionally timed and +/// counted on exactly the rows it evaluated (its marginal selectivity and cost +/// on the current working population). +/// +/// The working batch is physically compacted to the surviving rows only once +/// the accumulated mask becomes selective enough (see +/// [`COMPACTION_SELECTIVITY_THRESHOLD`]); until then masks are combined with a +/// cheap bitwise `AND`, so a run of non-selective conjuncts pays no +/// materialization cost. Unlike a fused `BinaryExpr` chain, survivors stay +/// compacted across the remaining conjuncts instead of being re-evaluated on +/// every row. +fn eval_conjuncts( Review Comment: This feels fairly similar to the code in evaluation of binary_expr: https://github.com/apache/datafusion/blob/b79d3965b7a2c29853e83def04656408d9d7ee7f/datafusion/physical-expr/src/expressions/binary.rs#L295-L294 (the ShortCircuitStrategy in particular) ########## datafusion/physical-plan/src/adaptive_filter.rs: ########## @@ -0,0 +1,956 @@ +// 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). +//! +//! Predicate evaluation order matters: a selective predicate run first gates +//! the work of the predicates after it. DataFusion's `BinaryExpr` `AND` +//! short-circuit only gates on the *leftmost* conjunct, so a conjunction whose +//! selective member is written last (e.g. Review Comment: Another suggestion would be to expand this example more explicitly Something like > For example, given a predicate like > > `regexp_like(s,'a') AND … AND regexp_like(s,'rare')` > > where the `regexp_like` functions are expensive to evaluate and > `regexp_like(s,'rare')`) ` filters out many more rows than `regexp_like(s,'a') ` > this module will reorder the evaluation so the more selective predicate is first. > > regexp_like(s,'rare')` AND … AND`regexp_like(s,'a') ########## datafusion/sqllogictest/test_files/adaptive_filter.slt: ########## @@ -0,0 +1,99 @@ +# 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. + +# Store the table as many small batches so that, with the flag on, the +# adaptive evaluator's warm-up (8 measured batches) completes and the settled +# (possibly reordered) path is exercised end-to-end — not just the measuring +# path a single-batch table would reach. +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 + +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=[] Review Comment: what do you think about adding some sort of metric to the FilterExec that would indicate that the adaptive filter reordering was being applied? ########## datafusion/physical-plan/src/adaptive_filter.rs: ########## Review Comment: See comment below -- I recommend we make this a PhysicalExpr (and put it in physical-expr) rather than in physical-plan -- 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]
