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
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
//! The lifecycle of an UNLOGGED TVIEW's rows: trusted until PostgreSQL resets the
//! table, then refilled once.
/// Whether an UNLOGGED TVIEW's rows can be trusted, and its fill when they can't.
///
/// PostgreSQL empties every UNLOGGED table on a crash restart and on promotion,
/// and leaves no other trace of it. `pg_tview_valid` is UNLOGGED too and holds
/// a row per UNLOGGED TVIEW whose rows are trusted, by entity, so the reset
/// empties both together. A missing row makes the next write fill that TVIEW
/// from its view; inserting the row is the claim, so concurrent writers wait for
/// the claimer instead of filling twice (#214). The rows are extension
/// configuration: `pg_dump` writes them exactly when it writes UNLOGGED tables'
/// rows, so a restore brings back both or neither (#236).
pub mod validity {
use crate::TViewResult;
use crate::catalog::TviewMeta;
use pgrx::pg_sys::{self, Oid};
use pgrx::prelude::*;
thread_local! {
/// The TVIEW tables this backend found trusted (or filled).
static CHECKED: std::cell::RefCell<std::collections::HashSet<u32>> =
std::cell::RefCell::new(std::collections::HashSet::new());
/// The TVIEW tables whose refill this transaction claimed, with the
/// (sub)transaction nesting level that claimed each.
static CLAIMS: std::cell::RefCell<Vec<(Oid, i32)>> =
const { std::cell::RefCell::new(Vec::new()) };
}
fn table() -> String {
format!("{}.pg_tview_valid", crate::utils::ext_schema())
}
fn unlogged(table_oid: Oid) -> bool {
// SAFETY: a syscache lookup; '\0' for a relation that no longer exists.
let persistence = unsafe { pg_sys::get_rel_persistence(table_oid) };
u8::try_from(persistence).ok() == Some(pg_sys::RELPERSISTENCE_UNLOGGED)
}
/// Whether `entity`'s table, `table_oid`, is UNLOGGED and its rows can't be
/// trusted.
///
/// # Errors
/// Returns an error if `pg_tview_valid` cannot be read.
pub fn needs_fill(entity: &str, table_oid: Oid) -> TViewResult<bool> {
if !unlogged(table_oid) {
return Ok(false);
}
let sql = format!(
"SELECT EXISTS (SELECT 1 FROM {} WHERE entity = $1)",
table()
);
// Read-only (`select`): a status call must not assign a transaction id.
let marked = Spi::connect(|client| {
client
.select(&sql, Some(1), &[crate::utils::spi::text(entity)])?
.first()
.get_one::<bool>()
})
.map_err(|e| crate::utils::spi::error(&sql, &e))?;
Ok(marked != Some(true))
}
/// Record that the rows of `entity`'s table, `table_oid`, filled in this
/// transaction, can be trusted. Nothing for a LOGGED table.
///
/// # Errors
/// Returns an error if `pg_tview_valid` cannot be written.
pub fn mark(entity: &str, table_oid: Oid) -> TViewResult<bool> {
if !unlogged(table_oid) {
return Ok(false);
}
let sql = format!(
"INSERT INTO {} VALUES ($1) ON CONFLICT DO NOTHING RETURNING entity",
table()
);
let _owner = crate::owner::AsOwner::of_extension()?;
// No row when it was there already (`first()` of an empty result is an error).
let inserted = Spi::connect_mut(|client| {
Ok::<_, pgrx::spi::Error>(
!client
.update(&sql, None, &[crate::utils::spi::text(entity)])?
.is_empty(),
)
})
.map_err(|e| crate::utils::spi::error(&sql, &e))?;
Ok(inserted)
}
/// Forget `entity`: it is dropped, or no longer UNLOGGED.
///
/// # Errors
/// Returns an error if `pg_tview_valid` cannot be written.
pub fn forget(entity: &str) -> TViewResult<()> {
let sql = format!("DELETE FROM {} WHERE entity = $1", table());
let _owner = crate::owner::AsOwner::of_extension()?;
crate::utils::spi::run(&sql, &[crate::utils::spi::text(entity)])
}
/// Make `entity`'s rows trustworthy before a write refreshes them: once per
/// entity and backend, since a reset ends every backend. See [`fill_if_reset`].
///
/// # Errors
/// Returns an error if the catalog cannot be read or a fill fails.
pub fn ensure(entity: &str) -> TViewResult<bool> {
let Some(meta) = TviewMeta::load_by_entity(entity)? else {
return Ok(false);
};
let table = meta.tview_oid.to_u32();
if CHECKED.with_borrow(|checked| checked.contains(&table)) {
return Ok(false);
}
let filled = fill_if_reset(entity)?.is_some();
CHECKED.with_borrow_mut(|checked| checked.insert(table));
Ok(filled)
}
/// Check the tables again: a rolled-back (sub)transaction may have undone a
/// fill, and a prepared one may not commit.
pub fn forget_checks() {
CHECKED.with_borrow_mut(std::collections::HashSet::clear);
}
/// The transaction ended: its claims ended with it.
pub fn forget_claims() {
CLAIMS.with_borrow_mut(Vec::clear);
}
/// A subtransaction rolled back: the claims it made are undone.
pub fn forget_subtransaction_claims() {
// SAFETY: reads the backend's transaction state.
let level = unsafe { pg_sys::GetCurrentTransactionNestLevel() };
CLAIMS.with_borrow_mut(|claims| claims.retain(|&(_, at)| at < level));
}
/// Refuse to prepare a transaction that claimed a refill: the claim, a row of
/// `pg_tview_valid`, would block every writer of that TVIEW until
/// `COMMIT PREPARED`. Runs in the `PRE_PREPARE` callback: no SPI.
pub fn refuse_prepare_with_claims() {
let Some(table) = CLAIMS.with_borrow(|claims| claims.first().map(|&(t, _)| t)) else {
return;
};
crate::TViewError::PrepareRefused {
reason: crate::error::PrepareRefusal::Refill {
table: crate::utils::qualified_relname_from_oid(table)
.unwrap_or_else(|_| format!("with OID {}", table.to_u32())),
},
}
.raise();
}
/// If `entity`'s table is UNLOGGED and was reset, fill the TVIEWs it reads,
/// then claim and fill it. Returns the rows inserted when this transaction
/// filled it.
///
/// # Errors
/// Returns an error if the catalog cannot be read or a fill fails.
pub fn fill_if_reset(entity: &str) -> TViewResult<Option<u64>> {
let Some(meta) = TviewMeta::load_by_entity(entity)? else {
return Ok(None);
};
if !needs_fill(entity, meta.tview_oid)? {
return Ok(None);
}
let graph = crate::cache::graph()?;
for dependency in graph.children.get(entity).into_iter().flatten() {
fill_if_reset(dependency)?;
}
// Another transaction may have claimed it since: the claim then waits
// for it and finds the row.
if !mark(entity, meta.tview_oid)? {
return Ok(None);
}
// SAFETY: reads the backend's transaction state.
let level = unsafe { pg_sys::GetCurrentTransactionNestLevel() };
CLAIMS.with_borrow_mut(|claims| claims.push((meta.tview_oid, level)));
crate::refresh::full::refill(entity).map(Some)
}
/// Before `entity`'s table, `table_oid`, becomes LOGGED (`logged`) or UNLOGGED:
/// a reset TVIEW is filled first, since a LOGGED one is never checked again.
///
/// # Errors
/// Returns an error if the catalog cannot be read or the fill fails.
pub fn before_persistence_change(entity: &str, logged: bool) -> TViewResult<()> {
if logged {
fill_if_reset(entity)?;
}
Ok(())
}
/// After `entity`'s table, `table_oid`, became LOGGED (`logged`) or UNLOGGED: a
/// LOGGED table has no row; the rows of one just made UNLOGGED were trusted
/// until now.
///
/// # Errors
/// Returns an error if `pg_tview_valid` cannot be written.
pub fn after_persistence_change(entity: &str, table_oid: Oid, logged: bool) -> TViewResult<()> {
if logged {
forget(entity)
} else {
mark(entity, table_oid).map(|_| ())
}
}
}