feat: support CreateIndex and IndexType::Normal & `IndexType::Com…

…posite`

Cargo.toml

@@ -2,7 +2,7 @@
 
 [package]
 name = "fnck_sql"
-version = "0.0.1-alpha.11.fix3"
+version = "0.0.1-alpha.12"
 edition = "2021"
 authors = ["Kould <[email protected]>", "Xwg <[email protected]>"]
 description = "Fast Insert OLTP SQL DBMS"

src/binder/alter_table.rs

@@ -19,73 +19,69 @@ impl<'a, T: Transaction> Binder<'a, T> {
         operation: &AlterTableOperation,
     ) -> Result<LogicalPlan, DatabaseError> {
         let table_name: Arc<String> = Arc::new(lower_case_name(name)?);
+        let table = self
+            .context
+            .table(table_name.clone())
+            .ok_or(DatabaseError::TableNotFound)?;
+        let plan = match operation {
+            AlterTableOperation::AddColumn {
+                column_keyword: _,
+                if_not_exists,
+                column_def,
+            } => {
+                let plan = ScanOperator::build(table_name.clone(), table);
+                let column = self.bind_column(column_def)?;
 
-        if let Some(table) = self.context.table(table_name.clone()) {
-            let plan = match operation {
-                AlterTableOperation::AddColumn {
-                    column_keyword: _,
-                    if_not_exists,
-                    column_def,
-                } => {
-                    let plan = ScanOperator::build(table_name.clone(), table);
-                    let column = self.bind_column(column_def)?;
-
-                    if !is_valid_identifier(column.name()) {
-                        return Err(DatabaseError::InvalidColumn(
-                            "illegal column naming".to_string(),
-                        ));
-                    }
-                    LogicalPlan::new(
-                        Operator::AddColumn(AddColumnOperator {
-                            table_name,
-                            if_not_exists: *if_not_exists,
-                            column,
-                        }),
-                        vec![plan],
-                    )
+                if !is_valid_identifier(column.name()) {
+                    return Err(DatabaseError::InvalidColumn(
+                        "illegal column naming".to_string(),
+                    ));
                 }
-                AlterTableOperation::DropColumn {
-                    column_name,
-                    if_exists,
-                    ..
-                } => {
-                    let plan = ScanOperator::build(table_name.clone(), table);
-                    let column_name = column_name.value.clone();
+                LogicalPlan::new(
+                    Operator::AddColumn(AddColumnOperator {
+                        table_name,
+                        if_not_exists: *if_not_exists,
+                        column,
+                    }),
+                    vec![plan],
+                )
+            }
+            AlterTableOperation::DropColumn {
+                column_name,
+                if_exists,
+                ..
+            } => {
+                let plan = ScanOperator::build(table_name.clone(), table);
+                let column_name = column_name.value.clone();
 
-                    LogicalPlan::new(
-                        Operator::DropColumn(DropColumnOperator {
-                            table_name,
-                            if_exists: *if_exists,
-                            column_name,
-                        }),
-                        vec![plan],
-                    )
-                }
-                AlterTableOperation::DropPrimaryKey => todo!(),
-                AlterTableOperation::RenameColumn {
-                    old_column_name: _,
-                    new_column_name: _,
-                } => todo!(),
-                AlterTableOperation::RenameTable { table_name: _ } => todo!(),
-                AlterTableOperation::ChangeColumn {
-                    old_name: _,
-                    new_name: _,
-                    data_type: _,
-                    options: _,
-                } => todo!(),
-                AlterTableOperation::AlterColumn {
-                    column_name: _,
-                    op: _,
-                } => todo!(),
-                _ => todo!(),
-            };
+                LogicalPlan::new(
+                    Operator::DropColumn(DropColumnOperator {
+                        table_name,
+                        if_exists: *if_exists,
+                        column_name,
+                    }),
+                    vec![plan],
+                )
+            }
+            AlterTableOperation::DropPrimaryKey => todo!(),
+            AlterTableOperation::RenameColumn {
+                old_column_name: _,
+                new_column_name: _,
+            } => todo!(),
+            AlterTableOperation::RenameTable { table_name: _ } => todo!(),
+            AlterTableOperation::ChangeColumn {
+                old_name: _,
+                new_name: _,
+                data_type: _,
+                options: _,
+            } => todo!(),
+            AlterTableOperation::AlterColumn {
+                column_name: _,
+                op: _,
+            } => todo!(),
+            _ => todo!(),
+        };
 
-            Ok(plan)
-        } else {
-            Err(DatabaseError::InvalidTable(format!(
-                "not found table {}",
-                table_name
-            )))
-        }
+        Ok(plan)
     }
 }

