[feat][coordinator]Add restore autoincrement

Author: LiuRuoyu01

src/coordinator/auto_increment_control.cc

@@ -25,6 +25,7 @@
 #include "common/logging.h"
 #include "common/synchronization.h"
 #include "engine/snapshot.h"
+#include "fmt/format.h"
 #include "proto/error.pb.h"
 #include "server/server.h"
 
@@ -255,6 +256,45 @@ void AutoIncrementControl::GetLeaderLocation(pb::common::Location& leader_server
   GetServerLocation(leader_raft_location, leader_server_location);
 }
 
+butil::Status AutoIncrementControl::CreateOrUpdateAutoIncrement(
+    int64_t table_id, int64_t start_id, pb::coordinator_internal::MetaIncrement& meta_increment) {
+  DINGO_LOG(INFO) << table_id << " | " << start_id << " | ";
+  int64_t source_start_id = 0;
+  butil::Status ret = GetAutoIncrement(table_id, source_start_id);
+  if (ret.ok()) {
+    // update
+    DINGO_LOG(INFO) << "[restore] update auto increment, table id: " << table_id << ", start id: " << start_id;
+    if (start_id <= source_start_id) {
+      DINGO_LOG(ERROR) << "table id: " << table_id << " : " << start_id << " <= " << source_start_id;
+      return butil::Status(
+          pb::error::Errno::EILLEGAL_PARAMTETERS,
+          fmt::format("[restore] illegal parameters, start id :{}  equal or smaller than source start id :{}", start_id,
+                      source_start_id));
+    }
+    auto* auto_increment = meta_increment.add_auto_increment();
+    auto_increment->set_id(table_id);
+    auto* increment = auto_increment->mutable_increment();
+    increment->set_start_id(start_id);
+    increment->set_source_start_id(source_start_id);
+    increment->set_update_type(pb::coordinator_internal::AutoIncrementUpdateType::UPDATE_ONLY);
+    auto_increment->set_op_type(pb::coordinator_internal::MetaIncrementOpType::UPDATE);
+
+  } else if (ret.error_code() == pb::error::Errno::EAUTO_INCREMENT_NOT_FOUND) {
+    // create
+    DINGO_LOG(INFO) << "[restore] create auto increment, table id: " << table_id << ", start id: " << start_id;
+    auto* auto_increment = meta_increment.add_auto_increment();
+    auto_increment->set_id(table_id);
+    auto* increment = auto_increment->mutable_increment();
+    increment->set_start_id(start_id);
+    auto_increment->set_op_type(pb::coordinator_internal::MetaIncrementOpType::CREATE);
+  } else {
+    DINGO_LOG(ERROR) << ret.error_cstr();
+    return ret;
+  }
+
+  return butil::Status::OK();
+}
+
 void AutoIncrementControl::GetServerLocation(pb::common::Location& raft_location,
                                              pb::common::Location& server_location) {
   // find in cache

src/coordinator/auto_increment_control.h

@@ -21,6 +21,7 @@
 #include <string>
 
 #include "butil/containers/flat_map.h"
+#include "butil/status.h"
 #include "common/meta_control.h"
 #include "engine/engine.h"
 #include "proto/coordinator_internal.pb.h"
@@ -73,6 +74,10 @@ class AutoIncrementControl : public MetaControl {
                                       pb::coordinator_internal::MetaIncrement &meta_increment);
   butil::Status DeleteAutoIncrement(int64_t table_id, pb::coordinator_internal::MetaIncrement &meta_increment);
 
+  // restore auto_increment_map_
+  butil::Status CreateOrUpdateAutoIncrement(int64_t table_id, int64_t start_id,
+                                            pb::coordinator_internal::MetaIncrement &meta_increment);
+
   // Get raft leader's server location
   void GetLeaderLocation(pb::common::Location &leader_server_location) override;
 

src/coordinator/coordinator_control_meta.cc

@@ -1789,6 +1789,20 @@ butil::Status CoordinatorControl::RestoreIndexMeta(int64_t schema_id, int64_t in
     return butil::Status(pb::error::Errno::EILLEGAL_PARAMTETERS, "[br] index_id is already used by other index");
   }
 
+  if (table_definition.auto_increment() > 0) {
+    auto status =
+        AutoIncrementControl::SyncSendCreateAutoIncrementInternal(index_id, table_definition.auto_increment());
+    if (!status.ok()) {
+      DINGO_LOG(ERROR) << fmt::format("send create auto increment internal error, code: {}, message: {} ",
+                                      status.error_code(), status.error_str())
+                       << ", table_definition=" << table_definition.ShortDebugString();
+      return butil::Status(pb::error::Errno::EAUTO_INCREMENT_WHILE_CREATING_TABLE,
+                           fmt::format("[restore] send create auto increment internal error, code: {}, message: {}",
+                                       status.error_code(), status.error_str()));
+    }
+    DINGO_LOG(INFO) << "Restore AutoIncrement send create auto increment internal success";
+  }
+
   // update index_name_map_safe_temp_
   if (index_name_map_safe_temp_.PutIfAbsent(new_index_check_name, index_id) < 0) {
     DINGO_LOG(INFO) << " [br] index_name" << table_definition.name() << " is exist, when insert index_id=" << index_id;

src/server/meta_service.cc

@@ -24,6 +24,7 @@
 #include "butil/containers/flat_map.h"
 #include "butil/status.h"
 #include "common/constant.h"
+#include "common/context.h"
 #include "common/logging.h"
 #include "coordinator/auto_increment_control.h"
 #include "coordinator/coordinator_control.h"
@@ -4029,6 +4030,73 @@ void MetaServiceImpl::ImportIdEpochType(google::protobuf::RpcController *control
   }
 }
 
+void DoCreateOrUpdateAutoIncrements(google::protobuf::RpcController *controller,
+                                    const pb::meta::CreateOrUpdateAutoIncrementsRequest *request,
+                                    pb::meta::CreateOrUpdateAutoIncrementsResponse *response, TrackClosure *done,
+                                    std::shared_ptr<AutoIncrementControl> auto_increment_control,
+                                    std::shared_ptr<Engine> raft_engine) {
+  brpc::ClosureGuard done_guard(done);
+
+  DINGO_LOG(INFO) << request->ShortDebugString();
+
+  pb::coordinator_internal::MetaIncrement meta_increment;
+
+  for (const auto &table_increment : request->table_increment_group().table_increments()) {
+    if (!auto_increment_control->IsLeader()) {
+      return auto_increment_control->RedirectResponse(response);
+    }
+    auto table_id = table_increment.table_id();
+    auto start_id = table_increment.start_id();
+    auto ret = auto_increment_control->CreateOrUpdateAutoIncrement(table_id, start_id, meta_increment);
+    if (!ret.ok()) {
+      DINGO_LOG(ERROR) << "failed, " << table_id << "|" << start_id << " , error_msg : " << ret.error_str();
+      response->mutable_error()->set_errcode(static_cast<pb::error::Errno>(ret.error_code()));
+      response->mutable_error()->set_errmsg(ret.error_str());
+      return;
+    }
+  }
+  std::shared_ptr<Context> ctx =
+      std::make_shared<Context>(static_cast<brpc::Controller *>(controller), nullptr, response);
+  ctx->SetRegionId(Constant::kAutoIncrementRegionId);
+  ctx->SetTracker(done->Tracker());
+
+  // this is a async operation will be block by closure
+  auto ret2 = raft_engine->Write(ctx, WriteDataBuilder::BuildWrite(ctx->CfName(), meta_increment));
+  if (!ret2.ok()) {
+    DINGO_LOG(ERROR) << "failed, " << " | " << ret2.error_str();
+    ServiceHelper::SetError(response->mutable_error(), ret2.error_code(), ret2.error_str());
+
+    if (ret2.error_code() == pb::error::Errno::ERAFT_NOTLEADER) {
+      auto_increment_control->RedirectResponse(response);
+    }
+    return;
+  }
+
+  DINGO_LOG(INFO) << "CreateOrUpdateAutoIncrements Success. ";
+}
+
+void MetaServiceImpl::CreateOrUpdateAutoIncrements(google::protobuf::RpcController *controller,
+                                                   const pb::meta::CreateOrUpdateAutoIncrementsRequest *request,
+                                                   pb::meta::CreateOrUpdateAutoIncrementsResponse *response,
+                                                   google::protobuf::Closure *done) {
+  brpc::ClosureGuard done_guard(done);
+  if (!auto_increment_control_->IsLeader()) {
+    return RedirectResponse(response);
+  }
+  DINGO_LOG(INFO) << request->ShortDebugString();
+
+  // Run in queue.
+  auto *svr_done = new CoordinatorServiceClosure(__func__, done_guard.release(), request, response);
+  auto task = std::make_shared<ServiceTask>([this, controller, request, response, svr_done]() {
+    DoCreateOrUpdateAutoIncrements(controller, request, response, svr_done, auto_increment_control_, engine_);
+  });
+  bool ret = worker_set_->ExecuteRR(task);
+  if (!ret) {
+    brpc::ClosureGuard done_guard(svr_done);
+    ServiceHelper::SetError(response->mutable_error(), pb::error::EREQUEST_FULL, "Commit execute queue failed");
+  }
+}
+
 void DoCreateTenant(google::protobuf::RpcController * /*controller*/, const pb::meta::CreateTenantRequest *request,
                     pb::meta::CreateTenantResponse *response, TrackClosure *done,
                     std::shared_ptr<CoordinatorControl> coordinator_control, std::shared_ptr<Engine> /*raft_engine*/) {

src/server/meta_service.h

@@ -229,6 +229,11 @@ class MetaServiceImpl : public pb::meta::MetaService {
   void ImportIdEpochType(google::protobuf::RpcController* controller, const pb::meta::ImportIdEpochTypeRequest* request,
                          pb::meta::ImportIdEpochTypeResponse* response, google::protobuf::Closure* done) override;
 
+  void CreateOrUpdateAutoIncrements(google::protobuf::RpcController* controller,
+                                    const pb::meta::CreateOrUpdateAutoIncrementsRequest* request,
+                                    pb::meta::CreateOrUpdateAutoIncrementsResponse* response,
+                                    google::protobuf::Closure* done) override;
+
   void SetWorkSet(WorkerSetPtr worker_set) { worker_set_ = worker_set; }
 
   // table and index definition convertor