(doris) branch master updated: [fix](load) disable num segments check in compatibility mode (#41053)

[dataroaring]

This is an automated email from the ASF dual-hosted git repository.

dataroaring pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git


The following commit(s) were added to refs/heads/master by this push:
     new eb7fd6e74b3 [fix](load) disable num segments check in compatibility 
mode (#41053)
eb7fd6e74b3 is described below

commit eb7fd6e74b333ee1cfaa3251de84eb79f7bc76b0
Author: Kaijie Chen <c...@apache.org>
AuthorDate: Mon Sep 23 11:34:51 2024 +0800

    [fix](load) disable num segments check in compatibility mode (#41053)
    
    ## Proposed changes
    
    When using mixed version of BEs, sink v2 on old BE won't report num
    segments to load streams on the new BE.
    This will cause false positive segment num mismatch.
    
    This PR addressed this issue by disabling num segments check when any
    tablets_to_commit proto has not set num_segments field.
---
 be/src/runtime/load_stream.cpp | 9 +++++++--
 be/src/runtime/load_stream.h   | 2 ++
 2 files changed, 9 insertions(+), 2 deletions(-)

diff --git a/be/src/runtime/load_stream.cpp b/be/src/runtime/load_stream.cpp
index a52fa01370c..80cd167260c 100644
--- a/be/src/runtime/load_stream.cpp
+++ b/be/src/runtime/load_stream.cpp
@@ -293,7 +293,7 @@ Status TabletStream::close() {
     }
 
     DBUG_EXECUTE_IF("TabletStream.close.segment_num_mismatch", { 
_num_segments++; });
-    if (_next_segid.load() != _num_segments) {
+    if (_check_num_segments && (_next_segid.load() != _num_segments)) {
         _status = Status::Corruption(
                 "segment num mismatch in tablet {}, expected: {}, actual: {}, 
load_id: {}", _id,
                 _num_segments, _next_segid.load(), print_id(_load_id));
@@ -380,9 +380,14 @@ void IndexStream::close(const std::vector<PTabletID>& 
tablets_to_commit,
         auto it = _tablet_streams_map.find(tablet.tablet_id());
         if (it == _tablet_streams_map.end()) {
             _init_tablet_stream(tablet_stream, tablet.tablet_id(), 
tablet.partition_id());
+        } else {
+            tablet_stream = it->second;
+        }
+        if (tablet.has_num_segments()) {
             tablet_stream->add_num_segments(tablet.num_segments());
         } else {
-            it->second->add_num_segments(tablet.num_segments());
+            // for compatibility reasons (sink from old version BE)
+            tablet_stream->disable_num_segments_check();
         }
     }
 
diff --git a/be/src/runtime/load_stream.h b/be/src/runtime/load_stream.h
index 427bc2dbb62..3b649c68835 100644
--- a/be/src/runtime/load_stream.h
+++ b/be/src/runtime/load_stream.h
@@ -53,6 +53,7 @@ public:
     Status append_data(const PStreamHeader& header, butil::IOBuf* data);
     Status add_segment(const PStreamHeader& header, butil::IOBuf* data);
     void add_num_segments(int64_t num_segments) { _num_segments += 
num_segments; }
+    void disable_num_segments_check() { _check_num_segments = false; }
     Status close();
     int64_t id() const { return _id; }
 
@@ -65,6 +66,7 @@ private:
     std::unordered_map<int64_t, std::unique_ptr<SegIdMapping>> _segids_mapping;
     std::atomic<uint32_t> _next_segid;
     int64_t _num_segments = 0;
+    bool _check_num_segments = true;
     bthread::Mutex _lock;
     Status _status;
     PUniqueId _load_id;


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org