Add "drop index" ddl

production/catalog/parser/src/parser.yy

@@ -232,6 +232,10 @@ drop_statement:
       $$ = std::make_unique<drop_statement_t>(drop_type_t::drop_relationship, $4);
       $$->if_exists = $3;
   }
+| DROP INDEX opt_if_exists IDENTIFIER {
+      $$ = std::make_unique<drop_statement_t>(drop_type_t::drop_index, $4);
+      $$->if_exists = $3;
+  }
 ;
 
 use_statement:

production/catalog/src/catalog.cpp

@@ -80,6 +80,11 @@ void drop_relationship(const string& name, bool throw_unless_exists)
     return ddl_executor_t::get().drop_relationship(name, throw_unless_exists);
 }
 
+void drop_index(const string& name, bool throw_unless_exists)
+{
+    return ddl_executor_t::get().drop_index(name, throw_unless_exists);
+}
+
 gaia_id_t create_index(
     const std::string& index_name,
     bool unique,

production/catalog/src/ddl_executor.cpp

@@ -1000,5 +1000,22 @@ gaia_id_t ddl_executor_t::create_index(
     return index_id;
 }
 
+void ddl_executor_t::drop_index(const std::string& name, bool throw_unless_exists)
+{
+    auto_transaction_t txn(false);
+    auto index_iter = gaia_index_t::list().where(gaia_index_expr::name == name).begin();
+    if (index_iter == gaia_index_t::list().end())
+    {
+        if (throw_unless_exists)
+        {
+            throw index_not_exists(name);
+        }
+        return;
+    }
+    index_iter->table().gaia_indexes().remove(index_iter->gaia_id());
+    index_iter->delete_row();
+    txn.commit();
+}
+
 } // namespace catalog
 } // namespace gaia

production/catalog/tests/test_ddl_execution.cpp

@@ -141,3 +141,43 @@ DROP TABLE IF EXISTS t2;
     ASSERT_NO_THROW(parser.parse_line("DROP RELATIONSHIP r3;"));
     ASSERT_THROW(execute(parser.statements), relationship_not_exists);
 }
+
+TEST_F(ddl_execution_test, drop_index)
+{
+    const string create_index_ddl = R"(
+DROP TABLE IF EXISTS t;
+CREATE TABLE IF NOT EXISTS t(c INT32);
+
+DROP INDEX IF EXISTS c_i;
+CREATE INDEX IF NOT EXISTS c_i ON t(c);
+)";
+
+    ddl::parser_t parser;
+    ASSERT_NO_THROW(parser.parse_line(create_index_ddl));
+    ASSERT_NO_THROW(execute(parser.statements));
+
+    {
+        gaia::direct_access::auto_transaction_t txn(false);
+        ASSERT_EQ(gaia_table_t::list().where(gaia_table_expr::name == "t").size(), 1);
+        ASSERT_EQ(
+            gaia_table_t::list()
+                .where(gaia::catalog::gaia_table_expr::name == "t")
+                .begin()
+                ->gaia_indexes()
+                .where(gaia_index_expr::name == "c_i")
+                .size(),
+            1);
+    }
+
+    ASSERT_NO_THROW(parser.parse_line("DROP INDEX IF EXISTS c_i;"));
+    ASSERT_NO_THROW(execute(parser.statements));
+
+    {
+        gaia::direct_access::auto_transaction_t txn(false);
+        ASSERT_EQ(gaia_table_t::list().where(gaia_table_expr::name == "t").begin()->gaia_indexes().size(), 0);
+        ASSERT_EQ(gaia_index_t::list().where(gaia_index_expr::name == "c_i").size(), 0);
+    }
+
+    ASSERT_NO_THROW(parser.parse_line("DROP INDEX c_i;"));
+    ASSERT_THROW(execute(parser.statements), index_not_exists);
+}

production/inc/gaia_internal/catalog/catalog.hpp

@@ -300,6 +300,20 @@ class index_already_exists : public gaia::common::gaia_exception
     }
 };
 
+/**
+ * Thrown when the index of the given name does not exists.
+ */
+class index_not_exists : public gaia::common::gaia_exception
+{
+public:
+    explicit index_not_exists(const std::string& name)
+    {
+        std::stringstream message;
+        message << "The index '" << name << "' does not exist.";
+        m_message = message.str();
+    }
+};
+
 /**
  * Thrown when the field map is invalid.
  */
@@ -569,6 +583,7 @@ enum class drop_type_t : uint8_t
     drop_table,
     drop_database,
     drop_relationship,
+    drop_index,
 };
 
 struct drop_statement_t : statement_t
@@ -686,7 +701,7 @@ void drop_database(const std::string& name, bool throw_unless_exists = true);
  *
  * @param db_name database name
  * @param name table name
- * @param throw_unless_exists throw an execption unless the database exists
+ * @param throw_unless_exists throw an execption unless the table exists
  * @throw table_not_exists
  */
 void drop_table(const std::string& db_name, const std::string& name, bool throw_unless_exists = true);
@@ -699,7 +714,7 @@ void drop_table(const std::string& db_name, const std::string& name, bool throw_
  * which is not available to the catalog implementation.
  *
  * @param name table name
- * @param throw_unless_exists throw an execption unless the database exists
+ * @param throw_unless_exists throw an execption unless the table exists
  * @throw table_not_exists
  */
 void drop_table(const std::string& name, bool throw_unless_exists = true);
@@ -780,11 +795,20 @@ gaia::common::gaia_id_t create_relationship(
  * Delete a given relationship.
  *
  * @param name of the relationship
- * @param throw_unless_exists throw an execption unless the database exists
- * @throw table_not_exists
+ * @param throw_unless_exists throw an execption unless the relationship exists
+ * @throw relationship_not_exists
  */
 void drop_relationship(const std::string& name, bool throw_unless_exists = true);
 
+/**
+ * Delete a given index.
+ *
+ * @param name of the index
+ * @param throw_unless_exists throw an execption unless the index exists
+ * @throw index_not_exists
+ */
+void drop_index(const std::string& name, bool throw_unless_exists = true);
+
 /**
  * Find the database id given its name
  *

production/inc/gaia_internal/catalog/ddl_execution.hpp

@@ -75,6 +75,10 @@ inline void execute(std::vector<std::unique_ptr<ddl::statement_t>>& statements)
             {
                 drop_relationship(drop_stmt->name, !drop_stmt->if_exists);
             }
+            else if (drop_stmt->type == ddl::drop_type_t::drop_index)
+            {
+                drop_index(drop_stmt->name, !drop_stmt->if_exists);
+            }
         }
         else if (stmt->is_type(ddl::statement_type_t::use))
         {

production/inc/gaia_internal/catalog/ddl_executor.hpp

@@ -67,6 +67,8 @@ class ddl_executor_t
 
     void drop_relationship(const std::string& name, bool throw_unless_exists);
 
+    void drop_index(const std::string& name, bool throw_unless_exists);
+
     void drop_table(const std::string& db_name, const std::string& name);
     void drop_database(const std::string& name);