fv_plan/
inline.rs

1//! Inline row-op executor. Executes a chain of row ops (select/rename/drop/filter/applyExpression)
2//! row-wise, delegating value compute to `fv-value`.
3//! Vectored against the published contract's `step-kind` conformance vectors.
4
5use crate::row::Row;
6use fv_value::{compile, ExprError, Value};
7
8/// A pipeline step as decoded JSON (op + its params), matching the control-plane spec.
9#[derive(Debug, Clone)]
10pub enum Step {
11    Select { columns: Vec<String> },
12    Rename { mapping: Vec<(String, String)> },
13    Drop { columns: Vec<String> },
14    Filter { expression: String },
15    ApplyExpression { column: String, expression: String },
16}
17
18#[derive(Debug)]
19pub enum StepError {
20    Expr(ExprError),
21    Message(String),
22}
23
24impl std::fmt::Display for StepError {
25    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
26        match self {
27            StepError::Expr(e) => write!(f, "{e}"),
28            StepError::Message(m) => write!(f, "{m}"),
29        }
30    }
31}
32
33impl From<ExprError> for StepError {
34    fn from(e: ExprError) -> Self {
35        StepError::Expr(e)
36    }
37}
38
39/// Apply one step to a set of rows, producing the transformed rows.
40pub fn apply_step(step: &Step, rows: &[Row]) -> Result<Vec<Row>, StepError> {
41    match step {
42        // Project to exactly `columns`, in order; a missing column lands null.
43        Step::Select { columns } => Ok(rows
44            .iter()
45            .map(|r| Row(columns.iter().map(|c| (c.clone(), r.get(c))).collect()))
46            .collect()),
47
48        // Rename mapped keys (others pass through), preserving column order.
49        Step::Rename { mapping } => Ok(rows
50            .iter()
51            .map(|r| {
52                Row(r
53                    .0
54                    .iter()
55                    .map(|(k, v)| {
56                        let nk = mapping
57                            .iter()
58                            .find(|(from, _)| from == k)
59                            .map(|(_, to)| to.clone())
60                            .unwrap_or_else(|| k.clone());
61                        (nk, v.clone())
62                    })
63                    .collect())
64            })
65            .collect()),
66
67        // Drop the named columns.
68        Step::Drop { columns } => Ok(rows
69            .iter()
70            .map(|r| Row(r.0.iter().filter(|(k, _)| !columns.contains(k)).cloned().collect()))
71            .collect()),
72
73        // Keep rows whose predicate is strictly `true`; a non-boolean result is an error (fail-closed).
74        Step::Filter { expression } => {
75            let ast = compile(expression)?;
76            let mut out = Vec::new();
77            for r in rows {
78                match ast.eval(r.lookup())? {
79                    Value::Bool(true) => out.push(r.clone()),
80                    Value::Bool(false) => {}
81                    other => {
82                        return Err(StepError::Message(format!(
83                            "filter: predicate must yield a boolean, got {other:?}"
84                        )))
85                    }
86                }
87            }
88            Ok(out)
89        }
90
91        // Add or overwrite a column from an expression over the row's identifiers.
92        Step::ApplyExpression { column, expression } => {
93            let ast = compile(expression)?;
94            let mut out = Vec::with_capacity(rows.len());
95            for r in rows {
96                let v = ast.eval(r.lookup())?;
97                let mut nr = r.clone();
98                nr.set(column, v);
99                out.push(nr);
100            }
101            Ok(out)
102        }
103    }
104}
105
106/// Apply a chain of inline steps left-to-right.
107pub fn apply_steps(steps: &[Step], rows: &[Row]) -> Result<Vec<Row>, StepError> {
108    let mut current = rows.to_vec();
109    for step in steps {
110        current = apply_step(step, &current)?;
111    }
112    Ok(current)
113}
114
115/// Apply `steps` with PER-ROW error isolation — for stream builds, where a single poison row
116/// (dirty data hitting a per-row `eval`/`filter` error) must NOT sink the whole batch. Fast
117/// path applies to the whole batch; only if that errors does it fall back to applying per row,
118/// keeping the survivors and returning `(survivors, dropped_row_count)`.
119///
120/// CONFIG vs DATA: callers must validate the steps COMPILE first (see [`validate_steps`]) so a bad
121/// expression fails LOUDLY at build start instead of silently dropping every row here — this helper
122/// is for *data* errors (a bad row), not authoring errors (a bad step).
123pub fn apply_steps_isolating(steps: &[Step], rows: &[Row]) -> (Vec<Row>, usize) {
124    match apply_steps(steps, rows) {
125        Ok(out) => (out, 0),
126        Err(_) => {
127            let mut survived = Vec::new();
128            let mut dropped = 0usize;
129            for row in rows {
130                match apply_steps(steps, std::slice::from_ref(row)) {
131                    Ok(mut o) => survived.append(&mut o),
132                    Err(_) => dropped += 1,
133                }
134            }
135            (survived, dropped)
136        }
137    }
138}
139
140/// Validate that every step's expression COMPILES (an authoring/config check), independent of data —
141/// run once at stream-build start so a bad expression fails the build instead of per-row-dropping
142/// forever. Compiling with zero rows exercises `compile()` without evaluating any row.
143pub fn validate_steps(steps: &[Step]) -> Result<(), StepError> {
144    apply_steps(steps, &[]).map(|_| ())
145}
146
147/// The inline (row-wise) step ops. Adding an inline step = a `Step` variant + an `apply_step` arm +
148/// a `parse` arm + one entry here — all in THIS file.
149pub const INLINE_OPS: &[&str] = &["select", "rename", "drop", "filter", "applyExpression"];
150
151/// Decode an inline step from its JSON spec (used by the build loop + the worker's preview tracer).
152pub fn parse(step: &serde_json::Value) -> Result<Step, String> {
153    let strs = |k: &str| {
154        step[k]
155            .as_array()
156            .map(|a| a.iter().filter_map(|v| v.as_str().map(String::from)).collect())
157            .unwrap_or_default()
158    };
159    Ok(match step["op"].as_str().unwrap_or("") {
160        "select" => Step::Select {
161            columns: strs("columns"),
162        },
163        "drop" => Step::Drop {
164            columns: strs("columns"),
165        },
166        "rename" => Step::Rename {
167            mapping: step["mapping"]
168                .as_object()
169                .map(|m| {
170                    m.iter()
171                        .map(|(k, v)| (k.clone(), v.as_str().unwrap_or("").to_string()))
172                        .collect()
173                })
174                .unwrap_or_default(),
175        },
176        "filter" => Step::Filter {
177            expression: step["expression"].as_str().unwrap_or("").to_string(),
178        },
179        "applyExpression" => Step::ApplyExpression {
180            column: step["column"].as_str().unwrap_or("").to_string(),
181            expression: step["expression"].as_str().unwrap_or("").to_string(),
182        },
183        other => return Err(format!("unknown inline step op '{other}'")),
184    })
185}
186
187#[cfg(test)]
188mod isolation_tests {
189    use super::*;
190    use crate::row::Row;
191    use fv_value::Value;
192
193    fn row(pairs: &[(&str, Value)]) -> Row {
194        Row(pairs.iter().map(|(k, v)| (k.to_string(), v.clone())).collect())
195    }
196
197    #[test]
198    fn isolating_drops_only_the_poison_row() {
199        // filter on `keep`: a non-boolean predicate is a per-row (fail-closed) error, so the whole
200        // batch errors — but isolation keeps the good rows and drops just the poison one.
201        let steps = vec![Step::Filter {
202            expression: "keep".into(),
203        }];
204        let rows = vec![
205            row(&[("keep", Value::Bool(true)), ("id", Value::Num(1.0))]),
206            row(&[("keep", Value::Num(9.0)), ("id", Value::Num(2.0))]), // poison: non-boolean
207            row(&[("keep", Value::Bool(true)), ("id", Value::Num(3.0))]),
208        ];
209        assert!(apply_steps(&steps, &rows).is_err(), "whole-batch apply is fail-closed");
210        let (out, dropped) = apply_steps_isolating(&steps, &rows);
211        assert_eq!(dropped, 1);
212        assert_eq!(out.len(), 2);
213        assert_eq!(out[0].get("id"), Value::Num(1.0));
214        assert_eq!(out[1].get("id"), Value::Num(3.0));
215    }
216
217    #[test]
218    fn isolating_fast_path_when_all_rows_ok() {
219        let steps = vec![Step::Filter {
220            expression: "keep".into(),
221        }];
222        let rows = vec![
223            row(&[("keep", Value::Bool(true)), ("id", Value::Num(1.0))]),
224            row(&[("keep", Value::Bool(false)), ("id", Value::Num(2.0))]),
225        ];
226        let (out, dropped) = apply_steps_isolating(&steps, &rows);
227        assert_eq!(dropped, 0);
228        assert_eq!(out.len(), 1, "filter keeps only the true row");
229        assert_eq!(out[0].get("id"), Value::Num(1.0));
230    }
231
232    #[test]
233    fn validate_rejects_bad_expression_accepts_good() {
234        assert!(validate_steps(&[Step::ApplyExpression {
235            column: "x".into(),
236            expression: "1 +".into()
237        }])
238        .is_err());
239        assert!(validate_steps(&[Step::ApplyExpression {
240            column: "x".into(),
241            expression: "1 + 2".into()
242        }])
243        .is_ok());
244    }
245}
fv_plan/inline.rs

fv_plan/
inline.rs
