[liboblog] support storage work mode

CMakeLists.txt

@@ -81,7 +81,7 @@ elseif(OB_INCLUDE_UNITTEST)
   add_subdirectory(unittest EXCLUDE_FROM_ALL)
 endif()
 
-if (OB_BUILD_TOOLS)
+if (OB_BUILD_TOOLS OR OB_BUILD_LIBOBLOG)
   add_subdirectory(tools)
 elseif (OB_INCLUDE_TOOLS)
   add_subdirectory(tools EXCLUDE_FROM_ALL)

cmake/RPM.cmake

@@ -60,7 +60,7 @@ install(FILES
   COMPONENT server)
 
 ## oceanbase-devel
-# liboblog.so and libob_sql_proxy_parser_static.a
+# libobcdc.so and libob_sql_proxy_parser_static.a
 set(OCEANBASE_DEVEL_LIB_FILES "")
 set(OCEANBASE_DEVEL_INCLUDE_FILES deps/oblib/src/lib/ob_errno.h)
 set(OCEANBASE_DEVEL_BIN_FILES "")
@@ -81,16 +81,16 @@ endif()
 
 if (OB_BUILD_LIBOBLOG)
   # lib
-  # list(APPEND OCEANBASE_DEVEL_LIB_FILES ${CMAKE_BINARY_DIR}/src/liboblog/src/liboblog.a)
-  list(APPEND OCEANBASE_DEVEL_LIB_FILES ${CMAKE_BINARY_DIR}/src/liboblog/src/liboblog.so)
-  list(APPEND OCEANBASE_DEVEL_LIB_FILES ${CMAKE_BINARY_DIR}/src/liboblog/src/liboblog.so.1)
-  list(APPEND OCEANBASE_DEVEL_LIB_FILES ${CMAKE_BINARY_DIR}/src/liboblog/src/liboblog.so.1.0.0)
+  # list(APPEND OCEANBASE_DEVEL_LIB_FILES ${CMAKE_BINARY_DIR}/tools/libobcdc/src/libobcdc.a)
+  list(APPEND OCEANBASE_DEVEL_LIB_FILES ${CMAKE_BINARY_DIR}/tools/obcdc/src/libobcdc.so)
+  list(APPEND OCEANBASE_DEVEL_LIB_FILES ${CMAKE_BINARY_DIR}/tools/obcdc/src/libobcdc.so.1)
+  list(APPEND OCEANBASE_DEVEL_LIB_FILES ${CMAKE_BINARY_DIR}/tools/obcdc/src/libobcdc.so.1.0.0)
 
   # include lilboblog header
-  list(APPEND OCEANBASE_DEVEL_INCLUDE_FILES src/liboblog/src/liboblog.h)
+  list(APPEND OCEANBASE_DEVEL_INCLUDE_FILES tools/obcdc/src/libobcdc.h)
 
   # bin
-  list(APPEND OCEANBASE_DEVEL_BIN_FILES ${CMAKE_BINARY_DIR}/src/liboblog/tests/oblog_tailf)
+  list(APPEND OCEANBASE_DEVEL_BIN_FILES ${CMAKE_BINARY_DIR}/tools/obcdc/tests/obcdc_tailf)
 endif()
 
 set(CPACK_RPM_DEVEL_DEFAULT_USER "root")

deps/3rd/oceanbase.el7.aarch64.deps

@@ -21,3 +21,6 @@ obdevtools-cmake-3.20.2-15.el7.aarch64.rpm
 obdevtools-flex-2.5.35-10.el7.aarch64.rpm
 obdevtools-gcc-5.2.0-15.el7.aarch64.rpm
 obdevtools-llvm-11.0.1-40.el7.aarch64.rpm
+
+[tools-deps]
+devdeps-rocksdb-6.22.1-15.el7.aarch64.rpm

deps/3rd/oceanbase.el8.x86_64.deps

@@ -21,3 +21,6 @@ obdevtools-cmake-3.20.2-3.el8.x86_64.rpm
 obdevtools-flex-2.5.35-3.el8.x86_64.rpm
 obdevtools-gcc-5.2.0-3.el8.x86_64.rpm
 obdevtools-llvm-11.0.1-40.el8.x86_64.rpm
+
+[tools-deps]
+devdeps-rocksdb-6.22.1-15.el8.x86_64.rpm

src/CMakeLists.txt

@@ -54,7 +54,6 @@ add_subdirectory(archive)
 add_subdirectory(election)
 add_subdirectory(storage)
 add_subdirectory(observer)
-add_subdirectory(liboblog)
 
 if (OB_ENABLE_SERVER_PCH)
   target_precompile_headers(ob_base INTERFACE ${ob_server_pchs})

src/liboblog/src/ob_log_mock_store_service.h

