feat(type): use arrow type to support vectorization

.github/workflows/ci.yml

@@ -8,47 +8,40 @@ env:
   CARGO_TERM_COLOR: always
 
 jobs:
-  check:
-    runs-on: ubuntu-20.04
+  fmt:
+    runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v3
       - uses: actions-rs/toolchain@v1
         with:
           profile: minimal
-          toolchain: nightly-2023-04-07
           components: rustfmt, clippy
-      - name: Check code format
-        uses: actions-rs/cargo@v1
+      - uses: actions/cache@v3
         with:
-          command: fmt
-          args: --all -- --check
+          path: |
+            ~/.cargo/registry/index/
+            ~/.cargo/registry/cache/
+            ~/.cargo/git/db/
+          key: ${{ runner.os }}-cargo-${{ hashFiles('**/Cargo.lock') }}
+      - name: Check code format
+        run: cargo fmt --all -- --check
 
 
-  build:
-    runs-on: ubuntu-20.04
-    steps:
-      - uses: actions/checkout@v2
-      - uses: actions-rs/toolchain@v1
-        with:
-          profile: minimal
-          toolchain: nightly-2023-04-07
-      - uses: actions/checkout@v2
-      - name: Build
-        uses: actions-rs/cargo@v1
-        with:
-          command: build
 
   test:
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v3
       - uses: actions-rs/toolchain@v1
         with:
           profile: minimal
-          toolchain: nightly-2023-04-07
-      - uses: actions/checkout@v2
-      - name: Test
-        uses: actions-rs/cargo@v1
+      - uses: actions/cache@v3
         with:
-          command: test
-          args: --release --no-fail-fast
+          path: |
+            ~/.cargo/registry/index/
+            ~/.cargo/registry/cache/
+            ~/.cargo/git/db/
+          key: ${{ runner.os }}-cargo-${{ hashFiles('**/Cargo.lock') }}
+      - uses: taiki-e/install-action@nextest
+      - name: Test
+        run: cargo nextest run --no-fail-fast --all-features

Cargo.toml

@@ -31,13 +31,17 @@ serde = { version = "1", features = ["derive", "rc"] }
 serde_json = "1"
 async-trait = "0.1.68"
 integer-encoding = "3.0.4"
+arrow = { version = "28", features = ["prettyprint", "simd"] }
+strum_macros = "0.24"
+ordered-float = "3.0"
 petgraph = "0.6.3"
 futures-async-stream = "0.2.6"
 async-channel = "1.8.0"
 async-backtrace = "0.2.6"
 futures = "0.3.25"
 futures-lite = "1.12.0"
 
+
 [dev-dependencies]
 ctor = "0.2.0"
 env_logger = "0.10"

rust-toolchain

@@ -0,0 +1 @@
+nightly-2023-04-07

src/binder/create.rs

@@ -1,12 +1,12 @@
+use std::collections::HashSet;
+
+use anyhow::Result;
+use sqlparser::ast::{ColumnDef, ObjectName};
+
 use super::Binder;
 use crate::binder::{lower_case_name, split_name};
-use crate::catalog::{Column, ColumnDesc};
+use crate::catalog::ColumnCatalog;
 use crate::planner::logical_create_table_plan::LogicalCreateTablePlan;
