1use crate::row::encode_row_into;
2use crate::table::Table;
3use crate::types::*;
4use crate::wal::{Wal, WalDurabilityTicket, WalRecord, WalRecordType, WalSyncMode};
5use rustc_hash::FxHashMap;
6use std::fs;
7use std::io::{self, Read, Write};
8use std::path::{Path, PathBuf};
9use tracing::{info, warn};
10
11fn check_encoded_row_size(encoded: &[u8]) -> io::Result<()> {
16 if encoded.len() > crate::page::MAX_ROW_DATA_SIZE {
17 return Err(crate::error::StorageError::RowTooLarge {
18 size: encoded.len(),
19 max: crate::page::MAX_ROW_DATA_SIZE,
20 }
21 .into());
22 }
23 Ok(())
24}
25
26fn validate_identifier(kind: &str, name: &str) -> io::Result<()> {
30 if name.is_empty() {
31 return Err(io::Error::new(
32 io::ErrorKind::InvalidInput,
33 format!("invalid {kind} name: must not be empty"),
34 ));
35 }
36 let mut chars = name.chars();
37 let first = chars.next().expect("non-empty name");
39 if !first.is_ascii_alphabetic() && first != '_' {
40 return Err(io::Error::new(
41 io::ErrorKind::InvalidInput,
42 format!("invalid {kind} name '{name}': must start with a letter or underscore"),
43 ));
44 }
45 for ch in chars {
46 if !ch.is_ascii_alphanumeric() && ch != '_' {
47 return Err(io::Error::new(
48 io::ErrorKind::InvalidInput,
49 format!(
50 "invalid {kind} name '{name}': must contain only letters, digits, and underscores"
51 ),
52 ));
53 }
54 }
55 Ok(())
56}
57
58fn validate_table_name(name: &str) -> io::Result<()> {
60 validate_identifier("table", name)
61}
62
63fn validate_column_name(name: &str) -> io::Result<()> {
65 validate_identifier("column", name)
66}
67
68const CATALOG_FILE: &str = "catalog.bin";
77pub const CATALOG_LSN_FILE: &str = "catalog.lsn";
78const CATALOG_MAGIC: &[u8; 4] = b"BCAT";
79pub const CATALOG_VERSION: u16 = 5;
83
84const WAL_FILE: &str = "wal.log";
87const SYNC_STATE_DIR: &str = ".powdb-sync";
88const SYNC_IDENTITY_FILE: &str = "identity.json";
89
90const WAL_BATCH_SIZE: usize = 64;
94type WalArchiveCallback<'a> = &'a mut dyn FnMut(&Path, &[WalRecord]) -> io::Result<()>;
95
96fn read_durable_lsn(data_dir: &Path) -> io::Result<u64> {
97 let path = data_dir.join(CATALOG_LSN_FILE);
98 let bytes = match fs::read(path) {
99 Ok(bytes) => bytes,
100 Err(err) if err.kind() == io::ErrorKind::NotFound => return Ok(0),
101 Err(err) => return Err(err),
102 };
103 if bytes.len() != 8 {
104 return Err(io::Error::new(
105 io::ErrorKind::InvalidData,
106 "catalog LSN sidecar has invalid length",
107 ));
108 }
109 let mut buf = [0u8; 8];
110 buf.copy_from_slice(&bytes);
111 Ok(u64::from_le_bytes(buf))
112}
113
114fn write_durable_lsn(data_dir: &Path, lsn: u64) -> io::Result<()> {
115 let path = data_dir.join(CATALOG_LSN_FILE);
116 let tmp_path = data_dir.join(format!("{CATALOG_LSN_FILE}.tmp"));
117 let mut file = fs::File::create(&tmp_path)?;
118 file.write_all(&lsn.to_le_bytes())?;
119 file.sync_all()?;
120 drop(file);
121 fs::rename(&tmp_path, &path)?;
122 sync_directory(data_dir)?;
123 Ok(())
124}
125
126#[cfg(unix)]
127fn sync_directory(path: &Path) -> io::Result<()> {
128 fs::File::open(path)?.sync_all()
129}
130
131#[cfg(not(unix))]
132fn sync_directory(path: &Path) -> io::Result<()> {
133 let _ = path;
134 Ok(())
135}
136
137fn max_record_lsn(records: &[WalRecord]) -> Option<u64> {
138 records.iter().map(|record| record.lsn).max()
139}
140
141pub struct Catalog {
155 tables: Vec<Table>,
158 name_to_slot: FxHashMap<String, usize>,
161 data_dir: PathBuf,
162 wal: Wal,
167 next_tx_id: u64,
171 active_tx_id: Option<u64>,
174 tx_start_len: Option<u64>,
177 pending_autocommit_tx_ids: Vec<u64>,
180 checkpointed: bool,
184 durable_lsn: u64,
187}
188
189impl Catalog {
190 pub fn create(data_dir: &Path) -> io::Result<Self> {
211 crate::create_data_dir_secure(data_dir)?;
212 let wal_path = data_dir.join(WAL_FILE);
213 let wal = Wal::create(&wal_path, WAL_BATCH_SIZE)?;
214 let cat = Catalog {
215 tables: Vec::new(),
216 name_to_slot: FxHashMap::default(),
217 data_dir: data_dir.to_path_buf(),
218 wal,
219 next_tx_id: 1,
220 active_tx_id: None,
221 tx_start_len: None,
222 pending_autocommit_tx_ids: Vec::new(),
223 checkpointed: false,
224 durable_lsn: 0,
225 };
226 cat.persist()?;
227 Ok(cat)
228 }
229
230 pub fn open(data_dir: &Path) -> io::Result<Self> {
239 Self::open_inner(data_dir, None)
240 }
241
242 pub fn open_with_wal_archive<F>(data_dir: &Path, mut archive: F) -> io::Result<Self>
251 where
252 F: FnMut(&Path, &[WalRecord]) -> io::Result<()>,
253 {
254 let archive: WalArchiveCallback<'_> = &mut archive;
255 Self::open_inner(data_dir, Some(archive))
256 }
257
258 fn open_inner(data_dir: &Path, archive: Option<WalArchiveCallback<'_>>) -> io::Result<Self> {
259 let cat_path = data_dir.join(CATALOG_FILE);
260 if !cat_path.exists() {
261 return Err(io::Error::new(io::ErrorKind::NotFound, "no catalog file"));
262 }
263 let entries = read_catalog_file(&cat_path)?;
264 let durable_lsn = read_durable_lsn(data_dir)?;
265 let mut tables: Vec<Table> = Vec::with_capacity(entries.len());
266 let mut name_to_slot =
267 FxHashMap::with_capacity_and_hasher(entries.len(), Default::default());
268 for CatalogEntry {
269 schema,
270 indexed_cols,
271 defaults,
272 auto_cols,
273 } in entries
274 {
275 let name = schema.table_name.clone();
276 let mut table = Table::open_with_indexes(schema, data_dir, &indexed_cols)?;
282 table.set_defaults(defaults);
283 table.set_auto_cols(auto_cols);
284 name_to_slot.insert(name, tables.len());
285 tables.push(table);
286 }
287 let wal_path = data_dir.join(WAL_FILE);
288 let wal = Wal::open(&wal_path, WAL_BATCH_SIZE)?;
289 let mut cat = Catalog {
290 tables,
291 name_to_slot,
292 data_dir: data_dir.to_path_buf(),
293 wal,
294 next_tx_id: 1,
295 active_tx_id: None,
296 tx_start_len: None,
297 pending_autocommit_tx_ids: Vec::new(),
298 checkpointed: false,
299 durable_lsn,
300 };
301 cat.replay_wal(archive)?;
302 let max_page_lsn = cat
312 .tables
313 .iter()
314 .map(|t| t.heap.max_page_lsn())
315 .max()
316 .unwrap_or(0);
317 let max_known_lsn = max_page_lsn.max(cat.durable_lsn);
318 cat.wal.set_next_lsn_at_least(max_known_lsn + 1);
319 Ok(cat)
320 }
321
322 fn replay_wal(&mut self, mut archive: Option<WalArchiveCallback<'_>>) -> io::Result<()> {
348 let records = self.wal.read_all()?;
349 if records.is_empty() {
350 return Ok(());
351 }
352 if archive.is_none() {
353 self.ensure_plain_wal_truncate_allowed(&records)?;
354 }
355 self.replay_records(&records)?;
356 if let Some(archive) = archive.as_mut() {
357 archive(&self.data_dir, &records)?;
358 }
359 self.wal.truncate()?;
360 Ok(())
361 }
362
363 pub fn apply_wal_records(&mut self, records: &[WalRecord]) -> io::Result<()> {
372 self.ensure_no_active_transaction_for_checkpoint()?;
373 self.ensure_no_pending_wal_records()?;
374 self.replay_records(records)
375 }
376
377 pub fn ensure_no_pending_wal_records(&self) -> io::Result<()> {
380 if self.wal.has_pending() || !self.wal.read_all()?.is_empty() {
381 return Err(io::Error::other(
382 "cannot apply replicated WAL records while local WAL records are pending",
383 ));
384 }
385 Ok(())
386 }
387
388 fn replay_records(&mut self, records: &[WalRecord]) -> io::Result<()> {
389 if records.is_empty() {
390 return Ok(());
391 }
392
393 info!(count = records.len(), "applying WAL records");
394
395 let has_boundaries = records.iter().any(|rec| {
404 matches!(
405 rec.record_type,
406 WalRecordType::Begin | WalRecordType::Commit | WalRecordType::Rollback
407 )
408 });
409 let mut committed_row_records = vec![true; records.len()];
410 if has_boundaries {
411 committed_row_records.fill(false);
412 let mut pending_tx_spans: Vec<(u64, Vec<usize>)> = Vec::new();
413 for (index, rec) in records.iter().enumerate() {
414 match rec.record_type {
415 WalRecordType::Insert | WalRecordType::Update | WalRecordType::Delete
416 if rec.tx_id == 0 =>
417 {
418 committed_row_records[index] = true;
419 }
420 WalRecordType::Insert | WalRecordType::Update | WalRecordType::Delete => {
421 if let Some((_, rows)) = pending_tx_spans
422 .iter_mut()
423 .rev()
424 .find(|(tx_id, _)| *tx_id == rec.tx_id)
425 {
426 rows.push(index);
427 } else {
428 pending_tx_spans.push((rec.tx_id, vec![index]));
429 }
430 }
431 WalRecordType::Begin if rec.tx_id != 0 => {
432 pending_tx_spans.push((rec.tx_id, Vec::new()));
433 }
434 WalRecordType::Commit if rec.tx_id != 0 => {
435 if let Some(span_index) = pending_tx_spans
436 .iter()
437 .rposition(|(tx_id, _)| *tx_id == rec.tx_id)
438 {
439 let (_, rows) = pending_tx_spans.remove(span_index);
440 for row_index in rows {
441 committed_row_records[row_index] = true;
442 }
443 }
444 }
445 WalRecordType::Rollback if rec.tx_id != 0 => {
446 if let Some(span_index) = pending_tx_spans
447 .iter()
448 .rposition(|(tx_id, _)| *tx_id == rec.tx_id)
449 {
450 pending_tx_spans.remove(span_index);
451 }
452 }
453 _ => {}
454 }
455 }
456 }
457
458 let mut replayed_inserts = 0usize;
459 let mut replayed_updates = 0usize;
460 let mut replayed_deletes = 0usize;
461 let mut skipped = 0usize;
462 let mut skipped_uncommitted = 0usize;
463 let mut saw_ddl = false;
464 for (index, rec) in records.iter().enumerate() {
465 if has_boundaries
466 && !committed_row_records[index]
467 && matches!(
468 rec.record_type,
469 WalRecordType::Insert | WalRecordType::Update | WalRecordType::Delete
470 )
471 {
472 skipped_uncommitted += 1;
473 continue;
474 }
475 match rec.record_type {
476 WalRecordType::Insert => {
477 if let Some((table_name, rid, row_bytes)) = decode_wal_payload(&rec.data) {
478 if let Some(slot) = self.name_to_slot.get(&table_name).copied() {
479 let tbl = &mut self.tables[slot];
480 if rec.lsn > 0 && tbl.heap.page_lsn(rid.page_id) >= rec.lsn {
484 skipped += 1;
485 continue;
486 }
487 tbl.heap.insert_at(rid, &row_bytes)?;
494 tbl.heap.set_page_lsn(rid.page_id, rec.lsn)?;
495 replayed_inserts += 1;
496 }
497 }
498 }
499 WalRecordType::Update => {
500 if let Some((table_name, rid, row_bytes)) = decode_wal_payload(&rec.data) {
501 if let Some(slot) = self.name_to_slot.get(&table_name).copied() {
502 let tbl = &mut self.tables[slot];
503 if rec.lsn > 0 && tbl.heap.page_lsn(rid.page_id) >= rec.lsn {
504 skipped += 1;
505 continue;
506 }
507 let new_rid = tbl.heap.update(rid, &row_bytes)?;
508 tbl.heap.set_page_lsn(new_rid.page_id, rec.lsn)?;
509 replayed_updates += 1;
510 }
511 }
512 }
513 WalRecordType::Delete => {
514 if let Some((table_name, rid, _)) = decode_wal_payload(&rec.data) {
515 if let Some(slot) = self.name_to_slot.get(&table_name).copied() {
516 let tbl = &mut self.tables[slot];
517 if rec.lsn > 0 && tbl.heap.page_lsn(rid.page_id) >= rec.lsn {
518 skipped += 1;
519 continue;
520 }
521 let _ = tbl.heap.delete(rid);
522 tbl.heap.set_page_lsn(rid.page_id, rec.lsn)?;
523 replayed_deletes += 1;
524 }
525 }
526 }
527 WalRecordType::Begin | WalRecordType::Commit | WalRecordType::Rollback => {
528 }
530 WalRecordType::DdlCreateTable => {
531 saw_ddl = true;
532 if let Some((schema, defaults, auto_cols)) = decode_ddl_create_table(&rec.data)
533 {
534 if !self.name_to_slot.contains_key(&schema.table_name) {
535 if let Ok(mut table) = Table::create(schema, &self.data_dir) {
536 table.set_defaults(defaults);
537 table.set_auto_cols(auto_cols);
538 let slot = self.tables.len();
539 let name = table.schema.table_name.clone();
540 self.tables.push(table);
541 self.name_to_slot.insert(name, slot);
542 }
543 }
544 }
545 }
546 WalRecordType::DdlDropTable => {
547 saw_ddl = true;
548 if let Some((table_name, _)) = decode_ddl_table_name(&rec.data) {
549 if let Some(&slot) = self.name_to_slot.get(&table_name) {
550 let heap_path = self.data_dir.join(format!("{table_name}.heap"));
551 if heap_path.exists() {
552 let _ = fs::remove_file(&heap_path);
553 }
554 for col_name in self.tables[slot].indexed_column_names() {
555 let idx_path =
556 self.data_dir.join(format!("{table_name}_{col_name}.idx"));
557 if idx_path.exists() {
558 let _ = fs::remove_file(&idx_path);
559 }
560 }
561 self.name_to_slot.remove(&table_name);
562 let last = self.tables.len() - 1;
563 if slot != last {
564 let moved_name = self.tables[last].schema.table_name.clone();
565 self.tables.swap(slot, last);
566 self.name_to_slot.insert(moved_name, slot);
567 }
568 self.tables.pop();
569 }
570 }
571 }
572 WalRecordType::DdlAddColumn => {
573 saw_ddl = true;
574 if let Some((table_name, col)) = decode_ddl_alter_add_column(&rec.data) {
575 if let Some(&slot) = self.name_to_slot.get(&table_name) {
576 let tbl = &mut self.tables[slot];
577 if !tbl.schema.columns.iter().any(|c| c.name == col.name) {
578 let old_schema = tbl.schema.clone();
579 let has_rows = tbl.heap.scan().next().is_some();
580 tbl.schema.columns.push(col);
581 tbl.refresh_layout();
582 if has_rows {
583 let fill = vec![Value::Empty; tbl.schema.columns.len()];
584 let data_dir = self.data_dir.clone();
585 let _ = tbl.rewrite_rows_for_schema_change(
586 &old_schema,
587 &fill,
588 &data_dir,
589 );
590 }
591 }
592 if rec.lsn > 0 {
599 let _ = tbl.heap.stamp_all_pages_min_lsn(rec.lsn);
600 }
601 }
602 }
603 }
604 WalRecordType::DdlDropColumn => {
605 saw_ddl = true;
606 if let Some((table_name, col_name)) = decode_ddl_alter_drop_column(&rec.data) {
607 if let Some(&slot) = self.name_to_slot.get(&table_name) {
608 let tbl = &mut self.tables[slot];
609 if let Some(idx) =
610 tbl.schema.columns.iter().position(|c| c.name == col_name)
611 {
612 let old_schema = tbl.schema.clone();
613 let has_rows = tbl.heap.scan().next().is_some();
614 tbl.schema.columns.remove(idx);
615 for (i, c) in tbl.schema.columns.iter_mut().enumerate() {
616 c.position = i as u16;
617 }
618 tbl.refresh_layout();
619 if has_rows {
620 let fill = vec![Value::Empty; tbl.schema.columns.len()];
621 let data_dir = self.data_dir.clone();
622 let _ = tbl.rewrite_rows_for_schema_change(
623 &old_schema,
624 &fill,
625 &data_dir,
626 );
627 }
628 }
629 if rec.lsn > 0 {
630 let _ = tbl.heap.stamp_all_pages_min_lsn(rec.lsn);
631 }
632 }
633 }
634 }
635 }
636 }
637 info!(
638 inserts = replayed_inserts,
639 updates = replayed_updates,
640 deletes = replayed_deletes,
641 skipped = skipped,
642 skipped_uncommitted = skipped_uncommitted,
643 "WAL record apply complete (commit-boundary + LSN idempotent)"
644 );
645 if saw_ddl {
646 self.persist()?;
647 }
648 for tbl in &mut self.tables {
662 tbl.heap.flush_all_dirty()?;
663 tbl.heap.flush()?;
664 tbl.rebuild_indexes_from_heap()?;
665 tbl.save_dirty_indexes()?;
670 }
671 if let Some(max_lsn) = max_record_lsn(records) {
672 self.record_durable_lsn_at_least(max_lsn)?;
673 self.wal.set_next_lsn_at_least(max_lsn.saturating_add(1));
674 }
675 Ok(())
676 }
677
678 pub fn checkpoint(&mut self) -> io::Result<()> {
687 self.ensure_no_active_transaction_for_checkpoint()?;
688 self.ensure_plain_checkpoint_allowed_before_flush()?;
689 self.flush_checkpoint_state()?;
690 self.wal.flush()?;
691 self.record_durable_lsn_at_least(self.wal.last_appended_lsn())?;
692 self.wal.truncate()?;
693 self.checkpointed = true;
694 Ok(())
695 }
696
697 pub fn checkpoint_with_wal_archive<F>(&mut self, mut archive: F) -> io::Result<()>
705 where
706 F: FnMut(&Path, &[WalRecord]) -> io::Result<()>,
707 {
708 self.ensure_no_active_transaction_for_checkpoint()?;
709 self.commit_autocommit()?;
710 self.flush_checkpoint_state()?;
711 self.wal.flush()?;
712 let records = self.wal.read_all()?;
713 let archive: WalArchiveCallback<'_> = &mut archive;
714 archive(&self.data_dir, &records)?;
715 if let Some(max_lsn) = max_record_lsn(&records) {
716 self.record_durable_lsn_at_least(max_lsn)?;
717 } else {
718 self.record_durable_lsn_at_least(self.wal.last_appended_lsn())?;
719 }
720 self.wal.truncate()?;
721 self.checkpointed = true;
722 Ok(())
723 }
724
725 fn ensure_no_active_transaction_for_checkpoint(&self) -> io::Result<()> {
726 if self.active_tx_id.is_some() {
727 return Err(io::Error::other(
728 "cannot checkpoint while an explicit transaction is active",
729 ));
730 }
731 Ok(())
732 }
733
734 fn flush_checkpoint_state(&mut self) -> io::Result<()> {
735 for tbl in &mut self.tables {
736 tbl.heap.flush_all_dirty()?;
737 tbl.heap.flush()?;
738 tbl.save_dirty_indexes()?;
743 }
744 Ok(())
745 }
746
747 fn ensure_plain_checkpoint_allowed_before_flush(&self) -> io::Result<()> {
748 if !self.sync_identity_file_exists() {
749 return Ok(());
750 }
751 if self.wal.has_pending() {
752 return Err(io::Error::other(
753 "sync identity exists but checkpoint/recovery was called without a WAL archive hook; refusing to truncate retained history",
754 ));
755 }
756 let records = self.wal.read_all()?;
757 self.ensure_plain_wal_truncate_allowed(&records)
758 }
759
760 fn ensure_plain_wal_truncate_allowed(&self, records: &[WalRecord]) -> io::Result<()> {
761 if records.is_empty() {
762 return Ok(());
763 }
764 if self.sync_identity_file_exists() {
765 return Err(io::Error::other(
766 "sync identity exists but checkpoint/recovery was called without a WAL archive hook; refusing to truncate retained history",
767 ));
768 }
769 Ok(())
770 }
771
772 fn sync_identity_file_exists(&self) -> bool {
773 self.data_dir
774 .join(SYNC_STATE_DIR)
775 .join(SYNC_IDENTITY_FILE)
776 .exists()
777 }
778
779 fn record_durable_lsn_at_least(&mut self, lsn: u64) -> io::Result<()> {
780 if lsn <= self.durable_lsn {
781 return Ok(());
782 }
783 self.durable_lsn = lsn;
784 write_durable_lsn(&self.data_dir, lsn)
785 }
786
787 #[inline]
789 fn next_tx(&mut self) -> u64 {
790 if let Some(id) = self.active_tx_id {
791 return id;
792 }
793 let id = self.next_tx_id;
794 self.next_tx_id = self.next_tx_id.wrapping_add(1);
795 id
796 }
797
798 pub fn begin_transaction(&mut self) -> io::Result<()> {
800 if self.active_tx_id.is_some() {
801 return Err(io::Error::new(
802 io::ErrorKind::InvalidInput,
803 "explicit transaction is already active",
804 ));
805 }
806 let start_len = self.wal.synced_len()?;
807 let id = self.next_tx_id;
808 self.next_tx_id = self.next_tx_id.wrapping_add(1);
809 self.active_tx_id = Some(id);
810 self.tx_start_len = Some(start_len);
811 self.pending_autocommit_tx_ids.clear();
812 if !self.wal.is_off() {
813 self.wal.append(id, WalRecordType::Begin, &[])?;
814 self.wal.flush()?;
815 }
816 Ok(())
817 }
818
819 pub fn commit_transaction(&mut self) -> io::Result<()> {
822 if let Some(id) = self.active_tx_id.take() {
823 if !self.wal.is_off() {
824 self.wal.append(id, WalRecordType::Commit, &[])?;
825 self.wal.flush()?;
826 }
827 }
828 self.tx_start_len = None;
829 Ok(())
830 }
831
832 pub fn commit_autocommit(&mut self) -> io::Result<()> {
836 if !self.wal.is_off() && !self.pending_autocommit_tx_ids.is_empty() {
837 self.pending_autocommit_tx_ids.sort_unstable();
838 self.pending_autocommit_tx_ids.dedup();
839 for id in self.pending_autocommit_tx_ids.drain(..) {
840 self.wal.append(id, WalRecordType::Commit, &[])?;
841 }
842 }
843 self.wal.flush()
844 }
845
846 fn wal_log(
861 &mut self,
862 tx_id: u64,
863 record_type: WalRecordType,
864 table: &str,
865 rid: RowId,
866 row_bytes: &[u8],
867 ) -> io::Result<()> {
868 if self.wal.is_off() {
875 return Ok(());
876 }
877 let payload = encode_wal_payload(table, rid, row_bytes);
878 self.wal.append(tx_id, record_type, &payload)?;
879 if self.active_tx_id.is_none() {
880 self.pending_autocommit_tx_ids.push(tx_id);
881 }
882 Ok(())
883 }
884
885 #[inline]
891 pub fn sync_wal(&mut self) -> io::Result<()> {
892 self.wal.flush()
893 }
894
895 pub fn set_wal_sync_mode(&mut self, mode: WalSyncMode) {
903 self.wal.set_sync_mode(mode);
904 }
905
906 pub fn set_wal_sync_deferred(&mut self, defer: bool) {
914 self.wal.set_defer_sync(defer);
915 }
916
917 pub fn take_wal_durability_ticket(&mut self) -> Option<WalDurabilityTicket> {
920 self.wal.take_durability_ticket()
921 }
922
923 pub fn wal_fsync_count(&self) -> u64 {
925 self.wal.fsync_count()
926 }
927
928 pub fn rollback_to_last_sync(&mut self) -> io::Result<()> {
943 self.rollback_to_last_sync_inner(None)
944 }
945
946 pub fn rollback_to_last_sync_with_wal_archive<F>(&mut self, mut archive: F) -> io::Result<()>
951 where
952 F: FnMut(&Path, &[WalRecord]) -> io::Result<()>,
953 {
954 let archive: WalArchiveCallback<'_> = &mut archive;
955 self.rollback_to_last_sync_inner(Some(archive))
956 }
957
958 fn rollback_to_last_sync_inner(
959 &mut self,
960 mut archive: Option<WalArchiveCallback<'_>>,
961 ) -> io::Result<()> {
962 let start_len = self.tx_start_len.unwrap_or(0);
963 let prearchived = if let Some(archive) = archive.as_mut() {
964 let records = self.wal.read_through_len(start_len)?;
965 if !records.is_empty() {
966 archive(&self.data_dir, &records)?;
967 }
968 true
969 } else {
970 false
971 };
972
973 let start_len = self.tx_start_len.take().unwrap_or(0);
974 if let Some(id) = self.active_tx_id.take() {
975 if !self.wal.is_off() {
976 let _ = self.wal.append(id, WalRecordType::Rollback, &[]);
977 }
978 }
979 self.wal.discard_and_truncate_to(start_len)?;
980
981 for tbl in &mut self.tables {
991 tbl.heap.discard_dirty();
992 tbl.discard_dirty_indexes();
993 }
994 self.wal.discard_pending()?;
1000 let data_dir = self.data_dir.clone();
1004 let sync_mode = self.wal.sync_mode();
1005 let restored = if prearchived {
1006 let mut already_archived = |_dir: &Path, _records: &[WalRecord]| Ok(());
1007 let archive: WalArchiveCallback<'_> = &mut already_archived;
1008 Self::open_inner(&data_dir, Some(archive))?
1009 } else {
1010 Self::open_inner(&data_dir, archive)?
1011 };
1012 *self = restored;
1013 self.wal.set_sync_mode(sync_mode);
1014 Ok(())
1015 }
1016
1017 fn abandon_active_transaction_for_drop(&mut self) -> io::Result<()> {
1018 for tbl in &mut self.tables {
1019 tbl.heap.discard_dirty();
1020 }
1021 self.pending_autocommit_tx_ids.clear();
1022 let truncate_result = match self.tx_start_len.take() {
1023 Some(start_len) => self.wal.discard_and_truncate_to(start_len),
1024 None => self.wal.discard_pending(),
1025 };
1026 self.active_tx_id = None;
1027 truncate_result
1028 }
1029
1030 pub fn data_dir(&self) -> &Path {
1032 &self.data_dir
1033 }
1034
1035 pub fn max_lsn(&self) -> u64 {
1040 let max_page_lsn = self
1041 .tables
1042 .iter()
1043 .map(|t| t.heap.max_page_lsn())
1044 .max()
1045 .unwrap_or(0);
1046 max_page_lsn
1047 .max(self.durable_lsn)
1048 .max(self.wal.last_appended_lsn())
1049 }
1050
1051 pub fn create_table(&mut self, schema: Schema) -> io::Result<()> {
1052 self.create_table_full(schema, Vec::new(), Vec::new())
1053 }
1054
1055 pub fn create_table_with_defaults(
1059 &mut self,
1060 schema: Schema,
1061 defaults: Vec<Option<Value>>,
1062 ) -> io::Result<()> {
1063 self.create_table_full(schema, defaults, Vec::new())
1064 }
1065
1066 pub fn create_table_full(
1071 &mut self,
1072 schema: Schema,
1073 defaults: Vec<Option<Value>>,
1074 auto_cols: Vec<bool>,
1075 ) -> io::Result<()> {
1076 validate_table_name(&schema.table_name)?;
1077 for col in &schema.columns {
1078 validate_column_name(&col.name)?;
1079 }
1080 let name = schema.table_name.clone();
1081 if self.name_to_slot.contains_key(&name) {
1082 return Err(io::Error::new(
1083 io::ErrorKind::AlreadyExists,
1084 format!("table '{name}' already exists"),
1085 ));
1086 }
1087 if !self.wal.is_off() {
1088 let payload = encode_ddl_create_table(&schema, &defaults, &auto_cols);
1089 self.wal
1090 .append(0, WalRecordType::DdlCreateTable, &payload)?;
1091 self.wal.flush()?;
1092 }
1093 let mut table = Table::create(schema, &self.data_dir)?;
1094 table.set_defaults(defaults);
1095 table.set_auto_cols(auto_cols);
1096 let slot = self.tables.len();
1097 self.tables.push(table);
1098 self.name_to_slot.insert(name, slot);
1099 self.persist()?;
1100 Ok(())
1101 }
1102
1103 pub fn column_defaults(&self, table: &str) -> Option<&[Option<Value>]> {
1107 let slot = *self.name_to_slot.get(table)?;
1108 Some(self.tables[slot].defaults())
1109 }
1110
1111 pub fn auto_columns(&self, table: &str) -> Option<&[bool]> {
1114 let slot = *self.name_to_slot.get(table)?;
1115 Some(self.tables[slot].auto_cols())
1116 }
1117
1118 pub fn assign_auto_columns(&mut self, table: &str, values: &mut [Value]) {
1122 if let Some(&slot) = self.name_to_slot.get(table) {
1123 self.tables[slot].assign_auto(values);
1124 }
1125 }
1126
1127 fn persist(&self) -> io::Result<()> {
1132 let cat_path = self.data_dir.join(CATALOG_FILE);
1133 let tmp_path = self.data_dir.join(format!("{CATALOG_FILE}.tmp"));
1134 let entries: Vec<CatalogEntryRef<'_>> = self
1135 .tables
1136 .iter()
1137 .map(|t| CatalogEntryRef {
1138 schema: &t.schema,
1139 indexed_cols: t.indexed_column_metas(),
1140 defaults: t.defaults(),
1141 auto_cols: t.auto_cols(),
1142 })
1143 .collect();
1144 write_catalog_file(&tmp_path, &entries)?;
1145 fs::rename(&tmp_path, &cat_path)?;
1146 Ok(())
1147 }
1148
1149 #[inline]
1153 pub fn table_slot(&self, name: &str) -> Option<usize> {
1154 self.name_to_slot.get(name).copied()
1155 }
1156
1157 #[inline]
1160 pub fn table_by_slot(&self, slot: usize) -> &Table {
1161 &self.tables[slot]
1162 }
1163
1164 #[inline]
1166 pub fn table_by_slot_mut(&mut self, slot: usize) -> &mut Table {
1167 &mut self.tables[slot]
1168 }
1169
1170 pub fn get_table(&self, name: &str) -> Option<&Table> {
1171 let slot = *self.name_to_slot.get(name)?;
1172 Some(&self.tables[slot])
1173 }
1174
1175 pub fn get_table_mut(&mut self, name: &str) -> Option<&mut Table> {
1176 let slot = *self.name_to_slot.get(name)?;
1177 Some(&mut self.tables[slot])
1178 }
1179
1180 #[inline]
1185 fn by_name(&self, table: &str) -> io::Result<&Table> {
1186 let slot = *self.name_to_slot.get(table).ok_or_else(|| {
1187 io::Error::new(
1188 io::ErrorKind::NotFound,
1189 format!("table '{table}' not found"),
1190 )
1191 })?;
1192 Ok(&self.tables[slot])
1193 }
1194
1195 #[inline]
1197 fn by_name_mut(&mut self, table: &str) -> io::Result<&mut Table> {
1198 let slot = *self.name_to_slot.get(table).ok_or_else(|| {
1199 io::Error::new(
1200 io::ErrorKind::NotFound,
1201 format!("table '{table}' not found"),
1202 )
1203 })?;
1204 Ok(&mut self.tables[slot])
1205 }
1206
1207 pub fn insert(&mut self, table: &str, values: &Row) -> io::Result<RowId> {
1208 if self.wal.is_off() {
1217 return self.by_name_mut(table)?.insert(values);
1218 }
1219 let tbl = self.by_name_mut(table)?;
1220 let mut wal_bytes: Vec<u8> = Vec::new();
1221 encode_row_into(&tbl.schema, values, &mut wal_bytes);
1222 let new_rid = tbl.insert(values)?;
1236 let tx_id = self.next_tx();
1237 self.wal_log(tx_id, WalRecordType::Insert, table, new_rid, &wal_bytes)?;
1238 let lsn = self.wal.last_appended_lsn();
1243 if lsn > 0 {
1244 self.by_name_mut(table)?
1245 .heap
1246 .set_page_lsn(new_rid.page_id, lsn)?;
1247 }
1248 Ok(new_rid)
1249 }
1250
1251 pub fn insert_by_slot(&mut self, slot: usize, values: &Row) -> io::Result<RowId> {
1259 if self.wal.is_off() {
1260 return self.tables[slot].insert(values);
1261 }
1262 let tx_id = self.next_tx();
1263 let autocommit = self.active_tx_id.is_none();
1264 let Catalog { tables, wal, .. } = self;
1265 let tbl = &mut tables[slot];
1266 let mut wal_bytes: Vec<u8> = Vec::new();
1267 encode_row_into(&tbl.schema, values, &mut wal_bytes);
1268 let new_rid = tbl.insert(values)?;
1271 let payload = encode_wal_payload(&tbl.schema.table_name, new_rid, &wal_bytes);
1272 wal.append(tx_id, WalRecordType::Insert, &payload)?;
1273 if autocommit {
1274 self.pending_autocommit_tx_ids.push(tx_id);
1275 }
1276 let lsn = wal.last_appended_lsn();
1277 if lsn > 0 {
1278 tbl.heap.set_page_lsn(new_rid.page_id, lsn)?;
1279 }
1280 Ok(new_rid)
1281 }
1282
1283 pub fn get(&self, table: &str, rid: RowId) -> Option<Row> {
1284 self.get_table(table)?.get(rid)
1285 }
1286
1287 pub fn delete(&mut self, table: &str, rid: RowId) -> io::Result<()> {
1288 if self.wal.is_off() {
1291 return self.by_name_mut(table)?.delete(rid);
1292 }
1293 let tx_id = self.next_tx();
1294 self.wal_log(tx_id, WalRecordType::Delete, table, rid, &[])?;
1296 self.by_name_mut(table)?.delete(rid)
1297 }
1298
1299 pub fn delete_many(&mut self, table: &str, rids: &[RowId]) -> io::Result<u64> {
1303 if self.wal.is_off() {
1311 return self.by_name_mut(table)?.delete_many(rids);
1312 }
1313 let tx_id = self.next_tx();
1314 for &rid in rids {
1315 let payload = encode_wal_payload(table, rid, &[]);
1316 self.wal.append(tx_id, WalRecordType::Delete, &payload)?;
1317 }
1318 if self.active_tx_id.is_none() && !rids.is_empty() {
1319 self.pending_autocommit_tx_ids.push(tx_id);
1320 }
1321 self.by_name_mut(table)?.delete_many(rids)
1322 }
1323
1324 pub fn scan_delete_matching<P>(&mut self, table: &str, pred: P) -> io::Result<u64>
1334 where
1335 P: FnMut(&[u8]) -> bool,
1336 {
1337 self.by_name_mut(table)?.scan_delete_matching(pred)
1338 }
1339
1340 pub fn scan_delete_matching_logged<P>(&mut self, table: &str, pred: P) -> io::Result<u64>
1351 where
1352 P: FnMut(&[u8]) -> bool,
1353 {
1354 if self.wal.is_off() {
1360 return self.by_name_mut(table)?.scan_delete_matching(pred);
1361 }
1362 let slot = *self.name_to_slot.get(table).ok_or_else(|| {
1366 io::Error::new(
1367 io::ErrorKind::NotFound,
1368 format!("table '{table}' not found"),
1369 )
1370 })?;
1371 let tx_id = self.next_tx();
1372 let autocommit = self.active_tx_id.is_none();
1373 let Catalog { tables, wal, .. } = self;
1376 let tbl = &mut tables[slot];
1377 let name_bytes = table.as_bytes();
1381 let count = tbl.scan_delete_matching_with_hook(pred, |rid, row_bytes| {
1382 let mut payload: Vec<u8> =
1383 Vec::with_capacity(4 + name_bytes.len() + 10 + row_bytes.len());
1384 payload.extend_from_slice(&(name_bytes.len() as u32).to_le_bytes());
1385 payload.extend_from_slice(name_bytes);
1386 payload.extend_from_slice(&rid.page_id.to_le_bytes());
1387 payload.extend_from_slice(&rid.slot_index.to_le_bytes());
1388 payload.extend_from_slice(&0u32.to_le_bytes());
1392 let _ = wal.append(tx_id, WalRecordType::Delete, &payload);
1399 })?;
1400 if autocommit && count > 0 {
1401 self.pending_autocommit_tx_ids.push(tx_id);
1402 }
1403 Ok(count)
1405 }
1406
1407 pub fn scan_patch_matching_logged<P, M>(
1415 &mut self,
1416 table: &str,
1417 pred: P,
1418 try_mutate: M,
1419 ) -> io::Result<(u64, Vec<RowId>)>
1420 where
1421 P: FnMut(&[u8]) -> bool,
1422 M: FnMut(&mut [u8]) -> Option<u16>,
1423 {
1424 if self.wal.is_off() {
1425 return self.by_name_mut(table)?.scan_patch_matching_with_hook(
1426 pred,
1427 try_mutate,
1428 |_, _| {},
1429 );
1430 }
1431 let slot = *self.name_to_slot.get(table).ok_or_else(|| {
1432 io::Error::new(
1433 io::ErrorKind::NotFound,
1434 format!("table '{table}' not found"),
1435 )
1436 })?;
1437 let tx_id = self.next_tx();
1438 let autocommit = self.active_tx_id.is_none();
1439 let Catalog { tables, wal, .. } = self;
1440 let tbl = &mut tables[slot];
1441 let name_bytes = table.as_bytes();
1442 let result = tbl.scan_patch_matching_with_hook(pred, try_mutate, |rid, row_bytes| {
1443 let mut payload: Vec<u8> =
1444 Vec::with_capacity(4 + name_bytes.len() + 10 + row_bytes.len());
1445 payload.extend_from_slice(&(name_bytes.len() as u32).to_le_bytes());
1446 payload.extend_from_slice(name_bytes);
1447 payload.extend_from_slice(&rid.page_id.to_le_bytes());
1448 payload.extend_from_slice(&rid.slot_index.to_le_bytes());
1449 payload.extend_from_slice(&(row_bytes.len() as u32).to_le_bytes());
1450 payload.extend_from_slice(row_bytes);
1451 let _ = wal.append(tx_id, WalRecordType::Update, &payload);
1452 })?;
1453 if autocommit && result.0 > 0 {
1454 self.pending_autocommit_tx_ids.push(tx_id);
1455 }
1456 Ok(result)
1457 }
1458
1459 pub fn update(&mut self, table: &str, rid: RowId, values: &Row) -> io::Result<RowId> {
1460 if self.wal.is_off() {
1463 return self.by_name_mut(table)?.update(rid, values);
1464 }
1465 let tbl = self.by_name_mut(table)?;
1466 let mut wal_bytes: Vec<u8> = Vec::new();
1467 encode_row_into(&tbl.schema, values, &mut wal_bytes);
1468 check_encoded_row_size(&wal_bytes)?;
1471 let tx_id = self.next_tx();
1472 self.wal_log(tx_id, WalRecordType::Update, table, rid, &wal_bytes)?;
1473 self.by_name_mut(table)?.update(rid, values)
1474 }
1475
1476 pub fn update_hinted(
1480 &mut self,
1481 table: &str,
1482 rid: RowId,
1483 values: &Row,
1484 changed_col_indices: Option<&[usize]>,
1485 ) -> io::Result<RowId> {
1486 if self.wal.is_off() {
1490 return self
1491 .by_name_mut(table)?
1492 .update_hinted(rid, values, changed_col_indices);
1493 }
1494 let tbl = self.by_name_mut(table)?;
1495 let mut wal_bytes: Vec<u8> = Vec::new();
1496 encode_row_into(&tbl.schema, values, &mut wal_bytes);
1497 check_encoded_row_size(&wal_bytes)?;
1499 let tx_id = self.next_tx();
1500 self.wal_log(tx_id, WalRecordType::Update, table, rid, &wal_bytes)?;
1501 self.by_name_mut(table)?
1502 .update_hinted(rid, values, changed_col_indices)
1503 }
1504
1505 #[inline]
1517 pub fn with_row_bytes_mut<F>(&mut self, table: &str, rid: RowId, f: F) -> io::Result<bool>
1518 where
1519 F: FnOnce(&mut [u8]),
1520 {
1521 self.by_name_mut(table)?.with_row_bytes_mut(rid, f)
1522 }
1523
1524 #[inline]
1538 pub fn update_row_bytes_logged<F>(&mut self, table: &str, rid: RowId, f: F) -> io::Result<bool>
1539 where
1540 F: FnOnce(&mut [u8]),
1541 {
1542 let slot = *self.name_to_slot.get(table).ok_or_else(|| {
1543 io::Error::new(
1544 io::ErrorKind::NotFound,
1545 format!("table '{table}' not found"),
1546 )
1547 })?;
1548 self.update_row_bytes_logged_by_slot(slot, rid, f)
1549 }
1550
1551 #[inline]
1556 pub fn update_row_bytes_logged_by_slot<F>(
1557 &mut self,
1558 slot: usize,
1559 rid: RowId,
1560 f: F,
1561 ) -> io::Result<bool>
1562 where
1563 F: FnOnce(&mut [u8]),
1564 {
1565 let tbl = &mut self.tables[slot];
1568 let ok = tbl.with_row_bytes_mut(rid, f)?;
1569 if !ok {
1570 return Ok(false);
1571 }
1572 if self.wal.is_off() {
1576 return Ok(true);
1577 }
1578 let new_bytes = match tbl.heap.get(rid) {
1581 Some(b) => b,
1582 None => return Ok(false),
1584 };
1585 let table_name = tbl.schema.table_name.clone();
1588 let tx_id = self.next_tx();
1589 self.wal_log(tx_id, WalRecordType::Update, &table_name, rid, &new_bytes)?;
1590 Ok(true)
1591 }
1592
1593 #[inline]
1605 pub fn patch_var_col_in_place(
1606 &mut self,
1607 table: &str,
1608 rid: RowId,
1609 col_idx: usize,
1610 new_value: Option<&[u8]>,
1611 ) -> io::Result<bool> {
1612 self.by_name_mut(table)?
1613 .patch_var_col_in_place(rid, col_idx, new_value)
1614 }
1615
1616 pub fn patch_var_col_logged(
1622 &mut self,
1623 table: &str,
1624 rid: RowId,
1625 col_idx: usize,
1626 new_value: Option<&[u8]>,
1627 ) -> io::Result<bool> {
1628 let slot = *self.name_to_slot.get(table).ok_or_else(|| {
1629 io::Error::new(
1630 io::ErrorKind::NotFound,
1631 format!("table '{table}' not found"),
1632 )
1633 })?;
1634 let tbl = &mut self.tables[slot];
1635 let ok = tbl.patch_var_col_in_place(rid, col_idx, new_value)?;
1636 if !ok {
1637 return Ok(false);
1638 }
1639 if self.wal.is_off() {
1642 return Ok(true);
1643 }
1644 let new_bytes = match tbl.heap.get(rid) {
1645 Some(b) => b,
1646 None => return Ok(false),
1647 };
1648 let table_name = tbl.schema.table_name.clone();
1649 let tx_id = self.next_tx();
1650 self.wal_log(tx_id, WalRecordType::Update, &table_name, rid, &new_bytes)?;
1651 Ok(true)
1652 }
1653
1654 pub fn scan(&self, table: &str) -> io::Result<impl Iterator<Item = (RowId, Row)> + '_> {
1655 Ok(self.by_name(table)?.scan())
1656 }
1657
1658 pub fn for_each_row_raw<F>(&self, table: &str, f: F) -> io::Result<()>
1661 where
1662 F: FnMut(RowId, &[u8]),
1663 {
1664 self.by_name(table)?.for_each_row_raw(f);
1665 Ok(())
1666 }
1667
1668 pub fn try_for_each_row_raw<F>(&self, table: &str, f: F) -> io::Result<()>
1673 where
1674 F: FnMut(RowId, &[u8]) -> std::ops::ControlFlow<()>,
1675 {
1676 self.by_name(table)?.try_for_each_row_raw(f);
1677 Ok(())
1678 }
1679
1680 pub fn create_index(&mut self, table: &str, column: &str) -> io::Result<()> {
1681 self.create_index_unique(table, column, false)
1682 }
1683
1684 pub fn create_index_unique(
1689 &mut self,
1690 table: &str,
1691 column: &str,
1692 unique: bool,
1693 ) -> io::Result<()> {
1694 let data_dir = self.data_dir.clone();
1695 self.by_name_mut(table)?
1696 .create_index_with_unique(column, &data_dir, unique)?;
1697 self.persist()
1701 }
1702
1703 pub fn is_index_unique(&self, table: &str, column: &str) -> Option<bool> {
1707 self.get_table(table)?.is_index_unique(column)
1708 }
1709
1710 pub fn has_index(&self, table: &str, column: &str) -> bool {
1712 self.get_table(table)
1713 .map(|t| t.has_index(column))
1714 .unwrap_or(false)
1715 }
1716
1717 pub fn index_lookup(&self, table: &str, column: &str, key: &Value) -> io::Result<Option<Row>> {
1718 Ok(self
1719 .by_name(table)?
1720 .index_lookup(column, key)
1721 .map(|(_, row)| row))
1722 }
1723
1724 pub fn list_tables(&self) -> Vec<&str> {
1725 self.tables
1729 .iter()
1730 .map(|t| t.schema.table_name.as_str())
1731 .collect()
1732 }
1733
1734 pub fn schema(&self, table: &str) -> Option<&Schema> {
1735 let slot = *self.name_to_slot.get(table)?;
1736 Some(&self.tables[slot].schema)
1737 }
1738
1739 pub fn drop_table(&mut self, name: &str) -> io::Result<()> {
1742 validate_table_name(name)?;
1743 let slot = *self.name_to_slot.get(name).ok_or_else(|| {
1744 io::Error::new(io::ErrorKind::NotFound, format!("table '{name}' not found"))
1745 })?;
1746 if !self.wal.is_off() {
1747 let payload = encode_ddl_drop_table(name);
1748 self.wal.append(0, WalRecordType::DdlDropTable, &payload)?;
1749 self.wal.flush()?;
1750 }
1751 let table = &self.tables[slot];
1753 let heap_path = self
1754 .data_dir
1755 .join(format!("{}.heap", table.schema.table_name));
1756 if heap_path.exists() {
1757 fs::remove_file(&heap_path)?;
1758 }
1759 for col_name in table.indexed_column_names() {
1765 let idx_path = self.data_dir.join(format!("{name}_{col_name}.idx"));
1766 if idx_path.exists() {
1767 let _ = fs::remove_file(&idx_path);
1768 }
1769 }
1770 self.name_to_slot.remove(name);
1772 let last = self.tables.len() - 1;
1773 if slot != last {
1774 let moved_name = self.tables[last].schema.table_name.clone();
1775 self.tables.swap(slot, last);
1776 self.name_to_slot.insert(moved_name, slot);
1777 }
1778 self.tables.pop();
1779 self.persist()?;
1780 Ok(())
1781 }
1782
1783 pub fn alter_table_add_column(&mut self, table: &str, col: ColumnDef) -> io::Result<()> {
1806 let data_dir = self.data_dir.clone();
1807 {
1808 let tbl = self.by_name_mut(table)?;
1809 if tbl.schema.columns.iter().any(|c| c.name == col.name) {
1810 return Err(io::Error::new(
1811 io::ErrorKind::AlreadyExists,
1812 format!("column '{}' already exists in table '{table}'", col.name),
1813 ));
1814 }
1815 }
1816 let barrier_lsn = if !self.wal.is_off() {
1817 let payload = encode_ddl_alter_add_column(table, &col);
1818 self.wal.append(0, WalRecordType::DdlAddColumn, &payload)?;
1819 self.wal.flush()?;
1820 self.wal.last_appended_lsn()
1821 } else {
1822 0
1823 };
1824 let tbl = self.by_name_mut(table)?;
1825
1826 let old_schema = tbl.schema.clone();
1827
1828 let has_rows = tbl.heap.scan().next().is_some();
1832
1833 if has_rows && col.required {
1834 return Err(io::Error::new(
1835 io::ErrorKind::InvalidInput,
1836 format!(
1837 "cannot add required column '{}' to non-empty table '{table}': \
1838 no default value to backfill existing rows with",
1839 col.name
1840 ),
1841 ));
1842 }
1843
1844 tbl.schema.columns.push(col);
1848 tbl.refresh_layout();
1849
1850 if has_rows {
1851 let fill: Vec<Value> = vec![Value::Empty; tbl.schema.columns.len()];
1856 tbl.rewrite_rows_for_schema_change(&old_schema, &fill, &data_dir)?;
1857 }
1858 if barrier_lsn > 0 {
1866 tbl.heap.stamp_all_pages_min_lsn(barrier_lsn)?;
1867 tbl.heap.flush()?;
1868 }
1869
1870 self.persist()?;
1871 Ok(())
1872 }
1873
1874 pub fn alter_table_drop_column(&mut self, table: &str, col_name: &str) -> io::Result<()> {
1897 let data_dir = self.data_dir.clone();
1898 {
1899 let tbl = self.by_name_mut(table)?;
1900 tbl.schema
1901 .columns
1902 .iter()
1903 .position(|c| c.name == col_name)
1904 .ok_or_else(|| {
1905 io::Error::new(
1906 io::ErrorKind::NotFound,
1907 format!("column '{col_name}' not found in table '{table}'"),
1908 )
1909 })?;
1910 }
1911 let barrier_lsn = if !self.wal.is_off() {
1912 let payload = encode_ddl_alter_drop_column(table, col_name);
1913 self.wal.append(0, WalRecordType::DdlDropColumn, &payload)?;
1914 self.wal.flush()?;
1915 self.wal.last_appended_lsn()
1916 } else {
1917 0
1918 };
1919 let tbl = self.by_name_mut(table)?;
1920 let idx = tbl
1921 .schema
1922 .columns
1923 .iter()
1924 .position(|c| c.name == col_name)
1925 .ok_or_else(|| {
1926 io::Error::new(
1927 io::ErrorKind::NotFound,
1928 format!("column '{col_name}' not found in table '{table}'"),
1929 )
1930 })?;
1931
1932 let old_schema = tbl.schema.clone();
1934 let has_rows = tbl.heap.scan().next().is_some();
1935
1936 tbl.schema.columns.remove(idx);
1938 for (i, col) in tbl.schema.columns.iter_mut().enumerate() {
1939 col.position = i as u16;
1940 }
1941 tbl.refresh_layout();
1942
1943 if has_rows {
1944 let fill: Vec<Value> = vec![Value::Empty; tbl.schema.columns.len()];
1950 tbl.rewrite_rows_for_schema_change(&old_schema, &fill, &data_dir)?;
1951 }
1952 if barrier_lsn > 0 {
1954 tbl.heap.stamp_all_pages_min_lsn(barrier_lsn)?;
1955 tbl.heap.flush()?;
1956 }
1957
1958 self.persist()?;
1959 Ok(())
1960 }
1961}
1962
1963impl Drop for Catalog {
1964 fn drop(&mut self) {
1965 if self.active_tx_id.is_some() {
1966 if let Err(e) = self.abandon_active_transaction_for_drop() {
1967 warn!(error = %e, "catalog drop active transaction cleanup failed");
1968 }
1969 return;
1970 }
1971 if let Err(e) = self.checkpoint() {
1983 warn!(error = %e, "catalog drop checkpoint failed");
1984 }
1985 }
1986}
1987
1988fn encode_wal_payload(table: &str, rid: RowId, row_bytes: &[u8]) -> Vec<u8> {
2003 let name = table.as_bytes();
2004 let mut out = Vec::with_capacity(4 + name.len() + 4 + 2 + 4 + row_bytes.len());
2005 out.extend_from_slice(&(name.len() as u32).to_le_bytes());
2006 out.extend_from_slice(name);
2007 out.extend_from_slice(&rid.page_id.to_le_bytes());
2008 out.extend_from_slice(&rid.slot_index.to_le_bytes());
2009 out.extend_from_slice(&(row_bytes.len() as u32).to_le_bytes());
2010 out.extend_from_slice(row_bytes);
2011 out
2012}
2013
2014fn decode_wal_payload(data: &[u8]) -> Option<(String, RowId, Vec<u8>)> {
2015 let mut pos = 0usize;
2016 if data.len() < 4 {
2017 return None;
2018 }
2019 let name_len = u32::from_le_bytes(data[pos..pos + 4].try_into().ok()?) as usize;
2020 pos += 4;
2021 if pos + name_len > data.len() {
2022 return None;
2023 }
2024 let name = std::str::from_utf8(&data[pos..pos + name_len])
2025 .ok()?
2026 .to_string();
2027 pos += name_len;
2028 if pos + 4 + 2 + 4 > data.len() {
2029 return None;
2030 }
2031 let page_id = u32::from_le_bytes(data[pos..pos + 4].try_into().ok()?);
2032 pos += 4;
2033 let slot_index = u16::from_le_bytes(data[pos..pos + 2].try_into().ok()?);
2034 pos += 2;
2035 let row_len = u32::from_le_bytes(data[pos..pos + 4].try_into().ok()?) as usize;
2036 pos += 4;
2037 if pos + row_len > data.len() {
2038 return None;
2039 }
2040 let row_bytes = data[pos..pos + row_len].to_vec();
2041 Some((
2042 name,
2043 RowId {
2044 page_id,
2045 slot_index,
2046 },
2047 row_bytes,
2048 ))
2049}
2050
2051fn encode_ddl_create_table(
2054 schema: &Schema,
2055 defaults: &[Option<Value>],
2056 auto_cols: &[bool],
2057) -> Vec<u8> {
2058 let name = schema.table_name.as_bytes();
2059 let mut out = Vec::new();
2060 out.extend_from_slice(&(name.len() as u32).to_le_bytes());
2061 out.extend_from_slice(name);
2062 out.extend_from_slice(&(schema.columns.len() as u16).to_le_bytes());
2063 for col in &schema.columns {
2064 let cn = col.name.as_bytes();
2065 out.extend_from_slice(&(cn.len() as u32).to_le_bytes());
2066 out.extend_from_slice(cn);
2067 out.push(col.type_id as u8);
2068 out.push(col.required as u8);
2069 out.extend_from_slice(&col.position.to_le_bytes());
2070 }
2071 encode_defaults_section(&mut out, defaults);
2075 encode_auto_section(&mut out, auto_cols);
2076 out
2077}
2078
2079fn decode_ddl_create_table(data: &[u8]) -> Option<(Schema, Vec<Option<Value>>, Vec<bool>)> {
2080 let mut pos = 0usize;
2081 if data.len() < 4 {
2082 return None;
2083 }
2084 let name_len = u32::from_le_bytes(data[pos..pos + 4].try_into().ok()?) as usize;
2085 pos += 4;
2086 if pos + name_len > data.len() {
2087 return None;
2088 }
2089 let table_name = std::str::from_utf8(&data[pos..pos + name_len])
2090 .ok()?
2091 .to_string();
2092 pos += name_len;
2093 if pos + 2 > data.len() {
2094 return None;
2095 }
2096 let n_cols = u16::from_le_bytes(data[pos..pos + 2].try_into().ok()?) as usize;
2097 pos += 2;
2098 let mut columns = Vec::with_capacity(n_cols);
2099 for _ in 0..n_cols {
2100 if pos + 4 > data.len() {
2101 return None;
2102 }
2103 let cn_len = u32::from_le_bytes(data[pos..pos + 4].try_into().ok()?) as usize;
2104 pos += 4;
2105 if pos + cn_len + 4 > data.len() {
2106 return None;
2107 }
2108 let col_name = std::str::from_utf8(&data[pos..pos + cn_len])
2109 .ok()?
2110 .to_string();
2111 pos += cn_len;
2112 let type_id = TypeId::from_u8(data[pos])?;
2113 pos += 1;
2114 let required = data[pos] != 0;
2115 pos += 1;
2116 if pos + 2 > data.len() {
2117 return None;
2118 }
2119 let position = u16::from_le_bytes(data[pos..pos + 2].try_into().ok()?);
2120 pos += 2;
2121 columns.push(ColumnDef {
2122 name: col_name,
2123 type_id,
2124 required,
2125 position,
2126 });
2127 }
2128 let defaults = if pos < data.len() {
2131 decode_defaults_section(data, &mut pos, columns.len())?
2132 } else {
2133 Vec::new()
2134 };
2135 let auto_cols = if pos < data.len() {
2136 decode_auto_section(data, &mut pos, columns.len())?
2137 } else {
2138 Vec::new()
2139 };
2140 Some((
2141 Schema {
2142 table_name,
2143 columns,
2144 },
2145 defaults,
2146 auto_cols,
2147 ))
2148}
2149
2150fn encode_ddl_drop_table(table_name: &str) -> Vec<u8> {
2151 let name = table_name.as_bytes();
2152 let mut out = Vec::with_capacity(4 + name.len());
2153 out.extend_from_slice(&(name.len() as u32).to_le_bytes());
2154 out.extend_from_slice(name);
2155 out
2156}
2157
2158fn encode_ddl_alter_add_column(table_name: &str, col: &ColumnDef) -> Vec<u8> {
2159 let name = table_name.as_bytes();
2160 let cn = col.name.as_bytes();
2161 let mut out = Vec::with_capacity(4 + name.len() + 4 + cn.len() + 4);
2162 out.extend_from_slice(&(name.len() as u32).to_le_bytes());
2163 out.extend_from_slice(name);
2164 out.extend_from_slice(&(cn.len() as u32).to_le_bytes());
2165 out.extend_from_slice(cn);
2166 out.push(col.type_id as u8);
2167 out.push(col.required as u8);
2168 out.extend_from_slice(&col.position.to_le_bytes());
2169 out
2170}
2171
2172fn encode_ddl_alter_drop_column(table_name: &str, col_name: &str) -> Vec<u8> {
2173 let name = table_name.as_bytes();
2174 let cn = col_name.as_bytes();
2175 let mut out = Vec::with_capacity(4 + name.len() + 4 + cn.len());
2176 out.extend_from_slice(&(name.len() as u32).to_le_bytes());
2177 out.extend_from_slice(name);
2178 out.extend_from_slice(&(cn.len() as u32).to_le_bytes());
2179 out.extend_from_slice(cn);
2180 out
2181}
2182
2183fn decode_ddl_table_name(data: &[u8]) -> Option<(String, usize)> {
2184 if data.len() < 4 {
2185 return None;
2186 }
2187 let name_len = u32::from_le_bytes(data[0..4].try_into().ok()?) as usize;
2188 if 4 + name_len > data.len() {
2189 return None;
2190 }
2191 let name = std::str::from_utf8(&data[4..4 + name_len])
2192 .ok()?
2193 .to_string();
2194 Some((name, 4 + name_len))
2195}
2196
2197fn decode_ddl_alter_add_column(data: &[u8]) -> Option<(String, ColumnDef)> {
2198 let (table_name, mut pos) = decode_ddl_table_name(data)?;
2199 if pos + 4 > data.len() {
2200 return None;
2201 }
2202 let cn_len = u32::from_le_bytes(data[pos..pos + 4].try_into().ok()?) as usize;
2203 pos += 4;
2204 if pos + cn_len + 4 > data.len() {
2205 return None;
2206 }
2207 let col_name = std::str::from_utf8(&data[pos..pos + cn_len])
2208 .ok()?
2209 .to_string();
2210 pos += cn_len;
2211 let type_id = TypeId::from_u8(data[pos])?;
2212 pos += 1;
2213 let required = data[pos] != 0;
2214 pos += 1;
2215 if pos + 2 > data.len() {
2216 return None;
2217 }
2218 let position = u16::from_le_bytes(data[pos..pos + 2].try_into().ok()?);
2219 Some((
2220 table_name,
2221 ColumnDef {
2222 name: col_name,
2223 type_id,
2224 required,
2225 position,
2226 },
2227 ))
2228}
2229
2230fn decode_ddl_alter_drop_column(data: &[u8]) -> Option<(String, String)> {
2231 let (table_name, pos) = decode_ddl_table_name(data)?;
2232 if pos + 4 > data.len() {
2233 return None;
2234 }
2235 let cn_len = u32::from_le_bytes(data[pos..pos + 4].try_into().ok()?) as usize;
2236 if pos + 4 + cn_len > data.len() {
2237 return None;
2238 }
2239 let col_name = std::str::from_utf8(&data[pos + 4..pos + 4 + cn_len])
2240 .ok()?
2241 .to_string();
2242 Some((table_name, col_name))
2243}
2244
2245pub(crate) struct IndexedColMeta {
2273 pub name: String,
2274 pub unique: bool,
2275}
2276
2277pub(crate) struct CatalogEntry {
2280 pub schema: Schema,
2281 pub indexed_cols: Vec<IndexedColMeta>,
2282 pub defaults: Vec<Option<Value>>,
2285 pub auto_cols: Vec<bool>,
2288}
2289
2290pub(crate) struct CatalogEntryRef<'a> {
2292 pub schema: &'a Schema,
2293 pub indexed_cols: Vec<IndexedColMeta>,
2294 pub defaults: &'a [Option<Value>],
2295 pub auto_cols: &'a [bool],
2296}
2297
2298fn encode_value_blob(out: &mut Vec<u8>, v: &Value) {
2304 out.push(v.type_id() as u8);
2305 match v {
2306 Value::Int(n) => out.extend_from_slice(&n.to_le_bytes()),
2307 Value::Float(f) => out.extend_from_slice(&f.to_bits().to_le_bytes()),
2308 Value::Bool(b) => out.push(*b as u8),
2309 Value::Str(s) => {
2310 out.extend_from_slice(&(s.len() as u32).to_le_bytes());
2311 out.extend_from_slice(s.as_bytes());
2312 }
2313 Value::DateTime(n) => out.extend_from_slice(&n.to_le_bytes()),
2314 Value::Uuid(u) => out.extend_from_slice(u),
2315 Value::Bytes(b) => {
2316 out.extend_from_slice(&(b.len() as u32).to_le_bytes());
2317 out.extend_from_slice(b);
2318 }
2319 Value::Empty => {}
2320 }
2321}
2322
2323fn decode_value_blob(data: &[u8], pos: &mut usize) -> Option<Value> {
2326 let tag = *data.get(*pos)?;
2327 *pos += 1;
2328 let type_id = TypeId::from_u8(tag)?;
2329 let take_fixed = |pos: &mut usize, n: usize| -> Option<Vec<u8>> {
2330 if *pos + n > data.len() {
2331 return None;
2332 }
2333 let slice = data[*pos..*pos + n].to_vec();
2334 *pos += n;
2335 Some(slice)
2336 };
2337 match type_id {
2338 TypeId::Empty => Some(Value::Empty),
2339 TypeId::Int => Some(Value::Int(i64::from_le_bytes(
2340 take_fixed(pos, 8)?.try_into().ok()?,
2341 ))),
2342 TypeId::Float => Some(Value::Float(f64::from_bits(u64::from_le_bytes(
2343 take_fixed(pos, 8)?.try_into().ok()?,
2344 )))),
2345 TypeId::Bool => Some(Value::Bool(take_fixed(pos, 1)?[0] != 0)),
2346 TypeId::DateTime => Some(Value::DateTime(i64::from_le_bytes(
2347 take_fixed(pos, 8)?.try_into().ok()?,
2348 ))),
2349 TypeId::Uuid => Some(Value::Uuid(take_fixed(pos, 16)?.try_into().ok()?)),
2350 TypeId::Str => {
2351 let len = u32::from_le_bytes(take_fixed(pos, 4)?.try_into().ok()?) as usize;
2352 Some(Value::Str(String::from_utf8(take_fixed(pos, len)?).ok()?))
2353 }
2354 TypeId::Bytes => {
2355 let len = u32::from_le_bytes(take_fixed(pos, 4)?.try_into().ok()?) as usize;
2356 Some(Value::Bytes(take_fixed(pos, len)?))
2357 }
2358 }
2359}
2360
2361fn encode_defaults_section(out: &mut Vec<u8>, defaults: &[Option<Value>]) {
2365 let present: Vec<(u16, &Value)> = defaults
2366 .iter()
2367 .enumerate()
2368 .filter_map(|(i, d)| d.as_ref().map(|v| (i as u16, v)))
2369 .collect();
2370 out.extend_from_slice(&(present.len() as u16).to_le_bytes());
2371 for (pos, v) in present {
2372 out.extend_from_slice(&pos.to_le_bytes());
2373 encode_value_blob(out, v);
2374 }
2375}
2376
2377fn decode_defaults_section(
2380 data: &[u8],
2381 pos: &mut usize,
2382 n_cols: usize,
2383) -> Option<Vec<Option<Value>>> {
2384 if *pos + 2 > data.len() {
2385 return None;
2386 }
2387 let count = u16::from_le_bytes(data[*pos..*pos + 2].try_into().ok()?) as usize;
2388 *pos += 2;
2389 let mut out = vec![None; n_cols];
2390 for _ in 0..count {
2391 if *pos + 2 > data.len() {
2392 return None;
2393 }
2394 let col = u16::from_le_bytes(data[*pos..*pos + 2].try_into().ok()?) as usize;
2395 *pos += 2;
2396 let value = decode_value_blob(data, pos)?;
2397 if col < n_cols {
2398 out[col] = Some(value);
2399 }
2400 }
2401 Some(out)
2402}
2403
2404fn encode_auto_section(out: &mut Vec<u8>, auto_cols: &[bool]) {
2408 let present: Vec<u16> = auto_cols
2409 .iter()
2410 .enumerate()
2411 .filter_map(|(i, &a)| if a { Some(i as u16) } else { None })
2412 .collect();
2413 out.extend_from_slice(&(present.len() as u16).to_le_bytes());
2414 for pos in present {
2415 out.extend_from_slice(&pos.to_le_bytes());
2416 }
2417}
2418
2419fn decode_auto_section(data: &[u8], pos: &mut usize, n_cols: usize) -> Option<Vec<bool>> {
2422 if *pos + 2 > data.len() {
2423 return None;
2424 }
2425 let count = u16::from_le_bytes(data[*pos..*pos + 2].try_into().ok()?) as usize;
2426 *pos += 2;
2427 let mut out = vec![false; n_cols];
2428 for _ in 0..count {
2429 if *pos + 2 > data.len() {
2430 return None;
2431 }
2432 let col = u16::from_le_bytes(data[*pos..*pos + 2].try_into().ok()?) as usize;
2433 *pos += 2;
2434 if col < n_cols {
2435 out[col] = true;
2436 }
2437 }
2438 Some(out)
2439}
2440
2441fn write_catalog_file(path: &Path, entries: &[CatalogEntryRef<'_>]) -> io::Result<()> {
2442 let mut buf: Vec<u8> = Vec::with_capacity(64);
2443 buf.extend_from_slice(CATALOG_MAGIC);
2444 buf.extend_from_slice(&CATALOG_VERSION.to_le_bytes());
2445 buf.extend_from_slice(&(entries.len() as u32).to_le_bytes());
2446
2447 for entry in entries {
2448 let schema = entry.schema;
2449 let name = schema.table_name.as_bytes();
2450 buf.extend_from_slice(&(name.len() as u32).to_le_bytes());
2451 buf.extend_from_slice(name);
2452 buf.extend_from_slice(&(schema.columns.len() as u16).to_le_bytes());
2453 for col in &schema.columns {
2454 let cn = col.name.as_bytes();
2455 buf.extend_from_slice(&(cn.len() as u32).to_le_bytes());
2456 buf.extend_from_slice(cn);
2457 buf.push(col.type_id as u8);
2458 buf.push(if col.required { 1 } else { 0 });
2459 buf.extend_from_slice(&col.position.to_le_bytes());
2460 }
2461 buf.extend_from_slice(&(entry.indexed_cols.len() as u16).to_le_bytes());
2463 for meta in &entry.indexed_cols {
2464 let cn = meta.name.as_bytes();
2465 buf.extend_from_slice(&(cn.len() as u32).to_le_bytes());
2466 buf.extend_from_slice(cn);
2467 buf.push(if meta.unique { 1 } else { 0 });
2468 }
2469 encode_defaults_section(&mut buf, entry.defaults);
2471 encode_auto_section(&mut buf, entry.auto_cols);
2473 }
2474
2475 let crc = crc32fast::hash(&buf);
2479 buf.extend_from_slice(&crc.to_le_bytes());
2480
2481 let mut f = fs::OpenOptions::new()
2482 .create(true)
2483 .write(true)
2484 .truncate(true)
2485 .open(path)?;
2486 f.write_all(&buf)?;
2487 f.sync_data()?;
2488 Ok(())
2489}
2490
2491fn read_catalog_file(path: &Path) -> io::Result<Vec<CatalogEntry>> {
2492 let mut f = fs::File::open(path)?;
2493 let mut buf = Vec::new();
2494 f.read_to_end(&mut buf)?;
2495
2496 let mut pos = 0usize;
2497 if buf.len() < 14 || &buf[0..4] != CATALOG_MAGIC {
2499 return Err(io::Error::new(
2500 io::ErrorKind::InvalidData,
2501 "bad catalog magic",
2502 ));
2503 }
2504
2505 let payload = &buf[..buf.len() - 4];
2507 let stored_crc = u32::from_le_bytes(
2508 buf[buf.len() - 4..]
2509 .try_into()
2510 .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "truncated catalog CRC"))?,
2511 );
2512 let computed_crc = crc32fast::hash(payload);
2513 if stored_crc != computed_crc {
2514 return Err(io::Error::new(
2515 io::ErrorKind::InvalidData,
2516 format!(
2517 "catalog CRC32 mismatch: expected {stored_crc:#010x}, got {computed_crc:#010x}"
2518 ),
2519 ));
2520 }
2521 let buf = &buf[..buf.len() - 4];
2523 pos += 4;
2524 let version = u16::from_le_bytes(
2525 buf[pos..pos + 2]
2526 .try_into()
2527 .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "truncated catalog header"))?,
2528 );
2529 pos += 2;
2530 if version == 0 || version > CATALOG_VERSION {
2540 return Err(io::Error::new(
2541 io::ErrorKind::InvalidData,
2542 format!("unsupported catalog version: {version}"),
2543 ));
2544 }
2545 let n_tables = u32::from_le_bytes(
2546 buf[pos..pos + 4]
2547 .try_into()
2548 .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "truncated catalog header"))?,
2549 ) as usize;
2550 pos += 4;
2551
2552 if n_tables > buf.len() {
2559 return Err(io::Error::new(
2560 io::ErrorKind::InvalidData,
2561 format!("catalog file corrupt: implausible table count {n_tables}"),
2562 ));
2563 }
2564
2565 let mut entries = Vec::with_capacity(n_tables);
2566 for _ in 0..n_tables {
2567 let name_len = read_u32(buf, &mut pos)? as usize;
2568 let table_name = read_string(buf, &mut pos, name_len)?;
2569 let n_cols = read_u16(buf, &mut pos)? as usize;
2570
2571 let mut columns = Vec::with_capacity(n_cols);
2572 for _ in 0..n_cols {
2573 let cname_len = read_u32(buf, &mut pos)? as usize;
2574 let name = read_string(buf, &mut pos, cname_len)?;
2575 let type_id_raw = read_u8(buf, &mut pos)?;
2576 let type_id = type_id_from_u8(type_id_raw)?;
2577 let required = read_u8(buf, &mut pos)? != 0;
2578 let position = read_u16(buf, &mut pos)?;
2579 columns.push(ColumnDef {
2580 name,
2581 type_id,
2582 required,
2583 position,
2584 });
2585 }
2586
2587 let indexed_cols: Vec<IndexedColMeta> = if version >= 3 {
2591 let n = read_u16(buf, &mut pos)? as usize;
2592 let mut v = Vec::with_capacity(n);
2593 for _ in 0..n {
2594 let l = read_u32(buf, &mut pos)? as usize;
2595 let name = read_string(buf, &mut pos, l)?;
2596 let unique = read_u8(buf, &mut pos)? != 0;
2597 v.push(IndexedColMeta { name, unique });
2598 }
2599 v
2600 } else if version >= 2 {
2601 let n = read_u16(buf, &mut pos)? as usize;
2602 let mut v = Vec::with_capacity(n);
2603 for _ in 0..n {
2604 let l = read_u32(buf, &mut pos)? as usize;
2605 let name = read_string(buf, &mut pos, l)?;
2606 v.push(IndexedColMeta {
2607 name,
2608 unique: false,
2609 });
2610 }
2611 v
2612 } else {
2613 Vec::new()
2614 };
2615
2616 let defaults = if version >= 4 {
2618 decode_defaults_section(buf, &mut pos, columns.len()).ok_or_else(|| {
2619 io::Error::new(io::ErrorKind::InvalidData, "truncated catalog defaults")
2620 })?
2621 } else {
2622 Vec::new()
2623 };
2624
2625 let auto_cols = if version >= 5 {
2627 decode_auto_section(buf, &mut pos, columns.len()).ok_or_else(|| {
2628 io::Error::new(io::ErrorKind::InvalidData, "truncated catalog auto columns")
2629 })?
2630 } else {
2631 Vec::new()
2632 };
2633
2634 entries.push(CatalogEntry {
2635 schema: Schema {
2636 table_name,
2637 columns,
2638 },
2639 indexed_cols,
2640 defaults,
2641 auto_cols,
2642 });
2643 }
2644
2645 Ok(entries)
2646}
2647
2648fn read_u8(buf: &[u8], pos: &mut usize) -> io::Result<u8> {
2649 if *pos >= buf.len() {
2650 return Err(io::Error::new(
2651 io::ErrorKind::UnexpectedEof,
2652 "truncated catalog",
2653 ));
2654 }
2655 let v = buf[*pos];
2656 *pos += 1;
2657 Ok(v)
2658}
2659fn read_u16(buf: &[u8], pos: &mut usize) -> io::Result<u16> {
2660 if *pos + 2 > buf.len() {
2661 return Err(io::Error::new(
2662 io::ErrorKind::UnexpectedEof,
2663 "truncated catalog",
2664 ));
2665 }
2666 let v = u16::from_le_bytes(
2667 buf[*pos..*pos + 2]
2668 .try_into()
2669 .expect("bounds checked above"),
2670 );
2671 *pos += 2;
2672 Ok(v)
2673}
2674fn read_u32(buf: &[u8], pos: &mut usize) -> io::Result<u32> {
2675 if *pos + 4 > buf.len() {
2676 return Err(io::Error::new(
2677 io::ErrorKind::UnexpectedEof,
2678 "truncated catalog",
2679 ));
2680 }
2681 let v = u32::from_le_bytes(
2682 buf[*pos..*pos + 4]
2683 .try_into()
2684 .expect("bounds checked above"),
2685 );
2686 *pos += 4;
2687 Ok(v)
2688}
2689fn read_string(buf: &[u8], pos: &mut usize, len: usize) -> io::Result<String> {
2690 if *pos + len > buf.len() {
2691 return Err(io::Error::new(
2692 io::ErrorKind::UnexpectedEof,
2693 "truncated catalog string",
2694 ));
2695 }
2696 let s = std::str::from_utf8(&buf[*pos..*pos + len])
2697 .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "non-utf8 in catalog"))?
2698 .to_string();
2699 *pos += len;
2700 Ok(s)
2701}
2702fn type_id_from_u8(v: u8) -> io::Result<TypeId> {
2703 match v {
2704 0 => Ok(TypeId::Empty),
2705 1 => Ok(TypeId::Int),
2706 2 => Ok(TypeId::Float),
2707 3 => Ok(TypeId::Bool),
2708 4 => Ok(TypeId::Str),
2709 5 => Ok(TypeId::DateTime),
2710 6 => Ok(TypeId::Uuid),
2711 7 => Ok(TypeId::Bytes),
2712 _ => Err(io::Error::new(
2713 io::ErrorKind::InvalidData,
2714 format!("unknown type id: {v}"),
2715 )),
2716 }
2717}
2718
2719#[cfg(test)]
2720mod tests {
2721 use super::*;
2722 fn temp_catalog(name: &str) -> Catalog {
2723 let dir = std::env::temp_dir().join(format!("powdb_cat_{name}_{}", std::process::id()));
2724 Catalog::create(&dir).unwrap()
2725 }
2726
2727 fn schema_two_cols() -> Schema {
2728 Schema {
2729 table_name: "T".into(),
2730 columns: vec![
2731 ColumnDef {
2732 name: "id".into(),
2733 type_id: TypeId::Int,
2734 required: true,
2735 position: 0,
2736 },
2737 ColumnDef {
2738 name: "status".into(),
2739 type_id: TypeId::Str,
2740 required: false,
2741 position: 1,
2742 },
2743 ],
2744 }
2745 }
2746
2747 #[test]
2748 fn replay_records_treats_reused_tx_ids_as_ordered_spans() {
2749 let mut cat = temp_catalog("reused_tx_ids");
2750 let schema = schema_two_cols();
2751 cat.create_table(schema.clone()).unwrap();
2752 cat.checkpoint().unwrap();
2753
2754 let mut committed_row = Vec::new();
2755 encode_row_into(
2756 &schema,
2757 &[Value::Int(1), Value::Str("committed".into())],
2758 &mut committed_row,
2759 );
2760 let mut incomplete_row = Vec::new();
2761 encode_row_into(
2762 &schema,
2763 &[Value::Int(2), Value::Str("incomplete".into())],
2764 &mut incomplete_row,
2765 );
2766
2767 let records = vec![
2768 WalRecord {
2769 tx_id: 1,
2770 record_type: WalRecordType::Begin,
2771 lsn: 1,
2772 data: Vec::new(),
2773 },
2774 WalRecord {
2775 tx_id: 1,
2776 record_type: WalRecordType::Insert,
2777 lsn: 2,
2778 data: encode_wal_payload(
2779 "T",
2780 RowId {
2781 page_id: 1,
2782 slot_index: 0,
2783 },
2784 &committed_row,
2785 ),
2786 },
2787 WalRecord {
2788 tx_id: 1,
2789 record_type: WalRecordType::Commit,
2790 lsn: 3,
2791 data: Vec::new(),
2792 },
2793 WalRecord {
2794 tx_id: 1,
2795 record_type: WalRecordType::Begin,
2796 lsn: 4,
2797 data: Vec::new(),
2798 },
2799 WalRecord {
2800 tx_id: 1,
2801 record_type: WalRecordType::Insert,
2802 lsn: 5,
2803 data: encode_wal_payload(
2804 "T",
2805 RowId {
2806 page_id: 1,
2807 slot_index: 1,
2808 },
2809 &incomplete_row,
2810 ),
2811 },
2812 ];
2813
2814 cat.apply_wal_records(&records).unwrap();
2815 let rows: Vec<_> = cat.scan("T").unwrap().collect();
2816 assert_eq!(rows.len(), 1);
2817 assert_eq!(rows[0].1[0], Value::Int(1));
2818 assert_eq!(rows[0].1[1], Value::Str("committed".into()));
2819 }
2820
2821 #[test]
2822 fn ddl_create_table_codec_roundtrips_defaults_and_auto() {
2823 let schema = schema_two_cols();
2824 let defaults = vec![None, Some(Value::Str("active".into()))];
2825 let auto_cols = vec![true, false];
2826 let encoded = encode_ddl_create_table(&schema, &defaults, &auto_cols);
2827 let (decoded_schema, decoded_defaults, decoded_auto) =
2828 decode_ddl_create_table(&encoded).unwrap();
2829 assert_eq!(decoded_schema.columns.len(), 2);
2830 assert_eq!(decoded_defaults, defaults);
2831 assert_eq!(decoded_auto, auto_cols);
2832 }
2833
2834 #[test]
2835 fn ddl_create_table_codec_back_compat_without_trailing_sections() {
2836 let schema = schema_two_cols();
2840 let full = encode_ddl_create_table(&schema, &[], &[]);
2841 let legacy = &full[..full.len() - 4];
2844 let (decoded_schema, decoded_defaults, decoded_auto) =
2845 decode_ddl_create_table(legacy).unwrap();
2846 assert_eq!(decoded_schema.columns.len(), 2);
2847 assert!(decoded_defaults.is_empty(), "no defaults section -> empty");
2848 assert!(decoded_auto.is_empty(), "no auto section -> empty");
2849 }
2850
2851 #[test]
2852 fn ddl_create_table_codec_back_compat_defaults_but_no_auto() {
2853 let schema = schema_two_cols();
2856 let defaults = vec![None, Some(Value::Str("active".into()))];
2857 let full = encode_ddl_create_table(&schema, &defaults, &[]);
2858 let legacy = &full[..full.len() - 2];
2860 let (_schema, decoded_defaults, decoded_auto) = decode_ddl_create_table(legacy).unwrap();
2861 assert_eq!(decoded_defaults, defaults);
2862 assert!(decoded_auto.is_empty());
2863 }
2864
2865 #[test]
2866 fn read_catalog_file_accepts_intermediate_versions_3_and_4() {
2867 use std::io::Write as _;
2875 fn write_legacy_catalog(path: &std::path::Path, version: u16) {
2876 let mut buf: Vec<u8> = Vec::new();
2877 buf.extend_from_slice(CATALOG_MAGIC);
2878 buf.extend_from_slice(&version.to_le_bytes());
2879 buf.extend_from_slice(&1u32.to_le_bytes()); buf.extend_from_slice(&1u32.to_le_bytes());
2882 buf.extend_from_slice(b"T");
2883 buf.extend_from_slice(&2u16.to_le_bytes()); buf.extend_from_slice(&2u32.to_le_bytes());
2886 buf.extend_from_slice(b"id");
2887 buf.push(TypeId::Int as u8);
2888 buf.push(1);
2889 buf.extend_from_slice(&0u16.to_le_bytes());
2890 buf.extend_from_slice(&6u32.to_le_bytes());
2892 buf.extend_from_slice(b"status");
2893 buf.push(TypeId::Str as u8);
2894 buf.push(0);
2895 buf.extend_from_slice(&1u16.to_le_bytes());
2896 buf.extend_from_slice(&0u16.to_le_bytes());
2898 if version >= 4 {
2900 encode_defaults_section(&mut buf, &[None, None]);
2901 }
2902 let crc = crc32fast::hash(&buf);
2904 buf.extend_from_slice(&crc.to_le_bytes());
2905 let mut f = fs::File::create(path).unwrap();
2906 f.write_all(&buf).unwrap();
2907 }
2908
2909 for version in [3u16, 4u16] {
2910 let path = std::env::temp_dir().join(format!(
2911 "powdb_cat_v{version}_compat_{}.bin",
2912 std::process::id()
2913 ));
2914 write_legacy_catalog(&path, version);
2915 let entries = read_catalog_file(&path)
2916 .unwrap_or_else(|e| panic!("version {version} catalog must load, got: {e}"));
2917 assert_eq!(entries.len(), 1);
2918 assert_eq!(entries[0].schema.table_name, "T");
2919 assert_eq!(entries[0].schema.columns.len(), 2);
2920 assert!(
2921 entries[0].auto_cols.is_empty(),
2922 "v{version} has no auto cols"
2923 );
2924 fs::remove_file(&path).ok();
2925 }
2926 }
2927
2928 #[test]
2929 fn read_catalog_file_rejects_implausible_table_count() {
2930 use std::io::Write as _;
2938 let mut buf: Vec<u8> = Vec::new();
2939 buf.extend_from_slice(CATALOG_MAGIC);
2940 buf.extend_from_slice(&CATALOG_VERSION.to_le_bytes());
2941 buf.extend_from_slice(&1000u32.to_le_bytes()); let crc = crc32fast::hash(&buf);
2944 buf.extend_from_slice(&crc.to_le_bytes());
2945 let path =
2946 std::env::temp_dir().join(format!("powdb_cat_badcount_{}.bin", std::process::id()));
2947 fs::File::create(&path).unwrap().write_all(&buf).unwrap();
2948
2949 let msg = match read_catalog_file(&path) {
2950 Ok(_) => panic!("implausible table count must be rejected, got Ok"),
2951 Err(e) => e.to_string(),
2952 };
2953 assert!(
2954 msg.contains("implausible table count"),
2955 "expected an implausible-table-count error, got: {msg}"
2956 );
2957 fs::remove_file(&path).ok();
2958 }
2959
2960 #[test]
2961 fn data_dir_and_max_lsn_accessors() {
2962 let dir = std::env::temp_dir().join(format!("powdb_cat_maxlsn_{}", std::process::id()));
2963 let mut cat = Catalog::create(&dir).unwrap();
2964
2965 assert_eq!(cat.data_dir(), dir.as_path());
2967
2968 assert_eq!(cat.max_lsn(), 0);
2970
2971 let schema = Schema {
2972 table_name: "users".into(),
2973 columns: vec![ColumnDef {
2974 name: "name".into(),
2975 type_id: TypeId::Str,
2976 required: true,
2977 position: 0,
2978 }],
2979 };
2980 cat.create_table(schema).unwrap();
2981
2982 cat.insert("users", &vec![Value::Str("Alice".into())])
2983 .unwrap();
2984 cat.sync_wal().unwrap();
2985
2986 assert!(cat.max_lsn() > 0);
2989 }
2990
2991 #[test]
2992 fn test_create_table_and_insert() {
2993 let mut cat = temp_catalog("basic");
2994 let schema = Schema {
2995 table_name: "users".into(),
2996 columns: vec![
2997 ColumnDef {
2998 name: "name".into(),
2999 type_id: TypeId::Str,
3000 required: true,
3001 position: 0,
3002 },
3003 ColumnDef {
3004 name: "age".into(),
3005 type_id: TypeId::Int,
3006 required: false,
3007 position: 1,
3008 },
3009 ],
3010 };
3011 cat.create_table(schema).unwrap();
3012
3013 let row = vec![Value::Str("Alice".into()), Value::Int(30)];
3014 let rid = cat.insert("users", &row).unwrap();
3015
3016 let result = cat.get("users", rid).unwrap();
3017 assert_eq!(result[0], Value::Str("Alice".into()));
3018 assert_eq!(result[1], Value::Int(30));
3019 }
3020
3021 #[test]
3022 fn test_scan_table() {
3023 let mut cat = temp_catalog("scan");
3024 let schema = Schema {
3025 table_name: "items".into(),
3026 columns: vec![
3027 ColumnDef {
3028 name: "name".into(),
3029 type_id: TypeId::Str,
3030 required: true,
3031 position: 0,
3032 },
3033 ColumnDef {
3034 name: "price".into(),
3035 type_id: TypeId::Float,
3036 required: true,
3037 position: 1,
3038 },
3039 ],
3040 };
3041 cat.create_table(schema).unwrap();
3042
3043 for i in 0..50 {
3044 cat.insert(
3045 "items",
3046 &vec![
3047 Value::Str(format!("item_{i}")),
3048 Value::Float(i as f64 * 1.5),
3049 ],
3050 )
3051 .unwrap();
3052 }
3053
3054 let rows: Vec<_> = cat.scan("items").unwrap().collect();
3055 assert_eq!(rows.len(), 50);
3056 }
3057
3058 #[test]
3059 fn test_index_lookup() {
3060 let mut cat = temp_catalog("idx");
3061 let schema = Schema {
3062 table_name: "users".into(),
3063 columns: vec![
3064 ColumnDef {
3065 name: "email".into(),
3066 type_id: TypeId::Str,
3067 required: true,
3068 position: 0,
3069 },
3070 ColumnDef {
3071 name: "name".into(),
3072 type_id: TypeId::Str,
3073 required: true,
3074 position: 1,
3075 },
3076 ],
3077 };
3078 cat.create_table(schema).unwrap();
3079 cat.create_index("users", "email").unwrap();
3080
3081 cat.insert(
3082 "users",
3083 &vec![
3084 Value::Str("alice@example.com".into()),
3085 Value::Str("Alice".into()),
3086 ],
3087 )
3088 .unwrap();
3089 cat.insert(
3090 "users",
3091 &vec![
3092 Value::Str("bob@example.com".into()),
3093 Value::Str("Bob".into()),
3094 ],
3095 )
3096 .unwrap();
3097
3098 let result = cat
3099 .index_lookup("users", "email", &Value::Str("bob@example.com".into()))
3100 .unwrap();
3101 assert!(result.is_some());
3102 let row = result.unwrap();
3103 assert_eq!(row[1], Value::Str("Bob".into()));
3104 }
3105
3106 #[test]
3107 fn test_delete_row() {
3108 let mut cat = temp_catalog("delete");
3109 let schema = Schema {
3110 table_name: "t".into(),
3111 columns: vec![ColumnDef {
3112 name: "v".into(),
3113 type_id: TypeId::Int,
3114 required: true,
3115 position: 0,
3116 }],
3117 };
3118 cat.create_table(schema).unwrap();
3119 let r1 = cat.insert("t", &vec![Value::Int(1)]).unwrap();
3120 let r2 = cat.insert("t", &vec![Value::Int(2)]).unwrap();
3121 cat.delete("t", r1).unwrap();
3122 assert!(cat.get("t", r1).is_none());
3123 assert!(cat.get("t", r2).is_some());
3124 }
3125
3126 #[test]
3127 fn test_update_row() {
3128 let mut cat = temp_catalog("update");
3129 let schema = Schema {
3130 table_name: "t".into(),
3131 columns: vec![ColumnDef {
3132 name: "v".into(),
3133 type_id: TypeId::Int,
3134 required: true,
3135 position: 0,
3136 }],
3137 };
3138 cat.create_table(schema).unwrap();
3139 let rid = cat.insert("t", &vec![Value::Int(1)]).unwrap();
3140 let new_rid = cat.update("t", rid, &vec![Value::Int(99)]).unwrap();
3141 let row = cat.get("t", new_rid).unwrap();
3142 assert_eq!(row[0], Value::Int(99));
3143 }
3144
3145 #[test]
3146 fn test_persist_and_reopen() {
3147 let dir = std::env::temp_dir().join(format!("powdb_cat_persist_{}", std::process::id()));
3148 let _ = std::fs::remove_dir_all(&dir);
3150
3151 {
3152 let mut cat = Catalog::create(&dir).unwrap();
3153 cat.create_table(Schema {
3154 table_name: "users".into(),
3155 columns: vec![
3156 ColumnDef {
3157 name: "name".into(),
3158 type_id: TypeId::Str,
3159 required: true,
3160 position: 0,
3161 },
3162 ColumnDef {
3163 name: "age".into(),
3164 type_id: TypeId::Int,
3165 required: false,
3166 position: 1,
3167 },
3168 ],
3169 })
3170 .unwrap();
3171 cat.insert("users", &vec![Value::Str("Alice".into()), Value::Int(30)])
3172 .unwrap();
3173 cat.insert("users", &vec![Value::Str("Bob".into()), Value::Int(25)])
3174 .unwrap();
3175 }
3176
3177 let cat = Catalog::open(&dir).unwrap();
3179 let schema = cat.schema("users").unwrap();
3180 assert_eq!(schema.columns.len(), 2);
3181 assert_eq!(schema.columns[0].name, "name");
3182 assert_eq!(schema.columns[0].type_id, TypeId::Str);
3183 assert_eq!(schema.columns[1].type_id, TypeId::Int);
3184
3185 let rows: Vec<_> = cat.scan("users").unwrap().collect();
3186 assert_eq!(rows.len(), 2);
3187
3188 std::fs::remove_dir_all(&dir).ok();
3189 }
3190
3191 #[test]
3192 fn test_open_missing_dir_errors() {
3193 let dir = std::env::temp_dir().join(format!("powdb_cat_missing_{}", std::process::id()));
3194 let _ = std::fs::remove_dir_all(&dir);
3195 std::fs::create_dir_all(&dir).unwrap();
3196 assert!(Catalog::open(&dir).is_err());
3198 std::fs::remove_dir_all(&dir).ok();
3199 }
3200
3201 #[test]
3202 fn test_list_tables() {
3203 let mut cat = temp_catalog("list");
3204 cat.create_table(Schema {
3205 table_name: "a".into(),
3206 columns: vec![ColumnDef {
3207 name: "x".into(),
3208 type_id: TypeId::Int,
3209 required: true,
3210 position: 0,
3211 }],
3212 })
3213 .unwrap();
3214 cat.create_table(Schema {
3215 table_name: "b".into(),
3216 columns: vec![ColumnDef {
3217 name: "y".into(),
3218 type_id: TypeId::Int,
3219 required: true,
3220 position: 0,
3221 }],
3222 })
3223 .unwrap();
3224 let mut tables = cat.list_tables();
3225 tables.sort();
3226 assert_eq!(tables, vec!["a", "b"]);
3227 }
3228
3229 #[test]
3230 fn test_path_traversal_table_name_rejected() {
3231 let mut cat = temp_catalog("path_trav");
3232 let bad_names = vec![
3234 "../etc/passwd",
3235 "foo/bar",
3236 "table\0name",
3237 "",
3238 "123starts_with_digit",
3239 "has-dashes",
3240 "has spaces",
3241 "has.dots",
3242 ];
3243 for name in bad_names {
3244 let schema = Schema {
3245 table_name: name.into(),
3246 columns: vec![ColumnDef {
3247 name: "x".into(),
3248 type_id: TypeId::Int,
3249 required: true,
3250 position: 0,
3251 }],
3252 };
3253 let result = cat.create_table(schema);
3254 assert!(result.is_err(), "expected error for table name '{name}'");
3255 assert_eq!(result.unwrap_err().kind(), io::ErrorKind::InvalidInput);
3256 }
3257 let good_names = vec!["users", "_private", "Table_123", "_"];
3259 for name in good_names {
3260 let schema = Schema {
3261 table_name: name.into(),
3262 columns: vec![ColumnDef {
3263 name: "x".into(),
3264 type_id: TypeId::Int,
3265 required: true,
3266 position: 0,
3267 }],
3268 };
3269 assert!(
3270 cat.create_table(schema).is_ok(),
3271 "expected ok for table name '{name}'"
3272 );
3273 }
3274 }
3275
3276 #[test]
3277 fn test_path_traversal_column_name_rejected() {
3278 let mut cat = temp_catalog("col_path_trav");
3279 let schema = Schema {
3280 table_name: "valid_table".into(),
3281 columns: vec![ColumnDef {
3282 name: "../bad".into(),
3283 type_id: TypeId::Int,
3284 required: true,
3285 position: 0,
3286 }],
3287 };
3288 let result = cat.create_table(schema);
3289 assert!(result.is_err());
3290 assert_eq!(result.unwrap_err().kind(), io::ErrorKind::InvalidInput);
3291 }
3292
3293 #[test]
3294 fn test_drop_table_validates_name() {
3295 let mut cat = temp_catalog("drop_trav");
3296 let result = cat.drop_table("../etc/passwd");
3297 assert!(result.is_err());
3298 assert_eq!(result.unwrap_err().kind(), io::ErrorKind::InvalidInput);
3300 }
3301}