Merge branch 'master' into improve-column-ut

be/src/cloud/cloud_cumulative_compaction.cpp

@@ -375,11 +375,9 @@ Status CloudCumulativeCompaction::modify_rowsets() {
 Status CloudCumulativeCompaction::process_old_version_delete_bitmap() {
     // agg previously rowset old version delete bitmap
     std::vector<RowsetSharedPtr> pre_rowsets {};
-    std::vector<std::string> pre_rowset_ids {};
     for (const auto& it : cloud_tablet()->rowset_map()) {
         if (it.first.second < _input_rowsets.front()->start_version()) {
             pre_rowsets.emplace_back(it.second);
-            pre_rowset_ids.emplace_back(it.second->rowset_id().to_string());
         }
     }
     std::sort(pre_rowsets.begin(), pre_rowsets.end(), Rowset::comparator);

be/src/cloud/cloud_engine_calc_delete_bitmap_task.cpp

@@ -227,7 +227,7 @@ Status CloudTabletCalcDeleteBitmapTask::handle() const {
         }
     }
     auto total_update_delete_bitmap_time_us = MonotonicMicros() - t3;
-    LOG(INFO) << "calculate delete bitmap successfully on tablet"
+    LOG(INFO) << "finish calculate delete bitmap on tablet"
               << ", table_id=" << tablet->table_id() << ", transaction_id=" << _transaction_id
               << ", tablet_id=" << tablet->tablet_id()
               << ", get_tablet_time_us=" << get_tablet_time_us

be/src/cloud/cloud_tablet.cpp

@@ -775,7 +775,8 @@ Status CloudTablet::calc_delete_bitmap_for_compaction(
     }
 
     std::unique_ptr<std::map<RowsetSharedPtr, RowLocationPairList>> location_map;
-    if (config::enable_rowid_conversion_correctness_check) {
+    if (config::enable_rowid_conversion_correctness_check &&
+        tablet_schema()->cluster_key_uids().empty()) {
         location_map = std::make_unique<std::map<RowsetSharedPtr, RowLocationPairList>>();
         LOG(INFO) << "Location Map inited succ for tablet:" << tablet_id();
     }

be/src/cloud/pb_convert.cpp

@@ -324,7 +324,7 @@ void doris_tablet_schema_to_cloud(TabletSchemaCloudPB* out, const TabletSchemaPB
     out->set_store_row_column(in.store_row_column());
     out->set_enable_single_replica_compaction(in.enable_single_replica_compaction());
     out->set_skip_write_index_on_load(in.skip_write_index_on_load());
-    out->mutable_cluster_key_idxes()->CopyFrom(in.cluster_key_idxes());
+    out->mutable_cluster_key_uids()->CopyFrom(in.cluster_key_uids());
     out->set_is_dynamic_schema(in.is_dynamic_schema());
     out->mutable_row_store_column_unique_ids()->CopyFrom(in.row_store_column_unique_ids());
     out->set_inverted_index_storage_format(in.inverted_index_storage_format());
@@ -353,7 +353,7 @@ void doris_tablet_schema_to_cloud(TabletSchemaCloudPB* out, TabletSchemaPB&& in)
     out->set_store_row_column(in.store_row_column());
     out->set_enable_single_replica_compaction(in.enable_single_replica_compaction());
     out->set_skip_write_index_on_load(in.skip_write_index_on_load());
-    out->mutable_cluster_key_idxes()->Swap(in.mutable_cluster_key_idxes());
+    out->mutable_cluster_key_uids()->Swap(in.mutable_cluster_key_uids());
     out->set_is_dynamic_schema(in.is_dynamic_schema());
     out->mutable_row_store_column_unique_ids()->Swap(in.mutable_row_store_column_unique_ids());
     out->set_inverted_index_storage_format(in.inverted_index_storage_format());
@@ -395,7 +395,7 @@ void cloud_tablet_schema_to_doris(TabletSchemaPB* out, const TabletSchemaCloudPB
     out->set_store_row_column(in.store_row_column());
     out->set_enable_single_replica_compaction(in.enable_single_replica_compaction());
     out->set_skip_write_index_on_load(in.skip_write_index_on_load());
-    out->mutable_cluster_key_idxes()->CopyFrom(in.cluster_key_idxes());
+    out->mutable_cluster_key_uids()->CopyFrom(in.cluster_key_uids());
     out->set_is_dynamic_schema(in.is_dynamic_schema());
     out->mutable_row_store_column_unique_ids()->CopyFrom(in.row_store_column_unique_ids());
     out->set_inverted_index_storage_format(in.inverted_index_storage_format());
@@ -425,7 +425,7 @@ void cloud_tablet_schema_to_doris(TabletSchemaPB* out, TabletSchemaCloudPB&& in)
     out->set_store_row_column(in.store_row_column());
     out->set_enable_single_replica_compaction(in.enable_single_replica_compaction());
     out->set_skip_write_index_on_load(in.skip_write_index_on_load());
-    out->mutable_cluster_key_idxes()->Swap(in.mutable_cluster_key_idxes());
+    out->mutable_cluster_key_uids()->Swap(in.mutable_cluster_key_uids());
     out->set_is_dynamic_schema(in.is_dynamic_schema());
     out->mutable_row_store_column_unique_ids()->Swap(in.mutable_row_store_column_unique_ids());
     out->set_inverted_index_storage_format(in.inverted_index_storage_format());

be/src/common/config.cpp

@@ -1166,6 +1166,9 @@ DEFINE_mBool(enable_missing_rows_correctness_check, "false");
 // When the number of missing versions is more than this value, do not directly
 // retry the publish and handle it through async publish.
 DEFINE_mInt32(mow_publish_max_discontinuous_version_num, "20");
+// When the size of primary keys in memory exceeds this value, finish current segment
+// and create a new segment, used in compaction. Default 50MB.
+DEFINE_mInt64(mow_primary_key_index_max_size_in_memory, "52428800");
 // When the version is not continuous for MOW table in publish phase and the gap between
 // current txn's publishing version and the max version of the tablet exceeds this value,
 // don't print warning log

be/src/common/config.h

@@ -1236,6 +1236,9 @@ DECLARE_mBool(enable_missing_rows_correctness_check);
 // When the number of missing versions is more than this value, do not directly
 // retry the publish and handle it through async publish.
 DECLARE_mInt32(mow_publish_max_discontinuous_version_num);
+// When the size of primary keys in memory exceeds this value, finish current segment
+// and create a new segment, used in compaction.
+DECLARE_mInt64(mow_primary_key_index_max_size_in_memory);
 // When the version is not continuous for MOW table in publish phase and the gap between
 // current txn's publishing version and the max version of the tablet exceeds this value,
 // don't print warning log

be/src/common/daemon.cpp

@@ -500,15 +500,18 @@ void Daemon::cache_adjust_capacity_thread() {
 void Daemon::cache_prune_stale_thread() {
     int32_t interval = config::cache_periodic_prune_stale_sweep_sec;
     while (!_stop_background_threads_latch.wait_for(std::chrono::seconds(interval))) {
-        if (interval <= 0) {
-            LOG(WARNING) << "config of cache clean interval is illegal: [" << interval
-                         << "], force set to 3600 ";
-            interval = 3600;
+        if (config::cache_periodic_prune_stale_sweep_sec <= 0) {
+            LOG(WARNING) << "config of cache clean interval is: [" << interval
+                         << "], it means the cache prune stale thread is disabled, will wait 3s "
+                            "and check again.";
+            interval = 3;
+            continue;
         }
         if (config::disable_memory_gc) {
             continue;
         }
         CacheManager::instance()->for_each_cache_prune_stale();
+        interval = config::cache_periodic_prune_stale_sweep_sec;
     }
 }
 

be/src/exec/schema_scanner.cpp

@@ -124,7 +124,6 @@ Status SchemaScanner::get_next_block_async(RuntimeState* state) {
                 }
                 SCOPED_ATTACH_TASK(state);
                 _async_thread_running = true;
-                _finish_dependency->block();
                 if (!_opened) {
                     _data_block = vectorized::Block::create_unique();
                     _init_block(_data_block.get());
@@ -140,9 +139,6 @@ Status SchemaScanner::get_next_block_async(RuntimeState* state) {
                 _eos = eos;
                 _async_thread_running = false;
                 _dependency->set_ready();
-                if (eos) {
-                    _finish_dependency->set_ready();
-                }
             }));
     return Status::OK();
 }

be/src/exec/schema_scanner.h

@@ -106,11 +106,7 @@ class SchemaScanner {
     // factory function
     static std::unique_ptr<SchemaScanner> create(TSchemaTableType::type type);
     TSchemaTableType::type type() const { return _schema_table_type; }
-    void set_dependency(std::shared_ptr<pipeline::Dependency> dep,
-                        std::shared_ptr<pipeline::Dependency> fin_dep) {
-        _dependency = dep;
-        _finish_dependency = fin_dep;
-    }
+    void set_dependency(std::shared_ptr<pipeline::Dependency> dep) { _dependency = dep; }
     Status get_next_block_async(RuntimeState* state);
 
 protected:
@@ -139,7 +135,6 @@ class SchemaScanner {
     RuntimeProfile::Counter* _fill_block_timer = nullptr;
 
     std::shared_ptr<pipeline::Dependency> _dependency = nullptr;
-    std::shared_ptr<pipeline::Dependency> _finish_dependency = nullptr;
 
     std::unique_ptr<vectorized::Block> _data_block;
     AtomicStatus _scanner_status;