mockgres 0.0.29

An in-memory database that replicates a reasonable subset of Postgres functionality to make unit tests that rely on a database to run.
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

use std::sync::Arc;

use parking_lot::RwLock;
use pgwire::error::PgWireResult;

use crate::db::{Db, LockOwner};
use crate::engine::fe_code;
use crate::server::errors::map_db_err;
use crate::session::{Session, now_utc_micros};
use crate::txn::{TransactionManager, TxId, TxnStatus};

pub(crate) fn writer_txid(
    session: &Arc<Session>,
    txn_manager: &Arc<TransactionManager>,
) -> (TxId, bool) {
    if let Some(txid) = session.current_tx() {
        (txid, false)
    } else {
        let txid = txn_manager.allocate();
        txn_manager.set_status(txid, TxnStatus::InProgress);
        (txid, true)
    }
}

pub(crate) fn finish_writer_tx(
    txn_manager: &Arc<TransactionManager>,
    txid: TxId,
    autocommit: bool,
    success: bool,
) {
    if autocommit {
        let status = if success {
            TxnStatus::Committed
        } else {
            TxnStatus::Aborted
        };
        txn_manager.set_status(txid, status);
    }
}

pub(crate) fn begin_transaction(
    session: &Arc<Session>,
    txn_manager: &Arc<TransactionManager>,
) -> PgWireResult<()> {
    if session.current_tx().is_some() {
        return Err(fe_code("25001", "transaction already in progress"));
    }
    let txid = txn_manager.allocate();
    txn_manager.set_status(txid, TxnStatus::InProgress);
    session.begin_transaction_epoch();
    session.set_current_tx(Some(txid));
    session.reset_changes();
    session.set_txn_start_micros(now_utc_micros());
    let iso = session.default_txn_isolation();
    session.set_txn_isolation(iso);
    Ok(())
}

pub(crate) fn commit_transaction(
    session: &Arc<Session>,
    txn_manager: &Arc<TransactionManager>,
    db: &Arc<RwLock<Db>>,
) -> PgWireResult<()> {
    let Some(txid) = session.current_tx() else {
        return Err(fe_code("25P01", "no transaction in progress"));
    };
    txn_manager.set_status(txid, TxnStatus::Committed);
    session.set_current_tx(None);
    session.reset_changes();
    session.clear_txn_start_micros();
    session.clear_txn_isolation();
    if let Some(epoch) = session.end_transaction_epoch() {
        let db_read = db.read();
        db_read.release_locks(LockOwner::new(session.id(), epoch));
    }
    Ok(())
}

pub(crate) fn rollback_transaction(
    session: &Arc<Session>,
    txn_manager: &Arc<TransactionManager>,
    db: &Arc<RwLock<Db>>,
) -> PgWireResult<()> {
    let Some(txid) = session.current_tx() else {
        return Err(fe_code("25P01", "no transaction in progress"));
    };
    let changes = session.take_changes();
    session.set_current_tx(None);
    session.clear_txn_start_micros();
    session.clear_txn_isolation();
    {
        let mut db = db.write();
        db.rollback_transaction_changes(&changes, txid)
            .map_err(map_db_err)?;
    }
    txn_manager.set_status(txid, TxnStatus::Aborted);
    if let Some(epoch) = session.end_transaction_epoch() {
        let db_read = db.read();
        db_read.release_locks(LockOwner::new(session.id(), epoch));
    }
    Ok(())
}