(cherry-pick)[branch-2.1] add calc tablet file crc and fix single compaction test

be/src/http/action/calc_file_crc_action.cpp

@@ -0,0 +1,134 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "http/action/calc_file_crc_action.h"
+
+#include <rapidjson/prettywriter.h>
+#include <rapidjson/rapidjson.h>
+#include <rapidjson/stringbuffer.h>
+
+#include <algorithm>
+#include <exception>
+#include <string>
+
+#include "common/logging.h"
+#include "common/status.h"
+#include "http/http_channel.h"
+#include "http/http_headers.h"
+#include "http/http_request.h"
+#include "http/http_status.h"
+#include "olap/storage_engine.h"
+#include "olap/tablet_manager.h"
+#include "util/stopwatch.hpp"
+
+namespace doris {
+using namespace ErrorCode;
+
+CalcFileCrcAction::CalcFileCrcAction(ExecEnv* exec_env, TPrivilegeHier::type hier,
+                                     TPrivilegeType::type ptype)
+        : HttpHandlerWithAuth(exec_env, hier, ptype) {}
+
+// calculate the crc value of the files in the tablet
+Status CalcFileCrcAction::_handle_calc_crc(HttpRequest* req, uint32_t* crc_value,
+                                           int64_t* start_version, int64_t* end_version,
+                                           int32_t* rowset_count, int64_t* file_count) {
+    uint64_t tablet_id = 0;
+    const auto& req_tablet_id = req->param(TABLET_ID_KEY);
+    if (req_tablet_id.empty()) {
+        return Status::InternalError("tablet id can not be empty!");
+    }
+
+    try {
+        tablet_id = std::stoull(req_tablet_id);
+    } catch (const std::exception& e) {
+        return Status::InternalError("convert tablet id or failed, {}", e.what());
+    }
+
+    TabletSharedPtr tablet = StorageEngine::instance()->tablet_manager()->get_tablet(tablet_id);
+    if (tablet == nullptr) {
+        return Status::NotFound("Tablet not found. tablet_id={}", tablet_id);
+    }
+
+    const auto& req_start_version = req->param(PARAM_START_VERSION);
+    const auto& req_end_version = req->param(PARAM_END_VERSION);
+
+    *start_version = 0;
+    *end_version = tablet->max_version().second;
+
+    if (!req_start_version.empty()) {
+        try {
+            *start_version = std::stoll(req_start_version);
+        } catch (const std::exception& e) {
+            return Status::InternalError("convert start version failed, {}", e.what());
+        }
+    }
+    if (!req_end_version.empty()) {
+        try {
+            *end_version =
+                    std::min(*end_version, static_cast<int64_t>(std::stoll(req_end_version)));
+        } catch (const std::exception& e) {
+            return Status::InternalError("convert end version failed, {}", e.what());
+        }
+    }
+
+    auto st = tablet->calc_local_file_crc(crc_value, *start_version, *end_version, rowset_count,
+                                          file_count);
+    if (!st.ok()) {
+        return st;
+    }
+    return Status::OK();
+}
+
+void CalcFileCrcAction::handle(HttpRequest* req) {
+    uint32_t crc_value = 0;
+    int64_t start_version = 0;
+    int64_t end_version = 0;
+    int32_t rowset_count = 0;
+    int64_t file_count = 0;
+
+    MonotonicStopWatch timer;
+    timer.start();
+    Status st = _handle_calc_crc(req, &crc_value, &start_version, &end_version, &rowset_count,
+                                 &file_count);
+    timer.stop();
+    LOG(INFO) << "Calc tablet file crc finished, status = " << st << ", crc_value = " << crc_value
+              << ", use time = " << timer.elapsed_time() / 1000000 << "ms";
+
+    if (!st.ok()) {
+        HttpChannel::send_reply(req, HttpStatus::OK, st.to_json());
+    } else {
+        rapidjson::StringBuffer s;
+        rapidjson::PrettyWriter<rapidjson::StringBuffer> writer(s);
+        writer.StartObject();
+        writer.Key("crc_value");
+        writer.String(std::to_string(crc_value).data());
+        writer.Key("used_time_ms");
+        writer.String(std::to_string(timer.elapsed_time() / 1000000).data());
+        writer.Key("start_version");
+        writer.String(std::to_string(start_version).data());
+        writer.Key("end_version");
+        writer.String(std::to_string(end_version).data());
+        writer.Key("rowset_count");
+        writer.String(std::to_string(rowset_count).data());
+        writer.Key("file_count");
+        writer.String(std::to_string(file_count).data());
+        writer.EndObject();
+        HttpChannel::send_reply(req, HttpStatus::OK, s.GetString());
+    }
+}
+
+} // end namespace doris

