3232//! which calls [`map`](DecoderProjection::map) on every decoded batch.
3333
3434use std:: sync:: Arc ;
35- use std:: sync:: atomic:: { AtomicBool , Ordering } ;
3635
3736use arrow:: array:: { RecordBatch , RecordBatchOptions } ;
3837use arrow:: compute:: filter_record_batch;
@@ -54,20 +53,6 @@ use crate::ParquetFileMetrics;
5453use crate :: opener:: { VirtualColumnsState , append_fields} ;
5554use crate :: projection_read_plan:: build_projection_read_plan;
5655
57- /// Number of rows the [`PostScanFilter`] must see before it starts
58- /// evaluating its own selectivity — small enough to react to a truly
59- /// non-selective filter within a couple of decoded batches, large
60- /// enough to smooth out a batch that happens to be atypical.
61- const POST_SCAN_FILTER_SAMPLE_ROWS : usize = 65_536 ;
62-
63- /// Selectivity threshold below which [`PostScanFilter`] disables itself
64- /// for the remainder of the file. `0.02` means: if fewer than 2% of the
65- /// sampled rows are being pruned, the per-batch predicate evaluation
66- /// cost outweighs the downstream savings and the filter is a net loss —
67- /// the downstream operator (HashJoin hash lookup, TopK heap compare)
68- /// will do the exact filtering for these rows anyway.
69- const POST_SCAN_FILTER_MIN_SELECTIVITY : f64 = 0.02 ;
70-
7156/// Predicate applied to decoded record batches inside the parquet scan.
7257///
7358/// Semantically identical to a `FilterExec` over the scan: the predicate is
@@ -78,50 +63,19 @@ const POST_SCAN_FILTER_MIN_SELECTIVITY: f64 = 0.02;
7863///
7964/// Holds metric handles so per-batch rows-pruned / matched / time accumulate
8065/// into [`ParquetFileMetrics`] for `EXPLAIN ANALYZE`.
81- ///
82- /// ## Adaptive selectivity gate
83- ///
84- /// After processing [`POST_SCAN_FILTER_SAMPLE_ROWS`] rows the filter
85- /// checks its own selectivity. If it has pruned fewer than
86- /// [`POST_SCAN_FILTER_MIN_SELECTIVITY`] of the sampled rows the per-batch
87- /// evaluation cost has clearly exceeded the downstream savings and the
88- /// filter permanently disables itself for the remainder of the file
89- /// (subsequent calls are a batch-clone fast path).
90- ///
91- /// This is a per-file MVP of the adaptive-predicate-evaluation design in
92- /// the #22883 EPIC — deliberately small so the cross-scan tracker
93- /// (`OptionalFilterPhysicalExpr` / `SelectivityTracker`) from that EPIC
94- /// can supersede it later without churn.
95- ///
96- /// Correctness note: `PostScanFilter` is a semantics-preserving
97- /// optimization (the downstream `HashJoin` hash lookup / `TopK` heap
98- /// compare / etc. does the exact match), so disabling the filter never
99- /// produces wrong results — only slightly more work downstream.
10066pub ( crate ) struct PostScanFilter {
10167 /// Combined predicate, rebased onto the decoder's stream schema.
10268 predicate : Arc < dyn PhysicalExpr > ,
10369 rows_pruned : Count ,
10470 rows_matched : Count ,
10571 eval_time : Time ,
106- /// Once flipped to `true`, [`Self::filter`] short-circuits and
107- /// returns the batch unfiltered. Shared across threads via `Arc`
108- /// because a single `PostScanFilter` instance can be applied to
109- /// multiple decoded streams within a partitioned scan.
110- disabled : Arc < AtomicBool > ,
11172}
11273
11374impl PostScanFilter {
11475 /// Evaluate the predicate on `batch` and return the surviving rows. May
11576 /// return an empty batch — callers should skip empty batches rather than
11677 /// yield them downstream.
11778 pub ( crate ) fn filter ( & self , batch : & RecordBatch ) -> Result < RecordBatch > {
118- // Adaptive fast path: sample said "not selective enough", skip
119- // the whole thing. The downstream operator will do the exact
120- // filtering, so correctness is preserved.
121- if self . disabled . load ( Ordering :: Relaxed ) {
122- return Ok ( batch. clone ( ) ) ;
123- }
124-
12579 // Scoped timer: stops on drop, so the early-return paths still record.
12680 let _timer = self . eval_time . timer ( ) ;
12781
@@ -135,23 +89,8 @@ impl PostScanFilter {
13589 let filtered = filter_record_batch ( batch, mask) ?;
13690
13791 let kept = filtered. num_rows ( ) ;
138- let pruned = input_rows - kept;
13992 self . rows_matched . add ( kept) ;
140- self . rows_pruned . add ( pruned) ;
141-
142- // Adaptive decision: once we have seen enough rows, judge whether
143- // the filter is earning its keep. Cheap load — `Count` is backed
144- // by an atomic, `total` is a local u64.
145- let total_matched = self . rows_matched . value ( ) ;
146- let total_pruned = self . rows_pruned . value ( ) ;
147- let total = total_matched + total_pruned;
148- if total >= POST_SCAN_FILTER_SAMPLE_ROWS {
149- let selectivity = total_pruned as f64 / total as f64 ;
150- if selectivity < POST_SCAN_FILTER_MIN_SELECTIVITY {
151- self . disabled . store ( true , Ordering :: Relaxed ) ;
152- }
153- }
154-
93+ self . rows_pruned . add ( input_rows - kept) ;
15594 Ok ( filtered)
15695 }
15796}
@@ -292,7 +231,6 @@ impl DecoderProjection {
292231 rows_pruned : file_metrics. post_scan_rows_pruned . clone ( ) ,
293232 rows_matched : file_metrics. post_scan_rows_matched . clone ( ) ,
294233 eval_time : file_metrics. post_scan_filter_eval_time . clone ( ) ,
295- disabled : Arc :: new ( AtomicBool :: new ( false ) ) ,
296234 } )
297235 } ;
298236
@@ -345,132 +283,3 @@ impl DecoderProjection {
345283 ) ?)
346284 }
347285}
348-
349- #[ cfg( test) ]
350- mod tests {
351- use super :: * ;
352-
353- use arrow:: array:: { ArrayRef , Int32Array } ;
354- use arrow:: datatypes:: { DataType , Field , Schema } ;
355- use datafusion_common:: ScalarValue ;
356- use datafusion_expr:: Operator ;
357- use datafusion_physical_expr:: PhysicalExpr ;
358- use datafusion_physical_expr:: expressions:: { BinaryExpr , Column , lit} ;
359-
360- /// Build a `PostScanFilter` that evaluates `predicate` on a single
361- /// `i32` column named "a", with fresh metric handles and the
362- /// adaptive gate armed.
363- fn test_filter ( predicate : Arc < dyn PhysicalExpr > ) -> PostScanFilter {
364- PostScanFilter {
365- predicate,
366- rows_pruned : Count :: new ( ) ,
367- rows_matched : Count :: new ( ) ,
368- eval_time : Time :: new ( ) ,
369- disabled : Arc :: new ( AtomicBool :: new ( false ) ) ,
370- }
371- }
372-
373- fn schema ( ) -> SchemaRef {
374- Arc :: new ( Schema :: new ( vec ! [ Field :: new( "a" , DataType :: Int32 , false ) ] ) )
375- }
376-
377- /// Build a `RecordBatch` with a single `Int32` column named "a",
378- /// filled with `size` copies of `value`.
379- fn batch_of ( value : i32 , size : usize ) -> RecordBatch {
380- let arr: ArrayRef = Arc :: new ( Int32Array :: from ( vec ! [ value; size] ) ) as ArrayRef ;
381- RecordBatch :: try_new ( schema ( ) , vec ! [ arr] ) . unwrap ( )
382- }
383-
384- /// `a = 1` — matches every row when the batch is all-1s
385- /// (selectivity = 0 → adaptive should disable) and matches no row
386- /// when the batch is all-2s (selectivity = 100% → adaptive keeps
387- /// the filter on).
388- fn eq_one ( ) -> Arc < dyn PhysicalExpr > {
389- let schema = schema ( ) ;
390- Arc :: new ( BinaryExpr :: new (
391- Arc :: new ( Column :: new ( "a" , 0 ) ) ,
392- Operator :: Eq ,
393- lit ( ScalarValue :: Int32 ( Some ( 1 ) ) ) ,
394- ) ) as Arc < dyn PhysicalExpr >
395- }
396-
397- /// Zero-selectivity filter (`predicate matches every row`): after
398- /// [`POST_SCAN_FILTER_SAMPLE_ROWS`] rows the adaptive gate must
399- /// flip to `disabled = true` and subsequent batches must be
400- /// returned unfiltered as a fast path.
401- #[ test]
402- fn adaptive_disables_when_selectivity_is_zero ( ) {
403- let filter = test_filter ( eq_one ( ) ) ;
404- let batch = batch_of ( 1 , 8192 ) ; // predicate matches all → prune 0
405-
406- // Feed enough rows to trip the sample threshold.
407- let rounds = POST_SCAN_FILTER_SAMPLE_ROWS . div_ceil ( batch. num_rows ( ) ) ;
408- for _ in 0 ..rounds {
409- let out = filter. filter ( & batch) . unwrap ( ) ;
410- assert_eq ! ( out. num_rows( ) , batch. num_rows( ) ) ;
411- }
412-
413- assert ! (
414- filter. disabled. load( Ordering :: Relaxed ) ,
415- "0% pruning must trip the adaptive gate"
416- ) ;
417- }
418-
419- /// High-selectivity filter (`predicate matches no row`): even
420- /// after the sample threshold is exceeded the filter is
421- /// obviously earning its keep and must stay enabled.
422- #[ test]
423- fn adaptive_stays_enabled_when_selectivity_is_high ( ) {
424- let filter = test_filter ( eq_one ( ) ) ;
425- let batch = batch_of ( 2 , 8192 ) ; // predicate matches none → prune 100%
426-
427- let rounds = POST_SCAN_FILTER_SAMPLE_ROWS . div_ceil ( batch. num_rows ( ) ) ;
428- for _ in 0 ..rounds {
429- let out = filter. filter ( & batch) . unwrap ( ) ;
430- assert_eq ! ( out. num_rows( ) , 0 ) ;
431- }
432-
433- assert ! (
434- !filter. disabled. load( Ordering :: Relaxed ) ,
435- "100% pruning must not trip the adaptive gate"
436- ) ;
437- }
438-
439- /// Sample size gate: below [`POST_SCAN_FILTER_SAMPLE_ROWS`] the
440- /// decision is deferred, no matter how bad the selectivity looks
441- /// on the tiny sample so far. Prevents a single unlucky early
442- /// batch from taking the filter down for the whole file.
443- #[ test]
444- fn adaptive_does_not_disable_before_sample_threshold ( ) {
445- let filter = test_filter ( eq_one ( ) ) ;
446- // Feed a small amount of "0% selectivity" data — well under
447- // the sample threshold.
448- let batch = batch_of ( 1 , 1024 ) ;
449- filter. filter ( & batch) . unwrap ( ) ;
450-
451- assert ! (
452- !filter. disabled. load( Ordering :: Relaxed ) ,
453- "must not disable before sampling {POST_SCAN_FILTER_SAMPLE_ROWS} rows" ,
454- ) ;
455- }
456-
457- /// Once disabled, a subsequent call must be a pure fast path:
458- /// the batch flows through unfiltered and — importantly — the
459- /// predicate is never evaluated. Verified indirectly by
460- /// installing a `lit(false)` predicate (which would zero out
461- /// every row if it ran) and asserting the row count survives.
462- #[ test]
463- fn adaptive_disable_short_circuits_evaluation ( ) {
464- let filter =
465- test_filter ( lit ( ScalarValue :: Boolean ( Some ( false ) ) ) as Arc < dyn PhysicalExpr > ) ;
466- filter. disabled . store ( true , Ordering :: Relaxed ) ;
467-
468- let batch = batch_of ( 42 , 128 ) ;
469- let out = filter. filter ( & batch) . unwrap ( ) ;
470- assert_eq ! (
471- out. num_rows( ) ,
472- batch. num_rows( ) ,
473- "disabled filter must bypass predicate evaluation"
474- ) ;
475- }
476- }
0 commit comments