feat: support drop table if exists (#108)

src/binder/drop_table.rs

@@ -7,13 +7,20 @@ use sqlparser::ast::ObjectName;
 use std::sync::Arc;
 
 impl<'a, T: Transaction> Binder<'a, T> {
-    pub(crate) fn bind_drop_table(&mut self, name: &ObjectName) -> Result<LogicalPlan, BindError> {
+    pub(crate) fn bind_drop_table(
+        &mut self,
+        name: &ObjectName,
+        if_exists: &bool,
+    ) -> Result<LogicalPlan, BindError> {
         let name = lower_case_name(name);
         let (_, name) = split_name(&name)?;
         let table_name = Arc::new(name.to_string());
 
         let plan = LogicalPlan {
-            operator: Operator::DropTable(DropTableOperator { table_name }),
+            operator: Operator::DropTable(DropTableOperator {
+                table_name,
+                if_exists: *if_exists,
+            }),
             childrens: vec![],
         };
         Ok(plan)

src/binder/mod.rs

@@ -129,9 +129,12 @@ impl<'a, T: Transaction> Binder<'a, T> {
                 ..
             } => self.bind_create_table(name, columns, constraints, *if_not_exists)?,
             Statement::Drop {
-                object_type, names, ..
+                object_type,
+                names,
+                if_exists,
+                ..
             } => match object_type {
-                ObjectType::Table => self.bind_drop_table(&names[0])?,
+                ObjectType::Table => self.bind_drop_table(&names[0], if_exists)?,
                 _ => todo!(),
             },
             Statement::Insert {

src/execution/executor/ddl/drop_table.rs

@@ -25,8 +25,11 @@ impl<T: Transaction> Executor<T> for DropTable {
 impl DropTable {
     #[try_stream(boxed, ok = Tuple, error = ExecutorError)]
     pub async fn _execute<T: Transaction>(self, transaction: &mut T) {
-        let DropTableOperator { table_name } = self.op;
+        let DropTableOperator {
+            table_name,
+            if_exists,
+        } = self.op;
 
-        transaction.drop_table(&table_name)?;
+        transaction.drop_table(&table_name, if_exists)?;
     }
 }

src/planner/operator/drop_table.rs

@@ -4,4 +4,5 @@ use crate::catalog::TableName;
 pub struct DropTableOperator {
     /// Table name to insert to
     pub table_name: TableName,
+    pub if_exists: bool,
 }

src/storage/kip.rs

@@ -237,7 +237,15 @@ impl Transaction for KipTransaction {
         Ok(table_name)
     }
 
-    fn drop_table(&mut self, table_name: &str) -> Result<(), StorageError> {
+    fn drop_table(&mut self, table_name: &str, if_exists: bool) -> Result<(), StorageError> {
+        if self.table(Arc::new(table_name.to_string())).is_none() {
+            if if_exists {
+                return Ok(());
+            } else {
+                return Err(StorageError::TableNotFound);
+            }
+        }
+
         self.drop_data(table_name)?;
 
         let (min, max) = TableCodec::columns_bound(table_name);

src/storage/mod.rs

@@ -76,7 +76,7 @@ pub trait Transaction: Sync + Send + 'static {
         if_not_exists: bool,
     ) -> Result<TableName, StorageError>;
 
-    fn drop_table(&mut self, table_name: &str) -> Result<(), StorageError>;
+    fn drop_table(&mut self, table_name: &str, if_exists: bool) -> Result<(), StorageError>;
     fn drop_data(&mut self, table_name: &str) -> Result<(), StorageError>;
     fn table(&self, table_name: TableName) -> Option<&TableCatalog>;