Adaptive (runtime, stats-based) conjunct reordering for FilterExec

[adriangb]

Which issue does this PR close?

• part of [EPIC] Adaptive predicate evaluation #22883
• Related to Reorder boolean expressions (including filter predicates) according to evaluation cost / selectivity #11262
• complements perf: Reorder predicates in conjuncts via simple heuristic #22343
• Depends on (benchmarks only) bench(predicate_eval): add expensive-first and wide-column shapes #25032

Rationale for this change

Predicate evaluation order matters: a selective conjunct run first gates the
work of the conjuncts after it. Two mechanisms already order and gate
conjuncts, and both decide statically:

• the logical optimizer sorts conjuncts cheap-before-expensive by a static
  cost class (perf: Reorder predicates in conjuncts via simple heuristic #22343). It is blind to selectivity, so a cheap-but-unselective
  conjunct still sorts ahead of an expensive-but-very-selective one, and
  conjuncts within one class keep their written order;
• BinaryExpr's AND pre-selects: when the conjuncts evaluated so far keep
  at most 20% of rows (and produce no nulls) it filters the batch before
  evaluating the next one. It cannot gate a conjunct on a more selective one
  written after it.

This PR adds runtime, statistics-based reordering for FilterExec: each
conjunct's selectivity and per-row cost are measured on the rows that reach
it, the conjuncts are ranked by rows discarded per nanosecond, and the ranking
is adopted only if it is materially cheaper than the written order. Once
adopted, the learned order is materialised once as an ordinary AND chain and
evaluated by BinaryExpr like any other predicate. The module contains no
conjunction-evaluation logic of its own on any path; it only measures, ranks,
and builds the chain. It is off by default
(datafusion.execution.adaptive_filter_reordering).

What changes are included in this PR?

Everything lives in a new private module
datafusion/physical-plan/src/adaptive_filter.rs; FilterExec gains a shared
per-execution state field, a metric, and a two-arm match in the stream poll
loop.

• Warm-up: for 8 batches (pooled across all partition streams of the
  operator) the written order is evaluated by BinaryExpr as a right-nested
  AND chain in which each conjunct is wrapped in a small measuring
  expression that records rows seen, rows passed and time. BinaryExpr's
  pre-selection does the compaction, so each conjunct is measured on exactly
  the rows BinaryExpr hands it. The wrappers disappear once the order is
  settled.
• Settle: rank by (1 + rows_in - rows_out) / time (the key Velox uses,
  Pedreira et al. VLDB 2022); adopt the ranking only if its expected cost is
  at least 5% below the written order's. If not, the written order is kept.
• Settled order as a right-nested AND: whichever order settles is built
  once as c1 AND (c2 AND (... AND cn)) and handed to BinaryExpr.
  Right-nesting is what makes this work: BinaryExpr pre-selection filters
  the batch it is given, so the survivors of the first conjunct stay compacted
  through the rest of the chain. A left-nested chain (what the planner and
  conjunction() build) pre-selects on the accumulated prefix and re-filters
  the original batch and scatters back at every level; see the measurements
  below. This also applies when the written order is kept: on
  predicate_eval it made no difference on 20 of 24 shapes and was 21–31%
  faster on the two many-cheap-conjunct shapes whose accumulated prefix
  crosses the 20% pre-selection threshold while no single conjunct does.
• Metric: adaptive_reorders on FilterExec (per partition, only present
  when the flag is on) shows in EXPLAIN ANALYZE whether a reorder was
  adopted.
• Safety rails: volatile predicates are never reordered; reset_state
  gives re-executions fresh measurements; predicate rewrites reset the pooled
  state; results are order-independent.
• Config flag plus regenerated configs.md / information_schema.

Known limitations (documented in the module): measurements are
conditional on the written order, so correlated conjuncts can be misjudged;
the settle is one-shot with no drift re-measurement; the settle cost model
does not yet include evaluation overhead, so on very cheap predicates a
reorder can be adopted that buys nothing (see k4 below).

Measurements behind the settled-path design

Same binary, settled path selected by an environment switch, 8 interleaved
rounds × 40 iterations on predicate_eval, ratios of medians. A = flag off,
B = a dedicated compact-once evaluation loop (an earlier revision of this PR),
C = learned order rebuilt as a left-nested AND, D = learned order rebuilt as
a right-nested AND (this PR).

query	B/A	C/A	D/A
costsel_q01 (5 regexps, selective last)	0.40	0.41	0.41
width q40 / q41 / q42	0.39 / 0.41 / 0.33	0.39 / 0.40 / 0.33	0.39 / 0.40 / 0.33
cardinality k2 / k4 / k8	1.01 / 1.01 / 1.03	0.98 / 1.19 / 1.37	0.98 / 1.13 / 1.04
cardinality k16	0.69	0.98	0.71
q02, q03 (already optimal)	1.00	1.01–1.03	0.99–1.02

The right-nested rebuild matches the dedicated loop everywhere except a
~10% residual on the 0.6 ms k4 query, which is evaluator fixed cost on a
reorder that buys nothing there; tightening the settle guard to account for
evaluation overhead is a follow-up.

tpch_sf10 (same binary, flag off → on): Q6 1.19× faster, Q12 1.45× faster,
all other queries unchanged; tpcds_sf1 and clickbench neutral within the A/A
noise floor. See the bot runs in the PR comments.

Are these changes tested?

• Unit tests for ranking, cost model, warm-up boundary, cross-stream pooling,
  the right-nested shape of the rebuilt chain, the metric transition, lazy
  pool init, and both directions of the fallible-predicate side effect (an
  adopted reorder can introduce or avoid a divide-by-zero).
• An end-to-end flag-on FilterExec test (4 partitions, nullable column,
  repeated execution with and without reset_state).
• adaptive_filter.slt: results identical on and off, EXPLAIN identical on
  and off, and an EXPLAIN ANALYZE assertion that adaptive_reorders fires
  on a predicate written selective-last.

Are there any user-facing changes?

One new config option, datafusion.execution.adaptive_filter_reordering
(experimental, default false), and one new FilterExec metric,
adaptive_reorders. When enabled, query results never change, but the
observable side effects of fallible predicates can, in either direction:
reordering b <> 0 AND 1/b > 2 can make a divide-by-zero error appear or
disappear. Predicates containing volatile expressions are never reordered.

[github-actions]

Thank you for opening this pull request!

Reviewer note: cargo-semver-checks reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch).