src/binder/analyze.rs

@@ -5,7 +5,6 @@ use crate::planner::operator::scan::ScanOperator;
 use crate::planner::operator::Operator;
 use crate::planner::LogicalPlan;
 use crate::storage::Transaction;
-use itertools::Itertools;
 use sqlparser::ast::ObjectName;
 use std::sync::Arc;
 
@@ -14,16 +13,13 @@ impl<'a, T: Transaction> Binder<'a, T> {
         let table_name = Arc::new(lower_case_name(name)?);
 
         let table_catalog = self.context.table_and_bind(table_name.clone(), None)?;
-        let columns = table_catalog
-            .columns()
-            .filter_map(|column| column.desc.is_index().then_some(column.clone()))
-            .collect_vec();
+        let index_metas = table_catalog.indexes.clone();
 
         let scan_op = ScanOperator::build(table_name.clone(), table_catalog);
         Ok(LogicalPlan::new(
             Operator::Analyze(AnalyzeOperator {
                 table_name,
-                columns,
+                index_metas,
             }),
             vec![scan_op],
         ))

src/binder/create_index.rs

@@ -0,0 +1,60 @@
+use crate::binder::{lower_case_name, Binder};
+use crate::errors::DatabaseError;
+use crate::expression::ScalarExpression;
+use crate::planner::operator::create_index::CreateIndexOperator;
+use crate::planner::operator::scan::ScanOperator;
+use crate::planner::operator::Operator;
+use crate::planner::LogicalPlan;
+use crate::storage::Transaction;
+use crate::types::index::IndexType;
+use sqlparser::ast::{ObjectName, OrderByExpr};
+use std::sync::Arc;
+
+impl<'a, T: Transaction> Binder<'a, T> {
+    pub(crate) fn bind_create_index(
+        &mut self,
+        table_name: &ObjectName,
+        name: &ObjectName,
+        exprs: &[OrderByExpr],
+        if_not_exists: bool,
+        is_unique: bool,
+    ) -> Result<LogicalPlan, DatabaseError> {
+        let table_name = Arc::new(lower_case_name(table_name)?);
+        let index_name = lower_case_name(name)?;
+        let ty = if is_unique {
+            IndexType::Unique
+        } else if exprs.len() == 1 {
+            IndexType::Normal
+        } else {
+            IndexType::Composite
+        };
+
+        let table = self.context.table_and_bind(table_name.clone(), None)?;
+        let plan = ScanOperator::build(table_name.clone(), table);
+        let mut columns = Vec::with_capacity(exprs.len());
+
+        for expr in exprs {
+            // TODO: Expression Index
+            match self.bind_expr(&expr.expr)? {
+                ScalarExpression::ColumnRef(column) => columns.push(column),
+                expr => {
+                    return Err(DatabaseError::UnsupportedStmt(format!(
+                        "create index on {}",
+                        expr
+                    )))
+                }
+            }
+        }
+
+        Ok(LogicalPlan::new(
+            Operator::CreateIndex(CreateIndexOperator {
+                table_name,
+                columns,
+                index_name,
+                if_not_exists,
+                ty,
+            }),
+            vec![plan],
+        ))
+    }
+}

src/binder/create_table.rs

@@ -80,15 +80,14 @@ impl<'a, T: Transaction> Binder<'a, T> {
             ));
         }
 
-        let plan = LogicalPlan::new(
+        Ok(LogicalPlan::new(
             Operator::CreateTable(CreateTableOperator {
                 table_name,
                 columns,
                 if_not_exists,
             }),
             vec![],
-        );
-        Ok(plan)
+        ))
     }
 
     pub fn bind_column(&mut self, column_def: &ColumnDef) -> Result<ColumnCatalog, DatabaseError> {

src/binder/mod.rs

@@ -2,6 +2,7 @@ pub mod aggregate;
 mod alter_table;
 mod analyze;
 pub mod copy;
+mod create_index;
 mod create_table;
 mod delete;
 mod describe;
@@ -248,6 +249,14 @@ impl<'a, T: Transaction> Binder<'a, T> {
                 describe_alias: true,
                 table_name,
             } => self.bind_describe(table_name)?,
+            Statement::CreateIndex {
+                table_name,
+                name,
+                columns,
+                if_not_exists,
+                unique,
+                ..
+            } => self.bind_create_index(table_name, name, columns, *if_not_exists, *unique)?,
             _ => return Err(DatabaseError::UnsupportedStmt(stmt.to_string())),
         };
         Ok(plan)

