1

Author: wyxxxcat

.github/CODEOWNERS

@@ -19,7 +19,7 @@
 # https://docs.github.com/en/repositories/managing-your-repositorys-settings-and-features/customizing-your-repository/about-code-owners#codeowners-syntax
 
 be/src/io/ @gavinchou @dataroaring
-be/src/agent/be_exec_version_manager.cpp @BiteTheDDDDt
+be/src/agent/be_exec_version_manager.cpp @BiteTheDDDDt @zclllyybb
 fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java @dataroaring @CalvinKirs @morningman
 **/pom.xml @CalvinKirs @morningman
 fe/fe-common/src/main/java/org/apache/doris/common/FeMetaVersion.java @dataroaring @morningman @yiguolei @xiaokang
@@ -37,3 +37,5 @@ gensrc/proto/olap_file.proto @gavinchou @dataroaring @yiguolei
 gensrc/proto/cloud.proto @gavinchou @dataroaring @w41ter
 fe/fe-core/src/main/java/org/apache/doris/fs @CalvinKirs
 fe/fe-core/src/main/java/org/apache/doris/fsv2 @CalvinKirs
+be/src/vec/functions @zclllyybb
+be/**/CMakeLists.txt @zclllyybb @BiteTheDDDDt

.licenserc.yaml

@@ -28,6 +28,7 @@ header:
     - "**/apache-orc/**"
     - "**/glibc-compatibility/**"
     - "**/gutil/**"
+    - "**/go.sum"
     - "**/test_data/**"
     - "**/jmockit/**"
     - "**/*.json"

be/benchmark/benchmark_bits.hpp

@@ -0,0 +1,67 @@
+// 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 <benchmark/benchmark.h>
+
+#include "util/simd/bits.h"
+
+namespace doris {} // namespace doris
+
+static void BM_Bits_CountZeroNum(benchmark::State& state) {
+    const auto n = static_cast<size_t>(state.range(0));
+    std::vector<int8_t> data(n, 0);
+
+    for (auto _ : state) {
+        auto r = doris::simd::count_zero_num<size_t>(data.data(), data.size());
+        benchmark::DoNotOptimize(r);
+    }
+
+    state.SetBytesProcessed(state.iterations() * n);
+}
+
+static void BM_Bits_CountZeroNumNullMap(benchmark::State& state) {
+    const auto n = static_cast<size_t>(state.range(0));
+    std::vector<int8_t> data(n, 0);
+    std::vector<uint8_t> null_map(n, 0);
+
+    for (auto _ : state) {
+        auto r = doris::simd::count_zero_num<size_t>(data.data(), null_map.data(), data.size());
+        benchmark::DoNotOptimize(r);
+    }
+
+    state.SetBytesProcessed(state.iterations() * n);
+}
+
+BENCHMARK(BM_Bits_CountZeroNum)
+        ->Unit(benchmark::kNanosecond)
+        ->Arg(16)   // 16 bytes
+        ->Arg(32)   // 32 bytes
+        ->Arg(64)   // 64 bytes
+        ->Arg(256)  // 256 bytes
+        ->Arg(1024) // 1KB
+        ->Repetitions(5)
+        ->DisplayAggregatesOnly();
+
+BENCHMARK(BM_Bits_CountZeroNumNullMap)
+        ->Unit(benchmark::kNanosecond)
+        ->Arg(16)   // 16 bytes
+        ->Arg(32)   // 32 bytes
+        ->Arg(64)   // 64 bytes
+        ->Arg(256)  // 256 bytes
+        ->Arg(1024) // 1KB
+        ->Repetitions(5)
+        ->DisplayAggregatesOnly();

be/benchmark/benchmark_main.cpp

@@ -18,6 +18,7 @@
 #include <benchmark/benchmark.h>
 
 #include "benchmark_bit_pack.hpp"
+#include "benchmark_bits.hpp"
 #include "benchmark_block_bloom_filter.hpp"
 #include "benchmark_fastunion.hpp"
 #include "benchmark_hll_merge.hpp"

be/cmake/thirdparty.cmake

@@ -68,6 +68,7 @@ add_thirdparty(curl)
 add_thirdparty(lz4)
 add_thirdparty(thrift)
 add_thirdparty(thriftnb)
+add_thirdparty(crc32c)
 
 add_thirdparty(libevent_core LIBNAME "lib/libevent_core.a")
 add_thirdparty(libevent_openssl LIBNAME "lib/libevent_openssl.a")

be/src/agent/cgroup_cpu_ctl.cpp

@@ -292,7 +292,7 @@ Status CgroupCpuCtl::delete_unused_cgroup_path(std::set<uint64_t>& used_wg_ids)
         std::string wg_path = query_path + unused_wg_id;
         int ret = rmdir(wg_path.c_str());
         if (ret < 0) {
-            LOG(WARNING) << "rmdir failed, path=" << wg_path;
+            LOG(WARNING) << "remove cgroup path failed, path=" << wg_path << ", error=" << ret;
             failed_count++;
         }
     }