Details

     Cloning apache/main
    Building datafusion-common v55.1.0 (current)
       Built [  35.588s] (current)
     Parsing datafusion-common v55.1.0 (current)
      Parsed [   0.064s] (current)
    Building datafusion-common v55.1.0 (baseline)
       Built [  35.198s] (baseline)
     Parsing datafusion-common v55.1.0 (baseline)
      Parsed [   0.065s] (baseline)
    Checking datafusion-common v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.663s] 223 checks: 222 pass, 1 fail, 0 warn, 31 skip

--- failure constructible_struct_adds_field: struct exhaustively constructible through public API adds field ---

Description:
A pub struct that could be exhaustively constructed with a literal using only public API has a new pub field, breaking existing exhaustive literals.
        ref: https://doc.rust-lang.org/reference/expressions/struct-expr.html
       impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.50.0/src/lints/constructible_struct_adds_field.ron

Failed in:
  field ExecutionOptions.adaptive_filter_reordering in /home/runner/work/datafusion/datafusion/datafusion/common/src/config.rs:894

     Summary semver requires new major version: 1 major and 0 minor checks failed
    Finished [  72.856s] datafusion-common
    Building datafusion-physical-expr v55.1.0 (current)
       Built [  30.878s] (current)
     Parsing datafusion-physical-expr v55.1.0 (current)
      Parsed [   0.050s] (current)
    Building datafusion-physical-expr v55.1.0 (baseline)
       Built [  30.107s] (baseline)
     Parsing datafusion-physical-expr v55.1.0 (baseline)
      Parsed [   0.051s] (baseline)
    Checking datafusion-physical-expr v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.334s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [  62.378s] datafusion-physical-expr
    Building datafusion-physical-plan v55.1.0 (current)
       Built [  39.359s] (current)
     Parsing datafusion-physical-plan v55.1.0 (current)
      Parsed [   0.169s] (current)
    Building datafusion-physical-plan v55.1.0 (baseline)
       Built [  39.482s] (baseline)
     Parsing datafusion-physical-plan v55.1.0 (baseline)
      Parsed [   0.161s] (baseline)
    Checking datafusion-physical-plan v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.664s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [  81.274s] datafusion-physical-plan
    Building datafusion-sqllogictest v55.1.0 (current)
       Built [ 103.788s] (current)
     Parsing datafusion-sqllogictest v55.1.0 (current)
      Parsed [   0.022s] (current)
    Building datafusion-sqllogictest v55.1.0 (baseline)
       Built [ 102.972s] (baseline)
     Parsing datafusion-sqllogictest v55.1.0 (baseline)
      Parsed [   0.024s] (baseline)
    Checking datafusion-sqllogictest v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.089s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [ 209.551s] datafusion-sqllogictest

[adriangb]

Can we add doc hidden so this is not as strongly part of the public API?

[adriangb]

Similarly, if we are making this pub just so it is cross-crate can we make this #[doc(hidden)]?