be/src/http/action/calc_file_crc_action.h

@@ -0,0 +1,49 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <stdint.h>
+
+#include <string>
+
+#include "common/status.h"
+#include "http/http_handler_with_auth.h"
+
+namespace doris {
+class HttpRequest;
+class StorageEngine;
+class ExecEnv;
+
+const std::string PARAM_START_VERSION = "start_version";
+const std::string PARAM_END_VERSION = "end_version";
+
+// This action is used to calculate the crc value of the files in the tablet.
+class CalcFileCrcAction : public HttpHandlerWithAuth {
+public:
+    CalcFileCrcAction(ExecEnv* exec_env, TPrivilegeHier::type hier, TPrivilegeType::type ptype);
+
+    ~CalcFileCrcAction() override = default;
+
+    void handle(HttpRequest* req) override;
+
+private:
+    Status _handle_calc_crc(HttpRequest* req, uint32_t* crc_value, int64_t* start_version,
+                            int64_t* end_version, int32_t* rowset_count, int64_t* file_count);
+};
+
+} // end namespace doris

be/src/olap/rowset/beta_rowset.cpp

@@ -42,6 +42,7 @@
 #include "olap/rowset/segment_v2/inverted_index_desc.h"
 #include "olap/tablet_schema.h"
 #include "olap/utils.h"
+#include "util/crc32c.h"
 #include "util/debug_points.h"
 #include "util/doris_metrics.h"
 
@@ -630,4 +631,75 @@ Status BetaRowset::add_to_binlog() {
     return Status::OK();
 }
 
+Status BetaRowset::calc_local_file_crc(uint32_t* crc_value, int64_t* file_count) {
+    DCHECK(is_local());
+    auto fs = _rowset_meta->fs();
+    if (!fs) {
+        return Status::Error<INIT_FAILED>("get fs failed");
+    }
+    if (fs->type() != io::FileSystemType::LOCAL) {
+        return Status::InternalError("should be local file system");
+    }
+
+    if (num_segments() < 1) {
+        *crc_value = 0x92a8fc17; // magic code from crc32c table
+        return Status::OK();
+    }
+
+    // 1. pick up all the files including dat file and idx file
+    std::vector<io::Path> local_paths;
+    for (int i = 0; i < num_segments(); ++i) {
+        auto local_seg_path = segment_file_path(i);
+        local_paths.emplace_back(local_seg_path);
+        if (_schema->get_inverted_index_storage_format() != InvertedIndexStorageFormatPB::V1) {
+            if (_schema->has_inverted_index()) {
+                std::string local_inverted_index_file =
+                        InvertedIndexDescriptor::get_index_file_name(local_seg_path);
+                local_paths.emplace_back(local_inverted_index_file);
+            }
+        } else {
+            for (auto& column : _schema->columns()) {
+                const TabletIndex* index_meta = _schema->get_inverted_index(*column);
+                if (index_meta) {
+                    std::string local_inverted_index_file =
+                            InvertedIndexDescriptor::get_index_file_name(
+                                    local_seg_path, index_meta->index_id(),
+                                    index_meta->get_index_suffix());
+                    local_paths.emplace_back(local_inverted_index_file);
+                }
+            }
+        }
+    }
+
+    // 2. calculate the md5sum of each file
+    auto* local_fs = static_cast<io::LocalFileSystem*>(fs.get());
+    DCHECK(local_paths.size() > 0);
+    std::vector<std::string> all_file_md5;
+    all_file_md5.reserve(local_paths.size());
+    for (const auto& file_path : local_paths) {
+        DBUG_EXECUTE_IF("fault_inject::BetaRowset::calc_local_file_crc", {
+            return Status::Error<OS_ERROR>("fault_inject calc_local_file_crc error");
+        });
+        std::string file_md5sum;
+        auto status = local_fs->md5sum(file_path, &file_md5sum);
+        if (!status.ok()) {
+            return status;
+        }
+        VLOG_CRITICAL << fmt::format("calc file_md5sum finished. file_path={}, md5sum={}",
+                                     file_path.string(), file_md5sum);
+        all_file_md5.emplace_back(std::move(file_md5sum));
+    }
+    std::sort(all_file_md5.begin(), all_file_md5.end());
+
+    // 3. calculate the crc_value based on all_file_md5
+    DCHECK(local_paths.size() == all_file_md5.size());
+    *crc_value = 0;
+    *file_count = local_paths.size();
+    for (int i = 0; i < all_file_md5.size(); i++) {
+        *crc_value = crc32c::Extend(*crc_value, all_file_md5[i].data(), all_file_md5[i].size());
+    }
+
+    return Status::OK();
+}
+
 } // namespace doris