-use crate::planner::LogicalPlan;
-use crate::types::{ColumnId, TableId};
-use anyhow::Result;
-use sqlparser::ast::{ColumnDef, ObjectName};
-use std::collections::HashSet;
 
 impl Binder {
     pub(crate) fn bind_create_table(
@@ -29,10 +29,10 @@ impl Binder {
             }
         }
 
-        let mut columns: Vec<Column> = columns
+        let columns: Vec<ColumnCatalog> = columns
             .iter()
             .enumerate()
-            .map(|(_, col)| Column::from(col))
+            .map(|(_, col)| ColumnCatalog::from(col.clone()))
             .collect();
 
         let plan = LogicalCreateTablePlan {
@@ -48,36 +48,31 @@ impl Binder {
 
 #[cfg(test)]
 mod tests {
+    use sqlparser::ast::CharacterLength;
+
     use super::*;
     use crate::binder::BinderContext;
-    use crate::catalog::Root;
-    use crate::types::{DataTypeExt, DataTypeKind};
-    use sqlparser::ast::CharacterLength;
-    use std::sync::Arc;
+    use crate::catalog::{ColumnDesc, RootCatalog};
+    use crate::planner::LogicalPlan;
+    use crate::types::LogicalType;
 
     #[test]
     fn test_create_bind() {
         let sql = "create table t1 (id int , name varchar(10))";
-        let mut binder = Binder::new(BinderContext::new(Root::new()));
+        let binder = Binder::new(BinderContext::new(RootCatalog::new()));
         let stmt = crate::parser::parse_sql(sql).unwrap();
         let plan1 = binder.bind(&stmt[0]).unwrap();
 
-        let character_length = CharacterLength {
-            length: 10,
-            unit: None,
-        };
         let plan2 = LogicalPlan::CreateTable(LogicalCreateTablePlan {
             table_name: "t1".to_string(),
             columns: vec![
                 (
                     "id".to_string(),
-                    DataTypeKind::Int(None).nullable().to_column(),
+                    ColumnDesc::new(LogicalType::Integer, false),
                 ),
                 (
                     "name".to_string(),
-                    DataTypeKind::Varchar(Option::from(character_length))
-                        .nullable()
-                        .to_column(),
+                    ColumnDesc::new(LogicalType::Varchar, false),
                 ),
             ],
         });

src/binder/expr.rs

@@ -1,8 +1,8 @@
+use anyhow::Result;
 use sqlparser::ast::Expr;
 
 use super::Binder;
 use crate::expression::ScalarExpression;
-use anyhow::Result;
 
 impl Binder {
     pub(crate) fn bind_expr(&mut self, expr: &Expr) -> Result<ScalarExpression> {

src/binder/mod.rs

@@ -5,15 +5,16 @@ mod select;
 
 use std::collections::HashMap;
 
-use crate::{catalog::CatalogRef, expression::ScalarExpression, planner::LogicalPlan};
-
-use crate::catalog::{Root, DEFAULT_SCHEMA_NAME};
-use crate::types::TableId;
 use anyhow::Result;
 use sqlparser::ast::{Ident, ObjectName, Statement};
+
+use crate::catalog::{RootCatalog, DEFAULT_SCHEMA_NAME};
+use crate::expression::ScalarExpression;
+use crate::planner::LogicalPlan;
+use crate::types::TableId;
 #[derive(Clone)]
 pub struct BinderContext {
-    catalog: Root,
+    catalog: RootCatalog,
     bind_table: HashMap<String, TableId>,
     aliases: HashMap<String, ScalarExpression>,
     group_by_exprs: Vec<ScalarExpression>,
@@ -22,7 +23,7 @@ pub struct BinderContext {
 }
 
 impl BinderContext {
-    pub fn new(catalog: Root) -> Self {
+    pub fn new(catalog: RootCatalog) -> Self {
         BinderContext {
             catalog,
             bind_table: Default::default(),
@@ -91,3 +92,21 @@ fn split_name(name: &ObjectName) -> Result<(&str, &str)> {
         _ => return Err(anyhow::anyhow!("Invalid table name: {:?}", name)),
     })
 }
+
+#[derive(thiserror::Error, Debug)]
+pub enum BindError {
+    #[error("unsupported statement {0}")]
+    UnsupportedStmt(String),
+    #[error("invalid table {0}")]
+    InvalidTable(String),
+    #[error("invalid table name: {0:?}")]
+    InvalidTableName(Vec<Ident>),
+    #[error("invalid column {0}")]
+    InvalidColumn(String),
+    #[error("ambiguous column {0}")]
+    AmbiguousColumn(String),
+    #[error("binary operator types mismatch: {0} != {1}")]
+    BinaryOpTypeMismatch(String, String),
+    #[error("subquery in FROM must have an alias")]
+    SubqueryMustHaveAlias,
+}

src/binder/select.rs

@@ -166,8 +166,7 @@ impl Binder {
     ///
     /// - Qualified name, e.g. `SELECT t.a FROM t`
     /// - Qualified name with wildcard, e.g. `SELECT t.* FROM t,t1`
-    /// - Scalar expression or aggregate expression, e.g. `SELECT COUNT(*) + 1
-    ///   AS count FROM t`
+    /// - Scalar expression or aggregate expression, e.g. `SELECT COUNT(*) + 1 AS count FROM t`
     ///  
     fn normalize_select_item(&mut self, items: &[SelectItem]) -> Result<Vec<ScalarExpression>> {
         let mut select_items = vec![];
@@ -293,8 +292,8 @@ impl Binder {
             let expr = self.bind_expr(expr)?;
             match expr {
                 ScalarExpression::Constant(dv) => match dv {
-                    DataValue::Int32(v) if v > 0 => limit = v as usize,
-                    DataValue::Int64(v) if v > 0 => limit = v as usize,
+                    DataValue::Int32(Some(v)) if v > 0 => limit = v as usize,
+                    DataValue::Int64(Some(v)) if v > 0 => limit = v as usize,
                     _ => return Err(anyhow::Error::msg("invalid limit expression.".to_owned())),
                 },
                 _ => return Err(anyhow::Error::msg("invalid limit expression.".to_owned())),
@@ -305,8 +304,8 @@ impl Binder {
             let expr = self.bind_expr(&expr.value)?;
             match expr {
                 ScalarExpression::Constant(dv) => match dv {
-                    DataValue::Int32(v) if v > 0 => offset = v as usize,
-                    DataValue::Int64(v) if v > 0 => offset = v as usize,
+                    DataValue::Int32(Some(v)) if v > 0 => offset = v as usize,
+                    DataValue::Int64(Some(v)) if v > 0 => offset = v as usize,
                     _ => return Err(anyhow::Error::msg("invalid limit expression.".to_owned())),
                 },
                 _ => return Err(anyhow::Error::msg("invalid offset expression.".to_owned())),