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
126
127
128
129
130
131
132
133
134
135
136

// SPDX-License-Identifier: BUSL-1.1

//! Protocol-neutral `LAST_VALUE` and `LAST_VALUES` query handlers.
//!
//! Syntax:
//! ```sql
//! SELECT LAST_VALUES('<collection>')
//! SELECT LAST_VALUE('<collection>', <series_id>)
//! ```
//!
//! These dispatch `MetaOp::QueryLastValues` / `QueryLastValue` to the Data Plane
//! and return results as protocol-neutral rows.

use std::time::Duration;

use serde_json::{Map, Value as JsonValue};

use crate::bridge::envelope::PhysicalPlan;
use crate::control::security::identity::AuthenticatedIdentity;
use crate::control::server::response_shape::types::{DdlColType, ShapedRows};
use crate::control::state::SharedState;
use crate::types::DatabaseId;
use nodedb_physical::physical_plan::MetaOp;

use super::super::result::{DdlError, DdlResult};

/// `SELECT LAST_VALUES('<collection>')` — returns all cached last values.
pub async fn query_last_values(
    state: &SharedState,
    identity: &AuthenticatedIdentity,
    database_id: DatabaseId,
    collection: &str,
) -> Result<Vec<DdlResult>, DdlError> {
    let tenant_id = identity.tenant_id;
    let plan = PhysicalPlan::Meta(MetaOp::QueryLastValues {
        collection: collection.to_string(),
    });

    let payload = crate::control::server::shared::ddl::sync_dispatch::dispatch_async(
        state,
        tenant_id,
        database_id,
        collection,
        plan,
        Duration::from_secs(5),
    )
    .await
    .map_err(|e| ddl_err("XX000", format!("dispatch failed: {e}")))?;

    let entries: Vec<(u64, i64, f64)> = sonic_rs::from_slice(&payload).unwrap_or_default();

    let mut rows = Vec::with_capacity(entries.len());
    for (series_id, ts, value) in &entries {
        let mut row = Map::new();
        row.insert(
            "series_id".to_string(),
            JsonValue::String((*series_id as i64).to_string()),
        );
        row.insert(
            "timestamp_ms".to_string(),
            JsonValue::String(ts.to_string()),
        );
        row.insert(
            "value".to_string(),
            JsonValue::String(format!("{value:.6}")),
        );
        rows.push(row);
    }

    Ok(vec![DdlResult::Rows(ShapedRows {
        columns: vec![
            "series_id".to_string(),
            "timestamp_ms".to_string(),
            "value".to_string(),
        ],
        column_types: vec![DdlColType::Int8, DdlColType::Int8, DdlColType::Text],
        rows,
        notice: None,
    })])
}

/// `SELECT LAST_VALUE('<collection>', <series_id>)` — returns single series value.
pub async fn query_last_value(
    state: &SharedState,
    identity: &AuthenticatedIdentity,
    database_id: DatabaseId,
    collection: &str,
    series_id: u64,
) -> Result<Vec<DdlResult>, DdlError> {
    let tenant_id = identity.tenant_id;
    let plan = PhysicalPlan::Meta(MetaOp::QueryLastValue {
        collection: collection.to_string(),
        series_id,
    });

    let payload = crate::control::server::shared::ddl::sync_dispatch::dispatch_async(
        state,
        tenant_id,
        database_id,
        collection,
        plan,
        Duration::from_secs(5),
    )
    .await
    .map_err(|e| ddl_err("XX000", format!("dispatch failed: {e}")))?;

    let entry: Option<(i64, f64)> = sonic_rs::from_slice(&payload).unwrap_or_default();

    let mut rows = Vec::new();
    if let Some((ts, value)) = entry {
        let mut row = Map::new();
        row.insert(
            "timestamp_ms".to_string(),
            JsonValue::String(ts.to_string()),
        );
        row.insert(
            "value".to_string(),
            JsonValue::String(format!("{value:.6}")),
        );
        rows.push(row);
    }

    Ok(vec![DdlResult::Rows(ShapedRows {
        columns: vec!["timestamp_ms".to_string(), "value".to_string()],
        column_types: vec![DdlColType::Int8, DdlColType::Text],
        rows,
        notice: None,
    })])
}

fn ddl_err(sqlstate: &str, message: impl Into<String>) -> DdlError {
    DdlError {
        sqlstate: sqlstate.to_string(),
        message: message.into(),
    }
}