1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
use super::*;
/// Whether a projection expression qualifies for the columnar fast path:
/// a bare column passthrough or a constant (gather / broadcast upstream,
/// no per-row expression work, NULL/selection/multiplicity preserved).
pub(super) fn is_passthrough_or_const(expression: &Expression) -> bool {
match expression {
Expression::Variable(_) | Expression::Literal(_) => true,
Expression::Property { object, .. } => {
matches!(object.as_ref(), Expression::Variable(_))
}
_ => false,
}
}
impl UnaryOperator {
/// Evaluate the project output expressions, preferring the compiled
/// closure tree over the scalar chunk path.
///
/// Columnar fast path: when every output is a passthrough (`Variable` /
/// `Property` over a variable) or a `Literal`, evaluation goes straight
/// through the chunk's column gather/broadcast helpers without building
/// per-expression compiled closures. Anything else keeps the existing
/// compiled-batch path.
pub(super) fn evaluate_project_expressions(
chunk: &mut DataChunk,
output_expressions: &[Expression],
state: &mut UnaryOperatorState,
) -> Result<Vec<Vec<Value>>, QueryError> {
if output_expressions.iter().all(is_passthrough_or_const) {
return chunk
.evaluate_expressions(output_expressions, Some(&state.env))
.map_err(|e| {
QueryError::execution(format!("Project expression evaluation failed: {}", e))
});
}
// Typed columnar fast path: expressions over raw scalars (bare
// columns, arithmetic/cast on typed columns) are evaluated on the
// batch. Falls through to the compiled path when the chunk has no
// typed layout or any expression is outside the typed batch set.
match chunk.try_evaluate_expressions_typed(output_expressions, Some(&state.env)) {
Ok(Some(cols)) => {
for _ in 0..output_expressions.len() {
chunk.count_typed_hit();
if let Some(stats) = &chunk.columnar_stats {
stats.record_b_path_hit();
}
}
return Ok(cols);
}
Ok(None) => {}
Err(_) => {}
}
if compiled_eval_enabled() {
if state.compiled_project.is_none() {
let layout = chunk.get_layout();
state.compiled_project = Some(
output_expressions
.iter()
.map(|e| CompiledExpr::compile(e, &layout))
.collect(),
);
}
if let Some(compiled) = &state.compiled_project {
let layout = chunk.get_layout();
let len = chunk.rows.len();
let mut columns = Vec::with_capacity(compiled.len());
let mut ok = true;
for expr in compiled {
match expr.evaluate_batch(
&chunk.rows,
layout.clone(),
Some(&state.env),
chunk.typed_columns.as_deref(),
) {
Ok(col) => columns.push(col.into_values(len)),
Err(_) => {
ok = false;
break;
}
}
}
if ok {
return Ok(columns);
}
}
}
chunk
.evaluate_expressions(output_expressions, Some(&state.env))
.map_err(|e| {
QueryError::execution(format!("Project expression evaluation failed: {}", e))
})
}
}
pub(super) fn handle(
op: &mut UnaryOperator,
input: &mut StreamingExecutor,
) -> Result<Option<DataChunk>, QueryError> {
let UnaryOperatorKind::Project {
output_expressions,
output_col_names: _,
state,
} = &mut op.kind
else {
return Err(QueryError::execution(
"project::handle called for a non-project unary operator".to_string(),
));
};
loop {
if let Some(mut chunk) = input.advance()? {
// When the child carries a selection vector, evaluate
// each output expression only for the visible rows — the
// output chunk is fully materialized (small).
if chunk.selection().is_some() {
let mut columns = Vec::with_capacity(output_expressions.len());
for expr in output_expressions.iter() {
let col = chunk
.evaluate_expression_visible(expr, Some(&state.env))
.map_err(|e| {
QueryError::execution(format!(
"Project expression evaluation failed: {}",
e
))
})?;
columns.push(col);
}
if !columns.is_empty() && !columns[0].is_empty() {
// 1:1 row-preserving rebuild: carry the input
// multiplicity (each output row occurs that often).
let multiplicity = chunk.multiplicity();
return Ok(Some(
DataChunk::project_columns(columns, Arc::clone(&op.output_layout))
.with_multiplicity(multiplicity),
));
}
continue;
}
let columns =
UnaryOperator::evaluate_project_expressions(&mut chunk, output_expressions, state)?;
if !columns.is_empty() && !columns[0].is_empty() {
let multiplicity = chunk.multiplicity();
return Ok(Some(
DataChunk::project_columns(columns, Arc::clone(&op.output_layout))
.with_multiplicity(multiplicity),
));
}
} else {
return Ok(None);
}
}
}