[Enhancement] support corrupt meta cache clear for bundle metadata (backport #63949) (#63988)

mergify[bot] · luohaha · web-flow · commit aa3d98a29bed · 2025-10-14T03:33:15.000Z
Signed-off-by: luohaha &lt;18810541851@163.com&gt;
Co-authored-by: Yixin Luo &lt;18810541851@163.com&gt;
diff --git a/be/src/storage/lake/tablet_manager.cpp b/be/src/storage/lake/tablet_manager.cpp
@@ -378,6 +378,22 @@ Status TabletManager::put_bundle_tablet_metadata(std::map<int64_t, TabletMetadat
     return Status::OK();
 }
 
+Status TabletManager::corrupted_tablet_meta_handler(const Status& s, const std::string& metadata_location) {
+    if (s.is_corruption() && config::lake_clear_corrupted_cache) {
+        auto drop_status = drop_local_cache(metadata_location);
+        TEST_SYNC_POINT_CALLBACK("TabletManager::corrupted_tablet_meta_handler", &drop_status);
+        if (!drop_status.ok()) {
+            LOG(WARNING) << "clear corrupted cache for " << metadata_location << " failed, "
+                         << "error: " << drop_status;
+            return s; // return error so load tablet meta can be retried
+        }
+        LOG(INFO) << "clear corrupted cache for " << metadata_location;
+        return Status::OK();
+    } else {
+        return s;
+    }
+}
+
 StatusOr<TabletMetadataPtr> TabletManager::load_tablet_metadata(const string& metadata_location, bool fill_cache,
                                                                 int64_t expected_gtid,
                                                                 const std::shared_ptr<FileSystem>& fs) {
@@ -387,21 +403,11 @@ StatusOr<TabletMetadataPtr> TabletManager::load_tablet_metadata(const string& me
     ProtobufFile file(metadata_location, fs);
     auto s = file.load(metadata.get(), fill_cache);
     if (!s.ok()) {
-        if (s.is_corruption() && config::lake_clear_corrupted_cache) {
-            auto drop_status = drop_local_cache(metadata_location);
-            if (!drop_status.ok()) {
-                LOG(WARNING) << "clear corrupted cache for " << metadata_location << " failed, "
-                             << "error: " << drop_status;
-                return s; // return error so load tablet meta can be retried
-            }
-            LOG(INFO) << "clear corrupted cache for " << metadata_location;
-            // reset metadata
-            metadata = std::make_shared<TabletMetadataPB>();
-            // read again
-            RETURN_IF_ERROR(file.load(metadata.get(), fill_cache));
-        } else {
-            return s;
-        }
+        RETURN_IF_ERROR(corrupted_tablet_meta_handler(s, metadata_location));
+        // reset metadata
+        metadata = std::make_shared<TabletMetadataPB>();
+        // read again
+        RETURN_IF_ERROR(file.load(metadata.get(), fill_cache));
     }
 
     if (expected_gtid > 0 && metadata->gtid() > 0 && expected_gtid != metadata->gtid()) {
@@ -502,7 +508,13 @@ StatusOr<BundleTabletMetadataPtr> TabletManager::parse_bundle_tablet_metadata(co
             std::string_view(serialized_string.data() + file_size - footer_size - bundle_metadata_size);
     RETURN_IF(!bundle_metadata->ParseFromArray(bundle_metadata_str.data(), bundle_metadata_size),
               Status::Corruption(strings::Substitute("deserialized shared metadata failed")));
-
+#ifdef BE_TEST
+    bool inject_error = false;
+    TEST_SYNC_POINT_CALLBACK("TabletManager::parse_bundle_tablet_metadata::corruption", &inject_error);
+    if (inject_error) {
+        return Status::Corruption("injected error");
+    }
+#endif
     return bundle_metadata;
 }
 
@@ -581,7 +593,18 @@ StatusOr<TabletMetadataPtr> TabletManager::get_single_tablet_metadata(int64_t ta
     g_read_bundle_tablet_meta_latency << (butil::gettimeofday_us() - t0);
 
     auto file_size = serialized_string.size();
-    ASSIGN_OR_RETURN(auto bundle_metadata, parse_bundle_tablet_metadata(path, serialized_string));
+    BundleTabletMetadataPtr bundle_metadata;
+    auto bundle_metadata_status = parse_bundle_tablet_metadata(path, serialized_string);
+    if (!bundle_metadata_status.ok()) {
+        RETURN_IF_ERROR(corrupted_tablet_meta_handler(bundle_metadata_status.status(), path));
+        // read bundle metadata again
+        ASSIGN_OR_RETURN(auto input_file, file_system->new_random_access_file(opts, path));
+        ASSIGN_OR_RETURN(serialized_string, input_file->read_all());
+        file_size = serialized_string.size();
+        ASSIGN_OR_RETURN(bundle_metadata, parse_bundle_tablet_metadata(path, serialized_string));
+    } else {
+        bundle_metadata = bundle_metadata_status.value();
+    }
 
     auto meta_it = bundle_metadata->tablet_meta_pages().find(tablet_id);
     size_t offset = 0;
diff --git a/be/src/storage/lake/tablet_manager.h b/be/src/storage/lake/tablet_manager.h
@@ -249,6 +249,7 @@ class TabletManager {
                                                      int64_t expected_gtid, const std::shared_ptr<FileSystem>& fs);
     StatusOr<TxnLogPtr> load_txn_log(const std::string& txn_log_location, bool fill_cache);
     StatusOr<CombinedTxnLogPtr> load_combined_txn_log(const std::string& path, bool fill_cache);
+    Status corrupted_tablet_meta_handler(const Status& s, const std::string& metadata_location);
 
 private:
     std::shared_ptr<LocationProvider> _location_provider;
diff --git a/be/src/storage/protobuf_file.cpp b/be/src/storage/protobuf_file.cpp
@@ -176,6 +176,12 @@ Status ProtobufFile::load(::google::protobuf::Message* message, bool fill_cache)
     if (bool parsed = message->ParseFromString(serialized_string); !parsed) {
         return Status::Corruption(fmt::format("failed to parse protobuf file {}", _path));
     }
+#ifndef BE_TEST
     return Status::OK();
+#else
+    Status st = Status::OK();
+    TEST_SYNC_POINT_CALLBACK("ProtobufFile::load::corruption", &st);
+    return st;
+#endif
 }
 } // namespace starrocks
diff --git a/be/test/storage/lake/tablet_manager_test.cpp b/be/test/storage/lake/tablet_manager_test.cpp
@@ -103,6 +103,41 @@ TEST_F(LakeTabletManagerTest, tablet_meta_write_and_read) {
     EXPECT_TRUE(res.status().is_not_found());
 }
 
+TEST_F(LakeTabletManagerTest, tablet_meta_read_corrupted_and_recover) {
+    starrocks::TabletMetadata metadata;
+    auto tablet_id = next_id();
+    metadata.set_id(tablet_id);
+    metadata.set_version(2);
+    auto rowset_meta_pb = metadata.add_rowsets();
+    rowset_meta_pb->set_id(2);
+    rowset_meta_pb->set_overlapped(false);
+    rowset_meta_pb->set_data_size(1024);
+    rowset_meta_pb->set_num_rows(5);
+    EXPECT_OK(_tablet_manager->put_tablet_metadata(metadata));
+    auto res = _tablet_manager->get_tablet_metadata(tablet_id, 2);
+    EXPECT_TRUE(res.ok());
+    _tablet_manager->metacache()->prune();
+    bool is_first_time = true;
+    SyncPoint::GetInstance()->EnableProcessing();
+    DeferOp defer([]() {
+        SyncPoint::GetInstance()->ClearCallBack("ProtobufFile::load::corruption");
+        SyncPoint::GetInstance()->ClearCallBack("TabletManager::corrupted_tablet_meta_handler");
+        SyncPoint::GetInstance()->DisableProcessing();
+    });
+    SyncPoint::GetInstance()->SetCallBack("ProtobufFile::load::corruption", [&](void* arg) {
+        if (is_first_time) {
+            *(Status*)arg = Status::Corruption("injected error");
+            is_first_time = false;
+        }
+    });
+    SyncPoint::GetInstance()->SetCallBack("TabletManager::corrupted_tablet_meta_handler",
+                                          [](void* arg) { *(Status*)arg = Status::OK(); });
+    res = _tablet_manager->get_tablet_metadata(tablet_id, 2);
+    EXPECT_TRUE(res.ok());
+    EXPECT_EQ(res.value()->id(), tablet_id);
+    EXPECT_EQ(res.value()->version(), 2);
+}
+
 // NOLINTNEXTLINE
 TEST_F(LakeTabletManagerTest, txnlog_write_and_read) {
     starrocks::TxnLog txnLog;
@@ -690,6 +725,32 @@ TEST_F(LakeTabletManagerTest, put_bundle_tablet_metadata) {
         ASSERT_EQ(metadata->historical_schemas_size(), 2);
     }
 
+    {
+        _tablet_manager->metacache()->prune();
+        // inject corruption error
+        SyncPoint::GetInstance()->EnableProcessing();
+        bool is_first_time = true;
+        DeferOp defer([]() {
+            SyncPoint::GetInstance()->ClearCallBack("TabletManager::parse_bundle_tablet_metadata::corruption");
+            SyncPoint::GetInstance()->ClearCallBack("TabletManager::corrupted_tablet_meta_handler");
+            SyncPoint::GetInstance()->DisableProcessing();
+        });
+        SyncPoint::GetInstance()->SetCallBack("TabletManager::parse_bundle_tablet_metadata::corruption",
+                                              [&](void* arg) {
+                                                  if (is_first_time) {
+                                                      *(bool*)arg = true;
+                                                      is_first_time = false;
+                                                  }
+                                              });
+        SyncPoint::GetInstance()->SetCallBack("TabletManager::corrupted_tablet_meta_handler",
+                                              [](void* arg) { *(Status*)arg = Status::OK(); });
+        auto res = _tablet_manager->get_tablet_metadata(1, 2);
+        EXPECT_TRUE(res.ok()) << res.status().to_string();
+        TabletMetadataPtr metadata = std::move(res).value();
+        ASSERT_EQ(metadata->schema().id(), 10);
+        ASSERT_EQ(metadata->historical_schemas_size(), 2);
+    }
+
     // multi thread read
     {
         std::vector<std::thread> threads;

Original file line number	Diff line number	Diff line change
`@@ -176,6 +176,12 @@ Status ProtobufFile::load(::google::protobuf::Message* message, bool fill_cache)`
`176`	`176`	`if (bool parsed = message->ParseFromString(serialized_string); !parsed) {`
`177`	`177`	`return Status::Corruption(fmt::format("failed to parse protobuf file {}", _path));`
`178`	`178`	`}`
	`179`	`+#ifndef BE_TEST`
`179`	`180`	`return Status::OK();`
	`181`	`+#else`
	`182`	`+ Status st = Status::OK();`
	`183`	`+ TEST_SYNC_POINT_CALLBACK("ProtobufFile::load::corruption", &st);`
	`184`	`+ return st;`
	`185`	`+#endif`
`180`	`186`	`}`
`181`	`187`	`} // namespace starrocks`