Merge branch 'master' into varbinary_order_by

Author: zhangstar333

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/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

be/src/cloud/cloud_schema_change_job.cpp

@@ -17,6 +17,7 @@
 
 #include "cloud/cloud_schema_change_job.h"
 
+#include <gen_cpp/Types_types.h>
 #include <gen_cpp/cloud.pb.h>
 
 #include <algorithm>
@@ -32,6 +33,7 @@
 #include "olap/delete_handler.h"
 #include "olap/olap_define.h"
 #include "olap/rowset/beta_rowset.h"
+#include "olap/rowset/rowset.h"
 #include "olap/rowset/rowset_factory.h"
 #include "olap/rowset/segment_v2/inverted_index_desc.h"
 #include "olap/storage_engine.h"
@@ -217,6 +219,20 @@ Status CloudSchemaChangeJob::process_alter_tablet(const TAlterTabletReqV2& reque
 
     SchemaChangeParams sc_params;
 
+    // cache schema change output to file cache
+    std::vector<RowsetSharedPtr> rowsets;
+    rowsets.resize(rs_splits.size());
+    std::transform(rs_splits.begin(), rs_splits.end(), rowsets.begin(),
+                   [](RowSetSplits& split) { return split.rs_reader->rowset(); });
+    sc_params.output_to_file_cache = _should_cache_sc_output(rowsets);
+    if (request.__isset.query_globals && request.__isset.query_options) {
+        sc_params.runtime_state =
+                std::make_shared<RuntimeState>(request.query_options, request.query_globals);
+    } else {
+        // for old version request compatibility
+        sc_params.runtime_state = std::make_shared<RuntimeState>();
+    }
+
     RETURN_IF_ERROR(DescriptorTbl::create(&sc_params.pool, request.desc_tbl, &sc_params.desc_tbl));
     sc_params.ref_rowset_readers.reserve(rs_splits.size());
     for (RowSetSplits& split : rs_splits) {
@@ -269,7 +285,8 @@ Status CloudSchemaChangeJob::_convert_historical_rowsets(const SchemaChangeParam
 
     // Add filter information in change, and filter column information will be set in _parse_request
     // And filter some data every time the row block changes
-    BlockChanger changer(_new_tablet->tablet_schema(), *sc_params.desc_tbl);
+    BlockChanger changer(_new_tablet->tablet_schema(), *sc_params.desc_tbl,
+                         sc_params.runtime_state);
 
     bool sc_sorting = false;
     bool sc_directly = false;
@@ -309,6 +326,8 @@ Status CloudSchemaChangeJob::_convert_historical_rowsets(const SchemaChangeParam
         context.tablet_schema = _new_tablet->tablet_schema();
         context.newest_write_timestamp = rs_reader->newest_write_timestamp();
         context.storage_resource = _cloud_storage_engine.get_storage_resource(sc_params.vault_id);
+        context.write_file_cache = sc_params.output_to_file_cache;
+        context.tablet = _new_tablet;
         if (!context.storage_resource) {
             return Status::InternalError("vault id not found, maybe not sync, vault id {}",
                                          sc_params.vault_id);
@@ -467,7 +486,7 @@ Status CloudSchemaChangeJob::_convert_historical_rowsets(const SchemaChangeParam
         // during double write phase by `CloudMetaMgr::sync_tablet_rowsets` in another thread
         std::unique_lock lock {_new_tablet->get_sync_meta_lock()};
         std::unique_lock wlock(_new_tablet->get_header_lock());
-        _new_tablet->add_rowsets(std::move(_output_rowsets), true, wlock);
+        _new_tablet->add_rowsets(std::move(_output_rowsets), true, wlock, false);
         _new_tablet->set_cumulative_layer_point(_output_cumulative_point);
         _new_tablet->reset_approximate_stats(stats.num_rowsets(), stats.num_segments(),
                                              stats.num_rows(), stats.data_size());
@@ -503,7 +522,7 @@ Status CloudSchemaChangeJob::_process_delete_bitmap(int64_t alter_version,
             std::make_shared<CloudTablet>(_cloud_storage_engine, tmp_meta);
     {
         std::unique_lock wlock(tmp_tablet->get_header_lock());
-        tmp_tablet->add_rowsets(_output_rowsets, true, wlock);
+        tmp_tablet->add_rowsets(_output_rowsets, true, wlock, false);
         // Set alter version to let the tmp_tablet can fill hole rowset greater than alter_version
         tmp_tablet->set_alter_version(alter_version);
     }
@@ -521,7 +540,7 @@ Status CloudSchemaChangeJob::_process_delete_bitmap(int64_t alter_version,
                         DBUG_BLOCK);
         {
             std::unique_lock wlock(tmp_tablet->get_header_lock());
-            tmp_tablet->add_rowsets(_output_rowsets, true, wlock);
+            tmp_tablet->add_rowsets(_output_rowsets, true, wlock, false);
         }
         for (auto rowset : ret.rowsets) {
             RETURN_IF_ERROR(CloudTablet::update_delete_bitmap_without_lock(tmp_tablet, rowset));
@@ -544,7 +563,7 @@ Status CloudSchemaChangeJob::_process_delete_bitmap(int64_t alter_version,
                 {max_version + 1, new_max_version}, CaptureRowsetOps {}));
         {
             std::unique_lock wlock(tmp_tablet->get_header_lock());
-            tmp_tablet->add_rowsets(_output_rowsets, true, wlock);
+            tmp_tablet->add_rowsets(_output_rowsets, true, wlock, false);
         }
         for (auto rowset : ret.rowsets) {
             RETURN_IF_ERROR(CloudTablet::update_delete_bitmap_without_lock(tmp_tablet, rowset));
@@ -595,4 +614,34 @@ void CloudSchemaChangeJob::clean_up_on_failure() {
     }
 }
 
+bool CloudSchemaChangeJob::_should_cache_sc_output(
+        const std::vector<RowsetSharedPtr>& input_rowsets) {
+    int64_t total_size = 0;
+    int64_t cached_index_size = 0;
+    int64_t cached_data_size = 0;
+
+    for (const auto& rs : input_rowsets) {
+        const RowsetMetaSharedPtr& rs_meta = rs->rowset_meta();
+        total_size += rs_meta->total_disk_size();
+        cached_index_size += rs->approximate_cache_index_size();
+        cached_data_size += rs->approximate_cached_data_size();
+    }
+
+    double input_hit_rate = static_cast<double>(cached_index_size + cached_data_size) / total_size;
+
+    LOG(INFO) << "CloudSchemaChangeJob check cache sc output strategy. "
+              << "job_id=" << _job_id << ", input_rowsets_count=" << input_rowsets.size()
+              << ", total_size=" << total_size << ", cached_index_size=" << cached_index_size
+              << ", cached_data_size=" << cached_data_size << ", input_hit_rate=" << input_hit_rate
+              << ", min_hit_ratio_threshold="
+              << config::file_cache_keep_schema_change_output_min_hit_ratio << ", should_cache="
+              << (input_hit_rate > config::file_cache_keep_schema_change_output_min_hit_ratio);
+
+    if (input_hit_rate > config::file_cache_keep_schema_change_output_min_hit_ratio) {
+        return true;
+    }
+
+    return false;
+}
+
 } // namespace doris

be/src/cloud/cloud_schema_change_job.h

@@ -39,13 +39,14 @@ class CloudSchemaChangeJob {
     void clean_up_on_failure();
 
 private:
+    bool _should_cache_sc_output(const std::vector<RowsetSharedPtr>& input_rowsets);
+
     Status _convert_historical_rowsets(const SchemaChangeParams& sc_params,
                                        cloud::TabletJobInfoPB& job);
 
     Status _process_delete_bitmap(int64_t alter_version, int64_t start_calc_delete_bitmap_version,
                                   int64_t initiator, const std::string& vault_id);
 
-private:
     CloudStorageEngine& _cloud_storage_engine;
     std::shared_ptr<CloudTablet> _base_tablet;
     std::shared_ptr<CloudTablet> _new_tablet;

be/src/cloud/cloud_tablet.cpp

@@ -387,7 +387,7 @@ void CloudTablet::add_rowsets(std::vector<RowsetSharedPtr> to_add, bool version_
 
     auto add_rowsets_directly = [=, this](std::vector<RowsetSharedPtr>& rowsets) {
         for (auto& rs : rowsets) {
-            if (version_overlap || warmup_delta_data) {
+            if (warmup_delta_data) {
 #ifndef BE_TEST
                 bool warm_up_state_updated = false;
                 // Warmup rowset data in background

be/src/cloud/config.cpp

@@ -56,6 +56,7 @@ DEFINE_mInt32(check_auto_compaction_interval_seconds, "5");
 DEFINE_mInt32(max_base_compaction_task_num_per_disk, "2");
 DEFINE_mBool(prioritize_query_perf_in_compaction, "false");
 DEFINE_mInt32(compaction_max_rowset_count, "10000");
+DEFINE_mInt64(compaction_txn_max_size_bytes, "7340032"); // 7MB
 
 DEFINE_mInt32(refresh_s3_info_interval_s, "60");
 DEFINE_mInt32(vacuum_stale_rowsets_interval_s, "300");