Method execute_bulk_delete

nodedb/src/data/executor/handlers/bulk_dml.rs:326–482

Bulk delete: scan documents matching filters, delete all matches. Cascades to inverted index, secondary indexes, and graph edges. When `returning` is `None`, returns affected row count as JSON payload: `{"affected": N}`. When `returning` is `Some(spec)`, returns a `RowsPayload` with the pre-deletion documents.

(
        &mut self,
        task: &ExecutionTask,
        tid: u64,
        collection: &str,
        filter_bytes: &[u8],
        returning: Option<&ReturningSpec>,
        ollp_predicted_surrogates

    /// When `returning` is `None`, returns affected row count as JSON payload: `{"affected": N}`.
    /// When `returning` is `Some(spec)`, returns a `RowsPayload` with the pre-deletion documents.
    pub(in crate::data::executor) fn execute_bulk_delete(
        &mut self,
        task: &ExecutionTask,
        tid: u64,
        collection: &str,
        filter_bytes: &[u8],
        returning: Option<&ReturningSpec>,
        ollp_predicted_surrogates: Option<&[u32]>,
    ) -> Response {
        debug!(core = self.core_id, %collection, has_returning = returning.is_some(), "bulk delete");

        // Empty `filter_bytes` means "no WHERE clause" — match every row.
        let filters: Vec<ScanFilter> = if filter_bytes.is_empty() {
            Vec::new()
        } else {
            match zerompk::from_msgpack(filter_bytes) {
                Ok(f) => f,
                Err(e) => {
                    return self.response_error(
                        task,
                        ErrorCode::Internal {
                            detail: format!("deserialize filters: {e}"),
                        },
                    );
                }
            }
        };

        let matching_ids = match self.scan_matching_documents(tid, collection, &filters) {
            Ok(ids) => ids,
            Err(e) => {
                return self.response_error(
                    task,
                    ErrorCode::Internal {
                        detail: e.to_string(),
                    },
                );
            }
        };

        // OLLP verification: when predicted surrogates are provided, compare
        // against the actual matching set. On mismatch return OllpRetryRequired
        // WITHOUT writing. The set comparison is deterministic: both sides are
        // sorted before comparison.
        if let Some(predicted) = ollp_predicted_surrogates {
            let actual = ollp_actual_surrogates(&matching_ids);
            let mut predicted_sorted: Vec<u32> = predicted.to_vec();
            predicted_sorted.sort_unstable();
            if actual != predicted_sorted {
                return self.response_error(task, ErrorCode::OllpRetryRequired);
            }
        }

        // Delete each matching document with full cascade.
        let mut affected = 0u64;
        let mut returned_docs: Vec<serde_json::Value> = if returning.is_some() {
            Vec::with_capacity(matching_ids.len())
        } else {