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
//
// Unified Query Algebra
//
// Copyright (c) 2023-2026 Cognica, Inc.
//
//! Tuple pins for `PostgreSQL`-style row-lock rechecks.
//!
//! When a candidate row's locked tuple changed while the statement waited, the recheck re-executes the plan below the `LockRows` boundary with every base relation pinned to the candidate's exact tuple, exactly like `PostgreSQL` 18's `EvalPlanQual`: marked relations substitute the latest committed image while unmarked join partners keep their original image.
use std::sync::Arc;
use uqa_execution::{PhysicalRow, RowSchema};
use uqa_storage::StoredDocument;
/// One tuple a pinned scan must emit during a recheck.
#[derive(Clone)]
pub(in crate::sql) struct RecheckDoc {
/// Row identity the pinned scan emits, already following any primary-key rewrite to the successor id.
pub doc_id: uqa_core::DocId,
/// Latest committed image for a changed tuple; `None` reads the statement snapshot image for `doc_id`.
pub document: Option<Arc<StoredDocument>>,
}
/// Pins for every lock-target relation of one candidate row.
pub(in crate::sql) struct RowLockRecheckPins {
targets: Vec<TargetPins>,
source_rows: Vec<SourceRowPin>,
}
struct TargetPins {
qualifier: String,
storage_name: String,
/// Qualifier of the base scan these tuples belong to. For a direct table target it equals `qualifier`; for an identity-source target it names the inner scan inside the view or derived table.
scan_qualifier: String,
identity_source: bool,
docs: Arc<Vec<RecheckDoc>>,
}
/// Exact output of one top-level FROM leaf that is not a row-lock target. `PostgreSQL` represents these as copy row marks during `EvalPlanQual`; keeping the positional row shares its physical fragments and preserves duplicate and volatile source values without named-row materialization.
#[derive(Clone)]
pub(in crate::sql) struct RecheckSourceRow {
pub schema: RowSchema,
pub row: PhysicalRow,
pub qualifier: String,
}
struct SourceRowPin {
path: Box<[u8]>,
source: RecheckSourceRow,
}
impl RowLockRecheckPins {
pub(in crate::sql) fn new() -> Self {
Self {
targets: Vec::new(),
source_rows: Vec::new(),
}
}
pub(in crate::sql) fn pin_source_row(
&mut self,
path: Vec<u8>,
qualifier: String,
schema: RowSchema,
row: PhysicalRow,
) {
self.source_rows.push(SourceRowPin {
path: path.into_boxed_slice(),
source: RecheckSourceRow {
schema,
row,
qualifier,
},
});
}
pub(in crate::sql) fn source_row(&self, path: &[u8]) -> Option<RecheckSourceRow> {
self.source_rows
.iter()
.find(|pin| pin.path.as_ref() == path)
.map(|pin| pin.source.clone())
}
pub(in crate::sql) fn pin_target(
&mut self,
qualifier: &str,
storage_name: &str,
scan_qualifier: &str,
identity_source: bool,
docs: Vec<RecheckDoc>,
) {
if let Some(target) = self.targets.iter_mut().find(|target| {
target.qualifier == qualifier
&& target.storage_name == storage_name
&& target.scan_qualifier == scan_qualifier
&& target.identity_source == identity_source
}) {
for doc in docs {
if !target
.docs
.iter()
.any(|existing| existing.doc_id == doc.doc_id)
{
Arc::make_mut(&mut target.docs).push(doc);
}
}
return;
}
self.targets.push(TargetPins {
qualifier: qualifier.to_string(),
storage_name: storage_name.to_string(),
scan_qualifier: scan_qualifier.to_string(),
identity_source,
docs: Arc::new(docs),
});
}
/// Pinned tuples for a base scan addressed by its own qualifier. Direct table targets match here; identity-source targets pin base scans through [`Self::storage_pins_for_identity_source`] instead because the origin qualifier was rebound at the derived-table boundary.
pub(in crate::sql) fn docs_for_scan(
&self,
qualifier: &str,
storage_name: &str,
) -> Option<Arc<Vec<RecheckDoc>>> {
self.targets
.iter()
.find(|target| {
!target.identity_source
&& target.qualifier == qualifier
&& recheck_storage_names_match(&target.storage_name, storage_name)
})
.map(|target| Arc::clone(&target.docs))
}
/// Scan-level pins for the subtree of one identity-source target (a view, derived table, or locked subquery visible as `qualifier`). Each entry names a base storage plus the inner scan qualifier whose emitted tuples it pins.
pub(in crate::sql) fn storage_pins_for_identity_source(
&self,
qualifier: &str,
) -> Vec<(String, String, Arc<Vec<RecheckDoc>>)> {
self.targets
.iter()
.filter(|target| target.identity_source && target.qualifier == qualifier)
.map(|target| {
(
target.storage_name.clone(),
target.scan_qualifier.clone(),
Arc::clone(&target.docs),
)
})
.collect()
}
}
pub(in crate::sql) fn recheck_storage_names_match(left: &str, right: &str) -> bool {
if left == right {
return true;
}
match (left.rsplit_once('.'), right.rsplit_once('.')) {
(Some((_, left_local)), None) => left_local == right,
(None, Some((_, right_local))) => left == right_local,
(None, None) | (Some(_), Some(_)) => false,
}
}