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

// SPDX-License-Identifier: BUSL-1.1

//! `SHOW PARTITIONS FOR <name>`
//!
//! Reachability note: the `SHOW PARTITIONS ` prefix is claimed by the
//! consumer-group handler earlier in the neutral router (mirroring the pgwire
//! streaming router, which ran before engine_ops), so this handler is shadowed
//! for that prefix exactly as it was on the pgwire path. It is ported here to
//! preserve the engine_ops handler set verbatim.

use crate::control::security::identity::AuthenticatedIdentity;
use crate::control::server::response_shape::types::{DdlColType, ShapedRows};
use crate::control::state::SharedState;
use nodedb_types::DatabaseId;
use serde_json::{Map, Value as JsonValue};

use super::super::super::result::{DdlError, DdlResult};
use super::helpers::{ddl_err, format_bytes};

/// SHOW PARTITIONS FOR <name>
pub fn show_partitions(
    state: &SharedState,
    identity: &AuthenticatedIdentity,
    parts: &[&str],
) -> Result<Vec<DdlResult>, DdlError> {
    if parts.len() < 4 {
        return Err(ddl_err("42601", "syntax: SHOW PARTITIONS FOR <collection>"));
    }

    let name = parts[3].to_lowercase();
    let tenant_id = identity.tenant_id;

    // Verify collection exists and is timeseries.
    {
        let catalog = state.credentials.catalog();
        match catalog.get_collection(DatabaseId::DEFAULT, tenant_id.as_u64(), &name) {
            Ok(Some(coll)) if coll.collection_type.is_timeseries() => {}
            Ok(Some(_)) => {
                return Err(ddl_err(
                    "42809",
                    format!("'{name}' is not a timeseries collection"),
                ));
            }
            _ => {
                return Err(ddl_err(
                    "42P01",
                    format!("collection '{name}' does not exist"),
                ));
            }
        }
    }

    let mut rows = Vec::new();

    if let Some(registries) = state.timeseries_registries() {
        let regs = crate::control::lock_utils::lock_or_recover(registries.lock(), "ts_registries");
        let key = format!("{}:{}", tenant_id.as_u64(), name);
        if let Some(registry) = regs.get(&key) {
            for (_, entry) in registry.iter() {
                if !entry.meta.is_queryable() {
                    continue;
                }
                let mut row = Map::new();
                row.insert(
                    "partition".to_string(),
                    JsonValue::String(entry.dir_name.clone()),
                );
                row.insert(
                    "min_ts".to_string(),
                    JsonValue::String(entry.meta.min_ts.to_string()),
                );
                row.insert(
                    "max_ts".to_string(),
                    JsonValue::String(entry.meta.max_ts.to_string()),
                );
                row.insert(
                    "rows".to_string(),
                    JsonValue::String((entry.meta.row_count as i64).to_string()),
                );
                row.insert(
                    "size".to_string(),
                    JsonValue::String(format_bytes(entry.meta.size_bytes)),
                );
                row.insert(
                    "state".to_string(),
                    JsonValue::String(format!("{:?}", entry.meta.state)),
                );
                rows.push(row);
            }
        }
    }

    Ok(vec![DdlResult::Rows(ShapedRows {
        columns: vec![
            "partition".to_string(),
            "min_ts".to_string(),
            "max_ts".to_string(),
            "rows".to_string(),
            "size".to_string(),
            "state".to_string(),
        ],
        column_types: vec![
            DdlColType::Text,
            DdlColType::Int8,
            DdlColType::Int8,
            DdlColType::Int8,
            DdlColType::Text,
            DdlColType::Text,
        ],
        rows,
        notice: None,
    })])
}