[feat][document] Add document service and raft handler.

Signed-off-by: Ketor <[email protected]>

CMakeLists.txt

@@ -509,6 +509,7 @@ file(GLOB CONFIG_SRCS ${PROJECT_SOURCE_DIR}/src/config/*.cc)
 file(GLOB LOG_SRCS ${PROJECT_SOURCE_DIR}/src/log/*.cc)
 file(GLOB REPORT_SRCS ${PROJECT_SOURCE_DIR}/src/report/*.cc)
 file(GLOB VECTOR_SRCS ${PROJECT_SOURCE_DIR}/src/vector/*.cc)
+file(GLOB DOCUMENT_SRCS ${PROJECT_SOURCE_DIR}/src/document/*.cc)
 file(GLOB SPLIT_SRCS ${PROJECT_SOURCE_DIR}/src/split/*.cc)
 file(GLOB RAFT_SRCS ${PROJECT_SOURCE_DIR}/src/raft/*.cc)
 file(GLOB ENGINE_SRCS ${PROJECT_SOURCE_DIR}/src/engine/*.cc)
@@ -542,6 +543,7 @@ add_library(DINGODB_OBJS
             ${LOG_SRCS}
             ${REPORT_SRCS}
             ${VECTOR_SRCS}
+            ${DOCUMENT_SRCS}
             ${SPLIT_SRCS}
             ${RAFT_SRCS}
             ${ENGINE_SRCS}
@@ -633,6 +635,7 @@ add_executable(dingodb_client
                 src/common/service_access.cc
                 src/coprocessor/utils.cc
                 src/vector/codec.cc
+                src/document/codec.cc
                 ${SERIAL1_SRCS}
                 ${SERIAL2_SRCS}
                 ${VERSION_SRCS} $<TARGET_OBJECTS:PROTO_OBJS>)

conf/document-gflags.conf

@@ -4,12 +4,8 @@
 -ip2hostname=false
 -min_system_disk_capacity_free_ratio=0.05
 -min_system_memory_capacity_free_ratio=0.20
--omp_num_threads=1
--vector_operation_parallel_thread_num=16
--max_hnsw_memory_size_of_region=2147483648
 -max_short_value_in_write_cf=8
--vector_max_batch_count=1024
--hnsw_max_init_max_elements=40000
+-document_max_batch_count=1024
 -braft_use_align_hearbeat=true
 -bdb_env_cache_size_gb=8
 -bdb_txn_max=65536

proto/error.proto

@@ -186,7 +186,7 @@ enum Errno {
   EREGION_NOT_NEIGHBOR = 60013;
   EREGION_EMPTY_ID = 60014;
 
-  // vector [70000, 80000)
+  // vector [70000, 71000)
   EVECTOR_INDEX_NOT_FOUND = 70000;
   EVECTOR_NOT_SUPPORT = 70001;
   EVECTOR_NOT_SUPPORT_DIMENSION = 70002;
@@ -216,6 +216,36 @@ enum Errno {
   EVECTOR_INDEX_LOAD_SNAPSHOT = 70026;
   EVECTOR_INDEX_SNAPSHOT_VERSION_NOT_MATCH = 70027;
 
+  // document [71000, 72000)
+  EDOCUMENT_INDEX_NOT_FOUND = 71000;
+  EDOCUMENT_NOT_SUPPORT = 71001;
+  EDOCUMENT_NOT_SUPPORT_DIMENSION = 71002;
+  EDOCUMENT_NOT_SUPPORT_INDEX_TYPE = 71003;
+  EDOCUMENT_EMPTY = 71004;
+  EDOCUMENT_INVALID = 71005;
+  EDOCUMENT_INDEX_OFFLINE = 71006;
+  EDOCUMENT_SCALAR_DATA_NOT_FOUND = 71007;
+  EDOCUMENT_SNAPSHOT_NOT_FOUND = 71008;
+  EDOCUMENT_SNAPSHOT_INVALID = 71009;
+  EDOCUMENT_INDEX_EXCEED_MAX_ELEMENTS = 71010;
+  EDOCUMENT_NOT_FOUND = 71011;
+  EDOCUMENT_INDEX_FULL = 71012;
+  EDOCUMENT_EXCEED_MAX_BATCH_COUNT = 71013;
+  EDOCUMENT_EXCEED_MAX_REQUEST_SIZE = 71014;
+  EDOCUMENT_SNAPSHOT_EXIST = 71015;
+  EDOCUMENT_NOT_NEED_SNAPSHOT = 71016;
+  EDOCUMENT_INDEX_EXIST = 71017;
+  EDOCUMENT_INDEX_SWITCHING = 71018;
+  EDOCUMENT_INDEX_NOT_READY = 71019;
+  EDOCUMENT_NOT_TRAIN = 71020;
+  EDOCUMENT_INDEX_BUILD_ERROR = 71021;
+  EDOCUMENT_INDEX_SAVE_SUCCESS = 71022;
+  EDOCUMENT_INDEX_BUILD_CONFLICT = 71023;
+  EDOCUMENT_INDEX_REBUILD_CONFLICT = 71024;
+  EDOCUMENT_INDEX_MERGE_RESULT = 71025;
+  EDOCUMENT_INDEX_LOAD_SNAPSHOT = 71026;
+  EDOCUMENT_INDEX_SNAPSHOT_VERSION_NOT_MATCH = 71027;
+
   // file [80000, 90000)
   EFILE_NOT_FOUND_READER = 80000;
   EFILE_READ = 80001;

proto/store.proto

@@ -554,8 +554,13 @@ message TxnGetResponse {
   // The vector value associated with the key, if not found, is empty and error is NOT_FOUND
   // dingo-store do not accept NULL value, so if value is NULL, it means the key is not exist
   dingodb.pb.common.VectorWithId vector = 5;
+  // for document index region:
+  // The document value associated with the key, if not found, is empty and error is NOT_FOUND
+  // dingo-store do not accept NULL value, so if value is NULL, it means the key is not exist
+  dingodb.pb.common.DocumentWithId document = 6;
+
   // True if the key does not exist in the database.
-  // bool not_found = 6;
+  // bool not_found = 10;
 }
 
 // TxnScan fetches values for a range of keys in the transaction with start_ts.
@@ -604,6 +609,9 @@ message TxnScanResponse {
   // the last iteratered key of this scan response.
   // if end_key is null, means scan do not successfully iterate any key.
   bytes end_key = 7;
+
+  // for document index region, documents is the return value of the scan request
+  repeated dingodb.pb.common.DocumentWithId documents = 8;
 }
 
 // Lock a set of keys to prepare to write to them.
@@ -865,6 +873,12 @@ message TxnBatchGetResponse {
   // all keys. In this case, `vectors` should be empty and the client should redo
   // batch get all the keys after resolving the lock.
   repeated dingodb.pb.common.VectorWithId vectors = 5;
+
+  // for document index region:
+  // When some key is locked but we cannot check locks of
+  // all keys. In this case, `documents` should be empty and the client should redo
+  // batch get all the keys after resolving the lock.
+  repeated dingodb.pb.common.DocumentWithId documents = 6;
 }
 
 // Rollback a prewritten transaction. This will remove the preliminary data from

src/common/helper.cc

@@ -901,7 +901,7 @@ bool Helper::Link(const std::string& old_path, const std::string& new_path) {
 std::vector<std::string> Helper::GetColumnFamilyNamesByRole() {
   if (GetRole() == pb::common::ClusterRole::COORDINATOR) {
     return {Constant::kStoreDataCF, Constant::kStoreMetaCF};
-  } else if (GetRole() == pb::common::ClusterRole::STORE) {
+  } else if (GetRole() == pb::common::ClusterRole::STORE || GetRole() == pb::common::ClusterRole::DOCUMENT) {
     return {Constant::kStoreDataCF, Constant::kStoreMetaCF, Constant::kTxnDataCF, Constant::kTxnLockCF,
             Constant::kTxnWriteCF};
   } else if (GetRole() == pb::common::ClusterRole::INDEX) {
@@ -921,7 +921,7 @@ std::vector<std::string> Helper::GetColumnFamilyNamesByRole() {
 std::vector<std::string> Helper::GetColumnFamilyNamesExecptMetaByRole() {
   if (GetRole() == pb::common::ClusterRole::COORDINATOR) {
     return {Constant::kStoreDataCF};
-  } else if (GetRole() == pb::common::ClusterRole::STORE) {
+  } else if (GetRole() == pb::common::ClusterRole::STORE || GetRole() == pb::common::ClusterRole::DOCUMENT) {
     return {Constant::kStoreDataCF, Constant::kTxnDataCF, Constant::kTxnLockCF, Constant::kTxnWriteCF};
   } else if (GetRole() == pb::common::ClusterRole::INDEX) {
     return {Constant::kStoreDataCF,  Constant::kTxnDataCF,      Constant::kTxnLockCF,
@@ -935,7 +935,7 @@ std::vector<std::string> Helper::GetColumnFamilyNamesExecptMetaByRole() {
 std::vector<std::string> Helper::GetColumnFamilyNames(const std::string& key) {
   if (GetRole() == pb::common::ClusterRole::COORDINATOR) {
     return {Constant::kStoreDataCF};
-  } else if (GetRole() == pb::common::ClusterRole::STORE) {
+  } else if (GetRole() == pb::common::ClusterRole::STORE || GetRole() == pb::common::ClusterRole::DOCUMENT) {
     if (IsExecutorTxn(key) || IsClientTxn(key)) {
       return {Constant::kTxnDataCF, Constant::kTxnLockCF, Constant::kTxnWriteCF};
     }
@@ -958,7 +958,7 @@ void Helper::GetColumnFamilyNames(const std::string& key, std::vector<std::strin
   if (GetRole() == pb::common::ClusterRole::COORDINATOR) {
     raw_cf_names.push_back(Constant::kStoreDataCF);
     return;
-  } else if (GetRole() == pb::common::ClusterRole::STORE) {
+  } else if (GetRole() == pb::common::ClusterRole::STORE || GetRole() == pb::common::ClusterRole::DOCUMENT) {
     if (IsExecutorTxn(key) || IsClientTxn(key)) {
       txn_cf_names.push_back(Constant::kTxnDataCF);
       txn_cf_names.push_back(Constant::kTxnLockCF);

src/common/tracker.cc

@@ -77,4 +77,7 @@ uint64_t Tracker::StoreWriteTime() const { return metrics_.store_write_time_ns;
 void Tracker::SetVectorIndexWriteTime(uint64_t elapsed_time) { metrics_.vector_index_write_time_ns = elapsed_time; }
 uint64_t Tracker::VectorIndexwriteTime() const { return metrics_.vector_index_write_time_ns; }
 
+void Tracker::SetDocumentIndexWriteTime(uint64_t elapsed_time) { metrics_.document_index_write_time_ns = elapsed_time; }
+uint64_t Tracker::DocumentIndexwriteTime() const { return metrics_.document_index_write_time_ns; }
+
 }  // namespace dingodb

src/common/tracker.h

@@ -38,6 +38,7 @@ class Tracker {
     uint64_t raft_apply_time_ns{0};
     uint64_t store_write_time_ns{0};
     uint64_t vector_index_write_time_ns{0};
+    uint64_t document_index_write_time_ns{0};
   };
 
   void SetTotalRpcTime();
@@ -64,6 +65,9 @@ class Tracker {
   void SetVectorIndexWriteTime(uint64_t elapsed_time);
   uint64_t VectorIndexwriteTime() const;
 
+  void SetDocumentIndexWriteTime(uint64_t elapsed_time);
+  uint64_t DocumentIndexwriteTime() const;
+
  private:
   uint64_t start_time_;
   uint64_t last_time_;