linkrs_query/executor/streaming/
context.rs

1//! ValueRowContext: Expression evaluation context for streaming rows
2//!
3//! Provides expression evaluation context for Vec<Value> rows.
4//! Runtime uses SlotId-based access via SlotLayout; column-name lookup
5//! at runtime is prohibited — all Variable references are resolved through
6//! the layout.
7
8use crate::executor::expression::evaluator::traits::ExpressionContext;
9use crate::executor::expression::ExpressionError;
10use crate::executor::streaming::slot::{SlotId, SlotLayout};
11use crate::executor::streaming::subquery::{EvalEnv, SubqueryExecutor};
12use linkrs_core::Value;
13use std::collections::HashMap;
14use std::sync::Arc;
15
16use super::parameters::{ParameterFrame, ParameterSchema};
17
18/// Row context for expression evaluation with Value types
19///
20/// Every context carries an [`Arc<SlotLayout>`] so that `get_variable(name)`
21/// resolves through `layout.slot_id(name)` → `row[slot]` without any
22/// separate name-to-index map or runtime fallback.
23pub struct ValueRowContext {
24    /// Column values (as Values, no conversion needed)
25    row: Vec<Value>,
26    /// Extra variables for expression evaluation
27    variables: HashMap<String, Value>,
28    /// Slot layout — always set; drives all variable resolution
29    layout: Arc<SlotLayout>,
30    /// Parameter name→value map (shared via Arc across rows).
31    parameters: Option<Arc<HashMap<String, Value>>>,
32    /// Session variable name→value snapshot (shared via Arc across rows).
33    session_variables: Option<Arc<HashMap<String, Value>>>,
34    /// Expression-level subquery executor of the hosting operator.
35    subquery_executor: Option<Arc<SubqueryExecutor>>,
36    /// Sequence catalog manager for `curr_val` / `next_val`.
37    sequence_manager: Option<Arc<linkrs_core::metadata::SequenceManager>>,
38}
39
40impl ValueRowContext {
41    /// Create a new context from a row and slot layout.
42    pub fn new(row: Vec<Value>, layout: Arc<SlotLayout>) -> Self {
43        Self {
44            row,
45            variables: HashMap::new(),
46            layout,
47            parameters: None,
48            session_variables: None,
49            subquery_executor: None,
50            sequence_manager: None,
51        }
52    }
53
54    /// Create a new context with parameter values for `@name` resolution.
55    pub fn with_parameters(
56        row: Vec<Value>,
57        layout: Arc<SlotLayout>,
58        parameters: Arc<HashMap<String, Value>>,
59    ) -> Self {
60        Self {
61            row,
62            variables: HashMap::new(),
63            layout,
64            parameters: Some(parameters),
65            session_variables: None,
66            subquery_executor: None,
67            sequence_manager: None,
68        }
69    }
70
71    /// Create a new context from a full evaluation environment.
72    pub fn with_env(row: Vec<Value>, layout: Arc<SlotLayout>, env: &EvalEnv) -> Self {
73        Self {
74            row,
75            variables: HashMap::new(),
76            layout,
77            parameters: env.params.clone(),
78            session_variables: env.session_variables.clone(),
79            subquery_executor: env.subquery_executor.clone(),
80            sequence_manager: env.sequence_manager.clone(),
81        }
82    }
83
84    /// Build a name→value map from schema and frame.
85    pub fn build_parameter_map(
86        schema: &ParameterSchema,
87        frame: &ParameterFrame,
88    ) -> HashMap<String, Value> {
89        schema
90            .params
91            .iter()
92            .filter_map(|p| frame.get(p.slot).map(|v| (p.name.clone(), v.clone())))
93            .collect()
94    }
95
96    /// Create a new context by building a layout from column names.
97    ///
98    /// This is a convenience for sites that only have column-name strings
99    /// (e.g. helper functions).  The layout is
100    /// created once and reused for all subsequent `get_variable()` calls
101    /// so there is never a name-map fallback at runtime.
102    pub fn from_names(row: Vec<Value>, col_names: Vec<String>) -> Self {
103        let layout = Arc::new(SlotLayout::from_names(&col_names));
104        Self::new(row, layout)
105    }
106}
107
108impl ExpressionContext for ValueRowContext {
109    fn get_variable(&self, name: &str) -> Option<Value> {
110        // First check explicit variables (e.g. bindings from Apply)
111        if let Some(value) = self.variables.get(name) {
112            return Some(value.clone());
113        }
114        // Slot-based access via layout — the ONLY resolution path at runtime.
115        self.layout
116            .slot_id(name)
117            .and_then(|slot_id| self.row.get(slot_id).cloned())
118    }
119
120    fn get_variable_by_slot(&self, slot: SlotId) -> Option<Value> {
121        self.row.get(slot).cloned()
122    }
123
124    fn set_variable(&mut self, name: String, value: Value) {
125        self.variables.insert(name, value);
126    }
127
128    fn get_parameter(&self, name: &str) -> Option<Value> {
129        self.parameters.as_ref().and_then(|p| p.get(name).cloned())
130    }
131
132    fn get_session_variable(&self, name: &str) -> Result<Value, ExpressionError> {
133        self.session_variables
134            .as_ref()
135            .and_then(|vars| vars.get(name).cloned())
136            .ok_or_else(|| {
137                ExpressionError::type_error(format!("Session variable `{}` is not defined", name))
138            })
139    }
140
141    fn get_sequence_manager(&self) -> Option<Arc<linkrs_core::metadata::SequenceManager>> {
142        self.sequence_manager.clone()
143    }
144
145    fn execute_exists(
146        &mut self,
147        body: &linkrs_core::types::expr::SubqueryBody,
148    ) -> Result<bool, ExpressionError> {
149        self.subquery_executor
150            .as_ref()
151            .ok_or_else(|| {
152                ExpressionError::type_error("Subquery execution not supported in this context")
153            })?
154            .execute_exists(body, self.layout.clone(), self.row.clone())
155    }
156
157    fn execute_scalar_subquery(
158        &mut self,
159        body: &linkrs_core::types::expr::SubqueryBody,
160    ) -> Result<Value, ExpressionError> {
161        self.subquery_executor
162            .as_ref()
163            .ok_or_else(|| {
164                ExpressionError::type_error(
165                    "Scalar subquery execution not supported in this context",
166                )
167            })?
168            .execute_scalar(body, self.layout.clone(), self.row.clone())
169    }
170
171    fn contains_subquery(
172        &mut self,
173        body: &linkrs_core::types::expr::SubqueryBody,
174        value: &Value,
175    ) -> Result<Value, ExpressionError> {
176        if value.is_null() {
177            return Ok(Value::Bool(false));
178        }
179        let found = self
180            .subquery_executor
181            .as_ref()
182            .ok_or_else(|| {
183                ExpressionError::type_error("Subquery execution not supported in this context")
184            })?
185            .execute_contains(body, self.layout.clone(), self.row.clone(), value)?;
186        Ok(Value::Bool(found))
187    }
188}
189
190/// A light-weight expression context that borrows the row data instead
191/// of cloning it.  Use this in hot paths (e.g. Filter) where the
192/// expression only reads from the row and does not take ownership.
193///
194/// Individual values returned by `get_variable` are still cloned (that
195/// is inherent in the `ExpressionContext` trait), but the `Vec<Value>`
196/// wrapper itself is not — saving one allocation per row.
197pub struct BorrowedRowContext<'a> {
198    row: &'a [Value],
199    variables: HashMap<String, Value>,
200    layout: Arc<SlotLayout>,
201    parameters: Option<Arc<HashMap<String, Value>>>,
202    /// Session variable name→value snapshot (shared via Arc across rows).
203    session_variables: Option<Arc<HashMap<String, Value>>>,
204    /// Expression-level subquery executor of the hosting operator.
205    subquery_executor: Option<Arc<SubqueryExecutor>>,
206    /// Sequence catalog manager for `curr_val` / `next_val`.
207    sequence_manager: Option<Arc<linkrs_core::metadata::SequenceManager>>,
208}
209
210impl<'a> BorrowedRowContext<'a> {
211    pub fn new(row: &'a [Value], layout: Arc<SlotLayout>) -> Self {
212        Self {
213            row,
214            variables: HashMap::new(),
215            layout,
216            parameters: None,
217            session_variables: None,
218            subquery_executor: None,
219            sequence_manager: None,
220        }
221    }
222
223    /// Create with parameter values for `@name` resolution.
224    pub fn with_parameters(
225        row: &'a [Value],
226        layout: Arc<SlotLayout>,
227        parameters: Arc<HashMap<String, Value>>,
228    ) -> Self {
229        Self {
230            row,
231            variables: HashMap::new(),
232            layout,
233            parameters: Some(parameters),
234            session_variables: None,
235            subquery_executor: None,
236            sequence_manager: None,
237        }
238    }
239
240    /// Create from a full evaluation environment (parameters + subquery
241    /// executor).
242    pub fn with_env(row: &'a [Value], layout: Arc<SlotLayout>, env: &EvalEnv) -> Self {
243        Self {
244            row,
245            variables: HashMap::new(),
246            layout,
247            parameters: env.params.clone(),
248            session_variables: env.session_variables.clone(),
249            subquery_executor: env.subquery_executor.clone(),
250            sequence_manager: env.sequence_manager.clone(),
251        }
252    }
253
254    /// Update the row reference and clear per-row variables.
255    ///
256    /// Call this in a hot loop to reuse the same context across many rows,
257    /// avoiding repeated `Arc::clone` on the slot layout.
258    pub fn set_row(&mut self, row: &'a [Value]) {
259        self.row = row;
260        self.variables.clear();
261    }
262}
263
264impl ExpressionContext for BorrowedRowContext<'_> {
265    fn get_variable(&self, name: &str) -> Option<Value> {
266        if let Some(value) = self.variables.get(name) {
267            return Some(value.clone());
268        }
269        self.layout
270            .slot_id(name)
271            .and_then(|slot_id| self.row.get(slot_id).cloned())
272    }
273
274    fn get_variable_by_slot(&self, slot: SlotId) -> Option<Value> {
275        self.row.get(slot).cloned()
276    }
277
278    fn set_variable(&mut self, name: String, value: Value) {
279        self.variables.insert(name, value);
280    }
281
282    fn get_parameter(&self, name: &str) -> Option<Value> {
283        self.parameters.as_ref().and_then(|p| p.get(name).cloned())
284    }
285
286    fn get_session_variable(&self, name: &str) -> Result<Value, ExpressionError> {
287        self.session_variables
288            .as_ref()
289            .and_then(|vars| vars.get(name).cloned())
290            .ok_or_else(|| {
291                ExpressionError::type_error(format!("Session variable `{}` is not defined", name))
292            })
293    }
294
295    fn get_sequence_manager(&self) -> Option<Arc<linkrs_core::metadata::SequenceManager>> {
296        self.sequence_manager.clone()
297    }
298
299    fn execute_exists(
300        &mut self,
301        body: &linkrs_core::types::expr::SubqueryBody,
302    ) -> Result<bool, ExpressionError> {
303        self.subquery_executor
304            .as_ref()
305            .ok_or_else(|| {
306                ExpressionError::type_error("Subquery execution not supported in this context")
307            })?
308            .execute_exists(body, self.layout.clone(), self.row.to_vec())
309    }
310
311    fn execute_scalar_subquery(
312        &mut self,
313        body: &linkrs_core::types::expr::SubqueryBody,
314    ) -> Result<Value, ExpressionError> {
315        self.subquery_executor
316            .as_ref()
317            .ok_or_else(|| {
318                ExpressionError::type_error(
319                    "Scalar subquery execution not supported in this context",
320                )
321            })?
322            .execute_scalar(body, self.layout.clone(), self.row.to_vec())
323    }
324
325    fn contains_subquery(
326        &mut self,
327        body: &linkrs_core::types::expr::SubqueryBody,
328        value: &Value,
329    ) -> Result<Value, ExpressionError> {
330        if value.is_null() {
331            return Ok(Value::Bool(false));
332        }
333        let found = self
334            .subquery_executor
335            .as_ref()
336            .ok_or_else(|| {
337                ExpressionError::type_error("Subquery execution not supported in this context")
338            })?
339            .execute_contains(body, self.layout.clone(), self.row.to_vec(), value)?;
340        Ok(Value::Bool(found))
341    }
342}
343
344/// A context that presents two disjoint row halves as a single combined row
345/// without allocating a combined `Vec<Value>`.
346///
347/// Slots `0..split` map to the left slice; slots `split..` map to the right
348/// slice.  Use this in hash join condition evaluation to avoid cloning
349/// every column before the condition is checked.
350pub struct SplitRowContext<'a> {
351    left: &'a [Value],
352    right: &'a [Value],
353    split: usize,
354    variables: HashMap<String, Value>,
355    layout: Arc<SlotLayout>,
356    parameters: Option<Arc<HashMap<String, Value>>>,
357    /// Session variable name→value snapshot (shared via Arc across rows).
358    session_variables: Option<Arc<HashMap<String, Value>>>,
359    /// Expression-level subquery executor of the hosting operator.
360    subquery_executor: Option<Arc<SubqueryExecutor>>,
361    /// Sequence catalog manager for `curr_val` / `next_val`.
362    sequence_manager: Option<Arc<linkrs_core::metadata::SequenceManager>>,
363}
364
365impl<'a> SplitRowContext<'a> {
366    pub fn new(left: &'a [Value], right: &'a [Value], layout: Arc<SlotLayout>) -> Self {
367        let split = left.len();
368        Self {
369            left,
370            right,
371            split,
372            variables: HashMap::new(),
373            layout,
374            parameters: None,
375            session_variables: None,
376            subquery_executor: None,
377            sequence_manager: None,
378        }
379    }
380
381    /// Create from a full evaluation environment (parameters + subquery
382    /// executor).
383    pub fn with_env(
384        left: &'a [Value],
385        right: &'a [Value],
386        layout: Arc<SlotLayout>,
387        env: &EvalEnv,
388    ) -> Self {
389        let split = left.len();
390        Self {
391            left,
392            right,
393            split,
394            variables: HashMap::new(),
395            layout,
396            parameters: env.params.clone(),
397            session_variables: env.session_variables.clone(),
398            subquery_executor: env.subquery_executor.clone(),
399            sequence_manager: env.sequence_manager.clone(),
400        }
401    }
402}
403
404impl ExpressionContext for SplitRowContext<'_> {
405    fn get_variable(&self, name: &str) -> Option<Value> {
406        if let Some(value) = self.variables.get(name) {
407            return Some(value.clone());
408        }
409        self.layout
410            .slot_id(name)
411            .and_then(|slot_id| self.get_variable_by_slot(slot_id))
412    }
413
414    fn get_variable_by_slot(&self, slot: SlotId) -> Option<Value> {
415        if slot < self.split {
416            self.left.get(slot).cloned()
417        } else {
418            self.right.get(slot - self.split).cloned()
419        }
420    }
421
422    fn set_variable(&mut self, name: String, value: Value) {
423        self.variables.insert(name, value);
424    }
425
426    fn get_parameter(&self, name: &str) -> Option<Value> {
427        self.parameters.as_ref().and_then(|p| p.get(name).cloned())
428    }
429
430    fn get_session_variable(&self, name: &str) -> Result<Value, ExpressionError> {
431        self.session_variables
432            .as_ref()
433            .and_then(|vars| vars.get(name).cloned())
434            .ok_or_else(|| {
435                ExpressionError::type_error(format!("Session variable `{}` is not defined", name))
436            })
437    }
438
439    fn get_sequence_manager(&self) -> Option<Arc<linkrs_core::metadata::SequenceManager>> {
440        self.sequence_manager.clone()
441    }
442
443    fn execute_exists(
444        &mut self,
445        body: &linkrs_core::types::expr::SubqueryBody,
446    ) -> Result<bool, ExpressionError> {
447        let mut row = self.left.to_vec();
448        row.extend_from_slice(self.right);
449        self.subquery_executor
450            .as_ref()
451            .ok_or_else(|| {
452                ExpressionError::type_error("Subquery execution not supported in this context")
453            })?
454            .execute_exists(body, self.layout.clone(), row)
455    }
456
457    fn execute_scalar_subquery(
458        &mut self,
459        body: &linkrs_core::types::expr::SubqueryBody,
460    ) -> Result<Value, ExpressionError> {
461        let mut row = self.left.to_vec();
462        row.extend_from_slice(self.right);
463        self.subquery_executor
464            .as_ref()
465            .ok_or_else(|| {
466                ExpressionError::type_error(
467                    "Scalar subquery execution not supported in this context",
468                )
469            })?
470            .execute_scalar(body, self.layout.clone(), row)
471    }
472
473    fn contains_subquery(
474        &mut self,
475        body: &linkrs_core::types::expr::SubqueryBody,
476        value: &Value,
477    ) -> Result<Value, ExpressionError> {
478        if value.is_null() {
479            return Ok(Value::Bool(false));
480        }
481        let mut row = self.left.to_vec();
482        row.extend_from_slice(self.right);
483        let found = self
484            .subquery_executor
485            .as_ref()
486            .ok_or_else(|| {
487                ExpressionError::type_error("Subquery execution not supported in this context")
488            })?
489            .execute_contains(body, self.layout.clone(), row, value)?;
490        Ok(Value::Bool(found))
491    }
492}
493
494#[cfg(test)]
495mod tests {
496    use super::*;
497
498    #[test]
499    fn test_context_creation() {
500        let row = vec![Value::Int(1), Value::string("test"), Value::Bool(true)];
501        let col_names = vec!["id".to_string(), "name".to_string(), "active".to_string()];
502        let layout = Arc::new(SlotLayout::from_names(&col_names));
503
504        let context = ValueRowContext::new(row, layout);
505
506        assert_eq!(context.get_variable("id"), Some(Value::Int(1)));
507        assert_eq!(context.get_variable("name"), Some(Value::string("test")));
508        assert_eq!(context.get_variable("active"), Some(Value::Bool(true)));
509    }
510
511    #[test]
512    fn test_variable_storage() {
513        let row = vec![Value::Int(1)];
514        let col_names = vec!["id".to_string()];
515        let layout = Arc::new(SlotLayout::from_names(&col_names));
516        let mut context = ValueRowContext::new(row, layout);
517
518        context.set_variable("var1".into(), Value::string("hello"));
519        context.set_variable("var2".into(), Value::Int(42));
520
521        assert_eq!(context.get_variable("var1"), Some(Value::string("hello")));
522        assert_eq!(context.get_variable("var2"), Some(Value::Int(42)));
523    }
524
525    #[test]
526    fn test_missing_column() {
527        let row = vec![Value::Int(1), Value::string("test")];
528        let col_names = vec!["id".to_string(), "name".to_string()];
529        let layout = Arc::new(SlotLayout::from_names(&col_names));
530
531        let context = ValueRowContext::new(row, layout);
532
533        assert_eq!(context.get_variable("nonexistent"), None);
534        assert_eq!(context.get_variable("var_not_set"), None);
535    }
536}
linkrs_query/executor/streaming/context.rs

linkrs_query/executor/streaming/
context.rs
