Skip to content

Commit

Permalink
instance HTTP APIs : ForceCheckpoint, CompactTable (#2036)
Browse files Browse the repository at this point in the history
### What problem does this PR solve?

New HTTP APIs for instance:
- Force GlobalCheckpoint. (Not documented.)
- Compact Table. (Not documented.)

Issue link: #1937 

### Type of change

- [x] New Feature (non-breaking change which adds functionality)
  • Loading branch information
vsian authored Oct 14, 2024
1 parent 61d536b commit b7706b5
Show file tree
Hide file tree
Showing 2 changed files with 87 additions and 2 deletions.
38 changes: 36 additions & 2 deletions src/main/infinity.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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(),
Expand Down
51 changes: 51 additions & 0 deletions src/network/http_server.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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>());
Expand Down

0 comments on commit b7706b5

Please sign in to comment.