@@ -1,115 +0,0 @@
-/**
- * Copyright (c) 2021 OceanBase
- * OceanBase CE is licensed under Mulan PubL v2.
- * You can use this software according to the terms and conditions of the Mulan PubL v2.
- * You may obtain a copy of Mulan PubL v2 at:
- *          http://license.coscl.org.cn/MulanPubL-2.0
- * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
- * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
- * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
- * See the Mulan PubL v2 for more details.
- */
-
-#ifndef OCEANBASE_LIBOBLOG_OB_LOG_MOCK_STORE_SERVICE_H_
-#define OCEANBASE_LIBOBLOG_OB_LOG_MOCK_STORE_SERVICE_H_
-
-#include "ob_log_store_service.h"
-
-namespace oceanbase
-{
-namespace liboblog
-{
-class MockObLogStoreService : public IObStoreService
-{
-public:
-  MockObLogStoreService() {}
-  void destroy() {}
-  ~MockObLogStoreService() { destroy(); }
-  int init(const std::string &path) { UNUSED(path); return 0; }
-  int close() { return 0; }
-public:
-  int put(const std::string &key, const ObSlice &value)
-  {
-    UNUSED(key);
-    UNUSED(value);
-    return 0;
-  }
-
-  int put(void *cf_handle, const std::string &key, const ObSlice &value)
-  {
-    UNUSED(key);
-    UNUSED(value);
-    UNUSED(cf_handle);
-    return 0;
-  }
-
-  int batch_write(void *cf_handle, const std::vector<std::string> &keys, const std::vector<ObSlice> &values)
-  {
-    UNUSED(cf_handle);
-    UNUSED(keys);
-    UNUSED(values);
-    return 0;
-  }
-
-  int get(const std::string &key, std::string &value)
-  {
-    UNUSED(key);
-    UNUSED(value);
-    return 0;
-  }
-
-  int get(void *cf_handle, const std::string &key, std::string &value)
-  {
-    UNUSED(cf_handle);
-    UNUSED(key);
-    UNUSED(value);
-    return 0;
-  }
-
-  int del(const std::string &key)
-  {
-    UNUSED(key);
-    return 0;
-  }
-
-  int del(void *cf_handle, const std::string &key)
-  {
-    UNUSED(cf_handle);
-    UNUSED(key);
-    return 0;
-  }
-
-  int create_column_family(const std::string& column_family_name,
-      void *&cf_handle)
-  {
-    UNUSED(column_family_name);
-    cf_handle = this;
-    return 0;
-  }
-
-  int drop_column_family(void *cf_handle)
-  {
-    UNUSED(cf_handle);
-    return 0;
-  }
-
-  int destory_column_family(void *cf_handle)
-  {
-    UNUSED(cf_handle);
-    return 0;
-  }
-
-  void get_mem_usage(const std::vector<uint64_t> ids,
-      const std::vector<void *> cf_handles)
-  {
-    UNUSED(ids);
-    UNUSED(cf_handles);
-  }
-
-};
-
-}
-}
-
-
-#endif

src/liboblog/tests/CMakeLists.txt

@@ -1,7 +0,0 @@
-add_executable(oblog_tailf
-  main.cpp
-  oblog_main.h oblog_main.cpp
-  ob_binlog_record_printer.h ob_binlog_record_printer.cpp
-  )
-target_link_libraries(oblog_tailf PRIVATE oblog)
-target_include_directories(oblog_tailf PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/../src)

tools/CMakeLists.txt

@@ -1,2 +1,3 @@
+add_subdirectory(obcdc)
 add_subdirectory(ob_error)
 add_subdirectory(ob_admin)

tools/obcdc/src/CMakeLists.txt

