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        // Build the unwound variable column once. `build_arrow_from_values`
33        // preserves complex values (map/struct/list) that
34        // `store_value_in_vector` would drop to NULL (P53.26).
35        let first_type = items
36            .first()
37            .map(|v| v.physical_type())
38            .unwrap_or(PhysicalTypeID::Int64);
39        let uw_arrow = crate::expression_evaluator::build_arrow_from_values(&items, first_type, items.len())
40            .map_err(|e| e.to_string())?;
41
42        let mut result_chunks = Vec::new();
43        // If we have input data, repeat for each input row
44        if let Some(chunk) = input.first() {
45            for row in 0..chunk.size {
46                let mut arrow_fields: Vec<arrow::array::ArrayRef> = Vec::with_capacity(chunk.fields.len() + 1);
47                let mut arrow_field_types: Vec<PhysicalTypeID> = Vec::with_capacity(chunk.fields.len() + 1);
48                for (col_idx, _field) in chunk.fields.iter().enumerate() {
49                    let val = chunk.get_value(col_idx, row).unwrap_or(Value::Null);
50                    let mut v = ValueVector::new(chunk.field_types[col_idx], items.len());
51                    v.resize(items.len());
52                    for i in 0..items.len() {
53                        store_value_in_vector(&mut v, i, &val)?;
54                    }
55                    arrow_fields.push(akar_common::arrow_vector::ArrowVector::from_legacy(&v).array);
56                    arrow_field_types.push(v.physical_type());
57                }
58                // Add unwound vector
59                arrow_fields.push(uw_arrow.array.clone());
60                arrow_field_types.push(uw_arrow.physical_type);
61                let mut named_chunk = DataChunk::new(arrow_fields, arrow_field_types);
62                named_chunk.field_names = chunk
63                    .field_names
64                    .iter()
65                    .cloned()
66                    .chain([self.variable.clone()])
67                    .collect();
68                result_chunks.push(named_chunk);
69            }
70        } else {
71            // No input — just the unwound vector
72            let mut named_chunk = DataChunk::new(vec![uw_arrow.array.clone()], vec![uw_arrow.physical_type]);
73            named_chunk.field_names = vec![self.variable.clone()];
74            result_chunks.push(named_chunk);
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 literal constants and
93/// nested list/map literals).
94fn expr_to_value(expr: &akar_parser::ast::Expression) -> Value {
95    match expr {
96        akar_parser::ast::Expression::Constant(c) => match c {
97            akar_parser::ast::Constant::Null => Value::Null,
98            akar_parser::ast::Constant::Bool(b) => Value::Bool(*b),
99            akar_parser::ast::Constant::Integer(i) => Value::Int64(*i),
100            akar_parser::ast::Constant::Float(f) => Value::Double(*f),
101            akar_parser::ast::Constant::String(s) => Value::String(s.clone()),
102        },
103        akar_parser::ast::Expression::List(items) => Value::List(items.iter().map(expr_to_value).collect()),
104        akar_parser::ast::Expression::Map(items) => Value::Map(
105            items
106                .iter()
107                .map(|(k, v)| (Value::String(k.clone()), expr_to_value(v)))
108                .collect(),
109        ),
110        _ => Value::Null,
111    }
112}
akar_processor/physical/write_ops/unwind.rs

akar_processor/physical/write_ops/
unwind.rs
