-
Notifications
You must be signed in to change notification settings - Fork 56
Expand file tree
/
Copy pathmod.rs
More file actions
4181 lines (3816 loc) · 167 KB
/
Copy pathmod.rs
File metadata and controls
4181 lines (3816 loc) · 167 KB
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
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
//! On-disk persistence for a `Database`, using fixed-size paged files.
//!
//! The file is a sequence of 4 KiB pages. Page 0 holds the header
//! (magic, version, page count, schema-root pointer). Every other page carries
//! a small per-page header (type tag + next-page pointer + payload length)
//! followed by a payload of up to 4089 bytes.
//!
//! **Storage strategy (format version 2, Phase 3c.5).**
//!
//! - Each `Table`'s rows live as **cells** in a chain of `TableLeaf` pages.
//! Cell layout and slot directory are in `cell.rs` / `table_page.rs`;
//! cells that exceed the inline threshold spill into an overflow chain
//! via `overflow.rs`.
//! - The schema catalog is itself a regular table named `sqlrite_master`,
//! with one row per user table:
//! `(name TEXT PRIMARY KEY, sql TEXT NOT NULL,
//! rootpage INTEGER NOT NULL, last_rowid INTEGER NOT NULL)`
//! This is the SQLite-style approach: the schema of `sqlrite_master`
//! itself is hardcoded into the engine so the open path can bootstrap.
//! - Page 0's `schema_root_page` field points at the first leaf of
//! `sqlrite_master`.
//!
//! **Format version.** Version 2 is not compatible with files produced by
//! earlier commits. Opening a v1 file returns a clean error — users on
//! old files have to regenerate them from CREATE/INSERT, as there's no
//! production data to migrate yet.
// Data-layer modules. Not every helper in these modules is used by save/open
// yet — some exist for tests, some for future maintenance operations.
// Module-level #[allow(dead_code)] keeps the build quiet without dotting
// the modules with per-item attributes.
#[allow(dead_code)]
pub mod allocator;
#[allow(dead_code)]
pub mod cell;
pub mod file;
#[allow(dead_code)]
pub mod freelist;
#[allow(dead_code)]
pub mod fts_cell;
pub mod header;
#[allow(dead_code)]
pub mod hnsw_cell;
#[allow(dead_code)]
pub mod index_cell;
#[allow(dead_code)]
pub mod interior_page;
pub mod overflow;
pub mod page;
pub mod pager;
#[allow(dead_code)]
pub mod table_page;
#[allow(dead_code)]
pub mod varint;
#[allow(dead_code)]
pub mod wal;
use std::collections::{BTreeMap, HashMap};
use std::path::Path;
use std::sync::{Arc, Mutex};
use crate::sql::dialect::SqlriteDialect;
use sqlparser::parser::Parser;
use crate::error::{Result, SQLRiteError};
use crate::sql::db::database::Database;
use crate::sql::db::secondary_index::{IndexOrigin, SecondaryIndex};
use crate::sql::db::table::{Column, DataType, Row, Table, Value};
use crate::sql::hnsw::DistanceMetric;
use crate::sql::pager::cell::Cell;
use crate::sql::pager::header::DbHeader;
use crate::sql::pager::index_cell::IndexCell;
use crate::sql::pager::interior_page::{InteriorCell, InteriorPage};
use crate::sql::pager::overflow::{
OVERFLOW_THRESHOLD, OverflowRef, PagedEntry, read_overflow_chain, write_overflow_chain,
};
use crate::sql::pager::page::{PAGE_HEADER_SIZE, PAGE_SIZE, PAYLOAD_PER_PAGE, PageType};
use crate::sql::pager::pager::Pager;
use crate::sql::pager::table_page::TablePage;
use crate::sql::parser::create::CreateQuery;
// Re-export so callers can spell `sql::pager::AccessMode` without
// reaching into the `pager::pager::pager` submodule path.
pub use crate::sql::pager::pager::AccessMode;
/// Name of the internal catalog table. Reserved — user CREATEs of this
/// name must be rejected upstream.
pub const MASTER_TABLE_NAME: &str = "sqlrite_master";
/// Opens a database file in read-write mode. Shorthand for
/// [`open_database_with_mode`] with [`AccessMode::ReadWrite`].
pub fn open_database(path: &Path, db_name: String) -> Result<Database> {
open_database_with_mode(path, db_name, AccessMode::ReadWrite)
}
/// Opens a database file in read-only mode. Acquires a shared OS-level
/// advisory lock, so other read-only openers coexist but any writer is
/// excluded. Attempts to mutate the returned `Database` (e.g. an
/// `INSERT`, or a `save_database` call against it) bottom out in a
/// `cannot commit: database is opened read-only` error from the Pager.
pub fn open_database_read_only(path: &Path, db_name: String) -> Result<Database> {
open_database_with_mode(path, db_name, AccessMode::ReadOnly)
}
/// Opens a database file and reconstructs the in-memory `Database`,
/// leaving the long-lived `Pager` attached for subsequent auto-save
/// (read-write) or consistent-snapshot reads (read-only).
pub fn open_database_with_mode(path: &Path, db_name: String, mode: AccessMode) -> Result<Database> {
let pager = Pager::open_with_mode(path, mode)?;
// 1. Load sqlrite_master from the tree at header.schema_root_page.
let mut master = build_empty_master_table();
load_table_rows(&pager, &mut master, pager.header().schema_root_page)?;
// 2. Two passes over master rows: first build every user table, then
// attach secondary indexes. Indexes need their base table to exist
// before we can populate them. Auto-indexes are created at table
// build time so we only have to load explicit indexes from disk
// (but we also reload the auto-index CONTENT because Table::new
// built it empty).
let mut db = Database::new(db_name);
let mut index_rows: Vec<IndexCatalogRow> = Vec::new();
for rowid in master.rowids() {
let ty = take_text(&master, "type", rowid)?;
let name = take_text(&master, "name", rowid)?;
let sql = take_text(&master, "sql", rowid)?;
let rootpage = take_integer(&master, "rootpage", rowid)? as u32;
let last_rowid = take_integer(&master, "last_rowid", rowid)?;
match ty.as_str() {
"table" => {
let (parsed_name, columns) = parse_create_sql(&sql)?;
if parsed_name != name {
return Err(SQLRiteError::Internal(format!(
"sqlrite_master row '{name}' carries SQL for '{parsed_name}' — corrupt catalog?"
)));
}
let mut table = build_empty_table(&name, columns, last_rowid);
if rootpage != 0 {
load_table_rows(&pager, &mut table, rootpage)?;
}
if last_rowid > table.last_rowid {
table.last_rowid = last_rowid;
}
db.tables.insert(name, table);
}
"index" => {
index_rows.push(IndexCatalogRow {
name,
sql,
rootpage,
});
}
other => {
return Err(SQLRiteError::Internal(format!(
"sqlrite_master row '{name}' has unknown type '{other}'"
)));
}
}
}
// Second pass: attach each index to its table. HNSW indexes
// (Phase 7d.2) take a different code path because their persisted
// form is just the CREATE INDEX SQL — the graph itself isn't
// persisted yet (Phase 7d.3). Detect HNSW via the SQL's USING clause
// and route to a graph-rebuild instead of the B-Tree-cell load.
//
// Phase 8b — same shape for FTS indexes. The posting lists aren't
// persisted yet (Phase 8c), so we replay the CREATE INDEX SQL on
// open and let `execute_create_index` walk current rows.
for row in index_rows {
if create_index_sql_uses_hnsw(&row.sql) {
rebuild_hnsw_index(&mut db, &pager, &row)?;
} else if create_index_sql_uses_fts(&row.sql) {
rebuild_fts_index(&mut db, &pager, &row)?;
} else {
attach_index(&mut db, &pager, row)?;
}
}
// Phase 11.9 — replay any MVCC commit batches recovered from
// the WAL into the freshly-built `MvStore`, and seed the
// `MvccClock` past the highest persisted timestamp. Without
// this step the in-memory MVCC state would always start blank
// on reopen — fine for legacy single-session workloads, but a
// correctness gap once `BEGIN CONCURRENT` is in play (a
// second process could hand out a `begin_ts` below an
// already-committed version's `end`, breaking the visibility
// rule).
//
// The clock seed is the larger of (header.clock_high_water,
// max(commit_ts among replayed batches)) so a crash between
// commits and the next checkpoint — where the header's
// high-water lags reality — still produces a clock that
// doesn't regress.
replay_mvcc_into_db(&mut db, &pager)?;
db.source_path = Some(path.to_path_buf());
db.pager = Some(pager);
Ok(db)
}
/// Phase 11.9 — drains every MVCC commit batch the Pager recovered
/// from the WAL into `db.mv_store`, and advances `db.mvcc_clock`
/// to at least the highest observed timestamp.
///
/// Batches are replayed in WAL order, which matches commit order
/// (the WAL appends sequentially). Each record's `commit_ts`
/// becomes the version's `begin`, with the previous latest
/// version's `end` capped at the same timestamp — identical to
/// the live-commit path's `MvStore::push_committed`.
fn replay_mvcc_into_db(db: &mut Database, pager: &Pager) -> Result<()> {
use crate::mvcc::RowVersion;
let mut clock_seed = pager.clock_high_water();
for batch in pager.recovered_mvcc_commits() {
if batch.commit_ts > clock_seed {
clock_seed = batch.commit_ts;
}
for rec in &batch.records {
let version = RowVersion::committed(batch.commit_ts, rec.payload.clone());
db.mv_store
.push_committed(rec.row.clone(), version)
.map_err(|e| {
SQLRiteError::Internal(format!(
"WAL MVCC replay: push_committed failed for {}/{}: {e}",
rec.row.table, rec.row.rowid,
))
})?;
}
}
if clock_seed > 0 {
db.mvcc_clock.observe(clock_seed);
}
Ok(())
}
/// Catalog row for a secondary index — deferred until after every table is
/// loaded so the index's base table exists by the time we populate it.
struct IndexCatalogRow {
name: String,
sql: String,
rootpage: u32,
}
/// Persists `db` to disk. Diff-pager skips writing pages whose bytes
/// haven't changed; the [`PageAllocator`] preserves per-table page
/// numbers across saves so unchanged tables produce zero dirty frames.
///
/// Pages that were live before this save but aren't restaged this round
/// (e.g., the leaves of a dropped table) move onto a persisted free
/// list rooted at `header.freelist_head`; subsequent saves draw from
/// the freelist before extending the file. `VACUUM` (see
/// [`vacuum_database`]) compacts the file by ignoring the freelist and
/// allocating linearly from page 1.
///
/// [`PageAllocator`]: crate::sql::pager::allocator::PageAllocator
pub fn save_database(db: &mut Database, path: &Path) -> Result<()> {
save_database_with_mode(db, path, /*compact=*/ false)
}
/// Reclaims space by rewriting every live B-Tree contiguously from
/// page 1, with no freelist. Equivalent to `save_database` but ignores
/// the existing freelist and per-table preferred pools — every page is
/// allocated by extending the high-water mark — so the resulting file
/// is tightly packed and the freelist is empty.
///
/// Used by the SQL-level `VACUUM;` statement.
pub fn vacuum_database(db: &mut Database, path: &Path) -> Result<()> {
save_database_with_mode(db, path, /*compact=*/ true)
}
/// Shared save core. `compact = false` is the normal save path (uses
/// the existing freelist + per-table preferred pools). `compact = true`
/// is the VACUUM path (empty freelist, empty preferred pools, linear
/// allocation from page 1).
fn save_database_with_mode(db: &mut Database, path: &Path, compact: bool) -> Result<()> {
// Phase 7d.3 — rebuild any HNSW index that DELETE / UPDATE-on-vector
// marked dirty. Done up front under the &mut Database borrow we
// already hold, before the immutable iteration loops below need
// their own borrow.
rebuild_dirty_hnsw_indexes(db)?;
// Phase 8b — same drill for FTS indexes flagged by DELETE / UPDATE.
rebuild_dirty_fts_indexes(db);
let same_path = db.source_path.as_deref() == Some(path);
let mut pager = if same_path {
match db.pager.take() {
Some(p) => p,
None if path.exists() => Pager::open(path)?,
None => Pager::create(path)?,
}
} else if path.exists() {
Pager::open(path)?
} else {
Pager::create(path)?
};
// Snapshot what was live BEFORE we reset staged. Used to compute the
// newly-freed set after staging completes. Page 0 (the header) is
// never on the freelist — it's always live.
let old_header = pager.header();
let old_live: std::collections::HashSet<u32> = (1..old_header.page_count).collect();
// Read the previously-persisted freelist so its leaf pages can be
// reused as preferred allocations and its trunk pages don't leak.
let (old_free_leaves, old_free_trunks) = if compact || old_header.freelist_head == 0 {
(Vec::new(), Vec::new())
} else {
crate::sql::pager::freelist::read_freelist(&pager, old_header.freelist_head)?
};
// Snapshot the previous rootpages of each table/index so we can
// seed per-table preferred pools (the unchanged-table case stages
// byte-identical pages → diff pager skips every write for it).
let old_rootpages = if compact {
HashMap::new()
} else {
read_old_rootpages(&pager, old_header.schema_root_page)?
};
// SQLR-1 — snapshot every prior B-Tree's page set NOW, before any
// staging starts. `Pager::read_page` shadows on-disk bytes with the
// current `staged` buffer, so if we deferred these walks until each
// object's turn in the staging loop, a *new* index added in this
// save would extend past the old high-water and overwrite the
// pages of any later-staged object whose old root sits in that
// range — including `sqlrite_master`, which is always staged last.
// The follow-up walk would then read the wrong B-Tree's bytes and
// either hand the allocator a bogus preferred pool or panic
// dispatching cells (a table-cell decoder vs. an index leaf, the
// shape of the original SQLR-1 panic). Walking up front pins each
// map to the committed bytes that were on disk before this save
// touched anything.
let old_preferred_pages: HashMap<(String, String), Vec<u32>> = if compact {
HashMap::new()
} else {
let mut map: HashMap<(String, String), Vec<u32>> = HashMap::new();
for ((kind, name), &root) in &old_rootpages {
// Tables can carry overflow chains; index/HNSW/FTS leaves
// never overflow in the current encoding, so the cheaper
// walk suffices for them.
let follow = kind == "table";
let pages = collect_pages_for_btree(&pager, root, follow)?;
map.insert((kind.clone(), name.clone()), pages);
}
map
};
let old_master_pages: Vec<u32> = if compact || old_header.schema_root_page == 0 {
Vec::new()
} else {
collect_pages_for_btree(
&pager,
old_header.schema_root_page,
/*follow_overflow=*/ true,
)?
};
pager.clear_staged();
// Allocator: in normal mode, seed with the old freelist; in compact
// mode, start empty so allocation extends linearly from page 1.
use std::collections::VecDeque;
let initial_freelist: VecDeque<u32> = if compact {
VecDeque::new()
} else {
crate::sql::pager::freelist::freelist_to_deque(old_free_leaves.clone())
};
let mut alloc = crate::sql::pager::allocator::PageAllocator::new(initial_freelist, 1);
// 1. Stage each user table's B-Tree, collecting master-row info.
// `kind` is "table" or "index" — master has one row per each.
let mut master_rows: Vec<CatalogEntry> = Vec::new();
let mut table_names: Vec<&String> = db.tables.keys().collect();
table_names.sort();
for name in table_names {
if name == MASTER_TABLE_NAME {
return Err(SQLRiteError::Internal(format!(
"user table cannot be named '{MASTER_TABLE_NAME}' (reserved)"
)));
}
if !compact {
if let Some(prev) = old_preferred_pages.get(&("table".to_string(), name.to_string())) {
alloc.set_preferred(prev.clone());
}
}
let table = &db.tables[name];
let rootpage = stage_table_btree(&mut pager, table, &mut alloc)?;
alloc.finish_preferred();
master_rows.push(CatalogEntry {
kind: "table".into(),
name: name.clone(),
sql: table_to_create_sql(table),
rootpage,
last_rowid: table.last_rowid,
});
}
// 2. Stage each secondary index's B-Tree. Indexes persist in a
// deterministic order: sorted by (owning_table, index_name).
let mut index_entries: Vec<(&Table, &SecondaryIndex)> = Vec::new();
for table in db.tables.values() {
for idx in &table.secondary_indexes {
index_entries.push((table, idx));
}
}
index_entries
.sort_by(|(ta, ia), (tb, ib)| ta.tb_name.cmp(&tb.tb_name).then(ia.name.cmp(&ib.name)));
for (_table, idx) in index_entries {
if !compact {
if let Some(prev) =
old_preferred_pages.get(&("index".to_string(), idx.name.to_string()))
{
alloc.set_preferred(prev.clone());
}
}
let rootpage = stage_index_btree(&mut pager, idx, &mut alloc)?;
alloc.finish_preferred();
master_rows.push(CatalogEntry {
kind: "index".into(),
name: idx.name.clone(),
sql: idx.synthesized_sql(),
rootpage,
last_rowid: 0,
});
}
// 2b. Phase 7d.3: persist HNSW indexes as their own cell-encoded
// page trees, with the rootpage recorded in sqlrite_master.
// Reopen loads the graph back from cells (fast, exact match)
// instead of rebuilding from rows.
//
// Dirty indexes (set by DELETE / UPDATE-on-vector-col) are
// rebuilt from current rows BEFORE staging, so the on-disk
// graph reflects the current row set.
let mut hnsw_entries: Vec<(&Table, &crate::sql::db::table::HnswIndexEntry)> = Vec::new();
for table in db.tables.values() {
for entry in &table.hnsw_indexes {
hnsw_entries.push((table, entry));
}
}
hnsw_entries
.sort_by(|(ta, ea), (tb, eb)| ta.tb_name.cmp(&tb.tb_name).then(ea.name.cmp(&eb.name)));
for (table, entry) in hnsw_entries {
if !compact {
if let Some(prev) =
old_preferred_pages.get(&("index".to_string(), entry.name.to_string()))
{
alloc.set_preferred(prev.clone());
}
}
let rootpage = stage_hnsw_btree(&mut pager, &entry.index, &mut alloc)?;
alloc.finish_preferred();
master_rows.push(CatalogEntry {
kind: "index".into(),
name: entry.name.clone(),
sql: synthesize_hnsw_create_index_sql(
&entry.name,
&table.tb_name,
&entry.column_name,
entry.metric,
),
rootpage,
last_rowid: 0,
});
}
// 2c. Phase 8c — persist FTS posting lists as their own
// cell-encoded page trees, with the rootpage recorded in
// sqlrite_master. Reopen loads the postings back from cells
// (fast, exact match) instead of re-tokenizing rows.
//
// Dirty indexes (set by DELETE / UPDATE-on-text-col) are
// rebuilt from current rows BEFORE staging by
// `rebuild_dirty_fts_indexes`, so the on-disk tree reflects
// the current row set.
let mut fts_entries: Vec<(&Table, &crate::sql::db::table::FtsIndexEntry)> = Vec::new();
for table in db.tables.values() {
for entry in &table.fts_indexes {
fts_entries.push((table, entry));
}
}
fts_entries
.sort_by(|(ta, ea), (tb, eb)| ta.tb_name.cmp(&tb.tb_name).then(ea.name.cmp(&eb.name)));
let any_fts = !fts_entries.is_empty();
for (table, entry) in fts_entries {
if !compact {
if let Some(prev) =
old_preferred_pages.get(&("index".to_string(), entry.name.to_string()))
{
alloc.set_preferred(prev.clone());
}
}
let rootpage = stage_fts_btree(&mut pager, &entry.index, &mut alloc)?;
alloc.finish_preferred();
master_rows.push(CatalogEntry {
kind: "index".into(),
name: entry.name.clone(),
sql: format!(
"CREATE INDEX {} ON {} USING fts ({})",
entry.name, table.tb_name, entry.column_name
),
rootpage,
last_rowid: 0,
});
}
// 3. Build an in-memory sqlrite_master with one row per table or index,
// then stage it via the same tree-build path. Seed master's
// preferred pool with the previous master tree's pages so the
// catalog page numbers stay stable across saves whenever the
// catalog content didn't change.
let mut master = build_empty_master_table();
for (i, entry) in master_rows.into_iter().enumerate() {
let rowid = (i as i64) + 1;
master.restore_row(
rowid,
vec![
Some(Value::Text(entry.kind)),
Some(Value::Text(entry.name)),
Some(Value::Text(entry.sql)),
Some(Value::Integer(entry.rootpage as i64)),
Some(Value::Integer(entry.last_rowid)),
],
)?;
}
if !compact && !old_master_pages.is_empty() {
// Use the page list snapshotted before any staging touched
// disk; re-walking here would read whatever a new index
// already restaged on top of master's old root (SQLR-1).
alloc.set_preferred(old_master_pages.clone());
}
let master_root = stage_table_btree(&mut pager, &master, &mut alloc)?;
alloc.finish_preferred();
// 4. Compute newly-freed pages: the previously-live set minus what
// we just restaged. The previous freelist's trunk pages get
// re-encoded too — they're in `old_live`, weren't restaged, so
// the filter naturally moves them to the new freelist.
//
// In `compact` mode (VACUUM), we *discard* newly_freed instead of
// routing it onto the new freelist. The whole point of VACUUM is
// to let the file truncate to the new high-water mark, so any page
// past it gets dropped at the next checkpoint.
if !compact {
let used = alloc.used().clone();
let mut newly_freed: Vec<u32> = old_live
.iter()
.copied()
.filter(|p| !used.contains(p))
.collect();
let _ = &old_free_trunks; // silenced — handled by the old_live filter
alloc.add_to_freelist(newly_freed.drain(..));
}
// 5. Encode the new freelist into trunk pages. `stage_freelist`
// consumes some of the free pages AS the trunk pages themselves —
// a trunk is just a free page borrowed for metadata. Pages that
// were on the freelist but become trunks no longer need to be
// "extension" pages; the high-water mark from the staging loop
// above is already correct.
let new_free_pages = alloc.drain_freelist();
let new_freelist_head =
crate::sql::pager::freelist::stage_freelist(&mut pager, new_free_pages)?;
// 6. Pick the format version. v6 is on demand: only bumps when the
// new freelist is non-empty. FTS-bearing files keep their v5
// promotion; v6 is a strict superset (v6 readers handle v4/v5/v6).
use crate::sql::pager::header::{FORMAT_VERSION_V5, FORMAT_VERSION_V6};
let format_version = if new_freelist_head != 0 {
FORMAT_VERSION_V6
} else if any_fts {
// Preserve a v6 file at v6 (don't downgrade) but otherwise
// bump v4 → v5 for FTS like Phase 8c does.
std::cmp::max(FORMAT_VERSION_V5, old_header.format_version)
} else {
// Preserve whatever the file already was.
old_header.format_version
};
pager.commit(DbHeader {
page_count: alloc.high_water(),
schema_root_page: master_root,
format_version,
freelist_head: new_freelist_head,
})?;
if same_path {
db.pager = Some(pager);
}
Ok(())
}
/// Build material for a single row in sqlrite_master.
struct CatalogEntry {
kind: String, // "table" or "index"
name: String,
sql: String,
rootpage: u32,
last_rowid: i64,
}
// -------------------------------------------------------------------------
// sqlrite_master — hardcoded catalog table schema
fn build_empty_master_table() -> Table {
// Phase 3e: `type` is the first column, matching SQLite's convention.
// It distinguishes `'table'` rows from `'index'` rows.
let columns = vec![
Column::new("type".into(), "text".into(), false, true, false),
Column::new("name".into(), "text".into(), true, true, true),
Column::new("sql".into(), "text".into(), false, true, false),
Column::new("rootpage".into(), "integer".into(), false, true, false),
Column::new("last_rowid".into(), "integer".into(), false, true, false),
];
build_empty_table(MASTER_TABLE_NAME, columns, 0)
}
/// SQLR-10 — builds an in-memory, read-only snapshot of `sqlrite_master`
/// from the live `Database`, so `SELECT … FROM sqlrite_master` can
/// introspect the catalog without a save round-trip. One row per user
/// table and per index (B-Tree / HNSW / FTS), reusing the exact same SQL
/// synthesis the persistence path uses — so the `name` / `type` / `sql`
/// columns match byte-for-byte what a saved-then-dumped catalog shows.
///
/// Rows are ordered deterministically: tables sorted by name, then
/// indexes sorted by `(owning_table, index_name)` — mirroring
/// [`save_database_with_mode`]'s staging order.
///
/// The `rootpage` column is `0` for every row: page assignment only
/// happens at save time, so a live/in-memory view has no meaningful page
/// number. The column is kept for schema parity with the persisted
/// catalog (and SQLite's `sqlite_master`); callers needing introspection
/// use `type` / `name` / `sql`. `last_rowid` carries the table's current
/// auto-increment high-water mark (0 for index rows).
pub(crate) fn build_master_table_snapshot(db: &Database) -> Result<Table> {
let mut master = build_empty_master_table();
let mut entries: Vec<CatalogEntry> = Vec::new();
// Tables, sorted by name.
let mut table_names: Vec<&String> = db.tables.keys().collect();
table_names.sort();
for name in &table_names {
let table = &db.tables[*name];
entries.push(CatalogEntry {
kind: "table".into(),
name: (*name).clone(),
sql: table_to_create_sql(table),
rootpage: 0,
last_rowid: table.last_rowid,
});
}
// Indexes across all three families, sorted by (table, index name) —
// collected into one list so the final order matches how a reopened
// catalog enumerates them.
let mut index_entries: Vec<(String, String, String)> = Vec::new(); // (table, index_name, sql)
for table in db.tables.values() {
for idx in &table.secondary_indexes {
index_entries.push((
table.tb_name.clone(),
idx.name.clone(),
idx.synthesized_sql(),
));
}
for entry in &table.hnsw_indexes {
index_entries.push((
table.tb_name.clone(),
entry.name.clone(),
synthesize_hnsw_create_index_sql(
&entry.name,
&table.tb_name,
&entry.column_name,
entry.metric,
),
));
}
for entry in &table.fts_indexes {
index_entries.push((
table.tb_name.clone(),
entry.name.clone(),
format!(
"CREATE INDEX {} ON {} USING fts ({})",
entry.name, table.tb_name, entry.column_name
),
));
}
}
index_entries.sort_by(|(ta, ia, _), (tb, ib, _)| ta.cmp(tb).then(ia.cmp(ib)));
for (_table, name, sql) in index_entries {
entries.push(CatalogEntry {
kind: "index".into(),
name,
sql,
rootpage: 0,
last_rowid: 0,
});
}
for (i, entry) in entries.into_iter().enumerate() {
let rowid = (i as i64) + 1;
master.restore_row(
rowid,
vec![
Some(Value::Text(entry.kind)),
Some(Value::Text(entry.name)),
Some(Value::Text(entry.sql)),
Some(Value::Integer(entry.rootpage as i64)),
Some(Value::Integer(entry.last_rowid)),
],
)?;
}
Ok(master)
}
/// Reads a required Text column from a known-good catalog row.
fn take_text(table: &Table, col: &str, rowid: i64) -> Result<String> {
match table.get_value(col, rowid) {
Some(Value::Text(s)) => Ok(s),
other => Err(SQLRiteError::Internal(format!(
"sqlrite_master column '{col}' at rowid {rowid}: expected Text, got {other:?}"
))),
}
}
/// Reads a required Integer column from a known-good catalog row.
fn take_integer(table: &Table, col: &str, rowid: i64) -> Result<i64> {
match table.get_value(col, rowid) {
Some(Value::Integer(v)) => Ok(v),
other => Err(SQLRiteError::Internal(format!(
"sqlrite_master column '{col}' at rowid {rowid}: expected Integer, got {other:?}"
))),
}
}
// -------------------------------------------------------------------------
// CREATE-TABLE SQL synthesis and re-parsing
/// Synthesizes a CREATE TABLE SQL string that recreates the table's schema.
/// Deterministic: same schema → same SQL, so diffing commits stay stable.
fn table_to_create_sql(table: &Table) -> String {
let mut parts = Vec::with_capacity(table.columns.len());
for c in &table.columns {
// Render the SQL type literally so the round-trip through
// CREATE TABLE re-parsing recreates the same schema. Vector
// carries its dimension inline.
let ty: String = match &c.datatype {
DataType::Integer => "INTEGER".to_string(),
DataType::Text => "TEXT".to_string(),
DataType::Real => "REAL".to_string(),
DataType::Bool => "BOOLEAN".to_string(),
DataType::Vector(dim) => format!("VECTOR({dim})"),
DataType::Json => "JSON".to_string(),
DataType::None | DataType::Invalid => "TEXT".to_string(),
};
let mut piece = format!("{} {}", c.column_name, ty);
if c.is_pk {
piece.push_str(" PRIMARY KEY");
} else {
if c.is_unique {
piece.push_str(" UNIQUE");
}
if c.not_null {
piece.push_str(" NOT NULL");
}
}
if let Some(default) = &c.default {
piece.push_str(" DEFAULT ");
piece.push_str(&render_default_literal(default));
}
parts.push(piece);
}
format!("CREATE TABLE {} ({});", table.tb_name, parts.join(", "))
}
/// Renders a DEFAULT value back to SQL-literal form so the synthesized
/// CREATE TABLE round-trips through `parse_create_sql`. Text values get
/// single-quoted with single-quote doubling for escaping. Vector defaults
/// are not currently expressible at CREATE TABLE time, so we render them
/// as their bracket-array form (matches the INSERT literal grammar).
fn render_default_literal(value: &Value) -> String {
match value {
Value::Integer(i) => i.to_string(),
Value::Real(f) => f.to_string(),
Value::Bool(b) => {
if *b {
"TRUE".to_string()
} else {
"FALSE".to_string()
}
}
Value::Text(s) => format!("'{}'", s.replace('\'', "''")),
Value::Null => "NULL".to_string(),
Value::Vector(_) => value.to_display_string(),
}
}
/// Reverses `table_to_create_sql`: feeds the SQL back through `sqlparser`
/// and produces our internal column list. Returns `(table_name, columns)`.
fn parse_create_sql(sql: &str) -> Result<(String, Vec<Column>)> {
let dialect = SqlriteDialect::new();
let mut ast = Parser::parse_sql(&dialect, sql).map_err(SQLRiteError::from)?;
let stmt = ast.pop().ok_or_else(|| {
SQLRiteError::Internal("sqlrite_master row held an empty SQL string".to_string())
})?;
let create = CreateQuery::new(&stmt)?;
let columns = create
.columns
.into_iter()
.map(|pc| {
Column::with_default(
pc.name,
pc.datatype,
pc.is_pk,
pc.not_null,
pc.is_unique,
pc.default,
)
})
.collect();
Ok((create.table_name, columns))
}
// -------------------------------------------------------------------------
// In-memory table (re)construction
/// Builds an empty in-memory `Table` given the declared columns.
fn build_empty_table(name: &str, columns: Vec<Column>, last_rowid: i64) -> Table {
let rows: Arc<Mutex<HashMap<String, Row>>> = Arc::new(Mutex::new(HashMap::new()));
let mut secondary_indexes: Vec<SecondaryIndex> = Vec::new();
{
let mut map = rows.lock().expect("rows mutex poisoned");
for col in &columns {
// Mirror the dispatch in `Table::new` so the reconstructed
// table has the same shape it'd have if it were built fresh
// from SQL. Phase 7a adds the Vector arm — without it,
// VECTOR columns silently restore as Row::None and every
// restore_row hits a "storage None vs value Some(Vector(...))"
// type mismatch.
let row = match &col.datatype {
DataType::Integer => Row::Integer(BTreeMap::new()),
DataType::Text => Row::Text(BTreeMap::new()),
DataType::Real => Row::Real(BTreeMap::new()),
DataType::Bool => Row::Bool(BTreeMap::new()),
DataType::Vector(_dim) => Row::Vector(BTreeMap::new()),
// JSON columns reuse Text storage — see Table::new and
// Phase 7e's scope-correction note.
DataType::Json => Row::Text(BTreeMap::new()),
DataType::None | DataType::Invalid => Row::None,
};
map.insert(col.column_name.clone(), row);
// Auto-create UNIQUE/PK indexes so the restored table has the
// same shape Table::new would have built from fresh SQL.
if (col.is_pk || col.is_unique)
&& matches!(col.datatype, DataType::Integer | DataType::Text)
{
if let Ok(idx) = SecondaryIndex::new(
SecondaryIndex::auto_name(name, &col.column_name),
name.to_string(),
col.column_name.clone(),
&col.datatype,
true,
IndexOrigin::Auto,
) {
secondary_indexes.push(idx);
}
}
}
}
let primary_key = columns
.iter()
.find(|c| c.is_pk)
.map(|c| c.column_name.clone())
.unwrap_or_else(|| "-1".to_string());
Table {
tb_name: name.to_string(),
columns,
rows,
secondary_indexes,
// HNSW indexes (Phase 7d.2) are reconstructed on open by re-
// executing each `CREATE INDEX … USING hnsw` SQL stored in
// `sqlrite_master`. This builder produces the empty shell;
// `replay_create_index_for_hnsw` (in this same module) walks
// sqlrite_master after every table is loaded and rebuilds the
// graph from current row data. Persistence of the graph itself
// (avoiding the on-open rebuild cost) is Phase 7d.3.
hnsw_indexes: Vec::new(),
// FTS indexes (Phase 8b) follow the same pattern — the
// CREATE INDEX … USING fts SQL is the source of truth on open
// and the in-memory posting list gets rebuilt from current
// rows. Cell-encoded persistence of the postings is Phase 8c.
fts_indexes: Vec::new(),
last_rowid,
primary_key,
}
}
// -------------------------------------------------------------------------
// Leaf-chain read / write
/// Walks a table's B-Tree from `root_page`, following the leftmost-child
/// chain down to the first leaf, then iterating leaves via their sibling
/// `next_page` pointers. Every cell is decoded and replayed into `table`.
///
/// Open-path note: we eagerly materialize the entire table into `Table`'s
/// in-memory maps. Phase 5 will introduce a `Cursor` that hits the pager
/// on demand so queries can stream through the tree without a full upfront
/// load.
/// Re-parses `CREATE INDEX` SQL from sqlrite_master and restores the
/// index on its base table by walking the tree of index cells at
/// `rootpage`. The base table is expected to already be in `db.tables`.
fn attach_index(db: &mut Database, pager: &Pager, row: IndexCatalogRow) -> Result<()> {
let (table_name, column_name, is_unique) = parse_create_index_sql(&row.sql)?;
let table = db.get_table_mut(table_name.clone()).map_err(|_| {
SQLRiteError::Internal(format!(
"index '{}' references unknown table '{table_name}' (sqlrite_master out of sync?)",
row.name
))
})?;
let datatype = table
.columns
.iter()
.find(|c| c.column_name == column_name)
.map(|c| clone_datatype(&c.datatype))
.ok_or_else(|| {
SQLRiteError::Internal(format!(
"index '{}' references unknown column '{column_name}' on '{table_name}'",
row.name
))
})?;
// An auto-index on this column may already exist (built by
// build_empty_table for UNIQUE/PK columns). If the names match, reuse
// the slot instead of adding a duplicate entry.
let existing_slot = table
.secondary_indexes
.iter()
.position(|i| i.name == row.name);
let idx = match existing_slot {
Some(i) => {
// Drain any entries that may have been populated during table
// restore_row calls — we're about to repopulate from the
// persisted tree.
table.secondary_indexes.remove(i)
}
None => SecondaryIndex::new(
row.name.clone(),
table_name.clone(),
column_name.clone(),
&datatype,
is_unique,
IndexOrigin::Explicit,
)?,
};
let mut idx = idx;
// Wipe any stale entries from the auto path so the load is idempotent.
let is_unique_flag = idx.is_unique;
let origin = idx.origin;
idx = SecondaryIndex::new(
idx.name,
idx.table_name,
idx.column_name,
&datatype,
is_unique_flag,
origin,
)?;
// Populate from the index tree's cells.
load_index_rows(pager, &mut idx, row.rootpage)?;
table.secondary_indexes.push(idx);
Ok(())
}
/// Walks the leaves of an index B-Tree rooted at `root_page` and inserts
/// every `(value, rowid)` pair into `idx`.
fn load_index_rows(pager: &Pager, idx: &mut SecondaryIndex, root_page: u32) -> Result<()> {
if root_page == 0 {
return Ok(());
}
let first_leaf = find_leftmost_leaf(pager, root_page)?;
let mut current = first_leaf;
while current != 0 {
let page_buf = pager
.read_page(current)
.ok_or_else(|| SQLRiteError::Internal(format!("missing index leaf page {current}")))?;
if page_buf[0] != PageType::TableLeaf as u8 {
return Err(SQLRiteError::Internal(format!(
"page {current} tagged {} but expected TableLeaf (index)",
page_buf[0]
)));
}