From 66c0ce02d9ea3d01f75712f3bafa1fa1494bea00 Mon Sep 17 00:00:00 2001
From: liaoxin <liaoxinbit@126.com>
Date: Thu, 13 Jul 2023 16:04:54 +0800
Subject: [PATCH] [fix](merge-on-write) fix be core and delete unused pending
 publish info for async publsih when tablet dropped

---
 be/src/olap/olap_server.cpp    |  6 +++++-
 be/src/olap/storage_engine.cpp | 27 +++++++++++++++++++++++++++
 be/src/olap/storage_engine.h   |  2 ++
 3 files changed, 34 insertions(+), 1 deletion(-)

diff --git a/be/src/olap/olap_server.cpp b/be/src/olap/olap_server.cpp
index 09875b9e5244fa..dabf07d17688d8 100644
--- a/be/src/olap/olap_server.cpp
+++ b/be/src/olap/olap_server.cpp
@@ -1216,6 +1216,11 @@ void StorageEngine::add_async_publish_task(int64_t partition_id, int64_t tablet_
                                            bool is_recovery) {
     if (!is_recovery) {
         TabletSharedPtr tablet = tablet_manager()->get_tablet(tablet_id);
+        if (tablet == nullptr) {
+            LOG(INFO) << "tablet may be dropped when add async publish task, tablet_id: "
+                      << tablet_id;
+            return;
+        }
         PendingPublishInfoPB pending_publish_info_pb;
         pending_publish_info_pb.set_partition_id(partition_id);
         pending_publish_info_pb.set_transaction_id(transaction_id);
@@ -1259,7 +1264,6 @@ void StorageEngine::_async_publish_callback() {
                 if (!tablet) {
                     LOG(WARNING) << "tablet does not exist when async publush, tablet_id: "
                                  << tablet_id;
-                    // TODO(liaoxin) remove pending publish info from db
                     tablet_iter = _async_publish_tasks.erase(tablet_iter);
                     continue;
                 }
diff --git a/be/src/olap/storage_engine.cpp b/be/src/olap/storage_engine.cpp
index f573b59f6027ad..2f111ae2266c81 100644
--- a/be/src/olap/storage_engine.cpp
+++ b/be/src/olap/storage_engine.cpp
@@ -695,6 +695,9 @@ Status StorageEngine::start_trash_sweep(double* usage, bool ignore_guard) {
     // cleand unused delete bitmap for deleted tablet
     _clean_unused_delete_bitmap();
 
+    // cleand unused pending publish info for deleted tablet
+    _clean_unused_pending_publish_info();
+
     // clean unused rowsets in remote storage backends
     for (auto data_dir : get_stores()) {
         data_dir->perform_remote_rowset_gc();
@@ -795,6 +798,30 @@ void StorageEngine::_clean_unused_delete_bitmap() {
     }
 }
 
+void StorageEngine::_clean_unused_pending_publish_info() {
+    std::vector<std::pair<int64_t, int64_t>> removed_infos;
+    auto clean_pending_publish_info_func = [this, &removed_infos](int64_t tablet_id,
+                                                                  int64_t publish_version,
+                                                                  const string& info) -> bool {
+        TabletSharedPtr tablet = _tablet_manager->get_tablet(tablet_id);
+        if (tablet == nullptr) {
+            removed_infos.emplace_back(tablet_id, publish_version);
+        }
+        return true;
+    };
+    auto data_dirs = get_stores();
+    for (auto data_dir : data_dirs) {
+        TabletMetaManager::traverse_delete_bitmap(data_dir->get_meta(),
+                                                  clean_pending_publish_info_func);
+        for (auto& [tablet_id, publish_version] : removed_infos) {
+            TabletMetaManager::remove_pending_publish_info(data_dir, tablet_id, publish_version);
+        }
+        LOG(INFO) << "removed invalid pending publish info from dir: " << data_dir->path()
+                  << ", deleted pending publish info size: " << removed_infos.size();
+        removed_infos.clear();
+    }
+}
+
 void StorageEngine::gc_binlogs(const std::unordered_map<int64_t, int64_t>& gc_tablet_infos) {
     for (auto [tablet_id, version] : gc_tablet_infos) {
         LOG(INFO) << fmt::format("start to gc binlogs for tablet_id: {}, version: {}", tablet_id,
diff --git a/be/src/olap/storage_engine.h b/be/src/olap/storage_engine.h
index 2bd01ba2cb29a2..a40e1792d90360 100644
--- a/be/src/olap/storage_engine.h
+++ b/be/src/olap/storage_engine.h
@@ -255,6 +255,8 @@ class StorageEngine {
 
     void _clean_unused_delete_bitmap();
 
+    void _clean_unused_pending_publish_info();
+
     Status _do_sweep(const std::string& scan_root, const time_t& local_tm_now,
                      const int32_t expire);