uqa_execution/
map_rows.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Streaming whole-row mapping adapter.
8
9use std::sync::Arc;
10
11use uqa_sql::ResultRow;
12
13use crate::{Batch, ExecResult, PhysicalOperator, RowSchema};
14
15pub type SharedRowMapper<'a> = Arc<dyn Fn(ResultRow) -> ExecResult<ResultRow> + Send + Sync + 'a>;
16
17/// Apply a fallible mapping to each input row without changing batch
18/// backpressure or collecting the child relation.
19pub struct MapRows<'a> {
20    child: Box<dyn PhysicalOperator + 'a>,
21    mapper: SharedRowMapper<'a>,
22    schema: RowSchema,
23}
24
25impl<'a> MapRows<'a> {
26    pub fn new(
27        child: Box<dyn PhysicalOperator + 'a>,
28        schema: Vec<String>,
29        mapper: SharedRowMapper<'a>,
30    ) -> Self {
31        Self {
32            child,
33            mapper,
34            schema: RowSchema::new(schema),
35        }
36    }
37}
38
39impl PhysicalOperator for MapRows<'_> {
40    fn row_schema(&self) -> &RowSchema {
41        &self.schema
42    }
43
44    fn estimated_cardinality(&self) -> Option<u64> {
45        self.child.estimated_cardinality()
46    }
47
48    fn open(&mut self) -> ExecResult<()> {
49        self.child.open()
50    }
51
52    fn next(&mut self) -> ExecResult<Option<Batch>> {
53        let Some(batch) = self.child.next()? else {
54            return Ok(None);
55        };
56        let rows = batch
57            .rows
58            .iter()
59            .map(|row| (self.mapper)(batch.schema.view(row).to_result_row()))
60            .collect::<ExecResult<Vec<_>>>()?;
61        Ok(Some(Batch::new(self.schema.clone(), rows)))
62    }
63
64    fn close(&mut self) -> ExecResult<()> {
65        self.child.close()
66    }
67}
68
69#[cfg(test)]
70mod tests {
71    use std::collections::BTreeMap;
72
73    use uqa_core::Value;
74
75    use super::*;
76    use crate::{physical::run_to_rows, ExecError, TableScan};
77
78    #[test]
79    fn mapper_preserves_late_errors() {
80        let rows = vec![
81            BTreeMap::from([("v".into(), Value::Int(1))]),
82            BTreeMap::from([("v".into(), Value::Int(2))]),
83        ];
84        let child = Box::new(TableScan::from_rows(vec!["v".into()], rows));
85        let mapper = Arc::new(|row: ResultRow| {
86            if row.get("v") == Some(&Value::Int(2)) {
87                Err(ExecError::Other("map failed".into()))
88            } else {
89                Ok(row)
90            }
91        });
92        let mut operator = MapRows::new(child, vec!["v".into()], mapper);
93        let error = run_to_rows(&mut operator).unwrap_err();
94        assert!(error.to_string().contains("map failed"));
95    }
96}
uqa_execution/map_rows.rs

uqa_execution/
map_rows.rs
