Test: add many e2e test (#88)

* test: add many e2e test

- insert.slt
- join_left_inner.slt
- limit.slt
- where.slt

tips: fix `limit` operator

* upgrade rust-toolchain version

* fix: `insert.slt` create table use multi types

* config: ci e2e test

* config: fix ci e2e test

Author: KKould

.github/workflows/ci.yml

@@ -61,4 +61,25 @@ jobs:
         uses: actions-rs/cargo@v1
         with:
           command: fmt
-          args: -- --check
+          args: -- --check
+  # 3
+  e2e:
+    name: Rust e2e sqllogictest
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - name: Install latest nightly
+        uses: actions-rs/toolchain@v1
+        with:
+          toolchain: nightly
+          override: true
+          components: rustfmt, clippy
+
+      # `cargo check` command here will use installed `nightly`
+      # as it is set as an "override" for current directory
+
+      - name: Run cargo run sqllogictest-test
+        uses: actions-rs/cargo@v1
+        with:
+          command: run
+          args: --bin sqllogictest-test --manifest-path ./tests/sqllogictest/Cargo.toml

Cargo.toml

@@ -26,7 +26,6 @@ tokio = { version = "1.28.2", features = ["full"] }
 serde = { version = "1", features = ["derive", "rc"] }
 serde_json = "1"
 bincode = "1.3.3"
-async-trait = "0.1.68"
 integer-encoding = "3.0.4"
 strum_macros = "0.24"
 ordered-float = "3.0"

rust-toolchain

@@ -1 +1 @@
-nightly-2023-09-29
+nightly-2023-10-13

src/binder/select.rs

@@ -26,6 +26,7 @@ use crate::planner::operator::join::JoinCondition;
 use crate::planner::operator::sort::{SortField, SortOperator};
 use crate::planner::LogicalPlan;
 use crate::storage::Transaction;
+use crate::types::errors::TypeError;
 use crate::types::LogicalType;
 use itertools::Itertools;
 use sqlparser::ast;
@@ -178,7 +179,6 @@ impl<'a, T: Transaction> Binder<'a, T> {
                     if tables.len() > 1 {
                         todo!("Implement virtual tables for multiple table aliases");
                     }
-                    // FIXME
                     self.context
                         .add_table_alias(alias.to_string(), tables.remove(0))?;
 
@@ -366,19 +366,15 @@ impl<'a, T: Transaction> Binder<'a, T> {
         limit_expr: &Option<Expr>,
         offset_expr: &Option<Offset>,
     ) -> Result<LogicalPlan, BindError> {
-        let mut limit = 0;
-        let mut offset = 0;
+        let mut limit = None;
+        let mut offset = None;
         if let Some(expr) = limit_expr {
             let expr = self.bind_expr(expr)?;
             match expr {
                 ScalarExpression::Constant(dv) => match dv.as_ref() {
-                    DataValue::Int32(Some(v)) if *v > 0 => limit = *v as usize,
-                    DataValue::Int64(Some(v)) if *v > 0 => limit = *v as usize,
-                    _ => {
-                        return Err(BindError::InvalidColumn(
-                            "invalid limit expression.".to_owned(),
-                        ))
-                    }
+                    DataValue::Int32(Some(v)) if *v >= 0 => limit = Some(*v as usize),
+                    DataValue::Int64(Some(v)) if *v >= 0 => limit = Some(*v as usize),
+                    _ => return Err(BindError::from(TypeError::InvalidType)),
                 },
                 _ => {
                     return Err(BindError::InvalidColumn(
@@ -392,13 +388,9 @@ impl<'a, T: Transaction> Binder<'a, T> {
             let expr = self.bind_expr(&expr.value)?;
             match expr {
                 ScalarExpression::Constant(dv) => match dv.as_ref() {
-                    DataValue::Int32(Some(v)) if *v > 0 => offset = *v as usize,
-                    DataValue::Int64(Some(v)) if *v > 0 => offset = *v as usize,
-                    _ => {
-                        return Err(BindError::InvalidColumn(
-                            "invalid limit expression.".to_owned(),
-                        ))
-                    }
+                    DataValue::Int32(Some(v)) if *v > 0 => offset = Some(*v as usize),
+                    DataValue::Int64(Some(v)) if *v > 0 => offset = Some(*v as usize),
+                    _ => return Err(BindError::from(TypeError::InvalidType)),
                 },
                 _ => {
                     return Err(BindError::InvalidColumn(

src/execution/executor/dql/limit.rs

@@ -16,8 +16,8 @@ pub struct Limit {
 impl From<(LimitOperator, BoxedExecutor)> for Limit {
     fn from((LimitOperator { offset, limit }, input): (LimitOperator, BoxedExecutor)) -> Self {
         Limit {
-            offset: Some(offset),
-            limit: Some(limit),
+            offset,
+            limit,
             input,
         }
     }

src/lib.rs

@@ -5,6 +5,7 @@
 #![feature(iterator_try_collect)]
 #![feature(slice_pattern)]
 #![feature(bound_map)]
+#![feature(async_fn_in_trait)]
 extern crate core;
 pub mod binder;
 pub mod catalog;

src/optimizer/rule/pushdown_limit.rs

@@ -75,10 +75,10 @@ impl Rule for EliminateLimits {
         if let Operator::Limit(op) = graph.operator(node_id) {
             let child_id = graph.children_at(node_id)[0];
             if let Operator::Limit(child_op) = graph.operator(child_id) {
-                let new_limit_op = LimitOperator {
-                    offset: op.offset + child_op.offset,
-                    limit: cmp::min(op.limit, child_op.limit),
-                };
+                let offset = Self::binary_options(op.offset, child_op.offset, |a, b| a + b);
+                let limit = Self::binary_options(op.limit, child_op.limit, |a, b| cmp::min(a, b));
+
+                let new_limit_op = LimitOperator { offset, limit };
 
                 graph.remove_node(child_id, false);
                 graph.replace_node(
@@ -92,6 +92,21 @@ impl Rule for EliminateLimits {
     }
 }
 
+impl EliminateLimits {
+    fn binary_options<F: Fn(usize, usize) -> usize>(
+        a: Option<usize>,
+        b: Option<usize>,
+        _fn: F,
+    ) -> Option<usize> {
+        match (a, b) {
+            (Some(a), Some(b)) => Some(_fn(a, b)),
+            (Some(a), None) => Some(a),
+            (None, Some(b)) => Some(b),
+            (None, None) => None,
+        }
+    }
+}
+
 /// Add extra limits below JOIN:
 /// 1. For LEFT OUTER and RIGHT OUTER JOIN, we push limits to the left and right sides,
 /// respectively.
@@ -147,7 +162,7 @@ impl Rule for PushLimitIntoScan {
             if let Operator::Scan(scan_op) = graph.operator(child_index) {
                 let mut new_scan_op = scan_op.clone();
 
-                new_scan_op.limit = (Some(limit_op.offset), Some(limit_op.limit));
+                new_scan_op.limit = (limit_op.offset, limit_op.limit);
 
                 graph.remove_node(node_id, false);
                 graph.replace_node(
@@ -208,8 +223,8 @@ mod tests {
         );
 
         let new_limit_op = LimitOperator {
-            offset: 2,
-            limit: 1,
+            offset: Some(2),
+            limit: Some(1),
         };
 
         optimizer
@@ -219,8 +234,8 @@ mod tests {
         let best_plan = optimizer.find_best()?;
 
         if let Operator::Limit(op) = &best_plan.operator {
-            assert_eq!(op.limit, 1);
-            assert_eq!(op.offset, 3);
+            assert_eq!(op.limit, Some(1));
+            assert_eq!(op.offset, Some(3));
         } else {
             unreachable!("Should be a project operator")
         }
@@ -253,7 +268,7 @@ mod tests {
         }
 
         if let Operator::Limit(op) = &best_plan.childrens[0].childrens[0].childrens[0].operator {
-            assert_eq!(op.limit, 1);
+            assert_eq!(op.limit, Some(1));
         } else {
             unreachable!("Should be a limit operator")
         }

src/planner/operator/limit.rs

@@ -4,12 +4,12 @@ use super::Operator;
 
 #[derive(Debug, PartialEq, Clone)]
 pub struct LimitOperator {
-    pub offset: usize,
-    pub limit: usize,
+    pub offset: Option<usize>,
+    pub limit: Option<usize>,
 }
 
 impl LimitOperator {
-    pub fn new(offset: usize, limit: usize, children: LogicalPlan) -> LogicalPlan {
+    pub fn new(offset: Option<usize>, limit: Option<usize>, children: LogicalPlan) -> LogicalPlan {
         LogicalPlan {
             operator: Operator::Limit(LimitOperator { offset, limit }),
             childrens: vec![children],

src/storage/kip.rs

@@ -8,7 +8,6 @@ use crate::types::errors::TypeError;
 use crate::types::index::{Index, IndexMeta, IndexMetaRef};
 use crate::types::tuple::{Tuple, TupleId};
 use crate::types::value::ValueRef;
-use async_trait::async_trait;
 use kip_db::kernel::lsm::iterator::Iter as KipDBIter;
 use kip_db::kernel::lsm::mvcc::TransactionIter;
 use kip_db::kernel::lsm::storage::Config;
@@ -37,7 +36,6 @@ impl KipStorage {
     }
 }
 
-#[async_trait]
 impl Storage for KipStorage {
     type TransactionType = KipTransaction;
 
@@ -56,7 +54,6 @@ pub struct KipTransaction {
     cache: ShardingLruCache<String, TableCatalog>,
 }
 
-#[async_trait]
 impl Transaction for KipTransaction {
     type IterType<'a> = KipIter<'a>;
 

src/storage/mod.rs

@@ -8,25 +8,23 @@ use crate::storage::table_codec::TableCodec;
 use crate::types::errors::TypeError;
 use crate::types::index::{Index, IndexMetaRef};
 use crate::types::tuple::{Tuple, TupleId};
-use async_trait::async_trait;
 use kip_db::error::CacheError;
 use kip_db::kernel::lsm::mvcc;
 use kip_db::KernelError;
 use std::collections::VecDeque;
 use std::ops::SubAssign;
 
-#[async_trait]
 pub trait Storage: Sync + Send + Clone + 'static {
     type TransactionType: Transaction;
 
+    #[allow(async_fn_in_trait)]
     async fn transaction(&self) -> Result<Self::TransactionType, StorageError>;
 }
 
 /// Optional bounds of the reader, of the form (offset, limit).
 pub(crate) type Bounds = (Option<usize>, Option<usize>);
 type Projections = Vec<ScalarExpression>;
 
-#[async_trait]
 pub trait Transaction: Sync + Send + 'static {
     type IterType<'a>: Iter;
 
@@ -80,6 +78,7 @@ pub trait Transaction: Sync + Send + 'static {
 
     fn show_tables(&self) -> Result<Vec<String>, StorageError>;
 
+    #[allow(async_fn_in_trait)]
     async fn commit(self) -> Result<(), StorageError>;
 }