@@ -1,18 +1,12 @@
-add_library(oblog_base INTERFACE)
-target_include_directories(oblog_base INTERFACE ${CMAKE_CURRENT_SOURCE_DIR})
-target_link_libraries(oblog_base
-  INTERFACE ob_base oblogmsg_static
-  )
-
-add_library(oblog_miner INTERFACE)
-
-target_include_directories(oblog_miner INTERFACE ${CMAKE_CURRENT_SOURCE_DIR})
-target_link_libraries(oblog_miner
-  INTERFACE ob_base oblogmsg_static
+add_library(obcdc_base INTERFACE)
+target_include_directories(obcdc_base INTERFACE ${CMAKE_CURRENT_SOURCE_DIR})
+target_link_libraries(obcdc_base
+  INTERFACE ob_base oceanbase::oblogmsg_static easy oblib
+  ${DEP_DIR}/lib64/librocksdb.a
   )
 
 set(SRC_LIST
-  liboblog.h liboblog.cpp
+  libobcdc.h libobcdc.cpp
   ob_concurrent_seq_queue.h ob_concurrent_seq_queue.cpp
   ob_easy_hazard_map.h
   ob_log_adapt_string.h ob_log_adapt_string.cpp
@@ -47,7 +41,6 @@ set(SRC_LIST
   ob_log_row_data_index.h ob_log_row_data_index.cpp
   ob_log_store_service.h
   ob_log_store_service_stat.h ob_log_store_service_stat.cpp
-  ob_log_mock_store_service.h
   ob_log_storager.h ob_log_storager.cpp
   ob_log_reader_plug_in.h ob_log_reader_plug_in.cpp
   ob_log_data_processor.h ob_log_data_processor.cpp
@@ -112,53 +105,53 @@ set(SRC_LIST
   ob_log_part_callback.h
   ob_log_ref_state.h ob_log_ref_state.cpp
   ob_log_committer.h ob_log_committer.cpp
+  ob_log_rocksdb_store_service.h ob_log_rocksdb_store_service.cpp
   )
-add_library(oblog_objects OBJECT ${SRC_LIST})
-target_link_libraries(oblog_objects PUBLIC oblog_base)
-target_compile_definitions(oblog_objects PRIVATE ENABLE_DEBUG_LOG)
-target_compile_options(oblog_objects PRIVATE -Werror)
+add_library(obcdc_objects OBJECT ${SRC_LIST})
+target_link_libraries(obcdc_objects PUBLIC obcdc_base)
+target_compile_definitions(obcdc_objects PRIVATE ENABLE_DEBUG_LOG)
+target_compile_options(obcdc_objects PRIVATE -Werror)
 
-disable_pch(oblog_objects)
+disable_pch(obcdc_objects)
 set(LGPL_DEPS "-L${DEP_DIR}/lib/mariadb -l:libmariadbclient.a -laio")
 
 if (OB_SO_CACHE)
-  add_library(oblog SHARED IMPORTED GLOBAL)
-  set_target_properties(oblog PROPERTIES
-    IMPORTED_LOCATION "${CMAKE_CURRENT_BINARY_DIR}/liboblog.so"
-    IMPORTED_LINK_INTERFACE_LIBRARIES oblog_miner)
+  add_library(obcdc SHARED IMPORTED GLOBAL)
+  set_target_properties(obcdc PROPERTIES
+    IMPORTED_LOCATION "${CMAKE_CURRENT_BINARY_DIR}/libobcdc.so"
+    IMPORTED_LINK_INTERFACE_LIBRARIES obcdc_base)
 else()
-  add_library(oblog SHARED ${CMAKE_BINARY_DIR}/src/observer/ob_version.cpp)
-  target_link_libraries(oblog
-    PUBLIC oblog_base oblib
+  add_library(obcdc SHARED ${CMAKE_BINARY_DIR}/src/observer/ob_version.cpp)
+  target_link_libraries(obcdc
+    PUBLIC obcdc_base
     PRIVATE
     -Wl,--whole-archive
-    $<TARGET_OBJECTS:oblog_objects>
+    $<TARGET_OBJECTS:obcdc_objects>
     -Wl,--no-whole-archive
     -Wl,--start-group
     oceanbase_static
     -Wl,--end-group
-    -static-libgcc -static-libstdc++
+    PRIVATE -static-libgcc -static-libstdc++
     -Wl,-Bsymbolic
     -Wl,-e,so_main
     ${LGPL_DEPS}
-    easy
     )
-  add_dependencies(oblog oblog_objects)
-  set_target_properties(oblog PROPERTIES SOVERSION 1 VERSION 1.0.0)
+  add_dependencies(obcdc obcdc_objects)
+  set_target_properties(obcdc PROPERTIES SOVERSION 1 VERSION 1.0.0)
 endif()
 
-add_library(oblog_objects_miner OBJECT ${SRC_LIST})
-disable_pch(oblog_objects_miner)
-target_link_libraries(oblog_objects_miner PUBLIC oblog_miner)
-add_library(oblog_static
+add_library(obcdc_objects_miner OBJECT ${SRC_LIST})
+disable_pch(obcdc_objects_miner)
+target_link_libraries(obcdc_objects_miner PUBLIC obcdc_base)
+add_library(obcdc_static
   STATIC
   ${CMAKE_BINARY_DIR}/src/observer/ob_version.cpp)
-target_link_libraries(oblog_static
-  PUBLIC oblog_objects_miner oblib
+target_link_libraries(obcdc_static
+  PUBLIC obcdc_objects_miner
   -Wl,--start-group
   oceanbase_static
   -Wl,--end-group
   PRIVATE -static-libgcc -static-libstdc++
   ${LGPL_DEPS}
   )
-set_target_properties(oblog_static PROPERTIES OUTPUT_NAME "oblog")
+set_target_properties(obcdc_static PROPERTIES OUTPUT_NAME "obcdc")

tools/obcdc/src/libobcdc.cpp

@@ -10,7 +10,7 @@
  * See the Mulan PubL v2 for more details.
  */
 
-#include "liboblog.h"
+#include "libobcdc.h"
 
 #include <locale.h>
 #include <curl/curl.h>

tools/obcdc/src/ob_log_binlog_record.cpp

@@ -205,7 +205,7 @@ int ObLogBR::init_data(const RecordType type,
     LOG_ERROR("ILogRecord has not been created");
     ret = OB_NOT_INIT;
   } else if (OB_UNLIKELY(EUNKNOWN == type)
-      || OB_UNLIKELY(! is_valid_cluster_id(cluster_id))
+      || OB_UNLIKELY(! is_valid_cluster_id_(cluster_id))
       || OB_UNLIKELY(OB_INVALID_TENANT_ID == tenant_id)
       || OB_UNLIKELY(commit_version <= 0)
       || OB_UNLIKELY(OB_INVALID_TIMESTAMP == ddl_schema_version)) {
@@ -333,6 +333,14 @@ int ObLogBR::init_binlog_record_second_(const RecordType type,
   return ret;
 }
 
+bool ObLogBR::is_valid_cluster_id_(const uint64_t cluster_id) const
+{
+  bool bool_ret = false;
+  bool_ret = is_valid_cluster_id(cluster_id)
+    || (MIN_DRC_CLUSTER_ID <= cluster_id && MAX_DRC_CLUSTER_ID >= cluster_id);
+  return bool_ret;
+}
+
 int ObLogBR::put_old(ILogRecord *br, const bool is_changed)
 {
   int ret = OB_SUCCESS;

tools/obcdc/src/ob_log_binlog_record.h

@@ -35,6 +35,8 @@ class ObLogBR : public ObLogResourceRecycleTask
 public:
   static const char *COLUMN_CHANGED_LABEL_PTR;
   static const char *COLUMN_UNCHANGED_LABEL_PTR;
+  static const uint64_t MIN_DRC_CLUSTER_ID = 4294901760; // 0xffff0000
+  static const uint64_t MAX_DRC_CLUSTER_ID = 4294967295; // 0xffffffff
 
 public:
   ObLogBR();
@@ -139,6 +141,10 @@ private:
       const common::ObString &unique_id,
       const int64_t commit_version,
       const common::ObString *major_version_str = NULL);
+  // ob defined cluster id:  0x00000001~0xfffeffff. [1, 4294901759]
+  // drc defined cluster id: 0xffff0000-0xffffffff, [4294901760~4294967295]
+  // liboblog valid cluster_id: [0x00000001~0xffffffff]
+  bool is_valid_cluster_id_(const uint64_t cluster_id) const;
 
 private:
   ILogRecord *data_;               ///< real BinlogRecord

tools/obcdc/src/ob_log_binlog_record_queue.h

@@ -68,4 +68,3 @@ private:
 } // namespace liboblog
 } // namespace oceanbase
 #endif /* OCEANBASE_LIBOBLOG_OB_LOG_BINLOG_RECORD_QUEUE_ */
-

tools/obcdc/src/ob_log_config.h

@@ -127,8 +127,11 @@ public:
       "partition trans task pre-alloc count");
   DEF_INT(part_trans_task_active_count_upper_bound, OB_CLUSTER_PARAMETER, "200000", "[1,]",
       "active partition trans task count upper bound");
-  DEF_INT(log_entry_task_active_count_upper_bound, OB_CLUSTER_PARAMETER, "200000", "[1,]",
-      "active log entry task count upper bound");
+  DEF_INT(storager_task_count_upper_bound, OB_CLUSTER_PARAMETER, "1000", "[1,]",
+      "storager task count upper bound");
+  DEF_INT(storager_mem_percentage, OB_CLUSTER_PARAMETER, "2", "[1,]",
+      "storager memory percentage");
+  T_DEF_BOOL(skip_recycle_data, OB_CLUSTER_PARAMETER, 0, "0:not_skip, 1:skip")
   DEF_INT(part_trans_task_reusable_count_upper_bound, OB_CLUSTER_PARAMETER, "10240", "[1,]",
       "reusable partition trans task count upper bound");
   DEF_INT(ready_to_seq_task_upper_bound, OB_CLUSTER_PARAMETER, "20000", "[1,]",
@@ -230,7 +233,8 @@ public:
   // liboblog support multiple working mode, default is storage
   // 1. storage: transaction data is stored, can support large transactions
   // 2. memory: transaction data is not stored, it means better performance, but may can not support large transactions
-  DEF_STR(working_mode, OB_CLUSTER_PARAMETER, "memory", "liboblog working mode");
+  DEF_STR(working_mode, OB_CLUSTER_PARAMETER, "storage", "liboblog working mode");
+  T_DEF_INT_INFT(rocksdb_write_buffer_size, OB_CLUSTER_PARAMETER, 64, 16, "write buffer size[M]");
 
   T_DEF_BOOL(fetcher_mock_mode, OB_CLUSTER_PARAMETER, 0, "0:disabled, 1:enabled");
   T_DEF_INT_INFT(io_thread_num, OB_CLUSTER_PARAMETER, 4, 1, "io thread number");
@@ -313,6 +317,8 @@ public:
 
   // Upper limit of progress difference between partitions, in seconds
   T_DEF_INT_INFT(progress_limit_sec_for_dml, OB_CLUSTER_PARAMETER, 3, 1, "dml progress limit in seconds");
+  T_DEF_INT_INFT(progress_limit_sec_for_ddl, OB_CLUSTER_PARAMETER, 3600, 1, "ddl progress limit in seconds");
+
 
   // Partition timeout in seconds
   // If the logs are not fetched after a certain period of time, the stream will be cut

tools/obcdc/src/ob_log_fetch_stream.cpp

@@ -38,6 +38,7 @@ int64_t FetchStream::g_stream_life_time = ObLogConfig::default_stream_life_time_
 int64_t FetchStream::g_rpc_timeout = ObLogConfig::default_fetch_log_rpc_timeout_sec * _SEC_;
 int64_t FetchStream::g_fetch_log_cnt_per_part_per_round = ObLogConfig::default_fetch_log_cnt_per_part_per_round;
 int64_t FetchStream::g_dml_progress_limit = ObLogConfig::default_progress_limit_sec_for_dml * _SEC_;
+int64_t FetchStream::g_ddl_progress_limit = ObLogConfig::default_progress_limit_sec_for_ddl * _SEC_;
 int64_t FetchStream::g_blacklist_survival_time = ObLogConfig::default_blacklist_survival_time_sec * _SEC_;
 int64_t FetchStream::g_check_switch_server_interval = ObLogConfig::default_check_switch_server_interval_min * _MIN_;
 bool FetchStream::g_print_rpc_handle_info = ObLogConfig::default_print_rpc_handle_info;
@@ -231,6 +232,7 @@ void FetchStream::configure(const ObLogConfig & config)
   int64_t fetch_log_rpc_timeout_sec = config.fetch_log_rpc_timeout_sec;
   int64_t fetch_log_cnt_per_part_per_round = config.fetch_log_cnt_per_part_per_round;
   int64_t dml_progress_limit_sec = config.progress_limit_sec_for_dml;
+  int64_t ddl_progress_limit_sec = config.progress_limit_sec_for_ddl;
   int64_t blacklist_survival_time_sec = config.blacklist_survival_time_sec;
   int64_t check_switch_server_interval_min = config.check_switch_server_interval_min;
   bool print_rpc_handle_info = config.print_rpc_handle_info;
@@ -246,6 +248,8 @@ void FetchStream::configure(const ObLogConfig & config)
   LOG_INFO("[CONFIG]", K(fetch_log_cnt_per_part_per_round));
   ATOMIC_STORE(&g_dml_progress_limit, dml_progress_limit_sec * _SEC_);
   LOG_INFO("[CONFIG]", K(dml_progress_limit_sec));
+  ATOMIC_STORE(&g_ddl_progress_limit, ddl_progress_limit_sec * _SEC_);
+  LOG_INFO("[CONFIG]", K(ddl_progress_limit_sec));
   ATOMIC_STORE(&g_blacklist_survival_time, blacklist_survival_time_sec * _SEC_);
   LOG_INFO("[CONFIG]", K(blacklist_survival_time_sec));
   ATOMIC_STORE(&g_check_switch_server_interval, check_switch_server_interval_min * _MIN_);
@@ -543,7 +547,7 @@ int FetchStream::get_upper_limit(int64_t &upper_limit_us)
   } else {
     // DDL partition is not limited by progress limit, here upper limit is set to a future value
     if (FETCH_STREAM_TYPE_DDL == stype_) {
-      upper_limit_us = min_progress + _YEAR_;
+      upper_limit_us = min_progress + ATOMIC_LOAD(&g_ddl_progress_limit);
     } else {
       // Other partition are limited by progress limit
       upper_limit_us = min_progress + ATOMIC_LOAD(&g_dml_progress_limit);

tools/obcdc/src/ob_log_fetch_stream.h

@@ -70,6 +70,7 @@ public:
   static int64_t g_rpc_timeout;
   static int64_t g_fetch_log_cnt_per_part_per_round;
   static int64_t g_dml_progress_limit;
+  static int64_t g_ddl_progress_limit;
   // Survival time of server added to blacklist
   static int64_t g_blacklist_survival_time;
   static int64_t g_check_switch_server_interval;

tools/obcdc/src/ob_log_fetcher_heartbeat_worker.cpp

@@ -519,4 +519,3 @@ void ObLogFetcherHeartbeatWorker::WorkerData::destroy()
 
 }
 }
-

tools/obcdc/src/ob_log_instance.cpp

@@ -50,7 +50,7 @@
 #include "ob_log_ddl_handler.h"           // ObLogDDLHandler
 #include "ob_log_start_schema_matcher.h"  // ObLogStartSchemaMatcher
 #include "ob_log_tenant_mgr.h"            // IObLogTenantMgr
-#include "ob_log_mock_store_service.h"    // MockObLogStoreService
+#include "ob_log_rocksdb_store_service.h" // RocksDbStoreService
 
 #include "ob_log_trace_id.h"
 
@@ -398,12 +398,12 @@ int ObLogInstance::init_logger_()
       }
     }
 
-    _LOG_INFO("====================liboblog start====================");
-    _LOG_INFO("liboblog CE %s %s", PACKAGE_VERSION, RELEASEID);
+    _LOG_INFO("====================libobcdc start====================");
+    _LOG_INFO("libobcdc CE %s %s", PACKAGE_VERSION, RELEASEID);
     _LOG_INFO("BUILD_VERSION: %s", build_version());
     _LOG_INFO("BUILD_TIME: %s %s", build_date(), build_time());
     _LOG_INFO("BUILD_FLAGS: %s", build_flags());
-    _LOG_INFO("Copyright (c) 2021 Ant Group Co., Ltd.");
+    _LOG_INFO("Copyright (c) 2022 Ant Group Co., Ltd.");
     _LOG_INFO("======================================================");
     _LOG_INFO("\n");
   }
@@ -415,11 +415,11 @@ int ObLogInstance::init_logger_()
 
 void ObLogInstance::print_version()
 {
-  MPRINT("liboblog CE %s %s", PACKAGE_VERSION, RELEASEID);
+  MPRINT("libobcdc CE %s %s", PACKAGE_VERSION, RELEASEID);
   MPRINT("REVISION: %s", build_version());
   MPRINT("BUILD_TIME: %s %s", build_date(), build_time());
   MPRINT("BUILD_FLAGS: %s\n", build_flags());
-  MPRINT("Copyright (c) 2021 Ant Group Co., Ltd.");
+  MPRINT("Copyright (c) 2022 Ant Group Co., Ltd.");
   MPRINT();
 }
 
@@ -790,7 +790,7 @@ int ObLogInstance::init_components_(const uint64_t start_tstamp_usec)
 
   INIT(log_entry_task_pool_, ObLogEntryTaskPool, TCONF.log_entry_task_prealloc_count);
 
-  INIT(store_service_, MockObLogStoreService, store_service_path);
+  INIT(store_service_, RocksDbStoreService, store_service_path);
 
   INIT(br_pool_, ObLogBRPool, TCONF.binlog_record_prealloc_count);
 
@@ -1038,7 +1038,7 @@ void ObLogInstance::destroy_components_()
   DESTROY(br_pool_, ObLogBRPool);
   DESTROY(storager_, ObLogStorager);
   DESTROY(data_processor_, ObLogDataProcessor);
-  DESTROY(store_service_, MockObLogStoreService);
+  DESTROY(store_service_, RocksDbStoreService);
 
   LOG_INFO("destroy all components end");
 }
