nodedb 0.4.0

Local-first, real-time, edge-to-cloud hybrid database for multi-modal workloads
Documentation 
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

// SPDX-License-Identifier: BUSL-1.1

//! MERGE RESOLVE pass: classify without writing, returning the NOT-MATCHED
//! insert rows for Control-Plane surrogate assignment.

use crate::bridge::envelope::{ErrorCode, Response};
use crate::data::executor::core_loop::CoreLoop;
use crate::data::executor::task::ExecutionTask;

use super::super::merge::MergeParams;

impl CoreLoop {
    /// RESOLVE pass: return ALL resolved arms without writing.
    ///
    /// Response payload is a msgpack 3-tuple `(updates, deletes, inserts)`:
    /// - `updates`: `Vec<(doc_id, Option<surrogate_u32>, body_msgpack)>` — the
    ///   existing target row's storage key, its registered surrogate, and the
    ///   post-update body.
    /// - `deletes`: `Vec<(doc_id, Option<surrogate_u32>, body_msgpack)>` — the
    ///   existing target row's storage key, its registered surrogate, and the
    ///   pre-delete body (so the CP can extract its primary key).
    /// - `inserts`: `Vec<(join_key, body_msgpack)>` — the NOT-MATCHED rows, for
    ///   which the orchestrator assigns a fresh, registered surrogate.
    ///
    /// The autocommit orchestrator consumes only `inserts`; the in-transaction
    /// COMMIT expander rewrites all three arms into concrete point ops.
    pub(in crate::data::executor) fn execute_merge_resolve(
        &mut self,
        task: &ExecutionTask,
        tid: u64,
        params: MergeParams<'_>,
    ) -> Response {
        let plan = match self.collect_merge_plan(
            task.request.database_id.as_u64(),
            tid,
            task.request.txn_id,
            &params,
        ) {
            Ok(p) => p,
            Err(e) => return self.response_error(task, e),
        };
        let updates: Vec<(String, Option<u32>, Vec<u8>)> = plan
            .updates
            .into_iter()
            .map(|u| (u.doc_id, u.surrogate.map(|s| s.as_u32()), u.body))
            .collect();
        let deletes: Vec<(String, Option<u32>, Vec<u8>)> = plan
            .deletes
            .into_iter()
            .map(|d| (d.doc_id, d.surrogate.map(|s| s.as_u32()), d.body))
            .collect();
        let inserts: Vec<(String, Vec<u8>)> = plan
            .inserts
            .into_iter()
            .map(|i| (i.join_key, i.body))
            .collect();
        match zerompk::to_msgpack_vec(&(updates, deletes, inserts)) {
            Ok(payload) => self.response_with_payload(task, payload),
            Err(e) => self.response_error(
                task,
                ErrorCode::Internal {
                    detail: format!("merge resolve encode: {e}"),
                },
            ),
        }
    }
}