Execute mutation pipeline: MATCH ... WITH ... [UNWIND ...] REMOVE/SET/DELETE
(
&self,
pipeline: &crate::ast::MutationPipeline,
context: &mut ExecutionContext,
_session: Option<&Arc<std::sync::RwLock<UserSession>>>,
)
| 7224 | |
| 7225 | /// Execute mutation pipeline: MATCH ... WITH ... [UNWIND ...] REMOVE/SET/DELETE |
| 7226 | fn execute_mutation_pipeline( |
| 7227 | &self, |
| 7228 | pipeline: &crate::ast::MutationPipeline, |
| 7229 | context: &mut ExecutionContext, |
| 7230 | _session: Option<&Arc<std::sync::RwLock<UserSession>>>, |
| 7231 | ) -> Result<QueryResult, ExecutionError> { |
| 7232 | use crate::ast::FinalMutation; |
| 7233 | use crate::storage::Value; |
| 7234 | |
| 7235 | log::debug!( |
| 7236 | "Starting mutation pipeline execution with {} segments", |
| 7237 | pipeline.segments.len() |
| 7238 | ); |
| 7239 | log::debug!("Final mutation type: {:?}", pipeline.final_mutation); |
| 7240 | |
| 7241 | // Execute the pipeline step by step |
| 7242 | if pipeline.segments.is_empty() { |
| 7243 | return Err(ExecutionError::RuntimeError( |
| 7244 | "Mutation pipeline requires at least one segment".to_string(), |
| 7245 | )); |
| 7246 | } |
| 7247 | |
| 7248 | // Step 1: Execute MATCH and WITH clauses to get aggregated data using passed context |
| 7249 | let first_segment = &pipeline.segments[0]; |
| 7250 | |
| 7251 | // Execute MATCH clause |
| 7252 | let mut match_results = |
| 7253 | self.execute_match_with_context(&first_segment.match_clause, context)?; |
| 7254 | |
| 7255 | // Apply pre-WITH WHERE clause if present |
| 7256 | if let Some(where_clause) = &first_segment.where_clause { |
| 7257 | match_results = |
| 7258 | self.apply_where_filter_to_rows(match_results, where_clause, context)?; |
| 7259 | } |
| 7260 | |
| 7261 | // Execute WITH clause for aggregation (if present) |
| 7262 | let with_results = if let Some(with_clause) = &first_segment.with_clause { |
| 7263 | self.execute_with_clause(with_clause, match_results, context)? |
| 7264 | } else { |
| 7265 | match_results |
| 7266 | }; |
| 7267 | |
| 7268 | // Step 2: Apply UNWIND if present |
| 7269 | let mut final_rows = with_results; |
| 7270 | if let Some(unwind_clause) = &first_segment.unwind_clause { |
| 7271 | final_rows = self.execute_unwind_on_rows(unwind_clause, final_rows, context)?; |
| 7272 | } |
| 7273 | |
| 7274 | // Step 3: Apply post-UNWIND WHERE clause if present |
| 7275 | if let Some(where_clause) = &first_segment.post_unwind_where { |
| 7276 | final_rows = self.apply_where_filter_to_rows_vec(final_rows, where_clause, context)?; |
| 7277 | } |
| 7278 | |
| 7279 | // Step 4: Apply the mutation to each row |
| 7280 | log::debug!("Applying final mutation to {} rows", final_rows.len()); |
| 7281 | let mut affected_count = 0; |
| 7282 | for (row_idx, row) in final_rows.iter().enumerate() { |
| 7283 | log::debug!( |
no test coverage detected