[improvement](spill) improve config and fix spill bugs (#33519)

jacktengg · web-flow · commit 9e7204484afc · 2024-04-16T15:46:13.000+08:00
diff --git a/be/src/common/config.cpp b/be/src/common/config.cpp
@@ -1160,13 +1160,13 @@ DEFINE_mDouble(high_disk_avail_level_diff_usages, "0.15");
 DEFINE_Int32(partition_disk_index_lru_size, "10000");
 // limit the storage space that query spill files can use
 DEFINE_String(spill_storage_root_path, "${DORIS_HOME}/storage");
-DEFINE_mInt64(spill_storage_limit, "10737418240"); // 10G
-DEFINE_mInt32(spill_gc_interval_ms, "2000");       // 2s
+DEFINE_String(spill_storage_limit, "20%");   // 20%
+DEFINE_mInt32(spill_gc_interval_ms, "2000"); // 2s
+DEFINE_mInt32(spill_gc_file_count, "2000");
 DEFINE_Int32(spill_io_thread_pool_per_disk_thread_num, "2");
 DEFINE_Int32(spill_io_thread_pool_queue_size, "1024");
 DEFINE_Int32(spill_async_task_thread_pool_thread_num, "2");
 DEFINE_Int32(spill_async_task_thread_pool_queue_size, "1024");
-DEFINE_mInt32(spill_mem_warning_water_mark_multiplier, "2");
 
 DEFINE_mBool(check_segment_when_build_rowset_meta, "false");
 
diff --git a/be/src/common/config.h b/be/src/common/config.h
@@ -1242,13 +1242,20 @@ DECLARE_mDouble(high_disk_avail_level_diff_usages);
 // create tablet in partition random robin idx lru size, default 10000
 DECLARE_Int32(partition_disk_index_lru_size);
 DECLARE_String(spill_storage_root_path);
-DECLARE_mInt64(spill_storage_limit);
+// Spill storage limit specified as number of bytes
+// ('<int>[bB]?'), megabytes ('<float>[mM]'), gigabytes ('<float>[gG]'),
+// or percentage of capaity ('<int>%').
+// Defaults to bytes if no unit is given.
+// Must larger than 0.
+// If specified as percentage, the final limit value is:
+//   disk_capacity_bytes * storage_flood_stage_usage_percent * spill_storage_limit
+DECLARE_String(spill_storage_limit);
 DECLARE_mInt32(spill_gc_interval_ms);
+DECLARE_mInt32(spill_gc_file_count);
 DECLARE_Int32(spill_io_thread_pool_per_disk_thread_num);
 DECLARE_Int32(spill_io_thread_pool_queue_size);
 DECLARE_Int32(spill_async_task_thread_pool_thread_num);
 DECLARE_Int32(spill_async_task_thread_pool_queue_size);
-DECLARE_mInt32(spill_mem_warning_water_mark_multiplier);
 
 DECLARE_mBool(check_segment_when_build_rowset_meta);
 
diff --git a/be/src/olap/storage_engine.h b/be/src/olap/storage_engine.h
@@ -170,7 +170,7 @@ class StorageEngine final : public BaseStorageEngine {
     // get all info of root_path
     Status get_all_data_dir_info(std::vector<DataDirInfo>* data_dir_infos, bool need_update);
 
-    int64_t get_file_or_directory_size(const std::string& file_path);
+    static int64_t get_file_or_directory_size(const std::string& file_path);
 
     // get root path for creating tablet. The returned vector of root path should be round robin,
     // for avoiding that all the tablet would be deployed one disk.
diff --git a/be/src/pipeline/exec/partitioned_aggregation_sink_operator.cpp b/be/src/pipeline/exec/partitioned_aggregation_sink_operator.cpp
@@ -249,14 +249,15 @@ Status PartitionedAggSinkLocalState::revoke_memory(RuntimeState* state) {
 
     auto execution_context = state->get_task_execution_context();
     _shared_state_holder = _shared_state->shared_from_this();
+    auto query_id = state->query_id();
 
     MonotonicStopWatch submit_timer;
     submit_timer.start();
     status = ExecEnv::GetInstance()->spill_stream_mgr()->get_async_task_thread_pool()->submit_func(
-            [this, &parent, state, execution_context, submit_timer] {
+            [this, &parent, state, query_id, execution_context, submit_timer] {
                 auto execution_context_lock = execution_context.lock();
                 if (!execution_context_lock) {
-                    LOG(INFO) << "query " << print_id(state->query_id())
+                    LOG(INFO) << "query " << print_id(query_id)
                               << " execution_context released, maybe query was cancelled.";
                     return Status::Cancelled("Cancelled");
                 }
@@ -267,13 +268,13 @@ Status PartitionedAggSinkLocalState::revoke_memory(RuntimeState* state) {
                     if (!_shared_state->sink_status.ok() || state->is_cancelled()) {
                         if (!_shared_state->sink_status.ok()) {
                             LOG(WARNING)
-                                    << "query " << print_id(state->query_id()) << " agg node "
+                                    << "query " << print_id(query_id) << " agg node "
                                     << Base::_parent->id()
                                     << " revoke_memory error: " << Base::_shared_state->sink_status;
                         }
                         _shared_state->close();
                     } else {
-                        VLOG_DEBUG << "query " << print_id(state->query_id()) << " agg node "
+                        VLOG_DEBUG << "query " << print_id(query_id) << " agg node "
                                    << Base::_parent->id() << " revoke_memory finish"
                                    << ", eos: " << _eos;
                     }
diff --git a/be/src/pipeline/exec/partitioned_aggregation_source_operator.cpp b/be/src/pipeline/exec/partitioned_aggregation_source_operator.cpp
@@ -205,16 +205,17 @@ Status PartitionedAggLocalState::initiate_merge_spill_partition_agg_data(Runtime
 
     auto execution_context = state->get_task_execution_context();
     _shared_state_holder = _shared_state->shared_from_this();
+    auto query_id = state->query_id();
 
     MonotonicStopWatch submit_timer;
     submit_timer.start();
 
     RETURN_IF_ERROR(
             ExecEnv::GetInstance()->spill_stream_mgr()->get_async_task_thread_pool()->submit_func(
-                    [this, state, execution_context, submit_timer] {
+                    [this, state, query_id, execution_context, submit_timer] {
                         auto execution_context_lock = execution_context.lock();
                         if (!execution_context_lock) {
-                            LOG(INFO) << "query " << print_id(state->query_id())
+                            LOG(INFO) << "query " << print_id(query_id)
                                       << " execution_context released, maybe query was cancelled.";
                             // FIXME: return status is meaningless?
                             return Status::Cancelled("Cancelled");
@@ -225,14 +226,14 @@ Status PartitionedAggLocalState::initiate_merge_spill_partition_agg_data(Runtime
                         Defer defer {[&]() {
                             if (!_status.ok() || state->is_cancelled()) {
                                 if (!_status.ok()) {
-                                    LOG(WARNING) << "query " << print_id(state->query_id())
-                                                 << " agg node " << _parent->node_id()
+                                    LOG(WARNING) << "query " << print_id(query_id) << " agg node "
+                                                 << _parent->node_id()
                                                  << " merge spilled agg data error: " << _status;
                                 }
                                 _shared_state->close();
                             } else if (_shared_state->spill_partitions.empty()) {
-                                VLOG_DEBUG << "query " << print_id(state->query_id())
-                                           << " agg node " << _parent->node_id()
+                                VLOG_DEBUG << "query " << print_id(query_id) << " agg node "
+                                           << _parent->node_id()
                                            << " merge spilled agg data finish";
                             }
                             Base::_shared_state->in_mem_shared_state->aggregate_data_container
diff --git a/be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp b/be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp
@@ -493,6 +493,7 @@ Status PartitionedHashJoinProbeOperatorX::prepare(RuntimeState* state) {
     // to avoid prepare _child_x twice
     auto child_x = std::move(_child_x);
     RETURN_IF_ERROR(JoinProbeOperatorX::prepare(state));
+    RETURN_IF_ERROR(vectorized::VExpr::prepare(_output_expr_ctxs, state, *_intermediate_row_desc));
     RETURN_IF_ERROR(_inner_probe_operator->set_child(child_x));
     DCHECK(_build_side_child != nullptr);
     _inner_probe_operator->set_build_side_child(_build_side_child);
@@ -648,6 +649,8 @@ Status PartitionedHashJoinProbeOperatorX::pull(doris::RuntimeState* state,
         }
     }
 
+    const auto partition_index = local_state._partition_cursor;
+    auto& probe_blocks = local_state._probe_blocks[partition_index];
     if (local_state._need_to_setup_internal_operators) {
         *eos = false;
         bool has_data = false;
@@ -659,12 +662,13 @@ Status PartitionedHashJoinProbeOperatorX::pull(doris::RuntimeState* state,
         }
         RETURN_IF_ERROR(_setup_internal_operators(local_state, state));
         local_state._need_to_setup_internal_operators = false;
+        auto& mutable_block = local_state._partitioned_blocks[partition_index];
+        if (mutable_block && !mutable_block->empty()) {
+            probe_blocks.emplace_back(mutable_block->to_block());
+        }
     }
-
-    auto partition_index = local_state._partition_cursor;
-    bool in_mem_eos_;
+    bool in_mem_eos = false;
     auto* runtime_state = local_state._runtime_state.get();
-    auto& probe_blocks = local_state._probe_blocks[partition_index];
     while (_inner_probe_operator->need_more_input_data(runtime_state)) {
         if (probe_blocks.empty()) {
             *eos = false;
@@ -682,14 +686,16 @@ Status PartitionedHashJoinProbeOperatorX::pull(doris::RuntimeState* state,
 
         auto block = std::move(probe_blocks.back());
         probe_blocks.pop_back();
-        RETURN_IF_ERROR(_inner_probe_operator->push(runtime_state, &block, false));
+        if (!block.empty()) {
+            RETURN_IF_ERROR(_inner_probe_operator->push(runtime_state, &block, false));
+        }
     }
 
     RETURN_IF_ERROR(_inner_probe_operator->pull(local_state._runtime_state.get(), output_block,
-                                                &in_mem_eos_));
+                                                &in_mem_eos));
 
     *eos = false;
-    if (in_mem_eos_) {
+    if (in_mem_eos) {
         local_state._partition_cursor++;
         if (local_state._partition_cursor == _partition_count) {
             *eos = true;
@@ -829,6 +835,10 @@ Status PartitionedHashJoinProbeOperatorX::get_block(RuntimeState* state, vectori
             RETURN_IF_ERROR(local_state.finish_spilling(0));
         }
 
+        if (local_state._child_block->rows() == 0 && !local_state._child_eos) {
+            return Status::OK();
+        }
+
         Defer defer([&] { local_state._child_block->clear_column_data(); });
         if (need_to_spill) {
             SCOPED_TIMER(local_state.exec_time_counter());
diff --git a/be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp b/be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp
@@ -93,6 +93,8 @@ Status PartitionedHashJoinSinkLocalState::revoke_memory(RuntimeState* state) {
     DCHECK_EQ(_spilling_streams_count, 0);
 
     if (!_shared_state->need_to_spill) {
+        profile()->add_info_string("Spilled", "true");
+        _shared_state->need_to_spill = true;
         auto& p = _parent->cast<PartitionedHashJoinSinkOperatorX>();
         _shared_state->inner_shared_state->hash_table_variants.reset();
         auto row_desc = p._child_x->row_desc();
@@ -172,7 +174,6 @@ Status PartitionedHashJoinSinkLocalState::revoke_memory(RuntimeState* state) {
     }
 
     if (_spilling_streams_count > 0) {
-        _shared_state->need_to_spill = true;
         std::unique_lock<std::mutex> lock(_spill_lock);
         if (_spilling_streams_count > 0) {
             _dependency->block();
@@ -202,7 +203,8 @@ Status PartitionedHashJoinSinkLocalState::_partition_block(RuntimeState* state,
     SCOPED_TIMER(_partition_shuffle_timer);
     auto* channel_ids = reinterpret_cast<uint64_t*>(_partitioner->get_channel_ids());
     std::vector<uint32_t> partition_indexes[p._partition_count];
-    for (uint32_t i = 0; i != rows; ++i) {
+    DCHECK_LT(begin, end);
+    for (size_t i = begin; i != end; ++i) {
         partition_indexes[channel_ids[i]].emplace_back(i);
     }
 
diff --git a/be/src/pipeline/exec/spill_sort_sink_operator.cpp b/be/src/pipeline/exec/spill_sort_sink_operator.cpp
@@ -105,9 +105,10 @@ Status SpillSortSinkLocalState::setup_in_memory_sort_op(RuntimeState* state) {
     auto* sink_local_state = _runtime_state->get_sink_local_state();
     DCHECK(sink_local_state != nullptr);
 
-    _profile->add_info_string("TOP-N", *sink_local_state->profile()->get_info_string("TOP-N"));
+    RETURN_IF_ERROR(sink_local_state->open(state));
 
-    return sink_local_state->open(state);
+    _profile->add_info_string("TOP-N", *sink_local_state->profile()->get_info_string("TOP-N"));
+    return Status::OK();
 }
 
 SpillSortSinkOperatorX::SpillSortSinkOperatorX(ObjectPool* pool, int operator_id,
@@ -230,17 +231,19 @@ Status SpillSortSinkLocalState::revoke_memory(RuntimeState* state) {
 
     auto execution_context = state->get_task_execution_context();
     _shared_state_holder = _shared_state->shared_from_this();
+    auto query_id = state->query_id();
 
     MonotonicStopWatch submit_timer;
     submit_timer.start();
 
     status = ExecEnv::GetInstance()
                      ->spill_stream_mgr()
                      ->get_spill_io_thread_pool(_spilling_stream->get_spill_root_dir())
-                     ->submit_func([this, state, &parent, execution_context, submit_timer] {
+                     ->submit_func([this, state, query_id, &parent, execution_context,
+                                    submit_timer] {
                          auto execution_context_lock = execution_context.lock();
                          if (!execution_context_lock) {
-                             LOG(INFO) << "query " << print_id(state->query_id())
+                             LOG(INFO) << "query " << print_id(query_id)
                                        << " execution_context released, maybe query was cancelled.";
                              return Status::OK();
                          }
@@ -250,16 +253,14 @@ Status SpillSortSinkLocalState::revoke_memory(RuntimeState* state) {
                          Defer defer {[&]() {
                              if (!_shared_state->sink_status.ok() || state->is_cancelled()) {
                                  if (!_shared_state->sink_status.ok()) {
-                                     LOG(WARNING) << "query " << print_id(state->query_id())
-                                                  << " sort node " << _parent->id()
-                                                  << " revoke memory error: "
+                                     LOG(WARNING) << "query " << print_id(query_id) << " sort node "
+                                                  << _parent->id() << " revoke memory error: "
                                                   << _shared_state->sink_status;
                                  }
                                  _shared_state->close();
                              } else {
-                                 VLOG_DEBUG << "query " << print_id(state->query_id())
-                                            << " sort node " << _parent->id()
-                                            << " revoke memory finish";
+                                 VLOG_DEBUG << "query " << print_id(query_id) << " sort node "
+                                            << _parent->id() << " revoke memory finish";
                              }
 
                              _spilling_stream->end_spill(_shared_state->sink_status);
diff --git a/be/src/pipeline/exec/spill_sort_source_operator.cpp b/be/src/pipeline/exec/spill_sort_source_operator.cpp
@@ -89,14 +89,15 @@ Status SpillSortLocalState::initiate_merge_sort_spill_streams(RuntimeState* stat
 
     auto execution_context = state->get_task_execution_context();
     _shared_state_holder = _shared_state->shared_from_this();
+    auto query_id = state->query_id();
 
     MonotonicStopWatch submit_timer;
     submit_timer.start();
 
-    auto spill_func = [this, state, &parent, execution_context, submit_timer] {
+    auto spill_func = [this, state, query_id, &parent, execution_context, submit_timer] {
         auto execution_context_lock = execution_context.lock();
         if (!execution_context_lock) {
-            LOG(INFO) << "query " << print_id(state->query_id())
+            LOG(INFO) << "query " << print_id(query_id)
                       << " execution_context released, maybe query was cancelled.";
             return Status::OK();
         }
@@ -107,7 +108,7 @@ Status SpillSortLocalState::initiate_merge_sort_spill_streams(RuntimeState* stat
         Defer defer {[&]() {
             if (!_status.ok() || state->is_cancelled()) {
                 if (!_status.ok()) {
-                    LOG(WARNING) << "query " << print_id(state->query_id()) << " sort node "
+                    LOG(WARNING) << "query " << print_id(query_id) << " sort node "
                                  << _parent->node_id() << " merge spill data error: " << _status;
                 }
                 _shared_state->close();
@@ -116,16 +117,16 @@ Status SpillSortLocalState::initiate_merge_sort_spill_streams(RuntimeState* stat
                 }
                 _current_merging_streams.clear();
             } else {
-                VLOG_DEBUG << "query " << print_id(state->query_id()) << " sort node "
-                           << _parent->node_id() << " merge spill data finish";
+                VLOG_DEBUG << "query " << print_id(query_id) << " sort node " << _parent->node_id()
+                           << " merge spill data finish";
             }
             _dependency->Dependency::set_ready();
         }};
         vectorized::Block merge_sorted_block;
         vectorized::SpillStreamSPtr tmp_stream;
         while (!state->is_cancelled()) {
             int max_stream_count = _calc_spill_blocks_to_merge();
-            VLOG_DEBUG << "query " << print_id(state->query_id()) << " sort node " << _parent->id()
+            VLOG_DEBUG << "query " << print_id(query_id) << " sort node " << _parent->id()
                        << " merge spill streams, streams count: "
                        << _shared_state->sorted_streams.size()
                        << ", curren merge max stream count: " << max_stream_count;
diff --git a/be/src/runtime/exec_env_init.cpp b/be/src/runtime/exec_env_init.cpp
@@ -151,6 +151,12 @@ Status ExecEnv::_init(const std::vector<StorePath>& store_paths,
     if (ready()) {
         return Status::OK();
     }
+    std::unordered_map<std::string, std::unique_ptr<vectorized::SpillDataDir>> spill_store_map;
+    for (const auto& spill_path : spill_store_paths) {
+        spill_store_map.emplace(spill_path.path, std::make_unique<vectorized::SpillDataDir>(
+                                                         spill_path.path, spill_path.capacity_bytes,
+                                                         spill_path.storage_medium));
+    }
     init_doris_metrics(store_paths);
     _store_paths = store_paths;
     _tmp_file_dirs = std::make_unique<segment_v2::TmpFileDirs>(_store_paths);
@@ -246,7 +252,7 @@ Status ExecEnv::_init(const std::vector<StorePath>& store_paths,
     _wal_manager = WalManager::create_shared(this, config::group_commit_wal_path);
     _dns_cache = new DNSCache();
     _write_cooldown_meta_executors = std::make_unique<WriteCooldownMetaExecutors>();
-    _spill_stream_mgr = new vectorized::SpillStreamManager(spill_store_paths);
+    _spill_stream_mgr = new vectorized::SpillStreamManager(std::move(spill_store_map));
     _backend_client_cache->init_metrics("backend");
     _frontend_client_cache->init_metrics("frontend");
     _broker_client_cache->init_metrics("broker");
diff --git a/be/src/runtime/workload_group/workload_group_manager.cpp b/be/src/runtime/workload_group/workload_group_manager.cpp
diff --git a/be/src/vec/spill/spill_stream_manager.cpp b/be/src/vec/spill/spill_stream_manager.cpp
diff --git a/be/src/vec/spill/spill_stream_manager.h b/be/src/vec/spill/spill_stream_manager.h
diff --git a/be/src/vec/spill/spill_writer.cpp b/be/src/vec/spill/spill_writer.cpp
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
