feat(storage): restart and finish part 1, 2

Signed-off-by: Alex Chi <[email protected]>

Author: skyzh

code/03-00/src/executor/create.rs

@@ -1,7 +1,6 @@
 use super::*;
 use crate::catalog::TableRefId;
 use crate::physical_planner::PhysicalCreateTable;
-use crate::storage::StorageRef;
 
 /// The executor of `CREATE TABLE` statement.
 pub struct CreateTableExecutor {

code/03-00/src/executor/insert.rs

@@ -3,7 +3,6 @@ use itertools::Itertools;
 use super::*;
 use crate::array::{ArrayBuilderImpl, DataChunk};
 use crate::catalog::{ColumnId, TableRefId};
-use crate::storage::StorageRef;
 use crate::types::{DataType, DataValue};
 
 /// The executor of `INSERT` statement.
@@ -42,7 +41,7 @@ impl InsertExecutor {
         for chunk in self.child {
             let chunk = transform_chunk(chunk?, &output_columns);
             count += chunk.cardinality();
-            table.append(chunk)?;
+            table.append(chunk).await?;
         }
         yield DataChunk::single(count as i32);
     }

code/03-00/src/executor/seq_scan.rs

@@ -13,7 +13,7 @@ impl SeqScanExecutor {
     #[try_stream(boxed, ok = DataChunk, error = ExecuteError)]
     pub async fn execute(self) {
         let table = self.storage.get_table(self.table_ref_id)?;
-        for chunk in table.all_chunks()? {
+        for chunk in table.all_chunks().await? {
             yield chunk;
         }
     }

code/03-00/src/storage/mod.rs

@@ -1,31 +1,29 @@
-//! In-memory storage.
-//!
-//! RisingLight's in-memory representation of data is very simple. Currently,
-//! it is simple a vector of `DataChunk`. Upon insertion, users' data are
-//! simply appended to the end of the vector.
+//! On-disk storage
 
-use std::collections::HashMap;
-use std::sync::{Arc, Mutex, RwLock};
+use std::sync::Arc;
 
 use crate::array::DataChunk;
 use crate::catalog::TableRefId;
 
 /// The error type of storage operations.
 #[derive(thiserror::Error, Debug)]
-pub enum StorageError {
-    #[error("table not found: {0:?}")]
-    NotFound(TableRefId),
-}
+#[error("{0:?}")]
+pub struct StorageError(#[from] anyhow::Error);
 
 /// A specialized `Result` type for storage operations.
 pub type StorageResult<T> = std::result::Result<T, StorageError>;
 
 pub type StorageRef = Arc<DiskStorage>;
-pub type DiskTableRef = Arc<DiskTable>;
+pub type StorageTableRef = Arc<DiskTable>;
+
+/// On-disk storage.
+#[derive(Clone)]
+pub struct DiskStorage;
 
-/// In-memory storage.
-pub struct DiskStorage {
-    tables: Mutex<HashMap<TableRefId, DiskTableRef>>,
+/// An on-disk table.
+pub struct DiskTable {
+    #[allow(dead_code)]
+    id: TableRefId,
 }
 
 impl Default for DiskStorage {
@@ -37,59 +35,28 @@ impl Default for DiskStorage {
 impl DiskStorage {
     /// Create a new in-memory storage.
     pub fn new() -> Self {
-        DiskStorage {
-            tables: Mutex::new(HashMap::new()),
-        }
+        DiskStorage
     }
 
     /// Add a table.
-    pub fn add_table(&self, id: TableRefId) -> StorageResult<()> {
-        let table = Arc::new(DiskTable::new(id));
-        self.tables.lock().unwrap().insert(id, table);
-        Ok(())
+    pub fn add_table(&self, _id: TableRefId) -> StorageResult<()> {
+        todo!()
     }
 
     /// Get a table.
-    pub fn get_table(&self, id: TableRefId) -> StorageResult<DiskTableRef> {
-        self.tables
-            .lock()
-            .unwrap()
-            .get(&id)
-            .cloned()
-            .ok_or(StorageError::NotFound(id))
+    pub fn get_table(&self, _id: TableRefId) -> StorageResult<StorageTableRef> {
+        todo!()
     }
 }
 
-/// A table in in-memory engine.
-pub struct DiskTable {
-    #[allow(dead_code)]
-    id: TableRefId,
-    inner: RwLock<DiskTableInner>,
-}
-
-#[derive(Default)]
-struct DiskTableInner {
-    chunks: Vec<DataChunk>,
-}
-
 impl DiskTable {
-    fn new(id: TableRefId) -> Self {
-        Self {
-            id,
-            inner: RwLock::new(DiskTableInner::default()),
-        }
-    }
-
     /// Append a chunk to the table.
-    pub fn append(&self, chunk: DataChunk) -> StorageResult<()> {
-        let mut inner = self.inner.write().unwrap();
-        inner.chunks.push(chunk);
-        Ok(())
+    pub async fn append(&self, _chunk: DataChunk) -> StorageResult<()> {
+        todo!()
     }
 
     /// Get all chunks of the table.
-    pub fn all_chunks(&self) -> StorageResult<Vec<DataChunk>> {
-        let inner = self.inner.read().unwrap();
-        Ok(inner.chunks.clone())
+    pub async fn all_chunks(&self) -> StorageResult<Vec<DataChunk>> {
+        todo!()
     }
 }

code/03-00/src/test.rs

@@ -1,3 +1,6 @@
+#![allow(unused_imports)]
+#![allow(dead_code)]
+
 use std::path::Path;
 
 use test_case::test_case;
@@ -6,16 +9,13 @@ use crate::array::DataChunk;
 use crate::types::DataValue;
 use crate::{Database, Error};
 
-#[test_case("01-01.slt")]
-#[test_case("01-03.slt")]
-#[test_case("01-05.slt")]
-#[test_case("01-06.slt")]
-#[test_case("01-07.slt")]
 fn test(name: &str) {
     init_logger();
     let script = std::fs::read_to_string(Path::new("../sql").join(name)).unwrap();
     let mut tester = sqllogictest::Runner::new(Database::new());
-    tester.run_script(&script).unwrap();
+    if let Err(err) = tester.run_script(&script) {
+        panic!("{}", err);
+    }
 }
 
 impl sqllogictest::DB for Database {

code/03-01/Cargo.toml

@@ -7,6 +7,7 @@ edition = "2021"
 [dependencies]
 anyhow = "1"
 bitvec = "1.0"
+bytes = "1"
 enum_dispatch = "0.3"
 env_logger = "0.9"
 futures = { version = "0.3", default-features = false, features = ["alloc"] }
@@ -17,7 +18,7 @@ prettytable-rs = { version = "0.8", default-features = false }
 rustyline = "9"
 sqlparser = "0.13"
 thiserror = "1"
-tokio = { version = "1", features = ["rt", "rt-multi-thread", "sync", "macros"] }
+tokio = { version = "1", features = ["rt", "rt-multi-thread", "sync", "macros", "fs"] }
 tokio-stream = "0.1"
 
 [dev-dependencies]