rupg_func/
snapshot.rs

1//! The functions of `pg_snapshot` and `txid_snapshot` from `xid8funcs.c`: `xmin`, `xmax`, the transactions in progress and the visibility of a transaction, and the functions of the ID, the snapshot and the state of a transaction. Also `age` and `mxid_age` of `xid` from `xid.c`. The functions of `txid_snapshot` have the same `prosrc` as the functions of `pg_snapshot`, and they give `bigint` where these give `xid8`. Both hold a `u64` in the bits of [`Value::Int8`].
2
3use rupg_common::{Error, Result, SqlState};
4use rupg_types::{Snapshot, Value};
5
6use crate::{Call, Kernel, bad_value, not_yet};
7
8/// `FirstNormalTransactionId`: the IDs before it are the IDs of no transaction, of the bootstrap and of the frozen rows.
9const FIRST_NORMAL_XID: u32 = 3;
10
11/// The state of a transaction for `pg_xact_status`, as `TransactionIdIsInProgress` and `TransactionIdDidCommit` give it.
12#[derive(Clone, Copy, Debug, PartialEq, Eq)]
13pub enum XactState {
14    /// The transaction did not end.
15    InProgress,
16    /// The transaction committed.
17    Committed,
18    /// The transaction aborted, or its session ended before it committed.
19    Aborted,
20}
21
22/// The kernel of a function of `pg_snapshot` by its `prosrc`.
23pub(crate) fn by_src(src: &str) -> Option<Kernel> {
24    Some(match src {
25        "pg_snapshot_xmin" => snapshot_xmin,
26        "pg_snapshot_xmax" => snapshot_xmax,
27        "pg_visible_in_snapshot" => visible_in_snapshot,
28        "pg_current_xact_id" => current_xact_id,
29        "pg_current_xact_id_if_assigned" => current_xact_id_if_assigned,
30        "pg_current_snapshot" => current_snapshot,
31        "pg_xact_status" => xact_status,
32        "xid_age" => xid_age,
33        "mxid_age" => mxid_age,
34        _ => return None,
35    })
36}
37
38/// `ReadNextMultiXactId`: rupg has no multixacts, so the next is the first.
39const FIRST_MULTI_XACT_ID: u32 = 1;
40
41/// `xid_age`: the number of transactions from `xid` to [`crate::Session::stable_transaction_id`]. An ID that is not normal is the oldest.
42fn xid_age(call: &Call<'_>, args: &[Value]) -> Result<Value> {
43    let Some(Value::Oid(xid)) = args.first() else { return Err(bad_value()) };
44    if *xid < FIRST_NORMAL_XID {
45        return Ok(Value::Int4(i32::MAX));
46    }
47    Ok(Value::Int4(call.session.stable_transaction_id().wrapping_sub(*xid).cast_signed()))
48}
49
50/// `mxid_age`: the number of multixacts from `xid` to the next multixact. The ID 0 is the oldest.
51fn mxid_age(_: &Call<'_>, args: &[Value]) -> Result<Value> {
52    let Some(Value::Oid(xid)) = args.first() else { return Err(bad_value()) };
53    if *xid == 0 {
54        return Ok(Value::Int4(i32::MAX));
55    }
56    Ok(Value::Int4(FIRST_MULTI_XACT_ID.wrapping_sub(*xid).cast_signed()))
57}
58
59/// The snapshot of an argument.
60fn snapshot(args: &[Value], n: usize) -> Result<Snapshot> {
61    match args.get(n) {
62        Some(Value::Bytea(v)) => Snapshot::from_bytes(v).ok_or_else(bad_value),
63        _ => Err(bad_value()),
64    }
65}
66
67fn snapshot_xmin(_: &Call<'_>, args: &[Value]) -> Result<Value> {
68    Ok(Value::Int8(snapshot(args, 0)?.xmin.cast_signed()))
69}
70
71fn snapshot_xmax(_: &Call<'_>, args: &[Value]) -> Result<Value> {
72    Ok(Value::Int8(snapshot(args, 0)?.xmax.cast_signed()))
73}
74
75/// `pg_visible_in_snapshot`: true when the transaction ended before the snapshot.
76fn visible_in_snapshot(_: &Call<'_>, args: &[Value]) -> Result<Value> {
77    let Some(Value::Int8(xid)) = args.first() else { return Err(bad_value()) };
78    Ok(Value::Bool(snapshot(args, 1)?.is_visible(xid.cast_unsigned())))
79}
80
81/// `pg_snapshot_xip`: a row for each transaction in progress, in order.
82pub(crate) fn snapshot_xip(_: &Call<'_>, args: &[Value]) -> Result<Vec<Vec<Value>>> {
83    let snapshot = snapshot(args, 0)?;
84    Ok(snapshot.xip.iter().map(|xid| vec![Value::Int8(xid.cast_signed())]).collect())
85}
86
87/// `pg_current_xact_id`: the ID of the transaction. A transaction that has no ID gets one, as `GetTopFullTransactionId` assigns it.
88fn current_xact_id(call: &Call<'_>, _: &[Value]) -> Result<Value> {
89    let Some(xid) = call.session.transaction_id(true) else {
90        return Err(not_yet("a transaction ID in a session with no server"));
91    };
92    Ok(Value::Int8(xid.cast_signed()))
93}
94
95/// `pg_current_xact_id_if_assigned`: the ID of the transaction, or NULL when it has none.
96fn current_xact_id_if_assigned(call: &Call<'_>, _: &[Value]) -> Result<Value> {
97    Ok(call.session.transaction_id(false).map_or(Value::Null, |xid| Value::Int8(xid.cast_signed())))
98}
99
100/// `pg_current_snapshot`: the snapshot of the statement.
101fn current_snapshot(call: &Call<'_>, _: &[Value]) -> Result<Value> {
102    let Some(snapshot) = call.session.current_snapshot() else {
103        return Err(Error::internal("no active snapshot set"));
104    };
105    Ok(Value::Bytea(snapshot.to_bytes()))
106}
107
108/// `pg_xact_status` with the checks of `TransactionIdInRecentPast`: NULL for the ID 0 and for a transaction that the session does not know, `committed` for the IDs of the bootstrap and of the frozen rows, and an error for an ID in the future.
109fn xact_status(call: &Call<'_>, args: &[Value]) -> Result<Value> {
110    let Some(Value::Int8(full)) = args.first() else { return Err(bad_value()) };
111    let full = full.cast_unsigned();
112    let xid = u32::try_from(full & u64::from(u32::MAX)).unwrap_or_default();
113    if xid == 0 {
114        return Ok(Value::Null);
115    }
116    if xid < FIRST_NORMAL_XID {
117        return Ok(Value::Text("committed".into()));
118    }
119    if full >= call.session.next_transaction_id() {
120        return Err(Error::new(
121            SqlState::INVALID_PARAMETER_VALUE,
122            format!("transaction ID {full} is in the future"),
123        ));
124    }
125    Ok(match call.session.transaction_state(xid) {
126        None => Value::Null,
127        Some(XactState::InProgress) => Value::Text("in progress".into()),
128        Some(XactState::Committed) => Value::Text("committed".into()),
129        Some(XactState::Aborted) => Value::Text("aborted".into()),
130    })
131}
rupg_func/snapshot.rs

rupg_func/
snapshot.rs
