feat(storage): add world 03-02 part 1 (#8)

skyzh · web-flow · commit 4655941ab9e5 · 2022-01-29T00:01:56.000+08:00
Signed-off-by: Alex Chi &lt;iskyzh@gmail.com&gt;
diff --git a/code/03-00/Cargo.toml b/code/03-00/Cargo.toml
@@ -5,6 +5,7 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+anyhow = "1"
 bitvec = "1.0"
 enum_dispatch = "0.3"
 env_logger = "0.9"
diff --git a/code/03-01/Cargo.toml b/code/03-01/Cargo.toml
@@ -5,6 +5,7 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+anyhow = "1"
 bitvec = "1.0"
 enum_dispatch = "0.3"
 env_logger = "0.9"
diff --git a/code/03-02/Cargo.toml b/code/03-02/Cargo.toml
@@ -0,0 +1,26 @@
+[package]
+name = "risinglight-03-02"
+version = "0.1.0"
+edition = "2021"
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[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"] }
+futures-async-stream = { git = "https://github.com/taiki-e/futures-async-stream", rev = "944f407" }
+itertools = "0.10"
+log = "0.4"
+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-stream = "0.1"
+
+[dev-dependencies]
+sqllogictest = "0.2"
+test-case = "1.2"
diff --git a/code/03-02/src/array b/code/03-02/src/array
@@ -0,0 +1 @@
+../../03-00/src/array
diff --git a/code/03-02/src/binder b/code/03-02/src/binder
@@ -0,0 +1 @@
+../../03-00/src/binder/
diff --git a/code/03-02/src/catalog b/code/03-02/src/catalog
@@ -0,0 +1 @@
+../../03-00/src/catalog/
diff --git a/code/03-02/src/db.rs b/code/03-02/src/db.rs
@@ -0,0 +1 @@
+../../03-00/src/db.rs
diff --git a/code/03-02/src/executor b/code/03-02/src/executor
@@ -0,0 +1 @@
+../../03-00/src/executor/
diff --git a/code/03-02/src/lib.rs b/code/03-02/src/lib.rs
@@ -0,0 +1 @@
+../../03-00/src/lib.rs
diff --git a/code/03-02/src/logical_planner b/code/03-02/src/logical_planner
@@ -0,0 +1 @@
+../../03-00/src/logical_planner
diff --git a/code/03-02/src/main.rs b/code/03-02/src/main.rs
@@ -0,0 +1,40 @@
+//! A simple interactive shell of the database.
+
+use risinglight_03_02::Database;
+use rustyline::error::ReadlineError;
+use rustyline::Editor;
+
+fn main() {
+    env_logger::init();
+
+    let db = Database::new();
+
+    let mut rl = Editor::<()>::new();
+    loop {
+        match rl.readline("> ") {
+            Ok(line) => {
+                rl.add_history_entry(line.as_str());
+                let ret = db.run(&line);
+                match ret {
+                    Ok(chunks) => {
+                        for chunk in chunks {
+                            println!("{}", chunk);
+                        }
+                    }
+                    Err(err) => println!("{}", err),
+                }
+            }
+            Err(ReadlineError::Interrupted) => {
+                println!("Interrupted");
+            }
+            Err(ReadlineError::Eof) => {
+                println!("Exited");
+                break;
+            }
+            Err(err) => {
+                println!("Error: {:?}", err);
+                break;
+            }
+        }
+    }
+}
diff --git a/code/03-02/src/parser.rs b/code/03-02/src/parser.rs
@@ -0,0 +1 @@
+../../03-00/src/parser.rs
diff --git a/code/03-02/src/physical_planner b/code/03-02/src/physical_planner
@@ -0,0 +1 @@
+../../03-00/src/physical_planner
diff --git a/code/03-02/src/storage/block/mod.rs b/code/03-02/src/storage/block/mod.rs
@@ -0,0 +1,85 @@
+// Copyright 2022 RisingLight Project Authors. Licensed under Apache-2.0.
+
+//! Secondary's Block builders and iterators
+//!
+//! [`Block`] is the minimum managing unit in the storage engine.
+
+mod primitive_block_builder;
+mod primitive_block_iterator;
+use anyhow::{anyhow, Context, Result};
+use bytes::{Buf, BufMut, Bytes};
+pub use primitive_block_builder::*;
+pub use primitive_block_iterator::*;
+
+use crate::array::Array;
+use crate::storage::proto::*;
+
+/// A block is simply a [`Bytes`] array.
+pub type Block = Bytes;
+
+/// Builds a block. All builders should implement the trait, while
+/// ensuring that the format follows the block encoding scheme.
+///
+/// In RisingLight, the block encoding scheme is as follows:
+///
+/// ```plain
+/// | block_type | cksum_type | cksum  |    data     |
+/// |    4B      |     4B     |   8B   |  variable   |
+/// ```
+pub trait BlockBuilder<A: Array> {
+    /// Append one data into the block.
+    fn append(&mut self, item: Option<&A::Item>);
+
+    /// Get estimated size of block. Will be useful on runlength or compression encoding.
+    fn estimated_size(&self) -> usize;
+
+    /// Check if we should finish the current block. If there is no item in the current
+    /// builder, this function must return `true`.
+    fn should_finish(&self, next_item: &Option<&A::Item>) -> bool;
+
+    /// Finish a block and return encoded data.
+    fn finish(self) -> Vec<u8>;
+}
+
+/// An iterator on a block. This iterator requires the block being pre-loaded in memory.
+pub trait BlockIterator<A: Array> {
+    /// Get a batch from the block. A `0` return value means that this batch contains no
+    /// element. Some iterators might support exact size output. By using `expected_size`,
+    /// developers can get an array of NO MORE THAN the `expected_size`.
+    fn next_batch(&mut self, expected_size: Option<usize>, builder: &mut A::Builder) -> usize;
+
+    /// Skip `cnt` items.
+    fn skip(&mut self, cnt: usize);
+
+    /// Number of items remaining in this block
+    fn remaining_items(&self) -> usize;
+}
+
+#[derive(Debug, Clone)]
+pub struct BlockHeader {
+    pub block_type: BlockType,
+    pub checksum_type: ChecksumType,
+    pub checksum: u64,
+}
+
+pub const BLOCK_HEADER_SIZE: usize = 4 + 4 + 8;
+
+impl BlockHeader {
+    pub fn encode(&self, buf: &mut impl BufMut) {
+        buf.put_i32(self.block_type.into());
+        buf.put_i32(self.checksum_type.into());
+        buf.put_u64(self.checksum);
+    }
+
+    pub fn decode(&mut self, buf: &mut impl Buf) -> Result<()> {
+        if buf.remaining() < 4 + 4 + 8 {
+            return Err(anyhow!("expected 16 bytes"));
+        }
+        self.block_type =
+            BlockType::from_i32(buf.get_i32()).context("expected valid checksum type")?;
+        self.checksum_type =
+            ChecksumType::from_i32(buf.get_i32()).context("expected valid checksum type")?;
+        self.checksum = buf.get_u64();
+        Ok(())
+    }
+}
diff --git a/code/03-02/src/storage/block/primitive_block_builder.rs b/code/03-02/src/storage/block/primitive_block_builder.rs
@@ -0,0 +1,44 @@
+// Copyright 2022 RisingLight Project Authors. Licensed under Apache-2.0.
+
+use std::marker::PhantomData;
+
+use super::super::encode::PrimitiveFixedWidthEncode;
+use super::BlockBuilder;
+
+/// Encodes fixed-width data into a block. The layout is simply an array of
+/// little endian fixed-width data.
+pub struct PlainPrimitiveBlockBuilder<T: PrimitiveFixedWidthEncode> {
+    data: Vec<u8>,
+    target_size: usize,
+    _phantom: PhantomData<T>,
+}
+
+impl<T: PrimitiveFixedWidthEncode> PlainPrimitiveBlockBuilder<T> {
+    pub fn new(target_size: usize) -> Self {
+        let data = Vec::with_capacity(target_size);
+        Self {
+            data,
+            target_size,
+            _phantom: PhantomData,
+        }
+    }
+}
+
+impl<T: PrimitiveFixedWidthEncode> BlockBuilder<T::ArrayType> for PlainPrimitiveBlockBuilder<T> {
+    fn append(&mut self, item: Option<&T>) {
+        item.expect("nullable item found in non-nullable block builder")
+            .encode(&mut self.data);
+    }
+
+    fn estimated_size(&self) -> usize {
+        self.data.len()
+    }
+
+    fn should_finish(&self, _next_item: &Option<&T>) -> bool {
+        !self.data.is_empty() && self.estimated_size() + T::WIDTH > self.target_size
+    }
+
+    fn finish(self) -> Vec<u8> {
+        self.data
+    }
+}
diff --git a/code/03-02/src/storage/block/primitive_block_iterator.rs b/code/03-02/src/storage/block/primitive_block_iterator.rs
@@ -0,0 +1,76 @@
+// Copyright 2022 RisingLight Project Authors. Licensed under Apache-2.0.
+
+use std::marker::PhantomData;
+
+use super::super::PrimitiveFixedWidthEncode;
+use super::{Block, BlockIterator};
+use crate::array::{Array, ArrayBuilder};
+
+/// Scans one or several arrays from the block content.
+pub struct PlainPrimitiveBlockIterator<T: PrimitiveFixedWidthEncode> {
+    /// Block content
+    block: Block,
+
+    /// Total count of elements in block
+    row_count: usize,
+
+    /// Indicates the beginning row of the next batch
+    next_row: usize,
+
+    _phantom: PhantomData<T>,
+}
+
+impl<T: PrimitiveFixedWidthEncode> PlainPrimitiveBlockIterator<T> {
+    pub fn new(block: Block, row_count: usize) -> Self {
+        Self {
+            block,
+            row_count,
+            next_row: 0,
+            _phantom: PhantomData,
+        }
+    }
+}
+
+impl<T: PrimitiveFixedWidthEncode> BlockIterator<T::ArrayType> for PlainPrimitiveBlockIterator<T> {
+    fn next_batch(
+        &mut self,
+        expected_size: Option<usize>,
+        builder: &mut <T::ArrayType as Array>::Builder,
+    ) -> usize {
+        if self.next_row >= self.row_count {
+            return 0;
+        }
+
+        // TODO(chi): error handling on corrupted block
+
+        let mut cnt = 0;
+        let mut buffer = &self.block[self.next_row * T::WIDTH..];
+
+        loop {
+            if let Some(expected_size) = expected_size {
+                assert!(expected_size > 0);
+                if cnt >= expected_size {
+                    break;
+                }
+            }
+
+            if self.next_row >= self.row_count {
+                break;
+            }
+
+            builder.push(Some(&T::decode(&mut buffer)));
+            cnt += 1;
+            self.next_row += 1;
+        }
+
+        cnt
+    }
+
+    fn skip(&mut self, cnt: usize) {
+        self.next_row += cnt;
+    }
+
+    fn remaining_items(&self) -> usize {
+        self.row_count - self.next_row
+    }
+}
diff --git a/code/03-02/src/storage/encode.rs b/code/03-02/src/storage/encode.rs
@@ -0,0 +1,64 @@
+// Copyright 2022 RisingLight Project Authors. Licensed under Apache-2.0.
+
+use bytes::{Buf, BufMut};
+
+use crate::array::{Array, BoolArray, F64Array, I32Array};
+
+/// Encode a primitive value into fixed-width buffer
+pub trait PrimitiveFixedWidthEncode: Copy + Clone + 'static + Send + Sync {
+    /// Width of each element
+    const WIDTH: usize;
+    const DEAFULT_VALUE: &'static Self;
+
+    type ArrayType: Array<Item = Self>;
+
+    /// Encode current primitive data to the end of an `Vec<u8>`.
+    fn encode(&self, buffer: &mut impl BufMut);
+
+    /// Decode a data from a bytes array.
+    fn decode(buffer: &mut impl Buf) -> Self;
+}
+
+impl PrimitiveFixedWidthEncode for bool {
+    const WIDTH: usize = std::mem::size_of::<u8>();
+    const DEAFULT_VALUE: &'static bool = &false;
+    type ArrayType = BoolArray;
+
+    fn encode(&self, buffer: &mut impl BufMut) {
+        buffer.put_u8(*self as u8)
+    }
+
+    fn decode(buffer: &mut impl Buf) -> Self {
+        buffer.get_u8() != 0
+    }
+}
+
+impl PrimitiveFixedWidthEncode for i32 {
+    const WIDTH: usize = std::mem::size_of::<i32>();
+    const DEAFULT_VALUE: &'static i32 = &0;
+
+    type ArrayType = I32Array;
+
+    fn encode(&self, buffer: &mut impl BufMut) {
+        buffer.put_i32_le(*self);
+    }
+
+    fn decode(buffer: &mut impl Buf) -> Self {
+        buffer.get_i32_le()
+    }
+}
+
+impl PrimitiveFixedWidthEncode for f64 {
+    const WIDTH: usize = std::mem::size_of::<f64>();
+    const DEAFULT_VALUE: &'static f64 = &0.0;
+
+    type ArrayType = F64Array;
+
+    fn encode(&self, buffer: &mut impl BufMut) {
+        buffer.put_f64_le(*self);
+    }
+
+    fn decode(buffer: &mut impl Buf) -> Self {
+        buffer.get_f64_le()
+    }
+}
diff --git a/code/03-02/src/storage/mod.rs b/code/03-02/src/storage/mod.rs
diff --git a/code/03-02/src/storage/proto.rs b/code/03-02/src/storage/proto.rs
diff --git a/code/03-02/src/storage/rowset/mem_rowset.rs b/code/03-02/src/storage/rowset/mem_rowset.rs
diff --git a/code/03-02/src/storage/rowset/mod.rs b/code/03-02/src/storage/rowset/mod.rs
diff --git a/code/03-02/src/storage/table_transaction.rs b/code/03-02/src/storage/table_transaction.rs
diff --git a/code/03-02/src/test.rs b/code/03-02/src/test.rs
diff --git a/code/03-02/src/types.rs b/code/03-02/src/types.rs
diff --git a/code/Cargo.toml b/code/Cargo.toml
