laminar_core/lookup/predicate/
mod.rs

1//! Predicate types for lookup source pushdown.
2
3use std::fmt;
4
5/// A scalar value used in predicate evaluation.
6///
7/// This enum covers the value types supported by lookup table predicates.
8/// It is intentionally kept small — only types that can be pushed down to
9/// external sources are included.
10#[derive(Debug, Clone, PartialEq)]
11pub enum ScalarValue {
12    /// SQL NULL
13    Null,
14    /// Boolean
15    Bool(bool),
16    /// 64-bit signed integer (covers i8/i16/i32/i64)
17    Int64(i64),
18    /// 64-bit float (covers f32/f64)
19    Float64(f64),
20    /// UTF-8 string
21    Utf8(String),
22    /// Raw binary data
23    Binary(Vec<u8>),
24    /// Timestamp as microseconds since Unix epoch
25    Timestamp(i64),
26}
27
28impl fmt::Display for ScalarValue {
29    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
30        match self {
31            Self::Null => write!(f, "NULL"),
32            Self::Bool(v) => write!(f, "{v}"),
33            Self::Int64(v) => write!(f, "{v}"),
34            Self::Float64(v) => write!(f, "{v}"),
35            Self::Utf8(v) => {
36                // Escape single quotes to prevent SQL injection
37                write!(f, "'{}'", v.replace('\'', "''"))
38            }
39            Self::Binary(v) => write!(f, "X'{}'", hex_encode(v)),
40            Self::Timestamp(us) => write!(f, "TIMESTAMP '{us}'"),
41        }
42    }
43}
44
45/// Encode bytes as lowercase hex string.
46fn hex_encode(bytes: &[u8]) -> String {
47    use std::fmt::Write;
48    bytes
49        .iter()
50        .fold(String::with_capacity(bytes.len() * 2), |mut s, b| {
51            let _ = write!(s, "{b:02x}");
52            s
53        })
54}
55
56/// A filter predicate for lookup table queries.
57///
58/// Each variant maps directly to a SQL comparison operator. The `column`
59/// field names the lookup table column, and the value(s) come from the
60/// stream side of the join.
61#[derive(Debug, Clone, PartialEq)]
62pub enum Predicate {
63    /// `column = value`
64    Eq {
65        /// Column name
66        column: String,
67        /// Value to compare against
68        value: ScalarValue,
69    },
70    /// `column != value`
71    NotEq {
72        /// Column name
73        column: String,
74        /// Value to compare against
75        value: ScalarValue,
76    },
77    /// `column < value`
78    Lt {
79        /// Column name
80        column: String,
81        /// Value to compare against
82        value: ScalarValue,
83    },
84    /// `column <= value`
85    LtEq {
86        /// Column name
87        column: String,
88        /// Value to compare against
89        value: ScalarValue,
90    },
91    /// `column > value`
92    Gt {
93        /// Column name
94        column: String,
95        /// Value to compare against
96        value: ScalarValue,
97    },
98    /// `column >= value`
99    GtEq {
100        /// Column name
101        column: String,
102        /// Value to compare against
103        value: ScalarValue,
104    },
105    /// `column IN (values...)`
106    In {
107        /// Column name
108        column: String,
109        /// Set of values to match against
110        values: Vec<ScalarValue>,
111    },
112    /// `column IS NULL`
113    IsNull {
114        /// Column name
115        column: String,
116    },
117    /// `column IS NOT NULL`
118    IsNotNull {
119        /// Column name
120        column: String,
121    },
122}
123
124impl Predicate {
125    /// Returns the column name this predicate references.
126    #[must_use]
127    pub fn column(&self) -> &str {
128        match self {
129            Self::Eq { column, .. }
130            | Self::NotEq { column, .. }
131            | Self::Lt { column, .. }
132            | Self::LtEq { column, .. }
133            | Self::Gt { column, .. }
134            | Self::GtEq { column, .. }
135            | Self::In { column, .. }
136            | Self::IsNull { column }
137            | Self::IsNotNull { column } => column,
138        }
139    }
140}
141
142/// Capabilities that a lookup source declares for predicate pushdown.
143///
144/// Used by [`split_predicates`] to decide which predicates can be
145/// pushed to the source vs. evaluated locally.
146#[derive(Debug, Clone, Default)]
147pub struct SourceCapabilities {
148    /// Columns that support equality pushdown.
149    pub eq_columns: Vec<String>,
150    /// Columns that support range pushdown (`Lt`, `LtEq`, `Gt`, `GtEq`).
151    pub range_columns: Vec<String>,
152    /// Columns that support IN-list pushdown.
153    pub in_columns: Vec<String>,
154    /// Whether the source supports IS NULL / IS NOT NULL pushdown.
155    pub supports_null_check: bool,
156}
157
158/// Result of splitting predicates into pushable and local sets.
159#[derive(Debug, Clone)]
160pub struct SplitPredicates {
161    /// Predicates that can be pushed down to the source.
162    pub pushable: Vec<Predicate>,
163    /// Predicates that must be evaluated locally after fetching.
164    pub local: Vec<Predicate>,
165}
166
167/// Classify predicates as pushable or local based on source capabilities.
168///
169/// # Arguments
170///
171/// * `predicates` - The full set of predicates from the query plan
172/// * `capabilities` - What the lookup source supports
173///
174/// # Returns
175///
176/// A [`SplitPredicates`] with predicates partitioned into pushable and local.
177#[must_use]
178pub fn split_predicates(
179    predicates: Vec<Predicate>,
180    capabilities: &SourceCapabilities,
181) -> SplitPredicates {
182    let mut pushable = Vec::new();
183    let mut local = Vec::new();
184
185    for pred in predicates {
186        let can_push = match &pred {
187            Predicate::Eq { column, .. } => capabilities.eq_columns.iter().any(|c| c == column),
188            // NotEq cannot use equality indexes — a != b requires a full
189            // scan in most databases. Always evaluate locally.
190            Predicate::NotEq { .. } => false,
191            Predicate::Lt { column, .. }
192            | Predicate::LtEq { column, .. }
193            | Predicate::Gt { column, .. }
194            | Predicate::GtEq { column, .. } => {
195                capabilities.range_columns.iter().any(|c| c == column)
196            }
197            Predicate::In { column, .. } => capabilities.in_columns.iter().any(|c| c == column),
198            Predicate::IsNull { .. } | Predicate::IsNotNull { .. } => {
199                capabilities.supports_null_check
200            }
201        };
202
203        if can_push {
204            pushable.push(pred);
205        } else {
206            local.push(pred);
207        }
208    }
209
210    SplitPredicates { pushable, local }
211}
212
213/// Convert a predicate to a SQL WHERE clause fragment.
214///
215/// This is used by SQL-based lookup sources such as `PostgreSQL` to
216/// construct parameterized queries.
217///
218/// # Returns
219///
220/// A SQL string fragment like `"column = 42"` or `"column IN (1, 2, 3)"`.
221#[must_use]
222pub fn predicate_to_sql(predicate: &Predicate) -> String {
223    let q = |col: &str| col.replace('"', "\"\"");
224    match predicate {
225        Predicate::Eq { column, value } => format!("\"{}\" = {value}", q(column)),
226        Predicate::NotEq { column, value } => format!("\"{}\" != {value}", q(column)),
227        Predicate::Lt { column, value } => format!("\"{}\" < {value}", q(column)),
228        Predicate::LtEq { column, value } => format!("\"{}\" <= {value}", q(column)),
229        Predicate::Gt { column, value } => format!("\"{}\" > {value}", q(column)),
230        Predicate::GtEq { column, value } => format!("\"{}\" >= {value}", q(column)),
231        Predicate::In { column, values } => {
232            let vals: Vec<String> = values.iter().map(ToString::to_string).collect();
233            format!("\"{}\" IN ({})", q(column), vals.join(", "))
234        }
235        Predicate::IsNull { column } => format!("\"{}\" IS NULL", q(column)),
236        Predicate::IsNotNull { column } => format!("\"{}\" IS NOT NULL", q(column)),
237    }
238}
239
240#[cfg(test)]
241mod tests;
laminar_core/lookup/predicate/mod.rs

laminar_core/lookup/predicate/
mod.rs
