Merge branch 'master' into add-report-tablet-metrics

Author: xy720

.github/workflows/build-thirdparty.yml

@@ -132,6 +132,8 @@ jobs:
           export PATH="$(find /usr/lib/jvm/java-8-openjdk* -maxdepth 1 -type d -name 'bin'):${PATH}"
           export JAVA_HOME="$(find /usr/lib/jvm/java-8-openjdk* -maxdepth 0)"
           export DORIS_TOOLCHAIN=gcc
+          export CMAKE_POLICY_VERSION_MINIMUM=3.10
+          export CUSTOM_CMAKE="/usr/local/bin/cmake"
 
           cd thirdparty
           ./build-thirdparty.sh -j "$(nproc)"
@@ -190,6 +192,8 @@ jobs:
       - name: Build
         run: |
           export MACOSX_DEPLOYMENT_TARGET=12.0
+          export CMAKE_POLICY_VERSION_MINIMUM=3.10
+          export CUSTOM_CMAKE="/usr/local/bin/cmake"
 
           cd thirdparty
           ./build-thirdparty.sh -j "$(nproc)"
@@ -248,6 +252,8 @@ jobs:
       - name: Build
         run: |
           export MACOSX_DEPLOYMENT_TARGET=12.0
+          export CMAKE_POLICY_VERSION_MINIMUM=3.10
+          export CUSTOM_CMAKE="/usr/local/bin/cmake"
 
           cd thirdparty
           ./build-thirdparty.sh -j "$(nproc)"

be/benchmark/benchmark_bit_pack.hpp

