akar_processor/physical/write_ops/
foreach.rs

1//! Auto-extracted from physical_operator.rs
2use crate::physical::common::store_value_in_vector;
3use crate::physical::scan_filter::PhysicalScan;
4use crate::physical::types::{OperatorResult, PhysicalOperatorExec};
5use crate::physical::write_ops::ast_constant_to_value;
6use akar_common::types::Value;
7use akar_common::vector::{DataChunk, ValueVector};
8use akar_parser::ast::Expression;
9use akar_storage::table::TableCatalog;
10use std::sync::{Arc, Mutex};
11
12// ==================== Foreach ====================
13
14/// Physical FOREACH operator — iterates over list elements and executes sub-plans.
15pub struct PhysicalForeach {
16    pub variable: String,
17    pub expression: Expression,
18    pub sub_plans: Vec<Vec<akar_planner::logical_operator::LogicalOperator>>,
19    pub function_registry: Option<Arc<Mutex<akar_function::registry::FunctionRegistry>>>,
20    pub table_catalog: Option<Arc<TableCatalog>>,
21    pub vfs: Option<Arc<akar_common::file_system::VirtualFileSystemRegistry>>,
22}
23
24impl PhysicalOperatorExec for PhysicalForeach {
25    fn operator_type(&self) -> &str {
26        "foreach"
27    }
28
29    fn execute(&self, _input: Vec<DataChunk>) -> OperatorResult {
30        // Evaluate the list expression
31        let list_val = match &self.expression {
32            Expression::List(items) => {
33                let mut vals = Vec::with_capacity(items.len());
34                for item in items {
35                    if let Expression::Constant(c) = item {
36                        vals.push(ast_constant_to_value(c));
37                    } else {
38                        vals.push(Value::Null);
39                    }
40                }
41                Value::List(vals)
42            }
43            _ => {
44                return Err(format!("FOREACH requires a list expression, got: {:?}", self.expression).into());
45            }
46        };
47
48        let list_items = match &list_val {
49            Value::List(items) => items.clone(),
50            _ => return Ok(vec![]),
51        };
52
53        if list_items.is_empty() || self.sub_plans.is_empty() {
54            return Ok(vec![]);
55        }
56
57        // For each list item, execute sub-plans with the item value in scope.
58        // We use a simplified approach: create a DataChunk with the item value
59        // and pass it to each sub-plan.
60        for item in &list_items {
61            for sub_plan in &self.sub_plans {
62                // Create a single-row DataChunk containing the current item
63                let phys_type = PhysicalScan::value_to_physical_type(item);
64                let mut v = ValueVector::new(phys_type, 1);
65                v.resize(1);
66                store_value_in_vector(&mut v, 0, item)?;
67                let _chunk = DataChunk::new(
68                    vec![akar_common::arrow_vector::ArrowVector::from_legacy(&v).array],
69                    vec![akar_common::types::PhysicalTypeID::Int64],
70                );
71
72                // Execute the sub-plan using the QueryProcessor-like pipeline
73                // Use the processor module directly from the same crate
74                let processor = crate::processor::QueryProcessor::with_catalog(
75                    self.function_registry.clone().unwrap(),
76                    self.table_catalog.clone().unwrap(),
77                    self.vfs.clone().unwrap(),
78                );
79                let _result = processor.execute(sub_plan)?;
80            }
81        }
82
83        // FOREACH produces no output rows (it's a write-only operation)
84        Ok(vec![])
85    }
86}
akar_processor/physical/write_ops/foreach.rs

akar_processor/physical/write_ops/
foreach.rs
