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
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125

// SPDX-License-Identifier: BUSL-1.1

//! Shared HTTP query-result shaping helper.
//!
//! Routes every HTTP query result path (materialized JSON, materialized
//! NDJSON) through the protocol-neutral [`shape_response_materialized`]
//! core — the same shaping pgwire and native use — so HTTP SELECT rows are
//! KV-wrapped, vector-translated, decoded, scan-envelope-unwrapped, and
//! projected identically to the other two protocols. Non-row plan kinds
//! (`Execution`, `DmlResult`) come back as [`HttpShaped::Passthrough`]; the
//! caller keeps its existing raw decode/base64 fallback for those.

use crate::bridge::envelope::PhysicalPlan;
use crate::control::server::response_shape::compose::{ShapeOutcome, shape_response_materialized};
use crate::control::server::response_shape::schema::OutputSchema;
use crate::control::server::response_shape::types::PlanKind;
use crate::control::state::SharedState;
use nodedb_types::{DatabaseId, NodeDbError, TenantId};

/// Outcome of shaping one Data-Plane payload for an HTTP response.
pub(super) enum HttpShaped {
    /// Row-producing plan kinds, already decoded, envelope-unwrapped, and
    /// (when a projection applies) column-selected.
    Rows(Vec<serde_json::Value>),
    /// Tag/execution plan kinds — the caller keeps its existing raw
    /// decode/base64 behavior for these.
    Passthrough,
}

/// Shape one Data-Plane payload for HTTP, mirroring the pgwire/native
/// materialized-shaping call site.
pub(super) fn shape_http_payload(
    payload: &[u8],
    plan: &PhysicalPlan,
    plan_kind: PlanKind,
    projection: Option<&OutputSchema>,
    state: &SharedState,
    database_id: DatabaseId,
    tenant_id: TenantId,
) -> Result<HttpShaped, NodeDbError> {
    match shape_response_materialized(
        payload,
        plan,
        plan_kind,
        projection,
        state,
        database_id,
        tenant_id,
    )? {
        ShapeOutcome::Rows(shaped) => Ok(HttpShaped::Rows(
            shaped
                .rows
                .into_iter()
                .map(serde_json::Value::Object)
                .collect(),
        )),
        ShapeOutcome::Passthrough => Ok(HttpShaped::Passthrough),
    }
}

/// Decode a Data-Plane payload to JSON for a `Passthrough` result (writes /
/// tags): MessagePack first, then JSON passthrough, then base64 as a last
/// resort for undecodable binary payloads. Matches the pre-shaping-core raw
/// behavior of the materialized JSON handler exactly.
pub(super) fn passthrough_json_row(payload: &[u8]) -> serde_json::Value {
    if let Ok(val) = nodedb_types::json_from_msgpack(payload) {
        return val;
    }
    if let Ok(val) = sonic_rs::from_slice::<serde_json::Value>(payload) {
        return val;
    }
    use base64::Engine;
    let encoded = base64::engine::general_purpose::STANDARD.encode(payload);
    serde_json::json!({ "data": encoded })
}

/// Append a `Passthrough` payload (writes / tags) to an NDJSON buffer as one
/// line per top-level JSON array element, or one line for a non-array
/// payload. Matches the pre-shaping-core raw behavior of the materialized
/// NDJSON handler exactly.
pub(super) fn passthrough_to_ndjson(payload: &[u8], ndjson: &mut String) {
    let json_str = crate::data::executor::response_codec::decode_payload_to_json(payload);
    if json_str.trim_start().starts_with('[') {
        for lv in sonic_rs::to_array_iter(json_str.as_str()).flatten() {
            ndjson.push_str(lv.as_raw_str());
            ndjson.push('\n');
        }
    } else {
        ndjson.push_str(&json_str);
        ndjson.push('\n');
    }
}

/// Render protocol-neutral DDL results to JSON rows.
///
/// Status and empty results keep their prior JSON shapes (`{type, tag}` /
/// `{type: "empty"}`). Row-returning results (SHOW / EXPLAIN / introspection)
/// emit one JSON object per row instead of a stub note — HTTP can return
/// DDL/SHOW rows directly.
pub(super) fn ddl_results_to_json(
    results: Vec<crate::control::server::shared::ddl::DdlResult>,
) -> Vec<serde_json::Value> {
    use crate::control::server::shared::ddl::DdlResult;

    let mut rows = Vec::new();
    for result in results {
        match result {
            DdlResult::Status { command, .. } => {
                rows.push(serde_json::json!({
                    "type": "execution",
                    "tag": command,
                }));
            }
            DdlResult::Rows(shaped) => {
                for row in shaped.rows {
                    rows.push(serde_json::Value::Object(row));
                }
            }
            DdlResult::Empty => {
                rows.push(serde_json::json!({ "type": "empty" }));
            }
        }
    }
    rows
}