Execute a projection operation
(
&self,
expressions: &[ProjectionItem],
input_rows: Vec<Row>,
context: &mut ExecutionContext,
)
| 4135 | |
| 4136 | /// Execute a projection operation |
| 4137 | fn execute_project( |
| 4138 | &self, |
| 4139 | expressions: &[ProjectionItem], |
| 4140 | input_rows: Vec<Row>, |
| 4141 | context: &mut ExecutionContext, |
| 4142 | ) -> Result<Vec<Row>, ExecutionError> { |
| 4143 | // Check if any expressions are aggregate functions |
| 4144 | let has_aggregates = expressions |
| 4145 | .iter() |
| 4146 | .any(|expr| self.is_aggregate_function(&expr.expression)); |
| 4147 | |
| 4148 | if has_aggregates { |
| 4149 | // Check if we have mixed aggregate and non-aggregate expressions |
| 4150 | let has_non_aggregates = expressions |
| 4151 | .iter() |
| 4152 | .any(|expr| !self.is_aggregate_function(&expr.expression)); |
| 4153 | |
| 4154 | if has_non_aggregates { |
| 4155 | // Mixed aggregate/non-aggregate - return one row per input row with aggregate evaluated per row |
| 4156 | return self.execute_mixed_aggregate_projection(expressions, input_rows, context); |
| 4157 | } else { |
| 4158 | // Pure aggregates - return single row |
| 4159 | return self.execute_aggregate_projection(expressions, input_rows, context); |
| 4160 | } |
| 4161 | } |
| 4162 | |
| 4163 | // Normal projection - process each row |
| 4164 | let mut projected_rows = Vec::new(); |
| 4165 | |
| 4166 | for row in input_rows { |
| 4167 | let mut new_row = Row::new(); |
| 4168 | |
| 4169 | // Clear local variables from previous row to prevent variable leakage |
| 4170 | context.clear_locals(); |
| 4171 | |
| 4172 | // Set row values in context for expression evaluation |
| 4173 | for (name, value) in &row.values { |
| 4174 | context.set_variable(name.clone(), value.clone()); |
| 4175 | } |
| 4176 | |
| 4177 | // Evaluate each projection expression |
| 4178 | for proj_item in expressions { |
| 4179 | let column_name = proj_item.alias.clone().unwrap_or_else(|| { |
| 4180 | // Generate a default name from the expression |
| 4181 | self.expression_to_string(&proj_item.expression) |
| 4182 | }); |
| 4183 | |
| 4184 | // Check if this is a post-aggregation projection where we should map existing columns |
| 4185 | // instead of re-evaluating expressions |
| 4186 | let raw_expression_name = self.expression_to_string(&proj_item.expression); |
| 4187 | let value = if let Some(existing_value) = row.values.get(&raw_expression_name) { |
| 4188 | // If the row already contains a column with the raw expression name, |
| 4189 | // use that value instead of re-evaluating (this happens after aggregation) |
| 4190 | existing_value.clone() |
| 4191 | } else { |
| 4192 | // Normal case: evaluate the expression |
| 4193 | self.evaluate_expression(&proj_item.expression, context)? |
| 4194 | }; |
no test coverage detected