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
84
85
86
87
88
89

// SPDX-License-Identifier: BUSL-1.1

//! CRDT-routing gate for document-collection DML.
//!
//! Detects `crdt = true` document collections and builds the `fields_json`
//! payload that `CrdtOp::DocUpsert` consumes. Lives at the `dml` module level
//! so INSERT / UPSERT (in `insert.rs`) and UPDATE / DELETE (in
//! `update_delete/`) all reach the same detection + payload builders.
//!
//! Routing contract: on a CRDT document collection, PK-targeted
//! INSERT / UPSERT (full replace) / UPDATE-SET (literal RHS) / DELETE lower to
//! `CrdtOp::DocUpsert` / `DocDelete`; every unsupported shape (predicate
//! UPDATE/DELETE, non-literal UPDATE RHS, UPDATE ... RETURNING, `IF NOT EXISTS`
//! INSERT, explicit `ON CONFLICT DO UPDATE`) is rejected with a typed error.
//! There is NO silent fallthrough to a non-CRDT `DocumentOp` — that would
//! bypass CRDT convergence and is a correctness bug.

use nodedb_sql::types::{SqlExpr, SqlValue};

use crate::control::planner::sql_plan_convert::convert::ConvertContext;
use crate::control::planner::sql_plan_convert::value::row_to_msgpack;

/// `true` when `collection` (already db-qualified by the caller) is a CRDT
/// document collection.
///
/// A genuine catalog READ error propagates: misrouting a write to the non-CRDT
/// path would silently bypass CRDT convergence. An ABSENT credential store or
/// catalog, or an absent collection row (`Ok(None)`), is treated as non-CRDT
/// (`Ok(false)`). Mirrors `document_collection_is_edge_bearing` in
/// `update_delete/shared.rs` exactly.
pub(in crate::control::planner::sql_plan_convert::dml) fn document_collection_is_crdt(
    ctx: &ConvertContext,
    collection: &str,
) -> crate::Result<bool> {
    let Some(credentials) = ctx.credentials.as_ref() else {
        return Ok(false);
    };
    let catalog = credentials.catalog();
    Ok(catalog
        .get_collection(ctx.database_id, ctx.tenant_id.as_u64(), collection)?
        .map(|c| c.crdt)
        .unwrap_or(false))
}

/// Full-row fields → JSON object string for `DocUpsert.fields_json`.
///
/// Reuses the DML msgpack writer + `json_from_msgpack` so the stored payload is
/// byte-identical to what the non-CRDT `PointInsert` path would produce for the
/// same row, then serializes to JSON text (the shape `DocUpsert` consumes).
pub(in crate::control::planner::sql_plan_convert::dml) fn row_to_fields_json(
    row: &[(String, SqlValue)],
) -> crate::Result<String> {
    let mpk = row_to_msgpack(row)?;
    let json = nodedb_types::json_from_msgpack(&mpk).map_err(|e| crate::Error::Serialization {
        format: "json".into(),
        detail: format!("crdt fields_json encode: {e}"),
    })?;
    sonic_rs::to_string(&json).map_err(|e| crate::Error::Serialization {
        format: "json".into(),
        detail: format!("crdt fields_json encode: {e}"),
    })
}

/// UPDATE SET (partial) `fields_json` from LITERAL assignments only.
///
/// An expression RHS is rejected: the Control Plane planner cannot evaluate a
/// per-row expression here (same restriction the KV and columnar/spatial UPDATE
/// paths enforce). Only the provided fields go into the object — `DocUpsert`
/// with `partial = true` writes exactly these (LWW-per-field), leaving untouched
/// keys intact.
pub(in crate::control::planner::sql_plan_convert::dml) fn literal_assignments_to_fields_json(
    assignments: &[(String, SqlExpr)],
) -> crate::Result<String> {
    let mut row: Vec<(String, SqlValue)> = Vec::with_capacity(assignments.len());
    for (field, expr) in assignments {
        match expr {
            SqlExpr::Literal(v) => row.push((field.clone(), v.clone())),
            _ => {
                return Err(crate::Error::BadRequest {
                    detail: format!(
                        "UPDATE with non-literal RHS on CRDT collection (field '{field}') \
                         is not supported; use a literal value"
                    ),
                });
            }
        }
    }
    row_to_fields_json(&row)
}