diff --git a/src/core/access.rs b/src/core/access.rs index 9aac565..9b2051f 100644 --- a/src/core/access.rs +++ b/src/core/access.rs @@ -1,59 +1,8 @@ -use crate::core::{ - IndexKeyRange, IndexSchema, OwnedTableRecord, TableKeyRange, TableSchema, TupleSchema, Value, - error::StorageResult, -}; -use crate::relational::record_manager::{IndexScan, TableScan}; +use crate::core::{IndexSchema, TableSchema, error::StorageResult}; -pub(crate) trait SchemaAccess { +/// Minimal catalog seam used by the planner and its lightweight tests. +pub(crate) trait CatalogRead { fn table_schema_by_name(&self, name: &str) -> StorageResult; fn index_schemas_for_table(&self, table: &TableSchema) -> StorageResult>; } - -pub(crate) trait DdlAccess { - fn create_table(&self, name: &str, row: TupleSchema) -> StorageResult; - - fn create_index( - &self, - name: &str, - table_name: &str, - columns: &[&str], - ) -> StorageResult; -} - -pub(crate) trait RecordAccess { - fn scan_table(&self, table: &TableSchema) -> StorageResult; - - fn scan_table_range( - &self, - table: &TableSchema, - range: TableKeyRange, - ) -> StorageResult; - - fn scan_index( - &self, - table: &TableSchema, - index: &IndexSchema, - key_range: IndexKeyRange, - ) -> StorageResult; - - fn insert_table_row( - &self, - table: &TableSchema, - values: Vec, - ) -> StorageResult; - - fn delete_table_row(&self, table: &TableSchema, record: &OwnedTableRecord) - -> StorageResult<()>; - - fn update_table_row( - &self, - table: &TableSchema, - record: &OwnedTableRecord, - values: Vec, - ) -> StorageResult; -} - -pub(crate) trait ExecutionAccess: DdlAccess + RecordAccess {} - -impl ExecutionAccess for T where T: DdlAccess + RecordAccess {} diff --git a/src/core/database.rs b/src/core/database.rs index 514e399..477351f 100644 --- a/src/core/database.rs +++ b/src/core/database.rs @@ -2,27 +2,23 @@ use std::path::Path; use crate::core::{ IndexKeyRange, IndexSchema, OwnedTableRecord, TableKeyRange, TableSchema, TupleSchema, Value, - access::{DdlAccess, RecordAccess, SchemaAccess}, - error::StorageResult, + access::CatalogRead, error::StorageResult, }; #[cfg(test)] use crate::relational::cursor::{IndexCursor, TableCursor}; use crate::relational::{ catalog_manager::CatalogManager, - index_manager::IndexManager, - record_manager::{IndexScan, RecordManager, TableScan}, + index_manager, + record_manager::{self, IndexScan, TableScan}, }; use crate::storage::{ - log_manager::TxnId, pager::Pager, transaction_manager::TransactionSavepoint, - transaction_runtime::TransactionRuntime, + engine::Storage, log_manager::TxnId, transaction_manager::TransactionSavepoint, }; /// Public database handle for one database file. pub struct Database { catalog: CatalogManager, - indexes: IndexManager, - records: RecordManager, - transactions: TransactionRuntime, + storage: Storage, } impl Database { @@ -33,8 +29,8 @@ impl Database { /// Returns an error when the file already exists, cannot be initialized, /// or its initial catalog and write-ahead log cannot be written. pub fn create(path: impl AsRef) -> StorageResult { - let pager = Pager::create(path)?; - Self::from_pager(pager) + let storage = Storage::create(path)?; + Self::from_storage(storage) } /// Opens an existing database file. @@ -44,8 +40,8 @@ impl Database { /// Returns an error when the file cannot be opened, its format is invalid, /// or recovery cannot restore it to a consistent state. pub fn open(path: impl AsRef) -> StorageResult { - let pager = Pager::open(path)?; - Self::from_pager(pager) + let storage = Storage::open(path)?; + Self::from_storage(storage) } /// Opens a database file, creating and initializing it if needed. @@ -55,16 +51,13 @@ impl Database { /// Returns an error when the file cannot be opened or initialized, its /// format is invalid, or recovery cannot restore it to a consistent state. pub fn open_or_create(path: impl AsRef) -> StorageResult { - let pager = Pager::open_or_create(path)?; - Self::from_pager(pager) + let storage = Storage::open_or_create(path)?; + Self::from_storage(storage) } - fn from_pager(pager: Pager) -> StorageResult { - let transactions = pager.transaction_runtime(); - let catalog = CatalogManager::from_pager(pager)?; - let indexes = IndexManager::new(catalog.clone()); - let records = RecordManager::new(catalog.clone(), indexes.clone()); - Ok(Self { catalog, indexes, records, transactions }) + fn from_storage(storage: Storage) -> StorageResult { + let catalog = CatalogManager::from_storage(storage.clone())?; + Ok(Self { catalog, storage }) } /// Returns the database-file path associated with this database. @@ -84,53 +77,58 @@ impl Database { #[cfg(test)] pub(crate) fn unlock_for_crash_for_test(&self) { - self.transactions.unlock_for_crash_for_test().unwrap(); + self.storage.unlock_for_crash_for_test().unwrap(); } pub(crate) fn begin_transaction(&self) -> StorageResult { - self.transactions.begin_transaction() + self.storage.begin_transaction() + } + + /// Returns the concrete relational gateway for an active transaction. + pub(crate) fn transaction(&self, txn_id: TxnId) -> crate::core::transaction::Transaction<'_> { + crate::core::transaction::Transaction::new(self, txn_id) } pub(crate) fn commit_transaction(&self, txn_id: TxnId) -> StorageResult<()> { - self.transactions.commit_transaction(txn_id) + self.storage.commit_transaction(txn_id) } pub(crate) fn statement_savepoint(&self, txn_id: TxnId) -> StorageResult { - self.transactions.statement_savepoint(txn_id) + self.storage.statement_savepoint(txn_id) } pub(crate) fn rollback_to_savepoint( &self, savepoint: TransactionSavepoint, ) -> StorageResult<()> { - self.transactions.rollback_to_savepoint(savepoint) + self.storage.rollback_to_savepoint(savepoint) } pub(crate) fn rollback_transaction(&self, txn_id: TxnId) -> StorageResult<()> { - self.transactions.rollback_transaction(txn_id) + self.storage.rollback_transaction(txn_id) } pub(crate) fn active_transaction_id(&self) -> Option { - self.transactions.active_transaction_id() + self.storage.active_transaction_id() } pub(crate) fn transaction_is_poisoned(&self, txn_id: TxnId) -> StorageResult { - self.transactions.transaction_is_poisoned(txn_id) + self.storage.transaction_is_poisoned(txn_id) } #[cfg(test)] pub(crate) fn force_next_lsn_exhausted_for_test(&self) { - self.transactions.force_next_lsn_exhausted_for_test(); + self.storage.force_next_lsn_exhausted_for_test(); } #[cfg(test)] pub(crate) fn fail_next_savepoint_rollback_for_test(&self) { - self.transactions.fail_next_savepoint_rollback_for_test(); + self.storage.fail_next_savepoint_rollback_for_test(); } #[cfg(test)] pub(crate) fn fail_next_wal_flush_for_test(&self) { - self.transactions.fail_next_wal_flush_for_test(); + self.storage.fail_next_wal_flush_for_test(); } #[cfg(test)] @@ -144,7 +142,7 @@ impl Database { } } -impl SchemaAccess for Database { +impl CatalogRead for Database { fn table_schema_by_name(&self, name: &str) -> StorageResult { self.catalog.table_schema_by_name(name) } @@ -154,66 +152,66 @@ impl SchemaAccess for Database { } } -impl DdlAccess for Database { - fn create_table(&self, name: &str, row: TupleSchema) -> StorageResult { +impl Database { + pub(crate) fn create_table(&self, name: &str, row: TupleSchema) -> StorageResult { self.catalog.create_table(name, row) } - fn create_index( + pub(crate) fn create_index( &self, name: &str, table_name: &str, columns: &[&str], ) -> StorageResult { - self.indexes.create_index(name, table_name, columns) + index_manager::create_index(&self.catalog, name, table_name, columns) } } -impl RecordAccess for Database { - fn scan_table(&self, table: &TableSchema) -> StorageResult { - self.records.scan_table(table) +impl Database { + pub(crate) fn scan_table(&self, table: &TableSchema) -> StorageResult { + record_manager::scan_table(&self.catalog, table) } - fn scan_table_range( + pub(crate) fn scan_table_range( &self, table: &TableSchema, range: TableKeyRange, ) -> StorageResult { - self.records.scan_table_range(table, range) + record_manager::scan_table_range(&self.catalog, table, range) } - fn scan_index( + pub(crate) fn scan_index( &self, table: &TableSchema, index: &IndexSchema, key_range: IndexKeyRange, ) -> StorageResult { - self.records.scan_index(table, index, key_range) + record_manager::scan_index(&self.catalog, table, index, key_range) } - fn insert_table_row( + pub(crate) fn insert_table_row( &self, table: &TableSchema, values: Vec, ) -> StorageResult { - self.records.insert_table_row(table, values) + record_manager::insert_table_row(&self.catalog, table, values) } - fn delete_table_row( + pub(crate) fn delete_table_row( &self, table: &TableSchema, record: &OwnedTableRecord, ) -> StorageResult<()> { - self.records.delete_table_row(table, record) + record_manager::delete_table_row(&self.catalog, table, record) } - fn update_table_row( + pub(crate) fn update_table_row( &self, table: &TableSchema, record: &OwnedTableRecord, values: Vec, ) -> StorageResult { - self.records.update_table_row(table, record, values) + record_manager::update_table_row(&self.catalog, table, record, values) } } diff --git a/src/core/mod.rs b/src/core/mod.rs index 2256ce7..9ffbff9 100644 --- a/src/core/mod.rs +++ b/src/core/mod.rs @@ -11,6 +11,7 @@ pub(crate) mod access; pub(crate) mod database; pub mod error; pub mod lock_manager; +pub(crate) mod transaction; mod types; pub use crate::relational::catalog::{ @@ -25,9 +26,9 @@ pub use error::{ ConstraintError, CorruptionComponent, CorruptionError, CorruptionKind, InternalError, InvalidArgumentError, LimitExceededError, StorageError, StorageResult, }; -pub use types::TxnId; +pub(crate) use transaction::Transaction; pub use types::{ - CatalogId, IndexKeyBound, IndexKeyRange, PageId, TableKey, TableKeyBound, TableKeyRange, + CatalogId, IndexKeyBound, IndexKeyRange, PageId, TableKey, TableKeyBound, TableKeyRange, TxnId, }; pub(crate) use types::{Lsn, SlotId}; diff --git a/src/core/transaction.rs b/src/core/transaction.rs new file mode 100644 index 0000000..a277e8a --- /dev/null +++ b/src/core/transaction.rs @@ -0,0 +1,105 @@ +//! Transaction-scoped access to relational operations. +//! +//! This type is the concrete boundary between session/execution policy and the +//! relational engine. Transaction identity is retained here even though lower +//! storage mutations still use the legacy ambient transaction during the +//! migration. + +use crate::{ + core::{ + IndexKeyRange, IndexSchema, OwnedTableRecord, TableKeyRange, TableSchema, TupleSchema, + Value, error::StorageResult, + }, + relational::record_manager::{IndexScan, TableScan}, + storage::{log_manager::TxnId, transaction_manager::TransactionSavepoint}, +}; + +use super::Database; + +/// An active transaction's concrete relational gateway. +pub(crate) struct Transaction<'db> { + database: &'db Database, + txn_id: TxnId, +} + +impl<'db> Transaction<'db> { + pub(super) fn new(database: &'db Database, txn_id: TxnId) -> Self { + Self { database, txn_id } + } + + /// Returns the storage transaction identity associated with this gateway. + pub(crate) fn id(&self) -> TxnId { + self.txn_id + } + + pub(crate) fn statement_savepoint(&self) -> StorageResult { + self.database.statement_savepoint(self.txn_id) + } + + pub(crate) fn is_poisoned(&self) -> StorageResult { + self.database.transaction_is_poisoned(self.txn_id) + } +} + +impl Transaction<'_> { + pub(crate) fn create_table(&self, name: &str, row: TupleSchema) -> StorageResult { + self.database.create_table(name, row) + } + + pub(crate) fn create_index( + &self, + name: &str, + table_name: &str, + columns: &[&str], + ) -> StorageResult { + self.database.create_index(name, table_name, columns) + } +} + +impl Transaction<'_> { + pub(crate) fn scan_table(&self, table: &TableSchema) -> StorageResult { + self.database.scan_table(table) + } + + pub(crate) fn scan_table_range( + &self, + table: &TableSchema, + range: TableKeyRange, + ) -> StorageResult { + self.database.scan_table_range(table, range) + } + + pub(crate) fn scan_index( + &self, + table: &TableSchema, + index: &IndexSchema, + key_range: IndexKeyRange, + ) -> StorageResult { + self.database.scan_index(table, index, key_range) + } + + pub(crate) fn insert_table_row( + &self, + table: &TableSchema, + values: Vec, + ) -> StorageResult { + self.database.insert_table_row(table, values) + } + + pub(crate) fn delete_table_row( + &self, + table: &TableSchema, + record: &OwnedTableRecord, + ) -> StorageResult<()> { + self.database.delete_table_row(table, record) + } + + pub(crate) fn update_table_row( + &self, + table: &TableSchema, + record: &OwnedTableRecord, + values: Vec, + ) -> StorageResult { + self.database.update_table_row(table, record, values) + } +} diff --git a/src/executor/expression.rs b/src/executor/expression.rs index ab2dc9d..86e452d 100644 --- a/src/executor/expression.rs +++ b/src/executor/expression.rs @@ -1,12 +1,12 @@ use crate::{ - core::{ - OwnedTableRecord, TableKey, TableSchema, Tuple, TupleView, Value, access::RecordAccess, - }, + core::{OwnedTableRecord, TableKey, TableSchema, Tuple, TupleView, Value}, planner::{BoundColumn, PlannedExpression, UpdateAssignment}, sql_parser::parser::op::Op, }; -use super::{ExecutionOutput, ExecutorError, ExecutorResult, ExecutorRow, RowStream}; +use super::{ + ExecutionDatabase, ExecutionOutput, ExecutorError, ExecutorResult, ExecutorRow, RowStream, +}; /// Evaluates one planned scalar expression against a record. /// @@ -59,8 +59,8 @@ pub(super) fn offset_rows(mut rows: RowStream, mut remaining: usize) -> RowStrea /// /// Each value row is evaluated, expanded into the target table layout, and then /// handed to storage for validation and insertion. -pub(super) fn execute_insert_values( - records: &R, +pub(super) fn execute_insert_values( + records: &ExecutionDatabase<'_>, table: TableSchema, columns: Vec, values: Vec>, @@ -100,8 +100,8 @@ pub(super) fn execute_insert_values( } /// Executes an `UPDATE` plan by consuming and mutating one target row at a time. -pub(super) fn execute_update( - records: &R, +pub(super) fn execute_update( + records: &ExecutionDatabase<'_>, table: TableSchema, assignments: Vec, target_rows: RowStream, @@ -136,8 +136,8 @@ pub(super) fn execute_update( } /// Executes a `DELETE` plan by consuming and deleting one target row at a time. -pub(super) fn execute_delete( - records: &R, +pub(super) fn execute_delete( + records: &ExecutionDatabase<'_>, table: TableSchema, target_rows: RowStream, ) -> ExecutorResult { diff --git a/src/executor/mod.rs b/src/executor/mod.rs index e0f1270..a215964 100644 --- a/src/executor/mod.rs +++ b/src/executor/mod.rs @@ -13,8 +13,8 @@ use crate::{ core::{ - Database, OwnedTableRecord, TableKey, TableRecord as BorrowedTableRecord, Tuple, Value, - access::ExecutionAccess, + Database, IndexKeyRange, IndexSchema, OwnedTableRecord, TableKey, TableKeyRange, + TableRecord as BorrowedTableRecord, TableSchema, Transaction, Tuple, TupleSchema, Value, error::{StorageError, StorageResult}, }, planner::PhysicalPlan, @@ -278,19 +278,118 @@ impl std::fmt::Display for ExecutionOutput { } } -/// Executes physical query plans against a database handle. +/// Executes physical query plans through a relational access gateway. /// -/// The executor borrows a [`Database`] and performs catalog, table, and index -/// operations through that handle. It owns no transaction state; mutation -/// ordering is encoded directly in each operator implementation. +/// The executor owns no transaction state; the caller supplies either a +/// transaction-scoped gateway or, temporarily during migration, a database +/// handle for non-transactional reads. pub struct Executor<'db> { - database: &'db dyn ExecutionAccess, + database: ExecutionDatabase<'db>, +} + +#[derive(Clone, Copy)] +pub(super) enum ExecutionDatabase<'db> { + Unscoped(&'db Database), + Transaction(&'db Transaction<'db>), +} + +impl ExecutionDatabase<'_> { + fn create_table(&self, name: &str, row: TupleSchema) -> StorageResult { + match self { + Self::Unscoped(database) => database.create_table(name, row), + Self::Transaction(transaction) => transaction.create_table(name, row), + } + } + + fn create_index( + &self, + name: &str, + table: &str, + columns: &[&str], + ) -> StorageResult { + match self { + Self::Unscoped(database) => database.create_index(name, table, columns), + Self::Transaction(transaction) => transaction.create_index(name, table, columns), + } + } + + fn scan_table( + &self, + table: &TableSchema, + ) -> StorageResult { + match self { + Self::Unscoped(database) => database.scan_table(table), + Self::Transaction(transaction) => transaction.scan_table(table), + } + } + + fn scan_table_range( + &self, + table: &TableSchema, + range: TableKeyRange, + ) -> StorageResult { + match self { + Self::Unscoped(database) => database.scan_table_range(table, range), + Self::Transaction(transaction) => transaction.scan_table_range(table, range), + } + } + + fn scan_index( + &self, + table: &TableSchema, + index: &IndexSchema, + range: IndexKeyRange, + ) -> StorageResult { + match self { + Self::Unscoped(database) => database.scan_index(table, index, range), + Self::Transaction(transaction) => transaction.scan_index(table, index, range), + } + } + + fn insert_table_row( + &self, + table: &TableSchema, + values: Vec, + ) -> StorageResult { + match self { + Self::Unscoped(database) => database.insert_table_row(table, values), + Self::Transaction(transaction) => transaction.insert_table_row(table, values), + } + } + + fn delete_table_row( + &self, + table: &TableSchema, + record: &OwnedTableRecord, + ) -> StorageResult<()> { + match self { + Self::Unscoped(database) => database.delete_table_row(table, record), + Self::Transaction(transaction) => transaction.delete_table_row(table, record), + } + } + + fn update_table_row( + &self, + table: &TableSchema, + record: &OwnedTableRecord, + values: Vec, + ) -> StorageResult { + match self { + Self::Unscoped(database) => database.update_table_row(table, record, values), + Self::Transaction(transaction) => transaction.update_table_row(table, record, values), + } + } } impl<'db> Executor<'db> { - /// Creates an executor that runs plans against `database`. - pub fn new(database: &'db Database) -> Self { - Self { database } + /// Creates an executor for legacy unscoped reads and focused executor tests. + pub(crate) fn new(database: &'db Database) -> Self { + Self { database: ExecutionDatabase::Unscoped(database) } + } + + /// Creates an executor scoped to an active transaction. + pub(crate) fn in_transaction(transaction: &'db Transaction<'db>) -> Self { + Self { database: ExecutionDatabase::Transaction(transaction) } } /// Executes a physical plan and returns its output. @@ -313,15 +412,20 @@ impl<'db> Executor<'db> { } PhysicalPlan::Values { rows } => execute_values(rows), PhysicalPlan::InsertValues { table, columns, values } => { - execute_insert_values(self.database, table, columns, values) + execute_insert_values(&self.database, table, columns, values) } PhysicalPlan::Update { table, assignments, input } => { let output_inner = self.execute(*input)?; - execute_update(self.database, table, assignments, output_inner.into_rows("UPDATE")?) + execute_update( + &self.database, + table, + assignments, + output_inner.into_rows("UPDATE")?, + ) } PhysicalPlan::Delete { table, input } => { let output_inner = self.execute(*input)?; - execute_delete(self.database, table, output_inner.into_rows("DELETE")?) + execute_delete(&self.database, table, output_inner.into_rows("DELETE")?) } PhysicalPlan::OneRow => Ok(ExecutionOutput::Rows { rows: Box::new(std::iter::once_with(|| empty_record(0))), diff --git a/src/executor/tests.rs b/src/executor/tests.rs index ad376b2..cca95da 100644 --- a/src/executor/tests.rs +++ b/src/executor/tests.rs @@ -6,7 +6,7 @@ use super::*; use crate::{ core::{ ColumnSchema, DataType, OwnedTableRecord, PAGE_SIZE, TableKey, Tuple, TupleSchema, - access::{DdlAccess, SchemaAccess}, + access::CatalogRead, error::{ConstraintError, InternalError, InvariantViolation, StorageError}, }, error::DatabaseError, @@ -2037,11 +2037,13 @@ fn uncommitted_flushed_insert_is_undone_during_recovery() { database.create_table("users", users_schema()).unwrap(); database.create_index("idx_users_name", "users", &["name"]).unwrap(); database.flush().unwrap(); - database.begin_transaction().unwrap(); + let txn_id = database.begin_transaction().unwrap(); let table = database.table_schema_by_name("users").unwrap(); + let transaction = database.transaction(txn_id); + let access = ExecutionDatabase::Transaction(&transaction); execute_insert_values( - &database, + &access, table, vec![ bound("id", 0, DataType::Integer), diff --git a/src/planner/mod.rs b/src/planner/mod.rs index a5c7485..d4b2e49 100644 --- a/src/planner/mod.rs +++ b/src/planner/mod.rs @@ -42,7 +42,7 @@ use crate::{ core::{ ColumnSchema, DataType, Database, IndexKeyBound, IndexKeyRange, IndexSchema, TableKey, TableKeyBound, TableKeyRange, TableSchema, Tuple, TupleSchema, Value, - access::SchemaAccess, + access::CatalogRead, error::{InvalidArgumentError, StorageError}, }, relational::cursor::encode_index_entry_key, diff --git a/src/planner/planning.rs b/src/planner/planning.rs index c7acdf6..f87c403 100644 --- a/src/planner/planning.rs +++ b/src/planner/planning.rs @@ -7,7 +7,7 @@ use super::*; /// perform writes. Even DDL and DML statements are represented only as plan /// nodes until an executor runs the physical plan. pub struct Planner<'db> { - schema: &'db dyn SchemaAccess, + schema: &'db dyn CatalogRead, } impl<'db> Planner<'db> { @@ -24,7 +24,7 @@ impl<'db> Planner<'db> { /// Keeping this constructor crate-visible lets focused tests and other /// internal planning clients provide lightweight schema access without /// constructing a storage-backed database. - pub(crate) fn with_schema(schema: &'db dyn SchemaAccess) -> Self { + pub(crate) fn with_schema(schema: &'db dyn CatalogRead) -> Self { Self { schema } } diff --git a/src/planner/tests.rs b/src/planner/tests.rs index 9f7dff7..b21e872 100644 --- a/src/planner/tests.rs +++ b/src/planner/tests.rs @@ -2,7 +2,7 @@ use tempfile::tempdir; use super::*; use crate::{ - core::{ColumnSchema, DataType, access::DdlAccess}, + core::{ColumnSchema, DataType}, sql_parser::parser::Parser, }; @@ -45,7 +45,7 @@ fn database_with_users() -> (tempfile::TempDir, Database) { struct EmptyCatalog; -impl SchemaAccess for EmptyCatalog { +impl CatalogRead for EmptyCatalog { fn table_schema_by_name(&self, name: &str) -> Result { Err(StorageError::InvalidArgument(InvalidArgumentError::TableNotFound { name: name.to_owned(), diff --git a/src/relational/catalog_manager.rs b/src/relational/catalog_manager.rs index 2fba650..6d03b9e 100644 --- a/src/relational/catalog_manager.rs +++ b/src/relational/catalog_manager.rs @@ -15,20 +15,20 @@ use crate::relational::{ }, cursor::{IndexCursor, TableCursor}, }; -use crate::storage::pager::Pager; +use crate::storage::engine::Storage; /// Internal catalog manager for one database file. /// -/// `CatalogManager` owns the low-level pager and manages catalog metadata for +/// `CatalogManager` owns the low-level storage and manages catalog metadata for /// table and index B+-trees. #[derive(Clone)] pub struct CatalogManager { - pager: Pager, + storage: Storage, } impl CatalogManager { - pub(crate) fn from_pager(pager: Pager) -> StorageResult { - let manager = Self { pager }; + pub(crate) fn from_storage(storage: Storage) -> StorageResult { + let manager = Self { storage }; manager.initialize_or_validate_system_catalog()?; manager.validate_page_formats()?; Ok(manager) @@ -36,12 +36,12 @@ impl CatalogManager { /// Returns the database-file path associated with this manager. pub fn path(&self) -> &Path { - self.pager.path() + self.storage.path() } /// Flushes all dirty, currently unpinned pages to disk. pub fn flush(&self) -> StorageResult<()> { - self.pager.flush() + self.storage.flush() } /// Creates a cataloged table, allocates its root page, and records its columns. @@ -54,7 +54,7 @@ impl CatalogManager { } let table_id = self.next_object_id()?; - let root_page_id = self.pager.create_tree()?.root_page_id(); + let root_page_id = self.storage.create_tree()?.root_page_id(); let schema = TableSchema { table_id, name: name.to_owned(), root_page_id, row }; self.insert_table_catalog_row(&schema.catalog_row())?; @@ -131,7 +131,7 @@ impl CatalogManager { }); } - let root_page_id = self.pager.create_tree()?.root_page_id(); + let root_page_id = self.storage.create_tree()?.root_page_id(); let schema = IndexSchema { index_id, name: name.to_owned(), @@ -160,16 +160,16 @@ impl CatalogManager { } fn initialize_or_validate_system_catalog(&self) -> StorageResult<()> { - match self.pager.opened_page_count() { + match self.storage.opened_page_count() { 0 => Err(crate::storage::database_header::missing_header()), 1 => self.initialize_system_catalog(), - 2..=3 => Err(missing_system_catalog_root(self.pager.opened_page_count())), + 2..=3 => Err(missing_system_catalog_root(self.storage.opened_page_count())), _ => Ok(()), } } fn initialize_system_root(&self, expected_page_id: PageId) -> StorageResult<()> { - let actual_page_id = self.pager.create_tree()?.root_page_id(); + let actual_page_id = self.storage.create_tree()?.root_page_id(); if actual_page_id == expected_page_id { Ok(()) } else { @@ -202,18 +202,18 @@ impl CatalogManager { } fn table_cursor(&self, root_page_id: PageId) -> TableCursor { - TableCursor::new(self.pager.tree_cursor(root_page_id)) + TableCursor::new(self.storage.tree_cursor(root_page_id)) } fn index_cursor(&self, root_page_id: PageId) -> IndexCursor { - IndexCursor::new(self.pager.tree_cursor(root_page_id)) + IndexCursor::new(self.storage.tree_cursor(root_page_id)) } fn validate_page_formats(&self) -> StorageResult<()> { let system_roots = [SYS_TABLES_ROOT_PAGE_ID, SYS_INDEXES_ROOT_PAGE_ID, SYS_COLUMNS_ROOT_PAGE_ID]; for root_page_id in system_roots { - self.pager.validate_tree_page_formats(root_page_id)?; + self.storage.validate_tree_page_formats(root_page_id)?; } let mut roots = system_roots.to_vec(); @@ -226,7 +226,7 @@ impl CatalogManager { if system_roots.contains(&root_page_id) { continue; } - self.pager.validate_tree_page_formats(root_page_id)?; + self.storage.validate_tree_page_formats(root_page_id)?; } Ok(()) } @@ -498,7 +498,7 @@ mod tests { use crate::storage::{database_header::DatabaseHeader, disk_manager::DiskManager}; fn open(path: impl AsRef) -> StorageResult { - CatalogManager::from_pager(Pager::open_or_create(path)?) + CatalogManager::from_storage(Storage::open_or_create(path)?) } #[test] @@ -506,7 +506,7 @@ mod tests { let file = NamedTempFile::new().unwrap(); let manager = open(file.path()).unwrap(); - assert_eq!(manager.pager.create_tree().unwrap().root_page_id(), 4); + assert_eq!(manager.storage.create_tree().unwrap().root_page_id(), 4); let mut tables = manager.table_cursor(SYS_TABLES_ROOT_PAGE_ID); assert_table_catalog_row( diff --git a/src/relational/index_manager.rs b/src/relational/index_manager.rs index f67ce30..b58b2e7 100644 --- a/src/relational/index_manager.rs +++ b/src/relational/index_manager.rs @@ -4,12 +4,72 @@ use crate::core::{ }; use crate::relational::{catalog_manager::CatalogManager, cursor::encode_index_entry_key}; -/// Internal manager for secondary-index data maintenance. +/// Creates and backfills a secondary index. +pub(crate) fn create_index( + catalog: &CatalogManager, + name: &str, + table_name: &str, + columns: &[&str], +) -> StorageResult { + let table = catalog.table_schema_by_name(table_name)?; + let index = catalog.create_index(name, table_name, columns)?; + backfill_index(catalog, &table, &index)?; + Ok(index) +} + +pub(crate) fn insert_index_entries( + catalog: &CatalogManager, + table: &TableSchema, + record: &OwnedTableRecord, +) -> StorageResult<()> { + for index in catalog.index_schemas_for_table(table)? { + let key = index_key_from_record(table, &index, record)?; + let key = encode_index_entry_key(&key, record.table_key); + let mut index_cursor = catalog.index_cursor_by_name(&index.name)?; + index_cursor.insert(&key, record.table_key)?; + } + Ok(()) +} + +pub(crate) fn delete_index_entries( + catalog: &CatalogManager, + table: &TableSchema, + record: &OwnedTableRecord, +) -> StorageResult<()> { + for index in catalog.index_schemas_for_table(table)? { + let key = index_key_from_record(table, &index, record)?; + let key = encode_index_entry_key(&key, record.table_key); + let mut index_cursor = catalog.index_cursor_by_name(&index.name)?; + index_cursor.delete(&key)?; + } + Ok(()) +} + +fn backfill_index( + catalog: &CatalogManager, + table: &TableSchema, + index: &IndexSchema, +) -> StorageResult<()> { + let mut table_cursor = catalog.table_cursor_by_name(&table.name)?; + let mut index_cursor = catalog.index_cursor_by_name(&index.name)?; + while let Some(record) = table_cursor.next_record()? { + let key = index_key_from_table_record(table, index, &record)?; + let table_key = record.table_key(); + let key = encode_index_entry_key(&key, table_key); + index_cursor.insert(&key, table_key)?; + } + Ok(()) +} + +/// Compatibility wrapper retained only for focused module tests during the +/// manager-to-operation migration. +#[cfg(test)] #[derive(Clone)] pub(crate) struct IndexManager { catalog: CatalogManager, } +#[cfg(test)] impl IndexManager { pub(crate) fn new(catalog: CatalogManager) -> Self { Self { catalog } @@ -21,54 +81,7 @@ impl IndexManager { table_name: &str, columns: &[&str], ) -> StorageResult { - let table = self.catalog.table_schema_by_name(table_name)?; - let index = self.catalog.create_index(name, table_name, columns)?; - self.backfill_index(&table, &index)?; - Ok(index) - } - - pub(crate) fn insert_index_entries( - &self, - table: &TableSchema, - record: &OwnedTableRecord, - ) -> StorageResult<()> { - for index in self.catalog.index_schemas_for_table(table)? { - let key = index_key_from_record(table, &index, record)?; - let key = encode_index_entry_key(&key, record.table_key); - let mut index_cursor = self.catalog.index_cursor_by_name(&index.name)?; - index_cursor.insert(&key, record.table_key)?; - } - - Ok(()) - } - - pub(crate) fn delete_index_entries( - &self, - table: &TableSchema, - record: &OwnedTableRecord, - ) -> StorageResult<()> { - for index in self.catalog.index_schemas_for_table(table)? { - let key = index_key_from_record(table, &index, record)?; - let key = encode_index_entry_key(&key, record.table_key); - let mut index_cursor = self.catalog.index_cursor_by_name(&index.name)?; - index_cursor.delete(&key)?; - } - - Ok(()) - } - - fn backfill_index(&self, table: &TableSchema, index: &IndexSchema) -> StorageResult<()> { - let mut table_cursor = self.catalog.table_cursor_by_name(&table.name)?; - let mut index_cursor = self.catalog.index_cursor_by_name(&index.name)?; - - while let Some(record) = table_cursor.next_record()? { - let key = index_key_from_table_record(table, index, &record)?; - let table_key = record.table_key(); - let key = encode_index_entry_key(&key, table_key); - index_cursor.insert(&key, table_key)?; - } - - Ok(()) + create_index(&self.catalog, name, table_name, columns) } } @@ -137,10 +150,10 @@ mod tests { use super::*; use crate::core::{ColumnSchema, DataType, TupleSchema, Value}; use crate::relational::{catalog_manager::CatalogManager, record_manager::RecordManager}; - use crate::storage::pager::Pager; + use crate::storage::engine::Storage; fn open(path: impl AsRef) -> StorageResult<(CatalogManager, IndexManager)> { - let catalog = CatalogManager::from_pager(Pager::open_or_create(path)?)?; + let catalog = CatalogManager::from_storage(Storage::open_or_create(path)?)?; let indexes = IndexManager::new(catalog.clone()); Ok((catalog, indexes)) } @@ -182,7 +195,7 @@ mod tests { fn create_index_backfills_existing_table_rows() { let file = NamedTempFile::new().unwrap(); let (catalog, indexes) = open(file.path()).unwrap(); - let records = RecordManager::new(catalog.clone(), indexes.clone()); + let records = RecordManager::new(catalog.clone()); let table = catalog.create_table("users", users_schema()).unwrap(); records .insert_table_row( @@ -208,7 +221,7 @@ mod tests { fn create_index_backfills_duplicate_index_values() { let file = NamedTempFile::new().unwrap(); let (catalog, indexes) = open(file.path()).unwrap(); - let records = RecordManager::new(catalog.clone(), indexes.clone()); + let records = RecordManager::new(catalog.clone()); let table = catalog.create_table("users", users_schema()).unwrap(); records .insert_table_row( diff --git a/src/relational/record_manager.rs b/src/relational/record_manager.rs index fdcbcd7..8f78b14 100644 --- a/src/relational/record_manager.rs +++ b/src/relational/record_manager.rs @@ -7,14 +7,14 @@ use crate::core::{ use crate::relational::{ catalog_manager::CatalogManager, cursor::{IndexCursor, TableCursor}, - index_manager::IndexManager, + index_manager, }; -/// Internal manager for table record access and mutation. +/// Compatibility wrapper retained for focused relational tests. +#[cfg(test)] #[derive(Clone)] pub(crate) struct RecordManager { catalog: CatalogManager, - indexes: IndexManager, } /// Iterator over records in one table. @@ -37,100 +37,136 @@ pub(crate) struct IndexScan { done: bool, } -impl RecordManager { - pub(crate) fn new(catalog: CatalogManager, indexes: IndexManager) -> Self { - Self { catalog, indexes } +pub(crate) fn scan_table( + catalog: &CatalogManager, + table: &TableSchema, +) -> StorageResult { + scan_table_range(catalog, table, TableKeyRange::unbounded()) +} + +pub(crate) fn scan_table_range( + catalog: &CatalogManager, + table: &TableSchema, + range: TableKeyRange, +) -> StorageResult { + let cursor = catalog.table_cursor_by_name(&table.name)?; + let observed_mutation_epoch = cursor.mutation_epoch(); + Ok(TableScan { + cursor, + range, + last_table_key: None, + observed_mutation_epoch, + initialized: false, + done: false, + }) +} + +pub(crate) fn scan_index( + catalog: &CatalogManager, + table: &TableSchema, + index: &IndexSchema, + key_range: IndexKeyRange, +) -> StorageResult { + Ok(IndexScan { + table: table.clone(), + table_cursor: catalog.table_cursor_by_name(&table.name)?, + index_cursor: catalog.index_cursor_by_name(&index.name)?, + key_range, + initialized: false, + done: false, + }) +} + +pub(crate) fn insert_table_row( + catalog: &CatalogManager, + table: &TableSchema, + values: Vec, +) -> StorageResult { + validate_table_row(table, &values)?; + + let table_key = table_key_from_values(table, &values)?; + let record = Tuple::new(values).to_bytes()?; + let mut table_cursor = catalog.table_cursor_by_name(&table.name)?; + table_cursor.insert(table_key, &record)?; + + let record = OwnedTableRecord { table_key, record: record.into_boxed_slice() }; + index_manager::insert_index_entries(catalog, table, &record)?; + Ok(record) +} + +pub(crate) fn delete_table_row( + catalog: &CatalogManager, + table: &TableSchema, + record: &OwnedTableRecord, +) -> StorageResult<()> { + index_manager::delete_index_entries(catalog, table, record)?; + let mut table_cursor = catalog.table_cursor_by_name(&table.name)?; + table_cursor.delete(record.table_key) +} + +pub(crate) fn update_table_row( + catalog: &CatalogManager, + table: &TableSchema, + record: &OwnedTableRecord, + values: Vec, +) -> StorageResult { + validate_table_row(table, &values)?; + let updated_table_key = table_key_from_values(table, &values)?; + if updated_table_key != record.table_key { + return Err(StorageError::InvalidArgument(InvalidArgumentError::PrimaryKeyUpdate { + table: table.name.clone(), + column: table.row.columns[0].name.clone(), + })); } + let updated = Tuple::new(values).to_bytes()?; + let updated = + OwnedTableRecord { table_key: record.table_key, record: updated.into_boxed_slice() }; + + index_manager::delete_index_entries(catalog, table, record)?; + let mut table_cursor = catalog.table_cursor_by_name(&table.name)?; + table_cursor.update(record.table_key, &updated.record)?; + index_manager::insert_index_entries(catalog, table, &updated)?; + + Ok(updated) +} + +#[cfg(test)] +impl RecordManager { + pub(crate) fn new(catalog: CatalogManager) -> Self { + Self { catalog } + } pub(crate) fn scan_table(&self, table: &TableSchema) -> StorageResult { - self.scan_table_range(table, TableKeyRange::unbounded()) + scan_table(&self.catalog, table) } - pub(crate) fn scan_table_range( &self, table: &TableSchema, range: TableKeyRange, ) -> StorageResult { - let cursor = self.catalog.table_cursor_by_name(&table.name)?; - let observed_mutation_epoch = cursor.mutation_epoch(); - Ok(TableScan { - cursor, - range, - last_table_key: None, - observed_mutation_epoch, - initialized: false, - done: false, - }) - } - - pub(crate) fn scan_index( - &self, - table: &TableSchema, - index: &IndexSchema, - key_range: IndexKeyRange, - ) -> StorageResult { - Ok(IndexScan { - table: table.clone(), - table_cursor: self.catalog.table_cursor_by_name(&table.name)?, - index_cursor: self.catalog.index_cursor_by_name(&index.name)?, - key_range, - initialized: false, - done: false, - }) + scan_table_range(&self.catalog, table, range) } - pub(crate) fn insert_table_row( &self, table: &TableSchema, values: Vec, ) -> StorageResult { - validate_table_row(table, &values)?; - - let table_key = table_key_from_values(table, &values)?; - let record = Tuple::new(values).to_bytes()?; - let mut table_cursor = self.catalog.table_cursor_by_name(&table.name)?; - table_cursor.insert(table_key, &record)?; - - let record = OwnedTableRecord { table_key, record: record.into_boxed_slice() }; - self.indexes.insert_index_entries(table, &record)?; - Ok(record) + insert_table_row(&self.catalog, table, values) } - pub(crate) fn delete_table_row( &self, table: &TableSchema, record: &OwnedTableRecord, ) -> StorageResult<()> { - self.indexes.delete_index_entries(table, record)?; - let mut table_cursor = self.catalog.table_cursor_by_name(&table.name)?; - table_cursor.delete(record.table_key) + delete_table_row(&self.catalog, table, record) } - pub(crate) fn update_table_row( &self, table: &TableSchema, record: &OwnedTableRecord, values: Vec, ) -> StorageResult { - validate_table_row(table, &values)?; - let updated_table_key = table_key_from_values(table, &values)?; - if updated_table_key != record.table_key { - return Err(StorageError::InvalidArgument(InvalidArgumentError::PrimaryKeyUpdate { - table: table.name.clone(), - column: table.row.columns[0].name.clone(), - })); - } - - let updated = Tuple::new(values).to_bytes()?; - let updated = - OwnedTableRecord { table_key: record.table_key, record: updated.into_boxed_slice() }; - - self.indexes.delete_index_entries(table, record)?; - let mut table_cursor = self.catalog.table_cursor_by_name(&table.name)?; - table_cursor.update(record.table_key, &updated.record)?; - self.indexes.insert_index_entries(table, &updated)?; - - Ok(updated) + update_table_row(&self.catalog, table, record, values) } } @@ -411,12 +447,11 @@ mod tests { cursor::{IndexCursor, encode_index_entry_key}, index_manager::IndexManager, }; - use crate::storage::pager::Pager; + use crate::storage::engine::Storage; fn open(path: impl AsRef) -> StorageResult<(CatalogManager, RecordManager)> { - let catalog = CatalogManager::from_pager(Pager::open_or_create(path)?)?; - let indexes = IndexManager::new(catalog.clone()); - let records = RecordManager::new(catalog.clone(), indexes); + let catalog = CatalogManager::from_storage(Storage::open_or_create(path)?)?; + let records = RecordManager::new(catalog.clone()); Ok((catalog, records)) } diff --git a/src/session.rs b/src/session.rs index 62d536b..9d2d9b8 100644 --- a/src/session.rs +++ b/src/session.rs @@ -141,10 +141,12 @@ impl<'db> Session<'db> { txn_id: u64, plan: PhysicalPlan, ) -> Result> { - let savepoint = self.database.statement_savepoint(txn_id)?; - match Executor::new(self.database).execute(plan) { + let transaction = self.database.transaction(txn_id); + debug_assert_eq!(transaction.id(), txn_id); + let savepoint = transaction.statement_savepoint()?; + match Executor::in_transaction(&transaction).execute(plan) { Ok(output) => { - if self.database.transaction_is_poisoned(txn_id)? { + if transaction.is_poisoned()? { return self.rollback_failed_explicit_transaction_statement( savepoint, transaction_poisoned(txn_id).into(), @@ -174,7 +176,8 @@ impl<'db> Session<'db> { plan: PhysicalPlan, ) -> Result> { let txn_id = self.database.begin_transaction()?; - match Executor::new(self.database).execute(plan) { + let transaction = self.database.transaction(txn_id); + match Executor::in_transaction(&transaction).execute(plan) { Ok(output) => match self.database.commit_transaction(txn_id) { Ok(()) => Ok(output), Err(commit_error) => { diff --git a/src/storage/pager.rs b/src/storage/engine.rs similarity index 56% rename from src/storage/pager.rs rename to src/storage/engine.rs index 9407092..3987077 100644 --- a/src/storage/pager.rs +++ b/src/storage/engine.rs @@ -1,25 +1,28 @@ use std::{path::Path, rc::Rc}; use crate::core::{PageId, error::StorageResult}; +#[cfg(test)] +use crate::storage::disk_manager::DiskManagerError; use crate::storage::{ btree::{TreeCursor, initialize_empty_root, validate_tree_page_formats}, database_header::{DATABASE_HEADER_PAGE_ID, DatabaseHeader, missing_header}, disk_manager::DiskManager, + log_manager::TxnId, page_cache::PageCache, storage_runtime::StorageRuntime, - transaction_runtime::TransactionRuntime, + transaction_manager::TransactionSavepoint, }; const DEFAULT_PAGE_CACHE_SIZE: usize = 16384; /// Configuration for [`crate::core::Database`]. #[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub struct PagerOptions { +pub struct StorageOptions { /// Number of frames to preallocate in the page cache. pub cache_frames: usize, } -impl Default for PagerOptions { +impl Default for StorageOptions { fn default() -> Self { Self { cache_frames: DEFAULT_PAGE_CACHE_SIZE } } @@ -27,25 +30,25 @@ impl Default for PagerOptions { /// Storage-engine handle for one database file. /// -/// `Pager` owns the disk manager and page cache indirectly, and is responsible +/// `Storage` owns the disk manager and page cache indirectly, and is responsible /// only for producing raw B+-tree cursors rooted at specific page ids. #[derive(Clone)] -pub(crate) struct Pager { +pub(crate) struct Storage { runtime: Rc, page_cache: PageCache, opened_page_count: u64, } -impl Pager { +impl Storage { /// Creates a new database file and initializes its database header. pub(crate) fn create(path: impl AsRef) -> StorageResult { - Self::create_with_options(path, PagerOptions::default()) + Self::create_with_options(path, StorageOptions::default()) } /// Creates a new database file with explicit cache settings. pub(crate) fn create_with_options( path: impl AsRef, - options: PagerOptions, + options: StorageOptions, ) -> StorageResult { let path = path.as_ref().to_path_buf(); let mut disk_manager = DiskManager::create_new(&path)?; @@ -54,15 +57,15 @@ impl Pager { Self::from_disk_manager(path, disk_manager, options) } - /// Opens an existing pager with default options. + /// Opens an existing storage with default options. pub(crate) fn open(path: impl AsRef) -> StorageResult { - Self::open_with_options(path, PagerOptions::default()) + Self::open_with_options(path, StorageOptions::default()) } - /// Opens an existing pager with explicit cache settings. + /// Opens an existing storage with explicit cache settings. pub(crate) fn open_with_options( path: impl AsRef, - options: PagerOptions, + options: StorageOptions, ) -> StorageResult { let path = path.as_ref().to_path_buf(); let mut disk_manager = DiskManager::open_existing(&path)?; @@ -71,15 +74,15 @@ impl Pager { Self::from_disk_manager(path, disk_manager, options) } - /// Opens a pager, creating and initializing an empty file if needed. + /// Opens a storage, creating and initializing an empty file if needed. pub(crate) fn open_or_create(path: impl AsRef) -> StorageResult { - Self::open_or_create_with_options(path, PagerOptions::default()) + Self::open_or_create_with_options(path, StorageOptions::default()) } - /// Opens a pager with explicit cache settings, creating an empty file if needed. + /// Opens a storage with explicit cache settings, creating an empty file if needed. pub(crate) fn open_or_create_with_options( path: impl AsRef, - options: PagerOptions, + options: StorageOptions, ) -> StorageResult { let path = path.as_ref().to_path_buf(); let mut disk_manager = DiskManager::new(&path)?; @@ -95,7 +98,7 @@ impl Pager { fn from_disk_manager( path: std::path::PathBuf, disk_manager: DiskManager, - options: PagerOptions, + options: StorageOptions, ) -> StorageResult { let opened_page_count = disk_manager.page_count(); let runtime = Rc::new(StorageRuntime::new(path, disk_manager)?); @@ -103,12 +106,12 @@ impl Pager { Ok(Self { runtime, page_cache, opened_page_count }) } - /// Returns the database-file path associated with this pager. + /// Returns the database-file path associated with this storage. pub(crate) fn path(&self) -> &Path { self.runtime.path() } - /// Returns the page count observed when this pager was opened. + /// Returns the page count observed when this storage was opened. pub(crate) fn opened_page_count(&self) -> u64 { self.opened_page_count } @@ -120,8 +123,65 @@ impl Pager { Ok(()) } - pub(crate) fn transaction_runtime(&self) -> TransactionRuntime { - TransactionRuntime::new(Rc::clone(&self.runtime), self.page_cache.clone()) + #[cfg(test)] + pub(crate) fn unlock_for_crash_for_test(&self) -> Result<(), DiskManagerError> { + self.runtime.unlock_for_crash_for_test() + } + + pub(crate) fn begin_transaction(&self) -> StorageResult { + self.runtime.begin_transaction() + } + + pub(crate) fn commit_transaction(&self, txn_id: TxnId) -> StorageResult<()> { + self.runtime.commit_transaction(txn_id) + } + + pub(crate) fn active_transaction_id(&self) -> Option { + self.runtime.active_transaction_id() + } + + pub(crate) fn transaction_is_poisoned(&self, txn_id: TxnId) -> StorageResult { + self.runtime.transaction_is_poisoned(txn_id) + } + + pub(crate) fn statement_savepoint(&self, txn_id: TxnId) -> StorageResult { + self.runtime.statement_savepoint(txn_id) + } + + pub(crate) fn rollback_to_savepoint( + &self, + savepoint: TransactionSavepoint, + ) -> StorageResult<()> { + let undo_pages = self.runtime.rollback_to_savepoint(savepoint)?; + if let Err(err) = self.page_cache.restore_rollback_pages(undo_pages) { + self.runtime.record_transaction_failure(); + return Err(err.into()); + } + self.runtime.complete_savepoint_rollback(savepoint) + } + + pub(crate) fn rollback_transaction(&self, txn_id: TxnId) -> StorageResult<()> { + let rollback = self.runtime.prepare_rollback_pages(txn_id)?; + self.page_cache.restore_rollback_pages(rollback.pages)?; + self.page_cache.flush_all()?; + self.runtime.sync_database_file()?; + self.runtime.finish_rollback(txn_id)?; + Ok(()) + } + + #[cfg(test)] + pub(crate) fn force_next_lsn_exhausted_for_test(&self) { + self.runtime.force_next_lsn_exhausted_for_test(); + } + + #[cfg(test)] + pub(crate) fn fail_next_savepoint_rollback_for_test(&self) { + self.runtime.fail_next_savepoint_rollback_for_test(); + } + + #[cfg(test)] + pub(crate) fn fail_next_wal_flush_for_test(&self) { + self.runtime.fail_next_wal_flush_for_test(); } /// Creates a new empty raw tree and returns a cursor rooted at it. @@ -168,17 +228,17 @@ mod tests { #[test] fn opens_database_and_manages_raw_trees() { let file = NamedTempFile::new().unwrap(); - let pager = Pager::open_or_create(file.path()).unwrap(); - - assert_eq!(pager.opened_page_count(), 1); - assert_eq!(pager.create_tree().unwrap().root_page_id(), 1); - assert_eq!(pager.create_tree().unwrap().root_page_id(), 2); - pager.flush().unwrap(); - drop(pager); - - let pager = Pager::open(file.path()).unwrap(); - assert_eq!(pager.opened_page_count(), 3); - assert_eq!(pager.tree_cursor(1).root_page_id(), 1); - assert_eq!(pager.tree_cursor(2).root_page_id(), 2); + let storage = Storage::open_or_create(file.path()).unwrap(); + + assert_eq!(storage.opened_page_count(), 1); + assert_eq!(storage.create_tree().unwrap().root_page_id(), 1); + assert_eq!(storage.create_tree().unwrap().root_page_id(), 2); + storage.flush().unwrap(); + drop(storage); + + let storage = Storage::open(file.path()).unwrap(); + assert_eq!(storage.opened_page_count(), 3); + assert_eq!(storage.tree_cursor(1).root_page_id(), 1); + assert_eq!(storage.tree_cursor(2).root_page_id(), 2); } } diff --git a/src/storage/mod.rs b/src/storage/mod.rs index 12361ba..878d830 100644 --- a/src/storage/mod.rs +++ b/src/storage/mod.rs @@ -3,13 +3,13 @@ //! The storage layer is organized around four cooperating areas: //! //! - [`page`] defines the on-disk page format and typed read/write access. -//! - [`page_cache`] and [`pager`] coordinate cached pages with database-file I/O. +//! - [`page_cache`] and [`storage`] coordinate cached pages with database-file I/O. //! - [`btree`] implements byte-oriented B+-tree traversal and mutation. //! - [`log_manager`], [`transaction_manager`], and [`recovery`] enforce //! write-ahead logging and restore committed state after a crash. //! //! [`storage_runtime`] owns the concrete disk, log, and transaction managers. -//! [`transaction_runtime`] adds page-cache-aware rollback orchestration for the +//! `Storage` adds page-cache-aware rollback orchestration for the //! higher relational and session layers. These runtimes are intentionally //! crate-private: callers use the database facade rather than assembling //! storage components directly. @@ -17,14 +17,13 @@ pub(crate) mod btree; pub(crate) mod database_header; pub(crate) mod disk_manager; +pub(crate) mod engine; mod error; pub(crate) mod log_manager; pub(crate) mod overflow; pub(crate) mod page; pub(crate) mod page_cache; pub(crate) mod page_replacement; -pub(crate) mod pager; pub(crate) mod recovery; pub(crate) mod storage_runtime; pub(crate) mod transaction_manager; -pub(crate) mod transaction_runtime; diff --git a/src/storage/page_cache.rs b/src/storage/page_cache.rs index fa05c58..73ba880 100644 --- a/src/storage/page_cache.rs +++ b/src/storage/page_cache.rs @@ -506,7 +506,6 @@ mod tests { use crate::storage::page::format::PageKind; use crate::storage::page::{Leaf, Page, Write}; use crate::storage::storage_runtime::StorageRuntime; - use crate::storage::transaction_runtime::TransactionRuntime; /// Generates a deterministic page payload from a seed byte. fn page_with_pattern(seed: u8) -> [u8; PAGE_SIZE] { @@ -1095,9 +1094,8 @@ mod tests { ]; let (file, runtime) = create_disk_with_pages(&pages); let cache = PageCache::new(Rc::clone(&runtime), 2).unwrap(); - let transactions = TransactionRuntime::new(Rc::clone(&runtime), cache.clone()); - let txn_id = transactions.begin_transaction().unwrap(); + let txn_id = runtime.begin_transaction().unwrap(); { let guard = cache.fetch_page(0).unwrap(); guard.write().unwrap().page_mut()[PAGE_SIZE - 1] = 100; diff --git a/src/storage/transaction_runtime.rs b/src/storage/transaction_runtime.rs deleted file mode 100644 index d2d2cf8..0000000 --- a/src/storage/transaction_runtime.rs +++ /dev/null @@ -1,136 +0,0 @@ -use std::rc::Rc; - -use crate::core::error::StorageResult; -#[cfg(test)] -use crate::storage::disk_manager::DiskManagerError; -use crate::storage::{ - log_manager::TxnId, page_cache::PageCache, storage_runtime::StorageRuntime, - transaction_manager::TransactionSavepoint, -}; - -/// Transaction-facing runtime for a database file. -/// -/// `TransactionRuntime` owns the transaction lifecycle surface used by higher -/// layers. It keeps rollback orchestration close to the page cache and storage -/// runtime without routing transaction calls through catalog code. -#[derive(Clone)] -pub(crate) struct TransactionRuntime { - runtime: Rc, - page_cache: PageCache, -} - -impl TransactionRuntime { - pub(crate) fn new(runtime: Rc, page_cache: PageCache) -> Self { - Self { runtime, page_cache } - } - - #[cfg(test)] - pub(crate) fn unlock_for_crash_for_test(&self) -> Result<(), DiskManagerError> { - self.runtime.unlock_for_crash_for_test() - } - - pub(crate) fn begin_transaction(&self) -> StorageResult { - self.runtime.begin_transaction() - } - - pub(crate) fn commit_transaction(&self, txn_id: TxnId) -> StorageResult<()> { - self.runtime.commit_transaction(txn_id) - } - - pub(crate) fn active_transaction_id(&self) -> Option { - self.runtime.active_transaction_id() - } - - pub(crate) fn transaction_is_poisoned(&self, txn_id: TxnId) -> StorageResult { - self.runtime.transaction_is_poisoned(txn_id) - } - - pub(crate) fn statement_savepoint(&self, txn_id: TxnId) -> StorageResult { - self.runtime.statement_savepoint(txn_id) - } - - pub(crate) fn rollback_to_savepoint( - &self, - savepoint: TransactionSavepoint, - ) -> StorageResult<()> { - let undo_pages = self.runtime.rollback_to_savepoint(savepoint)?; - if let Err(err) = self.page_cache.restore_rollback_pages(undo_pages) { - self.runtime.record_transaction_failure(); - return Err(err.into()); - } - self.runtime.complete_savepoint_rollback(savepoint) - } - - pub(crate) fn rollback_transaction(&self, txn_id: TxnId) -> StorageResult<()> { - let rollback = self.runtime.prepare_rollback_pages(txn_id)?; - self.page_cache.restore_rollback_pages(rollback.pages)?; - self.page_cache.flush_all()?; - self.runtime.sync_database_file()?; - self.runtime.finish_rollback(txn_id)?; - Ok(()) - } - - #[cfg(test)] - pub(crate) fn force_next_lsn_exhausted_for_test(&self) { - self.runtime.force_next_lsn_exhausted_for_test(); - } - - #[cfg(test)] - pub(crate) fn fail_next_savepoint_rollback_for_test(&self) { - self.runtime.fail_next_savepoint_rollback_for_test(); - } - - #[cfg(test)] - pub(crate) fn fail_next_wal_flush_for_test(&self) { - self.runtime.fail_next_wal_flush_for_test(); - } -} - -#[cfg(test)] -mod tests { - use std::rc::Rc; - - use tempfile::NamedTempFile; - - use super::*; - use crate::core::PAGE_SIZE; - use crate::storage::{ - disk_manager::DiskManager, page_cache::PageCache, storage_runtime::StorageRuntime, - }; - - // Issue: `rollback_to_savepoint` truncates the manager's undo log before the - // page cache installs the returned images. If installation then fails (here a - // pinned one-frame cache cannot fetch the evicted page), full rollback has no - // undo image left and can make the failed statement's page change permanent. - #[test] - fn failed_savepoint_restore_remains_undoable_by_full_rollback() { - let file = NamedTempFile::new().unwrap(); - let before = [1; PAGE_SIZE]; - let other = [2; PAGE_SIZE]; - let mut disk = DiskManager::new(file.path()).unwrap(); - disk.ensure_page_exists(1).unwrap(); - disk.write_page(0, &before).unwrap(); - disk.write_page(1, &other).unwrap(); - let runtime = Rc::new(StorageRuntime::new(file.path().to_path_buf(), disk).unwrap()); - let cache = PageCache::new(Rc::clone(&runtime), 1).unwrap(); - let transactions = TransactionRuntime::new(Rc::clone(&runtime), cache.clone()); - - let txn_id = transactions.begin_transaction().unwrap(); - let savepoint = transactions.statement_savepoint(txn_id).unwrap(); - { - let page = cache.fetch_page(0).unwrap(); - page.write().unwrap().page_mut()[PAGE_SIZE - 1] = 99; - } - - // Evict the changed page and keep the only frame pinned so restoring it fails. - let pinned_other_page = cache.fetch_page(1).unwrap(); - assert!(transactions.rollback_to_savepoint(savepoint).is_err()); - drop(pinned_other_page); - - transactions.rollback_transaction(txn_id).unwrap(); - - let mut actual = [0; PAGE_SIZE]; - runtime.read_page(0, &mut actual).unwrap(); - assert_eq!(actual[PAGE_SIZE - 1], before[PAGE_SIZE - 1]); - } -}