instance HTTP APIs : ForceCheckpoint, CompactTable

infiniflow · Oct 14, 2024 · 59515c6 · 59515c6
1 parent 61d536b
commit 59515c6
Show file tree

Hide file tree

Showing 2 changed files with 87 additions and 2 deletions.
diff --git a/src/main/infinity.cpp b/src/main/infinity.cpp
@@ -49,6 +49,7 @@ import optimize_statement;
 import alter_statement;
 import statement_common;
 import admin_statement;
+import compact_statement;
 
 import create_schema_info;
 import drop_schema_info;
@@ -737,7 +738,7 @@ QueryResult Infinity::ShowObjects() {
     return result;
 }
 
-QueryResult Infinity::ShowObject(const String& filename) {
+QueryResult Infinity::ShowObject(const String &filename) {
     UniquePtr<QueryContext> query_context_ptr = GetQueryContext();
     UniquePtr<ShowStatement> show_statement = MakeUnique<ShowStatement>();
     show_statement->show_type_ = ShowStmtType::kPersistenceObject;
@@ -778,7 +779,7 @@ QueryResult Infinity::ShowMemoryAllocations() {
     return result;
 }
 
-QueryResult Infinity::ShowFunction(const String& function_name) {
+QueryResult Infinity::ShowFunction(const String &function_name) {
     UniquePtr<QueryContext> query_context_ptr = GetQueryContext();
     UniquePtr<ShowStatement> show_statement = MakeUnique<ShowStatement>();
     show_statement->show_type_ = ShowStmtType::kFunction;
@@ -1039,6 +1040,39 @@ QueryResult Infinity::Cleanup() {
     return result;
 }
 
+QueryResult Infinity::ForceCheckpoint() {
+    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 flush_statement = MakeUnique<FlushStatement>();
+    flush_statement->type_ = infinity::FlushType::kData;
+
+    QueryResult result = query_context_ptr->QueryStatement(flush_statement.get());
+
+    return result;
+}
+
+QueryResult Infinity::CompactTable(const String &db_name, const String &table_name) {
+    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 compact_statement = MakeUnique<ManualCompactStatement>(db_name, table_name);
+
+    QueryResult result = query_context_ptr->QueryStatement(compact_statement.get());
+
+    return result;
+}
+
 QueryResult Infinity::AdminShowCatalogs() {
     auto query_context_ptr = MakeUnique<QueryContext>(session_.get());
     query_context_ptr->Init(InfinityContext::instance().config(),

diff --git a/src/network/http_server.cpp b/src/network/http_server.cpp
@@ -3922,6 +3922,55 @@ class ShowMemoryAllocationsHandler final : public HttpRequestHandler {
     }
 };
 
+class ForceGlobalCheckpointHandler final : public HttpRequestHandler {
+public:
+    SharedPtr<OutgoingResponse> handle(const SharedPtr<IncomingRequest> &request) final {
+        auto infinity = Infinity::RemoteConnect();
+        DeferFn defer_fn([&]() { infinity->RemoteDisconnect(); });
+
+        nlohmann::json json_response;
+        HTTPStatus http_status;
+        QueryResult result = infinity->ForceCheckpoint();
+
+        if (result.IsOk()) {
+            json_response["error_code"] = 0;
+            http_status = HTTPStatus::CODE_200;
+        } else {
+            json_response["error_code"] = result.ErrorCode();
+            json_response["error_message"] = result.ErrorMsg();
+            http_status = HTTPStatus::CODE_500;
+        }
+        return ResponseFactory::createResponse(http_status, json_response.dump());
+    }
+};
+
+class CompactTableHandler final : public HttpRequestHandler {
+public:
+    SharedPtr<OutgoingResponse> handle(const SharedPtr<IncomingRequest> &request) final {
+        auto infinity = Infinity::RemoteConnect();
+        DeferFn defer_fn([&]() { infinity->RemoteDisconnect(); });
+
+        String data_body = request->readBodyToString();
+        nlohmann::json json_body = nlohmann::json::parse(data_body);
+        String db_name = json_body["db_name"];
+        String table_name = json_body["table_name"];
+
+        nlohmann::json json_response;
+        HTTPStatus http_status;
+        QueryResult result = infinity->CompactTable(db_name, table_name);
+
+        if (result.IsOk()) {
+            json_response["error_code"] = 0;
+            http_status = HTTPStatus::CODE_200;
+        } else {
+            json_response["error_code"] = result.ErrorCode();
+            json_response["error_message"] = result.ErrorMsg();
+            http_status = HTTPStatus::CODE_500;
+        }
+        return ResponseFactory::createResponse(http_status, json_response.dump());
+    }
+};
+
 class AdminShowCurrentNodeHandler final : public HttpRequestHandler {
 public:
     SharedPtr<OutgoingResponse> handle(const SharedPtr<IncomingRequest> &request) final {
@@ -4267,6 +4316,8 @@ void HTTPServer::Start(const String &ip_address, u16 port) {
     router->route("GET", "/instance/memory", MakeShared<ShowMemoryHandler>());
     router->route("GET", "/instance/memory/objects", MakeShared<ShowMemoryObjectsHandler>());
     router->route("GET", "/instance/memory/allocations", MakeShared<ShowMemoryAllocationsHandler>());
+    router->route("POST", "/instance/flush", MakeShared<ForceGlobalCheckpointHandler>());
+    router->route("POST", "/instance/table/compact", MakeShared<CompactTableHandler>());
 
     // variable
     router->route("GET", "/variables/global", MakeShared<ShowGlobalVariablesHandler>());