You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by yi...@apache.org on 2024/02/20 01:19:00 UTC
(doris) 11/11: [pipelineX](refactor) make non-virtual function in Dependency (#31109)
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-2.1
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 6778b4ed9356cde4875ac0dd5129907109191c63
Author: Gabriel <ga...@gmail.com>
AuthorDate: Tue Feb 20 08:57:19 2024 +0800
[pipelineX](refactor) make non-virtual function in Dependency (#31109)
* [pipelineX](refactor) make non-virtual function in Dependency
* update
---
be/src/pipeline/exec/aggregation_sink_operator.cpp | 26 +++-----
be/src/pipeline/exec/aggregation_sink_operator.h | 71 +++++++++++-----------
be/src/pipeline/exec/aggregation_source_operator.h | 11 ----
be/src/pipeline/exec/data_queue.cpp | 20 ++++--
be/src/pipeline/exec/data_queue.h | 6 +-
...istinct_streaming_aggregation_sink_operator.cpp | 3 +-
.../distinct_streaming_aggregation_sink_operator.h | 14 ++++-
.../exec/streaming_aggregation_sink_operator.cpp | 2 +-
.../exec/streaming_aggregation_sink_operator.h | 12 +++-
be/src/pipeline/exec/union_sink_operator.h | 1 -
be/src/pipeline/pipeline_x/dependency.h | 7 ++-
be/src/pipeline/pipeline_x/operator.cpp | 6 ++
12 files changed, 99 insertions(+), 80 deletions(-)
diff --git a/be/src/pipeline/exec/aggregation_sink_operator.cpp b/be/src/pipeline/exec/aggregation_sink_operator.cpp
index c5822591d4a..4e6baf824ea 100644
--- a/be/src/pipeline/exec/aggregation_sink_operator.cpp
+++ b/be/src/pipeline/exec/aggregation_sink_operator.cpp
@@ -118,14 +118,10 @@ Status AggSinkLocalState<DependencyType, Derived>::init(RuntimeState* state,
Base::_shared_state->agg_profile_arena->alloc(p._total_size_of_aggregate_states));
if (p._is_merge) {
- _executor.execute =
- std::bind<Status>(&Derived::_merge_without_key, this, std::placeholders::_1);
+ _executor = std::make_unique<Executor<true, true>>();
} else {
- _executor.execute =
- std::bind<Status>(&Derived::_execute_without_key, this, std::placeholders::_1);
+ _executor = std::make_unique<Executor<true, false>>();
}
-
- _executor.update_memusage = std::bind<void>(&Derived::_update_memusage_without_key, this);
} else {
_init_hash_method(Base::_shared_state->probe_expr_ctxs);
@@ -144,16 +140,11 @@ Status AggSinkLocalState<DependencyType, Derived>::init(RuntimeState* state,
},
_agg_data->method_variant);
if (p._is_merge) {
- _executor.execute = std::bind<Status>(&Derived::_merge_with_serialized_key, this,
- std::placeholders::_1);
+ _executor = std::make_unique<Executor<false, true>>();
} else {
- _executor.execute = std::bind<Status>(&Derived::_execute_with_serialized_key, this,
- std::placeholders::_1);
+ _executor = std::make_unique<Executor<false, false>>();
}
- _executor.update_memusage =
- std::bind<void>(&Derived::_update_memusage_with_serialized_key, this);
-
_should_limit_output = p._limit != -1 && // has limit
(!p._have_conjuncts) && // no having conjunct
p._needs_finalize; // agg's finalize step
@@ -851,9 +842,9 @@ Status AggSinkOperatorX<LocalStateType>::sink(doris::RuntimeState* state,
COUNTER_UPDATE(local_state.rows_input_counter(), (int64_t)in_block->rows());
local_state._shared_state->input_num_rows += in_block->rows();
if (in_block->rows() > 0) {
- RETURN_IF_ERROR(local_state._executor.execute(in_block));
+ RETURN_IF_ERROR(local_state._executor->execute(&local_state, in_block));
RETURN_IF_ERROR(local_state.try_spill_disk());
- local_state._executor.update_memusage();
+ local_state._executor->update_memusage(&local_state);
}
if (source_state == SourceState::FINISHED) {
if (local_state._shared_state->spill_context.has_data) {
@@ -890,6 +881,7 @@ template class AggSinkOperatorX<BlockingAggSinkLocalState>;
template class AggSinkOperatorX<StreamingAggSinkLocalState>;
template class AggSinkOperatorX<DistinctStreamingAggSinkLocalState>;
template class AggSinkLocalState<AggSinkDependency, BlockingAggSinkLocalState>;
-template class AggSinkLocalState<AggSinkDependency, StreamingAggSinkLocalState>;
-template class AggSinkLocalState<AggSinkDependency, DistinctStreamingAggSinkLocalState>;
+template class AggSinkLocalState<StreamingAggSinkDependency, StreamingAggSinkLocalState>;
+template class AggSinkLocalState<DistinctStreamingAggSinkDependency,
+ DistinctStreamingAggSinkLocalState>;
} // namespace doris::pipeline
diff --git a/be/src/pipeline/exec/aggregation_sink_operator.h b/be/src/pipeline/exec/aggregation_sink_operator.h
index 8bdd624ca48..d9014ec1325 100644
--- a/be/src/pipeline/exec/aggregation_sink_operator.h
+++ b/be/src/pipeline/exec/aggregation_sink_operator.h
@@ -51,31 +51,6 @@ public:
AggSinkDependency(int id, int node_id, QueryContext* query_ctx)
: Dependency(id, node_id, "AggSinkDependency", true, query_ctx) {}
~AggSinkDependency() override = default;
-
- void set_ready() override {
- if (_shared_state && _is_streaming_agg_state(_shared_state)) {
- if (((SharedState*)_shared_state)->data_queue->has_enough_space_to_push()) {
- Dependency::set_ready();
- }
- } else {
- Dependency::set_ready();
- }
- }
-
- void block() override {
- if (_is_streaming_agg_state(_shared_state)) {
- if (!((SharedState*)_shared_state)->data_queue->has_enough_space_to_push()) {
- Dependency::block();
- }
- } else {
- Dependency::block();
- }
- }
-
-private:
- static bool _is_streaming_agg_state(const BasicSharedState* shared_state) {
- return ((SharedState*)shared_state)->data_queue != nullptr;
- }
};
template <typename LocalStateType>
@@ -99,6 +74,39 @@ protected:
template <typename LocalStateType>
friend class AggSinkOperatorX;
+ struct ExecutorBase {
+ virtual Status execute(AggSinkLocalState<DependencyType, Derived>* local_state,
+ vectorized::Block* block) = 0;
+ virtual void update_memusage(AggSinkLocalState<DependencyType, Derived>* local_state) = 0;
+ virtual ~ExecutorBase() = default;
+ };
+ template <bool WithoutKey, bool NeedToMerge>
+ struct Executor final : public ExecutorBase {
+ Status execute(AggSinkLocalState<DependencyType, Derived>* local_state,
+ vectorized::Block* block) override {
+ if constexpr (WithoutKey) {
+ if constexpr (NeedToMerge) {
+ return local_state->_merge_without_key(block);
+ } else {
+ return local_state->_execute_without_key(block);
+ }
+ } else {
+ if constexpr (NeedToMerge) {
+ return local_state->_merge_with_serialized_key(block);
+ } else {
+ return local_state->_execute_with_serialized_key(block);
+ }
+ }
+ }
+ void update_memusage(AggSinkLocalState<DependencyType, Derived>* local_state) override {
+ if constexpr (WithoutKey) {
+ local_state->_update_memusage_without_key();
+ } else {
+ local_state->_update_memusage_with_serialized_key();
+ }
+ }
+ };
+
Status _execute_without_key(vectorized::Block* block);
Status _merge_without_key(vectorized::Block* block);
void _update_memusage_without_key();
@@ -322,15 +330,7 @@ protected:
vectorized::AggregatedDataVariants* _agg_data = nullptr;
vectorized::Arena* _agg_arena_pool = nullptr;
- using vectorized_execute = std::function<Status(vectorized::Block* block)>;
- using vectorized_update_memusage = std::function<void()>;
-
- struct executor {
- vectorized_execute execute;
- vectorized_update_memusage update_memusage;
- };
-
- executor _executor;
+ std::unique_ptr<ExecutorBase> _executor = nullptr;
};
class BlockingAggSinkLocalState
@@ -344,6 +344,9 @@ public:
~BlockingAggSinkLocalState() override = default;
};
+class StreamingAggSinkDependency;
+class StreamingAggSinkLocalState;
+
template <typename LocalStateType = BlockingAggSinkLocalState>
class AggSinkOperatorX : public DataSinkOperatorX<LocalStateType> {
public:
diff --git a/be/src/pipeline/exec/aggregation_source_operator.h b/be/src/pipeline/exec/aggregation_source_operator.h
index c03aefcc04b..98fc72ecc97 100644
--- a/be/src/pipeline/exec/aggregation_source_operator.h
+++ b/be/src/pipeline/exec/aggregation_source_operator.h
@@ -54,17 +54,6 @@ public:
AggSourceDependency(int id, int node_id, QueryContext* query_ctx)
: Dependency(id, node_id, "AggSourceDependency", query_ctx) {}
~AggSourceDependency() override = default;
-
- void block() override {
- if (_is_streaming_agg_state()) {
- Dependency::block();
- }
- }
-
-private:
- bool _is_streaming_agg_state() {
- return ((SharedState*)Dependency::_shared_state)->data_queue != nullptr;
- }
};
class AggSourceOperatorX;
diff --git a/be/src/pipeline/exec/data_queue.cpp b/be/src/pipeline/exec/data_queue.cpp
index 5b12c6876d9..d92a8907675 100644
--- a/be/src/pipeline/exec/data_queue.cpp
+++ b/be/src/pipeline/exec/data_queue.cpp
@@ -30,7 +30,7 @@
namespace doris {
namespace pipeline {
-DataQueue::DataQueue(int child_count)
+DataQueue::DataQueue(int child_count, bool is_streaming_agg)
: _queue_blocks_lock(child_count),
_queue_blocks(child_count),
_free_blocks_lock(child_count),
@@ -40,7 +40,8 @@ DataQueue::DataQueue(int child_count)
_is_canceled(child_count),
_cur_bytes_in_queue(child_count),
_cur_blocks_nums_in_queue(child_count),
- _flag_queue_idx(0) {
+ _flag_queue_idx(0),
+ _is_streaming_agg(is_streaming_agg) {
for (int i = 0; i < child_count; ++i) {
_queue_blocks_lock[i].reset(new std::mutex());
_free_blocks_lock[i].reset(new std::mutex());
@@ -73,8 +74,9 @@ void DataQueue::push_free_block(std::unique_ptr<vectorized::Block> block, int ch
}
//use sink to check can_write
-bool DataQueue::has_enough_space_to_push(int child_idx) {
- return _cur_bytes_in_queue[child_idx].load() < MAX_BYTE_OF_QUEUE / 2;
+bool DataQueue::has_enough_space_to_push() {
+ DCHECK(_cur_bytes_in_queue.size() == 1);
+ return _cur_bytes_in_queue[0].load() < MAX_BYTE_OF_QUEUE / 2;
}
//use source to check can_read
@@ -121,7 +123,11 @@ Status DataQueue::get_block_from_queue(std::unique_ptr<vectorized::Block>* outpu
auto old_value = _cur_blocks_total_nums.fetch_sub(1);
if (old_value == 1 && _source_dependency) {
set_source_block();
- _sink_dependencies[_flag_queue_idx]->set_ready();
+ if (_is_streaming_agg && has_enough_space_to_push()) {
+ _sink_dependencies[_flag_queue_idx]->set_ready();
+ } else if (!_is_streaming_agg) {
+ _sink_dependencies[_flag_queue_idx]->set_ready();
+ }
}
} else {
if (_is_finished[_flag_queue_idx]) {
@@ -144,7 +150,9 @@ void DataQueue::push_block(std::unique_ptr<vectorized::Block> block, int child_i
_cur_blocks_total_nums++;
if (_source_dependency) {
set_source_ready();
- _sink_dependencies[child_idx]->block();
+ if (_is_streaming_agg && !has_enough_space_to_push()) {
+ _sink_dependencies[child_idx]->block();
+ }
}
//this only use to record the queue[0] for profile
_max_bytes_in_queue = std::max(_max_bytes_in_queue, _cur_bytes_in_queue[0].load());
diff --git a/be/src/pipeline/exec/data_queue.h b/be/src/pipeline/exec/data_queue.h
index 4eaa768fb8c..5fc58102ac6 100644
--- a/be/src/pipeline/exec/data_queue.h
+++ b/be/src/pipeline/exec/data_queue.h
@@ -35,7 +35,7 @@ class Dependency;
class DataQueue {
public:
//always one is enough, but in union node it's has more children
- DataQueue(int child_count = 1);
+ DataQueue(int child_count = 1, bool is_streaming_agg = true);
~DataQueue() = default;
Status get_block_from_queue(std::unique_ptr<vectorized::Block>* block,
@@ -52,7 +52,7 @@ public:
bool is_finish(int child_idx = 0);
bool is_all_finish();
- bool has_enough_space_to_push(int child_idx = 0);
+ bool has_enough_space_to_push();
bool has_data_or_finished(int child_idx = 0);
bool remaining_has_data();
@@ -107,6 +107,8 @@ private:
std::shared_ptr<Dependency> _source_dependency = nullptr;
std::vector<Dependency*> _sink_dependencies;
SpinLock _source_lock;
+
+ const bool _is_streaming_agg = true;
};
} // namespace doris::pipeline
diff --git a/be/src/pipeline/exec/distinct_streaming_aggregation_sink_operator.cpp b/be/src/pipeline/exec/distinct_streaming_aggregation_sink_operator.cpp
index da4968025af..6efacc12a04 100644
--- a/be/src/pipeline/exec/distinct_streaming_aggregation_sink_operator.cpp
+++ b/be/src/pipeline/exec/distinct_streaming_aggregation_sink_operator.cpp
@@ -97,7 +97,8 @@ OperatorPtr DistinctStreamingAggSinkOperatorBuilder::build_operator() {
DistinctStreamingAggSinkLocalState::DistinctStreamingAggSinkLocalState(
DataSinkOperatorXBase* parent, RuntimeState* state)
- : AggSinkLocalState<AggSinkDependency, DistinctStreamingAggSinkLocalState>(parent, state),
+ : AggSinkLocalState<DistinctStreamingAggSinkDependency, DistinctStreamingAggSinkLocalState>(
+ parent, state),
dummy_mapped_data(std::make_shared<char>('A')) {}
Status DistinctStreamingAggSinkLocalState::_distinct_pre_agg_with_serialized_key(
diff --git a/be/src/pipeline/exec/distinct_streaming_aggregation_sink_operator.h b/be/src/pipeline/exec/distinct_streaming_aggregation_sink_operator.h
index 900aa786634..13f48cad74a 100644
--- a/be/src/pipeline/exec/distinct_streaming_aggregation_sink_operator.h
+++ b/be/src/pipeline/exec/distinct_streaming_aggregation_sink_operator.h
@@ -77,11 +77,21 @@ private:
class DistinctStreamingAggSinkOperatorX;
+class DistinctStreamingAggSinkDependency final : public Dependency {
+public:
+ using SharedState = AggSharedState;
+ DistinctStreamingAggSinkDependency(int id, int node_id, QueryContext* query_ctx)
+ : Dependency(id, node_id, "DistinctStreamingAggSinkDependency", true, query_ctx) {}
+ ~DistinctStreamingAggSinkDependency() override = default;
+};
+
class DistinctStreamingAggSinkLocalState final
- : public AggSinkLocalState<AggSinkDependency, DistinctStreamingAggSinkLocalState> {
+ : public AggSinkLocalState<DistinctStreamingAggSinkDependency,
+ DistinctStreamingAggSinkLocalState> {
public:
using Parent = DistinctStreamingAggSinkOperatorX;
- using Base = AggSinkLocalState<AggSinkDependency, DistinctStreamingAggSinkLocalState>;
+ using Base = AggSinkLocalState<DistinctStreamingAggSinkDependency,
+ DistinctStreamingAggSinkLocalState>;
ENABLE_FACTORY_CREATOR(DistinctStreamingAggSinkLocalState);
DistinctStreamingAggSinkLocalState(DataSinkOperatorXBase* parent, RuntimeState* state);
diff --git a/be/src/pipeline/exec/streaming_aggregation_sink_operator.cpp b/be/src/pipeline/exec/streaming_aggregation_sink_operator.cpp
index 2cc8b9efcaf..41d46e2f3ad 100644
--- a/be/src/pipeline/exec/streaming_aggregation_sink_operator.cpp
+++ b/be/src/pipeline/exec/streaming_aggregation_sink_operator.cpp
@@ -157,7 +157,7 @@ Status StreamingAggSinkLocalState::do_pre_agg(vectorized::Block* input_block,
_num_rows_returned += output_block->rows();
_make_nullable_output_key(output_block);
// COUNTER_SET(_rows_returned_counter, _num_rows_returned);
- _executor.update_memusage();
+ _executor->update_memusage(this);
return Status::OK();
}
diff --git a/be/src/pipeline/exec/streaming_aggregation_sink_operator.h b/be/src/pipeline/exec/streaming_aggregation_sink_operator.h
index 766d2c2fbbe..bd2640517cf 100644
--- a/be/src/pipeline/exec/streaming_aggregation_sink_operator.h
+++ b/be/src/pipeline/exec/streaming_aggregation_sink_operator.h
@@ -74,11 +74,19 @@ private:
class StreamingAggSinkOperatorX;
+class StreamingAggSinkDependency final : public Dependency {
+public:
+ using SharedState = AggSharedState;
+ StreamingAggSinkDependency(int id, int node_id, QueryContext* query_ctx)
+ : Dependency(id, node_id, "StreamingAggSinkDependency", true, query_ctx) {}
+ ~StreamingAggSinkDependency() override = default;
+};
+
class StreamingAggSinkLocalState final
- : public AggSinkLocalState<AggSinkDependency, StreamingAggSinkLocalState> {
+ : public AggSinkLocalState<StreamingAggSinkDependency, StreamingAggSinkLocalState> {
public:
using Parent = StreamingAggSinkOperatorX;
- using Base = AggSinkLocalState<AggSinkDependency, StreamingAggSinkLocalState>;
+ using Base = AggSinkLocalState<StreamingAggSinkDependency, StreamingAggSinkLocalState>;
ENABLE_FACTORY_CREATOR(StreamingAggSinkLocalState);
StreamingAggSinkLocalState(DataSinkOperatorXBase* parent, RuntimeState* state);
~StreamingAggSinkLocalState() override = default;
diff --git a/be/src/pipeline/exec/union_sink_operator.h b/be/src/pipeline/exec/union_sink_operator.h
index ed51e8b7398..9e7e3f8b547 100644
--- a/be/src/pipeline/exec/union_sink_operator.h
+++ b/be/src/pipeline/exec/union_sink_operator.h
@@ -72,7 +72,6 @@ public:
UnionSinkDependency(int id, int node_id, QueryContext* query_ctx)
: Dependency(id, node_id, "UnionSinkDependency", true, query_ctx) {}
~UnionSinkDependency() override = default;
- void block() override {}
};
class UnionSinkOperatorX;
diff --git a/be/src/pipeline/pipeline_x/dependency.h b/be/src/pipeline/pipeline_x/dependency.h
index 11a9cbf39d5..b2ca808e209 100644
--- a/be/src/pipeline/pipeline_x/dependency.h
+++ b/be/src/pipeline/pipeline_x/dependency.h
@@ -116,7 +116,7 @@ public:
// Which dependency current pipeline task is blocked by. `nullptr` if this dependency is ready.
[[nodiscard]] virtual Dependency* is_blocked_by(PipelineXTask* task = nullptr);
// Notify downstream pipeline tasks this dependency is ready.
- virtual void set_ready();
+ void set_ready();
void set_ready_to_read() {
DCHECK(_is_write_dependency) << debug_string();
DCHECK(_shared_state->source_dep != nullptr) << debug_string();
@@ -341,7 +341,7 @@ public:
size_t input_num_rows = 0;
std::vector<vectorized::AggregateDataPtr> values;
std::unique_ptr<vectorized::Arena> agg_profile_arena;
- std::unique_ptr<DataQueue> data_queue = std::make_unique<DataQueue>(1);
+ std::unique_ptr<DataQueue> data_queue = std::make_unique<DataQueue>(1, true);
/// The total size of the row from the aggregate functions.
size_t total_size_of_aggregate_states = 0;
size_t align_aggregate_states = 1;
@@ -402,7 +402,8 @@ public:
struct UnionSharedState : public BasicSharedState {
public:
- UnionSharedState(int child_count = 1) : data_queue(child_count), _child_count(child_count) {};
+ UnionSharedState(int child_count = 1)
+ : data_queue(child_count, false), _child_count(child_count) {};
int child_count() const { return _child_count; }
DataQueue data_queue;
const int _child_count;
diff --git a/be/src/pipeline/pipeline_x/operator.cpp b/be/src/pipeline/pipeline_x/operator.cpp
index 4be062b9e8c..86c7f885792 100644
--- a/be/src/pipeline/pipeline_x/operator.cpp
+++ b/be/src/pipeline/pipeline_x/operator.cpp
@@ -72,6 +72,10 @@
namespace doris::pipeline {
+template class AggSinkLocalState<DistinctStreamingAggSinkDependency,
+ DistinctStreamingAggSinkLocalState>;
+template class AggSinkLocalState<StreamingAggSinkDependency, StreamingAggSinkLocalState>;
+
template <typename DependencyType>
std::string PipelineXLocalState<DependencyType>::debug_string(int indentation_level) const {
fmt::memory_buffer debug_string_buffer;
@@ -652,6 +656,8 @@ template class PipelineXSinkLocalState<SortSinkDependency>;
template class PipelineXSinkLocalState<NestedLoopJoinBuildSinkDependency>;
template class PipelineXSinkLocalState<AnalyticSinkDependency>;
template class PipelineXSinkLocalState<AggSinkDependency>;
+template class PipelineXSinkLocalState<StreamingAggSinkDependency>;
+template class PipelineXSinkLocalState<DistinctStreamingAggSinkDependency>;
template class PipelineXSinkLocalState<FakeDependency>;
template class PipelineXSinkLocalState<UnionSinkDependency>;
template class PipelineXSinkLocalState<PartitionSortSinkDependency>;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org