uqa_execution/relational/
filter.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Pipelined WHERE filtering.
8
9use super::{
10    truthy, BackwardScanSupport, Batch, DefaultExpressionEvaluator, ExecResult, PhysicalOperator,
11    PhysicalScanDirection, RowSchema, SQLParam, ScalarExpr, SharedExpressionEvaluator,
12    SharedRowPredicate,
13};
14
15/// Pipelined `WHERE` operator. Drops rows whose predicate evaluates
16/// to `false` or `NULL`; truthy rows pass through unchanged.
17pub struct Filter<'a> {
18    child: Box<dyn PhysicalOperator + 'a>,
19    condition: FilterCondition<'a>,
20    schema: RowSchema,
21}
22
23enum FilterCondition<'a> {
24    Expression {
25        predicate: ScalarExpr,
26        evaluator: SharedExpressionEvaluator<'a>,
27    },
28    Row(SharedRowPredicate<'a>),
29}
30
31impl Filter<'static> {
32    pub fn new(
33        child: Box<dyn PhysicalOperator>,
34        predicate: ScalarExpr,
35        params: Vec<SQLParam>,
36    ) -> Self {
37        Self::with_evaluator(child, predicate, DefaultExpressionEvaluator::shared(params))
38    }
39}
40
41impl<'a> Filter<'a> {
42    pub fn with_evaluator(
43        child: Box<dyn PhysicalOperator + 'a>,
44        predicate: ScalarExpr,
45        evaluator: SharedExpressionEvaluator<'a>,
46    ) -> Self {
47        let schema = child.row_schema().clone();
48        let predicate = evaluator.bind_type_introspection(predicate, &schema);
49        Self {
50            child,
51            condition: FilterCondition::Expression {
52                predicate,
53                evaluator,
54            },
55            schema,
56        }
57    }
58
59    pub fn with_row_predicate(
60        child: Box<dyn PhysicalOperator + 'a>,
61        predicate: SharedRowPredicate<'a>,
62    ) -> Self {
63        let schema = child.row_schema().clone();
64        Self {
65            child,
66            condition: FilterCondition::Row(predicate),
67            schema,
68        }
69    }
70
71    fn filter_batch(&self, batch: Batch) -> ExecResult<Option<Batch>> {
72        let mut kept = Vec::with_capacity(batch.rows.len());
73        for row in batch.rows {
74            let keep = match &self.condition {
75                FilterCondition::Expression {
76                    predicate,
77                    evaluator,
78                } => truthy(&evaluator.evaluate_physical(predicate, &batch.schema, &row)?),
79                FilterCondition::Row(predicate) => predicate.keep_physical(&batch.schema, &row)?,
80            };
81            if keep {
82                kept.push(row);
83            }
84        }
85        Ok((!kept.is_empty()).then(|| Batch::from_physical_rows(self.schema.clone(), kept)))
86    }
87}
88
89impl PhysicalOperator for Filter<'_> {
90    fn row_schema(&self) -> &RowSchema {
91        &self.schema
92    }
93
94    fn estimated_cardinality(&self) -> Option<u64> {
95        self.child.estimated_cardinality()
96    }
97
98    fn output_ordering(&self) -> &[crate::PhysicalOrder] {
99        self.child.output_ordering()
100    }
101
102    fn backward_scan_support(&self) -> BackwardScanSupport {
103        if self.child.backward_scan_support() == BackwardScanSupport::Native {
104            BackwardScanSupport::Native
105        } else {
106            BackwardScanSupport::Unsupported
107        }
108    }
109
110    fn open(&mut self) -> ExecResult<()> {
111        self.child.open()
112    }
113
114    fn next(&mut self) -> ExecResult<Option<Batch>> {
115        loop {
116            let Some(batch) = self.child.next()? else {
117                return Ok(None);
118            };
119            if let Some(batch) = self.filter_batch(batch)? {
120                return Ok(Some(batch));
121            }
122        }
123    }
124
125    fn next_direction(&mut self, direction: PhysicalScanDirection) -> ExecResult<Option<Batch>> {
126        loop {
127            let Some(batch) = self.child.next_direction(direction)? else {
128                return Ok(None);
129            };
130            if let Some(batch) = self.filter_batch(batch)? {
131                return Ok(Some(batch));
132            }
133        }
134    }
135
136    fn rewind(&mut self) -> ExecResult<()> {
137        self.child.rewind()
138    }
139
140    fn close(&mut self) -> ExecResult<()> {
141        self.child.close()
142    }
143}
uqa_execution/relational/filter.rs

uqa_execution/relational/
filter.rs
