Add cleanup cmd.

python/infinity_sdk/infinity/remote_thrift/client.py

@@ -287,3 +287,6 @@ def add_columns(self, db_name: str, table_name: str, column_defs: list):
     def drop_columns(self, db_name: str, table_name: str, column_names: list):
         return self.client.DropColumns(DropColumnsRequest(session_id=self.session_id, db_name=db_name, table_name=table_name,
                                                           column_names=column_names))
+
+    def cleanup(self):
+        return self.client.Cleanup(CommonRequest(session_id=self.session_id))

python/infinity_sdk/infinity/remote_thrift/infinity.py

@@ -91,6 +91,13 @@ def get_database(self, db_name: str):
         else:
             raise InfinityException(res.error_code, res.error_msg)
 
+    def cleanup(self):
+        res = self._client.cleanup()
+        if res.error_code == ErrorCode.OK:
+            return res
+        else:
+            raise InfinityException(res.error_code, res.error_msg)
+
     def disconnect(self):
         res = self._client.disconnect()
         if res.error_code == ErrorCode.OK:

python/test_pysdk/test_cleanup.py

@@ -0,0 +1,70 @@
+import pathlib
+from numpy import dtype
+import pytest
+import os
+import sys
+from common import common_values
+import infinity_embedded
+import infinity
+from infinity.common import ConflictType
+import infinity.index as index
+
+current_dir = os.path.dirname(os.path.abspath(__file__))
+parent_dir = os.path.dirname(current_dir)
+if parent_dir not in sys.path:
+    sys.path.insert(0, parent_dir)
+import importlib
+from infinity_http import infinity_http
+
+
+class TestCleanup:
+    @pytest.fixture(autouse=True)
+    def setup_and_teardown(self, local_infinity, http, suffix):
+        if local_infinity:
+            module = importlib.import_module("infinity_embedded.index")
+            globals()["index"] = module
+            self.uri = common_values.TEST_LOCAL_PATH
+            self.infinity_obj = infinity_embedded.connect(self.uri)
+        elif http:
+            self.uri = common_values.TEST_LOCAL_HOST
+            self.infinity_obj = infinity_http()
+        else:
+            self.uri = common_values.TEST_LOCAL_HOST
+            self.infinity_obj = infinity.connect(self.uri)
+
+        self.suffix = suffix
+        yield
+
+        res = self.infinity_obj.disconnect()
+        assert res.error_code == infinity.ErrorCode.OK
+
+    @pytest.mark.usefixtures("skip_if_http")
+    def test_invalidate_fulltext_cache(self):
+        table_name = "test_add_column" + self.suffix
+        db_obj = self.infinity_obj.get_database("default_db")
+        db_obj.drop_table(table_name, ConflictType.Ignore)
+        table_obj = db_obj.create_table(
+            table_name,
+            {
+                "c1": {"type": "varchar"},
+                "c2": {"type": "varchar"},
+            },
+        )
+        table_obj.insert([{"c1": "text1", "c2": "text2"}])
+
+        table_obj.create_index("idx1", index.IndexInfo("c1", index.IndexType.FullText))
+        table_obj.create_index("idx2", index.IndexInfo("c2", index.IndexType.FullText))
+
+        res = (
+            table_obj.output(["c1"])
+            .match_text(fields="c1", matching_text="text1", topn=1)
+            .to_result()
+        )
+
+        table_obj.drop_index("idx1")
+
+        self.infinity_obj.cleanup()
+        dropped_index_dirs = pathlib.Path("/var/infinity/data").rglob("*idx1*")
+        assert len(list(dropped_index_dirs)) == 0
+
+        db_obj.drop_table(table_name)

src/common/utility/exception.cpp

@@ -62,7 +62,6 @@ void UnrecoverableError(const String &message, const char *file_name, u32 line)
     if (storage != nullptr) {
         CleanupInfoTracer *cleanup_tracer = storage->cleanup_info_tracer();
         String error_msg = cleanup_tracer->GetCleanupInfo();
-        std::cout << error_msg << std::endl;
         LOG_ERROR(std::move(error_msg));
     }
 

src/executor/operator/physical_command.cpp

@@ -46,6 +46,9 @@ import table_entry;
 import txn;
 import cleanup_scanner;
 import infinity_context;
+import periodic_trigger;
+import bg_task;
+import wal_manager;
 
 namespace infinity {
 
@@ -299,6 +302,24 @@ bool PhysicalCommand::Execute(QueryContext *query_context, OperatorState *operat
             table_entry->SetUnlock();
             break;
         }
+        case CommandType::kCleanup: {
+            {
+                Txn *txn = query_context->GetTxn();
+                auto checkpoint_task = MakeShared<ForceCheckpointTask>(txn, false /*is_full_checkpoint*/);
+                WalManager *wal_manager = query_context->storage()->wal_manager();
+                wal_manager->Checkpoint(checkpoint_task.get());
+            }
+            {
+                CleanupPeriodicTrigger *cleanup_trigger = query_context->storage()->bg_processor()->cleanup_trigger();
+                SharedPtr<CleanupTask> cleanup_task = cleanup_trigger->CreateCleanupTask();
+                if (cleanup_task.get() != nullptr) {
+                    cleanup_task->Execute();
+                } else {
+                    LOG_DEBUG("Skip cleanup");
+                }
+            }
+            break;
+        }
         default: {
             String error_message = fmt::format("Invalid command type: {}", command_info_->ToString());
             UnrecoverableError(error_message);

src/main/infinity.cpp

@@ -1099,4 +1099,21 @@ QueryResult Infinity::DropColumns(const String &db_name, const String &table_nam
     return result;
 }
 
+QueryResult Infinity::Cleanup() {
+    auto query_context_ptr = MakeUnique<QueryContext>(session_.get());
+    query_context_ptr->Init(InfinityContext::instance().config(),
+                            InfinityContext::instance().task_scheduler(),
+                            InfinityContext::instance().storage(),
+                            InfinityContext::instance().resource_manager(),
+                            InfinityContext::instance().session_manager(),
+                            InfinityContext::instance().persistence_manager());
+
+    auto command_statement = MakeUnique<CommandStatement>();
+
+    command_statement->command_info_ = MakeUnique<CleanupCmd>();
+
+    QueryResult result = query_context_ptr->QueryStatement(command_statement.get());
+    return result;
+}
+
 } // namespace infinity

src/main/infinity.cppm

@@ -163,6 +163,8 @@ public:
 
     QueryResult DropColumns(const String &db_name, const String &table_name, Vector<String> column_names);
 
+    QueryResult Cleanup();
+
 private:
     SharedPtr<BaseSession> session_{};
 };