@@ -317,7 +317,8 @@ Status CgroupV1CpuCtl::init() {
     if (access(_cgroup_v1_cpu_tg_path.c_str(), F_OK) != 0) {
         int ret = mkdir(_cgroup_v1_cpu_tg_path.c_str(), S_IRWXU);
         if (ret != 0) {
-            LOG(ERROR) << "cgroup v1 mkdir workload group failed, path=" << _cgroup_v1_cpu_tg_path;
+            LOG(WARNING) << "cgroup v1 make workload group dir failed, path="
+                         << _cgroup_v1_cpu_tg_path << ", error=" << ret;
             return Status::InternalError<false>("cgroup v1 mkdir workload group failed, path={}",
                                                 _cgroup_v1_cpu_tg_path);
         }
@@ -382,6 +383,8 @@ Status CgroupV2CpuCtl::init() {
     if (access(_cgroup_v2_query_wg_path.c_str(), F_OK) != 0) {
         int ret = mkdir(_cgroup_v2_query_wg_path.c_str(), S_IRWXU);
         if (ret != 0) {
+            LOG(WARNING) << "cgroup v2 make workload group dir failed, path="
+                         << _cgroup_v2_query_wg_path << ", error=" << ret;
             return Status::InternalError<false>("cgroup v2 mkdir wg failed, path={}",
                                                 _cgroup_v2_query_wg_path);
         }

be/src/agent/task_worker_pool.cpp

@@ -124,10 +124,6 @@ bool register_task_info(const TTaskType::type task_type, int64_t signature) {
         // no need to report task of these types
         return true;
     }
-    if (task_type == TTaskType::type::DROP && config::is_cloud_mode()) {
-        // cloud no need to report drop task status
-        return true;
-    }
 
     if (signature == -1) { // No need to report task with unintialized signature
         return true;
@@ -1874,6 +1870,8 @@ void drop_tablet_callback(StorageEngine& engine, const TAgentTaskRequest& req) {
 
 void drop_tablet_callback(CloudStorageEngine& engine, const TAgentTaskRequest& req) {
     const auto& drop_tablet_req = req.drop_tablet_req;
+    // here drop_tablet_req.tablet_id is the signature of the task, see DropReplicaTask in fe
+    Defer defer = [&] { remove_task_info(req.task_type, req.signature); };
     DBUG_EXECUTE_IF("WorkPoolCloudDropTablet.drop_tablet_callback.failed", {
         LOG_WARNING("WorkPoolCloudDropTablet.drop_tablet_callback.failed")
                 .tag("tablet_id", drop_tablet_req.tablet_id);

be/src/cloud/cloud_base_compaction.cpp

@@ -226,6 +226,7 @@ Status CloudBaseCompaction::pick_rowsets_to_compact() {
                     << ", num_cumulative_rowsets=" << _input_rowsets.size() - 1
                     << ", base_compaction_num_cumulative_rowsets="
                     << config::base_compaction_min_rowset_num;
+        apply_txn_size_truncation_and_log("CloudBaseCompaction");
         return Status::OK();
     }
 
@@ -251,6 +252,7 @@ Status CloudBaseCompaction::pick_rowsets_to_compact() {
                     << ", base_size=" << base_size
                     << ", cumulative_base_ratio=" << cumulative_base_ratio
                     << ", policy_ratio=" << base_cumulative_delta_ratio;
+        apply_txn_size_truncation_and_log("CloudBaseCompaction");
         return Status::OK();
     }
 
@@ -263,6 +265,7 @@ Status CloudBaseCompaction::pick_rowsets_to_compact() {
                     << ", interval_since_last_base_compaction="
                     << interval_since_last_base_compaction
                     << ", interval_threshold=" << interval_threshold;
+        apply_txn_size_truncation_and_log("CloudBaseCompaction");
         return Status::OK();
     }
 

be/src/cloud/cloud_cumulative_compaction.cpp

@@ -375,7 +375,7 @@ Status CloudCumulativeCompaction::modify_rowsets() {
         if (_input_rowsets.size() == 1) {
             DCHECK_EQ(_output_rowset->version(), _input_rowsets[0]->version());
             // MUST NOT move input rowset to stale path
-            cloud_tablet()->add_rowsets({_output_rowset}, true, wrlock);
+            cloud_tablet()->add_rowsets({_output_rowset}, true, wrlock, true);
         } else {
             cloud_tablet()->delete_rowsets(_input_rowsets, wrlock);
             cloud_tablet()->add_rowsets({_output_rowset}, false, wrlock);
@@ -532,6 +532,8 @@ Status CloudCumulativeCompaction::pick_rowsets_to_compact() {
         return Status::Error<CUMULATIVE_NO_SUITABLE_VERSION>(
                 "no suitable versions: only one rowset and not overlapping");
     }
+
+    apply_txn_size_truncation_and_log("CloudCumulativeCompaction");
     return Status::OK();
 }
 

be/src/cloud/cloud_index_change_compaction.cpp

@@ -69,6 +69,11 @@ Status CloudIndexChangeCompaction::prepare_compact() {
 
     _input_rowsets.push_back(input_rowset);
 
+    // Apply transaction size truncation (usually only 1 rowset for index change)
+    int64_t kept_size = 0;
+    int64_t truncated_size = 0;
+    cloud::truncate_rowsets_by_txn_size(_input_rowsets, kept_size, truncated_size);
+
     for (auto& rs : _input_rowsets) {
         _input_row_num += rs->num_rows();
         _input_segments += rs->num_segments();
@@ -473,4 +478,4 @@ Status CloudIndexChangeCompaction::garbage_collection() {
     return st;
 }
 
-} // namespace doris
+} // namespace doris