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
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83

// SPDX-License-Identifier: BUSL-1.1

//! BSP superstep fan: dispatch across all local cores and merge the per-core
//! `BspSuperstepResult` parts by field concatenation.

use crate::control::state::SharedState;
use crate::types::{DatabaseId, Lsn, TenantId, TraceId};
use nodedb_physical::physical_plan::{BspSuperstepResult, PhysicalPlan};

use super::dispatch::NodeLevelResult;
use super::fanout::gather_graph_op_all_cores;

/// BSP superstep fan: dispatch to all local cores, decode each core's
/// [`BspSuperstepResult`], merge by field concatenation, and re-encode.
///
/// Owned-node sets are disjoint across cores because `gather_graph_op_all_cores`
/// scopes each core's `owned_vshards` to the vShards homed on that core, so each
/// graph node is owned by exactly one core; concatenation therefore requires no
/// dedup.
pub(super) async fn fan_bsp_all_cores(
    state: &SharedState,
    tenant_id: TenantId,
    database_id: DatabaseId,
    plan: PhysicalPlan,
    trace_id: TraceId,
) -> crate::Result<NodeLevelResult> {
    let responses =
        gather_graph_op_all_cores(state, tenant_id, database_id, plan, trace_id, None, "bsp")
            .await?;

    let mut parts: Vec<BspSuperstepResult> = Vec::with_capacity(responses.len());
    for resp in responses {
        // An empty payload decodes to BspSuperstepResult::default() (a
        // zero-vertex shard — contributes nothing to global_n or the ranks),
        // matching decode_single_result's contract.
        let part = if resp.payload.is_empty() {
            BspSuperstepResult::default()
        } else {
            zerompk::from_msgpack::<BspSuperstepResult>(resp.payload.as_ref()).map_err(|e| {
                crate::Error::Codec {
                    detail: format!("bsp gather: result decode: {e}"),
                }
            })?
        };
        parts.push(part);
    }

    let merged = merge_bsp_results(parts);
    let payload = zerompk::to_msgpack_vec(&merged).map_err(|e| crate::Error::Serialization {
        format: "msgpack".into(),
        detail: format!("bsp gather: merged result encode: {e}"),
    })?;

    Ok(NodeLevelResult {
        payload,
        watermark_lsn: Lsn::ZERO,
        read_version_lsn: Lsn::ZERO,
    })
}

/// Merge per-core [`BspSuperstepResult`] parts by field concatenation.
///
/// Owned-node sets are DISJOINT across cores because `fan_bsp_all_cores` scopes
/// each core's `owned_vshards` to the vShards homed on that core, so each graph
/// node is owned by exactly one core. Concatenation therefore requires no dedup.
fn merge_bsp_results(parts: Vec<BspSuperstepResult>) -> BspSuperstepResult {
    let mut out = BspSuperstepResult::default();
    for p in parts {
        out.local_delta += p.local_delta;
        out.vertex_count += p.vertex_count;
        out.outbound.extend(p.outbound);
        out.node_names.extend(p.node_names);
        out.rank_vec.extend(p.rank_vec);
        // Owned-node sets are DISJOINT across cores (each graph node is homed on
        // exactly one core), so summing per-core dangling sums counts every
        // dangling node exactly once.
        out.dangling_sum += p.dangling_sum;
        // Same disjointness for the count-phase seed-hit tally: each owned node is
        // counted on exactly one core, so per-core seed hits sum cleanly.
        out.seed_hits += p.seed_hits;
    }
    out
}