src/catalog/column.rs

@@ -115,8 +115,4 @@ impl ColumnDesc {
             default,
         }
     }
-
-    pub(crate) fn is_index(&self) -> bool {
-        self.is_unique || self.is_primary
-    }
 }

src/catalog/table.rs

@@ -6,7 +6,7 @@ use std::{slice, vec};
 
 use crate::catalog::{ColumnCatalog, ColumnRef};
 use crate::errors::DatabaseError;
-use crate::types::index::{IndexMeta, IndexMetaRef};
+use crate::types::index::{IndexMeta, IndexMetaRef, IndexType};
 use crate::types::tuple::SchemaRef;
 use crate::types::{ColumnId, LogicalType};
 
@@ -33,7 +33,7 @@ impl TableCatalog {
     pub(crate) fn get_unique_index(&self, col_id: &ColumnId) -> Option<&IndexMetaRef> {
         self.indexes
             .iter()
-            .find(|meta| meta.is_unique && &meta.column_ids[0] == col_id)
+            .find(|meta| matches!(meta.ty, IndexType::Unique) && &meta.column_ids[0] == col_id)
     }
 
     #[allow(dead_code)]
@@ -60,6 +60,10 @@ impl TableCatalog {
         self.schema_ref.iter()
     }
 
+    pub(crate) fn indexes(&self) -> slice::Iter<'_, IndexMetaRef> {
+        self.indexes.iter()
+    }
+
     pub(crate) fn schema_ref(&self) -> &SchemaRef {
         &self.schema_ref
     }
@@ -113,9 +117,14 @@ impl TableCatalog {
         &mut self,
         name: String,
         column_ids: Vec<ColumnId>,
-        is_unique: bool,
-        is_primary: bool,
+        ty: IndexType,
     ) -> Result<&IndexMeta, DatabaseError> {
+        for index in self.indexes.iter() {
+            if index.name == name {
+                return Err(DatabaseError::DuplicateIndex(name));
+            }
+        }
+
         let index_id = self.indexes.last().map(|index| index.id + 1).unwrap_or(0);
         let pk_ty = *self.primary_key()?.1.datatype();
         let index = IndexMeta {
@@ -124,8 +133,7 @@ impl TableCatalog {
             table_name: self.name.clone(),
             pk_ty,
             name,
-            is_unique,
-            is_primary,
+            ty,
         };
         self.indexes.push(Arc::new(index));
         Ok(self.indexes.last().unwrap())

src/errors.rs

@@ -101,6 +101,8 @@ pub enum DatabaseError {
     TableNotFound,
     #[error("the some column: {0} already exists")]
     DuplicateColumn(String),
+    #[error("the some index: {0} already exists")]
+    DuplicateIndex(String),
     #[error("add column must be nullable or specify a default value")]
     NeedNullAbleOrDefault,
     #[error("the table already exists")]

src/execution/volcano/ddl/add_column.rs

@@ -4,10 +4,11 @@ use crate::types::tuple::Tuple;
 use crate::types::tuple_builder::TupleBuilder;
 use crate::types::value::DataValue;
 use futures_async_stream::try_stream;
+use std::slice;
 use std::sync::Arc;
 
 use crate::planner::LogicalPlan;
-use crate::types::index::Index;
+use crate::types::index::{Index, IndexType};
 use crate::{planner::operator::alter_table::add_column::AddColumnOperator, storage::Transaction};
 
 pub struct AddColumn {
@@ -66,11 +67,8 @@ impl AddColumn {
                 .cloned(),
         ) {
             for (tuple_id, value) in unique_values {
-                let index = Index {
-                    id: unique_meta.id,
-                    column_values: vec![value],
-                };
-                transaction.add_index(table_name, index, &tuple_id, true)?;
+                let index = Index::new(unique_meta.id, slice::from_ref(&value), IndexType::Unique);
+                transaction.add_index(table_name, index, &tuple_id)?;
             }
         }