@@ -50,7 +50,7 @@ void bit_pack(const T* input, uint8_t in_num, int bit_width, uint8_t* output) {
 }
 
 static void BM_BitPack(benchmark::State& state) {
-    int w = state.range(0);
+    int w = (int)state.range(0);
     int n = 255;
 
     std::default_random_engine e;
@@ -67,15 +67,15 @@ static void BM_BitPack(benchmark::State& state) {
     for (auto _ : state) {
         benchmark::DoNotOptimize(test_data.data());
         benchmark::DoNotOptimize(output.data());
-        bit_pack(test_data.data(), n, w, output.data());
+        bit_pack(test_data.data(), (uint8_t)n, w, output.data());
         benchmark::ClobberMemory();
     }
 
     state.SetBytesProcessed(int64_t(state.iterations()) * size);
 }
 
 static void BM_BitPackOptimized(benchmark::State& state) {
-    int w = state.range(0);
+    int w = (int)state.range(0);
     int n = 255;
 
     std::default_random_engine e;
@@ -94,7 +94,7 @@ static void BM_BitPackOptimized(benchmark::State& state) {
     for (auto _ : state) {
         benchmark::DoNotOptimize(test_data.data());
         benchmark::DoNotOptimize(output.data());
-        forEncoder.bit_pack(test_data.data(), n, w, output.data());
+        forEncoder.bit_pack(test_data.data(), (uint8_t)n, w, output.data());
         benchmark::ClobberMemory();
     }
 

be/benchmark/benchmark_block_bloom_filter.hpp

@@ -0,0 +1,97 @@
+// 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 "common/status.h"
+#include "exprs/block_bloom_filter.hpp"
+
+namespace doris {
+    static std::unique_ptr<BlockBloomFilter> create_bloom_filter(int batch, int log_space_bytes = 20) {
+        auto bloom_filter = std::make_unique<BlockBloomFilter>();
+        [[maybe_unused]] Status status = bloom_filter->init(log_space_bytes, 0);
+
+        for (int i = 0; i < batch; i++) {
+            bloom_filter->insert(i);
+        }
+
+        return bloom_filter;
+    }
+} // namespace doris
+
+static void BM_BBF_BucketInsert(benchmark::State& state) {
+    const int batch = static_cast<int>(state.range(0));
+    auto bf = doris::create_bloom_filter(batch);
+
+    for (auto _ : state) {
+        for (int i = 0; i < batch; i++) {
+            bf->insert(i);
+        }
+    }
+
+    state.SetItemsProcessed(state.iterations() * static_cast<int64_t>(batch));
+}
+
+static void BM_BBF_BucketFind_Hit(benchmark::State& state) {
+    const int batch = static_cast<int>(state.range(0));
+    auto bf = doris::create_bloom_filter(batch);
+
+    for (auto _ : state) {
+        bool acc = false;
+        for (int i = 0; i < batch; i++) {
+            acc ^= bf->find(i);
+        }
+        benchmark::DoNotOptimize(acc);
+    }
+    state.SetItemsProcessed(state.iterations() * static_cast<int64_t>(batch));
+}
+
+static void BM_BBF_BucketFind_Miss(benchmark::State& state) {
+    const int batch = static_cast<int>(state.range(0));
+    auto bf = doris::create_bloom_filter(batch);
+
+    for (auto _ : state) {
+        bool acc = false;
+        for (int i = 0; i < batch; i++) {
+            acc ^= bf->find(i + batch);
+        }
+        benchmark::DoNotOptimize(acc);
+    }
+    state.SetItemsProcessed(state.iterations() * static_cast<int64_t>(batch));
+}
+
+BENCHMARK(BM_BBF_BucketInsert)
+    ->Unit(benchmark::kNanosecond)
+    ->Arg(1 << 12)
+    ->Arg(1 << 15)
+    ->Arg(1 << 18)
+    ->Repetitions(5)
+    ->DisplayAggregatesOnly();
+BENCHMARK(BM_BBF_BucketFind_Hit)
+    ->Unit(benchmark::kNanosecond)
+    ->Arg(1 << 12)
+    ->Arg(1 << 15)
+    ->Arg(1 << 18)
+    ->Repetitions(5)
+    ->DisplayAggregatesOnly();
+BENCHMARK(BM_BBF_BucketFind_Miss)
+    ->Unit(benchmark::kNanosecond)
+    ->Arg(1 << 12)
+    ->Arg(1 << 15)
+    ->Arg(1 << 18)
+    ->Repetitions(5)
+    ->DisplayAggregatesOnly();

be/benchmark/benchmark_main.cpp

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

be/src/cloud/cloud_full_compaction.cpp

@@ -349,8 +349,9 @@ Status CloudFullCompaction::_cloud_full_compaction_update_delete_bitmap(int64_t
     int64_t max_version = cloud_tablet()->max_version().second;
     DCHECK(max_version >= _output_rowset->version().second);
     if (max_version > _output_rowset->version().second) {
-        RETURN_IF_ERROR(cloud_tablet()->capture_consistent_rowsets_unlocked(
-                {_output_rowset->version().second + 1, max_version}, &tmp_rowsets));
+        auto ret = DORIS_TRY(cloud_tablet()->capture_consistent_rowsets_unlocked(
+                {_output_rowset->version().second + 1, max_version}, CaptureRowsetOps {}));
+        tmp_rowsets = std::move(ret.rowsets);
     }
     for (const auto& it : tmp_rowsets) {
         int64_t cur_version = it->rowset_meta()->start_version();

be/src/cloud/cloud_schema_change_job.cpp

@@ -150,7 +150,7 @@ Status CloudSchemaChangeJob::process_alter_tablet(const TAlterTabletReqV2& reque
         // [0-1] is a placeholder rowset, no need to convert
         RETURN_IF_ERROR(_base_tablet->capture_rs_readers({2, start_resp.alter_version()},
                                                          &rs_splits,
-                                                         {.skip_missing_version = false,
+                                                         {.skip_missing_versions = false,
                                                           .enable_prefer_cached_rowset = false,
                                                           .query_freshness_tolerance_ms = -1}));
     }
@@ -199,7 +199,7 @@ Status CloudSchemaChangeJob::process_alter_tablet(const TAlterTabletReqV2& reque
     reader_context.sequence_id_idx = reader_context.tablet_schema->sequence_col_idx();
     reader_context.is_unique = _base_tablet->keys_type() == UNIQUE_KEYS;
     reader_context.batch_size = ALTER_TABLE_BATCH_SIZE;
-    reader_context.delete_bitmap = &_base_tablet->tablet_meta()->delete_bitmap();
+    reader_context.delete_bitmap = _base_tablet->tablet_meta()->delete_bitmap_ptr();
     reader_context.version = Version(0, start_resp.alter_version());
     std::vector<uint32_t> cluster_key_idxes;
     if (!_base_tablet_schema->cluster_key_uids().empty()) {
@@ -509,22 +509,21 @@ Status CloudSchemaChangeJob::_process_delete_bitmap(int64_t alter_version,
     }
 
     // step 1, process incremental rowset without delete bitmap update lock
-    std::vector<RowsetSharedPtr> incremental_rowsets;
     RETURN_IF_ERROR(_cloud_storage_engine.meta_mgr().sync_tablet_rowsets(tmp_tablet.get()));
     int64_t max_version = tmp_tablet->max_version().second;
     LOG(INFO) << "alter table for mow table, calculate delete bitmap of "
               << "incremental rowsets without lock, version: " << start_calc_delete_bitmap_version
               << "-" << max_version << " new_table_id: " << _new_tablet->tablet_id();
     if (max_version >= start_calc_delete_bitmap_version) {
-        RETURN_IF_ERROR(tmp_tablet->capture_consistent_rowsets_unlocked(
-                {start_calc_delete_bitmap_version, max_version}, &incremental_rowsets));
+        auto ret = DORIS_TRY(tmp_tablet->capture_consistent_rowsets_unlocked(
+                {start_calc_delete_bitmap_version, max_version}, CaptureRowsetOps {}));
         DBUG_EXECUTE_IF("CloudSchemaChangeJob::_process_delete_bitmap.after.capture_without_lock",
                         DBUG_BLOCK);
         {
             std::unique_lock wlock(tmp_tablet->get_header_lock());
             tmp_tablet->add_rowsets(_output_rowsets, true, wlock);
         }
-        for (auto rowset : incremental_rowsets) {
+        for (auto rowset : ret.rowsets) {
             RETURN_IF_ERROR(CloudTablet::update_delete_bitmap_without_lock(tmp_tablet, rowset));
         }
     }
@@ -540,15 +539,14 @@ Status CloudSchemaChangeJob::_process_delete_bitmap(int64_t alter_version,
     LOG(INFO) << "alter table for mow table, calculate delete bitmap of "
               << "incremental rowsets with lock, version: " << max_version + 1 << "-"
               << new_max_version << " new_tablet_id: " << _new_tablet->tablet_id();
-    std::vector<RowsetSharedPtr> new_incremental_rowsets;
     if (new_max_version > max_version) {
-        RETURN_IF_ERROR(tmp_tablet->capture_consistent_rowsets_unlocked(
-                {max_version + 1, new_max_version}, &new_incremental_rowsets));
+        auto ret = DORIS_TRY(tmp_tablet->capture_consistent_rowsets_unlocked(
+                {max_version + 1, new_max_version}, CaptureRowsetOps {}));
         {
             std::unique_lock wlock(tmp_tablet->get_header_lock());
             tmp_tablet->add_rowsets(_output_rowsets, true, wlock);
         }
-        for (auto rowset : new_incremental_rowsets) {
+        for (auto rowset : ret.rowsets) {
             RETURN_IF_ERROR(CloudTablet::update_delete_bitmap_without_lock(tmp_tablet, rowset));
         }
     }