Experimental architecture refactor

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<TableSchema>;
 
     fn index_schemas_for_table(&self, table: &TableSchema) -> StorageResult<Vec<IndexSchema>>;
 }
-
-pub(crate) trait DdlAccess {
-    fn create_table(&self, name: &str, row: TupleSchema) -> StorageResult<TableSchema>;
-
-    fn create_index(
-        &self,
-        name: &str,
-        table_name: &str,
-        columns: &[&str],
-    ) -> StorageResult<IndexSchema>;
-}
-
-pub(crate) trait RecordAccess {
-    fn scan_table(&self, table: &TableSchema) -> StorageResult<TableScan>;
-
-    fn scan_table_range(
-        &self,
-        table: &TableSchema,
-        range: TableKeyRange,
-    ) -> StorageResult<TableScan>;
-
-    fn scan_index(
-        &self,
-        table: &TableSchema,
-        index: &IndexSchema,
-        key_range: IndexKeyRange,
-    ) -> StorageResult<IndexScan>;
-
-    fn insert_table_row(
-        &self,
-        table: &TableSchema,
-        values: Vec<Value>,
-    ) -> StorageResult<OwnedTableRecord>;
-
-    fn delete_table_row(&self, table: &TableSchema, record: &OwnedTableRecord)
-    -> StorageResult<()>;
-
-    fn update_table_row(
-        &self,
-        table: &TableSchema,
-        record: &OwnedTableRecord,
-        values: Vec<Value>,
-    ) -> StorageResult<OwnedTableRecord>;
-}
-
-pub(crate) trait ExecutionAccess: DdlAccess + RecordAccess {}
-
-impl<T> ExecutionAccess for T where T: DdlAccess + RecordAccess {}

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<Path>) -> StorageResult<Self> {
-        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<Path>) -> StorageResult<Self> {
-        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<Path>) -> StorageResult<Self> {
-        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<Self> {
-        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<Self> {
+        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<TxnId> {
-        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<TransactionSavepoint> {
-        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<TxnId> {
-        self.transactions.active_transaction_id()
+        self.storage.active_transaction_id()
     }
 
     pub(crate) fn transaction_is_poisoned(&self, txn_id: TxnId) -> StorageResult<bool> {
-        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<TableSchema> {
         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<TableSchema> {
+impl Database {
+    pub(crate) fn create_table(&self, name: &str, row: TupleSchema) -> StorageResult<TableSchema> {
         self.catalog.create_table(name, row)
     }
 
-    fn create_index(
+    pub(crate) fn create_index(
         &self,
         name: &str,
         table_name: &str,
         columns: &[&str],
     ) -> StorageResult<IndexSchema> {
-        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<TableScan> {
-        self.records.scan_table(table)
+impl Database {
+    pub(crate) fn scan_table(&self, table: &TableSchema) -> StorageResult<TableScan> {
+        record_manager::scan_table(&self.catalog, table)
     }
 
-    fn scan_table_range(
+    pub(crate) fn scan_table_range(
         &self,
         table: &TableSchema,
         range: TableKeyRange,
     ) -> StorageResult<TableScan> {
-        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<IndexScan> {
-        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<Value>,
     ) -> StorageResult<OwnedTableRecord> {
-        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<Value>,
     ) -> StorageResult<OwnedTableRecord> {
-        self.records.update_table_row(table, record, values)
+        record_manager::update_table_row(&self.catalog, table, record, values)
     }
 }
 

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};