pg_tviews 0.1.0-beta.32

Transactional materialized views with incremental refresh for PostgreSQL
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

//! Functions acting on the caller's session or transaction.

use pgrx::pg_sys::panic::ErrorReport;
use pgrx::prelude::*;

/// Suspend trigger-based refresh in this transaction: writes record which TVIEWs
/// they change instead of refreshing them, until the matching resume or the end
/// of the transaction, which refreshes them.
#[pg_extern]
fn pg_tviews_suspend_triggers() {
    crate::suspend::suspend();
}

/// Resume trigger-based refresh. When the outermost suspension ends, every TVIEW
/// changed while suspended (and every TVIEW reading one of them) is rebuilt.
#[pg_extern]
fn pg_tviews_resume_triggers() -> Result<(), ErrorReport> {
    crate::revision::check();
    crate::suspend::resume()?;
    if !crate::suspend::is_suspended() {
        crate::suspend::catch_up()?;
    }
    Ok(())
}

/// Whether this transaction's trigger-based refresh is suspended.
#[pg_extern]
fn pg_tviews_is_suspended() -> bool {
    crate::suspend::is_suspended()
}

/// The TVIEWs changed while this transaction's refresh is suspended.
#[pg_extern]
fn pg_tviews_suspended_entities() -> Vec<String> {
    crate::suspend::get_changed_entities()
}

/// Flush the transaction's refresh queue and report the TVIEW rows it changed,
/// as `FraiseQL`'s cascade response reads them (`docs/user-guides/graphql-cascade.md`).
#[pg_extern]
fn pg_tviews_flush_and_report(
    max_entities: default!(i32, 500),
    include_data: default!(bool, true),
    reset: default!(bool, true),
    types: default!(Option<Vec<String>>, "NULL"),
) -> Result<pgrx::JsonB, ErrorReport> {
    crate::revision::check();
    Ok(pgrx::JsonB(crate::report::flush_and_report(
        max_entities,
        include_data,
        reset,
        types,
    )?))
}

/// The session's refresh counters (debugging; `tviews.stats` is per TVIEW and
/// readable from any session).
#[pg_extern]
fn pg_tviews_queue_stats() -> pgrx::JsonB {
    pgrx::JsonB(crate::health::queue_stats())
}

/// The transactions the cluster committed without publishing their changes on
/// the change feed (the NOTIFY queue being nearly full), since the server
/// started (#247). A listener reads it when it connects and on each heartbeat,
/// and evicts everything when it moved.
#[pg_extern]
fn pg_tviews_feed_gaps() -> Result<i64, ErrorReport> {
    let gaps = crate::stats::feed_gaps()?;
    Ok(i64::try_from(gaps).unwrap_or(i64::MAX))
}

/// The transaction's refresh queue (debugging).
#[pg_extern]
fn pg_tviews_debug_queue() -> pgrx::JsonB {
    pgrx::JsonB(crate::health::debug_queue())
}