feat: implement drop multiple tables (#4085)

* feat: implement drop multiple tables

* fix: pass fmt and clippy checks

* add: drop multiple sqlness test

* update: accept review suggestions

* update: accept reviem suggestion

Co-authored-by: Weny Xu <wenymedia@gmail.com>

* fix: pass clippy check

---------

Co-authored-by: Weny Xu <wenymedia@gmail.com>

src/sql/src/parsers/drop_parser.rs

@@ -68,22 +68,29 @@ impl<'a> ParserContext<'a> {
         let _ = self.parser.next_token();
 
         let if_exists = self.parser.parse_keywords(&[Keyword::IF, Keyword::EXISTS]);
-        let raw_table_ident =
-            self.parse_object_name()
-                .with_context(|_| error::UnexpectedSnafu {
-                    sql: self.sql,
-                    expected: "a table name",
-                    actual: self.peek_token_as_string(),
-                })?;
-        let table_ident = Self::canonicalize_object_name(raw_table_ident);
-        ensure!(
-            !table_ident.0.is_empty(),
-            InvalidTableNameSnafu {
-                name: table_ident.to_string()
+        let mut table_names = Vec::with_capacity(1);
+        loop {
+            let raw_table_ident =
+                self.parse_object_name()
+                    .with_context(|_| error::UnexpectedSnafu {
+                        sql: self.sql,
+                        expected: "a table name",
+                        actual: self.peek_token_as_string(),
+                    })?;
+            let table_ident = Self::canonicalize_object_name(raw_table_ident);
+            ensure!(
+                !table_ident.0.is_empty(),
+                InvalidTableNameSnafu {
+                    name: table_ident.to_string()
+                }
+            );
+            table_names.push(table_ident);
+            if !self.parser.consume_token(&Token::Comma) {
+                break;
             }
-        );
+        }
 
-        Ok(Statement::DropTable(DropTable::new(table_ident, if_exists)))
+        Ok(Statement::DropTable(DropTable::new(table_names, if_exists)))
     }
 
     fn parse_drop_database(&mut self) -> Result<Statement> {
@@ -122,7 +129,10 @@ mod tests {
         let mut stmts = result.unwrap();
         assert_eq!(
             stmts.pop().unwrap(),
-            Statement::DropTable(DropTable::new(ObjectName(vec![Ident::new("foo")]), false))
+            Statement::DropTable(DropTable::new(
+                vec![ObjectName(vec![Ident::new("foo")])],
+                false
+            ))
         );
 
         let sql = "DROP TABLE IF EXISTS foo";
@@ -131,7 +141,10 @@ mod tests {
         let mut stmts = result.unwrap();
         assert_eq!(
             stmts.pop().unwrap(),
-            Statement::DropTable(DropTable::new(ObjectName(vec![Ident::new("foo")]), true))
+            Statement::DropTable(DropTable::new(
+                vec![ObjectName(vec![Ident::new("foo")])],
+                true
+            ))
         );
 
         let sql = "DROP TABLE my_schema.foo";
@@ -141,7 +154,7 @@ mod tests {
         assert_eq!(
             stmts.pop().unwrap(),
             Statement::DropTable(DropTable::new(
-                ObjectName(vec![Ident::new("my_schema"), Ident::new("foo")]),
+                vec![ObjectName(vec![Ident::new("my_schema"), Ident::new("foo")])],
                 false
             ))
         );
@@ -153,11 +166,11 @@ mod tests {
         assert_eq!(
             stmts.pop().unwrap(),
             Statement::DropTable(DropTable::new(
-                ObjectName(vec![
+                vec![ObjectName(vec![
                     Ident::new("my_catalog"),
                     Ident::new("my_schema"),
                     Ident::new("foo")
-                ]),
+                ])],
                 false
             ))
         )

src/sql/src/statements/drop.rs

@@ -20,22 +20,23 @@ use sqlparser_derive::{Visit, VisitMut};
 /// DROP TABLE statement.
 #[derive(Debug, Clone, PartialEq, Eq, Visit, VisitMut)]
 pub struct DropTable {
-    table_name: ObjectName,
+    table_names: Vec<ObjectName>,
+
     /// drop table if exists
     drop_if_exists: bool,
 }
 
 impl DropTable {
     /// Creates a statement for `DROP TABLE`
-    pub fn new(table_name: ObjectName, if_exists: bool) -> Self {
+    pub fn new(table_names: Vec<ObjectName>, if_exists: bool) -> Self {
         Self {
-            table_name,
+            table_names,
             drop_if_exists: if_exists,
         }
     }
 
-    pub fn table_name(&self) -> &ObjectName {
-        &self.table_name
+    pub fn table_names(&self) -> &[ObjectName] {
+        &self.table_names
     }
 
     pub fn drop_if_exists(&self) -> bool {
@@ -49,8 +50,14 @@ impl Display for DropTable {
         if self.drop_if_exists() {
             f.write_str(" IF EXISTS")?;
         }
-        let table_name = self.table_name();
-        write!(f, r#" {table_name}"#)
+        let table_names = self.table_names();
+        for (i, table_name) in table_names.iter().enumerate() {
+            if i > 0 {
+                f.write_str(",")?;
+            }
+            write!(f, " {}", table_name)?;
+        }
+        Ok(())
     }
 }
 
@@ -206,6 +213,27 @@ DROP TABLE test"#,
             }
         }
 
+        let sql = r"drop table test1, test2;";
+        let stmts =
+            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
+                .unwrap();
+        assert_eq!(1, stmts.len());
+        assert_matches!(&stmts[0], Statement::DropTable { .. });
+
+        match &stmts[0] {
+            Statement::DropTable(set) => {
+                let new_sql = format!("\n{}", set);
+                assert_eq!(
+                    r#"
+DROP TABLE test1, test2"#,
+                    &new_sql
+                );
+            }
+            _ => {
+                unreachable!();
+            }
+        }
+
         let sql = r"drop table if exists test;";
         let stmts =
             ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())