[BugFix] fix delvec crc32 upgrade compatibility (backport #65442) (#65482)

Signed-off-by: Yixin Luo <[email protected]>
Co-authored-by: Yixin Luo <[email protected]>

Author: mergify[bot]

be/src/storage/lake/meta_file.cpp

@@ -36,6 +36,8 @@ static std::string delvec_cache_key(int64_t tablet_id, const DelvecPagePB& page)
     DelvecCacheKeyPB cache_key_pb;
     cache_key_pb.set_id(tablet_id);
     cache_key_pb.mutable_delvec_page()->CopyFrom(page);
+    // Do not include crc32c_gen_version in cache key
+    cache_key_pb.mutable_delvec_page()->clear_crc32c_gen_version();
     return cache_key_pb.SerializeAsString();
 }
 
@@ -458,6 +460,7 @@ Status MetaFileBuilder::_finalize_delvec(int64_t version, int64_t txn_id) {
             each_delvec.second.set_offset(iter->second.offset());
             each_delvec.second.set_size(iter->second.size());
             each_delvec.second.set_crc32c(iter->second.crc32c());
+            each_delvec.second.set_crc32c_gen_version(version);
             // record from cache key to segment id, so we can fill up cache later
             _cache_key_to_segment_id[delvec_cache_key(_tablet_meta->id(), each_delvec.second)] = iter->first;
             _delvecs.erase(iter);
@@ -467,6 +470,7 @@ Status MetaFileBuilder::_finalize_delvec(int64_t version, int64_t txn_id) {
     // 2. insert new delvec to meta
     for (auto&& each_delvec : _delvecs) {
         each_delvec.second.set_version(version);
+        each_delvec.second.set_crc32c_gen_version(version);
         (*_tablet_meta->mutable_delvec_meta()->mutable_delvecs())[each_delvec.first] = each_delvec.second;
         // record from cache key to segment id, so we can fill up cache later
         _cache_key_to_segment_id[delvec_cache_key(_tablet_meta->id(), each_delvec.second)] = each_delvec.first;
@@ -606,7 +610,7 @@ Status get_del_vec(TabletManager* tablet_mgr, const TabletMetadata& metadata, ui
         ASSIGN_OR_RETURN(auto rf,
                          fs::new_random_access_file(opts, tablet_mgr->delvec_location(metadata.id(), delvec_name)));
         RETURN_IF_ERROR(rf->read_at_fully(iter->second.offset(), buf.data(), iter->second.size()));
-        if (iter->second.has_crc32c()) {
+        if (iter->second.has_crc32c() && iter->second.crc32c_gen_version() == iter->second.version()) {
             // check crc32c
             uint32_t crc32c = crc32c::Value(buf.data(), iter->second.size());
             if (crc32c != crc32c::Unmask(iter->second.crc32c())) {

gensrc/proto/lake_types.proto

@@ -34,6 +34,8 @@ message DelvecPagePB {
     optional uint64 offset = 2;
     optional uint64 size = 3;
     optional uint32 crc32c = 4;
+    // For upgrade compatibility.
+    optional int64 crc32c_gen_version = 5;
 }
 
 message DelvecCacheKeyPB {