1use crate::error::StorageError;
2use crate::heap::HeapFile;
3use crate::page::{OVERFLOW_CHAIN_END, OVERFLOW_PAYLOAD_CAP};
4use crate::row::{encode_row_into, encode_row_v2_into, plan_spill, OverflowStub, MAX_VALUE_SIZE};
5use crate::table::Table;
6use crate::types::*;
7use crate::wal::{Wal, WalDurabilityTicket, WalRecord, WalRecordType, WalSyncMode};
8use rustc_hash::FxHashMap;
9use std::fs;
10use std::io::{self, Read, Write};
11use std::path::{Path, PathBuf};
12use tracing::{info, warn};
13
14fn check_encoded_row_size(encoded: &[u8]) -> io::Result<()> {
19 if encoded.len() > crate::page::MAX_ROW_DATA_SIZE {
20 return Err(crate::error::StorageError::RowTooLarge {
21 size: encoded.len(),
22 max: crate::page::MAX_ROW_DATA_SIZE,
23 }
24 .into());
25 }
26 Ok(())
27}
28
29fn validate_identifier(kind: &str, name: &str) -> io::Result<()> {
33 if name.is_empty() {
34 return Err(io::Error::new(
35 io::ErrorKind::InvalidInput,
36 format!("invalid {kind} name: must not be empty"),
37 ));
38 }
39 let mut chars = name.chars();
40 let first = chars.next().expect("non-empty name");
42 if !first.is_ascii_alphabetic() && first != '_' {
43 return Err(io::Error::new(
44 io::ErrorKind::InvalidInput,
45 format!("invalid {kind} name '{name}': must start with a letter or underscore"),
46 ));
47 }
48 for ch in chars {
49 if !ch.is_ascii_alphanumeric() && ch != '_' {
50 return Err(io::Error::new(
51 io::ErrorKind::InvalidInput,
52 format!(
53 "invalid {kind} name '{name}': must contain only letters, digits, and underscores"
54 ),
55 ));
56 }
57 }
58 Ok(())
59}
60
61fn validate_table_name(name: &str) -> io::Result<()> {
63 validate_identifier("table", name)
64}
65
66fn validate_column_name(name: &str) -> io::Result<()> {
68 validate_identifier("column", name)
69}
70
71const CATALOG_FILE: &str = "catalog.bin";
80pub const CATALOG_LSN_FILE: &str = "catalog.lsn";
81const CATALOG_MAGIC: &[u8; 4] = b"BCAT";
82pub const CATALOG_VERSION: u16 = 5;
86
87const WAL_FILE: &str = "wal.log";
90const SYNC_STATE_DIR: &str = ".powdb-sync";
91const SYNC_IDENTITY_FILE: &str = "identity.json";
92
93const WAL_BATCH_SIZE: usize = 64;
97type WalArchiveCallback<'a> = &'a mut dyn FnMut(&Path, &[WalRecord]) -> io::Result<()>;
98
99fn read_durable_lsn(data_dir: &Path) -> io::Result<u64> {
100 let path = data_dir.join(CATALOG_LSN_FILE);
101 let bytes = match fs::read(path) {
102 Ok(bytes) => bytes,
103 Err(err) if err.kind() == io::ErrorKind::NotFound => return Ok(0),
104 Err(err) => return Err(err),
105 };
106 if bytes.len() != 8 {
107 return Err(io::Error::new(
108 io::ErrorKind::InvalidData,
109 "catalog LSN sidecar has invalid length",
110 ));
111 }
112 let mut buf = [0u8; 8];
113 buf.copy_from_slice(&bytes);
114 Ok(u64::from_le_bytes(buf))
115}
116
117fn write_durable_lsn(data_dir: &Path, lsn: u64) -> io::Result<()> {
118 let path = data_dir.join(CATALOG_LSN_FILE);
119 let tmp_path = data_dir.join(format!("{CATALOG_LSN_FILE}.tmp"));
120 let mut file = fs::File::create(&tmp_path)?;
121 file.write_all(&lsn.to_le_bytes())?;
122 file.sync_all()?;
123 drop(file);
124 fs::rename(&tmp_path, &path)?;
125 sync_directory(data_dir)?;
126 Ok(())
127}
128
129#[cfg(unix)]
130fn sync_directory(path: &Path) -> io::Result<()> {
131 fs::File::open(path)?.sync_all()
132}
133
134#[cfg(not(unix))]
135fn sync_directory(path: &Path) -> io::Result<()> {
136 let _ = path;
137 Ok(())
138}
139
140fn max_record_lsn(records: &[WalRecord]) -> Option<u64> {
141 records.iter().map(|record| record.lsn).max()
142}
143
144pub struct Catalog {
158 tables: Vec<Table>,
161 name_to_slot: FxHashMap<String, usize>,
164 data_dir: PathBuf,
165 wal: Wal,
170 next_tx_id: u64,
174 active_tx_id: Option<u64>,
177 tx_start_len: Option<u64>,
180 pending_autocommit_tx_ids: Vec<u64>,
183 checkpointed: bool,
187 durable_lsn: u64,
190 pending_free_overflow: Vec<(usize, Vec<u32>)>,
199}
200
201impl Catalog {
202 pub fn create(data_dir: &Path) -> io::Result<Self> {
223 crate::create_data_dir_secure(data_dir)?;
224 let wal_path = data_dir.join(WAL_FILE);
225 let wal = Wal::create(&wal_path, WAL_BATCH_SIZE)?;
226 let cat = Catalog {
227 tables: Vec::new(),
228 name_to_slot: FxHashMap::default(),
229 data_dir: data_dir.to_path_buf(),
230 wal,
231 next_tx_id: 1,
232 active_tx_id: None,
233 tx_start_len: None,
234 pending_autocommit_tx_ids: Vec::new(),
235 pending_free_overflow: Vec::new(),
236 checkpointed: false,
237 durable_lsn: 0,
238 };
239 cat.persist()?;
240 Ok(cat)
241 }
242
243 pub fn open(data_dir: &Path) -> io::Result<Self> {
252 Self::open_inner(data_dir, None)
253 }
254
255 pub fn open_with_wal_archive<F>(data_dir: &Path, mut archive: F) -> io::Result<Self>
264 where
265 F: FnMut(&Path, &[WalRecord]) -> io::Result<()>,
266 {
267 let archive: WalArchiveCallback<'_> = &mut archive;
268 Self::open_inner(data_dir, Some(archive))
269 }
270
271 fn open_inner(data_dir: &Path, archive: Option<WalArchiveCallback<'_>>) -> io::Result<Self> {
272 let cat_path = data_dir.join(CATALOG_FILE);
273 if !cat_path.exists() {
274 return Err(io::Error::new(io::ErrorKind::NotFound, "no catalog file"));
275 }
276 let entries = read_catalog_file(&cat_path)?;
277 let durable_lsn = read_durable_lsn(data_dir)?;
278 let mut tables: Vec<Table> = Vec::with_capacity(entries.len());
279 let mut name_to_slot =
280 FxHashMap::with_capacity_and_hasher(entries.len(), Default::default());
281 for CatalogEntry {
282 schema,
283 indexed_cols,
284 defaults,
285 auto_cols,
286 } in entries
287 {
288 let name = schema.table_name.clone();
289 let mut table = Table::open_with_indexes(schema, data_dir, &indexed_cols)?;
295 table.set_defaults(defaults);
296 table.set_auto_cols(auto_cols);
297 name_to_slot.insert(name, tables.len());
298 tables.push(table);
299 }
300 let wal_path = data_dir.join(WAL_FILE);
301 let wal = Wal::open(&wal_path, WAL_BATCH_SIZE)?;
302 let mut cat = Catalog {
303 tables,
304 name_to_slot,
305 data_dir: data_dir.to_path_buf(),
306 wal,
307 next_tx_id: 1,
308 active_tx_id: None,
309 tx_start_len: None,
310 pending_autocommit_tx_ids: Vec::new(),
311 pending_free_overflow: Vec::new(),
312 checkpointed: false,
313 durable_lsn,
314 };
315 cat.replay_wal(archive)?;
316 let max_page_lsn = cat
326 .tables
327 .iter()
328 .map(|t| t.heap.max_page_lsn())
329 .max()
330 .unwrap_or(0);
331 let max_known_lsn = max_page_lsn.max(cat.durable_lsn);
332 cat.wal.set_next_lsn_at_least(max_known_lsn + 1);
333 if let Err(e) = cat.sweep_all() {
338 warn!(error = %e, "post-recovery overflow sweep failed (non-fatal)");
339 }
340 Ok(cat)
341 }
342
343 fn replay_wal(&mut self, mut archive: Option<WalArchiveCallback<'_>>) -> io::Result<()> {
369 let records = self.wal.read_all()?;
370 if records.is_empty() {
371 return Ok(());
372 }
373 if archive.is_none() {
374 self.ensure_plain_wal_truncate_allowed(&records)?;
375 }
376 self.replay_records(&records)?;
377 if let Some(archive) = archive.as_mut() {
378 archive(&self.data_dir, &records)?;
379 }
380 self.wal.truncate()?;
381 Ok(())
382 }
383
384 pub fn apply_wal_records(&mut self, records: &[WalRecord]) -> io::Result<()> {
393 self.ensure_no_active_transaction_for_checkpoint()?;
394 self.ensure_no_pending_wal_records()?;
395 self.replay_records(records)
396 }
397
398 pub fn ensure_no_pending_wal_records(&self) -> io::Result<()> {
401 if self.wal.has_pending() || !self.wal.read_all()?.is_empty() {
402 return Err(io::Error::other(
403 "cannot apply replicated WAL records while local WAL records are pending",
404 ));
405 }
406 Ok(())
407 }
408
409 fn replay_records(&mut self, records: &[WalRecord]) -> io::Result<()> {
410 if records.is_empty() {
411 return Ok(());
412 }
413
414 info!(count = records.len(), "applying WAL records");
415
416 let has_boundaries = records.iter().any(|rec| {
425 matches!(
426 rec.record_type,
427 WalRecordType::Begin | WalRecordType::Commit | WalRecordType::Rollback
428 )
429 });
430 let mut committed_row_records = vec![true; records.len()];
431 if has_boundaries {
432 committed_row_records.fill(false);
433 let mut pending_tx_spans: Vec<(u64, Vec<usize>)> = Vec::new();
434 for (index, rec) in records.iter().enumerate() {
435 match rec.record_type {
436 WalRecordType::Insert
437 | WalRecordType::Update
438 | WalRecordType::Delete
439 | WalRecordType::OverflowWrite
440 | WalRecordType::OverflowFree
441 if rec.tx_id == 0 =>
442 {
443 committed_row_records[index] = true;
444 }
445 WalRecordType::Insert
446 | WalRecordType::Update
447 | WalRecordType::Delete
448 | WalRecordType::OverflowWrite
449 | WalRecordType::OverflowFree => {
450 if let Some((_, rows)) = pending_tx_spans
451 .iter_mut()
452 .rev()
453 .find(|(tx_id, _)| *tx_id == rec.tx_id)
454 {
455 rows.push(index);
456 } else {
457 pending_tx_spans.push((rec.tx_id, vec![index]));
458 }
459 }
460 WalRecordType::Begin if rec.tx_id != 0 => {
461 pending_tx_spans.push((rec.tx_id, Vec::new()));
462 }
463 WalRecordType::Commit if rec.tx_id != 0 => {
464 if let Some(span_index) = pending_tx_spans
465 .iter()
466 .rposition(|(tx_id, _)| *tx_id == rec.tx_id)
467 {
468 let (_, rows) = pending_tx_spans.remove(span_index);
469 for row_index in rows {
470 committed_row_records[row_index] = true;
471 }
472 }
473 }
474 WalRecordType::Rollback if rec.tx_id != 0 => {
475 if let Some(span_index) = pending_tx_spans
476 .iter()
477 .rposition(|(tx_id, _)| *tx_id == rec.tx_id)
478 {
479 pending_tx_spans.remove(span_index);
480 }
481 }
482 _ => {}
483 }
484 }
485 }
486
487 let mut replayed_inserts = 0usize;
488 let mut replayed_updates = 0usize;
489 let mut replayed_deletes = 0usize;
490 let mut skipped = 0usize;
491 let mut skipped_uncommitted = 0usize;
492 let mut saw_ddl = false;
493 for (index, rec) in records.iter().enumerate() {
494 if has_boundaries
495 && !committed_row_records[index]
496 && matches!(
497 rec.record_type,
498 WalRecordType::Insert
499 | WalRecordType::Update
500 | WalRecordType::Delete
501 | WalRecordType::OverflowWrite
502 | WalRecordType::OverflowFree
503 )
504 {
505 skipped_uncommitted += 1;
506 continue;
507 }
508 match rec.record_type {
509 WalRecordType::Insert => {
510 if let Some((table_name, rid, row_bytes)) = decode_wal_payload(&rec.data) {
511 if let Some(slot) = self.name_to_slot.get(&table_name).copied() {
512 let tbl = &mut self.tables[slot];
513 if rec.lsn > 0 && tbl.heap.page_lsn(rid.page_id) >= rec.lsn {
517 skipped += 1;
518 continue;
519 }
520 tbl.heap.insert_at(rid, &row_bytes)?;
527 tbl.heap.set_page_lsn(rid.page_id, rec.lsn)?;
528 replayed_inserts += 1;
529 }
530 }
531 }
532 WalRecordType::Update => {
533 if let Some((table_name, rid, row_bytes)) = decode_wal_payload(&rec.data) {
534 if let Some(slot) = self.name_to_slot.get(&table_name).copied() {
535 let tbl = &mut self.tables[slot];
536 if rec.lsn > 0 && tbl.heap.page_lsn(rid.page_id) >= rec.lsn {
537 skipped += 1;
538 continue;
539 }
540 let new_rid = tbl.heap.update(rid, &row_bytes)?;
541 tbl.heap.set_page_lsn(new_rid.page_id, rec.lsn)?;
542 replayed_updates += 1;
543 }
544 }
545 }
546 WalRecordType::Delete => {
547 if let Some((table_name, rid, _)) = decode_wal_payload(&rec.data) {
548 if let Some(slot) = self.name_to_slot.get(&table_name).copied() {
549 let tbl = &mut self.tables[slot];
550 if rec.lsn > 0 && tbl.heap.page_lsn(rid.page_id) >= rec.lsn {
551 skipped += 1;
552 continue;
553 }
554 let _ = tbl.heap.delete(rid);
555 tbl.heap.set_page_lsn(rid.page_id, rec.lsn)?;
556 replayed_deletes += 1;
557 }
558 }
559 }
560 WalRecordType::OverflowWrite => {
561 if let Some((table_name, page_id, next_page, chunk)) =
566 decode_overflow_write_payload(&rec.data)
567 {
568 if let Some(slot) = self.name_to_slot.get(&table_name).copied() {
569 let tbl = &mut self.tables[slot];
570 if rec.lsn > 0 && tbl.heap.overflow_page_lsn(page_id) >= rec.lsn {
571 skipped += 1;
572 continue;
573 }
574 tbl.heap
575 .write_overflow_page(page_id, next_page, &chunk, rec.lsn)?;
576 }
577 }
578 }
579 WalRecordType::OverflowFree => {
580 if let Some((table_name, pages)) = decode_overflow_free_payload(&rec.data) {
585 if let Some(slot) = self.name_to_slot.get(&table_name).copied() {
586 self.tables[slot].heap.release_overflow_pages(&pages);
587 }
588 }
589 }
590 WalRecordType::Begin | WalRecordType::Commit | WalRecordType::Rollback => {
591 }
593 WalRecordType::DdlCreateTable => {
594 saw_ddl = true;
595 if let Some((schema, defaults, auto_cols)) = decode_ddl_create_table(&rec.data)
596 {
597 if !self.name_to_slot.contains_key(&schema.table_name) {
598 if let Ok(mut table) = Table::create(schema, &self.data_dir) {
599 table.set_defaults(defaults);
600 table.set_auto_cols(auto_cols);
601 let slot = self.tables.len();
602 let name = table.schema.table_name.clone();
603 self.tables.push(table);
604 self.name_to_slot.insert(name, slot);
605 }
606 }
607 }
608 }
609 WalRecordType::DdlDropTable => {
610 saw_ddl = true;
611 if let Some((table_name, _)) = decode_ddl_table_name(&rec.data) {
612 if let Some(&slot) = self.name_to_slot.get(&table_name) {
613 let heap_path = self.data_dir.join(format!("{table_name}.heap"));
614 if heap_path.exists() {
615 let _ = fs::remove_file(&heap_path);
616 }
617 for col_name in self.tables[slot].indexed_column_names() {
618 let idx_path =
619 self.data_dir.join(format!("{table_name}_{col_name}.idx"));
620 if idx_path.exists() {
621 let _ = fs::remove_file(&idx_path);
622 }
623 }
624 self.name_to_slot.remove(&table_name);
625 let last = self.tables.len() - 1;
626 if slot != last {
627 let moved_name = self.tables[last].schema.table_name.clone();
628 self.tables.swap(slot, last);
629 self.name_to_slot.insert(moved_name, slot);
630 }
631 self.tables.pop();
632 }
633 }
634 }
635 WalRecordType::DdlAddColumn => {
636 saw_ddl = true;
637 if let Some((table_name, col)) = decode_ddl_alter_add_column(&rec.data) {
638 if let Some(&slot) = self.name_to_slot.get(&table_name) {
639 let tbl = &mut self.tables[slot];
640 if !tbl.schema.columns.iter().any(|c| c.name == col.name) {
641 let old_schema = tbl.schema.clone();
642 let has_rows = tbl.heap.scan().next().is_some();
643 tbl.schema.columns.push(col);
644 tbl.refresh_layout();
645 if has_rows {
646 let fill = vec![Value::Empty; tbl.schema.columns.len()];
647 let data_dir = self.data_dir.clone();
648 let _ = tbl.rewrite_rows_for_schema_change(
649 &old_schema,
650 &fill,
651 &data_dir,
652 );
653 }
654 }
655 if rec.lsn > 0 {
662 let _ = tbl.heap.stamp_all_pages_min_lsn(rec.lsn);
663 }
664 }
665 }
666 }
667 WalRecordType::DdlDropColumn => {
668 saw_ddl = true;
669 if let Some((table_name, col_name)) = decode_ddl_alter_drop_column(&rec.data) {
670 if let Some(&slot) = self.name_to_slot.get(&table_name) {
671 let tbl = &mut self.tables[slot];
672 if let Some(idx) =
673 tbl.schema.columns.iter().position(|c| c.name == col_name)
674 {
675 let old_schema = tbl.schema.clone();
676 let has_rows = tbl.heap.scan().next().is_some();
677 tbl.schema.columns.remove(idx);
678 for (i, c) in tbl.schema.columns.iter_mut().enumerate() {
679 c.position = i as u16;
680 }
681 tbl.refresh_layout();
682 if has_rows {
683 let fill = vec![Value::Empty; tbl.schema.columns.len()];
684 let data_dir = self.data_dir.clone();
685 let _ = tbl.rewrite_rows_for_schema_change(
686 &old_schema,
687 &fill,
688 &data_dir,
689 );
690 }
691 }
692 if rec.lsn > 0 {
693 let _ = tbl.heap.stamp_all_pages_min_lsn(rec.lsn);
694 }
695 }
696 }
697 }
698 }
699 }
700 info!(
701 inserts = replayed_inserts,
702 updates = replayed_updates,
703 deletes = replayed_deletes,
704 skipped = skipped,
705 skipped_uncommitted = skipped_uncommitted,
706 "WAL record apply complete (commit-boundary + LSN idempotent)"
707 );
708 if saw_ddl {
709 self.persist()?;
710 }
711 for tbl in &mut self.tables {
725 tbl.heap.flush_all_dirty()?;
726 tbl.heap.flush()?;
727 tbl.rebuild_indexes_from_heap()?;
728 tbl.save_dirty_indexes()?;
733 }
734 if let Some(max_lsn) = max_record_lsn(records) {
735 self.record_durable_lsn_at_least(max_lsn)?;
736 self.wal.set_next_lsn_at_least(max_lsn.saturating_add(1));
737 }
738 Ok(())
739 }
740
741 pub fn checkpoint(&mut self) -> io::Result<()> {
750 self.ensure_no_active_transaction_for_checkpoint()?;
751 self.ensure_plain_checkpoint_allowed_before_flush()?;
752 self.flush_checkpoint_state()?;
753 self.wal.flush()?;
754 self.record_durable_lsn_at_least(self.wal.last_appended_lsn())?;
755 self.wal.truncate()?;
756 self.checkpointed = true;
757 Ok(())
758 }
759
760 pub fn checkpoint_with_wal_archive<F>(&mut self, mut archive: F) -> io::Result<()>
768 where
769 F: FnMut(&Path, &[WalRecord]) -> io::Result<()>,
770 {
771 self.ensure_no_active_transaction_for_checkpoint()?;
772 self.commit_autocommit()?;
773 self.flush_checkpoint_state()?;
774 self.wal.flush()?;
775 let records = self.wal.read_all()?;
776 let archive: WalArchiveCallback<'_> = &mut archive;
777 archive(&self.data_dir, &records)?;
778 if let Some(max_lsn) = max_record_lsn(&records) {
779 self.record_durable_lsn_at_least(max_lsn)?;
780 } else {
781 self.record_durable_lsn_at_least(self.wal.last_appended_lsn())?;
782 }
783 self.wal.truncate()?;
784 self.checkpointed = true;
785 Ok(())
786 }
787
788 fn ensure_no_active_transaction_for_checkpoint(&self) -> io::Result<()> {
789 if self.active_tx_id.is_some() {
790 return Err(io::Error::other(
791 "cannot checkpoint while an explicit transaction is active",
792 ));
793 }
794 Ok(())
795 }
796
797 fn flush_checkpoint_state(&mut self) -> io::Result<()> {
798 for tbl in &mut self.tables {
799 tbl.heap.flush_all_dirty()?;
800 tbl.heap.flush()?;
801 tbl.save_dirty_indexes()?;
806 }
807 Ok(())
808 }
809
810 fn ensure_plain_checkpoint_allowed_before_flush(&self) -> io::Result<()> {
811 if !self.sync_identity_file_exists() {
812 return Ok(());
813 }
814 if self.wal.has_pending() {
815 return Err(io::Error::other(
816 "sync identity exists but checkpoint/recovery was called without a WAL archive hook; refusing to truncate retained history",
817 ));
818 }
819 let records = self.wal.read_all()?;
820 self.ensure_plain_wal_truncate_allowed(&records)
821 }
822
823 fn ensure_plain_wal_truncate_allowed(&self, records: &[WalRecord]) -> io::Result<()> {
824 if records.is_empty() {
825 return Ok(());
826 }
827 if self.sync_identity_file_exists() {
828 return Err(io::Error::other(
829 "sync identity exists but checkpoint/recovery was called without a WAL archive hook; refusing to truncate retained history",
830 ));
831 }
832 Ok(())
833 }
834
835 fn sync_identity_file_exists(&self) -> bool {
836 self.data_dir
837 .join(SYNC_STATE_DIR)
838 .join(SYNC_IDENTITY_FILE)
839 .exists()
840 }
841
842 fn record_durable_lsn_at_least(&mut self, lsn: u64) -> io::Result<()> {
843 if lsn <= self.durable_lsn {
844 return Ok(());
845 }
846 self.durable_lsn = lsn;
847 write_durable_lsn(&self.data_dir, lsn)
848 }
849
850 #[inline]
852 fn free_overflow_chain(&mut self, slot: usize, pages: Vec<u32>) {
863 if pages.is_empty() {
864 return;
865 }
866 if self.active_tx_id.is_some() {
867 self.pending_free_overflow.push((slot, pages));
868 } else {
869 self.tables[slot].release_overflow_pages(&pages);
870 }
871 }
872
873 fn next_tx(&mut self) -> u64 {
874 if let Some(id) = self.active_tx_id {
875 return id;
876 }
877 let id = self.next_tx_id;
878 self.next_tx_id = self.next_tx_id.wrapping_add(1);
879 id
880 }
881
882 pub fn begin_transaction(&mut self) -> io::Result<()> {
884 if self.active_tx_id.is_some() {
885 return Err(io::Error::new(
886 io::ErrorKind::InvalidInput,
887 "explicit transaction is already active",
888 ));
889 }
890 let start_len = self.wal.synced_len()?;
891 let id = self.next_tx_id;
892 self.next_tx_id = self.next_tx_id.wrapping_add(1);
893 self.active_tx_id = Some(id);
894 self.tx_start_len = Some(start_len);
895 self.pending_autocommit_tx_ids.clear();
896 if !self.wal.is_off() {
897 self.wal.append(id, WalRecordType::Begin, &[])?;
898 self.wal.flush()?;
899 }
900 Ok(())
901 }
902
903 pub fn commit_transaction(&mut self) -> io::Result<()> {
906 if let Some(id) = self.active_tx_id.take() {
907 if !self.wal.is_off() {
908 self.wal.append(id, WalRecordType::Commit, &[])?;
909 self.wal.flush()?;
910 }
911 }
912 self.tx_start_len = None;
913 for (slot, pages) in std::mem::take(&mut self.pending_free_overflow) {
917 self.tables[slot].release_overflow_pages(&pages);
918 }
919 Ok(())
920 }
921
922 pub fn commit_autocommit(&mut self) -> io::Result<()> {
926 if !self.wal.is_off() && !self.pending_autocommit_tx_ids.is_empty() {
927 self.pending_autocommit_tx_ids.sort_unstable();
928 self.pending_autocommit_tx_ids.dedup();
929 for id in self.pending_autocommit_tx_ids.drain(..) {
930 self.wal.append(id, WalRecordType::Commit, &[])?;
931 }
932 }
933 self.wal.flush()
934 }
935
936 fn wal_log(
951 &mut self,
952 tx_id: u64,
953 record_type: WalRecordType,
954 table: &str,
955 rid: RowId,
956 row_bytes: &[u8],
957 ) -> io::Result<()> {
958 if self.wal.is_off() {
965 return Ok(());
966 }
967 let payload = encode_wal_payload(table, rid, row_bytes);
968 self.wal.append(tx_id, record_type, &payload)?;
969 if self.active_tx_id.is_none() {
970 self.pending_autocommit_tx_ids.push(tx_id);
971 }
972 Ok(())
973 }
974
975 #[inline]
981 pub fn sync_wal(&mut self) -> io::Result<()> {
982 self.wal.flush()
983 }
984
985 pub fn set_wal_sync_mode(&mut self, mode: WalSyncMode) {
993 self.wal.set_sync_mode(mode);
994 }
995
996 pub fn set_wal_sync_deferred(&mut self, defer: bool) {
1004 self.wal.set_defer_sync(defer);
1005 }
1006
1007 pub fn take_wal_durability_ticket(&mut self) -> Option<WalDurabilityTicket> {
1010 self.wal.take_durability_ticket()
1011 }
1012
1013 pub fn wal_fsync_count(&self) -> u64 {
1015 self.wal.fsync_count()
1016 }
1017
1018 pub fn rollback_to_last_sync(&mut self) -> io::Result<()> {
1033 self.rollback_to_last_sync_inner(None)
1034 }
1035
1036 pub fn rollback_to_last_sync_with_wal_archive<F>(&mut self, mut archive: F) -> io::Result<()>
1041 where
1042 F: FnMut(&Path, &[WalRecord]) -> io::Result<()>,
1043 {
1044 let archive: WalArchiveCallback<'_> = &mut archive;
1045 self.rollback_to_last_sync_inner(Some(archive))
1046 }
1047
1048 fn rollback_to_last_sync_inner(
1049 &mut self,
1050 mut archive: Option<WalArchiveCallback<'_>>,
1051 ) -> io::Result<()> {
1052 let start_len = self.tx_start_len.unwrap_or(0);
1053 let prearchived = if let Some(archive) = archive.as_mut() {
1054 let records = self.wal.read_through_len(start_len)?;
1055 if !records.is_empty() {
1056 archive(&self.data_dir, &records)?;
1057 }
1058 true
1059 } else {
1060 false
1061 };
1062
1063 let start_len = self.tx_start_len.take().unwrap_or(0);
1064 if let Some(id) = self.active_tx_id.take() {
1065 if !self.wal.is_off() {
1066 let _ = self.wal.append(id, WalRecordType::Rollback, &[]);
1067 }
1068 }
1069 self.wal.discard_and_truncate_to(start_len)?;
1070
1071 for tbl in &mut self.tables {
1081 tbl.heap.discard_dirty();
1082 tbl.discard_dirty_indexes();
1083 }
1084 self.wal.discard_pending()?;
1090 let data_dir = self.data_dir.clone();
1094 let sync_mode = self.wal.sync_mode();
1095 let restored = if prearchived {
1096 let mut already_archived = |_dir: &Path, _records: &[WalRecord]| Ok(());
1097 let archive: WalArchiveCallback<'_> = &mut already_archived;
1098 Self::open_inner(&data_dir, Some(archive))?
1099 } else {
1100 Self::open_inner(&data_dir, archive)?
1101 };
1102 *self = restored;
1103 self.wal.set_sync_mode(sync_mode);
1104 Ok(())
1105 }
1106
1107 fn abandon_active_transaction_for_drop(&mut self) -> io::Result<()> {
1108 for tbl in &mut self.tables {
1109 tbl.heap.discard_dirty();
1110 }
1111 self.pending_autocommit_tx_ids.clear();
1112 let truncate_result = match self.tx_start_len.take() {
1113 Some(start_len) => self.wal.discard_and_truncate_to(start_len),
1114 None => self.wal.discard_pending(),
1115 };
1116 self.active_tx_id = None;
1117 truncate_result
1118 }
1119
1120 pub fn data_dir(&self) -> &Path {
1122 &self.data_dir
1123 }
1124
1125 pub fn max_lsn(&self) -> u64 {
1130 let max_page_lsn = self
1131 .tables
1132 .iter()
1133 .map(|t| t.heap.max_page_lsn())
1134 .max()
1135 .unwrap_or(0);
1136 max_page_lsn
1137 .max(self.durable_lsn)
1138 .max(self.wal.last_appended_lsn())
1139 }
1140
1141 pub fn create_table(&mut self, schema: Schema) -> io::Result<()> {
1142 self.create_table_full(schema, Vec::new(), Vec::new())
1143 }
1144
1145 pub fn create_table_with_defaults(
1149 &mut self,
1150 schema: Schema,
1151 defaults: Vec<Option<Value>>,
1152 ) -> io::Result<()> {
1153 self.create_table_full(schema, defaults, Vec::new())
1154 }
1155
1156 pub fn create_table_full(
1161 &mut self,
1162 schema: Schema,
1163 defaults: Vec<Option<Value>>,
1164 auto_cols: Vec<bool>,
1165 ) -> io::Result<()> {
1166 validate_table_name(&schema.table_name)?;
1167 for col in &schema.columns {
1168 validate_column_name(&col.name)?;
1169 }
1170 let name = schema.table_name.clone();
1171 if self.name_to_slot.contains_key(&name) {
1172 return Err(io::Error::new(
1173 io::ErrorKind::AlreadyExists,
1174 format!("table '{name}' already exists"),
1175 ));
1176 }
1177 if !self.wal.is_off() {
1178 let payload = encode_ddl_create_table(&schema, &defaults, &auto_cols);
1179 self.wal
1180 .append(0, WalRecordType::DdlCreateTable, &payload)?;
1181 self.wal.flush()?;
1182 }
1183 let mut table = Table::create(schema, &self.data_dir)?;
1184 table.set_defaults(defaults);
1185 table.set_auto_cols(auto_cols);
1186 let slot = self.tables.len();
1187 self.tables.push(table);
1188 self.name_to_slot.insert(name, slot);
1189 self.persist()?;
1190 Ok(())
1191 }
1192
1193 pub fn column_defaults(&self, table: &str) -> Option<&[Option<Value>]> {
1197 let slot = *self.name_to_slot.get(table)?;
1198 Some(self.tables[slot].defaults())
1199 }
1200
1201 pub fn auto_columns(&self, table: &str) -> Option<&[bool]> {
1204 let slot = *self.name_to_slot.get(table)?;
1205 Some(self.tables[slot].auto_cols())
1206 }
1207
1208 pub fn assign_auto_columns(&mut self, table: &str, values: &mut [Value]) {
1212 if let Some(&slot) = self.name_to_slot.get(table) {
1213 self.tables[slot].assign_auto(values);
1214 }
1215 }
1216
1217 fn persist(&self) -> io::Result<()> {
1222 let cat_path = self.data_dir.join(CATALOG_FILE);
1223 let tmp_path = self.data_dir.join(format!("{CATALOG_FILE}.tmp"));
1224 let entries: Vec<CatalogEntryRef<'_>> = self
1225 .tables
1226 .iter()
1227 .map(|t| CatalogEntryRef {
1228 schema: &t.schema,
1229 indexed_cols: t.indexed_column_metas(),
1230 defaults: t.defaults(),
1231 auto_cols: t.auto_cols(),
1232 })
1233 .collect();
1234 write_catalog_file(&tmp_path, &entries)?;
1235 fs::rename(&tmp_path, &cat_path)?;
1236 Ok(())
1237 }
1238
1239 #[inline]
1243 pub fn table_slot(&self, name: &str) -> Option<usize> {
1244 self.name_to_slot.get(name).copied()
1245 }
1246
1247 #[inline]
1250 pub fn table_by_slot(&self, slot: usize) -> &Table {
1251 &self.tables[slot]
1252 }
1253
1254 #[inline]
1256 pub fn table_by_slot_mut(&mut self, slot: usize) -> &mut Table {
1257 &mut self.tables[slot]
1258 }
1259
1260 pub fn get_table(&self, name: &str) -> Option<&Table> {
1261 let slot = *self.name_to_slot.get(name)?;
1262 Some(&self.tables[slot])
1263 }
1264
1265 pub fn get_table_mut(&mut self, name: &str) -> Option<&mut Table> {
1266 let slot = *self.name_to_slot.get(name)?;
1267 Some(&mut self.tables[slot])
1268 }
1269
1270 #[inline]
1274 pub fn table_has_overflow(&self, table: &str) -> bool {
1275 self.get_table(table)
1276 .map(|t| t.has_overflow_rows())
1277 .unwrap_or(false)
1278 }
1279
1280 #[inline]
1285 fn by_name(&self, table: &str) -> io::Result<&Table> {
1286 let slot = *self.name_to_slot.get(table).ok_or_else(|| {
1287 io::Error::new(
1288 io::ErrorKind::NotFound,
1289 format!("table '{table}' not found"),
1290 )
1291 })?;
1292 Ok(&self.tables[slot])
1293 }
1294
1295 #[inline]
1297 fn by_name_mut(&mut self, table: &str) -> io::Result<&mut Table> {
1298 let slot = self.slot_of(table)?;
1299 Ok(&mut self.tables[slot])
1300 }
1301
1302 pub fn sweep(&mut self, table: &str) -> io::Result<usize> {
1307 let slot = self.slot_of(table)?;
1308 let reclaimed = self.tables[slot].sweep_overflow()?;
1309 if !reclaimed.is_empty() && !self.wal.is_off() {
1310 let payload = encode_overflow_free_payload(table, &reclaimed);
1311 self.wal.append(0, WalRecordType::OverflowFree, &payload)?;
1312 self.wal.flush()?;
1313 }
1314 Ok(reclaimed.len())
1315 }
1316
1317 pub fn sweep_all(&mut self) -> io::Result<usize> {
1319 let names: Vec<String> = self
1320 .tables
1321 .iter()
1322 .map(|t| t.schema.table_name.clone())
1323 .collect();
1324 let mut total = 0;
1325 for name in names {
1326 total += self.sweep(&name)?;
1327 }
1328 Ok(total)
1329 }
1330
1331 fn slot_of(&self, table: &str) -> io::Result<usize> {
1332 self.name_to_slot.get(table).copied().ok_or_else(|| {
1333 io::Error::new(
1334 io::ErrorKind::NotFound,
1335 format!("table '{table}' not found"),
1336 )
1337 })
1338 }
1339
1340 pub fn insert(&mut self, table: &str, values: &Row) -> io::Result<RowId> {
1341 if self.wal.is_off() {
1350 return self.by_name_mut(table)?.insert(values);
1351 }
1352 let tx_id = self.next_tx();
1356 let slot = self.slot_of(table)?;
1357 let row_bytes = {
1358 let Catalog { tables, wal, .. } = self;
1359 encode_row_with_spill_logged(&mut tables[slot], wal, tx_id, values)?
1360 };
1361 let new_rid = self.tables[slot].insert_encoded(values, &row_bytes)?;
1366 self.wal_log(tx_id, WalRecordType::Insert, table, new_rid, &row_bytes)?;
1367 let lsn = self.wal.last_appended_lsn();
1368 if lsn > 0 {
1369 self.tables[slot].heap.set_page_lsn(new_rid.page_id, lsn)?;
1370 }
1371 Ok(new_rid)
1372 }
1373
1374 pub fn insert_by_slot(&mut self, slot: usize, values: &Row) -> io::Result<RowId> {
1382 if self.wal.is_off() {
1383 return self.tables[slot].insert(values);
1384 }
1385 let tx_id = self.next_tx();
1386 let autocommit = self.active_tx_id.is_none();
1387 let Catalog { tables, wal, .. } = self;
1388 let tbl = &mut tables[slot];
1389 let row_bytes = encode_row_with_spill_logged(tbl, wal, tx_id, values)?;
1392 let new_rid = tbl.insert_encoded(values, &row_bytes)?;
1395 let payload = encode_wal_payload(&tbl.schema.table_name, new_rid, &row_bytes);
1396 wal.append(tx_id, WalRecordType::Insert, &payload)?;
1397 if autocommit {
1398 self.pending_autocommit_tx_ids.push(tx_id);
1399 }
1400 let lsn = wal.last_appended_lsn();
1401 if lsn > 0 {
1402 tbl.heap.set_page_lsn(new_rid.page_id, lsn)?;
1403 }
1404 Ok(new_rid)
1405 }
1406
1407 pub fn get(&self, table: &str, rid: RowId) -> Option<Row> {
1408 self.get_table(table)?.get(rid)
1409 }
1410
1411 pub fn delete(&mut self, table: &str, rid: RowId) -> io::Result<()> {
1412 let slot = self.slot_of(table)?;
1413 let old_pages = self.tables[slot].overflow_chain_pages_at(rid)?;
1417 if self.wal.is_off() {
1420 self.tables[slot].delete(rid)?;
1421 self.free_overflow_chain(slot, old_pages);
1422 return Ok(());
1423 }
1424 let tx_id = self.next_tx();
1425 self.wal_log(tx_id, WalRecordType::Delete, table, rid, &[])?;
1427 self.tables[slot].delete(rid)?;
1428 self.free_overflow_chain(slot, old_pages);
1429 Ok(())
1430 }
1431
1432 pub fn delete_many(&mut self, table: &str, rids: &[RowId]) -> io::Result<u64> {
1436 let slot = self.slot_of(table)?;
1444 let old_pages = self.collect_overflow_pages(slot, rids)?;
1447 if self.wal.is_off() {
1448 let count = self.tables[slot].delete_many(rids)?;
1449 self.free_overflow_chain(slot, old_pages);
1450 return Ok(count);
1451 }
1452 let tx_id = self.next_tx();
1453 for &rid in rids {
1454 let payload = encode_wal_payload(table, rid, &[]);
1455 self.wal.append(tx_id, WalRecordType::Delete, &payload)?;
1456 }
1457 if self.active_tx_id.is_none() && !rids.is_empty() {
1458 self.pending_autocommit_tx_ids.push(tx_id);
1459 }
1460 let count = self.tables[slot].delete_many(rids)?;
1461 self.free_overflow_chain(slot, old_pages);
1462 Ok(count)
1463 }
1464
1465 fn collect_overflow_pages(&self, slot: usize, rids: &[RowId]) -> io::Result<Vec<u32>> {
1468 if !self.tables[slot].has_overflow_rows() {
1469 return Ok(Vec::new());
1470 }
1471 let mut pages = Vec::new();
1472 for &rid in rids {
1473 pages.extend(self.tables[slot].overflow_chain_pages_at(rid)?);
1474 }
1475 Ok(pages)
1476 }
1477
1478 pub fn scan_delete_matching<P>(&mut self, table: &str, pred: P) -> io::Result<u64>
1488 where
1489 P: FnMut(&[u8]) -> bool,
1490 {
1491 self.by_name_mut(table)?.scan_delete_matching(pred)
1492 }
1493
1494 pub fn scan_delete_matching_logged<P>(&mut self, table: &str, pred: P) -> io::Result<u64>
1505 where
1506 P: FnMut(&[u8]) -> bool,
1507 {
1508 if self.wal.is_off() {
1514 return self.by_name_mut(table)?.scan_delete_matching(pred);
1515 }
1516 let slot = *self.name_to_slot.get(table).ok_or_else(|| {
1520 io::Error::new(
1521 io::ErrorKind::NotFound,
1522 format!("table '{table}' not found"),
1523 )
1524 })?;
1525 let tx_id = self.next_tx();
1526 let autocommit = self.active_tx_id.is_none();
1527 let Catalog { tables, wal, .. } = self;
1530 let tbl = &mut tables[slot];
1531 let name_bytes = table.as_bytes();
1535 let count = tbl.scan_delete_matching_with_hook(pred, |rid, row_bytes| {
1536 let mut payload: Vec<u8> =
1537 Vec::with_capacity(4 + name_bytes.len() + 10 + row_bytes.len());
1538 payload.extend_from_slice(&(name_bytes.len() as u32).to_le_bytes());
1539 payload.extend_from_slice(name_bytes);
1540 payload.extend_from_slice(&rid.page_id.to_le_bytes());
1541 payload.extend_from_slice(&rid.slot_index.to_le_bytes());
1542 payload.extend_from_slice(&0u32.to_le_bytes());
1546 let _ = wal.append(tx_id, WalRecordType::Delete, &payload);
1553 })?;
1554 if autocommit && count > 0 {
1555 self.pending_autocommit_tx_ids.push(tx_id);
1556 }
1557 Ok(count)
1559 }
1560
1561 pub fn scan_patch_matching_logged<P, M>(
1569 &mut self,
1570 table: &str,
1571 pred: P,
1572 try_mutate: M,
1573 ) -> io::Result<(u64, Vec<RowId>)>
1574 where
1575 P: FnMut(&[u8]) -> bool,
1576 M: FnMut(&mut [u8]) -> Option<u16>,
1577 {
1578 if self.wal.is_off() {
1579 return self.by_name_mut(table)?.scan_patch_matching_with_hook(
1580 pred,
1581 try_mutate,
1582 |_, _| {},
1583 );
1584 }
1585 let slot = *self.name_to_slot.get(table).ok_or_else(|| {
1586 io::Error::new(
1587 io::ErrorKind::NotFound,
1588 format!("table '{table}' not found"),
1589 )
1590 })?;
1591 let tx_id = self.next_tx();
1592 let autocommit = self.active_tx_id.is_none();
1593 let Catalog { tables, wal, .. } = self;
1594 let tbl = &mut tables[slot];
1595 let name_bytes = table.as_bytes();
1596 let result = tbl.scan_patch_matching_with_hook(pred, try_mutate, |rid, row_bytes| {
1597 let mut payload: Vec<u8> =
1598 Vec::with_capacity(4 + name_bytes.len() + 10 + row_bytes.len());
1599 payload.extend_from_slice(&(name_bytes.len() as u32).to_le_bytes());
1600 payload.extend_from_slice(name_bytes);
1601 payload.extend_from_slice(&rid.page_id.to_le_bytes());
1602 payload.extend_from_slice(&rid.slot_index.to_le_bytes());
1603 payload.extend_from_slice(&(row_bytes.len() as u32).to_le_bytes());
1604 payload.extend_from_slice(row_bytes);
1605 let _ = wal.append(tx_id, WalRecordType::Update, &payload);
1606 })?;
1607 if autocommit && result.0 > 0 {
1608 self.pending_autocommit_tx_ids.push(tx_id);
1609 }
1610 Ok(result)
1611 }
1612
1613 pub fn update(&mut self, table: &str, rid: RowId, values: &Row) -> io::Result<RowId> {
1614 if self.wal.is_off() {
1617 let slot = self.slot_of(table)?;
1618 let old_pages = self.tables[slot].overflow_chain_pages_at(rid)?;
1619 let new_rid = self.tables[slot].update(rid, values)?;
1620 self.free_overflow_chain(slot, old_pages);
1621 return Ok(new_rid);
1622 }
1623 let tx_id = self.next_tx();
1624 let slot = self.slot_of(table)?;
1625 let old_pages = self.tables[slot].overflow_chain_pages_at(rid)?;
1629 let row_bytes = {
1634 let Catalog { tables, wal, .. } = self;
1635 encode_row_with_spill_logged(&mut tables[slot], wal, tx_id, values)?
1636 };
1637 check_encoded_row_size(&row_bytes)?;
1641 self.wal_log(tx_id, WalRecordType::Update, table, rid, &row_bytes)?;
1642 let new_rid = self.tables[slot].update_encoded(rid, values, &row_bytes, None)?;
1643 self.free_overflow_chain(slot, old_pages);
1644 Ok(new_rid)
1645 }
1646
1647 pub fn update_hinted(
1651 &mut self,
1652 table: &str,
1653 rid: RowId,
1654 values: &Row,
1655 changed_col_indices: Option<&[usize]>,
1656 ) -> io::Result<RowId> {
1657 if self.wal.is_off() {
1661 let slot = self.slot_of(table)?;
1662 let old_pages = self.tables[slot].overflow_chain_pages_at(rid)?;
1663 let new_rid = self.tables[slot].update_hinted(rid, values, changed_col_indices)?;
1664 self.free_overflow_chain(slot, old_pages);
1665 return Ok(new_rid);
1666 }
1667 let tx_id = self.next_tx();
1668 let slot = self.slot_of(table)?;
1669 let old_pages = self.tables[slot].overflow_chain_pages_at(rid)?;
1670 let row_bytes = {
1671 let Catalog { tables, wal, .. } = self;
1672 encode_row_with_spill_logged(&mut tables[slot], wal, tx_id, values)?
1673 };
1674 check_encoded_row_size(&row_bytes)?;
1676 self.wal_log(tx_id, WalRecordType::Update, table, rid, &row_bytes)?;
1677 let new_rid =
1678 self.tables[slot].update_encoded(rid, values, &row_bytes, changed_col_indices)?;
1679 self.free_overflow_chain(slot, old_pages);
1680 Ok(new_rid)
1681 }
1682
1683 #[inline]
1695 pub fn with_row_bytes_mut<F>(&mut self, table: &str, rid: RowId, f: F) -> io::Result<bool>
1696 where
1697 F: FnOnce(&mut [u8]),
1698 {
1699 self.by_name_mut(table)?.with_row_bytes_mut(rid, f)
1700 }
1701
1702 #[inline]
1716 pub fn update_row_bytes_logged<F>(&mut self, table: &str, rid: RowId, f: F) -> io::Result<bool>
1717 where
1718 F: FnOnce(&mut [u8]),
1719 {
1720 let slot = *self.name_to_slot.get(table).ok_or_else(|| {
1721 io::Error::new(
1722 io::ErrorKind::NotFound,
1723 format!("table '{table}' not found"),
1724 )
1725 })?;
1726 self.update_row_bytes_logged_by_slot(slot, rid, f)
1727 }
1728
1729 #[inline]
1734 pub fn update_row_bytes_logged_by_slot<F>(
1735 &mut self,
1736 slot: usize,
1737 rid: RowId,
1738 f: F,
1739 ) -> io::Result<bool>
1740 where
1741 F: FnOnce(&mut [u8]),
1742 {
1743 let tbl = &mut self.tables[slot];
1746 let ok = tbl.with_row_bytes_mut(rid, f)?;
1747 if !ok {
1748 return Ok(false);
1749 }
1750 if self.wal.is_off() {
1754 return Ok(true);
1755 }
1756 let new_bytes = match tbl.heap.get(rid) {
1759 Some(b) => b,
1760 None => return Ok(false),
1762 };
1763 let table_name = tbl.schema.table_name.clone();
1766 let tx_id = self.next_tx();
1767 self.wal_log(tx_id, WalRecordType::Update, &table_name, rid, &new_bytes)?;
1768 Ok(true)
1769 }
1770
1771 #[inline]
1783 pub fn patch_var_col_in_place(
1784 &mut self,
1785 table: &str,
1786 rid: RowId,
1787 col_idx: usize,
1788 new_value: Option<&[u8]>,
1789 ) -> io::Result<bool> {
1790 self.by_name_mut(table)?
1791 .patch_var_col_in_place(rid, col_idx, new_value)
1792 }
1793
1794 pub fn patch_var_col_logged(
1800 &mut self,
1801 table: &str,
1802 rid: RowId,
1803 col_idx: usize,
1804 new_value: Option<&[u8]>,
1805 ) -> io::Result<bool> {
1806 let slot = *self.name_to_slot.get(table).ok_or_else(|| {
1807 io::Error::new(
1808 io::ErrorKind::NotFound,
1809 format!("table '{table}' not found"),
1810 )
1811 })?;
1812 let tbl = &mut self.tables[slot];
1813 let ok = tbl.patch_var_col_in_place(rid, col_idx, new_value)?;
1814 if !ok {
1815 return Ok(false);
1816 }
1817 if self.wal.is_off() {
1820 return Ok(true);
1821 }
1822 let new_bytes = match tbl.heap.get(rid) {
1823 Some(b) => b,
1824 None => return Ok(false),
1825 };
1826 let table_name = tbl.schema.table_name.clone();
1827 let tx_id = self.next_tx();
1828 self.wal_log(tx_id, WalRecordType::Update, &table_name, rid, &new_bytes)?;
1829 Ok(true)
1830 }
1831
1832 pub fn scan(&self, table: &str) -> io::Result<impl Iterator<Item = (RowId, Row)> + '_> {
1833 Ok(self.by_name(table)?.scan())
1834 }
1835
1836 pub fn for_each_row_raw<F>(&self, table: &str, f: F) -> io::Result<()>
1839 where
1840 F: FnMut(RowId, &[u8]),
1841 {
1842 self.by_name(table)?.for_each_row_raw(f);
1843 Ok(())
1844 }
1845
1846 pub fn try_for_each_row_raw<F>(&self, table: &str, f: F) -> io::Result<()>
1851 where
1852 F: FnMut(RowId, &[u8]) -> std::ops::ControlFlow<()>,
1853 {
1854 self.by_name(table)?.try_for_each_row_raw(f);
1855 Ok(())
1856 }
1857
1858 pub fn create_index(&mut self, table: &str, column: &str) -> io::Result<()> {
1859 self.create_index_unique(table, column, false)
1860 }
1861
1862 pub fn create_index_unique(
1867 &mut self,
1868 table: &str,
1869 column: &str,
1870 unique: bool,
1871 ) -> io::Result<()> {
1872 let data_dir = self.data_dir.clone();
1873 self.by_name_mut(table)?
1874 .create_index_with_unique(column, &data_dir, unique)?;
1875 self.persist()
1879 }
1880
1881 pub fn is_index_unique(&self, table: &str, column: &str) -> Option<bool> {
1885 self.get_table(table)?.is_index_unique(column)
1886 }
1887
1888 pub fn has_index(&self, table: &str, column: &str) -> bool {
1890 self.get_table(table)
1891 .map(|t| t.has_index(column))
1892 .unwrap_or(false)
1893 }
1894
1895 pub fn index_lookup(&self, table: &str, column: &str, key: &Value) -> io::Result<Option<Row>> {
1896 Ok(self
1897 .by_name(table)?
1898 .index_lookup(column, key)
1899 .map(|(_, row)| row))
1900 }
1901
1902 pub fn list_tables(&self) -> Vec<&str> {
1903 self.tables
1907 .iter()
1908 .map(|t| t.schema.table_name.as_str())
1909 .collect()
1910 }
1911
1912 pub fn schema(&self, table: &str) -> Option<&Schema> {
1913 let slot = *self.name_to_slot.get(table)?;
1914 Some(&self.tables[slot].schema)
1915 }
1916
1917 pub fn drop_table(&mut self, name: &str) -> io::Result<()> {
1920 validate_table_name(name)?;
1921 let slot = *self.name_to_slot.get(name).ok_or_else(|| {
1922 io::Error::new(io::ErrorKind::NotFound, format!("table '{name}' not found"))
1923 })?;
1924 if !self.wal.is_off() {
1925 let payload = encode_ddl_drop_table(name);
1926 self.wal.append(0, WalRecordType::DdlDropTable, &payload)?;
1927 self.wal.flush()?;
1928 }
1929 let table = &self.tables[slot];
1931 let heap_path = self
1932 .data_dir
1933 .join(format!("{}.heap", table.schema.table_name));
1934 if heap_path.exists() {
1935 fs::remove_file(&heap_path)?;
1936 }
1937 for col_name in table.indexed_column_names() {
1943 let idx_path = self.data_dir.join(format!("{name}_{col_name}.idx"));
1944 if idx_path.exists() {
1945 let _ = fs::remove_file(&idx_path);
1946 }
1947 }
1948 self.name_to_slot.remove(name);
1950 let last = self.tables.len() - 1;
1951 if slot != last {
1952 let moved_name = self.tables[last].schema.table_name.clone();
1953 self.tables.swap(slot, last);
1954 self.name_to_slot.insert(moved_name, slot);
1955 }
1956 self.tables.pop();
1957 self.persist()?;
1958 Ok(())
1959 }
1960
1961 pub fn alter_table_add_column(&mut self, table: &str, col: ColumnDef) -> io::Result<()> {
1984 let data_dir = self.data_dir.clone();
1985 {
1986 let tbl = self.by_name_mut(table)?;
1987 if tbl.schema.columns.iter().any(|c| c.name == col.name) {
1988 return Err(io::Error::new(
1989 io::ErrorKind::AlreadyExists,
1990 format!("column '{}' already exists in table '{table}'", col.name),
1991 ));
1992 }
1993 }
1994 let barrier_lsn = if !self.wal.is_off() {
1995 let payload = encode_ddl_alter_add_column(table, &col);
1996 self.wal.append(0, WalRecordType::DdlAddColumn, &payload)?;
1997 self.wal.flush()?;
1998 self.wal.last_appended_lsn()
1999 } else {
2000 0
2001 };
2002 let tbl = self.by_name_mut(table)?;
2003
2004 let old_schema = tbl.schema.clone();
2005
2006 let has_rows = tbl.heap.scan().next().is_some();
2010
2011 if has_rows && col.required {
2012 return Err(io::Error::new(
2013 io::ErrorKind::InvalidInput,
2014 format!(
2015 "cannot add required column '{}' to non-empty table '{table}': \
2016 no default value to backfill existing rows with",
2017 col.name
2018 ),
2019 ));
2020 }
2021
2022 tbl.schema.columns.push(col);
2026 tbl.refresh_layout();
2027
2028 if has_rows {
2029 let fill: Vec<Value> = vec![Value::Empty; tbl.schema.columns.len()];
2034 tbl.rewrite_rows_for_schema_change(&old_schema, &fill, &data_dir)?;
2035 }
2036 if barrier_lsn > 0 {
2044 tbl.heap.stamp_all_pages_min_lsn(barrier_lsn)?;
2045 tbl.heap.flush()?;
2046 }
2047
2048 self.persist()?;
2049 Ok(())
2050 }
2051
2052 pub fn alter_table_drop_column(&mut self, table: &str, col_name: &str) -> io::Result<()> {
2075 let data_dir = self.data_dir.clone();
2076 {
2077 let tbl = self.by_name_mut(table)?;
2078 tbl.schema
2079 .columns
2080 .iter()
2081 .position(|c| c.name == col_name)
2082 .ok_or_else(|| {
2083 io::Error::new(
2084 io::ErrorKind::NotFound,
2085 format!("column '{col_name}' not found in table '{table}'"),
2086 )
2087 })?;
2088 }
2089 let barrier_lsn = if !self.wal.is_off() {
2090 let payload = encode_ddl_alter_drop_column(table, col_name);
2091 self.wal.append(0, WalRecordType::DdlDropColumn, &payload)?;
2092 self.wal.flush()?;
2093 self.wal.last_appended_lsn()
2094 } else {
2095 0
2096 };
2097 let tbl = self.by_name_mut(table)?;
2098 let idx = tbl
2099 .schema
2100 .columns
2101 .iter()
2102 .position(|c| c.name == col_name)
2103 .ok_or_else(|| {
2104 io::Error::new(
2105 io::ErrorKind::NotFound,
2106 format!("column '{col_name}' not found in table '{table}'"),
2107 )
2108 })?;
2109
2110 let old_schema = tbl.schema.clone();
2112 let has_rows = tbl.heap.scan().next().is_some();
2113
2114 tbl.schema.columns.remove(idx);
2116 for (i, col) in tbl.schema.columns.iter_mut().enumerate() {
2117 col.position = i as u16;
2118 }
2119 tbl.refresh_layout();
2120
2121 if has_rows {
2122 let fill: Vec<Value> = vec![Value::Empty; tbl.schema.columns.len()];
2128 tbl.rewrite_rows_for_schema_change(&old_schema, &fill, &data_dir)?;
2129 }
2130 if barrier_lsn > 0 {
2132 tbl.heap.stamp_all_pages_min_lsn(barrier_lsn)?;
2133 tbl.heap.flush()?;
2134 }
2135
2136 self.persist()?;
2137 Ok(())
2138 }
2139}
2140
2141impl Drop for Catalog {
2142 fn drop(&mut self) {
2143 if self.active_tx_id.is_some() {
2144 if let Err(e) = self.abandon_active_transaction_for_drop() {
2145 warn!(error = %e, "catalog drop active transaction cleanup failed");
2146 }
2147 return;
2148 }
2149 if let Err(e) = self.checkpoint() {
2161 warn!(error = %e, "catalog drop checkpoint failed");
2162 }
2163 }
2164}
2165
2166fn encode_wal_payload(table: &str, rid: RowId, row_bytes: &[u8]) -> Vec<u8> {
2181 let name = table.as_bytes();
2182 let mut out = Vec::with_capacity(4 + name.len() + 4 + 2 + 4 + row_bytes.len());
2183 out.extend_from_slice(&(name.len() as u32).to_le_bytes());
2184 out.extend_from_slice(name);
2185 out.extend_from_slice(&rid.page_id.to_le_bytes());
2186 out.extend_from_slice(&rid.slot_index.to_le_bytes());
2187 out.extend_from_slice(&(row_bytes.len() as u32).to_le_bytes());
2188 out.extend_from_slice(row_bytes);
2189 out
2190}
2191
2192fn decode_wal_payload(data: &[u8]) -> Option<(String, RowId, Vec<u8>)> {
2193 let mut pos = 0usize;
2194 if data.len() < 4 {
2195 return None;
2196 }
2197 let name_len = u32::from_le_bytes(data[pos..pos + 4].try_into().ok()?) as usize;
2198 pos += 4;
2199 if pos + name_len > data.len() {
2200 return None;
2201 }
2202 let name = std::str::from_utf8(&data[pos..pos + name_len])
2203 .ok()?
2204 .to_string();
2205 pos += name_len;
2206 if pos + 4 + 2 + 4 > data.len() {
2207 return None;
2208 }
2209 let page_id = u32::from_le_bytes(data[pos..pos + 4].try_into().ok()?);
2210 pos += 4;
2211 let slot_index = u16::from_le_bytes(data[pos..pos + 2].try_into().ok()?);
2212 pos += 2;
2213 let row_len = u32::from_le_bytes(data[pos..pos + 4].try_into().ok()?) as usize;
2214 pos += 4;
2215 if pos + row_len > data.len() {
2216 return None;
2217 }
2218 let row_bytes = data[pos..pos + row_len].to_vec();
2219 Some((
2220 name,
2221 RowId {
2222 page_id,
2223 slot_index,
2224 },
2225 row_bytes,
2226 ))
2227}
2228
2229fn write_overflow_chain_logged(
2235 heap: &mut HeapFile,
2236 wal: &mut Wal,
2237 table: &str,
2238 tx_id: u64,
2239 value: &[u8],
2240) -> io::Result<OverflowStub> {
2241 if value.len() > MAX_VALUE_SIZE {
2242 return Err(StorageError::ValueTooLarge {
2243 size: value.len(),
2244 max: MAX_VALUE_SIZE,
2245 }
2246 .into());
2247 }
2248 let n = value.len().div_ceil(OVERFLOW_PAYLOAD_CAP).max(1);
2249 let mut pages = Vec::with_capacity(n);
2250 for _ in 0..n {
2251 pages.push(heap.allocate_overflow_page()?);
2252 }
2253 for i in 0..n {
2254 let start = i * OVERFLOW_PAYLOAD_CAP;
2255 let end = (start + OVERFLOW_PAYLOAD_CAP).min(value.len());
2256 let chunk = &value[start..end];
2257 let next = if i + 1 < n {
2258 pages[i + 1]
2259 } else {
2260 OVERFLOW_CHAIN_END
2261 };
2262 let payload = encode_overflow_write_payload(table, pages[i], next, chunk);
2263 wal.append(tx_id, WalRecordType::OverflowWrite, &payload)?;
2264 let lsn = wal.last_appended_lsn();
2265 heap.write_overflow_page(pages[i], next, chunk, lsn)?;
2266 }
2267 Ok(OverflowStub::new(
2268 value.len() as u64,
2269 pages[0],
2270 crc32fast::hash(value),
2271 ))
2272}
2273
2274fn encode_row_with_spill_logged(
2279 tbl: &mut Table,
2280 wal: &mut Wal,
2281 tx_id: u64,
2282 values: &Row,
2283) -> io::Result<Vec<u8>> {
2284 let v1_len = crate::row::v1_encoded_len(tbl.row_layout(), values);
2287 let is_indexed = tbl.indexed_col_mask();
2288 let chosen = plan_spill(tbl.row_layout(), values, v1_len, &is_indexed);
2289 if chosen.is_empty() {
2290 let mut v1 = Vec::new();
2291 encode_row_into(&tbl.schema, values, &mut v1);
2292 return Ok(v1);
2293 }
2294 let table_name = tbl.schema.table_name.clone();
2295 let n_var = tbl.row_layout().n_var();
2296 let mut spilled: Vec<Option<OverflowStub>> = vec![None; n_var];
2297 for col_idx in chosen {
2298 let var_idx = tbl
2299 .row_layout()
2300 .var_index(col_idx)
2301 .expect("plan_spill only returns var columns");
2302 let bytes: Vec<u8> = match &values[col_idx] {
2303 Value::Str(s) => s.as_bytes().to_vec(),
2304 Value::Bytes(b) => b.clone(),
2305 _ => continue,
2306 };
2307 let stub = write_overflow_chain_logged(&mut tbl.heap, wal, &table_name, tx_id, &bytes)?;
2308 spilled[var_idx] = Some(stub);
2309 }
2310 let mut out = Vec::new();
2311 encode_row_v2_into(&tbl.schema, tbl.row_layout(), values, &spilled, &mut out);
2312 Ok(out)
2313}
2314
2315fn encode_overflow_write_payload(
2324 table: &str,
2325 page_id: u32,
2326 next_page: u32,
2327 chunk: &[u8],
2328) -> Vec<u8> {
2329 let name = table.as_bytes();
2330 let mut out = Vec::with_capacity(2 + name.len() + 4 + 4 + 2 + chunk.len());
2331 out.extend_from_slice(&(name.len() as u16).to_le_bytes());
2332 out.extend_from_slice(name);
2333 out.extend_from_slice(&page_id.to_le_bytes());
2334 out.extend_from_slice(&next_page.to_le_bytes());
2335 out.extend_from_slice(&(chunk.len() as u16).to_le_bytes());
2336 out.extend_from_slice(chunk);
2337 out
2338}
2339
2340fn decode_overflow_write_payload(data: &[u8]) -> Option<(String, u32, u32, Vec<u8>)> {
2341 let mut pos = 0usize;
2342 if data.len() < 2 {
2343 return None;
2344 }
2345 let name_len = u16::from_le_bytes(data[pos..pos + 2].try_into().ok()?) as usize;
2346 pos += 2;
2347 if pos + name_len + 4 + 4 + 2 > data.len() {
2348 return None;
2349 }
2350 let name = std::str::from_utf8(&data[pos..pos + name_len])
2351 .ok()?
2352 .to_string();
2353 pos += name_len;
2354 let page_id = u32::from_le_bytes(data[pos..pos + 4].try_into().ok()?);
2355 pos += 4;
2356 let next_page = u32::from_le_bytes(data[pos..pos + 4].try_into().ok()?);
2357 pos += 4;
2358 let chunk_len = u16::from_le_bytes(data[pos..pos + 2].try_into().ok()?) as usize;
2359 pos += 2;
2360 if pos + chunk_len > data.len() {
2361 return None;
2362 }
2363 Some((
2364 name,
2365 page_id,
2366 next_page,
2367 data[pos..pos + chunk_len].to_vec(),
2368 ))
2369}
2370
2371fn encode_overflow_free_payload(table: &str, pages: &[u32]) -> Vec<u8> {
2375 let name = table.as_bytes();
2376 let mut out = Vec::with_capacity(2 + name.len() + 4 + pages.len() * 4);
2377 out.extend_from_slice(&(name.len() as u16).to_le_bytes());
2378 out.extend_from_slice(name);
2379 out.extend_from_slice(&(pages.len() as u32).to_le_bytes());
2380 for p in pages {
2381 out.extend_from_slice(&p.to_le_bytes());
2382 }
2383 out
2384}
2385
2386fn decode_overflow_free_payload(data: &[u8]) -> Option<(String, Vec<u32>)> {
2387 let mut pos = 0usize;
2388 if data.len() < 2 {
2389 return None;
2390 }
2391 let name_len = u16::from_le_bytes(data[pos..pos + 2].try_into().ok()?) as usize;
2392 pos += 2;
2393 if pos + name_len + 4 > data.len() {
2394 return None;
2395 }
2396 let name = std::str::from_utf8(&data[pos..pos + name_len])
2397 .ok()?
2398 .to_string();
2399 pos += name_len;
2400 let count = u32::from_le_bytes(data[pos..pos + 4].try_into().ok()?) as usize;
2401 pos += 4;
2402 if pos + count * 4 > data.len() {
2403 return None;
2404 }
2405 let mut pages = Vec::with_capacity(count);
2406 for _ in 0..count {
2407 pages.push(u32::from_le_bytes(data[pos..pos + 4].try_into().ok()?));
2408 pos += 4;
2409 }
2410 Some((name, pages))
2411}
2412
2413fn encode_ddl_create_table(
2416 schema: &Schema,
2417 defaults: &[Option<Value>],
2418 auto_cols: &[bool],
2419) -> Vec<u8> {
2420 let name = schema.table_name.as_bytes();
2421 let mut out = Vec::new();
2422 out.extend_from_slice(&(name.len() as u32).to_le_bytes());
2423 out.extend_from_slice(name);
2424 out.extend_from_slice(&(schema.columns.len() as u16).to_le_bytes());
2425 for col in &schema.columns {
2426 let cn = col.name.as_bytes();
2427 out.extend_from_slice(&(cn.len() as u32).to_le_bytes());
2428 out.extend_from_slice(cn);
2429 out.push(col.type_id as u8);
2430 out.push(col.required as u8);
2431 out.extend_from_slice(&col.position.to_le_bytes());
2432 }
2433 encode_defaults_section(&mut out, defaults);
2437 encode_auto_section(&mut out, auto_cols);
2438 out
2439}
2440
2441fn decode_ddl_create_table(data: &[u8]) -> Option<(Schema, Vec<Option<Value>>, Vec<bool>)> {
2442 let mut pos = 0usize;
2443 if data.len() < 4 {
2444 return None;
2445 }
2446 let name_len = u32::from_le_bytes(data[pos..pos + 4].try_into().ok()?) as usize;
2447 pos += 4;
2448 if pos + name_len > data.len() {
2449 return None;
2450 }
2451 let table_name = std::str::from_utf8(&data[pos..pos + name_len])
2452 .ok()?
2453 .to_string();
2454 pos += name_len;
2455 if pos + 2 > data.len() {
2456 return None;
2457 }
2458 let n_cols = u16::from_le_bytes(data[pos..pos + 2].try_into().ok()?) as usize;
2459 pos += 2;
2460 let mut columns = Vec::with_capacity(n_cols);
2461 for _ in 0..n_cols {
2462 if pos + 4 > data.len() {
2463 return None;
2464 }
2465 let cn_len = u32::from_le_bytes(data[pos..pos + 4].try_into().ok()?) as usize;
2466 pos += 4;
2467 if pos + cn_len + 4 > data.len() {
2468 return None;
2469 }
2470 let col_name = std::str::from_utf8(&data[pos..pos + cn_len])
2471 .ok()?
2472 .to_string();
2473 pos += cn_len;
2474 let type_id = TypeId::from_u8(data[pos])?;
2475 pos += 1;
2476 let required = data[pos] != 0;
2477 pos += 1;
2478 if pos + 2 > data.len() {
2479 return None;
2480 }
2481 let position = u16::from_le_bytes(data[pos..pos + 2].try_into().ok()?);
2482 pos += 2;
2483 columns.push(ColumnDef {
2484 name: col_name,
2485 type_id,
2486 required,
2487 position,
2488 });
2489 }
2490 let defaults = if pos < data.len() {
2493 decode_defaults_section(data, &mut pos, columns.len())?
2494 } else {
2495 Vec::new()
2496 };
2497 let auto_cols = if pos < data.len() {
2498 decode_auto_section(data, &mut pos, columns.len())?
2499 } else {
2500 Vec::new()
2501 };
2502 Some((
2503 Schema {
2504 table_name,
2505 columns,
2506 },
2507 defaults,
2508 auto_cols,
2509 ))
2510}
2511
2512fn encode_ddl_drop_table(table_name: &str) -> Vec<u8> {
2513 let name = table_name.as_bytes();
2514 let mut out = Vec::with_capacity(4 + name.len());
2515 out.extend_from_slice(&(name.len() as u32).to_le_bytes());
2516 out.extend_from_slice(name);
2517 out
2518}
2519
2520fn encode_ddl_alter_add_column(table_name: &str, col: &ColumnDef) -> Vec<u8> {
2521 let name = table_name.as_bytes();
2522 let cn = col.name.as_bytes();
2523 let mut out = Vec::with_capacity(4 + name.len() + 4 + cn.len() + 4);
2524 out.extend_from_slice(&(name.len() as u32).to_le_bytes());
2525 out.extend_from_slice(name);
2526 out.extend_from_slice(&(cn.len() as u32).to_le_bytes());
2527 out.extend_from_slice(cn);
2528 out.push(col.type_id as u8);
2529 out.push(col.required as u8);
2530 out.extend_from_slice(&col.position.to_le_bytes());
2531 out
2532}
2533
2534fn encode_ddl_alter_drop_column(table_name: &str, col_name: &str) -> Vec<u8> {
2535 let name = table_name.as_bytes();
2536 let cn = col_name.as_bytes();
2537 let mut out = Vec::with_capacity(4 + name.len() + 4 + cn.len());
2538 out.extend_from_slice(&(name.len() as u32).to_le_bytes());
2539 out.extend_from_slice(name);
2540 out.extend_from_slice(&(cn.len() as u32).to_le_bytes());
2541 out.extend_from_slice(cn);
2542 out
2543}
2544
2545fn decode_ddl_table_name(data: &[u8]) -> Option<(String, usize)> {
2546 if data.len() < 4 {
2547 return None;
2548 }
2549 let name_len = u32::from_le_bytes(data[0..4].try_into().ok()?) as usize;
2550 if 4 + name_len > data.len() {
2551 return None;
2552 }
2553 let name = std::str::from_utf8(&data[4..4 + name_len])
2554 .ok()?
2555 .to_string();
2556 Some((name, 4 + name_len))
2557}
2558
2559fn decode_ddl_alter_add_column(data: &[u8]) -> Option<(String, ColumnDef)> {
2560 let (table_name, mut pos) = decode_ddl_table_name(data)?;
2561 if pos + 4 > data.len() {
2562 return None;
2563 }
2564 let cn_len = u32::from_le_bytes(data[pos..pos + 4].try_into().ok()?) as usize;
2565 pos += 4;
2566 if pos + cn_len + 4 > data.len() {
2567 return None;
2568 }
2569 let col_name = std::str::from_utf8(&data[pos..pos + cn_len])
2570 .ok()?
2571 .to_string();
2572 pos += cn_len;
2573 let type_id = TypeId::from_u8(data[pos])?;
2574 pos += 1;
2575 let required = data[pos] != 0;
2576 pos += 1;
2577 if pos + 2 > data.len() {
2578 return None;
2579 }
2580 let position = u16::from_le_bytes(data[pos..pos + 2].try_into().ok()?);
2581 Some((
2582 table_name,
2583 ColumnDef {
2584 name: col_name,
2585 type_id,
2586 required,
2587 position,
2588 },
2589 ))
2590}
2591
2592fn decode_ddl_alter_drop_column(data: &[u8]) -> Option<(String, String)> {
2593 let (table_name, pos) = decode_ddl_table_name(data)?;
2594 if pos + 4 > data.len() {
2595 return None;
2596 }
2597 let cn_len = u32::from_le_bytes(data[pos..pos + 4].try_into().ok()?) as usize;
2598 if pos + 4 + cn_len > data.len() {
2599 return None;
2600 }
2601 let col_name = std::str::from_utf8(&data[pos + 4..pos + 4 + cn_len])
2602 .ok()?
2603 .to_string();
2604 Some((table_name, col_name))
2605}
2606
2607pub(crate) struct IndexedColMeta {
2635 pub name: String,
2636 pub unique: bool,
2637}
2638
2639pub(crate) struct CatalogEntry {
2642 pub schema: Schema,
2643 pub indexed_cols: Vec<IndexedColMeta>,
2644 pub defaults: Vec<Option<Value>>,
2647 pub auto_cols: Vec<bool>,
2650}
2651
2652pub(crate) struct CatalogEntryRef<'a> {
2654 pub schema: &'a Schema,
2655 pub indexed_cols: Vec<IndexedColMeta>,
2656 pub defaults: &'a [Option<Value>],
2657 pub auto_cols: &'a [bool],
2658}
2659
2660fn encode_value_blob(out: &mut Vec<u8>, v: &Value) {
2666 out.push(v.type_id() as u8);
2667 match v {
2668 Value::Int(n) => out.extend_from_slice(&n.to_le_bytes()),
2669 Value::Float(f) => out.extend_from_slice(&f.to_bits().to_le_bytes()),
2670 Value::Bool(b) => out.push(*b as u8),
2671 Value::Str(s) => {
2672 out.extend_from_slice(&(s.len() as u32).to_le_bytes());
2673 out.extend_from_slice(s.as_bytes());
2674 }
2675 Value::DateTime(n) => out.extend_from_slice(&n.to_le_bytes()),
2676 Value::Uuid(u) => out.extend_from_slice(u),
2677 Value::Bytes(b) => {
2678 out.extend_from_slice(&(b.len() as u32).to_le_bytes());
2679 out.extend_from_slice(b);
2680 }
2681 Value::Empty => {}
2682 }
2683}
2684
2685fn decode_value_blob(data: &[u8], pos: &mut usize) -> Option<Value> {
2688 let tag = *data.get(*pos)?;
2689 *pos += 1;
2690 let type_id = TypeId::from_u8(tag)?;
2691 let take_fixed = |pos: &mut usize, n: usize| -> Option<Vec<u8>> {
2692 if *pos + n > data.len() {
2693 return None;
2694 }
2695 let slice = data[*pos..*pos + n].to_vec();
2696 *pos += n;
2697 Some(slice)
2698 };
2699 match type_id {
2700 TypeId::Empty => Some(Value::Empty),
2701 TypeId::Int => Some(Value::Int(i64::from_le_bytes(
2702 take_fixed(pos, 8)?.try_into().ok()?,
2703 ))),
2704 TypeId::Float => Some(Value::Float(f64::from_bits(u64::from_le_bytes(
2705 take_fixed(pos, 8)?.try_into().ok()?,
2706 )))),
2707 TypeId::Bool => Some(Value::Bool(take_fixed(pos, 1)?[0] != 0)),
2708 TypeId::DateTime => Some(Value::DateTime(i64::from_le_bytes(
2709 take_fixed(pos, 8)?.try_into().ok()?,
2710 ))),
2711 TypeId::Uuid => Some(Value::Uuid(take_fixed(pos, 16)?.try_into().ok()?)),
2712 TypeId::Str => {
2713 let len = u32::from_le_bytes(take_fixed(pos, 4)?.try_into().ok()?) as usize;
2714 Some(Value::Str(String::from_utf8(take_fixed(pos, len)?).ok()?))
2715 }
2716 TypeId::Bytes => {
2717 let len = u32::from_le_bytes(take_fixed(pos, 4)?.try_into().ok()?) as usize;
2718 Some(Value::Bytes(take_fixed(pos, len)?))
2719 }
2720 }
2721}
2722
2723fn encode_defaults_section(out: &mut Vec<u8>, defaults: &[Option<Value>]) {
2727 let present: Vec<(u16, &Value)> = defaults
2728 .iter()
2729 .enumerate()
2730 .filter_map(|(i, d)| d.as_ref().map(|v| (i as u16, v)))
2731 .collect();
2732 out.extend_from_slice(&(present.len() as u16).to_le_bytes());
2733 for (pos, v) in present {
2734 out.extend_from_slice(&pos.to_le_bytes());
2735 encode_value_blob(out, v);
2736 }
2737}
2738
2739fn decode_defaults_section(
2742 data: &[u8],
2743 pos: &mut usize,
2744 n_cols: usize,
2745) -> Option<Vec<Option<Value>>> {
2746 if *pos + 2 > data.len() {
2747 return None;
2748 }
2749 let count = u16::from_le_bytes(data[*pos..*pos + 2].try_into().ok()?) as usize;
2750 *pos += 2;
2751 let mut out = vec![None; n_cols];
2752 for _ in 0..count {
2753 if *pos + 2 > data.len() {
2754 return None;
2755 }
2756 let col = u16::from_le_bytes(data[*pos..*pos + 2].try_into().ok()?) as usize;
2757 *pos += 2;
2758 let value = decode_value_blob(data, pos)?;
2759 if col < n_cols {
2760 out[col] = Some(value);
2761 }
2762 }
2763 Some(out)
2764}
2765
2766fn encode_auto_section(out: &mut Vec<u8>, auto_cols: &[bool]) {
2770 let present: Vec<u16> = auto_cols
2771 .iter()
2772 .enumerate()
2773 .filter_map(|(i, &a)| if a { Some(i as u16) } else { None })
2774 .collect();
2775 out.extend_from_slice(&(present.len() as u16).to_le_bytes());
2776 for pos in present {
2777 out.extend_from_slice(&pos.to_le_bytes());
2778 }
2779}
2780
2781fn decode_auto_section(data: &[u8], pos: &mut usize, n_cols: usize) -> Option<Vec<bool>> {
2784 if *pos + 2 > data.len() {
2785 return None;
2786 }
2787 let count = u16::from_le_bytes(data[*pos..*pos + 2].try_into().ok()?) as usize;
2788 *pos += 2;
2789 let mut out = vec![false; n_cols];
2790 for _ in 0..count {
2791 if *pos + 2 > data.len() {
2792 return None;
2793 }
2794 let col = u16::from_le_bytes(data[*pos..*pos + 2].try_into().ok()?) as usize;
2795 *pos += 2;
2796 if col < n_cols {
2797 out[col] = true;
2798 }
2799 }
2800 Some(out)
2801}
2802
2803fn write_catalog_file(path: &Path, entries: &[CatalogEntryRef<'_>]) -> io::Result<()> {
2804 let mut buf: Vec<u8> = Vec::with_capacity(64);
2805 buf.extend_from_slice(CATALOG_MAGIC);
2806 buf.extend_from_slice(&CATALOG_VERSION.to_le_bytes());
2807 buf.extend_from_slice(&(entries.len() as u32).to_le_bytes());
2808
2809 for entry in entries {
2810 let schema = entry.schema;
2811 let name = schema.table_name.as_bytes();
2812 buf.extend_from_slice(&(name.len() as u32).to_le_bytes());
2813 buf.extend_from_slice(name);
2814 buf.extend_from_slice(&(schema.columns.len() as u16).to_le_bytes());
2815 for col in &schema.columns {
2816 let cn = col.name.as_bytes();
2817 buf.extend_from_slice(&(cn.len() as u32).to_le_bytes());
2818 buf.extend_from_slice(cn);
2819 buf.push(col.type_id as u8);
2820 buf.push(if col.required { 1 } else { 0 });
2821 buf.extend_from_slice(&col.position.to_le_bytes());
2822 }
2823 buf.extend_from_slice(&(entry.indexed_cols.len() as u16).to_le_bytes());
2825 for meta in &entry.indexed_cols {
2826 let cn = meta.name.as_bytes();
2827 buf.extend_from_slice(&(cn.len() as u32).to_le_bytes());
2828 buf.extend_from_slice(cn);
2829 buf.push(if meta.unique { 1 } else { 0 });
2830 }
2831 encode_defaults_section(&mut buf, entry.defaults);
2833 encode_auto_section(&mut buf, entry.auto_cols);
2835 }
2836
2837 let crc = crc32fast::hash(&buf);
2841 buf.extend_from_slice(&crc.to_le_bytes());
2842
2843 let mut f = fs::OpenOptions::new()
2844 .create(true)
2845 .write(true)
2846 .truncate(true)
2847 .open(path)?;
2848 f.write_all(&buf)?;
2849 f.sync_data()?;
2850 Ok(())
2851}
2852
2853fn read_catalog_file(path: &Path) -> io::Result<Vec<CatalogEntry>> {
2854 let mut f = fs::File::open(path)?;
2855 let mut buf = Vec::new();
2856 f.read_to_end(&mut buf)?;
2857
2858 let mut pos = 0usize;
2859 if buf.len() < 14 || &buf[0..4] != CATALOG_MAGIC {
2861 return Err(io::Error::new(
2862 io::ErrorKind::InvalidData,
2863 "bad catalog magic",
2864 ));
2865 }
2866
2867 let payload = &buf[..buf.len() - 4];
2869 let stored_crc = u32::from_le_bytes(
2870 buf[buf.len() - 4..]
2871 .try_into()
2872 .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "truncated catalog CRC"))?,
2873 );
2874 let computed_crc = crc32fast::hash(payload);
2875 if stored_crc != computed_crc {
2876 return Err(io::Error::new(
2877 io::ErrorKind::InvalidData,
2878 format!(
2879 "catalog CRC32 mismatch: expected {stored_crc:#010x}, got {computed_crc:#010x}"
2880 ),
2881 ));
2882 }
2883 let buf = &buf[..buf.len() - 4];
2885 pos += 4;
2886 let version = u16::from_le_bytes(
2887 buf[pos..pos + 2]
2888 .try_into()
2889 .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "truncated catalog header"))?,
2890 );
2891 pos += 2;
2892 if version == 0 || version > CATALOG_VERSION {
2902 return Err(io::Error::new(
2903 io::ErrorKind::InvalidData,
2904 format!("unsupported catalog version: {version}"),
2905 ));
2906 }
2907 let n_tables = u32::from_le_bytes(
2908 buf[pos..pos + 4]
2909 .try_into()
2910 .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "truncated catalog header"))?,
2911 ) as usize;
2912 pos += 4;
2913
2914 if n_tables > buf.len() {
2921 return Err(io::Error::new(
2922 io::ErrorKind::InvalidData,
2923 format!("catalog file corrupt: implausible table count {n_tables}"),
2924 ));
2925 }
2926
2927 let mut entries = Vec::with_capacity(n_tables);
2928 for _ in 0..n_tables {
2929 let name_len = read_u32(buf, &mut pos)? as usize;
2930 let table_name = read_string(buf, &mut pos, name_len)?;
2931 let n_cols = read_u16(buf, &mut pos)? as usize;
2932
2933 let mut columns = Vec::with_capacity(n_cols);
2934 for _ in 0..n_cols {
2935 let cname_len = read_u32(buf, &mut pos)? as usize;
2936 let name = read_string(buf, &mut pos, cname_len)?;
2937 let type_id_raw = read_u8(buf, &mut pos)?;
2938 let type_id = type_id_from_u8(type_id_raw)?;
2939 let required = read_u8(buf, &mut pos)? != 0;
2940 let position = read_u16(buf, &mut pos)?;
2941 columns.push(ColumnDef {
2942 name,
2943 type_id,
2944 required,
2945 position,
2946 });
2947 }
2948
2949 let indexed_cols: Vec<IndexedColMeta> = if version >= 3 {
2953 let n = read_u16(buf, &mut pos)? as usize;
2954 let mut v = Vec::with_capacity(n);
2955 for _ in 0..n {
2956 let l = read_u32(buf, &mut pos)? as usize;
2957 let name = read_string(buf, &mut pos, l)?;
2958 let unique = read_u8(buf, &mut pos)? != 0;
2959 v.push(IndexedColMeta { name, unique });
2960 }
2961 v
2962 } else if version >= 2 {
2963 let n = read_u16(buf, &mut pos)? as usize;
2964 let mut v = Vec::with_capacity(n);
2965 for _ in 0..n {
2966 let l = read_u32(buf, &mut pos)? as usize;
2967 let name = read_string(buf, &mut pos, l)?;
2968 v.push(IndexedColMeta {
2969 name,
2970 unique: false,
2971 });
2972 }
2973 v
2974 } else {
2975 Vec::new()
2976 };
2977
2978 let defaults = if version >= 4 {
2980 decode_defaults_section(buf, &mut pos, columns.len()).ok_or_else(|| {
2981 io::Error::new(io::ErrorKind::InvalidData, "truncated catalog defaults")
2982 })?
2983 } else {
2984 Vec::new()
2985 };
2986
2987 let auto_cols = if version >= 5 {
2989 decode_auto_section(buf, &mut pos, columns.len()).ok_or_else(|| {
2990 io::Error::new(io::ErrorKind::InvalidData, "truncated catalog auto columns")
2991 })?
2992 } else {
2993 Vec::new()
2994 };
2995
2996 entries.push(CatalogEntry {
2997 schema: Schema {
2998 table_name,
2999 columns,
3000 },
3001 indexed_cols,
3002 defaults,
3003 auto_cols,
3004 });
3005 }
3006
3007 Ok(entries)
3008}
3009
3010fn read_u8(buf: &[u8], pos: &mut usize) -> io::Result<u8> {
3011 if *pos >= buf.len() {
3012 return Err(io::Error::new(
3013 io::ErrorKind::UnexpectedEof,
3014 "truncated catalog",
3015 ));
3016 }
3017 let v = buf[*pos];
3018 *pos += 1;
3019 Ok(v)
3020}
3021fn read_u16(buf: &[u8], pos: &mut usize) -> io::Result<u16> {
3022 if *pos + 2 > buf.len() {
3023 return Err(io::Error::new(
3024 io::ErrorKind::UnexpectedEof,
3025 "truncated catalog",
3026 ));
3027 }
3028 let v = u16::from_le_bytes(
3029 buf[*pos..*pos + 2]
3030 .try_into()
3031 .expect("bounds checked above"),
3032 );
3033 *pos += 2;
3034 Ok(v)
3035}
3036fn read_u32(buf: &[u8], pos: &mut usize) -> io::Result<u32> {
3037 if *pos + 4 > buf.len() {
3038 return Err(io::Error::new(
3039 io::ErrorKind::UnexpectedEof,
3040 "truncated catalog",
3041 ));
3042 }
3043 let v = u32::from_le_bytes(
3044 buf[*pos..*pos + 4]
3045 .try_into()
3046 .expect("bounds checked above"),
3047 );
3048 *pos += 4;
3049 Ok(v)
3050}
3051fn read_string(buf: &[u8], pos: &mut usize, len: usize) -> io::Result<String> {
3052 if *pos + len > buf.len() {
3053 return Err(io::Error::new(
3054 io::ErrorKind::UnexpectedEof,
3055 "truncated catalog string",
3056 ));
3057 }
3058 let s = std::str::from_utf8(&buf[*pos..*pos + len])
3059 .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "non-utf8 in catalog"))?
3060 .to_string();
3061 *pos += len;
3062 Ok(s)
3063}
3064fn type_id_from_u8(v: u8) -> io::Result<TypeId> {
3065 match v {
3066 0 => Ok(TypeId::Empty),
3067 1 => Ok(TypeId::Int),
3068 2 => Ok(TypeId::Float),
3069 3 => Ok(TypeId::Bool),
3070 4 => Ok(TypeId::Str),
3071 5 => Ok(TypeId::DateTime),
3072 6 => Ok(TypeId::Uuid),
3073 7 => Ok(TypeId::Bytes),
3074 _ => Err(io::Error::new(
3075 io::ErrorKind::InvalidData,
3076 format!("unknown type id: {v}"),
3077 )),
3078 }
3079}
3080
3081#[cfg(test)]
3082mod tests {
3083 use super::*;
3084 fn temp_catalog(name: &str) -> Catalog {
3085 let dir = std::env::temp_dir().join(format!("powdb_cat_{name}_{}", std::process::id()));
3086 Catalog::create(&dir).unwrap()
3087 }
3088
3089 fn schema_two_cols() -> Schema {
3090 Schema {
3091 table_name: "T".into(),
3092 columns: vec![
3093 ColumnDef {
3094 name: "id".into(),
3095 type_id: TypeId::Int,
3096 required: true,
3097 position: 0,
3098 },
3099 ColumnDef {
3100 name: "status".into(),
3101 type_id: TypeId::Str,
3102 required: false,
3103 position: 1,
3104 },
3105 ],
3106 }
3107 }
3108
3109 #[test]
3110 fn replay_records_treats_reused_tx_ids_as_ordered_spans() {
3111 let mut cat = temp_catalog("reused_tx_ids");
3112 let schema = schema_two_cols();
3113 cat.create_table(schema.clone()).unwrap();
3114 cat.checkpoint().unwrap();
3115
3116 let mut committed_row = Vec::new();
3117 encode_row_into(
3118 &schema,
3119 &[Value::Int(1), Value::Str("committed".into())],
3120 &mut committed_row,
3121 );
3122 let mut incomplete_row = Vec::new();
3123 encode_row_into(
3124 &schema,
3125 &[Value::Int(2), Value::Str("incomplete".into())],
3126 &mut incomplete_row,
3127 );
3128
3129 let records = vec![
3130 WalRecord {
3131 tx_id: 1,
3132 record_type: WalRecordType::Begin,
3133 lsn: 1,
3134 data: Vec::new(),
3135 },
3136 WalRecord {
3137 tx_id: 1,
3138 record_type: WalRecordType::Insert,
3139 lsn: 2,
3140 data: encode_wal_payload(
3141 "T",
3142 RowId {
3143 page_id: 1,
3144 slot_index: 0,
3145 },
3146 &committed_row,
3147 ),
3148 },
3149 WalRecord {
3150 tx_id: 1,
3151 record_type: WalRecordType::Commit,
3152 lsn: 3,
3153 data: Vec::new(),
3154 },
3155 WalRecord {
3156 tx_id: 1,
3157 record_type: WalRecordType::Begin,
3158 lsn: 4,
3159 data: Vec::new(),
3160 },
3161 WalRecord {
3162 tx_id: 1,
3163 record_type: WalRecordType::Insert,
3164 lsn: 5,
3165 data: encode_wal_payload(
3166 "T",
3167 RowId {
3168 page_id: 1,
3169 slot_index: 1,
3170 },
3171 &incomplete_row,
3172 ),
3173 },
3174 ];
3175
3176 cat.apply_wal_records(&records).unwrap();
3177 let rows: Vec<_> = cat.scan("T").unwrap().collect();
3178 assert_eq!(rows.len(), 1);
3179 assert_eq!(rows[0].1[0], Value::Int(1));
3180 assert_eq!(rows[0].1[1], Value::Str("committed".into()));
3181 }
3182
3183 #[test]
3184 fn ddl_create_table_codec_roundtrips_defaults_and_auto() {
3185 let schema = schema_two_cols();
3186 let defaults = vec![None, Some(Value::Str("active".into()))];
3187 let auto_cols = vec![true, false];
3188 let encoded = encode_ddl_create_table(&schema, &defaults, &auto_cols);
3189 let (decoded_schema, decoded_defaults, decoded_auto) =
3190 decode_ddl_create_table(&encoded).unwrap();
3191 assert_eq!(decoded_schema.columns.len(), 2);
3192 assert_eq!(decoded_defaults, defaults);
3193 assert_eq!(decoded_auto, auto_cols);
3194 }
3195
3196 #[test]
3197 fn ddl_create_table_codec_back_compat_without_trailing_sections() {
3198 let schema = schema_two_cols();
3202 let full = encode_ddl_create_table(&schema, &[], &[]);
3203 let legacy = &full[..full.len() - 4];
3206 let (decoded_schema, decoded_defaults, decoded_auto) =
3207 decode_ddl_create_table(legacy).unwrap();
3208 assert_eq!(decoded_schema.columns.len(), 2);
3209 assert!(decoded_defaults.is_empty(), "no defaults section -> empty");
3210 assert!(decoded_auto.is_empty(), "no auto section -> empty");
3211 }
3212
3213 #[test]
3214 fn ddl_create_table_codec_back_compat_defaults_but_no_auto() {
3215 let schema = schema_two_cols();
3218 let defaults = vec![None, Some(Value::Str("active".into()))];
3219 let full = encode_ddl_create_table(&schema, &defaults, &[]);
3220 let legacy = &full[..full.len() - 2];
3222 let (_schema, decoded_defaults, decoded_auto) = decode_ddl_create_table(legacy).unwrap();
3223 assert_eq!(decoded_defaults, defaults);
3224 assert!(decoded_auto.is_empty());
3225 }
3226
3227 #[test]
3228 fn read_catalog_file_accepts_intermediate_versions_3_and_4() {
3229 use std::io::Write as _;
3237 fn write_legacy_catalog(path: &std::path::Path, version: u16) {
3238 let mut buf: Vec<u8> = Vec::new();
3239 buf.extend_from_slice(CATALOG_MAGIC);
3240 buf.extend_from_slice(&version.to_le_bytes());
3241 buf.extend_from_slice(&1u32.to_le_bytes()); buf.extend_from_slice(&1u32.to_le_bytes());
3244 buf.extend_from_slice(b"T");
3245 buf.extend_from_slice(&2u16.to_le_bytes()); buf.extend_from_slice(&2u32.to_le_bytes());
3248 buf.extend_from_slice(b"id");
3249 buf.push(TypeId::Int as u8);
3250 buf.push(1);
3251 buf.extend_from_slice(&0u16.to_le_bytes());
3252 buf.extend_from_slice(&6u32.to_le_bytes());
3254 buf.extend_from_slice(b"status");
3255 buf.push(TypeId::Str as u8);
3256 buf.push(0);
3257 buf.extend_from_slice(&1u16.to_le_bytes());
3258 buf.extend_from_slice(&0u16.to_le_bytes());
3260 if version >= 4 {
3262 encode_defaults_section(&mut buf, &[None, None]);
3263 }
3264 let crc = crc32fast::hash(&buf);
3266 buf.extend_from_slice(&crc.to_le_bytes());
3267 let mut f = fs::File::create(path).unwrap();
3268 f.write_all(&buf).unwrap();
3269 }
3270
3271 for version in [3u16, 4u16] {
3272 let path = std::env::temp_dir().join(format!(
3273 "powdb_cat_v{version}_compat_{}.bin",
3274 std::process::id()
3275 ));
3276 write_legacy_catalog(&path, version);
3277 let entries = read_catalog_file(&path)
3278 .unwrap_or_else(|e| panic!("version {version} catalog must load, got: {e}"));
3279 assert_eq!(entries.len(), 1);
3280 assert_eq!(entries[0].schema.table_name, "T");
3281 assert_eq!(entries[0].schema.columns.len(), 2);
3282 assert!(
3283 entries[0].auto_cols.is_empty(),
3284 "v{version} has no auto cols"
3285 );
3286 fs::remove_file(&path).ok();
3287 }
3288 }
3289
3290 #[test]
3291 fn read_catalog_file_rejects_implausible_table_count() {
3292 use std::io::Write as _;
3300 let mut buf: Vec<u8> = Vec::new();
3301 buf.extend_from_slice(CATALOG_MAGIC);
3302 buf.extend_from_slice(&CATALOG_VERSION.to_le_bytes());
3303 buf.extend_from_slice(&1000u32.to_le_bytes()); let crc = crc32fast::hash(&buf);
3306 buf.extend_from_slice(&crc.to_le_bytes());
3307 let path =
3308 std::env::temp_dir().join(format!("powdb_cat_badcount_{}.bin", std::process::id()));
3309 fs::File::create(&path).unwrap().write_all(&buf).unwrap();
3310
3311 let msg = match read_catalog_file(&path) {
3312 Ok(_) => panic!("implausible table count must be rejected, got Ok"),
3313 Err(e) => e.to_string(),
3314 };
3315 assert!(
3316 msg.contains("implausible table count"),
3317 "expected an implausible-table-count error, got: {msg}"
3318 );
3319 fs::remove_file(&path).ok();
3320 }
3321
3322 #[test]
3323 fn data_dir_and_max_lsn_accessors() {
3324 let dir = std::env::temp_dir().join(format!("powdb_cat_maxlsn_{}", std::process::id()));
3325 let mut cat = Catalog::create(&dir).unwrap();
3326
3327 assert_eq!(cat.data_dir(), dir.as_path());
3329
3330 assert_eq!(cat.max_lsn(), 0);
3332
3333 let schema = Schema {
3334 table_name: "users".into(),
3335 columns: vec![ColumnDef {
3336 name: "name".into(),
3337 type_id: TypeId::Str,
3338 required: true,
3339 position: 0,
3340 }],
3341 };
3342 cat.create_table(schema).unwrap();
3343
3344 cat.insert("users", &vec![Value::Str("Alice".into())])
3345 .unwrap();
3346 cat.sync_wal().unwrap();
3347
3348 assert!(cat.max_lsn() > 0);
3351 }
3352
3353 #[test]
3354 fn test_create_table_and_insert() {
3355 let mut cat = temp_catalog("basic");
3356 let schema = Schema {
3357 table_name: "users".into(),
3358 columns: vec![
3359 ColumnDef {
3360 name: "name".into(),
3361 type_id: TypeId::Str,
3362 required: true,
3363 position: 0,
3364 },
3365 ColumnDef {
3366 name: "age".into(),
3367 type_id: TypeId::Int,
3368 required: false,
3369 position: 1,
3370 },
3371 ],
3372 };
3373 cat.create_table(schema).unwrap();
3374
3375 let row = vec![Value::Str("Alice".into()), Value::Int(30)];
3376 let rid = cat.insert("users", &row).unwrap();
3377
3378 let result = cat.get("users", rid).unwrap();
3379 assert_eq!(result[0], Value::Str("Alice".into()));
3380 assert_eq!(result[1], Value::Int(30));
3381 }
3382
3383 #[test]
3384 fn test_scan_table() {
3385 let mut cat = temp_catalog("scan");
3386 let schema = Schema {
3387 table_name: "items".into(),
3388 columns: vec![
3389 ColumnDef {
3390 name: "name".into(),
3391 type_id: TypeId::Str,
3392 required: true,
3393 position: 0,
3394 },
3395 ColumnDef {
3396 name: "price".into(),
3397 type_id: TypeId::Float,
3398 required: true,
3399 position: 1,
3400 },
3401 ],
3402 };
3403 cat.create_table(schema).unwrap();
3404
3405 for i in 0..50 {
3406 cat.insert(
3407 "items",
3408 &vec![
3409 Value::Str(format!("item_{i}")),
3410 Value::Float(i as f64 * 1.5),
3411 ],
3412 )
3413 .unwrap();
3414 }
3415
3416 let rows: Vec<_> = cat.scan("items").unwrap().collect();
3417 assert_eq!(rows.len(), 50);
3418 }
3419
3420 #[test]
3421 fn test_index_lookup() {
3422 let mut cat = temp_catalog("idx");
3423 let schema = Schema {
3424 table_name: "users".into(),
3425 columns: vec![
3426 ColumnDef {
3427 name: "email".into(),
3428 type_id: TypeId::Str,
3429 required: true,
3430 position: 0,
3431 },
3432 ColumnDef {
3433 name: "name".into(),
3434 type_id: TypeId::Str,
3435 required: true,
3436 position: 1,
3437 },
3438 ],
3439 };
3440 cat.create_table(schema).unwrap();
3441 cat.create_index("users", "email").unwrap();
3442
3443 cat.insert(
3444 "users",
3445 &vec![
3446 Value::Str("alice@example.com".into()),
3447 Value::Str("Alice".into()),
3448 ],
3449 )
3450 .unwrap();
3451 cat.insert(
3452 "users",
3453 &vec![
3454 Value::Str("bob@example.com".into()),
3455 Value::Str("Bob".into()),
3456 ],
3457 )
3458 .unwrap();
3459
3460 let result = cat
3461 .index_lookup("users", "email", &Value::Str("bob@example.com".into()))
3462 .unwrap();
3463 assert!(result.is_some());
3464 let row = result.unwrap();
3465 assert_eq!(row[1], Value::Str("Bob".into()));
3466 }
3467
3468 #[test]
3469 fn test_delete_row() {
3470 let mut cat = temp_catalog("delete");
3471 let schema = Schema {
3472 table_name: "t".into(),
3473 columns: vec![ColumnDef {
3474 name: "v".into(),
3475 type_id: TypeId::Int,
3476 required: true,
3477 position: 0,
3478 }],
3479 };
3480 cat.create_table(schema).unwrap();
3481 let r1 = cat.insert("t", &vec![Value::Int(1)]).unwrap();
3482 let r2 = cat.insert("t", &vec![Value::Int(2)]).unwrap();
3483 cat.delete("t", r1).unwrap();
3484 assert!(cat.get("t", r1).is_none());
3485 assert!(cat.get("t", r2).is_some());
3486 }
3487
3488 #[test]
3489 fn test_update_row() {
3490 let mut cat = temp_catalog("update");
3491 let schema = Schema {
3492 table_name: "t".into(),
3493 columns: vec![ColumnDef {
3494 name: "v".into(),
3495 type_id: TypeId::Int,
3496 required: true,
3497 position: 0,
3498 }],
3499 };
3500 cat.create_table(schema).unwrap();
3501 let rid = cat.insert("t", &vec![Value::Int(1)]).unwrap();
3502 let new_rid = cat.update("t", rid, &vec![Value::Int(99)]).unwrap();
3503 let row = cat.get("t", new_rid).unwrap();
3504 assert_eq!(row[0], Value::Int(99));
3505 }
3506
3507 #[test]
3508 fn test_persist_and_reopen() {
3509 let dir = std::env::temp_dir().join(format!("powdb_cat_persist_{}", std::process::id()));
3510 let _ = std::fs::remove_dir_all(&dir);
3512
3513 {
3514 let mut cat = Catalog::create(&dir).unwrap();
3515 cat.create_table(Schema {
3516 table_name: "users".into(),
3517 columns: vec![
3518 ColumnDef {
3519 name: "name".into(),
3520 type_id: TypeId::Str,
3521 required: true,
3522 position: 0,
3523 },
3524 ColumnDef {
3525 name: "age".into(),
3526 type_id: TypeId::Int,
3527 required: false,
3528 position: 1,
3529 },
3530 ],
3531 })
3532 .unwrap();
3533 cat.insert("users", &vec![Value::Str("Alice".into()), Value::Int(30)])
3534 .unwrap();
3535 cat.insert("users", &vec![Value::Str("Bob".into()), Value::Int(25)])
3536 .unwrap();
3537 }
3538
3539 let cat = Catalog::open(&dir).unwrap();
3541 let schema = cat.schema("users").unwrap();
3542 assert_eq!(schema.columns.len(), 2);
3543 assert_eq!(schema.columns[0].name, "name");
3544 assert_eq!(schema.columns[0].type_id, TypeId::Str);
3545 assert_eq!(schema.columns[1].type_id, TypeId::Int);
3546
3547 let rows: Vec<_> = cat.scan("users").unwrap().collect();
3548 assert_eq!(rows.len(), 2);
3549
3550 std::fs::remove_dir_all(&dir).ok();
3551 }
3552
3553 #[test]
3554 fn test_open_missing_dir_errors() {
3555 let dir = std::env::temp_dir().join(format!("powdb_cat_missing_{}", std::process::id()));
3556 let _ = std::fs::remove_dir_all(&dir);
3557 std::fs::create_dir_all(&dir).unwrap();
3558 assert!(Catalog::open(&dir).is_err());
3560 std::fs::remove_dir_all(&dir).ok();
3561 }
3562
3563 #[test]
3564 fn test_list_tables() {
3565 let mut cat = temp_catalog("list");
3566 cat.create_table(Schema {
3567 table_name: "a".into(),
3568 columns: vec![ColumnDef {
3569 name: "x".into(),
3570 type_id: TypeId::Int,
3571 required: true,
3572 position: 0,
3573 }],
3574 })
3575 .unwrap();
3576 cat.create_table(Schema {
3577 table_name: "b".into(),
3578 columns: vec![ColumnDef {
3579 name: "y".into(),
3580 type_id: TypeId::Int,
3581 required: true,
3582 position: 0,
3583 }],
3584 })
3585 .unwrap();
3586 let mut tables = cat.list_tables();
3587 tables.sort();
3588 assert_eq!(tables, vec!["a", "b"]);
3589 }
3590
3591 #[test]
3592 fn test_path_traversal_table_name_rejected() {
3593 let mut cat = temp_catalog("path_trav");
3594 let bad_names = vec![
3596 "../etc/passwd",
3597 "foo/bar",
3598 "table\0name",
3599 "",
3600 "123starts_with_digit",
3601 "has-dashes",
3602 "has spaces",
3603 "has.dots",
3604 ];
3605 for name in bad_names {
3606 let schema = Schema {
3607 table_name: name.into(),
3608 columns: vec![ColumnDef {
3609 name: "x".into(),
3610 type_id: TypeId::Int,
3611 required: true,
3612 position: 0,
3613 }],
3614 };
3615 let result = cat.create_table(schema);
3616 assert!(result.is_err(), "expected error for table name '{name}'");
3617 assert_eq!(result.unwrap_err().kind(), io::ErrorKind::InvalidInput);
3618 }
3619 let good_names = vec!["users", "_private", "Table_123", "_"];
3621 for name in good_names {
3622 let schema = Schema {
3623 table_name: name.into(),
3624 columns: vec![ColumnDef {
3625 name: "x".into(),
3626 type_id: TypeId::Int,
3627 required: true,
3628 position: 0,
3629 }],
3630 };
3631 assert!(
3632 cat.create_table(schema).is_ok(),
3633 "expected ok for table name '{name}'"
3634 );
3635 }
3636 }
3637
3638 #[test]
3639 fn test_path_traversal_column_name_rejected() {
3640 let mut cat = temp_catalog("col_path_trav");
3641 let schema = Schema {
3642 table_name: "valid_table".into(),
3643 columns: vec![ColumnDef {
3644 name: "../bad".into(),
3645 type_id: TypeId::Int,
3646 required: true,
3647 position: 0,
3648 }],
3649 };
3650 let result = cat.create_table(schema);
3651 assert!(result.is_err());
3652 assert_eq!(result.unwrap_err().kind(), io::ErrorKind::InvalidInput);
3653 }
3654
3655 #[test]
3656 fn test_drop_table_validates_name() {
3657 let mut cat = temp_catalog("drop_trav");
3658 let result = cat.drop_table("../etc/passwd");
3659 assert!(result.is_err());
3660 assert_eq!(result.unwrap_err().kind(), io::ErrorKind::InvalidInput);
3662 }
3663}