alopex_sql/executor/query/
project.rs

1use crate::catalog::ColumnMetadata;
2use crate::executor::{ColumnInfo, Result};
3use crate::planner::typed_expr::Projection;
4use crate::storage::SqlValue;
5
6use super::{Row, column_info_from_projection, column_infos_from_all, eval_expr};
7
8/// Build the public output schema for a projection without executing rows.
9pub fn projected_columns(
10    projection: &Projection,
11    schema: &[ColumnMetadata],
12) -> Result<Vec<ColumnInfo>> {
13    match projection {
14        Projection::All(names) => column_infos_from_all(schema, names),
15        Projection::Columns(cols) => Ok(cols
16            .iter()
17            .enumerate()
18            .map(|(index, column)| column_info_from_projection(column, index))
19            .collect()),
20    }
21}
22
23/// Apply one projection with the same expression kernel as local query
24/// execution. The distributed worker/assembler boundary uses this helper for
25/// its already fenced, already authorized rows; it does not deserialize a
26/// logical plan.
27pub fn project_row_values(
28    row: &Row,
29    projection: &Projection,
30    schema: &[ColumnMetadata],
31) -> Result<Vec<SqlValue>> {
32    match projection {
33        Projection::All(names) if names.len() == schema.len() => Ok(row.values.clone()),
34        Projection::All(names) => names
35            .iter()
36            .map(|name| {
37                let index = schema
38                    .iter()
39                    .position(|column| &column.name == name)
40                    .ok_or_else(|| crate::executor::ExecutorError::ColumnNotFound(name.clone()))?;
41                row.values.get(index).cloned().ok_or_else(|| {
42                    crate::executor::ExecutorError::InvalidOperation {
43                        operation: "project".into(),
44                        reason: format!("row is missing projected column '{name}'"),
45                    }
46                })
47            })
48            .collect(),
49        Projection::Columns(columns) => columns
50            .iter()
51            .map(|column| eval_expr(&column.expr, row))
52            .collect(),
53    }
54}
55
56/// Project rows according to Projection, returning QueryResult.
57pub fn execute_project(
58    rows: Vec<Row>,
59    projection: &Projection,
60    schema: &[crate::catalog::ColumnMetadata],
61) -> Result<crate::executor::QueryResult> {
62    match projection {
63        Projection::All(names) => project_all(rows, schema, names),
64        Projection::Columns(cols) => project_columns(rows, cols),
65    }
66}
67
68fn project_all(
69    rows: Vec<Row>,
70    schema: &[crate::catalog::ColumnMetadata],
71    names: &[String],
72) -> Result<crate::executor::QueryResult> {
73    let projection = Projection::All(names.to_vec());
74    let columns = projected_columns(&projection, schema)?;
75    let projected_rows = rows
76        .iter()
77        .map(|row| project_row_values(row, &projection, schema))
78        .collect::<Result<Vec<_>>>()?;
79    Ok(crate::executor::QueryResult::new(columns, projected_rows))
80}
81
82fn project_columns(
83    rows: Vec<Row>,
84    cols: &[crate::planner::typed_expr::ProjectedColumn],
85) -> Result<crate::executor::QueryResult> {
86    let projection = Projection::Columns(cols.to_vec());
87    let columns = projected_columns(&projection, &[])?;
88    let projected_rows = rows
89        .iter()
90        .map(|row| project_row_values(row, &projection, &[]))
91        .collect::<Result<Vec<_>>>()?;
92
93    Ok(crate::executor::QueryResult::new(columns, projected_rows))
94}
alopex_sql/executor/query/project.rs

alopex_sql/executor/query/
project.rs
