diff --git a/CMakeLists.txt b/CMakeLists.txt
index 50f3f2c87..d5b8b2c39 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -4,7 +4,7 @@ include(cmake/Utils.cmake)
 include(cmake/Env.cmake)
 
 project("OceanBase CE"
-  VERSION 3.1.4
+  VERSION 3.1.5
   DESCRIPTION "OceanBase distributed database system"
   HOMEPAGE_URL "https://open.oceanbase.com/"
   LANGUAGES CXX C ASM)
diff --git a/cmake/RPM.cmake b/cmake/RPM.cmake
index b536153cc..06e91a1d3 100644
--- a/cmake/RPM.cmake
+++ b/cmake/RPM.cmake
@@ -17,10 +17,10 @@ list(APPEND CPACK_RPM_EXCLUDE_FROM_AUTO_FILELIST_ADDITION "/home/admin/oceanbase
 set(CPACK_PACKAGE_NAME "oceanbase-ce")
 set(CPACK_PACKAGE_DESCRIPTION_SUMMARY "OceanBase CE is a distributed relational database")
 set(CPACK_PACKAGE_VENDOR "Ant Group CO., Ltd.")
-set(CPACK_PACKAGE_VERSION 3.1.4)
+set(CPACK_PACKAGE_VERSION 3.1.5)
 set(CPACK_PACKAGE_VERSION_MAJOR 3)
 set(CPACK_PACKAGE_VERSION_MINOR 1)
-set(CPACK_PACKAGE_VERSION_PATCH 4)
+set(CPACK_PACKAGE_VERSION_PATCH 5)
 set(CPACK_RPM_PACKAGE_GROUP "Applications/Databases")
 set(CPACK_RPM_PACKAGE_URL "https://open.oceanbase.com")
 set(CPACK_RPM_PACKAGE_DESCRIPTION "OceanBase CE is a distributed relational database")
diff --git a/src/share/ob_cluster_version.h b/src/share/ob_cluster_version.h
index 8e86f2469..af4891013 100644
--- a/src/share/ob_cluster_version.h
+++ b/src/share/ob_cluster_version.h
@@ -96,11 +96,12 @@ uint64_t cal_version(const uint64_t major, const uint64_t minor, const uint64_t
 #define CLUSTER_VERSION_312 (oceanbase::common::cal_version(3, 1, 2))
 #define CLUSTER_VERSION_313 (oceanbase::common::cal_version(3, 1, 3))
 #define CLUSTER_VERSION_314 (oceanbase::common::cal_version(3, 1, 4))
+#define CLUSTER_VERSION_315 (oceanbase::common::cal_version(3, 1, 5))
 #define CLUSTER_VERSION_MAX UINT64_MAX
 // FIXME If you update the above version, please update me, CLUSTER_CURRENT_VERSION & ObUpgradeChecker!!!!!!
 
 //!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
-#define CLUSTER_CURRENT_VERSION CLUSTER_VERSION_314
+#define CLUSTER_CURRENT_VERSION CLUSTER_VERSION_315
 #define GET_MIN_CLUSTER_VERSION() (oceanbase::common::ObClusterVersion::get_instance().get_cluster_version())
 #define GET_UNIS_CLUSTER_VERSION() (::oceanbase::lib::get_unis_compat_version() ?: GET_MIN_CLUSTER_VERSION())
 
diff --git a/src/share/ob_upgrade_utils.cpp b/src/share/ob_upgrade_utils.cpp
index 9cac0d79e..69f4d9dae 100644
--- a/src/share/ob_upgrade_utils.cpp
+++ b/src/share/ob_upgrade_utils.cpp
@@ -33,7 +33,8 @@ const uint64_t ObUpgradeChecker::UPGRADE_PATH[CLUTER_VERSION_NUM] = {
     CALC_CLUSTER_VERSION(3UL, 1UL, 1UL),    //3.1.1
     CALC_CLUSTER_VERSION(3UL, 1UL, 2UL),   //3.1.2
     CALC_CLUSTER_VERSION(3UL, 1UL, 3UL),   //3.1.3
-    CALC_CLUSTER_VERSION(3UL, 1UL, 4UL)   //3.1.4
+    CALC_CLUSTER_VERSION(3UL, 1UL, 4UL),   //3.1.4
+  CALC_CLUSTER_VERSION(3UL, 1UL, 5UL)   //3.1.5
 };
 
 bool ObUpgradeChecker::check_cluster_version_exist(const uint64_t version)
@@ -1092,6 +1093,7 @@ int ObUpgradeProcesserSet::init(ObBaseUpgradeProcessor::UpgradeMode mode, common
     INIT_PROCESSOR_BY_VERSION(3, 1, 2);
     INIT_PROCESSOR_BY_VERSION(3, 1, 3);
     INIT_PROCESSOR_BY_VERSION(3, 1, 4);
+    INIT_PROCESSOR_BY_VERSION(3, 1, 5);
 #undef INIT_PROCESSOR_BY_VERSION
     inited_ = true;
   }
diff --git a/src/share/ob_upgrade_utils.h b/src/share/ob_upgrade_utils.h
index 468719d12..0ff8c1161 100644
--- a/src/share/ob_upgrade_utils.h
+++ b/src/share/ob_upgrade_utils.h
@@ -181,7 +181,7 @@ public:
   static bool check_cluster_version_exist(const uint64_t version);
 
 public:
-  static const int64_t CLUTER_VERSION_NUM = 4;
+  static const int64_t CLUTER_VERSION_NUM = 5;
   static const uint64_t UPGRADE_PATH[CLUTER_VERSION_NUM];
 };
 
@@ -234,6 +234,7 @@ DEF_SIMPLE_UPGRARD_PROCESSER(3, 1, 1);
 DEF_SIMPLE_UPGRARD_PROCESSER(3, 1, 2);
 DEF_SIMPLE_UPGRARD_PROCESSER(3, 1, 3);
 DEF_SIMPLE_UPGRARD_PROCESSER(3, 1, 4);
+DEF_SIMPLE_UPGRARD_PROCESSER(3, 1, 5);
 
 /* =========== upgrade processor end ============= */
 
diff --git a/src/share/parameter/ob_parameter_seed.ipp b/src/share/parameter/ob_parameter_seed.ipp
index 7af0cebc0..c3ff9c2e7 100644
--- a/src/share/parameter/ob_parameter_seed.ipp
+++ b/src/share/parameter/ob_parameter_seed.ipp
@@ -528,7 +528,7 @@ DEF_TIME(get_leader_candidate_rpc_timeout, OB_CLUSTER_PARAMETER, "9s", "[2s, 180
     "the time during a get leader candidate rpc request "
     "is permitted to execute before it is terminated. Range: [2s, 180s]",
     ObParameterAttr(Section::ROOT_SERVICE, Source::DEFAULT, EditLevel::DYNAMIC_EFFECTIVE));
-DEF_STR(min_observer_version, OB_CLUSTER_PARAMETER, "3.1.4", "the min observer version",
+DEF_STR(min_observer_version, OB_CLUSTER_PARAMETER, "3.1.5", "the min observer version",
     ObParameterAttr(Section::ROOT_SERVICE, Source::DEFAULT, EditLevel::DYNAMIC_EFFECTIVE));
 DEF_BOOL(enable_ddl, OB_CLUSTER_PARAMETER, "True",
     "specifies whether DDL operation is turned on. "
diff --git a/src/sql/engine/join/ob_hash_join_op.cpp b/src/sql/engine/join/ob_hash_join_op.cpp
index 7d498f003..924d19988 100644
--- a/src/sql/engine/join/ob_hash_join_op.cpp
+++ b/src/sql/engine/join/ob_hash_join_op.cpp
@@ -2727,10 +2727,13 @@ int ObHashJoinOp::get_next_probe_partition()
       ret = OB_ITER_END;
     } else if (!part_histograms_[cur_full_right_partition_].empty()) {
       if (right_splitter_.is_valid()) {
-        HashJoinHistogram::HistPrefixArray* prefix_hist_count = right_splitter_.part_histogram_.prefix_hist_count2_;
-        if (OB_ISNULL(right_splitter_.part_histogram_.h2_) || OB_ISNULL(prefix_hist_count)) {
+        HashJoinHistogram::HistPrefixArray *prefix_hist_count = right_splitter_.part_histogram_.prefix_hist_count2_;
+        if (0 == right_splitter_.get_total_row_count()) {
+          ret = OB_ITER_END;
+          LOG_DEBUG("hj_part_array_ has no row in memory", K(ret));
+        } else if (OB_ISNULL(right_splitter_.part_histogram_.h2_) || OB_ISNULL(prefix_hist_count)) {
           ret = OB_ERR_UNEXPECTED;
-          LOG_WARN("h2 is null", K(ret));
+          LOG_WARN("h2 is null", K(ret), K(level1_part_count_), K(level2_part_count_), K(part_count));
         } else {
           if (cur_full_right_partition_ >= prefix_hist_count->count()) {
             ret = OB_ERR_UNEXPECTED;
@@ -2770,7 +2773,7 @@ int ObHashJoinOp::get_next_probe_partition()
         }
       }
     }
-  } while (cur_full_right_partition_ < part_count);
+  } while (OB_SUCC(ret) && cur_full_right_partition_ < part_count);
   return ret;
 }
 
diff --git a/tools/upgrade/each_tenant_ddl_actions_post.py b/tools/upgrade/each_tenant_ddl_actions_post.py
index 25f1f2b01..5d31a63c6 100755
--- a/tools/upgrade/each_tenant_ddl_actions_post.py
+++ b/tools/upgrade/each_tenant_ddl_actions_post.py
@@ -111,123 +111,6 @@ class EachTenantDDLActionPostCreateAllTenantBackupBackupLogArchiveStatus(BaseEac
 #这两行之间的这些action，如果不写在这两行之间的话会导致清空不掉相应的action。
 
 ####========******####======== actions begin ========####******========####
-class EachTenantDDLActionPostCreateAllKvTTLTasks(BaseEachTenantDDLAction):
-  @staticmethod
-  def get_seq_num():
-    return 0
-  def dump_before_do_action(self):
-    my_utils.query_and_dump_results(self._query_cursor, """select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task'""".format(self.get_all_table_name()))
-  def check_before_do_action(self):
-    (desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task'""".format(self.get_all_table_name()))
-    if len(results) > 0:
-      raise MyError('__all_kv_ttl_task already created')
-  def dump_before_do_each_tenant_action(self, tenant_id):
-    my_utils.query_and_dump_results(self._query_cursor, """select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
-  def skip_pre_check(self):
-    return True
-  def skip_each_tenant_action(self, tenant_id):
-    (desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
-    return (1 == len(results))
-  def check_before_do_each_tenant_action(self, tenant_id):
-    (desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
-    if len(results) > 0:
-      raise MyError('tenant_id:{0} has already create table __all_kv_ttl_task'.format(tenant_id))
-  @staticmethod
-  def get_each_tenant_action_ddl(tenant_id):
-    pure_table_id = 410
-    table_id = (tenant_id << 40) | pure_table_id
-    return """CREATE TABLE `__all_kv_ttl_task` (
-              `gmt_create` timestamp(6) NULL DEFAULT CURRENT_TIMESTAMP(6),
-              `gmt_modified` timestamp(6) NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6),
-              `tenant_id` bigint(20) NOT NULL,
-              `task_id` bigint(20) NOT NULL,
-              `table_id` bigint(20) NOT NULL,
-              `partition_id` bigint(20) NOT NULL,
-              `task_start_time` bigint(20) NOT NULL,
-              `task_update_time` bigint(20) NOT NULL,
-              `trigger_type` bigint(20) NOT NULL,
-              `status` bigint(20) NOT NULL,
-              `ttl_del_cnt` bigint(20) NOT NULL,
-              `max_version_del_cnt` bigint(20) NOT NULL,
-              `scan_cnt` bigint(20) NOT NULL,
-              `row_key` varbinary(2048) NOT NULL,
-              `ret_code` varchar(512) NOT NULL,
-              PRIMARY KEY (`tenant_id`, `task_id`, `table_id`, `partition_id`)
-            ) TABLE_ID={0} DEFAULT CHARSET = utf8mb4 ROW_FORMAT = COMPACT COMPRESSION = 'none' REPLICA_NUM = 1 BLOCK_SIZE = 16384 USE_BLOOM_FILTER = FALSE TABLET_SIZE = 134217728 PCTFREE = 10 TABLEGROUP = 'oceanbase' """.format(table_id)
-  @staticmethod
-  def get_each_tenant_rollback_sql(tenant_id):
-    return """select 1"""
-  def dump_after_do_each_tenant_action(self, tenant_id):
-    my_utils.query_and_dump_results(self._query_cursor, """select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
-  def check_after_do_each_tenant_action(self, tenant_id):
-    (desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
-    if len(results) != 1:
-      raise MyError('tenant_id:{0} create table __all_kv_ttl_task failed'.format(tenant_id))
-  def dump_after_do_action(self):
-    my_utils.query_and_dump_results(self._query_cursor, """select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task'""".format(self.get_all_table_name()))
-  def check_after_do_action(self):
-    (desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task'""".format(self.get_all_table_name()))
-    if len(results) != len(self.get_tenant_id_list()):
-      raise MyError('there should be {0} rows in {1} whose table_name is __all_kv_ttl_task, but there has {2} rows like that'.format(len(self.get_tenant_id_list()), self.get_all_table_name(), len(results)))  
-
-class EachTenantDDLActionPostCreateAllKvTTLTaskHistory(BaseEachTenantDDLAction):
-  @staticmethod
-  def get_seq_num():
-    return 1
-  def dump_before_do_action(self):
-    my_utils.query_and_dump_results(self._query_cursor, """select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task_history'""".format(self.get_all_table_name()))
-  def skip_pre_check(self):
-    return True
-  def skip_each_tenant_action(self, tenant_id):
-    (desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task_history' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
-    return (1 == len(results))
-  def check_before_do_action(self):
-    (desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task_history'""".format(self.get_all_table_name()))
-    if len(results) > 0:
-      raise MyError('__all_kv_ttl_task_history already created')
-  def dump_before_do_each_tenant_action(self, tenant_id):
-    my_utils.query_and_dump_results(self._query_cursor, """select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task_history' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
-  def check_before_do_each_tenant_action(self, tenant_id):
-    (desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task_history' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
-    if len(results) > 0:
-      raise MyError('tenant_id:{0} has already create table __all_kv_ttl_task_history'.format(tenant_id))
-  @staticmethod
-  def get_each_tenant_action_ddl(tenant_id):
-    pure_table_id = 411
-    table_id = (tenant_id << 40) | pure_table_id
-    return """CREATE TABLE `__all_kv_ttl_task_history` (
-              `gmt_create` timestamp(6) NULL DEFAULT CURRENT_TIMESTAMP(6),
-              `gmt_modified` timestamp(6) NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6),
-              `tenant_id` bigint(20) NOT NULL,
-              `task_id` bigint(20) NOT NULL,
-              `table_id` bigint(20) NOT NULL,
-              `partition_id` bigint(20) NOT NULL,
-              `task_start_time` bigint(20) NOT NULL,
-              `task_update_time` bigint(20) NOT NULL,
-              `trigger_type` bigint(20) NOT NULL,
-              `status` bigint(20) NOT NULL,
-              `ttl_del_cnt` bigint(20) NOT NULL,
-              `max_version_del_cnt` bigint(20) NOT NULL,
-              `scan_cnt` bigint(20) NOT NULL,
-              `row_key` varbinary(2048) NOT NULL,
-              `ret_code` varchar(512) NOT NULL,
-              PRIMARY KEY (`tenant_id`, `task_id`, `table_id`, `partition_id`)
-            ) TABLE_ID={0} DEFAULT CHARSET = utf8mb4 ROW_FORMAT = COMPACT COMPRESSION = 'none' REPLICA_NUM = 1 BLOCK_SIZE = 16384 USE_BLOOM_FILTER = FALSE TABLET_SIZE = 134217728 PCTFREE = 10 TABLEGROUP = 'oceanbase' """.format(table_id)
-  @staticmethod
-  def get_each_tenant_rollback_sql(tenant_id):
-    return """select 1"""
-  def dump_after_do_each_tenant_action(self, tenant_id):
-    my_utils.query_and_dump_results(self._query_cursor, """select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task_history' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
-  def check_after_do_each_tenant_action(self, tenant_id):
-    (desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task_history' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
-    if len(results) != 1:
-      raise MyError('tenant_id:{0} create table __all_kv_ttl_task_history failed'.format(tenant_id))
-  def dump_after_do_action(self):
-    my_utils.query_and_dump_results(self._query_cursor, """select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task_history'""".format(self.get_all_table_name()))
-  def check_after_do_action(self):
-    (desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task_history'""".format(self.get_all_table_name()))
-    if len(results) != len(self.get_tenant_id_list()):
-      raise MyError('there should be {0} rows in {1} whose table_name is __all_kv_ttl_task_history, but there has {2} rows like that'.format(len(self.get_tenant_id_list()), self.get_all_table_name(), len(results)))    
 ####========******####========= actions end =========####******========####
 
 def do_each_tenant_ddl_actions(cur, tenant_id_list):
diff --git a/tools/upgrade/oceanbase_upgrade_dep.yml b/tools/upgrade/oceanbase_upgrade_dep.yml
index b036e3dd8..add448cc8 100644
--- a/tools/upgrade/oceanbase_upgrade_dep.yml
+++ b/tools/upgrade/oceanbase_upgrade_dep.yml
@@ -29,3 +29,6 @@
 - version: 3.1.4
   can_be_upgraded_to:
       - 3.1.5
+- version: 3.1.5
+  can_be_upgraded_to:
+      - 3.1.6
diff --git a/tools/upgrade/special_upgrade_action_post.py b/tools/upgrade/special_upgrade_action_post.py
index 9cbe1dcb9..4ea93daec 100755
--- a/tools/upgrade/special_upgrade_action_post.py
+++ b/tools/upgrade/special_upgrade_action_post.py
@@ -17,7 +17,7 @@ def do_special_upgrade(conn, cur, tenant_id_list, user, pwd):
 #因为基准版本更新的时候会调用reset_upgrade_scripts.py来清空actions begin和actions end
 #这两行之间的这些代码，如果不写在这两行之间的话会导致清空不掉相应的代码。
 ####========******####======== actions begin ========####******========####
-  run_upgrade_job(conn, cur, "3.1.4")
+  run_upgrade_job(conn, cur, "3.1.5")
   return
 ####========******####========= actions end =========####******========####
 
diff --git a/tools/upgrade/upgrade_post.py b/tools/upgrade/upgrade_post.py
index d3b2d95fc..e6c7867f2 100755
--- a/tools/upgrade/upgrade_post.py
+++ b/tools/upgrade/upgrade_post.py
@@ -1115,123 +1115,6 @@
 ##这两行之间的这些action，如果不写在这两行之间的话会导致清空不掉相应的action。
 #
 #####========******####======== actions begin ========####******========####
-#class EachTenantDDLActionPostCreateAllKvTTLTasks(BaseEachTenantDDLAction):
-#  @staticmethod
-#  def get_seq_num():
-#    return 0
-#  def dump_before_do_action(self):
-#    my_utils.query_and_dump_results(self._query_cursor, """select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task'""".format(self.get_all_table_name()))
-#  def check_before_do_action(self):
-#    (desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task'""".format(self.get_all_table_name()))
-#    if len(results) > 0:
-#      raise MyError('__all_kv_ttl_task already created')
-#  def dump_before_do_each_tenant_action(self, tenant_id):
-#    my_utils.query_and_dump_results(self._query_cursor, """select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
-#  def skip_pre_check(self):
-#    return True
-#  def skip_each_tenant_action(self, tenant_id):
-#    (desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
-#    return (1 == len(results))
-#  def check_before_do_each_tenant_action(self, tenant_id):
-#    (desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
-#    if len(results) > 0:
-#      raise MyError('tenant_id:{0} has already create table __all_kv_ttl_task'.format(tenant_id))
-#  @staticmethod
-#  def get_each_tenant_action_ddl(tenant_id):
-#    pure_table_id = 410
-#    table_id = (tenant_id << 40) | pure_table_id
-#    return """CREATE TABLE `__all_kv_ttl_task` (
-#              `gmt_create` timestamp(6) NULL DEFAULT CURRENT_TIMESTAMP(6),
-#              `gmt_modified` timestamp(6) NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6),
-#              `tenant_id` bigint(20) NOT NULL,
-#              `task_id` bigint(20) NOT NULL,
-#              `table_id` bigint(20) NOT NULL,
-#              `partition_id` bigint(20) NOT NULL,
-#              `task_start_time` bigint(20) NOT NULL,
-#              `task_update_time` bigint(20) NOT NULL,
-#              `trigger_type` bigint(20) NOT NULL,
-#              `status` bigint(20) NOT NULL,
-#              `ttl_del_cnt` bigint(20) NOT NULL,
-#              `max_version_del_cnt` bigint(20) NOT NULL,
-#              `scan_cnt` bigint(20) NOT NULL,
-#              `row_key` varbinary(2048) NOT NULL,
-#              `ret_code` varchar(512) NOT NULL,
-#              PRIMARY KEY (`tenant_id`, `task_id`, `table_id`, `partition_id`)
-#            ) TABLE_ID={0} DEFAULT CHARSET = utf8mb4 ROW_FORMAT = COMPACT COMPRESSION = 'none' REPLICA_NUM = 1 BLOCK_SIZE = 16384 USE_BLOOM_FILTER = FALSE TABLET_SIZE = 134217728 PCTFREE = 10 TABLEGROUP = 'oceanbase' """.format(table_id)
-#  @staticmethod
-#  def get_each_tenant_rollback_sql(tenant_id):
-#    return """select 1"""
-#  def dump_after_do_each_tenant_action(self, tenant_id):
-#    my_utils.query_and_dump_results(self._query_cursor, """select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
-#  def check_after_do_each_tenant_action(self, tenant_id):
-#    (desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
-#    if len(results) != 1:
-#      raise MyError('tenant_id:{0} create table __all_kv_ttl_task failed'.format(tenant_id))
-#  def dump_after_do_action(self):
-#    my_utils.query_and_dump_results(self._query_cursor, """select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task'""".format(self.get_all_table_name()))
-#  def check_after_do_action(self):
-#    (desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task'""".format(self.get_all_table_name()))
-#    if len(results) != len(self.get_tenant_id_list()):
-#      raise MyError('there should be {0} rows in {1} whose table_name is __all_kv_ttl_task, but there has {2} rows like that'.format(len(self.get_tenant_id_list()), self.get_all_table_name(), len(results)))  
-#
-#class EachTenantDDLActionPostCreateAllKvTTLTaskHistory(BaseEachTenantDDLAction):
-#  @staticmethod
-#  def get_seq_num():
-#    return 1
-#  def dump_before_do_action(self):
-#    my_utils.query_and_dump_results(self._query_cursor, """select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task_history'""".format(self.get_all_table_name()))
-#  def skip_pre_check(self):
-#    return True
-#  def skip_each_tenant_action(self, tenant_id):
-#    (desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task_history' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
-#    return (1 == len(results))
-#  def check_before_do_action(self):
-#    (desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task_history'""".format(self.get_all_table_name()))
-#    if len(results) > 0:
-#      raise MyError('__all_kv_ttl_task_history already created')
-#  def dump_before_do_each_tenant_action(self, tenant_id):
-#    my_utils.query_and_dump_results(self._query_cursor, """select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task_history' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
-#  def check_before_do_each_tenant_action(self, tenant_id):
-#    (desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task_history' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
-#    if len(results) > 0:
-#      raise MyError('tenant_id:{0} has already create table __all_kv_ttl_task_history'.format(tenant_id))
-#  @staticmethod
-#  def get_each_tenant_action_ddl(tenant_id):
-#    pure_table_id = 411
-#    table_id = (tenant_id << 40) | pure_table_id
-#    return """CREATE TABLE `__all_kv_ttl_task_history` (
-#              `gmt_create` timestamp(6) NULL DEFAULT CURRENT_TIMESTAMP(6),
-#              `gmt_modified` timestamp(6) NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6),
-#              `tenant_id` bigint(20) NOT NULL,
-#              `task_id` bigint(20) NOT NULL,
-#              `table_id` bigint(20) NOT NULL,
-#              `partition_id` bigint(20) NOT NULL,
-#              `task_start_time` bigint(20) NOT NULL,
-#              `task_update_time` bigint(20) NOT NULL,
-#              `trigger_type` bigint(20) NOT NULL,
-#              `status` bigint(20) NOT NULL,
-#              `ttl_del_cnt` bigint(20) NOT NULL,
-#              `max_version_del_cnt` bigint(20) NOT NULL,
-#              `scan_cnt` bigint(20) NOT NULL,
-#              `row_key` varbinary(2048) NOT NULL,
-#              `ret_code` varchar(512) NOT NULL,
-#              PRIMARY KEY (`tenant_id`, `task_id`, `table_id`, `partition_id`)
-#            ) TABLE_ID={0} DEFAULT CHARSET = utf8mb4 ROW_FORMAT = COMPACT COMPRESSION = 'none' REPLICA_NUM = 1 BLOCK_SIZE = 16384 USE_BLOOM_FILTER = FALSE TABLET_SIZE = 134217728 PCTFREE = 10 TABLEGROUP = 'oceanbase' """.format(table_id)
-#  @staticmethod
-#  def get_each_tenant_rollback_sql(tenant_id):
-#    return """select 1"""
-#  def dump_after_do_each_tenant_action(self, tenant_id):
-#    my_utils.query_and_dump_results(self._query_cursor, """select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task_history' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
-#  def check_after_do_each_tenant_action(self, tenant_id):
-#    (desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task_history' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
-#    if len(results) != 1:
-#      raise MyError('tenant_id:{0} create table __all_kv_ttl_task_history failed'.format(tenant_id))
-#  def dump_after_do_action(self):
-#    my_utils.query_and_dump_results(self._query_cursor, """select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task_history'""".format(self.get_all_table_name()))
-#  def check_after_do_action(self):
-#    (desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task_history'""".format(self.get_all_table_name()))
-#    if len(results) != len(self.get_tenant_id_list()):
-#      raise MyError('there should be {0} rows in {1} whose table_name is __all_kv_ttl_task_history, but there has {2} rows like that'.format(len(self.get_tenant_id_list()), self.get_all_table_name(), len(results)))    
 #####========******####========= actions end =========####******========####
 #
 #def do_each_tenant_ddl_actions(cur, tenant_id_list):
@@ -6396,7 +6279,7 @@
 ##因为基准版本更新的时候会调用reset_upgrade_scripts.py来清空actions begin和actions end
 ##这两行之间的这些代码，如果不写在这两行之间的话会导致清空不掉相应的代码。
 #####========******####======== actions begin ========####******========####
-#  run_upgrade_job(conn, cur, "3.1.4")
+#  run_upgrade_job(conn, cur, "3.1.5")
 #  return
 #####========******####========= actions end =========####******========####
 #
@@ -8693,7 +8576,7 @@
 #
 #class UpgradeParams:
 #  log_filename = 'upgrade_post_checker.log'
-#  new_version = '3.1.4'
+#  new_version = '3.1.5'
 ##### --------------start : my_error.py --------------
 #class MyError(Exception):
 #  def __init__(self, value):
diff --git a/tools/upgrade/upgrade_post_checker.py b/tools/upgrade/upgrade_post_checker.py
index 8ab675674..4aa25ae87 100755
--- a/tools/upgrade/upgrade_post_checker.py
+++ b/tools/upgrade/upgrade_post_checker.py
@@ -12,7 +12,7 @@ import time
 
 class UpgradeParams:
   log_filename = 'upgrade_post_checker.log'
-  new_version = '3.1.4'
+  new_version = '3.1.5'
 #### --------------start : my_error.py --------------
 class MyError(Exception):
   def __init__(self, value):
diff --git a/tools/upgrade/upgrade_pre.py b/tools/upgrade/upgrade_pre.py
index b9073703e..153e871c7 100755
--- a/tools/upgrade/upgrade_pre.py
+++ b/tools/upgrade/upgrade_pre.py
@@ -1115,123 +1115,6 @@
 ##这两行之间的这些action，如果不写在这两行之间的话会导致清空不掉相应的action。
 #
 #####========******####======== actions begin ========####******========####
-#class EachTenantDDLActionPostCreateAllKvTTLTasks(BaseEachTenantDDLAction):
-#  @staticmethod
-#  def get_seq_num():
-#    return 0
-#  def dump_before_do_action(self):
-#    my_utils.query_and_dump_results(self._query_cursor, """select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task'""".format(self.get_all_table_name()))
-#  def check_before_do_action(self):
-#    (desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task'""".format(self.get_all_table_name()))
-#    if len(results) > 0:
-#      raise MyError('__all_kv_ttl_task already created')
-#  def dump_before_do_each_tenant_action(self, tenant_id):
-#    my_utils.query_and_dump_results(self._query_cursor, """select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
-#  def skip_pre_check(self):
-#    return True
-#  def skip_each_tenant_action(self, tenant_id):
-#    (desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
-#    return (1 == len(results))
-#  def check_before_do_each_tenant_action(self, tenant_id):
-#    (desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
-#    if len(results) > 0:
-#      raise MyError('tenant_id:{0} has already create table __all_kv_ttl_task'.format(tenant_id))
-#  @staticmethod
-#  def get_each_tenant_action_ddl(tenant_id):
-#    pure_table_id = 410
-#    table_id = (tenant_id << 40) | pure_table_id
-#    return """CREATE TABLE `__all_kv_ttl_task` (
-#              `gmt_create` timestamp(6) NULL DEFAULT CURRENT_TIMESTAMP(6),
-#              `gmt_modified` timestamp(6) NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6),
-#              `tenant_id` bigint(20) NOT NULL,
-#              `task_id` bigint(20) NOT NULL,
-#              `table_id` bigint(20) NOT NULL,
-#              `partition_id` bigint(20) NOT NULL,
-#              `task_start_time` bigint(20) NOT NULL,
-#              `task_update_time` bigint(20) NOT NULL,
-#              `trigger_type` bigint(20) NOT NULL,
-#              `status` bigint(20) NOT NULL,
-#              `ttl_del_cnt` bigint(20) NOT NULL,
-#              `max_version_del_cnt` bigint(20) NOT NULL,
-#              `scan_cnt` bigint(20) NOT NULL,
-#              `row_key` varbinary(2048) NOT NULL,
-#              `ret_code` varchar(512) NOT NULL,
-#              PRIMARY KEY (`tenant_id`, `task_id`, `table_id`, `partition_id`)
-#            ) TABLE_ID={0} DEFAULT CHARSET = utf8mb4 ROW_FORMAT = COMPACT COMPRESSION = 'none' REPLICA_NUM = 1 BLOCK_SIZE = 16384 USE_BLOOM_FILTER = FALSE TABLET_SIZE = 134217728 PCTFREE = 10 TABLEGROUP = 'oceanbase' """.format(table_id)
-#  @staticmethod
-#  def get_each_tenant_rollback_sql(tenant_id):
-#    return """select 1"""
-#  def dump_after_do_each_tenant_action(self, tenant_id):
-#    my_utils.query_and_dump_results(self._query_cursor, """select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
-#  def check_after_do_each_tenant_action(self, tenant_id):
-#    (desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
-#    if len(results) != 1:
-#      raise MyError('tenant_id:{0} create table __all_kv_ttl_task failed'.format(tenant_id))
-#  def dump_after_do_action(self):
-#    my_utils.query_and_dump_results(self._query_cursor, """select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task'""".format(self.get_all_table_name()))
-#  def check_after_do_action(self):
-#    (desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task'""".format(self.get_all_table_name()))
-#    if len(results) != len(self.get_tenant_id_list()):
-#      raise MyError('there should be {0} rows in {1} whose table_name is __all_kv_ttl_task, but there has {2} rows like that'.format(len(self.get_tenant_id_list()), self.get_all_table_name(), len(results)))  
-#
-#class EachTenantDDLActionPostCreateAllKvTTLTaskHistory(BaseEachTenantDDLAction):
-#  @staticmethod
-#  def get_seq_num():
-#    return 1
-#  def dump_before_do_action(self):
-#    my_utils.query_and_dump_results(self._query_cursor, """select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task_history'""".format(self.get_all_table_name()))
-#  def skip_pre_check(self):
-#    return True
-#  def skip_each_tenant_action(self, tenant_id):
-#    (desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task_history' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
-#    return (1 == len(results))
-#  def check_before_do_action(self):
-#    (desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task_history'""".format(self.get_all_table_name()))
-#    if len(results) > 0:
-#      raise MyError('__all_kv_ttl_task_history already created')
-#  def dump_before_do_each_tenant_action(self, tenant_id):
-#    my_utils.query_and_dump_results(self._query_cursor, """select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task_history' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
-#  def check_before_do_each_tenant_action(self, tenant_id):
-#    (desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task_history' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
-#    if len(results) > 0:
-#      raise MyError('tenant_id:{0} has already create table __all_kv_ttl_task_history'.format(tenant_id))
-#  @staticmethod
-#  def get_each_tenant_action_ddl(tenant_id):
-#    pure_table_id = 411
-#    table_id = (tenant_id << 40) | pure_table_id
-#    return """CREATE TABLE `__all_kv_ttl_task_history` (
-#              `gmt_create` timestamp(6) NULL DEFAULT CURRENT_TIMESTAMP(6),
-#              `gmt_modified` timestamp(6) NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6),
-#              `tenant_id` bigint(20) NOT NULL,
-#              `task_id` bigint(20) NOT NULL,
-#              `table_id` bigint(20) NOT NULL,
-#              `partition_id` bigint(20) NOT NULL,
-#              `task_start_time` bigint(20) NOT NULL,
-#              `task_update_time` bigint(20) NOT NULL,
-#              `trigger_type` bigint(20) NOT NULL,
-#              `status` bigint(20) NOT NULL,
-#              `ttl_del_cnt` bigint(20) NOT NULL,
-#              `max_version_del_cnt` bigint(20) NOT NULL,
-#              `scan_cnt` bigint(20) NOT NULL,
-#              `row_key` varbinary(2048) NOT NULL,
-#              `ret_code` varchar(512) NOT NULL,
-#              PRIMARY KEY (`tenant_id`, `task_id`, `table_id`, `partition_id`)
-#            ) TABLE_ID={0} DEFAULT CHARSET = utf8mb4 ROW_FORMAT = COMPACT COMPRESSION = 'none' REPLICA_NUM = 1 BLOCK_SIZE = 16384 USE_BLOOM_FILTER = FALSE TABLET_SIZE = 134217728 PCTFREE = 10 TABLEGROUP = 'oceanbase' """.format(table_id)
-#  @staticmethod
-#  def get_each_tenant_rollback_sql(tenant_id):
-#    return """select 1"""
-#  def dump_after_do_each_tenant_action(self, tenant_id):
-#    my_utils.query_and_dump_results(self._query_cursor, """select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task_history' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
-#  def check_after_do_each_tenant_action(self, tenant_id):
-#    (desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task_history' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
-#    if len(results) != 1:
-#      raise MyError('tenant_id:{0} create table __all_kv_ttl_task_history failed'.format(tenant_id))
-#  def dump_after_do_action(self):
-#    my_utils.query_and_dump_results(self._query_cursor, """select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task_history'""".format(self.get_all_table_name()))
-#  def check_after_do_action(self):
-#    (desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task_history'""".format(self.get_all_table_name()))
-#    if len(results) != len(self.get_tenant_id_list()):
-#      raise MyError('there should be {0} rows in {1} whose table_name is __all_kv_ttl_task_history, but there has {2} rows like that'.format(len(self.get_tenant_id_list()), self.get_all_table_name(), len(results)))    
 #####========******####========= actions end =========####******========####
 #
 #def do_each_tenant_ddl_actions(cur, tenant_id_list):
@@ -6396,7 +6279,7 @@
 ##因为基准版本更新的时候会调用reset_upgrade_scripts.py来清空actions begin和actions end
 ##这两行之间的这些代码，如果不写在这两行之间的话会导致清空不掉相应的代码。
 #####========******####======== actions begin ========####******========####
-#  run_upgrade_job(conn, cur, "3.1.4")
+#  run_upgrade_job(conn, cur, "3.1.5")
 #  return
 #####========******####========= actions end =========####******========####
 #
@@ -8693,7 +8576,7 @@
 #
 #class UpgradeParams:
 #  log_filename = 'upgrade_post_checker.log'
-#  new_version = '3.1.4'
+#  new_version = '3.1.5'
 ##### --------------start : my_error.py --------------
 #class MyError(Exception):
 #  def __init__(self, value):