@@ -2015,8 +2015,10 @@ void ObLogInstance::global_flow_control_()
         TCONF.part_trans_task_reusable_count_upper_bound;
       int64_t ready_to_seq_task_upper_bound =
         TCONF.ready_to_seq_task_upper_bound;
-      int64_t log_entry_task_active_count_upper_bound =
-        TCONF.log_entry_task_active_count_upper_bound;
+      int64_t storager_task_count_upper_bound =
+        TCONF.storager_task_count_upper_bound;
+      int64_t storager_mem_percentage =
+        TCONF.storager_mem_percentage;
       double system_memory_avail_percentage_lower_bound =
         static_cast<double>(TCONF.system_memory_avail_percentage_lower_bound) / 100;
       int64_t memory_limit = TCONF.memory_limit.get();
@@ -2052,6 +2054,9 @@ void ObLogInstance::global_flow_control_()
       } else if (OB_FAIL(dml_parser_->get_log_entry_task_count(dml_parser_part_trans_task_count))) {
         LOG_ERROR("DML parser get_log_entry_task_count fail", KR(ret), K(dml_parser_part_trans_task_count));
       } else {
+        int64_t storager_task_count = 0;
+        storager_->get_task_count(storager_task_count);
+
         // Use the following policy for global traffic control:
         // need_slow_down = (active partitioned transaction tasks exceed the upper limit || liboblog takes up more memory than the upper limit || system free memory is less than a certain percentage)
         // && (reusable transaction tasks exceeds limit || Parser and Sequencer module cache tasks exceeds limit)
@@ -2060,7 +2065,7 @@ void ObLogInstance::global_flow_control_()
           || (system_memory_avail < system_memory_avail_lower_bound);
         bool condition2 = (reusable_part_trans_task_count >= part_trans_task_reusable_count_upper_bound)
           || (ready_to_seq_task_count > ready_to_seq_task_upper_bound);
-        bool condition3 = (active_log_entry_task_count >= log_entry_task_active_count_upper_bound);
+        bool condition3 = (storager_task_count > storager_task_count_upper_bound) && (memory_hold >= storager_mem_percentage * memory_limit);
 
         need_slow_down_fetcher = (condition1 && condition2) || (condition3);
 
@@ -2081,7 +2086,8 @@ void ObLogInstance::global_flow_control_()
               "AVAIL_MEM=%s/%s "
               "READY_TO_SEQ=%ld/%ld "
               "PART_TRANS(TOTAL=%ld,ACTIVE=%ld/%ld,REUSABLE=%ld/%ld) "
-              "LOG_TASK(ACTIVE=%ld/%ld) "
+              "LOG_TASK(ACTIVE=%ld) "
+              "STORE(%ld/%ld) "
               "[FETCHER=%ld DML_PARSER=%ld "
               "COMMITER=%ld USER_QUEUE=%ld OUT=%ld RC=%ld] "
               "DIST_TRANS(SEQ=%ld,COMMITTED=%ld)",
@@ -2092,7 +2098,8 @@ void ObLogInstance::global_flow_control_()
               total_part_trans_task_count,
               active_part_trans_task_count, part_trans_task_active_count_upper_bound,
               reusable_part_trans_task_count, part_trans_task_reusable_count_upper_bound,
-              active_log_entry_task_count, log_entry_task_active_count_upper_bound,
+              active_log_entry_task_count,
+              storager_task_count, storager_task_count_upper_bound,
               fetcher_part_trans_task_count, dml_parser_part_trans_task_count,
               committer_ddl_part_trans_task_count + committer_dml_part_trans_task_count,
               br_queue_part_trans_task_count, out_part_trans_task_count,
@@ -2546,7 +2553,7 @@ int ObLogInstance::check_observer_version_valid_()
   uint32_t ob_major = 0;
   uint16_t ob_minor = 0;
   uint16_t ob_patch = 0;
-  cal_version_components_(ob_version, ob_major, ob_minor, ob_patch);
+  cal_version_components(ob_version, ob_major, ob_minor, ob_patch);
   if (0 != TCONF.skip_ob_version_compat_check) {
     _LOG_INFO("skip_ob_version_compat_check is true, skip check, observer_version(%u.%hu.%hu)",
         ob_major, ob_minor, ob_patch);
@@ -2597,7 +2604,7 @@ int ObLogInstance::init_oblog_version_components_()
   if (OB_FAIL(ObClusterVersion::get_version(PACKAGE_VERSION, oblog_version))) {
     LOG_ERROR("get_version fail", KR(ret), K(PACKAGE_VERSION), K(oblog_version));
   } else {
-    cal_version_components_(oblog_version, oblog_major_, oblog_minor_, oblog_patch_);
+    cal_version_components(oblog_version, oblog_major_, oblog_minor_, oblog_patch_);
   }
 
   if (OB_SUCC(ret)) {
@@ -2607,16 +2614,6 @@ int ObLogInstance::init_oblog_version_components_()
   return ret;
 }
 
-void ObLogInstance::cal_version_components_(const uint64_t version,
-    uint32_t &major,
-    uint16_t &minor,
-    uint16_t &patch)
-{
-  major = OB_VSN_MAJOR(version);
-  minor = OB_VSN_MINOR(version);
-  patch = OB_VSN_PATCH(version);
-}
-
 int ObLogInstance::get_tenant_guard(const uint64_t tenant_id, ObLogTenantGuard &guard)
 {
   int ret = OB_SUCCESS;

tools/obcdc/src/ob_log_instance.h

@@ -13,7 +13,7 @@
 #ifndef OCEANBASE_LIBOBLOG_INSTANCE_H__
 #define OCEANBASE_LIBOBLOG_INSTANCE_H__
 
-#include "liboblog.h"
+#include "libobcdc.h"
 
 #include "lib/allocator/ob_concurrent_fifo_allocator.h"   // ObConcurrentFIFOAllocator
 #include "lib/alloc/memory_dump.h"                        // memory_meta_dump
@@ -248,7 +248,6 @@ private:
   int check_observer_version_valid_();
   int init_ob_cluster_version_();
   int init_oblog_version_components_();
-  void cal_version_components_(const uint64_t version, uint32_t &major, uint16_t &minor, uint16_t &patch);
   int query_cluster_min_observer_version_(uint64_t &min_observer_version);
   // Initialize global variables for compatibility with GCTX dependencies
   void init_global_context_();

tools/obcdc/src/ob_log_main.c

@@ -25,10 +25,10 @@ int so_main()
 {
   fprintf(stdout, "\n");
 
-  fprintf(stdout, "liboblog (%s %s)\n",   PACKAGE_STRING, RELEASEID);
+  fprintf(stdout, "libobcdc (%s %s)\n",   PACKAGE_STRING, RELEASEID);
   fprintf(stdout, "\n");
 
-  fprintf(stdout, "BUILD_VERSION: %s\n",    build_version());
+  fprintf(stdout, "BUILD_VERSION: %s\n",  build_version());
   fprintf(stdout, "BUILD_TIME: %s %s\n",  build_date(), build_time());
   fprintf(stdout, "BUILD_FLAGS: %s\n",    build_flags());
   exit(0);

tools/obcdc/src/ob_log_part_mgr.cpp

@@ -3449,7 +3449,7 @@ int ObLogPartMgr::add_partition_(const ObPartitionKey& pkey,
   return ret;
 }
 
-// ensure_recycled_when_offlined: Whether recycling is guaranteed to be successful in the event of going offline, 
+// ensure_recycled_when_offlined: Whether recycling is guaranteed to be successful in the event of going offline,
 // i.e. whether the number of transactions left behind is 0
 int ObLogPartMgr::offline_partition_(const ObPartitionKey &pkey,
     const bool ensure_recycled_when_offlined)

tools/obcdc/src/ob_log_part_trans_parser.cpp

@@ -349,16 +349,16 @@ int ObLogPartTransParser::handle_ddl_part_rollback_savepoint_(const uint64_t row
 {
   int ret = OB_SUCCESS;
   const int32_t flag = row.flag_;
-  const int32_t sql_no = row.sql_no_;
+  const int32_t rollback_to_seq = get_row_seq_(task, row);
 
   if (OB_UNLIKELY(! task.is_ddl_trans())) {
     LOG_ERROR("task is not ddl trans, unexcepted", K(row_index), K(task), K(row));
     ret = OB_ERR_UNEXPECTED;
-  } else if (OB_UNLIKELY(1 != flag) || OB_UNLIKELY(sql_no < 0) || OB_UNLIKELY(OB_INVALID_ID == row_index)) {
-    LOG_ERROR("invalid argument", K(flag), K(sql_no), K(row_index), K(row));
+  } else if (OB_UNLIKELY(1 != flag) || OB_UNLIKELY(rollback_to_seq < 0) || OB_UNLIKELY(OB_INVALID_ID == row_index)) {
+    LOG_ERROR("invalid argument", K(flag), K(rollback_to_seq), K(row_index), K(row));
     ret = OB_INVALID_ARGUMENT;
-  } else if (OB_FAIL(task.revert_by_rollback_savepoint(row_index, sql_no))) {
-    LOG_ERROR("task revert_stmt_by_rollback_savepoint failed", KR(ret), K(sql_no), K(task));
+  } else if (OB_FAIL(task.revert_by_rollback_savepoint(row_index, rollback_to_seq))) {
+    LOG_ERROR("task revert_stmt_by_rollback_savepoint failed", KR(ret), K(rollback_to_seq), K(task));
   } else {
     // succ
     LOG_DEBUG("handle rollback savepoint succ", K(row), K(task));
@@ -609,7 +609,7 @@ int ObLogPartTransParser::parse_dml_stmts_(const uint64_t row_index,
   const char *participant_key_str = task.get_participant_key_str();
   const uint64_t log_id = redo_log_entry_task.get_log_id();
   const int32_t log_offset = redo_log_entry_task.get_log_offset();
-  const int32_t row_sql_no = row.sql_no_;
+  const int32_t row_seq_no = get_row_seq_(task, row);
 
   if (OB_ISNULL(stmt_task) || OB_ISNULL(row_data_index)) {
     LOG_ERROR("allocate memory for DmlStmtTask or ObLogRowDataIndex fail", "Dmlsize", sizeof(DmlStmtTask),
@@ -618,9 +618,9 @@ int ObLogPartTransParser::parse_dml_stmts_(const uint64_t row_index,
   } else {
     new (row_data_index) ObLogRowDataIndex();
 
-    if (OB_FAIL(row_data_index->init(tenant_id, participant_key_str, log_id, log_offset, row_index, is_rollback, row_sql_no))) {
+    if (OB_FAIL(row_data_index->init(tenant_id, participant_key_str, log_id, log_offset, row_index, is_rollback, row_seq_no))) {
       LOG_ERROR("row_data_index init fail", KR(ret), K(tenant_id), K(row_data_index),
-          K(participant_key_str), K(log_id), K(log_offset), K(row_index), K(is_rollback), K(row_sql_no),
+          K(participant_key_str), K(log_id), K(log_offset), K(row_index), K(is_rollback), K(row_seq_no),
           K(task), K(redo_log_entry_task));
     } else {
       new (stmt_task) DmlStmtTask(task, redo_log_entry_task, *row_data_index, row);
@@ -654,5 +654,13 @@ int ObLogPartTransParser::parse_dml_stmts_(const uint64_t row_index,
 
   return ret;
 }
+
+int64_t ObLogPartTransParser::get_row_seq_(PartTransTask &task, MutatorRow &row) const
+{
+  // only has sql_no
+  UNUSED(task);
+  return row.sql_no_;
+}
+
 }
 }

tools/obcdc/src/ob_log_part_trans_parser.h

@@ -98,6 +98,7 @@ private:
       ObLogEntryTask &redo_log_entry_task,
       PartTransTask &part_trans_task,
       const bool is_rollback = false);
+  int64_t get_row_seq_(PartTransTask &task, MutatorRow &row) const;
 
 private:
   bool              inited_;