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

// SPDX-License-Identifier: BUSL-1.1

//! Savepoint control adapters for the native protocol: SAVEPOINT, RELEASE
//! SAVEPOINT, ROLLBACK TO SAVEPOINT — thin shims over the protocol-neutral
//! savepoint orchestrator in `control/server/shared/session/savepoint_ops.rs`.
//!
//! The overlay marker capture/decode and the `SessionStore` savepoint stack
//! live in the neutral core; native only parses the statement, shares the same
//! Data-Plane dispatch seam as COMMIT/ROLLBACK, and maps the neutral error to a
//! native SQLSTATE frame (`25P01` / `3B001`).

use nodedb_types::protocol::NativeResponse;

use crate::control::server::shared::session::savepoint_ops::{self, SavepointError};

use super::DispatchCtx;
use super::transaction::NativeTxnDp;

/// Map a neutral savepoint error to a native error frame.
fn savepoint_error_to_native(seq: u64, e: &SavepointError) -> NativeResponse {
    match e {
        SavepointError::NoActiveTransaction => NativeResponse::error(
            seq,
            "25P01",
            "SAVEPOINT can only be used in transaction blocks",
        ),
        SavepointError::NotFound { message } => {
            NativeResponse::error(seq, "3B001", message.clone())
        }
    }
}

/// Handle SAVEPOINT <name>.
pub(crate) async fn handle_savepoint(
    ctx: &DispatchCtx<'_>,
    seq: u64,
    sql_trimmed: &str,
) -> NativeResponse {
    let sp_name = sql_trimmed.split_whitespace().nth(1).unwrap_or("sp");
    let dp = NativeTxnDp { state: ctx.state };
    match savepoint_ops::run_savepoint(ctx.sessions, ctx.peer_addr, ctx.tenant_id(), &dp, sp_name)
        .await
    {
        Ok(()) => NativeResponse::status_row(seq, "SAVEPOINT"),
        Err(e) => savepoint_error_to_native(seq, &e),
    }
}

/// Handle RELEASE SAVEPOINT <name>.
pub(crate) fn handle_release_savepoint(
    ctx: &DispatchCtx<'_>,
    seq: u64,
    sql_trimmed: &str,
) -> NativeResponse {
    let sp_name = sql_trimmed.split_whitespace().last().unwrap_or("sp");
    match savepoint_ops::run_release_savepoint(ctx.sessions, ctx.peer_addr, sp_name) {
        Ok(()) => NativeResponse::status_row(seq, "RELEASE"),
        Err(e) => savepoint_error_to_native(seq, &e),
    }
}

/// Handle ROLLBACK TO SAVEPOINT <name>.
pub(crate) async fn handle_rollback_to_savepoint(
    ctx: &DispatchCtx<'_>,
    seq: u64,
    sql_trimmed: &str,
) -> NativeResponse {
    let sp_name = sql_trimmed.split_whitespace().last().unwrap_or("sp");
    let dp = NativeTxnDp { state: ctx.state };
    match savepoint_ops::run_rollback_to_savepoint(
        ctx.sessions,
        ctx.peer_addr,
        ctx.tenant_id(),
        &dp,
        sp_name,
    )
    .await
    {
        Ok(()) => NativeResponse::status_row(seq, "ROLLBACK"),
        Err(e) => savepoint_error_to_native(seq, &e),
    }
}