diff --git a/storage/ctc/ctc_mysql_proxy.cc b/storage/ctc/ctc_mysql_proxy.cc
index a668b1b1ffbaaa115cbad9da529ee661e7a91f63..32cff2e36e90f3592b331bebbd2228da479a8a2d 100644
--- a/storage/ctc/ctc_mysql_proxy.cc
+++ b/storage/ctc/ctc_mysql_proxy.cc
@@ -485,6 +485,11 @@ __attribute__((visibility("default"))) int ctc_set_cluster_role_by_cantian(bool
   return 0;
 }
 
+__attribute__((visibility("default"))) int ctc_update_sql_statistic_stat(bool enable_stat) {
+  set_enable_wsr_stat(enable_stat);
+  return 0;
+}
+
 static int ctc_ddl_get_lock(MYSQL *curr_conn, const uint64_t &conn_map_key, const char *lock_name, int *err_code) {
   uchar digest[MD5_HASH_SIZE];
   compute_md5_hash(pointer_cast<char *>(digest), lock_name, strlen(lock_name));
diff --git a/storage/ctc/ctc_srv.h b/storage/ctc/ctc_srv.h
index 0c2be04425699a5d422660d3d638bbbb8f6c391f..f0bd75246b5cf3098a89eefab1354261d29e793a 100644
--- a/storage/ctc/ctc_srv.h
+++ b/storage/ctc/ctc_srv.h
@@ -20,7 +20,7 @@
 
 #include <stdint.h>
 #include <stdbool.h>
-
+#include "sql/tztime.h"
 #ifdef __cplusplus
 extern "C" {
 #endif
@@ -38,6 +38,8 @@ extern "C" {
 #define ERROR_MESSAGE_LEN 512
 #define MAX_DDL_SQL_LEN_CONTEXT (63488)  // 62kb, 预留2kb
 #define MAX_DDL_SQL_LEN (MAX_DDL_SQL_LEN_CONTEXT + 30)  // ddl sql语句的长度 不能超过64kb, 超过了会报错
+#define MAX_WSR_DML_SQL_LEN_CONTEXT (8192)
+#define MAX_WSR_DML_SQL_LEN (MAX_WSR_DML_SQL_LEN_CONTEXT + 30)  // 用于wsr的dml sql语句的长度 不能超过8000, 超过了会截断
 #define DD_BROADCAST_RECORD_LENGTH (3072)
 #define LOCK_TABLE_SQL_FMT_LEN 20
 #define MAX_LOCK_TABLE_NAME (MAX_DDL_SQL_LEN - LOCK_TABLE_SQL_FMT_LEN)
@@ -313,6 +315,8 @@ enum CTC_FUNC_TYPE {
     CTC_FUNC_TYPE_SCAN_RECORDS,
     CTC_FUNC_TYPE_TRX_COMMIT,
     CTC_FUNC_TYPE_TRX_ROLLBACK,
+    CTC_FUNC_TYPE_STATISTIC_BEGIN,
+    CTC_FUNC_TYPE_STATISTIC_COMMIT,
     CTC_FUNC_TYPE_TRX_BEGIN,
     CTC_FUNC_TYPE_LOCK_TABLE,
     CTC_FUNC_TYPE_UNLOCK_TABLE,
@@ -364,6 +368,7 @@ enum CTC_FUNC_TYPE {
     CTC_FUNC_TYPE_GET_INDEX_PARAL_SCHEDULE,
     CTC_FUNC_TYPE_PQ_INDEX_READ,
     CTC_FUNC_TYPE_PQ_SET_CURSOR_RANGE,
+    CTC_FUNC_QUERY_SQL_STATISTIC_STAT,
     /* for instance registration, should be the last but before duplex */
     CTC_FUNC_TYPE_REGISTER_INSTANCE,
     CTC_FUNC_QUERY_SHM_FILE_NUM,
@@ -378,6 +383,7 @@ enum CTC_FUNC_TYPE {
     CTC_FUNC_TYPE_INVALIDATE_OBJECTS,
     CTC_FUNC_TYPE_INVALIDATE_ALL_OBJECTS,
     CTC_FUNC_TYPE_UPDATE_DDCACHE,
+    CTC_FUNC_TYPE_UPDATE_SQL_STATISTIC_STAT,
     CTC_FUNC_TYPE_NUMBER,
 };
 
@@ -621,6 +627,7 @@ typedef int (*ctc_ddl_execute_lock_tables_t)(ctc_handler_t *tch, char *db_name,
 typedef int (*ctc_ddl_execute_unlock_tables_t)(ctc_handler_t *tch, uint32_t mysql_inst_id, ctc_lock_table_info *lock_info);
 typedef int (*ctc_invalidate_mysql_dd_cache_t)(ctc_handler_t *tch, ctc_invalidate_broadcast_request *broadcast_req, int *err_code);
 typedef int (*ctc_set_cluster_role_by_cantian_t)(bool is_slave);
+typedef int (*ctc_update_sql_statistic_stat_t)(bool enable_stat);
 
 typedef struct mysql_engine_intf_t {
     ctc_execute_rewrite_open_conn_t ctc_execute_rewrite_open_conn;
@@ -631,6 +638,7 @@ typedef struct mysql_engine_intf_t {
     ctc_ddl_execute_unlock_tables_t ctc_ddl_execute_unlock_tables;
     ctc_invalidate_mysql_dd_cache_t ctc_invalidate_mysql_dd_cache;
     ctc_set_cluster_role_by_cantian_t ctc_set_cluster_role_by_cantian;
+    ctc_update_sql_statistic_stat_t ctc_update_sql_statistic_stat;
 } sql_engine_intf;
 
 /* General Control Interface */
@@ -678,8 +686,12 @@ int ctc_general_prefetch(ctc_handler_t *tch, uint8_t *records, uint16_t *record_
 int ctc_free_session_cursors(ctc_handler_t *tch, uint64_t *cursors, int32_t csize);
 
 /* Transaction Related Interface */
-int ctc_trx_begin(ctc_handler_t *tch, ctc_trx_context_t trx_context, bool is_mysql_local);
-int ctc_trx_commit(ctc_handler_t *tch, uint64_t *cursors, int32_t csize, bool *is_ddl_commit);
+int ctc_trx_begin(ctc_handler_t *tch, ctc_trx_context_t trx_context, bool is_mysql_local,
+                  struct timeval begin_time, bool enable_stat);
+int ctc_statistic_begin(ctc_handler_t *tch, struct timeval begin_time, bool enable_stat);
+int ctc_trx_commit(ctc_handler_t *tch, uint64_t *cursors, int32_t csize,
+                   bool *is_ddl_commit, const char *sql_str, bool enable_stat);
+int ctc_statistic_commit(ctc_handler_t *tch, const char *sql_str, bool enable_stat);
 int ctc_trx_rollback(ctc_handler_t *tch, uint64_t *cursors, int32_t csize);
 
 int ctc_srv_set_savepoint(ctc_handler_t *tch, const char *name);
@@ -746,6 +758,8 @@ int ctc_broadcast_mysql_dd_invalidate(ctc_handler_t *tch, ctc_invalidate_broadca
 
 /* Disaster Recovery Related Interface*/
 int ctc_set_cluster_role_by_cantian(bool is_slave);
+
+int ctc_update_sql_statistic_stat(bool enable_stat);
 void ctc_set_mysql_read_only();
 void ctc_reset_mysql_read_only();
 
@@ -768,6 +782,8 @@ int ctc_query_cluster_role(bool *is_slave, bool *cantian_cluster_ready);
 int ctc_query_shm_file_num(uint32_t *shm_file_num);
 int ctc_query_shm_usage(uint32_t *shm_usage);
 
+int ctc_query_sql_statistic_stat(bool* enable_stat);
+
 #ifdef __cplusplus
 }
 #endif
diff --git a/storage/ctc/ctc_srv_mq_module.cc b/storage/ctc/ctc_srv_mq_module.cc
index 0fdbaaaddea2f5db08ad1154845d5663f4b06ea0..817dcd238c018a743c3e74adf72fbdf381edc96b 100644
--- a/storage/ctc/ctc_srv_mq_module.cc
+++ b/storage/ctc/ctc_srv_mq_module.cc
@@ -117,6 +117,12 @@ static void* mq_msg_handler(void *arg) {
       CTC_IGNORE_ERROR_WHEN_MYSQL_SHUTDOWN(req, "ctc_set_cluster_role_by_cantian");
       break;
     }
+    case CTC_FUNC_TYPE_UPDATE_SQL_STATISTIC_STAT: {
+      struct update_sql_statistic_stat *req = (struct update_sql_statistic_stat *)message_block->seg_buf[0];
+      req->result = ctc_update_sql_statistic_stat(req->enable_stat);
+      ctc_log_info("enable_stat has been set to %d", req->result);
+      break;
+    }
     default: {
       ctc_log_error("cmd type invalid, cmd_type:%d.", message_block->head.cmd_type);
       break;
diff --git a/storage/ctc/ctc_srv_mq_stub.cc b/storage/ctc/ctc_srv_mq_stub.cc
index c7e99525f27898b7d38645d197bc1edb1ef9cfeb..974bfddc0ec4981bae0f02fc5d492376ab06b961 100644
--- a/storage/ctc/ctc_srv_mq_stub.cc
+++ b/storage/ctc/ctc_srv_mq_stub.cc
@@ -526,9 +526,10 @@ int ctc_index_read(ctc_handler_t *tch, record_info_t *record_info, index_key_inf
   return result;
 }
 
-int ctc_trx_begin(ctc_handler_t *tch, ctc_trx_context_t trx_context, bool is_mysql_local) {
+int ctc_trx_begin(ctc_handler_t *tch, ctc_trx_context_t trx_context, bool is_mysql_local,
+                  struct timeval begin_time, bool enable_stat) {
   void *shm_inst = get_one_shm_inst(tch);
-  trx_begin_request *req = (trx_begin_request*)alloc_share_mem(shm_inst, sizeof(trx_begin_request));
+  trx_begin_request *req = (trx_begin_request *)alloc_share_mem(shm_inst, sizeof(trx_begin_request));
   if (req == NULL) {
     ctc_log_error("alloc shm mem error, shm_inst(%p), size(%lu)", shm_inst, sizeof(trx_begin_request));
     return ERR_ALLOC_MEMORY;
@@ -536,7 +537,8 @@ int ctc_trx_begin(ctc_handler_t *tch, ctc_trx_context_t trx_context, bool is_mys
   req->tch = *tch;
   req->trx_context = trx_context;
   req->is_mysql_local = is_mysql_local;
-
+  req->begin_time = begin_time;
+  req->enable_stat = enable_stat;
   int result = ERR_CONNECTION_FAILED;
   int ret = ctc_mq_deal_func(shm_inst, CTC_FUNC_TYPE_TRX_BEGIN, req, tch->msg_buf);
   *tch = req->tch; // 此处不管参天处理成功与否，都需要拷贝一次，避免session泄漏
@@ -547,17 +549,43 @@ int ctc_trx_begin(ctc_handler_t *tch, ctc_trx_context_t trx_context, bool is_mys
   return result;
 }
 
-int ctc_trx_commit(ctc_handler_t *tch, uint64_t *cursors, int32_t csize, bool *is_ddl_commit) {
+int ctc_statistic_begin(ctc_handler_t *tch, struct timeval begin_time, bool enable_stat) {
+  void *shm_inst = get_one_shm_inst(tch);
+  trx_begin_request *req = (trx_begin_request *)alloc_share_mem(shm_inst, sizeof(trx_begin_request));
+  if (req == NULL) {
+    ctc_log_error("alloc shm mem error, shm_inst(%p), size(%lu)", shm_inst, sizeof(trx_begin_request));
+    return ERR_ALLOC_MEMORY;
+  }
+  req->tch = *tch;
+  req->begin_time = begin_time;
+  req->enable_stat = enable_stat;
+  int result = ERR_CONNECTION_FAILED;
+  int ret = ctc_mq_deal_func(shm_inst, CTC_FUNC_TYPE_STATISTIC_BEGIN, req, tch->msg_buf);
+  *tch = req->tch; // 此处不管参天处理成功与否，都需要拷贝一次，避免session泄漏
+  if (ret == CT_SUCCESS) {
+    result = req->result;
+  }
+  free_share_mem(shm_inst, req);
+  return result;
+}
+
+int ctc_trx_commit(ctc_handler_t *tch, uint64_t *cursors, int32_t csize,
+                   bool *is_ddl_commit, const char *sql_str, bool enable_stat) {
   void *shm_inst = get_one_shm_inst(tch);
   trx_commit_request *req = (trx_commit_request*)alloc_share_mem(shm_inst, sizeof(trx_commit_request));
   if (req == NULL) {
     ctc_log_error("alloc shm mem error, shm_inst(%p), size(%lu)", shm_inst, sizeof(trx_commit_request));
     return ERR_ALLOC_MEMORY;
   }
+  uint64_t len = sizeof(trx_commit_request);
+  memset(req, 0, len);
   req->tch = *tch;
   req->csize = csize;
   req->cursors = cursors;
-
+  req->enable_stat = enable_stat;
+  if (sql_str != NULL) {
+    strncpy(req->sql, sql_str, MAX_WSR_DML_SQL_LEN - 1);
+  }
   int result = ERR_CONNECTION_FAILED;
   int ret = ctc_mq_deal_func(shm_inst, CTC_FUNC_TYPE_TRX_COMMIT, req, tch->msg_buf);
   *is_ddl_commit = req->is_ddl_commit;
@@ -568,6 +596,29 @@ int ctc_trx_commit(ctc_handler_t *tch, uint64_t *cursors, int32_t csize, bool *i
   return result;
 }
 
+int ctc_statistic_commit(ctc_handler_t *tch, const char *sql_str, bool enable_stat) {
+  void *shm_inst = get_one_shm_inst(tch);
+  trx_commit_request *req = (trx_commit_request*)alloc_share_mem(shm_inst, sizeof(trx_commit_request));
+  if (req == NULL) {
+    ctc_log_error("alloc shm mem error, shm_inst(%p), size(%lu)", shm_inst, sizeof(trx_commit_request));
+    return ERR_ALLOC_MEMORY;
+  }
+  uint64_t len = sizeof(trx_commit_request);
+  memset(req, 0, len);
+  req->tch = *tch;
+  if (sql_str != NULL) {
+    strncpy(req->sql, sql_str, MAX_WSR_DML_SQL_LEN - 1);
+  }
+  req->enable_stat = enable_stat;
+  int result = ERR_CONNECTION_FAILED;
+  int ret = ctc_mq_deal_func(shm_inst, CTC_FUNC_TYPE_STATISTIC_COMMIT, req, tch->msg_buf);
+  if (ret == CT_SUCCESS) {
+    result = req->result;
+  }
+  free_share_mem(shm_inst, req);
+  return result;
+}
+
 int ctc_trx_rollback(ctc_handler_t *tch, uint64_t *cursors, int32_t csize) {
   void *shm_inst = get_one_shm_inst(tch);
   trx_rollback_request *req = (trx_rollback_request*)alloc_share_mem(shm_inst, sizeof(trx_rollback_request));
@@ -1904,4 +1955,22 @@ int ctc_query_shm_usage(uint32_t *shm_usage)
   }
   free_share_mem(shm_inst, req);
   return result;
+}
+
+int ctc_query_sql_statistic_stat(bool* enable_stat) {
+  void *shm_inst = get_one_shm_inst(NULL);
+  query_sql_statistic_stat *req = (query_sql_statistic_stat*)
+                                     alloc_share_mem(shm_inst, sizeof(query_sql_statistic_stat));
+  if (req == nullptr) {
+    ctc_log_error("alloc shm mem error, shm_inst(%p), size(%lu)", shm_inst, sizeof(query_sql_statistic_stat));
+    return ERR_ALLOC_MEMORY;
+  }
+  int result = ERR_CONNECTION_FAILED;
+  int ret = ctc_mq_deal_func(shm_inst, CTC_FUNC_QUERY_SQL_STATISTIC_STAT, req, nullptr, SERVER_REGISTER_PROC_ID);
+  if (ret == CT_SUCCESS) {
+    result = req->result;
+    *enable_stat = req->enable_stat;
+  }
+  free_share_mem(shm_inst, req);
+  return result;
 }
\ No newline at end of file
diff --git a/storage/ctc/ctc_util.cc b/storage/ctc/ctc_util.cc
index afee965501b6a48a33dd2b6375c963569abdac13..c142d53f6cc040bce47bc66da5b863399e0e1755 100644
--- a/storage/ctc/ctc_util.cc
+++ b/storage/ctc/ctc_util.cc
@@ -35,6 +35,7 @@
 
 using namespace std;
 extern bool ctc_enable_x_lock_instance;
+bool g_enable_wsr_stat = false;
 
 string cnvrt_name_for_sql(string name) {
   string res = "";
@@ -1462,4 +1463,12 @@ int32_t ctc_cmp_cantian_rowid(const rowid_t *rowid1, const rowid_t *rowid2) {
 
   result = rowid1->slot > rowid2->slot ? 1 : (rowid1->slot < rowid2->slot ? (-1) : 0);
   return result;
+}
+
+void set_enable_wsr_stat(bool enable_stat) {
+  g_enable_wsr_stat = enable_stat;
+}
+
+bool get_enable_wsr_stat() {
+  return g_enable_wsr_stat;
 }
\ No newline at end of file
diff --git a/storage/ctc/ctc_util.h b/storage/ctc/ctc_util.h
index 4591285f8b4b2873e8cd1212fc3cc612636a4660..2fcc99128094971a2259458ae29ce33019ba1240 100644
--- a/storage/ctc/ctc_util.h
+++ b/storage/ctc/ctc_util.h
@@ -147,6 +147,9 @@ typedef struct en_ctc_cond_push_map {
 
 void cond_push_term(Item *term, Item *&pushed_cond, Item *&remainder_cond, Item_func::Functype parent_type);
 
+void set_enable_wsr_stat(bool enable_stat);
+bool get_enable_wsr_stat();
+
 #pragma GCC visibility pop
 
 #endif // __CTC_UTIL_H__
\ No newline at end of file
diff --git a/storage/ctc/ha_ctc.cc b/storage/ctc/ha_ctc.cc
index 472190783d03071bd86c975e8cf20a32feacaa22..994054ce147002c2ef39b12819d2e6c2619ebcef 100644
--- a/storage/ctc/ha_ctc.cc
+++ b/storage/ctc/ha_ctc.cc
@@ -77,7 +77,8 @@
 #include "ha_ctc_ddl.h"
 #include "ha_ctcpart.h"
 #include "ha_ctc_pq.h"
-
+#include "my_systime.h"
+#include "my_time.h"
 #include <errno.h>
 #include <limits.h>
 #include <sql/sql_thd_internal_api.h>
@@ -147,7 +148,6 @@
 #define CTC_MAX_SAMPLE_SIZE (4096)    // MB
 #define CTC_MIN_SAMPLE_SIZE (32)      // MB
 #define CTC_DEFAULT_SAMPLE_SIZE (128) // MB
-
 static void ctc_statistics_enabled_update(THD * thd, SYS_VAR *, void *var_ptr, const void *save) {
     bool enabled = *static_cast<bool *>(var_ptr) = *static_cast<const bool *>(save);
     ctc_stats::get_instance()->set_statistics_enabled(enabled);
@@ -1331,7 +1331,12 @@ static int ctc_start_trx_and_assign_scn(
   uint32_t lock_wait_timeout = THDVAR(thd, lock_wait_timeout);
   ctc_trx_context_t trx_context = {isolation_level, autocommit, lock_wait_timeout, false};
   bool is_mysql_local = (sess_ctx->set_flag & CTC_DDL_LOCAL_ENABLED);
-  ct_errno_t ret = (ct_errno_t)ctc_trx_begin(&tch, trx_context, is_mysql_local);
+  struct timeval begin_time = {INVALID_INT64, INVALID_INT64};
+  bool enable_stat = get_enable_wsr_stat();
+  if (enable_stat) {
+    gettimeofday(&begin_time, NULL);
+  }
+  ct_errno_t ret = (ct_errno_t)ctc_trx_begin(&tch, trx_context, is_mysql_local, begin_time, enable_stat);
   update_sess_ctx_by_tch(tch, hton, thd);
   if (ret != CT_SUCCESS) {
     ctc_log_error("start trx failed with error code: %d", ret);
@@ -1477,6 +1482,26 @@ static void ctc_free_cursors_no_autocommit(THD *thd, ctc_handler_t *tch, thd_ses
   ctc_free_buf(tch, (uint8_t *)cursors);
 }
 
+bool is_dml_cmd(const String& sql)
+{
+    constexpr int PREFIX_LENGTH = 6;
+    if (sql.length() < PREFIX_LENGTH) {
+        return false;
+    }
+    String prefix(sql.ptr(), PREFIX_LENGTH, sql.charset());
+    for (size_t i = 0; i < PREFIX_LENGTH; ++i) {
+        prefix[i] = tolower(prefix[i]);
+    }
+    constexpr char SELECT_PREFIX[] = "select";
+    constexpr char INSERT_PREFIX[] = "insert";
+    constexpr char DELETE_PREFIX[] = "delete";
+    constexpr char UPDATE_PREFIX[] = "update";
+    return strncmp(prefix.ptr(), SELECT_PREFIX, PREFIX_LENGTH) == 0 ||
+           strncmp(prefix.ptr(), INSERT_PREFIX, PREFIX_LENGTH) == 0 ||
+           strncmp(prefix.ptr(), DELETE_PREFIX, PREFIX_LENGTH) == 0 ||
+           strncmp(prefix.ptr(), UPDATE_PREFIX, PREFIX_LENGTH) == 0;
+}
+
 /**
   Commits a transaction in an ctc database or marks an SQL statement ended.
   @param: hton in, ctc handlerton
@@ -1502,7 +1527,13 @@ static int ctc_commit(handlerton *hton, THD *thd, bool commit_trx) {
   ct_errno_t ret = CT_SUCCESS;
   thd_sess_ctx_s *sess_ctx = (thd_sess_ctx_s *)thd_get_ha_data(thd, hton);
   assert(sess_ctx != nullptr);
-
+  bool is_dmlsql = false;
+  bool enable_stat = get_enable_wsr_stat();
+  if (thd->query().str != NULL && thd->query().length > 0) {
+    string dml_sql = string(thd->query().str).substr(0, thd->query().length);
+    String dml_sql_string(dml_sql.c_str(), dml_sql.length(), thd->charset());
+    is_dmlsql = is_dml_cmd(dml_sql_string);
+  }
   if (will_commit) {
     commit_preprocess(thd, &tch);
     attachable_trx_update_pre_addr(thd, &tch, true);
@@ -1517,7 +1548,11 @@ static int ctc_commit(handlerton *hton, THD *thd, bool commit_trx) {
     }
     assert((total_csize == 0) ^ (cursors != nullptr));
     ctc_copy_cursors_to_free(sess_ctx, cursors, 0);
-    ret = (ct_errno_t)ctc_trx_commit(&tch, cursors, total_csize, &is_ddl_commit);
+    const char *sql_str = nullptr;
+    if (is_dmlsql && enable_stat) {
+      sql_str = thd->query().str;
+    }
+    ret = (ct_errno_t)ctc_trx_commit(&tch, cursors, total_csize, &is_ddl_commit, sql_str, enable_stat);
     ctc_free_buf(&tch, (uint8_t *)cursors);
     if (ret != CT_SUCCESS) {
       ctc_log_error("commit atomic ddl failed with error code: %d", ret);
@@ -1542,6 +1577,15 @@ static int ctc_commit(handlerton *hton, THD *thd, bool commit_trx) {
     }
     sess_ctx->is_ctc_trx_begin = 0;
   } else {
+    const char* sql_str = thd->query().str;
+    if (is_dmlsql && enable_stat) {
+      ret = (ct_errno_t)ctc_statistic_commit(&tch, sql_str, enable_stat);
+      if (ret != CT_SUCCESS) {
+        ctc_log_error("commit statistic failed with error code: %d", ret);
+        END_RECORD_STATS(EVENT_TYPE_COMMIT)
+        return convert_ctc_error_code_to_mysql(ret);
+      }
+    }
     ctc_free_cursors_no_autocommit(thd, &tch, sess_ctx);
   }
 
@@ -4445,7 +4489,11 @@ int ha_ctc::external_lock(THD *thd, int lock_type) {
 */
 int ha_ctc::start_stmt(THD *thd, thr_lock_type) {
   DBUG_TRACE;
-
+  struct timeval begin_time = {INVALID_INT64, INVALID_INT64};
+  bool enable_stat = get_enable_wsr_stat();
+  if (enable_stat) {
+    gettimeofday(&begin_time, NULL);
+  }
   trans_register_ha(thd, false, ht, nullptr); // register trans to STMT
 
   update_member_tch(m_tch, ctc_hton, thd, false);
@@ -4468,6 +4516,11 @@ int ha_ctc::start_stmt(THD *thd, thr_lock_type) {
   if (sess_ctx->is_ctc_trx_begin) {
     assert(m_tch.sess_addr != INVALID_VALUE64);
     assert(m_tch.thd_id == thd->thread_id());
+    ct_errno_t ret = (ct_errno_t)ctc_statistic_begin(&m_tch, begin_time, enable_stat);
+    if (ret != CT_SUCCESS) {
+      ctc_log_error("start statistic failed with error code: %d", ret);
+      return convert_ctc_error_code_to_mysql(ret);
+    }
     return 0;
   }
 
@@ -4478,14 +4531,14 @@ int ha_ctc::start_stmt(THD *thd, thr_lock_type) {
   ctc_trx_context_t trx_context = {isolation_level, autocommit, lock_wait_timeout, m_select_lock == lock_mode::EXCLUSIVE_LOCK};
   
   bool is_mysql_local = (sess_ctx->set_flag & CTC_DDL_LOCAL_ENABLED);
-  ct_errno_t ret = (ct_errno_t)ctc_trx_begin(&m_tch, trx_context, is_mysql_local);
+  ct_errno_t ret = (ct_errno_t)ctc_trx_begin(&m_tch, trx_context, is_mysql_local, begin_time, enable_stat);
   
   check_error_code_to_mysql(ha_thd(), &ret);
 
   update_sess_ctx_by_tch(m_tch, ctc_hton, thd);
   
   if (ret != CT_SUCCESS) {
-    ctc_log_error("start trx failed with error code: %d", ret);
+      ctc_log_error("start trx failed with error code: %d", ret);
     return convert_ctc_error_code_to_mysql(ret);
   }
   
@@ -4970,6 +5023,23 @@ extern int (*ctc_deinit)();
 
 int ctc_push_to_engine(THD *thd, AccessPath *root_path, JOIN *);
 
+static int ctc_initialize_sql_statistic_stat() {
+  bool enable_stat = false;
+  int result = ctc_query_sql_statistic_stat(&enable_stat);
+  if (result != 0) {
+    return HA_ERR_INITIALIZATION;
+  }
+  set_enable_wsr_stat(enable_stat);
+  return 0;
+}
+
+static int ctc_initialize_global_variable_from_cantian_parameter() {
+  if (ctc_initialize_sql_statistic_stat() != 0) {
+    return HA_ERR_INITIALIZATION;
+  }
+  return 0;
+}
+
 static int ctc_init_func(void *p) {
   DBUG_TRACE;
   ctc_hton = (handlerton *)p;
@@ -5049,6 +5119,12 @@ static int ctc_init_func(void *p) {
 
   ctc_get_sample_size_value();
 
+  ret = ctc_initialize_global_variable_from_cantian_parameter();
+  if (ret != 0) {
+    ctc_log_error("[CTC_INIT]:ctc_initialize_global_variable_from_cantian_parameter failed:%d", ret);
+    return HA_ERR_INITIALIZATION;
+  }
+
   ctc_log_system("[CTC_INIT]:SUCCESS!");
   return 0;
 }
diff --git a/storage/ctc/ha_ctc.h b/storage/ctc/ha_ctc.h
index ec70e51734b948b738e6a18b6ac2370a3826cce8..bede2173b2a7acbfadd91779b5f6c6496fb17f73 100644
--- a/storage/ctc/ha_ctc.h
+++ b/storage/ctc/ha_ctc.h
@@ -19,7 +19,7 @@
 
 #ifndef __HA_CTC_H__
 #define __HA_CTC_H__
-
+#include "sql/tztime.h"
 #include <sys/types.h>
 #include <vector>
 #include "my_inttypes.h"
@@ -66,6 +66,7 @@ using namespace std;
 #define PREFER_RANGE_DENSITY 0.8
 #define CT_MAX_RECORD_LENGTH 64000
 #define INVALID_PART_ID (uint32)0xFFFFFFFF
+#define INVALID_INT64 (int64)0x7FFFFFFFFFFFFFFF
 
 /* update if restraints changed in Cantian */
 #define CTC_MAX_KEY_PART_LENGTH 4095  // CT_MAX_KEY_SIZE
diff --git a/storage/ctc/mysql_cantian_plugin.cc b/storage/ctc/mysql_cantian_plugin.cc
index ad3491b29f505414cb8e3de93aca8e13289b312c..ac2441030b65221b64389e13103dd55d7dc9d4ab 100644
--- a/storage/ctc/mysql_cantian_plugin.cc
+++ b/storage/ctc/mysql_cantian_plugin.cc
@@ -90,6 +90,7 @@ sql_engine_intf *get_local_sql_intf()
     g_local_sql_intf.ctc_ddl_execute_lock_tables = ctc_ddl_execute_lock_tables;
     g_local_sql_intf.ctc_ddl_execute_unlock_tables = ctc_ddl_execute_unlock_tables;
     g_local_sql_intf.ctc_set_cluster_role_by_cantian = ctc_set_cluster_role_by_cantian;
+    g_local_sql_intf.ctc_update_sql_statistic_stat = ctc_update_sql_statistic_stat;
     ctc_log_system("get local sql function success.");
     return &g_local_sql_intf;
 }
diff --git a/storage/ctc/srv_mq_msg.h b/storage/ctc/srv_mq_msg.h
index 60ad2787dc020ede1bcbc025a8cae59a28f5ce20..cb477dc00a9692f9efb19fe28b4204ba198d3279 100644
--- a/storage/ctc/srv_mq_msg.h
+++ b/storage/ctc/srv_mq_msg.h
@@ -168,6 +168,8 @@ struct trx_begin_request {
     int result;
     ctc_trx_context_t trx_context;
     bool is_mysql_local;
+    struct timeval begin_time;
+    bool enable_stat;
 };
 
 struct trx_commit_request {
@@ -176,6 +178,8 @@ struct trx_commit_request {
     bool is_ddl_commit;
     int32_t csize;
     uint64_t *cursors;
+    char sql[MAX_WSR_DML_SQL_LEN];
+    bool enable_stat;
 };
 
 struct trx_rollback_request {
@@ -468,6 +472,11 @@ struct set_cluster_role_by_cantian_request {
     int result;
 };
 
+struct update_sql_statistic_stat {
+    bool enable_stat;
+    int result;
+};
+
 struct execute_ddl_mysql_sql_request {
     ctc_ddl_broadcast_request broadcast_req;
     uint32_t thd_id;
@@ -516,6 +525,11 @@ struct invalidate_mysql_dd_request {
     int result;
 };
 
+struct query_sql_statistic_stat {
+    bool* enable_stat;
+    int result;
+};
+
 void* alloc_share_mem(void* shm_inst, uint32_t mem_size);
 
 void free_share_mem(void* shm_inst, void* shm_mem);