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
use super::ShardedVertexTable;
use crate::schema::ChangeDetails;
use crate::types::StoragePropertyDef;
use linkrs_core::{StorageError, StorageResult};
impl ShardedVertexTable {
pub fn schema(&self) -> crate::vertex::VertexSchema {
// Shard 0 is the schema authority: every shard holds the same schema
// and all schema mutations are applied to each shard in order.
self.shards[0].read().schema().clone()
}
/// Replace the schema on every shard without staging.
///
/// Recovery and undo compensation only: it replays an already-committed
/// schema (WAL redo, transaction rollback) rather than evolving live
/// state. All live schema evolution must go through the staged
/// prepare/fill/publish machine (`add_property`, `remove_property`,
/// `rename_property` and their staged fan-outs).
pub(crate) fn apply_schema(&self, schema: crate::vertex::VertexSchema) -> StorageResult<()> {
for shard in &self.shards {
shard.write().set_schema(schema.clone())?;
}
Ok(())
}
pub fn label(&self) -> linkrs_core::types::LabelId {
self.label
}
pub fn label_name(&self) -> &str {
&self.label_name
}
pub fn add_property(&self, prop: StoragePropertyDef) -> StorageResult<()> {
// Single-step convenience over the staged machine: prepare, fill and
// publish each fan out to every shard, aborting everywhere when a
// phase fails midway.
self.prepare_add_property_staged(prop)?;
if let Err(error) = self.fill_pending_schema_change() {
self.abort_pending_schema_change();
return Err(error);
}
if let Err(error) = self.publish_pending_schema_change() {
self.abort_pending_schema_change();
return Err(error);
}
Ok(())
}
pub fn remove_property(&self, prop_name: &str) -> StorageResult<()> {
self.prepare_remove_property_staged(prop_name)?;
if let Err(error) = self.fill_pending_schema_change() {
self.abort_pending_schema_change();
return Err(error);
}
if let Err(error) = self.publish_pending_schema_change() {
self.abort_pending_schema_change();
return Err(error);
}
Ok(())
}
pub fn rename_property(&self, old_name: &str, new_name: &str) -> StorageResult<()> {
self.prepare_rename_property_staged(old_name, new_name)?;
if let Err(error) = self.fill_pending_schema_change() {
self.abort_pending_schema_change();
return Err(error);
}
if let Err(error) = self.publish_pending_schema_change() {
self.abort_pending_schema_change();
return Err(error);
}
Ok(())
}
pub fn rebuild_schema_change_from_redo(&self, details: ChangeDetails) -> StorageResult<()> {
for shard in &self.shards {
shard
.write()
.rebuild_schema_change_from_redo(details.clone())?;
}
Ok(())
}
/// Abort staged changes on every shard holding one. Shards without a
/// pending change are skipped so compensation after a partial fan-out
/// never fails on already-clean shards.
fn abort_staged_on_all_shards(&self) {
for shard in &self.shards {
// Bind the check so the read guard is released before the write:
// parking_lot locks are not reentrant.
let pending = shard.read().has_pending_schema_change();
if pending {
let _ = shard.write().abort_pending_schema_change();
}
}
}
/// Staged add-column fan-out: prepare, fill and publish run as separate
/// passes so callers can interleave checkpoints between phases. Any phase
/// failing midway aborts the staged change on all shards, leaving no
/// residual column behind.
pub fn prepare_add_property_staged(&self, prop: StoragePropertyDef) -> StorageResult<()> {
for shard in &self.shards {
// Bind the result so the write guard drops before any abort
// re-locks this shard.
let result = shard.write().prepare_add_property_staged(prop.clone());
if let Err(error) = result {
self.abort_staged_on_all_shards();
return Err(error);
}
}
Ok(())
}
/// Staged drop-column fan-out with the same abort compensation.
pub fn prepare_remove_property_staged(&self, prop_name: &str) -> StorageResult<()> {
for shard in &self.shards {
let result = shard.write().prepare_remove_property_staged(prop_name);
if let Err(error) = result {
self.abort_staged_on_all_shards();
return Err(error);
}
}
Ok(())
}
/// Staged rename-column fan-out with the same abort compensation.
pub fn prepare_rename_property_staged(
&self,
old_name: &str,
new_name: &str,
) -> StorageResult<()> {
for shard in &self.shards {
let result = shard
.write()
.prepare_rename_property_staged(old_name, new_name);
if let Err(error) = result {
self.abort_staged_on_all_shards();
return Err(error);
}
}
Ok(())
}
/// Fill the staged change on every shard, aborting everywhere on failure.
pub fn fill_pending_schema_change(&self) -> StorageResult<()> {
for shard in &self.shards {
let result = shard.write().fill_pending_schema_change();
if let Err(error) = result {
self.abort_staged_on_all_shards();
return Err(error);
}
}
Ok(())
}
/// Publish the staged change on every shard with forward recovery.
///
/// A published shard cannot be unpublished, so a mid-fan-out failure
/// completes the publish on every remaining shard and reports the first
/// error instead of aborting: abort would leave published and staged
/// shards diverged, while forward completion converges every shard on
/// the published schema.
pub fn publish_pending_schema_change(&self) -> StorageResult<()> {
let mut first_error: Option<StorageError> = None;
for shard in &self.shards {
if let Err(error) = shard.write().publish_pending_schema_change() {
if first_error.is_none() {
first_error = Some(error);
}
}
}
if let Some(error) = first_error {
return Err(error);
}
Ok(())
}
/// Fail when shards disagree on the published schema version.
///
/// Column decoding trusts one layout, so a divergence must surface as
/// an error at the scan entry instead of a silent per-row downgrade.
pub fn verify_shard_schema_uniform(&self) -> StorageResult<()> {
let mut versions = std::collections::HashSet::new();
for shard in &self.shards {
versions.insert(shard.read().schema().schema_version);
}
if versions.len() > 1 {
return Err(StorageError::invalid_operation(format!(
"vertex table '{}' shards disagree on schema version: {:?}",
self.label_name, versions
)));
}
Ok(())
}
/// Abort the staged change on every shard holding one.
pub fn abort_pending_schema_change(&self) {
self.abort_staged_on_all_shards();
}
}