linkrs_query/executor/streaming/
join_helpers.rs

1use std::sync::Arc;
2
3use crate::executor::expression::evaluator::ExpressionEvaluator;
4use crate::executor::streaming::executor::ValueRowContext;
5use crate::executor::streaming::slot::{combine_layouts, SlotLayout};
6use linkrs_core::error::QueryError;
7use linkrs_core::types::expr::Expression;
8use linkrs_core::Value;
9
10/// Evaluate join key expressions against a row, returning the key values.
11pub fn evaluate_join_key(
12    row: &[Value],
13    layout: Arc<SlotLayout>,
14    key_expressions: &[Expression],
15) -> Result<Vec<Value>, QueryError> {
16    if key_expressions.is_empty() {
17        return Ok(Vec::new());
18    }
19    let mut key = Vec::with_capacity(key_expressions.len());
20    for expr in key_expressions {
21        let mut context = ValueRowContext::new(row.to_vec(), layout.clone());
22        let value = ExpressionEvaluator::evaluate(expr, &mut context)
23            .map_err(|e| QueryError::execution(format!("HashJoin key evaluation failed: {e}")))?;
24        key.push(value);
25    }
26    Ok(key)
27}
28
29/// Evaluate a residual join condition given left and right rows with their schemas.
30pub fn evaluate_residual_condition(
31    condition: &Expression,
32    left_row: &[Value],
33    right_row: &[Value],
34    left_schema: &[String],
35    right_schema: &[String],
36) -> Result<bool, QueryError> {
37    let mut combined = left_row.to_vec();
38    combined.extend_from_slice(right_row);
39    let combined_layout = build_combined_layout_from_schemas(left_schema, right_schema);
40    let mut context = ValueRowContext::new(combined, combined_layout);
41    match ExpressionEvaluator::evaluate(condition, &mut context) {
42        Ok(Value::Bool(b)) => Ok(b),
43        Ok(Value::Null(_)) => Ok(false),
44        Ok(_) => Ok(true),
45        Err(e) => Err(QueryError::execution(format!(
46            "HashJoin condition evaluation failed: {e}"
47        ))),
48    }
49}
50
51/// Build a combined slot layout from left and right column name schemas.
52pub fn build_combined_layout_from_schemas(
53    left_schema: &[String],
54    right_schema: &[String],
55) -> Arc<SlotLayout> {
56    let left_layout = Arc::new(SlotLayout::from_names(left_schema));
57    let right_layout = Arc::new(SlotLayout::from_names(right_schema));
58    Arc::new(combine_layouts(&left_layout, &right_layout))
59}
linkrs_query/executor/streaming/join_helpers.rs

linkrs_query/executor/streaming/
join_helpers.rs
