akar_processor/physical/write_ops/
unwind.rs

1//! Auto-extracted from physical_operator.rs
2use crate::physical::common::store_value_in_vector;
3use crate::physical::types::{OperatorResult, PhysicalOperatorExec};
4use akar_common::types::{PhysicalTypeID, Value};
5use akar_common::vector::{DataChunk, ValueVector};
6
7// ==================== Unwind ====================
8
9/// Physical operator for UNWIND — expands a list expression into rows.
10pub struct PhysicalUnwind {
11    pub expression: akar_parser::ast::Expression,
12    pub variable: String,
13}
14
15impl PhysicalOperatorExec for PhysicalUnwind {
16    fn operator_type(&self) -> &str {
17        "unwind"
18    }
19
20    fn execute(&self, input: Vec<DataChunk>) -> OperatorResult {
21        // Evaluate the expression to get a list value
22        let list_val = evaluate_unwind_expr(&self.expression);
23        let items = match &list_val {
24            akar_common::types::Value::List(items) => items.clone(),
25            _ => return Err("UNWIND expression must evaluate to a list".into()),
26        };
27
28        if items.is_empty() {
29            return Ok(Vec::new());
30        }
31
32        // Create a new ValueVector for the unwound variable
33        let first_type = items
34            .first()
35            .map(|v| v.physical_type())
36            .unwrap_or(PhysicalTypeID::Int64);
37
38        let mut result_chunks = Vec::new();
39        // If we have input data, repeat for each input row
40        if let Some(chunk) = input.first() {
41            for row in 0..chunk.size {
42                let mut chunk_fields = Vec::new();
43                for (col_idx, _field) in chunk.fields.iter().enumerate() {
44                    let val = chunk.get_value(col_idx, row).unwrap_or(Value::Null);
45                    let mut v = ValueVector::new(chunk.field_types[col_idx], items.len());
46                    v.resize(items.len());
47                    for i in 0..items.len() {
48                        store_value_in_vector(&mut v, i, &val)?;
49                    }
50                    chunk_fields.push(v);
51                }
52                // Add unwound vector
53                let mut uw_v = ValueVector::new(first_type, items.len());
54                uw_v.resize(items.len());
55                for (i, item) in items.iter().enumerate() {
56                    store_value_in_vector(&mut uw_v, i, item)?;
57                }
58                chunk_fields.push(uw_v);
59                let arrow_fields = chunk_fields
60                    .iter()
61                    .map(|v| akar_common::arrow_vector::ArrowVector::from_legacy(v).array)
62                    .collect::<Vec<_>>();
63                let arrow_field_types = chunk_fields.iter().map(|v| v.physical_type()).collect::<Vec<_>>();
64                result_chunks.push(DataChunk::new(arrow_fields, arrow_field_types));
65            }
66        } else {
67            // No input — just the unwound vector
68            let mut uw_v = ValueVector::new(first_type, items.len());
69            uw_v.resize(items.len());
70            for (i, item) in items.iter().enumerate() {
71                store_value_in_vector(&mut uw_v, i, item)?;
72            }
73            let arr = akar_common::arrow_vector::ArrowVector::from_legacy(&uw_v).array;
74            result_chunks.push(DataChunk::new(vec![arr], vec![uw_v.physical_type()]));
75        }
76
77        Ok(result_chunks)
78    }
79}
80
81/// Evaluate an UNWIND expression to get the list value.
82fn evaluate_unwind_expr(expr: &akar_parser::ast::Expression) -> Value {
83    match expr {
84        akar_parser::ast::Expression::List(items) => {
85            let values: Vec<Value> = items.iter().map(expr_to_value).collect();
86            Value::List(values)
87        }
88        _ => Value::List(Vec::new()),
89    }
90}
91
92/// Convert an AST expression to a runtime Value (for simple constants).
93fn expr_to_value(expr: &akar_parser::ast::Expression) -> Value {
94    match expr {
95        akar_parser::ast::Expression::Constant(c) => match c {
96            akar_parser::ast::Constant::Null => Value::Null,
97            akar_parser::ast::Constant::Bool(b) => Value::Bool(*b),
98            akar_parser::ast::Constant::Integer(i) => Value::Int64(*i),
99            akar_parser::ast::Constant::Float(f) => Value::Double(*f),
100            akar_parser::ast::Constant::String(s) => Value::String(s.clone()),
101        },
102        _ => Value::Null,
103    }
104}
akar_processor/physical/write_ops/unwind.rs

akar_processor/physical/write_ops/
unwind.rs
