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, Batch, DefaultExpressionEvaluator, ExecResult, PhysicalOperator, RowSchema, SQLParam,
11    ScalarExpr, SharedExpressionEvaluator, SharedRowPredicate,
12};
13
14/// Pipelined `WHERE` operator. Drops rows whose predicate evaluates
15/// to `false` or `NULL`; truthy rows pass through unchanged.
16pub struct Filter<'a> {
17    child: Box<dyn PhysicalOperator + 'a>,
18    condition: FilterCondition<'a>,
19    schema: RowSchema,
20}
21
22enum FilterCondition<'a> {
23    Expression {
24        predicate: ScalarExpr,
25        evaluator: SharedExpressionEvaluator<'a>,
26    },
27    Row(SharedRowPredicate<'a>),
28}
29
30impl Filter<'static> {
31    pub fn new(
32        child: Box<dyn PhysicalOperator>,
33        predicate: ScalarExpr,
34        params: Vec<SQLParam>,
35    ) -> Self {
36        Self::with_evaluator(child, predicate, DefaultExpressionEvaluator::shared(params))
37    }
38}
39
40impl<'a> Filter<'a> {
41    pub fn with_evaluator(
42        child: Box<dyn PhysicalOperator + 'a>,
43        predicate: ScalarExpr,
44        evaluator: SharedExpressionEvaluator<'a>,
45    ) -> Self {
46        let schema = child.row_schema().clone();
47        let predicate = crate::bind_type_introspection(predicate, &schema, evaluator.parameters());
48        Self {
49            child,
50            condition: FilterCondition::Expression {
51                predicate,
52                evaluator,
53            },
54            schema,
55        }
56    }
57
58    pub fn with_row_predicate(
59        child: Box<dyn PhysicalOperator + 'a>,
60        predicate: SharedRowPredicate<'a>,
61    ) -> Self {
62        let schema = child.row_schema().clone();
63        Self {
64            child,
65            condition: FilterCondition::Row(predicate),
66            schema,
67        }
68    }
69}
70
71impl PhysicalOperator for Filter<'_> {
72    fn row_schema(&self) -> &RowSchema {
73        &self.schema
74    }
75
76    fn estimated_cardinality(&self) -> Option<u64> {
77        self.child.estimated_cardinality()
78    }
79
80    fn output_ordering(&self) -> &[crate::PhysicalOrder] {
81        self.child.output_ordering()
82    }
83
84    fn open(&mut self) -> ExecResult<()> {
85        self.child.open()
86    }
87
88    fn next(&mut self) -> ExecResult<Option<Batch>> {
89        loop {
90            let Some(batch) = self.child.next()? else {
91                return Ok(None);
92            };
93            let mut kept = Vec::with_capacity(batch.rows.len());
94            for row in batch.rows {
95                let keep = match &self.condition {
96                    FilterCondition::Expression {
97                        predicate,
98                        evaluator,
99                    } => truthy(&evaluator.evaluate_physical(predicate, &batch.schema, &row)?),
100                    FilterCondition::Row(predicate) => {
101                        predicate.keep_physical(&batch.schema, &row)?
102                    }
103                };
104                if keep {
105                    kept.push(row);
106                }
107            }
108            if !kept.is_empty() {
109                return Ok(Some(Batch::from_physical_rows(self.schema.clone(), kept)));
110            }
111        }
112    }
113
114    fn close(&mut self) -> ExecResult<()> {
115        self.child.close()
116    }
117}
uqa_execution/relational/filter.rs

uqa_execution/relational/
filter.rs
