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
207
208
use async_trait::async_trait;
use crate::LixError;
use crate::hot_state::MaterializedHotStateBatchBuilder;
use crate::hot_state::{HotStateExactBatchRequest, HotStateScanRequest};
use crate::hot_state::{MaterializedHotStateBatch, MaterializedHotStateExactBatch};
/// Selects the authenticated serving domain for an internal read.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum HotStateReadDomain {
Combined,
Tracked,
Untracked,
}
/// Minimal engine read model for transaction planning and SQL providers.
///
/// Engine only needs visible state-row reads here. Changelog freshness/catch-up
/// should be added at this boundary later instead of leaking projection internals
/// into sessions or SQL providers.
#[async_trait]
pub(crate) trait HotStateReader: Send + Sync {
fn is_partial_replica(&self) -> bool {
false
}
fn read_interest_registry(&self) -> Option<std::sync::Arc<super::ReadInterestRegistry>> {
None
}
/// Columnar scan lane used by live-state composition and SQL providers.
///
async fn scan_batch(
&self,
request: &HotStateScanRequest,
) -> Result<MaterializedHotStateBatch, LixError>;
/// Scans committed rows for constraint validation into one shared owner.
///
/// Durable readers can override this lane to use publication metadata for
/// a conservative empty-schema proof. Other readers preserve correctness
/// by falling back to their ordinary columnar scan implementation.
async fn scan_constraint_batch(
&self,
request: &HotStateScanRequest,
tracked_only: bool,
) -> Result<MaterializedHotStateBatch, LixError> {
if tracked_only {
self.scan_tracked_batch(request).await
} else {
self.scan_batch(request).await
}
}
/// Reads one explicit authenticated serving domain. Ordinary visibility
/// callers use `Combined`; internal historical/row validation and
/// current-only durable callers select their domain explicitly.
async fn scan_domain_batch(
&self,
request: &HotStateScanRequest,
domain: HotStateReadDomain,
) -> Result<MaterializedHotStateBatch, LixError> {
match domain {
HotStateReadDomain::Combined => self.scan_constraint_batch(request, false).await,
HotStateReadDomain::Tracked => {
let mut request = request.clone();
request.filter.untracked = Some(false);
self.scan_constraint_batch(&request, true).await
}
HotStateReadDomain::Untracked => {
let mut request = request.clone();
request.filter.untracked = Some(true);
self.scan_constraint_batch(&request, false).await
}
}
}
/// Serves a declared-column equality through a correlated identity index
/// when the reader owns one. `Some` is a complete candidate superset for
/// the selected predicate (including an empty result); stale entries are
/// still rechecked by the caller. `None` asks the caller to use its
/// ordinary scan path. The default keeps lightweight test and overlay
/// readers conservative.
async fn scan_indexed_declared_column_batch(
&self,
_request: &HotStateScanRequest,
_domain: HotStateReadDomain,
) -> Result<Option<MaterializedHotStateBatch>, LixError> {
Ok(None)
}
/// Scans the immutable tracked head selected by the current branch ref.
///
/// Normal SQL reads use [`Self::scan_batch`] and therefore see exactly one
/// canonical current row. Validation and schema planning use this explicit
/// durability view when a tracked commit must not depend on untracked
/// live state. Readers that wrap canonical current scans must override
/// this method instead of relying on the fallback below.
/// The default keeps the explicit tracked filter in the batch
/// lane. Readers with a distinct immutable tracked source override this.
async fn scan_tracked_batch(
&self,
request: &HotStateScanRequest,
) -> Result<MaterializedHotStateBatch, LixError> {
let mut request = request.clone();
request.filter.untracked = Some(false);
self.scan_batch(&request).await
}
/// Loads concrete visible identities while preserving request alignment.
///
/// Readers must provide the correlated batch path directly.
/// There is no scan-based default because silently lowering this operation
/// to one scan per row would reintroduce the amplification this API exists
/// to prevent.
async fn load_exact_batch(
&self,
request: &HotStateExactBatchRequest,
) -> Result<MaterializedHotStateExactBatch, LixError>;
/// Resolve a parent chain in the same visible domain as the initial rows.
/// The caller retains the semantic dependency recipe; point interests alone
/// cannot describe parents introduced by a later generation.
async fn load_exact_parent_closure(
&self,
request: &HotStateExactBatchRequest,
parent: ParentRowPk,
) -> Result<MaterializedHotStateBatch, LixError> {
load_parent_closure_with(request, parent, |batch| async move {
self.load_exact_batch(&batch).await
}).await
}
/// Loads collection control from this reader's coherent storage snapshot.
///
/// Readers without a published tracked-head projection conservatively
/// decline the proof so callers retain their ordinary identity scans.
async fn collection_generation(
&self,
_branch_id: &str,
_scope: crate::collection_generation::CollectionScopeRef<'_>,
) -> Result<Option<crate::collection_generation::CollectionGeneration>, LixError> {
Ok(None)
}
}
#[cfg(test)]
pub(crate) async fn load_exact_batch_via_scan_for_test<R>(
reader: &R,
request: &HotStateExactBatchRequest,
) -> Result<MaterializedHotStateExactBatch, LixError>
where
R: HotStateReader + ?Sized,
{
let mut rows = MaterializedHotStateBatchBuilder::with_capacity(request.rows.len());
let mut slots = Vec::with_capacity(request.rows.len());
for row in &request.rows {
let scanned = reader.scan_batch(&request.row_scan_request(row)).await?;
slots.push(
scanned
.get(0)
.map(|row| u32::try_from(rows.push_ref(row, None)))
.transpose()
.map_err(|_| {
LixError::new(
LixError::CODE_INTERNAL_ERROR,
"test exact live-state result exceeds u32 rows",
)
})?,
);
}
MaterializedHotStateExactBatch::new(rows.finish(), slots)
}
/// Pure identity extraction, never SQL expression evaluation.
pub(crate) type ParentRowPk = fn(super::MaterializedHotStateRowRef<'_>) -> Result<Option<crate::row_pk::RowPk>, LixError>;
pub(crate) async fn load_parent_closure_with<F, Fut>(
request: &HotStateExactBatchRequest,
parent: ParentRowPk,
mut load: F,
) -> Result<MaterializedHotStateBatch, LixError>
where
F: FnMut(HotStateExactBatchRequest) -> Fut + Send,
Fut: Future<Output = Result<MaterializedHotStateExactBatch, LixError>> + Send,
{
let mut pending = request.rows.clone();
let mut visited = std::collections::BTreeSet::new();
let mut batches = Vec::new();
loop {
pending.retain(|row| visited.insert((row.branch_id.clone(), row.schema_key.clone(), row.file_id.clone(), row.row_pk.clone())));
if pending.is_empty() { break; }
let batch = load(HotStateExactBatchRequest { rows: std::mem::take(&mut pending), ..request.clone() }).await?.into_present_batch();
for row in batch.iter() {
if let Some(row_pk) = parent(row)? {
pending.push(super::HotStateExactRowRequest {
branch_id: row.branch_id().to_owned(),
schema_key: row.schema_key().to_owned(),
file_id: row.file_id().map(str::to_owned),
row_pk,
});
}
}
batches.push(batch);
}
let mut builder = MaterializedHotStateBatchBuilder::with_capacity(batches.iter().map(MaterializedHotStateBatch::len).sum());
for batch in &batches { for row in batch.iter() { builder.push_ref(row, None); } }
Ok(builder.finish())
}