be/src/olap/rowset/beta_rowset.h

@@ -100,6 +100,8 @@ class BetaRowset final : public Rowset {
 
     [[nodiscard]] virtual Status add_to_binlog() override;
 
+    Status calc_local_file_crc(uint32_t* crc_value, int64_t* file_count);
+
 protected:
     BetaRowset(const TabletSchemaSPtr& schema, const std::string& tablet_path,
                const RowsetMetaSharedPtr& rowset_meta);

be/src/olap/tablet.cpp

@@ -110,6 +110,7 @@
 #include "service/point_query_executor.h"
 #include "tablet.h"
 #include "util/bvar_helper.h"
+#include "util/crc32c.h"
 #include "util/debug_points.h"
 #include "util/defer_op.h"
 #include "util/doris_metrics.h"
@@ -3867,4 +3868,35 @@ Status Tablet::check_delete_bitmap_correctness(DeleteBitmapPtr delete_bitmap, in
     }
     return Status::OK();
 }
+Status Tablet::calc_local_file_crc(uint32_t* crc_value, int64_t start_version, int64_t end_version,
+                                   int32_t* rowset_count, int64_t* file_count) {
+    Version v(start_version, end_version);
+    std::vector<RowsetSharedPtr> rowsets;
+    traverse_rowsets([&rowsets, &v](const auto& rs) {
+        // get local rowsets
+        if (rs->is_local() && v.contains(rs->version())) {
+            rowsets.emplace_back(rs);
+        }
+    });
+    std::sort(rowsets.begin(), rowsets.end(), Rowset::comparator);
+    *rowset_count = rowsets.size();
+
+    *crc_value = 0;
+    *file_count = 0;
+    for (const auto& rs : rowsets) {
+        uint32_t rs_crc_value;
+        int64_t rs_file_count = 0;
+        auto rowset = std::static_pointer_cast<BetaRowset>(rs);
+        auto st = rowset->calc_local_file_crc(&rs_crc_value, &rs_file_count);
+        if (!st.ok()) {
+            return st;
+        }
+        // crc_value is calculated based on the crc_value of each rowset.
+        *crc_value = crc32c::Extend(*crc_value, reinterpret_cast<const char*>(&rs_crc_value),
+                                    sizeof(rs_crc_value));
+        *file_count += rs_file_count;
+    }
+    return Status::OK();
+}
+
 } // namespace doris

be/src/olap/tablet.h

@@ -595,6 +595,8 @@ class Tablet final : public BaseTablet {
     }
     inline bool is_full_compaction_running() const { return _is_full_compaction_running; }
     void clear_cache();
+    Status calc_local_file_crc(uint32_t* crc_value, int64_t start_version, int64_t end_version,
+                               int32_t* rowset_count, int64_t* file_count);
 
 private:
     Status _init_once_action();

be/src/service/http_service.cpp

@@ -27,6 +27,7 @@
 #include "common/status.h"
 #include "http/action/adjust_log_level.h"
 #include "http/action/adjust_tracing_dump.h"
+#include "http/action/calc_file_crc_action.h"
 #include "http/action/check_rpc_channel_action.h"
 #include "http/action/check_tablet_segment_action.h"
 #include "http/action/checksum_action.h"
@@ -315,6 +316,10 @@ Status HttpService::start() {
             _env, TPrivilegeHier::GLOBAL, TPrivilegeType::ADMIN, "REPORT_DISK_STATE"));
     _ev_http_server->register_handler(HttpMethod::GET, "/api/report/disk", report_disk_action);
 
+    CalcFileCrcAction* calc_crc_action =
+            _pool.add(new CalcFileCrcAction(_env, TPrivilegeHier::GLOBAL, TPrivilegeType::ADMIN));
+    _ev_http_server->register_handler(HttpMethod::GET, "/api/calc_crc", calc_crc_action);
+
     ReportAction* report_task_action = _pool.add(
             new ReportAction(_env, TPrivilegeHier::GLOBAL, TPrivilegeType::ADMIN, "REPORT_TASK"));
     _ev_http_server->register_handler(HttpMethod::GET, "/api/report/task", report_task_action);