Per-conjunct pruning stats via PruningPredicateTree + observer

[adriangb]

Which issue does this PR close?

• Supersedes Per-conjunct pruning statistics for PruningPredicate #22235 (same goal — per-conjunct pruning stats for adaptive filter scheduling — with a different, more reusable API). Please close Per-conjunct pruning statistics for PruningPredicate #22235 in favor of this if the direction lands.
• Part of [Experiment] Adaptive filter pushdown #22144 (Adaptive filter pushdown).

Rationale for this change

#22235 surfaced per-conjunct pruning effectiveness by bolting a second, AND-only code path onto PruningPredicate (a marker wrapper holding sub_predicates, a placeholder true literal as its predicate_expr, and a prune_per_conjunct that re-walks the leaves a second time). Review feedback (thanks @asolimando) flagged three things, all tracing back to that shape:

1. Double traversal — prune walks the leaves, then prune_per_conjunct walks them again.
2. Duplicated page-pruning method (prune_plan_with_per_conjunct_stats) that is a near-copy of the untagged path.
3. The wrapper's predicate_expr() / literal_guarantees() / required_columns() silently return placeholder values, which is misuse-prone.

Plus two design questions: AND-only (OR/NOT?), and order-dependent stats from the page-index short-circuit (later conjuncts get biased rows_seen).

This PR reshapes the feature around a small, single-pass, observer-driven primitive that resolves all of the above.

What changes are included in this PR?

Two reviewable commits, each builds green:

1. PruningConjunction (datafusion-pruning) — an AND of tagged PruningPredicate leaves. One evaluation pass ANDs the leaves' masks and fires PruningObserver::on_leaf(tag, mask) only for leaves actually evaluated; the AND short-circuits once everything is pruned, so stats are never biased by predicates that didn't run. Built with a small builder:

   let conj = PruningConjunction::builder(schema)
       .push(filter_id, expr)   // runs try_new, skips always-true / errors
       .build();                // Option<PruningConjunction>

   PruningConjunction::single(predicate) wraps one existing predicate for the untagged path (prune() + NoopObserver, zero overhead). ConjunctStatsObserver accumulates PerConjunctPruneStats { tag, containers_seen, containers_pruned }.

   Deliberately AND-only. OR / NOT are handled inside a leaf via PruningPredicate::try_new, not as structural nodes, because (a) try_new(a < 5 OR b > 100) prunes better than ORing two leaf masks, and (b) the per-leaf "containers pruned" count is only sound under monotonic AND — under OR a leaf only helps where every sibling also prunes, so a per-leaf stat would mislead a scheduler. Per-branch OR short-circuit, where it is genuinely useful, belongs to the row-filter decode path over exact masks — a separate concern that should land with its consumer (the adaptive scheduler), not here.

2. Parquet row-group + page filters consume the conjunction via *_with_observer methods. The page-index loop has one shared body (no duplicated method); its !selects_any short-circuit fires after a leaf was observed, so unevaluated leaves are correctly absent from stats.

How this maps to the review on #22235

Concern	Resolution
Double traversal	Single observer-driven pass
Duplicated page method	One shared body, observer param
Placeholder predicate_expr	Gone — leaves are real, standalone PruningPredicates
AND-only (Q1)	Intentional: OR/NOT fold into leaves (prune better, sound stats). Exact-mask OR short-circuit is a decode-path concern, deferred.
Order-dependent stats (Q2)	Short-circuited leaves are never observed

Are these changes tested?

Yes — unit tests for PruningConjunction (single-leaf equivalence to a plain PruningPredicate, AND short-circuit leaving later-leaf stats unbiased, builder stat accumulation, always-true dropping, all-trivial → None) and a row-group filter test asserting the observer surfaces correct per-conjunct stats while preserving the existing pruning result. All existing datafusion-datasource-parquet lib tests pass unchanged.

Are there any user-facing changes?

New public API in datafusion-pruning (PruningConjunction, PruningConjunctionBuilder, PruningObserver, NoopObserver, ConjunctStatsObserver, PerConjunctPruneStats, Tag) and an observer-accepting prune method on the parquet row-group / page filters. Purely additive; existing prune / prune_by_statistics / prune_plan_with_page_index paths are unchanged.

🤖 Generated with Claude Code