datafusion_physical_optimizer/filter_pushdown.rs
1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements. See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership. The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License. You may obtain a copy of the License at
8//
9// http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied. See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18//! Filter Pushdown Optimization Process
19//!
20//! The filter pushdown mechanism involves four key steps:
21//! 1. **Optimizer Asks Parent for a Filter Pushdown Plan**: The optimizer calls [`ExecutionPlan::gather_filters_for_pushdown`]
22//! on the parent node, passing in parent predicates and phase. The parent node creates a [`FilterDescription`]
23//! by inspecting its logic and children's schemas, determining which filters can be pushed to each child.
24//! 2. **Optimizer Executes Pushdown**: The optimizer recursively calls `push_down_filters` in this module on each child,
25//! passing the appropriate filters (`Vec<Arc<dyn PhysicalExpr>>`) for that child.
26//! 3. **Optimizer Gathers Results**: The optimizer collects [`FilterPushdownPropagation`] results from children,
27//! containing information about which filters were successfully pushed down vs. unsupported.
28//! 4. **Parent Responds**: The optimizer calls [`ExecutionPlan::handle_child_pushdown_result`] on the parent,
29//! passing a [`ChildPushdownResult`] containing the aggregated pushdown outcomes. The parent decides
30//! how to handle filters that couldn't be pushed down (e.g., keep them as FilterExec nodes).
31//!
32//! [`FilterDescription`]: datafusion_physical_plan::filter_pushdown::FilterDescription
33
34use std::sync::Arc;
35
36use crate::PhysicalOptimizerRule;
37
38use datafusion_common::tree_node::{TreeNode, TreeNodeRecursion};
39use datafusion_common::{Result, assert_eq_or_internal_err, config::ConfigOptions};
40use datafusion_physical_expr::PhysicalExpr;
41use datafusion_physical_expr_common::physical_expr::is_volatile;
42use datafusion_physical_plan::ExecutionPlan;
43use datafusion_physical_plan::execution_plan::replace_children_if_necessary;
44use datafusion_physical_plan::filter_pushdown::{
45 ChildFilterPushdownResult, ChildPushdownResult, FilterPushdownPhase,
46 FilterPushdownPropagation, PushedDown,
47};
48
49use itertools::{Itertools, izip};
50
51/// Attempts to recursively push given filters from the top of the tree into leaves.
52///
53/// # Default Implementation
54///
55/// The default implementation in [`ExecutionPlan::gather_filters_for_pushdown`]
56/// and [`ExecutionPlan::handle_child_pushdown_result`] assumes that:
57///
58/// * Parent filters can't be passed onto children (determined by [`ExecutionPlan::gather_filters_for_pushdown`])
59/// * This node has no filters to contribute (determined by [`ExecutionPlan::gather_filters_for_pushdown`]).
60/// * Any filters that could not be pushed down to the children are marked as unsupported (determined by [`ExecutionPlan::handle_child_pushdown_result`]).
61///
62/// # Example: Push filter into a `DataSourceExec`
63///
64/// For example, consider the following plan:
65///
66/// ```text
67/// ┌──────────────────────┐
68/// │ CoalesceBatchesExec │
69/// └──────────────────────┘
70/// │
71/// ▼
72/// ┌──────────────────────┐
73/// │ FilterExec │
74/// │ filters = [ id=1] │
75/// └──────────────────────┘
76/// │
77/// ▼
78/// ┌──────────────────────┐
79/// │ DataSourceExec │
80/// │ projection = * │
81/// └──────────────────────┘
82/// ```
83///
84/// Our goal is to move the `id = 1` filter from the [`FilterExec`] node to the `DataSourceExec` node.
85///
86/// If this filter is selective pushing it into the scan can avoid massive
87/// amounts of data being read from the source (the projection is `*` so all
88/// matching columns are read).
89///
90/// The new plan looks like:
91///
92/// ```text
93/// ┌──────────────────────┐
94/// │ CoalesceBatchesExec │
95/// └──────────────────────┘
96/// │
97/// ▼
98/// ┌──────────────────────┐
99/// │ DataSourceExec │
100/// │ projection = * │
101/// │ filters = [ id=1] │
102/// └──────────────────────┘
103/// ```
104///
105/// # Example: Push filters with `ProjectionExec`
106///
107/// Let's consider a more complex example involving a [`ProjectionExec`]
108/// node in between the [`FilterExec`] and `DataSourceExec` nodes that
109/// creates a new column that the filter depends on.
110///
111/// ```text
112/// ┌──────────────────────┐
113/// │ CoalesceBatchesExec │
114/// └──────────────────────┘
115/// │
116/// ▼
117/// ┌──────────────────────┐
118/// │ FilterExec │
119/// │ filters = │
120/// │ [cost>50,id=1] │
121/// └──────────────────────┘
122/// │
123/// ▼
124/// ┌──────────────────────┐
125/// │ ProjectionExec │
126/// │ cost = price * 1.2 │
127/// └──────────────────────┘
128/// │
129/// ▼
130/// ┌──────────────────────┐
131/// │ DataSourceExec │
132/// │ projection = * │
133/// └──────────────────────┘
134/// ```
135///
136/// We want to push down the filters `[id=1]` to the `DataSourceExec` node,
137/// but can't push down `cost>50` because it requires the [`ProjectionExec`]
138/// node to be executed first. A simple thing to do would be to split up the
139/// filter into two separate filters and push down the first one:
140///
141/// ```text
142/// ┌──────────────────────┐
143/// │ CoalesceBatchesExec │
144/// └──────────────────────┘
145/// │
146/// ▼
147/// ┌──────────────────────┐
148/// │ FilterExec │
149/// │ filters = │
150/// │ [cost>50] │
151/// └──────────────────────┘
152/// │
153/// ▼
154/// ┌──────────────────────┐
155/// │ ProjectionExec │
156/// │ cost = price * 1.2 │
157/// └──────────────────────┘
158/// │
159/// ▼
160/// ┌──────────────────────┐
161/// │ DataSourceExec │
162/// │ projection = * │
163/// │ filters = [ id=1] │
164/// └──────────────────────┘
165/// ```
166///
167/// We can actually however do better by pushing down `price * 1.2 > 50`
168/// instead of `cost > 50`:
169///
170/// ```text
171/// ┌──────────────────────┐
172/// │ CoalesceBatchesExec │
173/// └──────────────────────┘
174/// │
175/// ▼
176/// ┌──────────────────────┐
177/// │ ProjectionExec │
178/// │ cost = price * 1.2 │
179/// └──────────────────────┘
180/// │
181/// ▼
182/// ┌──────────────────────┐
183/// │ DataSourceExec │
184/// │ projection = * │
185/// │ filters = [id=1, │
186/// │ price * 1.2 > 50] │
187/// └──────────────────────┘
188/// ```
189///
190/// # Example: Push filters within a subtree
191///
192/// There are also cases where we may be able to push down filters within a
193/// subtree but not the entire tree. A good example of this is aggregation
194/// nodes:
195///
196/// ```text
197/// ┌──────────────────────┐
198/// │ ProjectionExec │
199/// │ projection = * │
200/// └──────────────────────┘
201/// │
202/// ▼
203/// ┌──────────────────────┐
204/// │ FilterExec │
205/// │ filters = [sum > 10] │
206/// └──────────────────────┘
207/// │
208/// ▼
209/// ┌───────────────────────┐
210/// │ AggregateExec │
211/// │ group by = [id] │
212/// │ aggregate = │
213/// │ [sum(price)] │
214/// └───────────────────────┘
215/// │
216/// ▼
217/// ┌──────────────────────┐
218/// │ FilterExec │
219/// │ filters = [id=1] │
220/// └──────────────────────┘
221/// │
222/// ▼
223/// ┌──────────────────────┐
224/// │ DataSourceExec │
225/// │ projection = * │
226/// └──────────────────────┘
227/// ```
228///
229/// The transformation here is to push down the `id=1` filter to the
230/// `DataSourceExec` node:
231///
232/// ```text
233/// ┌──────────────────────┐
234/// │ ProjectionExec │
235/// │ projection = * │
236/// └──────────────────────┘
237/// │
238/// ▼
239/// ┌──────────────────────┐
240/// │ FilterExec │
241/// │ filters = [sum > 10] │
242/// └──────────────────────┘
243/// │
244/// ▼
245/// ┌───────────────────────┐
246/// │ AggregateExec │
247/// │ group by = [id] │
248/// │ aggregate = │
249/// │ [sum(price)] │
250/// └───────────────────────┘
251/// │
252/// ▼
253/// ┌──────────────────────┐
254/// │ DataSourceExec │
255/// │ projection = * │
256/// │ filters = [id=1] │
257/// └──────────────────────┘
258/// ```
259///
260/// The point here is that:
261/// 1. We cannot push down `sum > 10` through the [`AggregateExec`] node into the `DataSourceExec` node.
262/// Any filters above the [`AggregateExec`] node are not pushed down.
263/// This is determined by calling [`ExecutionPlan::gather_filters_for_pushdown`] on the [`AggregateExec`] node.
264/// 2. We need to keep recursing into the tree so that we can discover the other [`FilterExec`] node and push
265/// down the `id=1` filter.
266///
267/// # Example: Push filters through Joins
268///
269/// It is also possible to push down filters through joins and filters that
270/// originate from joins. For example, a hash join where we build a hash
271/// table of the left side and probe the right side (ignoring why we would
272/// choose this order, typically it depends on the size of each table,
273/// etc.).
274///
275/// ```text
276/// ┌─────────────────────┐
277/// │ FilterExec │
278/// │ filters = │
279/// │ [d.size > 100] │
280/// └─────────────────────┘
281/// │
282/// │
283/// ┌──────────▼──────────┐
284/// │ │
285/// │ HashJoinExec │
286/// │ [u.dept@hash(d.id)] │
287/// │ │
288/// └─────────────────────┘
289/// │
290/// ┌────────────┴────────────┐
291/// ┌──────────▼──────────┐ ┌──────────▼──────────┐
292/// │ DataSourceExec │ │ DataSourceExec │
293/// │ alias [users as u] │ │ alias [dept as d] │
294/// │ │ │ │
295/// └─────────────────────┘ └─────────────────────┘
296/// ```
297///
298/// There are two pushdowns we can do here:
299/// 1. Push down the `d.size > 100` filter through the `HashJoinExec` node to the `DataSourceExec`
300/// node for the `departments` table.
301/// 2. Push down the hash table state from the `HashJoinExec` node to the `DataSourceExec` node to avoid reading
302/// rows from the `users` table that will be eliminated by the join.
303/// This can be done via a bloom filter or similar and is not (yet) supported
304/// in DataFusion. See <https://github.com/apache/datafusion/issues/7955>.
305///
306/// ```text
307/// ┌─────────────────────┐
308/// │ │
309/// │ HashJoinExec │
310/// │ [u.dept@hash(d.id)] │
311/// │ │
312/// └─────────────────────┘
313/// │
314/// ┌────────────┴────────────┐
315/// ┌──────────▼──────────┐ ┌──────────▼──────────┐
316/// │ DataSourceExec │ │ DataSourceExec │
317/// │ alias [users as u] │ │ alias [dept as d] │
318/// │ filters = │ │ filters = │
319/// │ [depg@hash(d.id)] │ │ [ d.size > 100] │
320/// └─────────────────────┘ └─────────────────────┘
321/// ```
322///
323/// You may notice in this case that the filter is *dynamic*: the hash table
324/// is built _after_ the `departments` table is read and at runtime. We
325/// don't have a concrete `InList` filter or similar to push down at
326/// optimization time. These sorts of dynamic filters are handled by
327/// building a specialized [`PhysicalExpr`] that can be evaluated at runtime
328/// and internally maintains a reference to the hash table or other state.
329///
330/// To make working with these sorts of dynamic filters more tractable we have the method [`PhysicalExpr::snapshot`]
331/// which attempts to simplify a dynamic filter into a "basic" non-dynamic filter.
332/// For a join this could mean converting it to an `InList` filter or a min/max filter for example.
333/// See `datafusion/physical-plan/src/dynamic_filters.rs` for more details.
334///
335/// # Example: Push TopK filters into Scans
336///
337/// Another form of dynamic filter is pushing down the state of a `TopK`
338/// operator for queries like `SELECT * FROM t ORDER BY id LIMIT 10`:
339///
340/// ```text
341/// ┌──────────────────────┐
342/// │ TopK │
343/// │ limit = 10 │
344/// │ order by = [id] │
345/// └──────────────────────┘
346/// │
347/// ▼
348/// ┌──────────────────────┐
349/// │ DataSourceExec │
350/// │ projection = * │
351/// └──────────────────────┘
352/// ```
353///
354/// We can avoid large amounts of data processing by transforming this into:
355///
356/// ```text
357/// ┌──────────────────────┐
358/// │ TopK │
359/// │ limit = 10 │
360/// │ order by = [id] │
361/// └──────────────────────┘
362/// │
363/// ▼
364/// ┌──────────────────────┐
365/// │ DataSourceExec │
366/// │ projection = * │
367/// │ filters = │
368/// │ [id < @ TopKHeap] │
369/// └──────────────────────┘
370/// ```
371///
372/// Now as we fill our `TopK` heap we can push down the state of the heap to
373/// the `DataSourceExec` node to avoid reading files / row groups / pages /
374/// rows that could not possibly be in the top 10.
375///
376/// This is not yet implemented in DataFusion. See
377/// <https://github.com/apache/datafusion/issues/15037>
378///
379/// [`PhysicalExpr`]: datafusion_physical_plan::PhysicalExpr
380/// [`PhysicalExpr::snapshot`]: datafusion_physical_plan::PhysicalExpr::snapshot
381/// [`FilterExec`]: datafusion_physical_plan::filter::FilterExec
382/// [`ProjectionExec`]: datafusion_physical_plan::projection::ProjectionExec
383/// [`AggregateExec`]: datafusion_physical_plan::aggregates::AggregateExec
384#[derive(Debug)]
385pub struct FilterPushdown {
386 phase: FilterPushdownPhase,
387 name: String,
388}
389
390impl FilterPushdown {
391 fn new_with_phase(phase: FilterPushdownPhase) -> Self {
392 let name = match phase {
393 FilterPushdownPhase::Pre => "FilterPushdown",
394 FilterPushdownPhase::Post => "FilterPushdown(Post)",
395 }
396 .to_string();
397 Self { phase, name }
398 }
399
400 /// Create a new [`FilterPushdown`] optimizer rule that runs in the pre-optimization phase.
401 /// See [`FilterPushdownPhase`] for more details.
402 pub fn new() -> Self {
403 Self::new_with_phase(FilterPushdownPhase::Pre)
404 }
405
406 /// Create a new [`FilterPushdown`] optimizer rule that runs in the post-optimization phase.
407 /// See [`FilterPushdownPhase`] for more details.
408 pub fn new_post_optimization() -> Self {
409 Self::new_with_phase(FilterPushdownPhase::Post)
410 }
411}
412
413impl Default for FilterPushdown {
414 fn default() -> Self {
415 Self::new()
416 }
417}
418
419impl PhysicalOptimizerRule for FilterPushdown {
420 fn optimize(
421 &self,
422 plan: Arc<dyn ExecutionPlan>,
423 config: &ConfigOptions,
424 ) -> Result<Arc<dyn ExecutionPlan>> {
425 Ok(
426 push_down_filters(&Arc::clone(&plan), vec![], config, self.phase)?
427 .updated_node
428 .unwrap_or(plan),
429 )
430 }
431
432 fn name(&self) -> &str {
433 &self.name
434 }
435
436 fn schema_check(&self) -> bool {
437 true // Filter pushdown does not change the schema of the plan
438 }
439}
440
441fn push_down_filters(
442 node: &Arc<dyn ExecutionPlan>,
443 parent_predicates: Vec<Arc<dyn PhysicalExpr>>,
444 config: &ConfigOptions,
445 phase: FilterPushdownPhase,
446) -> Result<FilterPushdownPropagation<Arc<dyn ExecutionPlan>>> {
447 let mut parent_filter_pushdown_supports: Vec<Vec<PushedDown>> =
448 vec![vec![]; parent_predicates.len()];
449 let mut self_filters_pushdown_supports = vec![];
450 let mut new_children = Vec::with_capacity(node.children().len());
451
452 let children = node.children();
453
454 // Filter out expressions that are not allowed for pushdown
455 let parent_filtered = FilteredVec::new(&parent_predicates, allow_pushdown_for_expr);
456
457 let filter_description = node.gather_filters_for_pushdown(
458 phase,
459 parent_filtered.items().to_vec(),
460 config,
461 )?;
462
463 let filter_description_parent_filters = filter_description.parent_filters();
464 let filter_description_self_filters = filter_description.self_filters();
465 assert_eq_or_internal_err!(
466 filter_description_parent_filters.len(),
467 children.len(),
468 "Filter pushdown expected parent filters count to match number of children for node {}",
469 node.name()
470 );
471 assert_eq_or_internal_err!(
472 filter_description_self_filters.len(),
473 children.len(),
474 "Filter pushdown expected self filters count to match number of children for node {}",
475 node.name()
476 );
477
478 for (child_idx, (child, parent_filters, self_filters)) in izip!(
479 children,
480 filter_description.parent_filters(),
481 filter_description.self_filters()
482 )
483 .enumerate()
484 {
485 // Here, `parent_filters` are the predicates which are provided by the parent node of
486 // the current node, and tried to be pushed down over the child which the loop points
487 // currently. `self_filters` are the predicates which are provided by the current node,
488 // and tried to be pushed down over the child similarly.
489
490 assert_eq_or_internal_err!(
491 parent_filters.len(),
492 parent_filtered.len(),
493 "Filter pushdown expected {} to return one parent filter result per input filter for child {}",
494 node.name(),
495 child_idx
496 );
497
498 // Filter out self_filters that contain volatile expressions and track indices
499 let self_filtered = FilteredVec::new(&self_filters, allow_pushdown_for_expr);
500
501 let num_self_filters = self_filtered.len();
502 let mut all_predicates = self_filtered.items().to_vec();
503
504 // Apply second filter pass: collect indices of parent filters that can be pushed down
505 let parent_filters_for_child = parent_filtered
506 .chain_filter_slice(&parent_filters, |filter| {
507 matches!(filter.discriminant, PushedDown::Yes)
508 });
509
510 // Add the filtered parent predicates to all_predicates
511 for filter in parent_filters_for_child.items() {
512 all_predicates.push(Arc::clone(&filter.predicate));
513 }
514
515 let num_parent_filters = all_predicates.len() - num_self_filters;
516
517 // Any filters that could not be pushed down to a child are marked as not-supported to our parents
518 let result =
519 push_down_filters(&Arc::clone(child), all_predicates, config, phase)?;
520
521 if let Some(new_child) = result.updated_node {
522 // If we have a filter pushdown result, we need to update our children
523 new_children.push(new_child);
524 } else {
525 // If we don't have a filter pushdown result, we need to update our children
526 new_children.push(Arc::clone(child));
527 }
528
529 // Our child doesn't know the difference between filters that were passed down
530 // from our parents and filters that the current node injected. We need to de-entangle
531 // this since we do need to distinguish between them.
532 let mut all_filters = result.filters.into_iter().collect_vec();
533 assert_eq_or_internal_err!(
534 all_filters.len(),
535 num_self_filters + num_parent_filters,
536 "Filter pushdown did not return the expected number of filters from {}",
537 child.name()
538 );
539 let parent_filters = all_filters
540 .split_off(num_self_filters)
541 .into_iter()
542 .collect_vec();
543 // Map the results from filtered self filters back to their original positions using FilteredVec
544 let mapped_self_results =
545 self_filtered.map_results_to_original(all_filters, PushedDown::No);
546
547 // Wrap each result with its corresponding expression
548 let self_filter_results: Vec<_> = mapped_self_results
549 .into_iter()
550 .zip(self_filters)
551 .map(|(support, filter)| support.wrap_expression(filter))
552 .collect();
553
554 self_filters_pushdown_supports.push(self_filter_results);
555
556 // Start by marking all parent filters as unsupported for this child
557 for parent_filter_pushdown_support in parent_filter_pushdown_supports.iter_mut() {
558 parent_filter_pushdown_support.push(PushedDown::No);
559 assert_eq!(
560 parent_filter_pushdown_support.len(),
561 child_idx + 1,
562 "Parent filter pushdown supports should have the same length as the number of children"
563 );
564 }
565 // Map results from pushed-down filters back to original parent filter indices
566 let mapped_parent_results = parent_filters_for_child
567 .map_results_to_original(parent_filters, PushedDown::No);
568
569 // Update parent_filter_pushdown_supports with the mapped results
570 // mapped_parent_results already has the results at their original indices
571 for (idx, support) in parent_filter_pushdown_supports.iter_mut().enumerate() {
572 support[child_idx] = mapped_parent_results[idx];
573 }
574 }
575
576 // Re-create this node with new children
577 let updated_node = replace_children_if_necessary(Arc::clone(node), new_children)?;
578
579 // TODO: by calling `handle_child_pushdown_result` we are assuming that the
580 // `ExecutionPlan` implementation will not change the plan itself.
581 // Should we have a separate method for dynamic pushdown that does not allow modifying the plan?
582 let mut res = updated_node.handle_child_pushdown_result(
583 phase,
584 ChildPushdownResult {
585 parent_filters: parent_predicates
586 .into_iter()
587 .enumerate()
588 .map(
589 |(parent_filter_idx, parent_filter)| ChildFilterPushdownResult {
590 filter: parent_filter,
591 child_results: parent_filter_pushdown_supports[parent_filter_idx]
592 .clone(),
593 },
594 )
595 .collect(),
596 self_filters: self_filters_pushdown_supports,
597 },
598 config,
599 )?;
600 // Compare pointers for new_node and node, if they are different we must replace
601 // ourselves because of changes in our children.
602 if res.updated_node.is_none() && !Arc::ptr_eq(&updated_node, node) {
603 res.updated_node = Some(updated_node)
604 }
605 Ok(res)
606}
607
608/// A helper structure for filtering elements from a vector through multiple passes while
609/// tracking their original indices, allowing results to be mapped back to the original positions.
610struct FilteredVec<T> {
611 items: Vec<T>,
612 // Chain of index mappings: each Vec maps from current level to previous level
613 // index_mappings[0] maps from first filter to original indices
614 // index_mappings[1] maps from second filter to first filter indices, etc.
615 index_mappings: Vec<Vec<usize>>,
616 original_len: usize,
617}
618
619impl<T: Clone> FilteredVec<T> {
620 /// Creates a new FilteredVec by filtering items based on the given predicate
621 fn new<F>(items: &[T], predicate: F) -> Self
622 where
623 F: Fn(&T) -> bool,
624 {
625 let mut filtered_items = Vec::new();
626 let mut original_indices = Vec::new();
627
628 for (idx, item) in items.iter().enumerate() {
629 if predicate(item) {
630 filtered_items.push(item.clone());
631 original_indices.push(idx);
632 }
633 }
634
635 Self {
636 items: filtered_items,
637 index_mappings: vec![original_indices],
638 original_len: items.len(),
639 }
640 }
641
642 /// Returns a reference to the filtered items
643 fn items(&self) -> &[T] {
644 &self.items
645 }
646
647 /// Returns the number of filtered items
648 fn len(&self) -> usize {
649 self.items.len()
650 }
651
652 /// Maps results from the filtered items back to their original positions
653 /// Returns a vector with the same length as the original input, filled with default_value
654 /// and updated with results at their original positions
655 fn map_results_to_original<R: Clone>(
656 &self,
657 results: Vec<R>,
658 default_value: R,
659 ) -> Vec<R> {
660 let mut mapped_results = vec![default_value; self.original_len];
661
662 for (result_idx, result) in results.into_iter().enumerate() {
663 let original_idx = self.trace_to_original_index(result_idx);
664 mapped_results[original_idx] = result;
665 }
666
667 mapped_results
668 }
669
670 /// Traces a filtered index back to its original index through all filter passes
671 fn trace_to_original_index(&self, mut current_idx: usize) -> usize {
672 // Work backwards through the chain of index mappings
673 for mapping in self.index_mappings.iter().rev() {
674 current_idx = mapping[current_idx];
675 }
676 current_idx
677 }
678
679 /// Apply a filter to a new set of items while chaining the index mapping from self (parent)
680 /// This is useful when you have filtered items and then get a transformed slice
681 /// (e.g., from gather_filters_for_pushdown) that you need to filter again
682 fn chain_filter_slice<U: Clone, F>(&self, items: &[U], predicate: F) -> FilteredVec<U>
683 where
684 F: Fn(&U) -> bool,
685 {
686 let mut filtered_items = Vec::new();
687 let mut filtered_indices = Vec::new();
688
689 for (idx, item) in items.iter().enumerate() {
690 if predicate(item) {
691 filtered_items.push(item.clone());
692 filtered_indices.push(idx);
693 }
694 }
695
696 // Chain the index mappings from parent (self)
697 let mut index_mappings = self.index_mappings.clone();
698 index_mappings.push(filtered_indices);
699
700 FilteredVec {
701 items: filtered_items,
702 index_mappings,
703 original_len: self.original_len,
704 }
705 }
706}
707
708fn allow_pushdown_for_expr(expr: &Arc<dyn PhysicalExpr>) -> bool {
709 let mut allow_pushdown = true;
710 expr.apply(|e| {
711 allow_pushdown = allow_pushdown && !is_volatile(e);
712 if allow_pushdown {
713 Ok(TreeNodeRecursion::Continue)
714 } else {
715 Ok(TreeNodeRecursion::Stop)
716 }
717 })
718 .expect("Infallible traversal of PhysicalExpr tree failed");
719 allow_pushdown
720}
721
722#[cfg(test)]
723mod tests {
724 use super::*;
725
726 #[test]
727 fn test_filtered_vec_single_pass() {
728 let items = vec![1, 2, 3, 4, 5, 6];
729 let filtered = FilteredVec::new(&items, |&x| x % 2 == 0);
730
731 // Check filtered items
732 assert_eq!(filtered.items(), &[2, 4, 6]);
733 assert_eq!(filtered.len(), 3);
734
735 // Check index mapping
736 let results = vec!["a", "b", "c"];
737 let mapped = filtered.map_results_to_original(results, "default");
738 assert_eq!(mapped, vec!["default", "a", "default", "b", "default", "c"]);
739 }
740
741 #[test]
742 fn test_filtered_vec_empty_filter() {
743 let items = vec![1, 3, 5];
744 let filtered = FilteredVec::new(&items, |&x| x % 2 == 0);
745
746 assert_eq!(filtered.items(), &[] as &[i32]);
747 assert_eq!(filtered.len(), 0);
748
749 let results: Vec<&str> = vec![];
750 let mapped = filtered.map_results_to_original(results, "default");
751 assert_eq!(mapped, vec!["default", "default", "default"]);
752 }
753
754 #[test]
755 fn test_filtered_vec_all_pass() {
756 let items = vec![2, 4, 6];
757 let filtered = FilteredVec::new(&items, |&x| x % 2 == 0);
758
759 assert_eq!(filtered.items(), &[2, 4, 6]);
760 assert_eq!(filtered.len(), 3);
761
762 let results = vec!["a", "b", "c"];
763 let mapped = filtered.map_results_to_original(results, "default");
764 assert_eq!(mapped, vec!["a", "b", "c"]);
765 }
766
767 #[test]
768 fn test_chain_filter_slice_different_types() {
769 // First pass: filter numbers
770 let numbers = vec![1, 2, 3, 4, 5, 6];
771 let first_pass = FilteredVec::new(&numbers, |&x| x > 3);
772 assert_eq!(first_pass.items(), &[4, 5, 6]);
773
774 // Transform to strings (simulating gather_filters_for_pushdown transformation)
775 let strings = vec!["four", "five", "six"];
776
777 // Second pass: filter strings that contain 'i'
778 let second_pass = first_pass.chain_filter_slice(&strings, |s| s.contains('i'));
779 assert_eq!(second_pass.items(), &["five", "six"]);
780
781 // Map results back to original indices
782 let results = vec![100, 200];
783 let mapped = second_pass.map_results_to_original(results, 0);
784 // "five" was at index 4 (1-based: 5), "six" was at index 5 (1-based: 6)
785 assert_eq!(mapped, vec![0, 0, 0, 0, 100, 200]);
786 }
787
788 #[test]
789 fn test_chain_filter_slice_complex_scenario() {
790 // Simulating the filter pushdown scenario
791 // Parent predicates: [A, B, C, D, E]
792 let parent_predicates = vec!["A", "B", "C", "D", "E"];
793
794 // First pass: filter out some predicates (simulating allow_pushdown_for_expr)
795 let first_pass = FilteredVec::new(&parent_predicates, |s| *s != "B" && *s != "D");
796 assert_eq!(first_pass.items(), &["A", "C", "E"]);
797
798 // After gather_filters_for_pushdown, we get transformed results for a specific child
799 // Let's say child gets [A_transformed, C_transformed, E_transformed]
800 // but only C and E can be pushed down
801 #[derive(Clone, Debug, PartialEq)]
802 struct TransformedPredicate {
803 name: String,
804 can_push: bool,
805 }
806
807 let child_predicates = vec![
808 TransformedPredicate {
809 name: "A_transformed".to_string(),
810 can_push: false,
811 },
812 TransformedPredicate {
813 name: "C_transformed".to_string(),
814 can_push: true,
815 },
816 TransformedPredicate {
817 name: "E_transformed".to_string(),
818 can_push: true,
819 },
820 ];
821
822 // Second pass: filter based on can_push
823 let second_pass =
824 first_pass.chain_filter_slice(&child_predicates, |p| p.can_push);
825 assert_eq!(second_pass.len(), 2);
826 assert_eq!(second_pass.items()[0].name, "C_transformed");
827 assert_eq!(second_pass.items()[1].name, "E_transformed");
828
829 // Simulate getting results back from child
830 let child_results = vec!["C_result", "E_result"];
831 let mapped = second_pass.map_results_to_original(child_results, "no_result");
832
833 // Results should be at original positions: C was at index 2, E was at index 4
834 assert_eq!(
835 mapped,
836 vec![
837 "no_result",
838 "no_result",
839 "C_result",
840 "no_result",
841 "E_result"
842 ]
843 );
844 }
845
846 #[test]
847 fn test_trace_to_original_index() {
848 let items = vec![10, 20, 30, 40, 50];
849 let filtered = FilteredVec::new(&items, |&x| x != 20 && x != 40);
850
851 // filtered items are [10, 30, 50] at original indices [0, 2, 4]
852 assert_eq!(filtered.trace_to_original_index(0), 0); // 10 was at index 0
853 assert_eq!(filtered.trace_to_original_index(1), 2); // 30 was at index 2
854 assert_eq!(filtered.trace_to_original_index(2), 4); // 50 was at index 4
855 }
856
857 #[test]
858 fn test_chain_filter_preserves_original_len() {
859 let items = vec![1, 2, 3, 4, 5];
860 let first = FilteredVec::new(&items, |&x| x > 2);
861
862 let strings = vec!["three", "four", "five"];
863 let second = first.chain_filter_slice(&strings, |s| s.len() == 4);
864
865 // Original length should still be 5
866 let results = vec!["x", "y"];
867 let mapped = second.map_results_to_original(results, "-");
868 assert_eq!(mapped.len(), 5);
869 }
870}