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/21 06:03:29 UTC
(doris) 14/19: [pipelineX](refactor) refactor streaming agg structure (#31151)
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 8b6d6d01657b020304599d08ac2cc2c5826506d8
Author: Gabriel <ga...@gmail.com>
AuthorDate: Wed Feb 21 10:01:52 2024 +0800
[pipelineX](refactor) refactor streaming agg structure (#31151)
---
be/src/pipeline/exec/aggregation_sink_operator.cpp | 172 +--
be/src/pipeline/exec/aggregation_sink_operator.h | 57 +-
.../pipeline/exec/aggregation_source_operator.cpp | 6 +-
be/src/pipeline/exec/aggregation_source_operator.h | 7 +-
be/src/pipeline/exec/data_queue.cpp | 14 +-
be/src/pipeline/exec/data_queue.h | 4 +-
.../distinct_streaming_aggregation_operator.cpp | 406 ++++++
.../exec/distinct_streaming_aggregation_operator.h | 128 ++
...istinct_streaming_aggregation_sink_operator.cpp | 151 ---
.../distinct_streaming_aggregation_sink_operator.h | 56 -
...tinct_streaming_aggregation_source_operator.cpp | 49 -
...istinct_streaming_aggregation_source_operator.h | 14 -
be/src/pipeline/exec/empty_source_operator.h | 2 +-
be/src/pipeline/exec/join_probe_operator.cpp | 6 +-
be/src/pipeline/exec/join_probe_operator.h | 2 +-
be/src/pipeline/exec/multi_cast_data_stream_sink.h | 2 +-
.../exec/multi_cast_data_stream_source.cpp | 2 +-
.../pipeline/exec/multi_cast_data_stream_source.h | 2 +-
.../exec/nested_loop_join_probe_operator.h | 2 +-
be/src/pipeline/exec/operator.cpp | 2 +-
be/src/pipeline/exec/operator.h | 8 +-
.../exec/streaming_aggregation_operator.cpp | 1302 ++++++++++++++++++++
.../pipeline/exec/streaming_aggregation_operator.h | 231 ++++
.../exec/streaming_aggregation_sink_operator.cpp | 298 -----
.../exec/streaming_aggregation_sink_operator.h | 61 -
.../exec/streaming_aggregation_source_operator.cpp | 39 -
.../exec/streaming_aggregation_source_operator.h | 13 -
be/src/pipeline/pipeline_x/dependency.h | 4 +-
.../local_exchange_source_operator.h | 2 +-
be/src/pipeline/pipeline_x/operator.cpp | 17 +-
be/src/pipeline/pipeline_x/operator.h | 2 +-
.../pipeline_x/pipeline_x_fragment_context.cpp | 39 +-
32 files changed, 2174 insertions(+), 926 deletions(-)
diff --git a/be/src/pipeline/exec/aggregation_sink_operator.cpp b/be/src/pipeline/exec/aggregation_sink_operator.cpp
index 4e6baf824ea..30c4d9d77a3 100644
--- a/be/src/pipeline/exec/aggregation_sink_operator.cpp
+++ b/be/src/pipeline/exec/aggregation_sink_operator.cpp
@@ -48,30 +48,25 @@ OPERATOR_CODE_GENERATOR(AggSinkOperator, StreamingOperator)
/// using the planner's estimated input cardinality and the assumption that input
/// is in a random order. This means that we assume that the reduction factor will
/// increase over time.
-template <typename DependencyType, typename Derived>
-AggSinkLocalState<DependencyType, Derived>::AggSinkLocalState(DataSinkOperatorXBase* parent,
- RuntimeState* state)
+AggSinkLocalState::AggSinkLocalState(DataSinkOperatorXBase* parent, RuntimeState* state)
: Base(parent, state),
_hash_table_compute_timer(nullptr),
_hash_table_input_counter(nullptr),
_build_timer(nullptr),
_expr_timer(nullptr),
- _exec_timer(nullptr),
_serialize_key_timer(nullptr),
_merge_timer(nullptr),
_serialize_data_timer(nullptr),
_deserialize_data_timer(nullptr),
_max_row_size_counter(nullptr) {}
-template <typename DependencyType, typename Derived>
-Status AggSinkLocalState<DependencyType, Derived>::init(RuntimeState* state,
- LocalSinkStateInfo& info) {
+Status AggSinkLocalState::init(RuntimeState* state, LocalSinkStateInfo& info) {
RETURN_IF_ERROR(Base::init(state, info));
SCOPED_TIMER(Base::exec_time_counter());
SCOPED_TIMER(Base::_open_timer);
_agg_data = Base::_shared_state->agg_data.get();
_agg_arena_pool = Base::_shared_state->agg_arena_pool.get();
- auto& p = Base::_parent->template cast<typename Derived::Parent>();
+ auto& p = Base::_parent->template cast<AggSinkOperatorX>();
Base::_shared_state->align_aggregate_states = p._align_aggregate_states;
Base::_shared_state->total_size_of_aggregate_states = p._total_size_of_aggregate_states;
Base::_shared_state->offsets_of_aggregate_states = p._offsets_of_aggregate_states;
@@ -80,9 +75,6 @@ Status AggSinkLocalState<DependencyType, Derived>::init(RuntimeState* state,
for (auto& evaluator : p._aggregate_evaluators) {
Base::_shared_state->aggregate_evaluators.push_back(evaluator->clone(state, p._pool));
}
- if (p._is_streaming) {
- Base::_shared_state->data_queue->set_sink_dependency(Base::_dependency, 0);
- }
Base::_shared_state->probe_expr_ctxs.resize(p._probe_expr_ctxs.size());
for (size_t i = 0; i < Base::_shared_state->probe_expr_ctxs.size(); i++) {
RETURN_IF_ERROR(
@@ -153,8 +145,7 @@ Status AggSinkLocalState<DependencyType, Derived>::init(RuntimeState* state,
return Status::OK();
}
-template <typename DependencyType, typename Derived>
-Status AggSinkLocalState<DependencyType, Derived>::open(RuntimeState* state) {
+Status AggSinkLocalState::open(RuntimeState* state) {
SCOPED_TIMER(Base::exec_time_counter());
SCOPED_TIMER(Base::_open_timer);
RETURN_IF_ERROR(Base::open(state));
@@ -169,9 +160,7 @@ Status AggSinkLocalState<DependencyType, Derived>::open(RuntimeState* state) {
return Status::OK();
}
-template <typename DependencyType, typename Derived>
-Status AggSinkLocalState<DependencyType, Derived>::_create_agg_status(
- vectorized::AggregateDataPtr data) {
+Status AggSinkLocalState::_create_agg_status(vectorized::AggregateDataPtr data) {
auto& shared_state = *Base::_shared_state;
for (int i = 0; i < shared_state.aggregate_evaluators.size(); ++i) {
try {
@@ -188,23 +177,20 @@ Status AggSinkLocalState<DependencyType, Derived>::_create_agg_status(
return Status::OK();
}
-template <typename DependencyType, typename Derived>
-Status AggSinkLocalState<DependencyType, Derived>::_execute_without_key(vectorized::Block* block) {
+Status AggSinkLocalState::_execute_without_key(vectorized::Block* block) {
DCHECK(_agg_data->without_key != nullptr);
SCOPED_TIMER(_build_timer);
for (int i = 0; i < Base::_shared_state->aggregate_evaluators.size(); ++i) {
RETURN_IF_ERROR(Base::_shared_state->aggregate_evaluators[i]->execute_single_add(
block,
- _agg_data->without_key + Base::_parent->template cast<typename Derived::Parent>()
+ _agg_data->without_key + Base::_parent->template cast<AggSinkOperatorX>()
._offsets_of_aggregate_states[i],
_agg_arena_pool));
}
return Status::OK();
}
-template <typename DependencyType, typename Derived>
-Status AggSinkLocalState<DependencyType, Derived>::_merge_with_serialized_key(
- vectorized::Block* block) {
+Status AggSinkLocalState::_merge_with_serialized_key(vectorized::Block* block) {
if (_reach_limit) {
return _merge_with_serialized_key_helper<true, false>(block);
} else {
@@ -212,8 +198,7 @@ Status AggSinkLocalState<DependencyType, Derived>::_merge_with_serialized_key(
}
}
-template <typename DependencyType, typename Derived>
-size_t AggSinkLocalState<DependencyType, Derived>::_memory_usage() const {
+size_t AggSinkLocalState::_memory_usage() const {
size_t usage = 0;
if (_agg_arena_pool) {
usage += _agg_arena_pool->size();
@@ -226,8 +211,7 @@ size_t AggSinkLocalState<DependencyType, Derived>::_memory_usage() const {
return usage;
}
-template <typename DependencyType, typename Derived>
-void AggSinkLocalState<DependencyType, Derived>::_update_memusage_with_serialized_key() {
+void AggSinkLocalState::_update_memusage_with_serialized_key() {
std::visit(
[&](auto&& agg_method) -> void {
auto& data = *agg_method.hash_table;
@@ -251,9 +235,7 @@ void AggSinkLocalState<DependencyType, Derived>::_update_memusage_with_serialize
_agg_data->method_variant);
}
-template <typename DependencyType, typename Derived>
-Status AggSinkLocalState<DependencyType, Derived>::_destroy_agg_status(
- vectorized::AggregateDataPtr data) {
+Status AggSinkLocalState::_destroy_agg_status(vectorized::AggregateDataPtr data) {
auto& shared_state = *Base::_shared_state;
for (int i = 0; i < shared_state.aggregate_evaluators.size(); ++i) {
shared_state.aggregate_evaluators[i]->function()->destroy(
@@ -262,10 +244,8 @@ Status AggSinkLocalState<DependencyType, Derived>::_destroy_agg_status(
return Status::OK();
}
-template <typename DependencyType, typename Derived>
template <bool limit, bool for_spill>
-Status AggSinkLocalState<DependencyType, Derived>::_merge_with_serialized_key_helper(
- vectorized::Block* block) {
+Status AggSinkLocalState::_merge_with_serialized_key_helper(vectorized::Block* block) {
SCOPED_TIMER(_merge_timer);
size_t key_size = Base::_shared_state->probe_expr_ctxs.size();
@@ -312,7 +292,7 @@ Status AggSinkLocalState<DependencyType, Derived>::_merge_with_serialized_key_he
->function()
->deserialize_and_merge_vec_selected(
_places.data(),
- Base::_parent->template cast<typename Derived::Parent>()
+ Base::_parent->template cast<AggSinkOperatorX>()
._offsets_of_aggregate_states[i],
_deserialize_buffer.data(),
(vectorized::ColumnString*)(column.get()), _agg_arena_pool,
@@ -322,7 +302,7 @@ Status AggSinkLocalState<DependencyType, Derived>::_merge_with_serialized_key_he
RETURN_IF_ERROR(
Base::_shared_state->aggregate_evaluators[i]->execute_batch_add_selected(
block,
- Base::_parent->template cast<typename Derived::Parent>()
+ Base::_parent->template cast<AggSinkOperatorX>()
._offsets_of_aggregate_states[i],
_places.data(), _agg_arena_pool));
}
@@ -356,7 +336,7 @@ Status AggSinkLocalState<DependencyType, Derived>::_merge_with_serialized_key_he
->function()
->deserialize_and_merge_vec(
_places.data(),
- Base::_parent->template cast<typename Derived::Parent>()
+ Base::_parent->template cast<AggSinkOperatorX>()
._offsets_of_aggregate_states[i],
_deserialize_buffer.data(),
(vectorized::ColumnString*)(column.get()), _agg_arena_pool,
@@ -365,7 +345,7 @@ Status AggSinkLocalState<DependencyType, Derived>::_merge_with_serialized_key_he
} else {
RETURN_IF_ERROR(Base::_shared_state->aggregate_evaluators[i]->execute_batch_add(
block,
- Base::_parent->template cast<typename Derived::Parent>()
+ Base::_parent->template cast<AggSinkOperatorX>()
._offsets_of_aggregate_states[i],
_places.data(), _agg_arena_pool));
}
@@ -373,7 +353,7 @@ Status AggSinkLocalState<DependencyType, Derived>::_merge_with_serialized_key_he
if (_should_limit_output) {
_reach_limit = _get_hash_table_size() >=
- Base::_parent->template cast<typename Derived::Parent>()._limit;
+ Base::_parent->template cast<AggSinkOperatorX>()._limit;
}
}
@@ -383,9 +363,7 @@ Status AggSinkLocalState<DependencyType, Derived>::_merge_with_serialized_key_he
// We should call this function only at 1st phase.
// 1st phase: is_merge=true, only have one SlotRef.
// 2nd phase: is_merge=false, maybe have multiple exprs.
-template <typename DependencyType, typename Derived>
-int AggSinkLocalState<DependencyType, Derived>::_get_slot_column_id(
- const vectorized::AggFnEvaluator* evaluator) {
+int AggSinkLocalState::_get_slot_column_id(const vectorized::AggFnEvaluator* evaluator) {
auto ctxs = evaluator->input_exprs_ctxs();
CHECK(ctxs.size() == 1 && ctxs[0]->root()->is_slot_ref())
<< "input_exprs_ctxs is invalid, input_exprs_ctx[0]="
@@ -393,8 +371,7 @@ int AggSinkLocalState<DependencyType, Derived>::_get_slot_column_id(
return ((vectorized::VSlotRef*)ctxs[0]->root().get())->column_id();
}
-template <typename DependencyType, typename Derived>
-Status AggSinkLocalState<DependencyType, Derived>::_merge_without_key(vectorized::Block* block) {
+Status AggSinkLocalState::_merge_without_key(vectorized::Block* block) {
SCOPED_TIMER(_merge_timer);
DCHECK(_agg_data->without_key != nullptr);
for (int i = 0; i < Base::_shared_state->aggregate_evaluators.size(); ++i) {
@@ -410,23 +387,21 @@ Status AggSinkLocalState<DependencyType, Derived>::_merge_without_key(vectorized
->function()
->deserialize_and_merge_from_column(
_agg_data->without_key +
- Base::_parent->template cast<typename Derived::Parent>()
+ Base::_parent->template cast<AggSinkOperatorX>()
._offsets_of_aggregate_states[i],
*column, _agg_arena_pool);
} else {
RETURN_IF_ERROR(Base::_shared_state->aggregate_evaluators[i]->execute_single_add(
block,
- _agg_data->without_key +
- Base::_parent->template cast<typename Derived::Parent>()
- ._offsets_of_aggregate_states[i],
+ _agg_data->without_key + Base::_parent->template cast<AggSinkOperatorX>()
+ ._offsets_of_aggregate_states[i],
_agg_arena_pool));
}
}
return Status::OK();
}
-template <typename DependencyType, typename Derived>
-void AggSinkLocalState<DependencyType, Derived>::_update_memusage_without_key() {
+void AggSinkLocalState::_update_memusage_without_key() {
auto arena_memory_usage =
_agg_arena_pool->size() - Base::_shared_state->mem_usage_record.used_in_arena;
Base::_mem_tracker->consume(arena_memory_usage);
@@ -434,9 +409,7 @@ void AggSinkLocalState<DependencyType, Derived>::_update_memusage_without_key()
Base::_shared_state->mem_usage_record.used_in_arena = _agg_arena_pool->size();
}
-template <typename DependencyType, typename Derived>
-Status AggSinkLocalState<DependencyType, Derived>::_execute_with_serialized_key(
- vectorized::Block* block) {
+Status AggSinkLocalState::_execute_with_serialized_key(vectorized::Block* block) {
if (_reach_limit) {
return _execute_with_serialized_key_helper<true>(block);
} else {
@@ -444,10 +417,8 @@ Status AggSinkLocalState<DependencyType, Derived>::_execute_with_serialized_key(
}
}
-template <typename DependencyType, typename Derived>
template <bool limit>
-Status AggSinkLocalState<DependencyType, Derived>::_execute_with_serialized_key_helper(
- vectorized::Block* block) {
+Status AggSinkLocalState::_execute_with_serialized_key_helper(vectorized::Block* block) {
SCOPED_TIMER(_build_timer);
DCHECK(!Base::_shared_state->probe_expr_ctxs.empty());
@@ -478,7 +449,7 @@ Status AggSinkLocalState<DependencyType, Derived>::_execute_with_serialized_key_
RETURN_IF_ERROR(
Base::_shared_state->aggregate_evaluators[i]->execute_batch_add_selected(
block,
- Base::_parent->template cast<typename Derived::Parent>()
+ Base::_parent->template cast<AggSinkOperatorX>()
._offsets_of_aggregate_states[i],
_places.data(), _agg_arena_pool));
}
@@ -488,16 +459,16 @@ Status AggSinkLocalState<DependencyType, Derived>::_execute_with_serialized_key_
for (int i = 0; i < Base::_shared_state->aggregate_evaluators.size(); ++i) {
RETURN_IF_ERROR(Base::_shared_state->aggregate_evaluators[i]->execute_batch_add(
block,
- Base::_parent->template cast<typename Derived::Parent>()
+ Base::_parent->template cast<AggSinkOperatorX>()
._offsets_of_aggregate_states[i],
_places.data(), _agg_arena_pool));
}
if (_should_limit_output) {
_reach_limit = _get_hash_table_size() >=
- Base::_parent->template cast<typename Derived::Parent>()._limit;
+ Base::_parent->template cast<AggSinkOperatorX>()._limit;
if (_reach_limit &&
- Base::_parent->template cast<typename Derived::Parent>()._can_short_circuit) {
+ Base::_parent->template cast<AggSinkOperatorX>()._can_short_circuit) {
Base::_dependency->set_ready_to_read();
return Status::Error<ErrorCode::END_OF_FILE>("");
}
@@ -507,16 +478,14 @@ Status AggSinkLocalState<DependencyType, Derived>::_execute_with_serialized_key_
return Status::OK();
}
-template <typename DependencyType, typename Derived>
-size_t AggSinkLocalState<DependencyType, Derived>::_get_hash_table_size() {
+size_t AggSinkLocalState::_get_hash_table_size() {
return std::visit([&](auto&& agg_method) { return agg_method.hash_table->size(); },
_agg_data->method_variant);
}
-template <typename DependencyType, typename Derived>
-void AggSinkLocalState<DependencyType, Derived>::_emplace_into_hash_table(
- vectorized::AggregateDataPtr* places, vectorized::ColumnRawPtrs& key_columns,
- const size_t num_rows) {
+void AggSinkLocalState::_emplace_into_hash_table(vectorized::AggregateDataPtr* places,
+ vectorized::ColumnRawPtrs& key_columns,
+ const size_t num_rows) {
std::visit(
[&](auto&& agg_method) -> void {
SCOPED_TIMER(_hash_table_compute_timer);
@@ -538,9 +507,9 @@ void AggSinkLocalState<DependencyType, Derived>::_emplace_into_hash_table(
auto creator_for_null_key = [&](auto& mapped) {
mapped = _agg_arena_pool->aligned_alloc(
- Base::_parent->template cast<typename Derived::Parent>()
+ Base::_parent->template cast<AggSinkOperatorX>()
._total_size_of_aggregate_states,
- Base::_parent->template cast<typename Derived::Parent>()
+ Base::_parent->template cast<AggSinkOperatorX>()
._align_aggregate_states);
auto st = _create_agg_status(mapped);
if (!st) {
@@ -558,10 +527,9 @@ void AggSinkLocalState<DependencyType, Derived>::_emplace_into_hash_table(
_agg_data->method_variant);
}
-template <typename DependencyType, typename Derived>
-void AggSinkLocalState<DependencyType, Derived>::_find_in_hash_table(
- vectorized::AggregateDataPtr* places, vectorized::ColumnRawPtrs& key_columns,
- size_t num_rows) {
+void AggSinkLocalState::_find_in_hash_table(vectorized::AggregateDataPtr* places,
+ vectorized::ColumnRawPtrs& key_columns,
+ size_t num_rows) {
std::visit(
[&](auto&& agg_method) -> void {
using HashMethodType = std::decay_t<decltype(agg_method)>;
@@ -583,9 +551,7 @@ void AggSinkLocalState<DependencyType, Derived>::_find_in_hash_table(
_agg_data->method_variant);
}
-template <typename DependencyType, typename Derived>
-void AggSinkLocalState<DependencyType, Derived>::_init_hash_method(
- const vectorized::VExprContextSPtrs& probe_exprs) {
+void AggSinkLocalState::_init_hash_method(const vectorized::VExprContextSPtrs& probe_exprs) {
DCHECK(probe_exprs.size() >= 1);
using Type = vectorized::AggregatedDataVariants::Type;
@@ -641,8 +607,7 @@ void AggSinkLocalState<DependencyType, Derived>::_init_hash_method(
_agg_data->init(
get_hash_key_type_with_phase(
- t,
- !Base::_parent->template cast<typename Derived::Parent>()._is_first_phase),
+ t, !Base::_parent->template cast<AggSinkOperatorX>()._is_first_phase),
is_nullable);
} else {
if (!try_get_hash_map_context_fixed<PHNormalHashMap, HashCRC32,
@@ -653,8 +618,7 @@ void AggSinkLocalState<DependencyType, Derived>::_init_hash_method(
}
}
-template <typename DependencyType, typename Derived>
-Status AggSinkLocalState<DependencyType, Derived>::_reset_hash_table() {
+Status AggSinkLocalState::_reset_hash_table() {
auto& ss = *Base::_shared_state;
return std::visit(
[&](auto&& agg_method) {
@@ -682,18 +646,15 @@ Status AggSinkLocalState<DependencyType, Derived>::_reset_hash_table() {
ss.agg_data->method_variant);
}
-template <typename DependencyType, typename Derived>
-Status AggSinkLocalState<DependencyType, Derived>::try_spill_disk(bool eos) {
- if (Base::_parent->template cast<typename Derived::Parent>()._external_agg_bytes_threshold ==
- 0) {
+Status AggSinkLocalState::try_spill_disk(bool eos) {
+ if (Base::_parent->template cast<AggSinkOperatorX>()._external_agg_bytes_threshold == 0) {
return Status::OK();
}
return std::visit(
[&](auto&& agg_method) -> Status {
auto& hash_table = *agg_method.hash_table;
- if (!eos &&
- _memory_usage() < Base::_parent->template cast<typename Derived::Parent>()
- ._external_agg_bytes_threshold) {
+ if (!eos && _memory_usage() < Base::_parent->template cast<AggSinkOperatorX>()
+ ._external_agg_bytes_threshold) {
return Status::OK();
}
@@ -707,11 +668,9 @@ Status AggSinkLocalState<DependencyType, Derived>::try_spill_disk(bool eos) {
_agg_data->method_variant);
}
-template <typename LocalStateType>
-AggSinkOperatorX<LocalStateType>::AggSinkOperatorX(ObjectPool* pool, int operator_id,
- const TPlanNode& tnode,
- const DescriptorTbl& descs, bool is_streaming)
- : DataSinkOperatorX<LocalStateType>(operator_id, tnode.node_id),
+AggSinkOperatorX::AggSinkOperatorX(ObjectPool* pool, int operator_id, const TPlanNode& tnode,
+ const DescriptorTbl& descs)
+ : DataSinkOperatorX<AggSinkLocalState>(operator_id, tnode.node_id),
_intermediate_tuple_id(tnode.agg_node.intermediate_tuple_id),
_intermediate_tuple_desc(nullptr),
_output_tuple_id(tnode.agg_node.output_tuple_id),
@@ -722,14 +681,12 @@ AggSinkOperatorX<LocalStateType>::AggSinkOperatorX(ObjectPool* pool, int operato
_pool(pool),
_limit(tnode.limit),
_have_conjuncts(tnode.__isset.vconjunct && !tnode.vconjunct.nodes.empty()),
- _is_streaming(is_streaming),
_partition_exprs(tnode.__isset.distribute_expr_lists ? tnode.distribute_expr_lists[0]
: std::vector<TExpr> {}),
_is_colocate(tnode.agg_node.__isset.is_colocate && tnode.agg_node.is_colocate) {}
-template <typename LocalStateType>
-Status AggSinkOperatorX<LocalStateType>::init(const TPlanNode& tnode, RuntimeState* state) {
- RETURN_IF_ERROR(DataSinkOperatorX<LocalStateType>::init(tnode, state));
+Status AggSinkOperatorX::init(const TPlanNode& tnode, RuntimeState* state) {
+ RETURN_IF_ERROR(DataSinkOperatorX<AggSinkLocalState>::init(tnode, state));
// ignore return status for now , so we need to introduce ExecNode::init()
RETURN_IF_ERROR(
vectorized::VExpr::create_expr_trees(tnode.agg_node.grouping_exprs, _probe_expr_ctxs));
@@ -767,13 +724,12 @@ Status AggSinkOperatorX<LocalStateType>::init(const TPlanNode& tnode, RuntimeSta
return Status::OK();
}
-template <typename LocalStateType>
-Status AggSinkOperatorX<LocalStateType>::prepare(RuntimeState* state) {
+Status AggSinkOperatorX::prepare(RuntimeState* state) {
_intermediate_tuple_desc = state->desc_tbl().get_tuple_descriptor(_intermediate_tuple_id);
_output_tuple_desc = state->desc_tbl().get_tuple_descriptor(_output_tuple_id);
DCHECK_EQ(_intermediate_tuple_desc->slots().size(), _output_tuple_desc->slots().size());
RETURN_IF_ERROR(vectorized::VExpr::prepare(
- _probe_expr_ctxs, state, DataSinkOperatorX<LocalStateType>::_child_x->row_desc()));
+ _probe_expr_ctxs, state, DataSinkOperatorX<AggSinkLocalState>::_child_x->row_desc()));
int j = _probe_expr_ctxs.size();
for (int i = 0; i < j; ++i) {
@@ -788,7 +744,7 @@ Status AggSinkOperatorX<LocalStateType>::prepare(RuntimeState* state) {
SlotDescriptor* intermediate_slot_desc = _intermediate_tuple_desc->slots()[j];
SlotDescriptor* output_slot_desc = _output_tuple_desc->slots()[j];
RETURN_IF_ERROR(_aggregate_evaluators[i]->prepare(
- state, DataSinkOperatorX<LocalStateType>::_child_x->row_desc(),
+ state, DataSinkOperatorX<AggSinkLocalState>::_child_x->row_desc(),
intermediate_slot_desc, output_slot_desc));
}
@@ -821,8 +777,7 @@ Status AggSinkOperatorX<LocalStateType>::prepare(RuntimeState* state) {
return Status::OK();
}
-template <typename LocalStateType>
-Status AggSinkOperatorX<LocalStateType>::open(RuntimeState* state) {
+Status AggSinkOperatorX::open(RuntimeState* state) {
RETURN_IF_ERROR(vectorized::VExpr::open(_probe_expr_ctxs, state));
for (int i = 0; i < _aggregate_evaluators.size(); ++i) {
@@ -833,10 +788,8 @@ Status AggSinkOperatorX<LocalStateType>::open(RuntimeState* state) {
return Status::OK();
}
-template <typename LocalStateType>
-Status AggSinkOperatorX<LocalStateType>::sink(doris::RuntimeState* state,
- vectorized::Block* in_block,
- SourceState source_state) {
+Status AggSinkOperatorX::sink(doris::RuntimeState* state, vectorized::Block* in_block,
+ SourceState source_state) {
auto& local_state = get_local_state(state);
SCOPED_TIMER(local_state.exec_time_counter());
COUNTER_UPDATE(local_state.rows_input_counter(), (int64_t)in_block->rows());
@@ -856,8 +809,7 @@ Status AggSinkOperatorX<LocalStateType>::sink(doris::RuntimeState* state,
return Status::OK();
}
-template <typename DependencyType, typename Derived>
-Status AggSinkLocalState<DependencyType, Derived>::close(RuntimeState* state, Status exec_status) {
+Status AggSinkLocalState::close(RuntimeState* state, Status exec_status) {
SCOPED_TIMER(Base::exec_time_counter());
SCOPED_TIMER(Base::_close_timer);
if (Base::_closed) {
@@ -874,14 +826,4 @@ Status AggSinkLocalState<DependencyType, Derived>::close(RuntimeState* state, St
return Base::close(state, exec_status);
}
-class StreamingAggSinkLocalState;
-class DistinctStreamingAggSinkLocalState;
-
-template class AggSinkOperatorX<BlockingAggSinkLocalState>;
-template class AggSinkOperatorX<StreamingAggSinkLocalState>;
-template class AggSinkOperatorX<DistinctStreamingAggSinkLocalState>;
-template class AggSinkLocalState<AggSinkDependency, BlockingAggSinkLocalState>;
-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 d9014ec1325..2cbfe6c94e0 100644
--- a/be/src/pipeline/exec/aggregation_sink_operator.h
+++ b/be/src/pipeline/exec/aggregation_sink_operator.h
@@ -53,13 +53,13 @@ public:
~AggSinkDependency() override = default;
};
-template <typename LocalStateType>
class AggSinkOperatorX;
-template <typename DependencyType, typename Derived>
-class AggSinkLocalState : public PipelineXSinkLocalState<DependencyType> {
+class AggSinkLocalState : public PipelineXSinkLocalState<AggSinkDependency> {
public:
+ ENABLE_FACTORY_CREATOR(AggSinkLocalState);
using Base = PipelineXSinkLocalState<DependencyType>;
+ AggSinkLocalState(DataSinkOperatorXBase* parent, RuntimeState* state);
~AggSinkLocalState() override = default;
Status init(RuntimeState* state, LocalSinkStateInfo& info) override;
@@ -69,21 +69,16 @@ public:
Status try_spill_disk(bool eos = false);
protected:
- AggSinkLocalState(DataSinkOperatorXBase* parent, RuntimeState* state);
-
- 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 Status execute(AggSinkLocalState* local_state, vectorized::Block* block) = 0;
+ virtual void update_memusage(AggSinkLocalState* 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 {
+ Status execute(AggSinkLocalState* local_state, vectorized::Block* block) override {
if constexpr (WithoutKey) {
if constexpr (NeedToMerge) {
return local_state->_merge_without_key(block);
@@ -98,7 +93,7 @@ protected:
}
}
}
- void update_memusage(AggSinkLocalState<DependencyType, Derived>* local_state) override {
+ void update_memusage(AggSinkLocalState* local_state) override {
if constexpr (WithoutKey) {
local_state->_update_memusage_without_key();
} else {
@@ -309,7 +304,6 @@ protected:
RuntimeProfile::Counter* _hash_table_input_counter = nullptr;
RuntimeProfile::Counter* _build_timer = nullptr;
RuntimeProfile::Counter* _expr_timer = nullptr;
- RuntimeProfile::Counter* _exec_timer = nullptr;
RuntimeProfile::Counter* _build_table_convert_timer = nullptr;
RuntimeProfile::Counter* _serialize_key_timer = nullptr;
RuntimeProfile::Counter* _merge_timer = nullptr;
@@ -333,29 +327,14 @@ protected:
std::unique_ptr<ExecutorBase> _executor = nullptr;
};
-class BlockingAggSinkLocalState
- : public AggSinkLocalState<AggSinkDependency, BlockingAggSinkLocalState> {
-public:
- ENABLE_FACTORY_CREATOR(BlockingAggSinkLocalState);
- using Parent = AggSinkOperatorX<BlockingAggSinkLocalState>;
-
- BlockingAggSinkLocalState(DataSinkOperatorXBase* parent, RuntimeState* state)
- : AggSinkLocalState<AggSinkDependency, BlockingAggSinkLocalState>(parent, state) {}
- ~BlockingAggSinkLocalState() override = default;
-};
-
-class StreamingAggSinkDependency;
-class StreamingAggSinkLocalState;
-
-template <typename LocalStateType = BlockingAggSinkLocalState>
-class AggSinkOperatorX : public DataSinkOperatorX<LocalStateType> {
+class AggSinkOperatorX final : public DataSinkOperatorX<AggSinkLocalState> {
public:
AggSinkOperatorX(ObjectPool* pool, int operator_id, const TPlanNode& tnode,
- const DescriptorTbl& descs, bool is_streaming = false);
+ const DescriptorTbl& descs);
~AggSinkOperatorX() override = default;
Status init(const TDataSink& tsink) override {
return Status::InternalError("{} should not init with TPlanNode",
- DataSinkOperatorX<LocalStateType>::_name);
+ DataSinkOperatorX<AggSinkLocalState>::_name);
}
Status init(const TPlanNode& tnode, RuntimeState* state) override;
@@ -368,25 +347,22 @@ public:
DataDistribution required_data_distribution() const override {
if (_probe_expr_ctxs.empty()) {
- return _needs_finalize || DataSinkOperatorX<LocalStateType>::_child_x
+ return _needs_finalize || DataSinkOperatorX<AggSinkLocalState>::_child_x
->ignore_data_distribution()
? DataDistribution(ExchangeType::PASSTHROUGH)
- : DataSinkOperatorX<LocalStateType>::required_data_distribution();
+ : DataSinkOperatorX<AggSinkLocalState>::required_data_distribution();
}
return _is_colocate ? DataDistribution(ExchangeType::BUCKET_HASH_SHUFFLE, _partition_exprs)
: DataDistribution(ExchangeType::HASH_SHUFFLE, _partition_exprs);
}
- using DataSinkOperatorX<LocalStateType>::id;
- using DataSinkOperatorX<LocalStateType>::operator_id;
- using DataSinkOperatorX<LocalStateType>::get_local_state;
+ using DataSinkOperatorX<AggSinkLocalState>::id;
+ using DataSinkOperatorX<AggSinkLocalState>::operator_id;
+ using DataSinkOperatorX<AggSinkLocalState>::get_local_state;
protected:
- using LocalState = LocalStateType;
- template <typename DependencyType, typename Derived>
+ using LocalState = AggSinkLocalState;
friend class AggSinkLocalState;
- friend class StreamingAggSinkLocalState;
- friend class DistinctStreamingAggSinkLocalState;
std::vector<vectorized::AggFnEvaluator*> _aggregate_evaluators;
bool _can_short_circuit = false;
@@ -415,7 +391,6 @@ protected:
size_t _spill_partition_count_bits;
int64_t _limit; // -1: no limit
bool _have_conjuncts;
- const bool _is_streaming;
const std::vector<TExpr> _partition_exprs;
const bool _is_colocate;
diff --git a/be/src/pipeline/exec/aggregation_source_operator.cpp b/be/src/pipeline/exec/aggregation_source_operator.cpp
index 7e2fca9bc67..5d27c2d33a2 100644
--- a/be/src/pipeline/exec/aggregation_source_operator.cpp
+++ b/be/src/pipeline/exec/aggregation_source_operator.cpp
@@ -50,9 +50,6 @@ Status AggLocalState::init(RuntimeState* state, LocalStateInfo& info) {
_serialize_data_timer = ADD_TIMER(profile(), "SerializeDataTime");
_hash_table_size_counter = ADD_COUNTER(profile(), "HashTableSize", TUnit::UNIT);
auto& p = _parent->template cast<AggSourceOperatorX>();
- if (p._is_streaming) {
- _shared_state->data_queue->set_source_dependency(info.dependency);
- }
if (p._without_key) {
if (p._needs_finalize) {
_executor.get_result = std::bind<Status>(&AggLocalState::_get_without_key_result, this,
@@ -529,9 +526,8 @@ Status AggLocalState::_get_without_key_result(RuntimeState* state, vectorized::B
}
AggSourceOperatorX::AggSourceOperatorX(ObjectPool* pool, const TPlanNode& tnode, int operator_id,
- const DescriptorTbl& descs, bool is_streaming)
+ const DescriptorTbl& descs)
: Base(pool, tnode, operator_id, descs),
- _is_streaming(is_streaming),
_needs_finalize(tnode.agg_node.need_finalize),
_without_key(tnode.agg_node.grouping_exprs.empty()) {}
diff --git a/be/src/pipeline/exec/aggregation_source_operator.h b/be/src/pipeline/exec/aggregation_source_operator.h
index 98fc72ecc97..d6aa19e57fd 100644
--- a/be/src/pipeline/exec/aggregation_source_operator.h
+++ b/be/src/pipeline/exec/aggregation_source_operator.h
@@ -72,10 +72,6 @@ public:
protected:
friend class AggSourceOperatorX;
- friend class StreamingAggSourceOperatorX;
- friend class StreamingAggSinkOperatorX;
- friend class DistinctStreamingAggSourceOperatorX;
- friend class DistinctStreamingAggSinkOperatorX;
Status _get_without_key_result(RuntimeState* state, vectorized::Block* block,
SourceState& source_state);
@@ -133,7 +129,7 @@ class AggSourceOperatorX : public OperatorX<AggLocalState> {
public:
using Base = OperatorX<AggLocalState>;
AggSourceOperatorX(ObjectPool* pool, const TPlanNode& tnode, int operator_id,
- const DescriptorTbl& descs, bool is_streaming = false);
+ const DescriptorTbl& descs);
~AggSourceOperatorX() = default;
Status get_block(RuntimeState* state, vectorized::Block* block,
@@ -143,7 +139,6 @@ public:
private:
friend class AggLocalState;
- const bool _is_streaming;
bool _needs_finalize;
bool _without_key;
diff --git a/be/src/pipeline/exec/data_queue.cpp b/be/src/pipeline/exec/data_queue.cpp
index d92a8907675..06c16c7dfa6 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, bool is_streaming_agg)
+DataQueue::DataQueue(int child_count)
: _queue_blocks_lock(child_count),
_queue_blocks(child_count),
_free_blocks_lock(child_count),
@@ -40,8 +40,7 @@ DataQueue::DataQueue(int child_count, bool is_streaming_agg)
_is_canceled(child_count),
_cur_bytes_in_queue(child_count),
_cur_blocks_nums_in_queue(child_count),
- _flag_queue_idx(0),
- _is_streaming_agg(is_streaming_agg) {
+ _flag_queue_idx(0) {
for (int i = 0; i < child_count; ++i) {
_queue_blocks_lock[i].reset(new std::mutex());
_free_blocks_lock[i].reset(new std::mutex());
@@ -123,11 +122,7 @@ 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();
- 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();
- }
+ _sink_dependencies[_flag_queue_idx]->set_ready();
}
} else {
if (_is_finished[_flag_queue_idx]) {
@@ -150,9 +145,6 @@ void DataQueue::push_block(std::unique_ptr<vectorized::Block> block, int child_i
_cur_blocks_total_nums++;
if (_source_dependency) {
set_source_ready();
- 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 5fc58102ac6..144fe62728c 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, bool is_streaming_agg = true);
+ DataQueue(int child_count = 1);
~DataQueue() = default;
Status get_block_from_queue(std::unique_ptr<vectorized::Block>* block,
@@ -107,8 +107,6 @@ 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_operator.cpp b/be/src/pipeline/exec/distinct_streaming_aggregation_operator.cpp
new file mode 100644
index 00000000000..36d5065087a
--- /dev/null
+++ b/be/src/pipeline/exec/distinct_streaming_aggregation_operator.cpp
@@ -0,0 +1,406 @@
+// 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 "distinct_streaming_aggregation_operator.h"
+
+#include <gen_cpp/Metrics_types.h>
+
+#include <memory>
+#include <utility>
+
+#include "common/compiler_util.h" // IWYU pragma: keep
+
+namespace doris {
+class ExecNode;
+class RuntimeState;
+} // namespace doris
+
+namespace doris::pipeline {
+
+DistinctStreamingAggLocalState::DistinctStreamingAggLocalState(RuntimeState* state,
+ OperatorXBase* parent)
+ : PipelineXLocalState<FakeDependency>(state, parent),
+ dummy_mapped_data(std::make_shared<char>('A')),
+ _agg_arena_pool(std::make_unique<vectorized::Arena>()),
+ _agg_data(std::make_unique<vectorized::AggregatedDataVariants>()),
+ _agg_profile_arena(std::make_unique<vectorized::Arena>()),
+ _child_block(vectorized::Block::create_unique()),
+ _child_source_state(SourceState::DEPEND_ON_SOURCE),
+ _aggregated_block(vectorized::Block::create_unique()) {}
+
+Status DistinctStreamingAggLocalState::init(RuntimeState* state, LocalStateInfo& info) {
+ RETURN_IF_ERROR(Base::init(state, info));
+ SCOPED_TIMER(Base::exec_time_counter());
+ SCOPED_TIMER(Base::_open_timer);
+ auto& p = Base::_parent->template cast<DistinctStreamingAggOperatorX>();
+ for (auto& evaluator : p._aggregate_evaluators) {
+ _aggregate_evaluators.push_back(evaluator->clone(state, p._pool));
+ }
+ _probe_expr_ctxs.resize(p._probe_expr_ctxs.size());
+ for (size_t i = 0; i < _probe_expr_ctxs.size(); i++) {
+ RETURN_IF_ERROR(p._probe_expr_ctxs[i]->clone(state, _probe_expr_ctxs[i]));
+ }
+
+ _build_timer = ADD_TIMER(Base::profile(), "BuildTime");
+ _exec_timer = ADD_TIMER(Base::profile(), "ExecTime");
+ _hash_table_compute_timer = ADD_TIMER(Base::profile(), "HashTableComputeTime");
+ _hash_table_emplace_timer = ADD_TIMER(Base::profile(), "HashTableEmplaceTime");
+ _hash_table_input_counter = ADD_COUNTER(Base::profile(), "HashTableInputCount", TUnit::UNIT);
+
+ if (_probe_expr_ctxs.empty()) {
+ _agg_data->without_key = reinterpret_cast<vectorized::AggregateDataPtr>(
+ _agg_profile_arena->alloc(p._total_size_of_aggregate_states));
+ } else {
+ _init_hash_method(_probe_expr_ctxs);
+ }
+ return Status::OK();
+}
+
+void DistinctStreamingAggLocalState::_init_hash_method(
+ const vectorized::VExprContextSPtrs& probe_exprs) {
+ DCHECK(probe_exprs.size() >= 1);
+
+ using Type = vectorized::AggregatedDataVariants::Type;
+ Type t(Type::serialized);
+
+ if (probe_exprs.size() == 1) {
+ auto is_nullable = probe_exprs[0]->root()->is_nullable();
+ PrimitiveType type = probe_exprs[0]->root()->result_type();
+ switch (type) {
+ case TYPE_TINYINT:
+ case TYPE_BOOLEAN:
+ case TYPE_SMALLINT:
+ case TYPE_INT:
+ case TYPE_FLOAT:
+ case TYPE_DATEV2:
+ case TYPE_BIGINT:
+ case TYPE_DOUBLE:
+ case TYPE_DATE:
+ case TYPE_DATETIME:
+ case TYPE_DATETIMEV2:
+ case TYPE_LARGEINT:
+ case TYPE_DECIMALV2:
+ case TYPE_DECIMAL32:
+ case TYPE_DECIMAL64:
+ case TYPE_DECIMAL128I: {
+ size_t size = get_primitive_type_size(type);
+ if (size == 1) {
+ t = Type::int8_key;
+ } else if (size == 2) {
+ t = Type::int16_key;
+ } else if (size == 4) {
+ t = Type::int32_key;
+ } else if (size == 8) {
+ t = Type::int64_key;
+ } else if (size == 16) {
+ t = Type::int128_key;
+ } else {
+ throw Exception(ErrorCode::INTERNAL_ERROR,
+ "meet invalid type size, size={}, type={}", size,
+ type_to_string(type));
+ }
+ break;
+ }
+ case TYPE_CHAR:
+ case TYPE_VARCHAR:
+ case TYPE_STRING: {
+ t = Type::string_key;
+ break;
+ }
+ default:
+ t = Type::serialized;
+ }
+
+ _agg_data->init(get_hash_key_type_with_phase(
+ t, !Base::_parent->template cast<DistinctStreamingAggOperatorX>()
+ ._is_first_phase),
+ is_nullable);
+ } else {
+ if (!try_get_hash_map_context_fixed<PHNormalHashMap, HashCRC32,
+ vectorized::AggregateDataPtr>(_agg_data->method_variant,
+ probe_exprs)) {
+ _agg_data->init(Type::serialized);
+ }
+ }
+}
+
+Status DistinctStreamingAggLocalState::_distinct_pre_agg_with_serialized_key(
+ doris::vectorized::Block* in_block, doris::vectorized::Block* out_block) {
+ SCOPED_TIMER(_build_timer);
+ DCHECK(!_probe_expr_ctxs.empty());
+
+ size_t key_size = _probe_expr_ctxs.size();
+ vectorized::ColumnRawPtrs key_columns(key_size);
+ {
+ SCOPED_TIMER(_expr_timer);
+ for (size_t i = 0; i < key_size; ++i) {
+ int result_column_id = -1;
+ RETURN_IF_ERROR(_probe_expr_ctxs[i]->execute(in_block, &result_column_id));
+ in_block->get_by_position(result_column_id).column =
+ in_block->get_by_position(result_column_id)
+ .column->convert_to_full_column_if_const();
+ key_columns[i] = in_block->get_by_position(result_column_id).column.get();
+ }
+ }
+
+ int rows = in_block->rows();
+ _distinct_row.clear();
+ _distinct_row.reserve(rows);
+
+ RETURN_IF_CATCH_EXCEPTION(
+ _emplace_into_hash_table_to_distinct(_distinct_row, key_columns, rows));
+
+ bool mem_reuse = _parent->cast<DistinctStreamingAggOperatorX>()._make_nullable_keys.empty() &&
+ out_block->mem_reuse();
+ if (mem_reuse) {
+ for (int i = 0; i < key_size; ++i) {
+ auto dst = out_block->get_by_position(i).column->assume_mutable();
+ key_columns[i]->append_data_by_selector(dst, _distinct_row);
+ }
+ } else {
+ vectorized::ColumnsWithTypeAndName columns_with_schema;
+ for (int i = 0; i < key_size; ++i) {
+ auto distinct_column = key_columns[i]->clone_empty();
+ key_columns[i]->append_data_by_selector(distinct_column, _distinct_row);
+ columns_with_schema.emplace_back(std::move(distinct_column),
+ _probe_expr_ctxs[i]->root()->data_type(),
+ _probe_expr_ctxs[i]->root()->expr_name());
+ }
+ out_block->swap(vectorized::Block(columns_with_schema));
+ }
+ return Status::OK();
+}
+
+void DistinctStreamingAggLocalState::_make_nullable_output_key(vectorized::Block* block) {
+ if (block->rows() != 0) {
+ for (auto cid : Base::_parent->cast<DistinctStreamingAggOperatorX>()._make_nullable_keys) {
+ block->get_by_position(cid).column = make_nullable(block->get_by_position(cid).column);
+ block->get_by_position(cid).type = make_nullable(block->get_by_position(cid).type);
+ }
+ }
+}
+
+void DistinctStreamingAggLocalState::_emplace_into_hash_table_to_distinct(
+ vectorized::IColumn::Selector& distinct_row, vectorized::ColumnRawPtrs& key_columns,
+ const size_t num_rows) {
+ std::visit(
+ [&](auto&& agg_method) -> void {
+ SCOPED_TIMER(_hash_table_compute_timer);
+ using HashMethodType = std::decay_t<decltype(agg_method)>;
+ using AggState = typename HashMethodType::State;
+ AggState state(key_columns);
+ agg_method.init_serialized_keys(key_columns, num_rows);
+ size_t row = 0;
+ auto creator = [&](const auto& ctor, auto& key, auto& origin) {
+ HashMethodType::try_presis_key(key, origin, _arena);
+ ctor(key, dummy_mapped_data.get());
+ distinct_row.push_back(row);
+ };
+ auto creator_for_null_key = [&](auto& mapped) {
+ mapped = dummy_mapped_data.get();
+ distinct_row.push_back(row);
+ };
+
+ SCOPED_TIMER(_hash_table_emplace_timer);
+ for (; row < num_rows; ++row) {
+ agg_method.lazy_emplace(state, row, creator, creator_for_null_key);
+ }
+
+ COUNTER_UPDATE(_hash_table_input_counter, num_rows);
+ },
+ _agg_data->method_variant);
+}
+
+DistinctStreamingAggOperatorX::DistinctStreamingAggOperatorX(ObjectPool* pool, int operator_id,
+ const TPlanNode& tnode,
+ const DescriptorTbl& descs)
+ : StatefulOperatorX<DistinctStreamingAggLocalState>(pool, tnode, operator_id, descs),
+ _intermediate_tuple_id(tnode.agg_node.intermediate_tuple_id),
+ _intermediate_tuple_desc(nullptr),
+ _output_tuple_id(tnode.agg_node.output_tuple_id),
+ _output_tuple_desc(nullptr),
+ _needs_finalize(tnode.agg_node.need_finalize),
+ _is_first_phase(tnode.agg_node.__isset.is_first_phase && tnode.agg_node.is_first_phase),
+ _partition_exprs(tnode.__isset.distribute_expr_lists ? tnode.distribute_expr_lists[0]
+ : std::vector<TExpr> {}),
+ _is_colocate(tnode.agg_node.__isset.is_colocate && tnode.agg_node.is_colocate) {
+ if (tnode.agg_node.__isset.use_streaming_preaggregation) {
+ _is_streaming_preagg = tnode.agg_node.use_streaming_preaggregation;
+ if (_is_streaming_preagg) {
+ DCHECK(!tnode.agg_node.grouping_exprs.empty()) << "Streaming preaggs do grouping";
+ DCHECK(_limit == -1) << "Preaggs have no limits";
+ }
+ } else {
+ _is_streaming_preagg = false;
+ }
+}
+
+Status DistinctStreamingAggOperatorX::init(const TPlanNode& tnode, RuntimeState* state) {
+ RETURN_IF_ERROR(StatefulOperatorX<DistinctStreamingAggLocalState>::init(tnode, state));
+ // ignore return status for now , so we need to introduce ExecNode::init()
+ RETURN_IF_ERROR(
+ vectorized::VExpr::create_expr_trees(tnode.agg_node.grouping_exprs, _probe_expr_ctxs));
+
+ // init aggregate functions
+ _aggregate_evaluators.reserve(tnode.agg_node.aggregate_functions.size());
+
+ TSortInfo dummy;
+ for (int i = 0; i < tnode.agg_node.aggregate_functions.size(); ++i) {
+ vectorized::AggFnEvaluator* evaluator = nullptr;
+ RETURN_IF_ERROR(vectorized::AggFnEvaluator::create(
+ _pool, tnode.agg_node.aggregate_functions[i],
+ tnode.agg_node.__isset.agg_sort_infos ? tnode.agg_node.agg_sort_infos[i] : dummy,
+ &evaluator));
+ _aggregate_evaluators.push_back(evaluator);
+ }
+
+ _op_name = "DISTINCT_STREAMING_AGGREGATION_OPERATOR";
+ return Status::OK();
+}
+
+Status DistinctStreamingAggOperatorX::prepare(RuntimeState* state) {
+ RETURN_IF_ERROR(StatefulOperatorX<DistinctStreamingAggLocalState>::prepare(state));
+ _intermediate_tuple_desc = state->desc_tbl().get_tuple_descriptor(_intermediate_tuple_id);
+ _output_tuple_desc = state->desc_tbl().get_tuple_descriptor(_output_tuple_id);
+ DCHECK_EQ(_intermediate_tuple_desc->slots().size(), _output_tuple_desc->slots().size());
+ RETURN_IF_ERROR(vectorized::VExpr::prepare(_probe_expr_ctxs, state, _child_x->row_desc()));
+
+ int j = _probe_expr_ctxs.size();
+ for (int i = 0; i < j; ++i) {
+ auto nullable_output = _output_tuple_desc->slots()[i]->is_nullable();
+ auto nullable_input = _probe_expr_ctxs[i]->root()->is_nullable();
+ if (nullable_output != nullable_input) {
+ DCHECK(nullable_output);
+ _make_nullable_keys.emplace_back(i);
+ }
+ }
+ for (int i = 0; i < _aggregate_evaluators.size(); ++i, ++j) {
+ SlotDescriptor* intermediate_slot_desc = _intermediate_tuple_desc->slots()[j];
+ SlotDescriptor* output_slot_desc = _output_tuple_desc->slots()[j];
+ RETURN_IF_ERROR(_aggregate_evaluators[i]->prepare(
+ state, _child_x->row_desc(), intermediate_slot_desc, output_slot_desc));
+ }
+
+ for (size_t i = 0; i < _aggregate_evaluators.size(); ++i) {
+ const auto& agg_function = _aggregate_evaluators[i]->function();
+ _total_size_of_aggregate_states += agg_function->size_of_data();
+
+ // If not the last aggregate_state, we need pad it so that next aggregate_state will be aligned.
+ if (i + 1 < _aggregate_evaluators.size()) {
+ size_t alignment_of_next_state =
+ _aggregate_evaluators[i + 1]->function()->align_of_data();
+ if ((alignment_of_next_state & (alignment_of_next_state - 1)) != 0) {
+ return Status::RuntimeError("Logical error: align_of_data is not 2^N");
+ }
+
+ /// Extend total_size to next alignment requirement
+ /// Add padding by rounding up 'total_size_of_aggregate_states' to be a multiplier of alignment_of_next_state.
+ _total_size_of_aggregate_states =
+ (_total_size_of_aggregate_states + alignment_of_next_state - 1) /
+ alignment_of_next_state * alignment_of_next_state;
+ }
+ }
+
+ return Status::OK();
+}
+
+Status DistinctStreamingAggOperatorX::open(RuntimeState* state) {
+ RETURN_IF_ERROR(StatefulOperatorX<DistinctStreamingAggLocalState>::open(state));
+ RETURN_IF_ERROR(vectorized::VExpr::open(_probe_expr_ctxs, state));
+
+ for (int i = 0; i < _aggregate_evaluators.size(); ++i) {
+ RETURN_IF_ERROR(_aggregate_evaluators[i]->open(state));
+ _aggregate_evaluators[i]->set_version(state->be_exec_version());
+ }
+
+ return Status::OK();
+}
+
+Status DistinctStreamingAggOperatorX::push(RuntimeState* state, vectorized::Block* in_block,
+ SourceState source_state) const {
+ auto& local_state = get_local_state(state);
+ local_state._input_num_rows += in_block->rows();
+ Status ret = Status::OK();
+ if (in_block->rows() > 0) {
+ RETURN_IF_ERROR(local_state._distinct_pre_agg_with_serialized_key(
+ in_block, local_state._aggregated_block.get()));
+
+ // get enough data or reached limit rows, need push block to queue
+ if (_limit != -1 &&
+ (local_state._aggregated_block->rows() + local_state._output_distinct_rows) >= _limit) {
+ auto limit_rows = _limit - local_state._output_distinct_rows;
+ local_state._aggregated_block->set_num_rows(limit_rows);
+ local_state._output_distinct_rows += limit_rows;
+ } else if (local_state._aggregated_block->rows() >= state->batch_size()) {
+ local_state._output_distinct_rows += local_state._aggregated_block->rows();
+ }
+ }
+
+ // reach limit or source finish
+ if ((UNLIKELY(source_state == SourceState::FINISHED)) ||
+ (_limit != -1 && local_state._output_distinct_rows >= _limit)) {
+ local_state._output_distinct_rows += local_state._aggregated_block->rows();
+ return Status::OK(); // need given finish signal
+ }
+ return Status::OK();
+}
+
+Status DistinctStreamingAggOperatorX::pull(RuntimeState* state, vectorized::Block* block,
+ SourceState& source_state) const {
+ auto& local_state = get_local_state(state);
+ if (!local_state._aggregated_block->empty()) {
+ block->swap(*local_state._aggregated_block);
+ local_state._aggregated_block->clear_column_data(block->columns());
+ }
+
+ local_state._make_nullable_output_key(block);
+ if (_is_streaming_preagg == false) {
+ // dispose the having clause, should not be execute in prestreaming agg
+ RETURN_IF_ERROR(
+ vectorized::VExprContext::filter_block(_conjuncts, block, block->columns()));
+ }
+
+ if (UNLIKELY(local_state._child_source_state == SourceState::FINISHED ||
+ (_limit != -1 && local_state._output_distinct_rows >= _limit))) {
+ source_state = SourceState::FINISHED;
+ } else {
+ source_state = SourceState::DEPEND_ON_SOURCE;
+ }
+ return Status::OK();
+}
+
+bool DistinctStreamingAggOperatorX::need_more_input_data(RuntimeState* state) const {
+ auto& local_state = get_local_state(state);
+ return local_state._aggregated_block->empty() &&
+ local_state._child_source_state != SourceState::FINISHED &&
+ (_limit == -1 || local_state._output_distinct_rows < _limit);
+}
+
+Status DistinctStreamingAggLocalState::close(RuntimeState* state) {
+ if (_closed) {
+ return Status::OK();
+ }
+ SCOPED_TIMER(Base::exec_time_counter());
+ SCOPED_TIMER(Base::_close_timer);
+ if (Base::_closed) {
+ return Status::OK();
+ }
+ _aggregated_block->clear();
+ return Base::close(state);
+}
+
+} // namespace doris::pipeline
diff --git a/be/src/pipeline/exec/distinct_streaming_aggregation_operator.h b/be/src/pipeline/exec/distinct_streaming_aggregation_operator.h
new file mode 100644
index 00000000000..564c29b3e2a
--- /dev/null
+++ b/be/src/pipeline/exec/distinct_streaming_aggregation_operator.h
@@ -0,0 +1,128 @@
+// 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.
+
+#pragma once
+
+#include <stdint.h>
+
+#include <cstdint>
+#include <memory>
+
+#include "common/status.h"
+#include "pipeline/pipeline_x/dependency.h"
+#include "pipeline/pipeline_x/operator.h"
+#include "util/runtime_profile.h"
+#include "vec/core/block.h"
+
+namespace doris {
+class ExecNode;
+class RuntimeState;
+
+namespace pipeline {
+
+class DistinctStreamingAggOperatorX;
+
+class DistinctStreamingAggLocalState final : public PipelineXLocalState<FakeDependency> {
+public:
+ using Parent = DistinctStreamingAggOperatorX;
+ using Base = PipelineXLocalState<FakeDependency>;
+ ENABLE_FACTORY_CREATOR(DistinctStreamingAggLocalState);
+ DistinctStreamingAggLocalState(RuntimeState* state, OperatorXBase* parent);
+
+ Status init(RuntimeState* state, LocalStateInfo& info) override;
+ Status close(RuntimeState* state) override;
+
+private:
+ friend class DistinctStreamingAggOperatorX;
+ template <typename LocalStateType>
+ friend class StatefulOperatorX;
+ Status _distinct_pre_agg_with_serialized_key(vectorized::Block* in_block,
+ vectorized::Block* out_block);
+ void _init_hash_method(const vectorized::VExprContextSPtrs& probe_exprs);
+ void _emplace_into_hash_table_to_distinct(vectorized::IColumn::Selector& distinct_row,
+ vectorized::ColumnRawPtrs& key_columns,
+ const size_t num_rows);
+ void _make_nullable_output_key(vectorized::Block* block);
+
+ std::shared_ptr<char> dummy_mapped_data;
+ vectorized::IColumn::Selector _distinct_row;
+ vectorized::Arena _arena;
+ int64_t _output_distinct_rows = 0;
+ size_t _input_num_rows = 0;
+
+ std::unique_ptr<vectorized::Arena> _agg_arena_pool = nullptr;
+ vectorized::AggregatedDataVariantsUPtr _agg_data = nullptr;
+ std::vector<vectorized::AggFnEvaluator*> _aggregate_evaluators;
+ // group by k1,k2
+ vectorized::VExprContextSPtrs _probe_expr_ctxs;
+ std::unique_ptr<vectorized::Arena> _agg_profile_arena = nullptr;
+ std::unique_ptr<vectorized::Block> _child_block = nullptr;
+ SourceState _child_source_state;
+ std::unique_ptr<vectorized::Block> _aggregated_block = nullptr;
+
+ RuntimeProfile::Counter* _build_timer = nullptr;
+ RuntimeProfile::Counter* _expr_timer = nullptr;
+ RuntimeProfile::Counter* _hash_table_compute_timer = nullptr;
+ RuntimeProfile::Counter* _hash_table_emplace_timer = nullptr;
+ RuntimeProfile::Counter* _hash_table_input_counter = nullptr;
+};
+
+class DistinctStreamingAggOperatorX final
+ : public StatefulOperatorX<DistinctStreamingAggLocalState> {
+public:
+ DistinctStreamingAggOperatorX(ObjectPool* pool, int operator_id, const TPlanNode& tnode,
+ const DescriptorTbl& descs);
+ Status init(const TPlanNode& tnode, RuntimeState* state) override;
+ Status prepare(RuntimeState* state) override;
+ Status open(RuntimeState* state) override;
+ Status pull(RuntimeState* state, vectorized::Block* block,
+ SourceState& source_state) const override;
+ Status push(RuntimeState* state, vectorized::Block* input_block,
+ SourceState source_state) const override;
+ bool need_more_input_data(RuntimeState* state) const override;
+
+ DataDistribution required_data_distribution() const override {
+ if (_needs_finalize) {
+ return _is_colocate
+ ? DataDistribution(ExchangeType::BUCKET_HASH_SHUFFLE, _partition_exprs)
+ : DataDistribution(ExchangeType::HASH_SHUFFLE, _partition_exprs);
+ }
+ return StatefulOperatorX<DistinctStreamingAggLocalState>::required_data_distribution();
+ }
+
+private:
+ friend class DistinctStreamingAggLocalState;
+ TupleId _intermediate_tuple_id;
+ TupleDescriptor* _intermediate_tuple_desc = nullptr;
+
+ TupleId _output_tuple_id;
+ TupleDescriptor* _output_tuple_desc = nullptr;
+ const bool _needs_finalize;
+ const bool _is_first_phase;
+ const std::vector<TExpr> _partition_exprs;
+ const bool _is_colocate;
+ // group by k1,k2
+ vectorized::VExprContextSPtrs _probe_expr_ctxs;
+ std::vector<vectorized::AggFnEvaluator*> _aggregate_evaluators;
+ std::vector<size_t> _make_nullable_keys;
+ /// The total size of the row from the aggregate functions.
+ size_t _total_size_of_aggregate_states = 0;
+ bool _is_streaming_preagg = false;
+};
+
+} // namespace pipeline
+} // namespace doris
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 6efacc12a04..765055a004b 100644
--- a/be/src/pipeline/exec/distinct_streaming_aggregation_sink_operator.cpp
+++ b/be/src/pipeline/exec/distinct_streaming_aggregation_sink_operator.cpp
@@ -95,155 +95,4 @@ OperatorPtr DistinctStreamingAggSinkOperatorBuilder::build_operator() {
return std::make_shared<DistinctStreamingAggSinkOperator>(this, _node, _data_queue);
}
-DistinctStreamingAggSinkLocalState::DistinctStreamingAggSinkLocalState(
- DataSinkOperatorXBase* parent, RuntimeState* state)
- : AggSinkLocalState<DistinctStreamingAggSinkDependency, DistinctStreamingAggSinkLocalState>(
- parent, state),
- dummy_mapped_data(std::make_shared<char>('A')) {}
-
-Status DistinctStreamingAggSinkLocalState::_distinct_pre_agg_with_serialized_key(
- doris::vectorized::Block* in_block, doris::vectorized::Block* out_block) {
- SCOPED_TIMER(_build_timer);
- DCHECK(!_shared_state->probe_expr_ctxs.empty());
-
- size_t key_size = _shared_state->probe_expr_ctxs.size();
- vectorized::ColumnRawPtrs key_columns(key_size);
- {
- SCOPED_TIMER(_expr_timer);
- for (size_t i = 0; i < key_size; ++i) {
- int result_column_id = -1;
- RETURN_IF_ERROR(
- _shared_state->probe_expr_ctxs[i]->execute(in_block, &result_column_id));
- in_block->get_by_position(result_column_id).column =
- in_block->get_by_position(result_column_id)
- .column->convert_to_full_column_if_const();
- key_columns[i] = in_block->get_by_position(result_column_id).column.get();
- }
- }
-
- int rows = in_block->rows();
- _distinct_row.clear();
- _distinct_row.reserve(rows);
-
- RETURN_IF_CATCH_EXCEPTION(
- _emplace_into_hash_table_to_distinct(_distinct_row, key_columns, rows));
-
- bool mem_reuse = _shared_state->make_nullable_keys.empty() && out_block->mem_reuse();
- if (mem_reuse) {
- for (int i = 0; i < key_size; ++i) {
- auto dst = out_block->get_by_position(i).column->assume_mutable();
- key_columns[i]->append_data_by_selector(dst, _distinct_row);
- }
- } else {
- vectorized::ColumnsWithTypeAndName columns_with_schema;
- for (int i = 0; i < key_size; ++i) {
- auto distinct_column = key_columns[i]->clone_empty();
- key_columns[i]->append_data_by_selector(distinct_column, _distinct_row);
- columns_with_schema.emplace_back(
- std::move(distinct_column),
- _shared_state->probe_expr_ctxs[i]->root()->data_type(),
- _shared_state->probe_expr_ctxs[i]->root()->expr_name());
- }
- out_block->swap(vectorized::Block(columns_with_schema));
- }
- return Status::OK();
-}
-
-void DistinctStreamingAggSinkLocalState::_emplace_into_hash_table_to_distinct(
- vectorized::IColumn::Selector& distinct_row, vectorized::ColumnRawPtrs& key_columns,
- const size_t num_rows) {
- std::visit(
- [&](auto&& agg_method) -> void {
- SCOPED_TIMER(_hash_table_compute_timer);
- using HashMethodType = std::decay_t<decltype(agg_method)>;
- using AggState = typename HashMethodType::State;
- AggState state(key_columns);
- agg_method.init_serialized_keys(key_columns, num_rows);
- size_t row = 0;
- auto creator = [&](const auto& ctor, auto& key, auto& origin) {
- HashMethodType::try_presis_key(key, origin, _arena);
- ctor(key, dummy_mapped_data.get());
- distinct_row.push_back(row);
- };
- auto creator_for_null_key = [&](auto& mapped) {
- mapped = dummy_mapped_data.get();
- distinct_row.push_back(row);
- };
-
- SCOPED_TIMER(_hash_table_emplace_timer);
- for (; row < num_rows; ++row) {
- agg_method.lazy_emplace(state, row, creator, creator_for_null_key);
- }
-
- COUNTER_UPDATE(_hash_table_input_counter, num_rows);
- },
- _agg_data->method_variant);
-}
-
-DistinctStreamingAggSinkOperatorX::DistinctStreamingAggSinkOperatorX(ObjectPool* pool,
- int operator_id,
- const TPlanNode& tnode,
- const DescriptorTbl& descs)
- : AggSinkOperatorX<DistinctStreamingAggSinkLocalState>(pool, operator_id, tnode, descs,
- true) {}
-
-Status DistinctStreamingAggSinkOperatorX::init(const TPlanNode& tnode, RuntimeState* state) {
- RETURN_IF_ERROR(AggSinkOperatorX<DistinctStreamingAggSinkLocalState>::init(tnode, state));
- _name = "DISTINCT_STREAMING_AGGREGATION_SINK_OPERATOR";
- return Status::OK();
-}
-
-Status DistinctStreamingAggSinkOperatorX::sink(RuntimeState* state, vectorized::Block* in_block,
- SourceState source_state) {
- auto& local_state = get_local_state(state);
- SCOPED_TIMER(local_state.exec_time_counter());
- COUNTER_UPDATE(local_state.rows_input_counter(), (int64_t)in_block->rows());
- local_state._shared_state->input_num_rows += in_block->rows();
- Status ret = Status::OK();
- if (in_block && in_block->rows() > 0) {
- if (local_state._output_block == nullptr) {
- local_state._output_block = local_state._shared_state->data_queue->get_free_block();
- }
- RETURN_IF_ERROR(local_state._distinct_pre_agg_with_serialized_key(
- in_block, local_state._output_block.get()));
-
- // get enough data or reached limit rows, need push block to queue
- if (_limit != -1 &&
- (local_state._output_block->rows() + local_state._output_distinct_rows) >= _limit) {
- auto limit_rows = _limit - local_state._output_distinct_rows;
- local_state._output_block->set_num_rows(limit_rows);
- local_state._output_distinct_rows += limit_rows;
- local_state._shared_state->data_queue->push_block(std::move(local_state._output_block));
- } else if (local_state._output_block->rows() >= state->batch_size()) {
- local_state._output_distinct_rows += local_state._output_block->rows();
- local_state._shared_state->data_queue->push_block(std::move(local_state._output_block));
- }
- }
-
- // reach limit or source finish
- if ((UNLIKELY(source_state == SourceState::FINISHED)) ||
- (_limit != -1 && local_state._output_distinct_rows >= _limit)) {
- if (local_state._output_block != nullptr) { //maybe the last block with eos
- local_state._output_distinct_rows += local_state._output_block->rows();
- local_state._shared_state->data_queue->push_block(std::move(local_state._output_block));
- }
- local_state._shared_state->data_queue->set_finish();
- return Status::Error<ErrorCode::END_OF_FILE>(""); // need given finish signal
- }
- return Status::OK();
-}
-
-Status DistinctStreamingAggSinkLocalState::close(RuntimeState* state, Status exec_status) {
- if (_closed) {
- return Status::OK();
- }
- SCOPED_TIMER(exec_time_counter());
- SCOPED_TIMER(_close_timer);
- if (_shared_state->data_queue && !_shared_state->data_queue->is_finish()) {
- // finish should be set, if not set here means error.
- _shared_state->data_queue->set_canceled();
- }
- return Base::close(state, exec_status);
-}
-
} // namespace doris::pipeline
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 13f48cad74a..c872a2b299e 100644
--- a/be/src/pipeline/exec/distinct_streaming_aggregation_sink_operator.h
+++ b/be/src/pipeline/exec/distinct_streaming_aggregation_sink_operator.h
@@ -75,61 +75,5 @@ private:
std::unique_ptr<vectorized::Block> _output_block = vectorized::Block::create_unique();
};
-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<DistinctStreamingAggSinkDependency,
- DistinctStreamingAggSinkLocalState> {
-public:
- using Parent = DistinctStreamingAggSinkOperatorX;
- using Base = AggSinkLocalState<DistinctStreamingAggSinkDependency,
- DistinctStreamingAggSinkLocalState>;
- ENABLE_FACTORY_CREATOR(DistinctStreamingAggSinkLocalState);
- DistinctStreamingAggSinkLocalState(DataSinkOperatorXBase* parent, RuntimeState* state);
-
- Status close(RuntimeState* state, Status exec_status) override;
- Status _distinct_pre_agg_with_serialized_key(vectorized::Block* in_block,
- vectorized::Block* out_block);
-
-private:
- friend class DistinctStreamingAggSinkOperatorX;
- void _emplace_into_hash_table_to_distinct(vectorized::IColumn::Selector& distinct_row,
- vectorized::ColumnRawPtrs& key_columns,
- const size_t num_rows);
-
- std::unique_ptr<vectorized::Block> _output_block = vectorized::Block::create_unique();
- std::shared_ptr<char> dummy_mapped_data;
- vectorized::IColumn::Selector _distinct_row;
- vectorized::Arena _arena;
- int64_t _output_distinct_rows = 0;
-};
-
-class DistinctStreamingAggSinkOperatorX final
- : public AggSinkOperatorX<DistinctStreamingAggSinkLocalState> {
-public:
- DistinctStreamingAggSinkOperatorX(ObjectPool* pool, int operator_id, const TPlanNode& tnode,
- const DescriptorTbl& descs);
- Status init(const TPlanNode& tnode, RuntimeState* state) override;
- Status sink(RuntimeState* state, vectorized::Block* in_block,
- SourceState source_state) override;
-
- DataDistribution required_data_distribution() const override {
- if (_needs_finalize) {
- return _is_colocate
- ? DataDistribution(ExchangeType::BUCKET_HASH_SHUFFLE, _partition_exprs)
- : DataDistribution(ExchangeType::HASH_SHUFFLE, _partition_exprs);
- }
- return DataSinkOperatorX<DistinctStreamingAggSinkLocalState>::required_data_distribution();
- }
-};
-
} // namespace pipeline
} // namespace doris
diff --git a/be/src/pipeline/exec/distinct_streaming_aggregation_source_operator.cpp b/be/src/pipeline/exec/distinct_streaming_aggregation_source_operator.cpp
index 722cc5b7956..fb653bdcbdb 100644
--- a/be/src/pipeline/exec/distinct_streaming_aggregation_source_operator.cpp
+++ b/be/src/pipeline/exec/distinct_streaming_aggregation_source_operator.cpp
@@ -88,54 +88,5 @@ OperatorPtr DistinctStreamingAggSourceOperatorBuilder::build_operator() {
return std::make_shared<DistinctStreamingAggSourceOperator>(this, _node, _data_queue);
}
-DistinctStreamingAggSourceOperatorX::DistinctStreamingAggSourceOperatorX(ObjectPool* pool,
- const TPlanNode& tnode,
- int operator_id,
- const DescriptorTbl& descs)
- : Base(pool, tnode, operator_id, descs, true) {
- if (tnode.agg_node.__isset.use_streaming_preaggregation) {
- _is_streaming_preagg = tnode.agg_node.use_streaming_preaggregation;
- if (_is_streaming_preagg) {
- DCHECK(!tnode.agg_node.grouping_exprs.empty()) << "Streaming preaggs do grouping";
- DCHECK(_limit == -1) << "Preaggs have no limits";
- }
- } else {
- _is_streaming_preagg = false;
- }
-}
-
-Status DistinctStreamingAggSourceOperatorX::get_block(RuntimeState* state, vectorized::Block* block,
- SourceState& source_state) {
- auto& local_state = get_local_state(state);
- SCOPED_TIMER(local_state.exec_time_counter());
- std::unique_ptr<vectorized::Block> agg_block;
- RETURN_IF_ERROR(local_state._shared_state->data_queue->get_block_from_queue(&agg_block));
- if (agg_block != nullptr) {
- block->swap(*agg_block);
- agg_block->clear_column_data(block->columns());
- local_state._shared_state->data_queue->push_free_block(std::move(agg_block));
- }
-
- local_state._make_nullable_output_key(block);
- if (_is_streaming_preagg == false) {
- // dispose the having clause, should not be execute in prestreaming agg
- RETURN_IF_ERROR(
- vectorized::VExprContext::filter_block(_conjuncts, block, block->columns()));
- }
-
- if (UNLIKELY(local_state._shared_state->data_queue->data_exhausted())) {
- source_state = SourceState::FINISHED;
- } else {
- source_state = SourceState::DEPEND_ON_SOURCE;
- }
- return Status::OK();
-}
-
-Status DistinctStreamingAggSourceOperatorX::init(const TPlanNode& tnode, RuntimeState* state) {
- RETURN_IF_ERROR(Base::init(tnode, state));
- _op_name = "DISTINCT_STREAMING_AGGREGATION_OPERATOR";
- return Status::OK();
-}
-
} // namespace pipeline
} // namespace doris
diff --git a/be/src/pipeline/exec/distinct_streaming_aggregation_source_operator.h b/be/src/pipeline/exec/distinct_streaming_aggregation_source_operator.h
index 81edf18567f..0283e31c41f 100644
--- a/be/src/pipeline/exec/distinct_streaming_aggregation_source_operator.h
+++ b/be/src/pipeline/exec/distinct_streaming_aggregation_source_operator.h
@@ -64,19 +64,5 @@ private:
std::shared_ptr<DataQueue> _data_queue;
};
-class DistinctStreamingAggSourceOperatorX final : public AggSourceOperatorX {
-public:
- using Base = AggSourceOperatorX;
- DistinctStreamingAggSourceOperatorX(ObjectPool* pool, const TPlanNode& tnode, int operator_id,
- const DescriptorTbl& descs);
- ~DistinctStreamingAggSourceOperatorX() = default;
-
- Status init(const TPlanNode& tnode, RuntimeState* state) override;
-
- Status get_block(RuntimeState* state, vectorized::Block* block,
- SourceState& source_state) override;
- bool _is_streaming_preagg = false;
-};
-
} // namespace pipeline
} // namespace doris
diff --git a/be/src/pipeline/exec/empty_source_operator.h b/be/src/pipeline/exec/empty_source_operator.h
index d76d3689253..b352f7cab29 100644
--- a/be/src/pipeline/exec/empty_source_operator.h
+++ b/be/src/pipeline/exec/empty_source_operator.h
@@ -46,7 +46,7 @@ public:
OperatorPtr build_operator() override;
- const RowDescriptor& row_desc() override { return _row_descriptor; }
+ const RowDescriptor& row_desc() const override { return _row_descriptor; }
private:
RowDescriptor _row_descriptor;
diff --git a/be/src/pipeline/exec/join_probe_operator.cpp b/be/src/pipeline/exec/join_probe_operator.cpp
index 3e71ec41dd4..c041fed5d7a 100644
--- a/be/src/pipeline/exec/join_probe_operator.cpp
+++ b/be/src/pipeline/exec/join_probe_operator.cpp
@@ -112,13 +112,15 @@ Status JoinProbeLocalState<DependencyType, Derived>::_build_output_block(
if (rows != 0) {
auto& mutable_columns = mutable_block.mutable_columns();
if (_output_expr_ctxs.empty()) {
- DCHECK(mutable_columns.size() == p.row_desc().num_materialized_slots());
+ DCHECK(mutable_columns.size() == p.row_desc().num_materialized_slots())
+ << mutable_columns.size() << " " << p.row_desc().num_materialized_slots();
for (int i = 0; i < mutable_columns.size(); ++i) {
insert_column_datas(mutable_columns[i], origin_block->get_by_position(i).column,
rows);
}
} else {
- DCHECK(mutable_columns.size() == p.row_desc().num_materialized_slots());
+ DCHECK(mutable_columns.size() == p.row_desc().num_materialized_slots())
+ << mutable_columns.size() << " " << p.row_desc().num_materialized_slots();
SCOPED_TIMER(Base::_projection_timer);
for (int i = 0; i < mutable_columns.size(); ++i) {
auto result_column_id = -1;
diff --git a/be/src/pipeline/exec/join_probe_operator.h b/be/src/pipeline/exec/join_probe_operator.h
index 5dc6089c348..6f379bc9749 100644
--- a/be/src/pipeline/exec/join_probe_operator.h
+++ b/be/src/pipeline/exec/join_probe_operator.h
@@ -73,7 +73,7 @@ public:
Status init(const TPlanNode& tnode, RuntimeState* state) override;
Status open(doris::RuntimeState* state) override;
- [[nodiscard]] const RowDescriptor& row_desc() override { return *_output_row_desc; }
+ [[nodiscard]] const RowDescriptor& row_desc() const override { return *_output_row_desc; }
[[nodiscard]] const RowDescriptor& intermediate_row_desc() const override {
return *_intermediate_row_desc;
diff --git a/be/src/pipeline/exec/multi_cast_data_stream_sink.h b/be/src/pipeline/exec/multi_cast_data_stream_sink.h
index 84a75720a66..7203ed0bf6f 100644
--- a/be/src/pipeline/exec/multi_cast_data_stream_sink.h
+++ b/be/src/pipeline/exec/multi_cast_data_stream_sink.h
@@ -99,7 +99,7 @@ public:
return Status::OK();
}
- RowDescriptor& row_desc() override { return _row_desc; }
+ const RowDescriptor& row_desc() const override { return _row_desc; }
std::shared_ptr<MultiCastSharedState> create_multi_cast_data_streamer() {
auto multi_cast_data_streamer =
diff --git a/be/src/pipeline/exec/multi_cast_data_stream_source.cpp b/be/src/pipeline/exec/multi_cast_data_stream_source.cpp
index 6ac06ee5f10..ccc0169c289 100644
--- a/be/src/pipeline/exec/multi_cast_data_stream_source.cpp
+++ b/be/src/pipeline/exec/multi_cast_data_stream_source.cpp
@@ -38,7 +38,7 @@ OperatorPtr MultiCastDataStreamerSourceOperatorBuilder::build_operator() {
this, _consumer_id, _multi_cast_data_streamer, _t_data_stream_sink);
}
-const RowDescriptor& MultiCastDataStreamerSourceOperatorBuilder::row_desc() {
+const RowDescriptor& MultiCastDataStreamerSourceOperatorBuilder::row_desc() const {
return _multi_cast_data_streamer->row_desc();
}
diff --git a/be/src/pipeline/exec/multi_cast_data_stream_source.h b/be/src/pipeline/exec/multi_cast_data_stream_source.h
index 5dae2be31f7..8eaa82520d1 100644
--- a/be/src/pipeline/exec/multi_cast_data_stream_source.h
+++ b/be/src/pipeline/exec/multi_cast_data_stream_source.h
@@ -48,7 +48,7 @@ public:
OperatorPtr build_operator() override;
- const RowDescriptor& row_desc() override;
+ const RowDescriptor& row_desc() const override;
private:
const int _consumer_id;
diff --git a/be/src/pipeline/exec/nested_loop_join_probe_operator.h b/be/src/pipeline/exec/nested_loop_join_probe_operator.h
index 1a3a0e9f9a1..35302434e70 100644
--- a/be/src/pipeline/exec/nested_loop_join_probe_operator.h
+++ b/be/src/pipeline/exec/nested_loop_join_probe_operator.h
@@ -234,7 +234,7 @@ public:
return {ExchangeType::ADAPTIVE_PASSTHROUGH};
}
- const RowDescriptor& row_desc() override {
+ const RowDescriptor& row_desc() const override {
return _old_version_flag
? (_output_row_descriptor ? *_output_row_descriptor : _row_descriptor)
: *_output_row_desc;
diff --git a/be/src/pipeline/exec/operator.cpp b/be/src/pipeline/exec/operator.cpp
index 11643bd4c36..4a01fa4578d 100644
--- a/be/src/pipeline/exec/operator.cpp
+++ b/be/src/pipeline/exec/operator.cpp
@@ -50,7 +50,7 @@ Status OperatorBase::close(RuntimeState* state) {
return Status::OK();
}
-const RowDescriptor& OperatorBase::row_desc() {
+const RowDescriptor& OperatorBase::row_desc() const {
return _operator_builder->row_desc();
}
diff --git a/be/src/pipeline/exec/operator.h b/be/src/pipeline/exec/operator.h
index 8755872a979..ffbe51d1b91 100644
--- a/be/src/pipeline/exec/operator.h
+++ b/be/src/pipeline/exec/operator.h
@@ -112,7 +112,7 @@ public:
std::string get_name() const { return _name; }
- virtual const RowDescriptor& row_desc() = 0;
+ virtual const RowDescriptor& row_desc() const = 0;
int32_t id() const { return _id; }
@@ -133,7 +133,7 @@ public:
~OperatorBuilder() override = default;
- const RowDescriptor& row_desc() override { return _node->row_desc(); }
+ const RowDescriptor& row_desc() const override { return _node->row_desc(); }
NodeType* exec_node() const { return _node; }
@@ -151,7 +151,7 @@ public:
bool is_sink() const override { return true; }
- const RowDescriptor& row_desc() override { return _sink->row_desc(); }
+ const RowDescriptor& row_desc() const override { return _sink->row_desc(); }
SinkType* exec_node() const { return _sink; }
@@ -243,7 +243,7 @@ public:
const OperatorBuilderBase* operator_builder() const { return _operator_builder; }
- virtual const RowDescriptor& row_desc();
+ virtual const RowDescriptor& row_desc() const;
virtual std::string debug_string() const;
virtual int32_t id() const { return _operator_builder->id(); }
diff --git a/be/src/pipeline/exec/streaming_aggregation_operator.cpp b/be/src/pipeline/exec/streaming_aggregation_operator.cpp
new file mode 100644
index 00000000000..f9ba600e3c0
--- /dev/null
+++ b/be/src/pipeline/exec/streaming_aggregation_operator.cpp
@@ -0,0 +1,1302 @@
+// 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 "streaming_aggregation_operator.h"
+
+#include <gen_cpp/Metrics_types.h>
+
+#include <utility>
+
+#include "common/compiler_util.h" // IWYU pragma: keep
+#include "pipeline/exec/operator.h"
+
+namespace doris {
+class RuntimeState;
+} // namespace doris
+
+namespace doris::pipeline {
+/// The minimum reduction factor (input rows divided by output rows) to grow hash tables
+/// in a streaming preaggregation, given that the hash tables are currently the given
+/// size or above. The sizes roughly correspond to hash table sizes where the bucket
+/// arrays will fit in a cache level. Intuitively, we don't want the working set of the
+/// aggregation to expand to the next level of cache unless we're reducing the input
+/// enough to outweigh the increased memory latency we'll incur for each hash table
+/// lookup.
+///
+/// Note that the current reduction achieved is not always a good estimate of the
+/// final reduction. It may be biased either way depending on the ordering of the
+/// input. If the input order is random, we will underestimate the final reduction
+/// factor because the probability of a row having the same key as a previous row
+/// increases as more input is processed. If the input order is correlated with the
+/// key, skew may bias the estimate. If high cardinality keys appear first, we
+/// may overestimate and if low cardinality keys appear first, we underestimate.
+/// To estimate the eventual reduction achieved, we estimate the final reduction
+/// using the planner's estimated input cardinality and the assumption that input
+/// is in a random order. This means that we assume that the reduction factor will
+/// increase over time.
+struct StreamingHtMinReductionEntry {
+ // Use 'streaming_ht_min_reduction' if the total size of hash table bucket directories in
+ // bytes is greater than this threshold.
+ int min_ht_mem;
+ // The minimum reduction factor to expand the hash tables.
+ double streaming_ht_min_reduction;
+};
+
+// TODO: experimentally tune these values and also programmatically get the cache size
+// of the machine that we're running on.
+static constexpr StreamingHtMinReductionEntry STREAMING_HT_MIN_REDUCTION[] = {
+ // Expand up to L2 cache always.
+ {0, 0.0},
+ // Expand into L3 cache if we look like we're getting some reduction.
+ // At present, The L2 cache is generally 1024k or more
+ {1024 * 1024, 1.1},
+ // Expand into main memory if we're getting a significant reduction.
+ // The L3 cache is generally 16MB or more
+ {16 * 1024 * 1024, 2.0},
+};
+
+static constexpr int STREAMING_HT_MIN_REDUCTION_SIZE =
+ sizeof(STREAMING_HT_MIN_REDUCTION) / sizeof(STREAMING_HT_MIN_REDUCTION[0]);
+
+StreamingAggLocalState::StreamingAggLocalState(RuntimeState* state, OperatorXBase* parent)
+ : Base(state, parent),
+ _agg_arena_pool(std::make_unique<vectorized::Arena>()),
+ _agg_data(std::make_unique<vectorized::AggregatedDataVariants>()),
+ _agg_profile_arena(std::make_unique<vectorized::Arena>()),
+ _child_block(vectorized::Block::create_unique()),
+ _child_source_state(SourceState::DEPEND_ON_SOURCE),
+ _pre_aggregated_block(vectorized::Block::create_unique()) {}
+
+Status StreamingAggLocalState::init(RuntimeState* state, LocalStateInfo& info) {
+ RETURN_IF_ERROR(Base::init(state, info));
+ SCOPED_TIMER(Base::exec_time_counter());
+ SCOPED_TIMER(Base::_open_timer);
+ auto& p = Base::_parent->template cast<StreamingAggOperatorX>();
+ for (auto& evaluator : p._aggregate_evaluators) {
+ _aggregate_evaluators.push_back(evaluator->clone(state, p._pool));
+ }
+ _probe_expr_ctxs.resize(p._probe_expr_ctxs.size());
+ for (size_t i = 0; i < _probe_expr_ctxs.size(); i++) {
+ RETURN_IF_ERROR(p._probe_expr_ctxs[i]->clone(state, _probe_expr_ctxs[i]));
+ }
+ _hash_table_memory_usage = ADD_CHILD_COUNTER_WITH_LEVEL(Base::profile(), "HashTable",
+ TUnit::BYTES, "MemoryUsage", 1);
+ _serialize_key_arena_memory_usage = Base::profile()->AddHighWaterMarkCounter(
+ "SerializeKeyArena", TUnit::BYTES, "MemoryUsage", 1);
+
+ _build_timer = ADD_TIMER(Base::profile(), "BuildTime");
+ _build_table_convert_timer = ADD_TIMER(Base::profile(), "BuildConvertToPartitionedTime");
+ _serialize_key_timer = ADD_TIMER(Base::profile(), "SerializeKeyTime");
+ _exec_timer = ADD_TIMER(Base::profile(), "ExecTime");
+ _merge_timer = ADD_TIMER(Base::profile(), "MergeTime");
+ _expr_timer = ADD_TIMER(Base::profile(), "ExprTime");
+ _serialize_data_timer = ADD_TIMER(Base::profile(), "SerializeDataTime");
+ _deserialize_data_timer = ADD_TIMER(Base::profile(), "DeserializeAndMergeTime");
+ _hash_table_compute_timer = ADD_TIMER(Base::profile(), "HashTableComputeTime");
+ _hash_table_emplace_timer = ADD_TIMER(Base::profile(), "HashTableEmplaceTime");
+ _hash_table_input_counter = ADD_COUNTER(Base::profile(), "HashTableInputCount", TUnit::UNIT);
+ _max_row_size_counter = ADD_COUNTER(Base::profile(), "MaxRowSizeInBytes", TUnit::UNIT);
+ _hash_table_size_counter = ADD_COUNTER(profile(), "HashTableSize", TUnit::UNIT);
+ _queue_byte_size_counter = ADD_COUNTER(profile(), "MaxSizeInBlockQueue", TUnit::BYTES);
+ _queue_size_counter = ADD_COUNTER(profile(), "MaxSizeOfBlockQueue", TUnit::UNIT);
+ _streaming_agg_timer = ADD_TIMER(profile(), "StreamingAggTime");
+ _build_timer = ADD_TIMER(profile(), "BuildTime");
+ _expr_timer = ADD_TIMER(Base::profile(), "ExprTime");
+ _get_results_timer = ADD_TIMER(profile(), "GetResultsTime");
+ _serialize_result_timer = ADD_TIMER(profile(), "SerializeResultTime");
+ _hash_table_iterate_timer = ADD_TIMER(profile(), "HashTableIterateTime");
+ _insert_keys_to_column_timer = ADD_TIMER(profile(), "InsertKeysToColumnTime");
+ COUNTER_SET(_max_row_size_counter, (int64_t)0);
+
+ for (auto& evaluator : _aggregate_evaluators) {
+ evaluator->set_timer(_merge_timer, _expr_timer);
+ }
+
+ if (_probe_expr_ctxs.empty()) {
+ _agg_data->without_key = reinterpret_cast<vectorized::AggregateDataPtr>(
+ _agg_profile_arena->alloc(p._total_size_of_aggregate_states));
+
+ if (p._is_merge) {
+ if (p._needs_finalize) {
+ _executor = std::make_unique<Executor<true, true, true>>();
+ } else {
+ _executor = std::make_unique<Executor<true, true, false>>();
+ }
+ } else {
+ if (p._needs_finalize) {
+ _executor = std::make_unique<Executor<true, false, true>>();
+ } else {
+ _executor = std::make_unique<Executor<true, false, false>>();
+ }
+ }
+ } else {
+ _init_hash_method(_probe_expr_ctxs);
+
+ std::visit(
+ [&](auto&& agg_method) {
+ using HashTableType = std::decay_t<decltype(agg_method)>;
+ using KeyType = typename HashTableType::Key;
+
+ /// some aggregate functions (like AVG for decimal) have align issues.
+ _aggregate_data_container.reset(new vectorized::AggregateDataContainer(
+ sizeof(KeyType),
+ ((p._total_size_of_aggregate_states + p._align_aggregate_states - 1) /
+ p._align_aggregate_states) *
+ p._align_aggregate_states));
+ },
+ _agg_data->method_variant);
+ if (p._is_merge) {
+ if (p._needs_finalize) {
+ _executor = std::make_unique<Executor<false, true, true>>();
+ } else {
+ _executor = std::make_unique<Executor<false, true, false>>();
+ }
+ } else {
+ if (p._needs_finalize) {
+ _executor = std::make_unique<Executor<false, false, true>>();
+ } else {
+ _executor = std::make_unique<Executor<false, false, false>>();
+ }
+ }
+
+ _should_limit_output = p._limit != -1 && // has limit
+ (!p._have_conjuncts) && // no having conjunct
+ p._needs_finalize; // agg's finalize step
+ }
+ return Status::OK();
+}
+
+void StreamingAggLocalState::_make_nullable_output_key(vectorized::Block* block) {
+ if (block->rows() != 0) {
+ for (auto cid : Base::_parent->cast<StreamingAggOperatorX>()._make_nullable_keys) {
+ block->get_by_position(cid).column = make_nullable(block->get_by_position(cid).column);
+ block->get_by_position(cid).type = make_nullable(block->get_by_position(cid).type);
+ }
+ }
+}
+
+Status StreamingAggLocalState::_execute_without_key(vectorized::Block* block) {
+ DCHECK(_agg_data->without_key != nullptr);
+ SCOPED_TIMER(_build_timer);
+ for (int i = 0; i < _aggregate_evaluators.size(); ++i) {
+ RETURN_IF_ERROR(_aggregate_evaluators[i]->execute_single_add(
+ block,
+ _agg_data->without_key + Base::_parent->template cast<StreamingAggOperatorX>()
+ ._offsets_of_aggregate_states[i],
+ _agg_arena_pool.get()));
+ }
+ return Status::OK();
+}
+
+template <bool limit, bool for_spill>
+Status StreamingAggLocalState::_merge_with_serialized_key_helper(vectorized::Block* block) {
+ SCOPED_TIMER(_merge_timer);
+
+ size_t key_size = _probe_expr_ctxs.size();
+ vectorized::ColumnRawPtrs key_columns(key_size);
+
+ for (size_t i = 0; i < key_size; ++i) {
+ if constexpr (for_spill) {
+ key_columns[i] = block->get_by_position(i).column.get();
+ } else {
+ int result_column_id = -1;
+ RETURN_IF_ERROR(_probe_expr_ctxs[i]->execute(block, &result_column_id));
+ block->replace_by_position_if_const(result_column_id);
+ key_columns[i] = block->get_by_position(result_column_id).column.get();
+ }
+ }
+
+ int rows = block->rows();
+ if (_places.size() < rows) {
+ _places.resize(rows);
+ }
+
+ if constexpr (limit) {
+ _find_in_hash_table(_places.data(), key_columns, rows);
+
+ for (int i = 0; i < _aggregate_evaluators.size(); ++i) {
+ if (_aggregate_evaluators[i]->is_merge()) {
+ int col_id = _get_slot_column_id(_aggregate_evaluators[i]);
+ auto column = block->get_by_position(col_id).column;
+ if (column->is_nullable()) {
+ column = ((vectorized::ColumnNullable*)column.get())->get_nested_column_ptr();
+ }
+
+ size_t buffer_size = _aggregate_evaluators[i]->function()->size_of_data() * rows;
+ if (_deserialize_buffer.size() < buffer_size) {
+ _deserialize_buffer.resize(buffer_size);
+ }
+
+ {
+ SCOPED_TIMER(_deserialize_data_timer);
+ _aggregate_evaluators[i]->function()->deserialize_and_merge_vec_selected(
+ _places.data(),
+ Base::_parent->template cast<StreamingAggOperatorX>()
+ ._offsets_of_aggregate_states[i],
+ _deserialize_buffer.data(), (vectorized::ColumnString*)(column.get()),
+ _agg_arena_pool.get(), rows);
+ }
+ } else {
+ RETURN_IF_ERROR(_aggregate_evaluators[i]->execute_batch_add_selected(
+ block,
+ Base::_parent->template cast<StreamingAggOperatorX>()
+ ._offsets_of_aggregate_states[i],
+ _places.data(), _agg_arena_pool.get()));
+ }
+ }
+ } else {
+ _emplace_into_hash_table(_places.data(), key_columns, rows);
+
+ for (int i = 0; i < _aggregate_evaluators.size(); ++i) {
+ if (_aggregate_evaluators[i]->is_merge() || for_spill) {
+ int col_id = 0;
+ if constexpr (for_spill) {
+ col_id = _probe_expr_ctxs.size() + i;
+ } else {
+ col_id = _get_slot_column_id(_aggregate_evaluators[i]);
+ }
+ auto column = block->get_by_position(col_id).column;
+ if (column->is_nullable()) {
+ column = ((vectorized::ColumnNullable*)column.get())->get_nested_column_ptr();
+ }
+
+ size_t buffer_size = _aggregate_evaluators[i]->function()->size_of_data() * rows;
+ if (_deserialize_buffer.size() < buffer_size) {
+ _deserialize_buffer.resize(buffer_size);
+ }
+
+ {
+ SCOPED_TIMER(_deserialize_data_timer);
+ _aggregate_evaluators[i]->function()->deserialize_and_merge_vec(
+ _places.data(),
+ Base::_parent->template cast<StreamingAggOperatorX>()
+ ._offsets_of_aggregate_states[i],
+ _deserialize_buffer.data(), (vectorized::ColumnString*)(column.get()),
+ _agg_arena_pool.get(), rows);
+ }
+ } else {
+ RETURN_IF_ERROR(_aggregate_evaluators[i]->execute_batch_add(
+ block,
+ Base::_parent->template cast<StreamingAggOperatorX>()
+ ._offsets_of_aggregate_states[i],
+ _places.data(), _agg_arena_pool.get()));
+ }
+ }
+
+ if (_should_limit_output) {
+ _reach_limit = _get_hash_table_size() >=
+ Base::_parent->template cast<StreamingAggOperatorX>()._limit;
+ }
+ }
+
+ return Status::OK();
+}
+
+size_t StreamingAggLocalState::_get_hash_table_size() {
+ return std::visit([&](auto&& agg_method) { return agg_method.hash_table->size(); },
+ _agg_data->method_variant);
+}
+
+Status StreamingAggLocalState::_merge_without_key(vectorized::Block* block) {
+ SCOPED_TIMER(_merge_timer);
+ DCHECK(_agg_data->without_key != nullptr);
+ for (int i = 0; i < _aggregate_evaluators.size(); ++i) {
+ if (_aggregate_evaluators[i]->is_merge()) {
+ int col_id = _get_slot_column_id(_aggregate_evaluators[i]);
+ auto column = block->get_by_position(col_id).column;
+ if (column->is_nullable()) {
+ column = ((vectorized::ColumnNullable*)column.get())->get_nested_column_ptr();
+ }
+
+ SCOPED_TIMER(_deserialize_data_timer);
+ _aggregate_evaluators[i]->function()->deserialize_and_merge_from_column(
+ _agg_data->without_key + Base::_parent->template cast<StreamingAggOperatorX>()
+ ._offsets_of_aggregate_states[i],
+ *column, _agg_arena_pool.get());
+ } else {
+ RETURN_IF_ERROR(_aggregate_evaluators[i]->execute_single_add(
+ block,
+ _agg_data->without_key + Base::_parent->template cast<StreamingAggOperatorX>()
+ ._offsets_of_aggregate_states[i],
+ _agg_arena_pool.get()));
+ }
+ }
+ return Status::OK();
+}
+
+void StreamingAggLocalState::_update_memusage_without_key() {
+ auto arena_memory_usage = _agg_arena_pool->size() - _mem_usage_record.used_in_arena;
+ Base::_mem_tracker->consume(arena_memory_usage);
+ _serialize_key_arena_memory_usage->add(arena_memory_usage);
+ _mem_usage_record.used_in_arena = _agg_arena_pool->size();
+}
+
+Status StreamingAggLocalState::_execute_with_serialized_key(vectorized::Block* block) {
+ if (_reach_limit) {
+ return _execute_with_serialized_key_helper<true>(block);
+ } else {
+ return _execute_with_serialized_key_helper<false>(block);
+ }
+}
+
+void StreamingAggLocalState::_update_memusage_with_serialized_key() {
+ std::visit(
+ [&](auto&& agg_method) -> void {
+ auto& data = *agg_method.hash_table;
+ auto arena_memory_usage = _agg_arena_pool->size() +
+ _aggregate_data_container->memory_usage() -
+ _mem_usage_record.used_in_arena;
+ Base::_mem_tracker->consume(arena_memory_usage);
+ Base::_mem_tracker->consume(data.get_buffer_size_in_bytes() -
+ _mem_usage_record.used_in_state);
+ _serialize_key_arena_memory_usage->add(arena_memory_usage);
+ COUNTER_UPDATE(_hash_table_memory_usage,
+ data.get_buffer_size_in_bytes() - _mem_usage_record.used_in_state);
+ _mem_usage_record.used_in_state = data.get_buffer_size_in_bytes();
+ _mem_usage_record.used_in_arena =
+ _agg_arena_pool->size() + _aggregate_data_container->memory_usage();
+ },
+ _agg_data->method_variant);
+}
+
+template <bool limit>
+Status StreamingAggLocalState::_execute_with_serialized_key_helper(vectorized::Block* block) {
+ SCOPED_TIMER(_build_timer);
+ DCHECK(!_probe_expr_ctxs.empty());
+
+ size_t key_size = _probe_expr_ctxs.size();
+ vectorized::ColumnRawPtrs key_columns(key_size);
+ {
+ SCOPED_TIMER(_expr_timer);
+ for (size_t i = 0; i < key_size; ++i) {
+ int result_column_id = -1;
+ RETURN_IF_ERROR(_probe_expr_ctxs[i]->execute(block, &result_column_id));
+ block->get_by_position(result_column_id).column =
+ block->get_by_position(result_column_id)
+ .column->convert_to_full_column_if_const();
+ key_columns[i] = block->get_by_position(result_column_id).column.get();
+ }
+ }
+
+ int rows = block->rows();
+ if (_places.size() < rows) {
+ _places.resize(rows);
+ }
+
+ if constexpr (limit) {
+ _find_in_hash_table(_places.data(), key_columns, rows);
+
+ for (int i = 0; i < _aggregate_evaluators.size(); ++i) {
+ RETURN_IF_ERROR(_aggregate_evaluators[i]->execute_batch_add_selected(
+ block,
+ Base::_parent->template cast<StreamingAggOperatorX>()
+ ._offsets_of_aggregate_states[i],
+ _places.data(), _agg_arena_pool.get()));
+ }
+ } else {
+ _emplace_into_hash_table(_places.data(), key_columns, rows);
+
+ for (int i = 0; i < _aggregate_evaluators.size(); ++i) {
+ RETURN_IF_ERROR(_aggregate_evaluators[i]->execute_batch_add(
+ block,
+ Base::_parent->template cast<StreamingAggOperatorX>()
+ ._offsets_of_aggregate_states[i],
+ _places.data(), _agg_arena_pool.get()));
+ }
+
+ if (_should_limit_output) {
+ _reach_limit = _get_hash_table_size() >=
+ Base::_parent->template cast<StreamingAggOperatorX>()._limit;
+ if (_reach_limit &&
+ Base::_parent->template cast<StreamingAggOperatorX>()._can_short_circuit) {
+ Base::_dependency->set_ready_to_read();
+ return Status::Error<ErrorCode::END_OF_FILE>("");
+ }
+ }
+ }
+
+ return Status::OK();
+}
+
+// We should call this function only at 1st phase.
+// 1st phase: is_merge=true, only have one SlotRef.
+// 2nd phase: is_merge=false, maybe have multiple exprs.
+int StreamingAggLocalState::_get_slot_column_id(const vectorized::AggFnEvaluator* evaluator) {
+ auto ctxs = evaluator->input_exprs_ctxs();
+ CHECK(ctxs.size() == 1 && ctxs[0]->root()->is_slot_ref())
+ << "input_exprs_ctxs is invalid, input_exprs_ctx[0]="
+ << ctxs[0]->root()->debug_string();
+ return ((vectorized::VSlotRef*)ctxs[0]->root().get())->column_id();
+}
+
+void StreamingAggLocalState::_find_in_hash_table(vectorized::AggregateDataPtr* places,
+ vectorized::ColumnRawPtrs& key_columns,
+ size_t num_rows) {
+ std::visit(
+ [&](auto&& agg_method) -> void {
+ using HashMethodType = std::decay_t<decltype(agg_method)>;
+ using AggState = typename HashMethodType::State;
+ AggState state(key_columns);
+ agg_method.init_serialized_keys(key_columns, num_rows);
+
+ /// For all rows.
+ for (size_t i = 0; i < num_rows; ++i) {
+ auto find_result = agg_method.find(state, i);
+
+ if (find_result.is_found()) {
+ places[i] = find_result.get_mapped();
+ } else {
+ places[i] = nullptr;
+ }
+ }
+ },
+ _agg_data->method_variant);
+}
+
+Status StreamingAggLocalState::_merge_with_serialized_key(vectorized::Block* block) {
+ if (_reach_limit) {
+ return _merge_with_serialized_key_helper<true, false>(block);
+ } else {
+ return _merge_with_serialized_key_helper<false, false>(block);
+ }
+}
+
+void StreamingAggLocalState::_init_hash_method(const vectorized::VExprContextSPtrs& probe_exprs) {
+ DCHECK(probe_exprs.size() >= 1);
+
+ using Type = vectorized::AggregatedDataVariants::Type;
+ Type t(Type::serialized);
+
+ if (probe_exprs.size() == 1) {
+ auto is_nullable = probe_exprs[0]->root()->is_nullable();
+ PrimitiveType type = probe_exprs[0]->root()->result_type();
+ switch (type) {
+ case TYPE_TINYINT:
+ case TYPE_BOOLEAN:
+ case TYPE_SMALLINT:
+ case TYPE_INT:
+ case TYPE_FLOAT:
+ case TYPE_DATEV2:
+ case TYPE_BIGINT:
+ case TYPE_DOUBLE:
+ case TYPE_DATE:
+ case TYPE_DATETIME:
+ case TYPE_DATETIMEV2:
+ case TYPE_LARGEINT:
+ case TYPE_DECIMALV2:
+ case TYPE_DECIMAL32:
+ case TYPE_DECIMAL64:
+ case TYPE_DECIMAL128I: {
+ size_t size = get_primitive_type_size(type);
+ if (size == 1) {
+ t = Type::int8_key;
+ } else if (size == 2) {
+ t = Type::int16_key;
+ } else if (size == 4) {
+ t = Type::int32_key;
+ } else if (size == 8) {
+ t = Type::int64_key;
+ } else if (size == 16) {
+ t = Type::int128_key;
+ } else {
+ throw Exception(ErrorCode::INTERNAL_ERROR,
+ "meet invalid type size, size={}, type={}", size,
+ type_to_string(type));
+ }
+ break;
+ }
+ case TYPE_CHAR:
+ case TYPE_VARCHAR:
+ case TYPE_STRING: {
+ t = Type::string_key;
+ break;
+ }
+ default:
+ t = Type::serialized;
+ }
+
+ _agg_data->init(
+ get_hash_key_type_with_phase(
+ t, !Base::_parent->template cast<StreamingAggOperatorX>()._is_first_phase),
+ is_nullable);
+ } else {
+ if (!try_get_hash_map_context_fixed<PHNormalHashMap, HashCRC32,
+ vectorized::AggregateDataPtr>(_agg_data->method_variant,
+ probe_exprs)) {
+ _agg_data->init(Type::serialized);
+ }
+ }
+}
+
+Status StreamingAggLocalState::do_pre_agg(vectorized::Block* input_block,
+ vectorized::Block* output_block) {
+ RETURN_IF_ERROR(_pre_agg_with_serialized_key(input_block, output_block));
+
+ // pre stream agg need use _num_row_return to decide whether to do pre stream agg
+ _cur_num_rows_returned += output_block->rows();
+ _make_nullable_output_key(output_block);
+ // COUNTER_SET(_rows_returned_counter, _num_rows_returned);
+ _executor->update_memusage(this);
+ return Status::OK();
+}
+
+bool StreamingAggLocalState::_should_expand_preagg_hash_tables() {
+ if (!_should_expand_hash_table) {
+ return false;
+ }
+
+ return std::visit(
+ [&](auto&& agg_method) -> bool {
+ auto& hash_tbl = *agg_method.hash_table;
+ auto [ht_mem, ht_rows] =
+ std::pair {hash_tbl.get_buffer_size_in_bytes(), hash_tbl.size()};
+
+ // Need some rows in tables to have valid statistics.
+ if (ht_rows == 0) {
+ return true;
+ }
+
+ // Find the appropriate reduction factor in our table for the current hash table sizes.
+ int cache_level = 0;
+ while (cache_level + 1 < STREAMING_HT_MIN_REDUCTION_SIZE &&
+ ht_mem >= STREAMING_HT_MIN_REDUCTION[cache_level + 1].min_ht_mem) {
+ ++cache_level;
+ }
+
+ // Compare the number of rows in the hash table with the number of input rows that
+ // were aggregated into it. Exclude passed through rows from this calculation since
+ // they were not in hash tables.
+ const int64_t input_rows = _input_num_rows;
+ const int64_t aggregated_input_rows = input_rows - _cur_num_rows_returned;
+ // TODO chenhao
+ // const int64_t expected_input_rows = estimated_input_cardinality_ - num_rows_returned_;
+ double current_reduction = static_cast<double>(aggregated_input_rows) / ht_rows;
+
+ // TODO: workaround for IMPALA-2490: subplan node rows_returned counter may be
+ // inaccurate, which could lead to a divide by zero below.
+ if (aggregated_input_rows <= 0) {
+ return true;
+ }
+
+ // Extrapolate the current reduction factor (r) using the formula
+ // R = 1 + (N / n) * (r - 1), where R is the reduction factor over the full input data
+ // set, N is the number of input rows, excluding passed-through rows, and n is the
+ // number of rows inserted or merged into the hash tables. This is a very rough
+ // approximation but is good enough to be useful.
+ // TODO: consider collecting more statistics to better estimate reduction.
+ // double estimated_reduction = aggregated_input_rows >= expected_input_rows
+ // ? current_reduction
+ // : 1 + (expected_input_rows / aggregated_input_rows) * (current_reduction - 1);
+ double min_reduction =
+ STREAMING_HT_MIN_REDUCTION[cache_level].streaming_ht_min_reduction;
+
+ // COUNTER_SET(preagg_estimated_reduction_, estimated_reduction);
+ // COUNTER_SET(preagg_streaming_ht_min_reduction_, min_reduction);
+ // return estimated_reduction > min_reduction;
+ _should_expand_hash_table = current_reduction > min_reduction;
+ return _should_expand_hash_table;
+ },
+ _agg_data->method_variant);
+}
+
+size_t StreamingAggLocalState::_memory_usage() const {
+ size_t usage = 0;
+ if (_agg_arena_pool) {
+ usage += _agg_arena_pool->size();
+ }
+
+ if (_aggregate_data_container) {
+ usage += _aggregate_data_container->memory_usage();
+ }
+
+ return usage;
+}
+
+Status StreamingAggLocalState::_pre_agg_with_serialized_key(doris::vectorized::Block* in_block,
+ doris::vectorized::Block* out_block) {
+ SCOPED_TIMER(_build_timer);
+ DCHECK(!_probe_expr_ctxs.empty());
+
+ auto& p = Base::_parent->template cast<StreamingAggOperatorX>();
+
+ size_t key_size = _probe_expr_ctxs.size();
+ vectorized::ColumnRawPtrs key_columns(key_size);
+ {
+ SCOPED_TIMER(_expr_timer);
+ for (size_t i = 0; i < key_size; ++i) {
+ int result_column_id = -1;
+ RETURN_IF_ERROR(_probe_expr_ctxs[i]->execute(in_block, &result_column_id));
+ in_block->get_by_position(result_column_id).column =
+ in_block->get_by_position(result_column_id)
+ .column->convert_to_full_column_if_const();
+ key_columns[i] = in_block->get_by_position(result_column_id).column.get();
+ }
+ }
+
+ int rows = in_block->rows();
+ _places.resize(rows);
+
+ // Stop expanding hash tables if we're not reducing the input sufficiently. As our
+ // hash tables expand out of each level of cache hierarchy, every hash table lookup
+ // will take longer. We also may not be able to expand hash tables because of memory
+ // pressure. In either case we should always use the remaining space in the hash table
+ // to avoid wasting memory.
+ // But for fixed hash map, it never need to expand
+ bool ret_flag = false;
+ RETURN_IF_ERROR(std::visit(
+ [&](auto&& agg_method) -> Status {
+ if (auto& hash_tbl = *agg_method.hash_table;
+ hash_tbl.add_elem_size_overflow(rows)) {
+ /// If too much memory is used during the pre-aggregation stage,
+ /// it is better to output the data directly without performing further aggregation.
+ const bool used_too_much_memory =
+ (_parent->cast<StreamingAggOperatorX>()._external_agg_bytes_threshold >
+ 0 &&
+ _memory_usage() > _parent->cast<StreamingAggOperatorX>()
+ ._external_agg_bytes_threshold);
+ // do not try to do agg, just init and serialize directly return the out_block
+ if (!_should_expand_preagg_hash_tables() || used_too_much_memory) {
+ SCOPED_TIMER(_streaming_agg_timer);
+ ret_flag = true;
+
+ // will serialize value data to string column.
+ // non-nullable column(id in `_make_nullable_keys`)
+ // will be converted to nullable.
+ bool mem_reuse = p._make_nullable_keys.empty() && out_block->mem_reuse();
+
+ std::vector<vectorized::DataTypePtr> data_types;
+ vectorized::MutableColumns value_columns;
+ for (int i = 0; i < _aggregate_evaluators.size(); ++i) {
+ auto data_type =
+ _aggregate_evaluators[i]->function()->get_serialized_type();
+ if (mem_reuse) {
+ value_columns.emplace_back(
+ std::move(*out_block->get_by_position(i + key_size).column)
+ .mutate());
+ } else {
+ // slot type of value it should always be string type
+ value_columns.emplace_back(_aggregate_evaluators[i]
+ ->function()
+ ->create_serialize_column());
+ }
+ data_types.emplace_back(data_type);
+ }
+
+ for (int i = 0; i != _aggregate_evaluators.size(); ++i) {
+ SCOPED_TIMER(_serialize_data_timer);
+ RETURN_IF_ERROR(
+ _aggregate_evaluators[i]->streaming_agg_serialize_to_column(
+ in_block, value_columns[i], rows,
+ _agg_arena_pool.get()));
+ }
+
+ if (!mem_reuse) {
+ vectorized::ColumnsWithTypeAndName columns_with_schema;
+ for (int i = 0; i < key_size; ++i) {
+ columns_with_schema.emplace_back(
+ key_columns[i]->clone_resized(rows),
+ _probe_expr_ctxs[i]->root()->data_type(),
+ _probe_expr_ctxs[i]->root()->expr_name());
+ }
+ for (int i = 0; i < value_columns.size(); ++i) {
+ columns_with_schema.emplace_back(std::move(value_columns[i]),
+ data_types[i], "");
+ }
+ out_block->swap(vectorized::Block(columns_with_schema));
+ } else {
+ for (int i = 0; i < key_size; ++i) {
+ std::move(*out_block->get_by_position(i).column)
+ .mutate()
+ ->insert_range_from(*key_columns[i], 0, rows);
+ }
+ }
+ }
+ }
+ return Status::OK();
+ },
+ _agg_data->method_variant));
+
+ if (!ret_flag) {
+ RETURN_IF_CATCH_EXCEPTION(_emplace_into_hash_table(_places.data(), key_columns, rows));
+
+ for (int i = 0; i < _aggregate_evaluators.size(); ++i) {
+ RETURN_IF_ERROR(_aggregate_evaluators[i]->execute_batch_add(
+ in_block, p._offsets_of_aggregate_states[i], _places.data(),
+ _agg_arena_pool.get(), _should_expand_hash_table));
+ }
+ }
+
+ return Status::OK();
+}
+
+Status StreamingAggLocalState::_create_agg_status(vectorized::AggregateDataPtr data) {
+ auto& p = Base::_parent->template cast<StreamingAggOperatorX>();
+ for (int i = 0; i < _aggregate_evaluators.size(); ++i) {
+ try {
+ _aggregate_evaluators[i]->create(data + p._offsets_of_aggregate_states[i]);
+ } catch (...) {
+ for (int j = 0; j < i; ++j) {
+ _aggregate_evaluators[j]->destroy(data + p._offsets_of_aggregate_states[j]);
+ }
+ throw;
+ }
+ }
+ return Status::OK();
+}
+
+Status StreamingAggLocalState::_get_with_serialized_key_result(RuntimeState* state,
+ vectorized::Block* block,
+ SourceState& source_state) {
+ auto& p = _parent->cast<StreamingAggOperatorX>();
+ // non-nullable column(id in `_make_nullable_keys`) will be converted to nullable.
+ bool mem_reuse = p._make_nullable_keys.empty() && block->mem_reuse();
+
+ auto columns_with_schema =
+ vectorized::VectorizedUtils::create_columns_with_type_and_name(p._row_descriptor);
+ int key_size = _probe_expr_ctxs.size();
+
+ vectorized::MutableColumns key_columns;
+ for (int i = 0; i < key_size; ++i) {
+ if (!mem_reuse) {
+ key_columns.emplace_back(columns_with_schema[i].type->create_column());
+ } else {
+ key_columns.emplace_back(std::move(*block->get_by_position(i).column).mutate());
+ }
+ }
+ vectorized::MutableColumns value_columns;
+ for (int i = key_size; i < columns_with_schema.size(); ++i) {
+ if (!mem_reuse) {
+ value_columns.emplace_back(columns_with_schema[i].type->create_column());
+ } else {
+ value_columns.emplace_back(std::move(*block->get_by_position(i).column).mutate());
+ }
+ }
+
+ SCOPED_TIMER(_get_results_timer);
+ std::visit(
+ [&](auto&& agg_method) -> void {
+ auto& data = *agg_method.hash_table;
+ agg_method.init_iterator();
+ const auto size = std::min(data.size(), size_t(state->batch_size()));
+ using KeyType = std::decay_t<decltype(agg_method.iterator->get_first())>;
+ std::vector<KeyType> keys(size);
+ if (_values.size() < size) {
+ _values.resize(size);
+ }
+
+ size_t num_rows = 0;
+ _aggregate_data_container->init_once();
+ auto& iter = _aggregate_data_container->iterator;
+
+ {
+ SCOPED_TIMER(_hash_table_iterate_timer);
+ while (iter != _aggregate_data_container->end() &&
+ num_rows < state->batch_size()) {
+ keys[num_rows] = iter.template get_key<KeyType>();
+ _values[num_rows] = iter.get_aggregate_data();
+ ++iter;
+ ++num_rows;
+ }
+ }
+
+ {
+ SCOPED_TIMER(_insert_keys_to_column_timer);
+ agg_method.insert_keys_into_columns(keys, key_columns, num_rows);
+ }
+
+ for (size_t i = 0; i < _aggregate_evaluators.size(); ++i) {
+ _aggregate_evaluators[i]->insert_result_info_vec(
+ _values, p._offsets_of_aggregate_states[i], value_columns[i].get(),
+ num_rows);
+ }
+
+ if (iter == _aggregate_data_container->end()) {
+ if (agg_method.hash_table->has_null_key_data()) {
+ // only one key of group by support wrap null key
+ // here need additional processing logic on the null key / value
+ DCHECK(key_columns.size() == 1);
+ DCHECK(key_columns[0]->is_nullable());
+ if (key_columns[0]->size() < state->batch_size()) {
+ key_columns[0]->insert_data(nullptr, 0);
+ auto mapped = agg_method.hash_table->template get_null_key_data<
+ vectorized::AggregateDataPtr>();
+ for (size_t i = 0; i < _aggregate_evaluators.size(); ++i)
+ _aggregate_evaluators[i]->insert_result_info(
+ mapped + p._offsets_of_aggregate_states[i],
+ value_columns[i].get());
+ source_state = SourceState::FINISHED;
+ }
+ } else {
+ source_state = SourceState::FINISHED;
+ }
+ }
+ },
+ _agg_data->method_variant);
+
+ if (!mem_reuse) {
+ *block = columns_with_schema;
+ vectorized::MutableColumns columns(block->columns());
+ for (int i = 0; i < block->columns(); ++i) {
+ if (i < key_size) {
+ columns[i] = std::move(key_columns[i]);
+ } else {
+ columns[i] = std::move(value_columns[i - key_size]);
+ }
+ }
+ block->set_columns(std::move(columns));
+ }
+
+ return Status::OK();
+}
+
+Status StreamingAggLocalState::_serialize_without_key(RuntimeState* state, vectorized::Block* block,
+ SourceState& source_state) {
+ // 1. `child(0)->rows_returned() == 0` mean not data from child
+ // in level two aggregation node should return NULL result
+ // level one aggregation node set `eos = true` return directly
+ SCOPED_TIMER(_serialize_result_timer);
+ if (UNLIKELY(_input_num_rows == 0)) {
+ source_state = SourceState::FINISHED;
+ return Status::OK();
+ }
+ block->clear();
+
+ DCHECK(_agg_data->without_key != nullptr);
+ int agg_size = _aggregate_evaluators.size();
+
+ vectorized::MutableColumns value_columns(agg_size);
+ std::vector<vectorized::DataTypePtr> data_types(agg_size);
+ // will serialize data to string column
+ for (int i = 0; i < _aggregate_evaluators.size(); ++i) {
+ data_types[i] = _aggregate_evaluators[i]->function()->get_serialized_type();
+ value_columns[i] = _aggregate_evaluators[i]->function()->create_serialize_column();
+ }
+
+ for (int i = 0; i < _aggregate_evaluators.size(); ++i) {
+ _aggregate_evaluators[i]->function()->serialize_without_key_to_column(
+ _agg_data->without_key +
+ _parent->cast<StreamingAggOperatorX>()._offsets_of_aggregate_states[i],
+ *value_columns[i]);
+ }
+
+ {
+ vectorized::ColumnsWithTypeAndName data_with_schema;
+ for (int i = 0; i < _aggregate_evaluators.size(); ++i) {
+ vectorized::ColumnWithTypeAndName column_with_schema = {nullptr, data_types[i], ""};
+ data_with_schema.push_back(std::move(column_with_schema));
+ }
+ *block = vectorized::Block(data_with_schema);
+ }
+
+ block->set_columns(std::move(value_columns));
+ source_state = SourceState::FINISHED;
+ return Status::OK();
+}
+
+Status StreamingAggLocalState::_serialize_with_serialized_key_result(RuntimeState* state,
+ vectorized::Block* block,
+ SourceState& source_state) {
+ SCOPED_TIMER(_serialize_result_timer);
+ auto& p = _parent->cast<StreamingAggOperatorX>();
+ int key_size = _probe_expr_ctxs.size();
+ int agg_size = _aggregate_evaluators.size();
+ vectorized::MutableColumns value_columns(agg_size);
+ vectorized::DataTypes value_data_types(agg_size);
+
+ // non-nullable column(id in `_make_nullable_keys`) will be converted to nullable.
+ bool mem_reuse = p._make_nullable_keys.empty() && block->mem_reuse();
+
+ vectorized::MutableColumns key_columns;
+ for (int i = 0; i < key_size; ++i) {
+ if (mem_reuse) {
+ key_columns.emplace_back(std::move(*block->get_by_position(i).column).mutate());
+ } else {
+ key_columns.emplace_back(_probe_expr_ctxs[i]->root()->data_type()->create_column());
+ }
+ }
+
+ SCOPED_TIMER(_get_results_timer);
+ std::visit(
+ [&](auto&& agg_method) -> void {
+ agg_method.init_iterator();
+ auto& data = *agg_method.hash_table;
+ const auto size = std::min(data.size(), size_t(state->batch_size()));
+ using KeyType = std::decay_t<decltype(agg_method.iterator->get_first())>;
+ std::vector<KeyType> keys(size);
+ if (_values.size() < size + 1) {
+ _values.resize(size + 1);
+ }
+
+ size_t num_rows = 0;
+ _aggregate_data_container->init_once();
+ auto& iter = _aggregate_data_container->iterator;
+
+ {
+ SCOPED_TIMER(_hash_table_iterate_timer);
+ while (iter != _aggregate_data_container->end() &&
+ num_rows < state->batch_size()) {
+ keys[num_rows] = iter.template get_key<KeyType>();
+ _values[num_rows] = iter.get_aggregate_data();
+ ++iter;
+ ++num_rows;
+ }
+ }
+
+ {
+ SCOPED_TIMER(_insert_keys_to_column_timer);
+ agg_method.insert_keys_into_columns(keys, key_columns, num_rows);
+ }
+
+ if (iter == _aggregate_data_container->end()) {
+ if (agg_method.hash_table->has_null_key_data()) {
+ // only one key of group by support wrap null key
+ // here need additional processing logic on the null key / value
+ DCHECK(key_columns.size() == 1);
+ DCHECK(key_columns[0]->is_nullable());
+ if (agg_method.hash_table->has_null_key_data()) {
+ key_columns[0]->insert_data(nullptr, 0);
+ _values[num_rows] = agg_method.hash_table->template get_null_key_data<
+ vectorized::AggregateDataPtr>();
+ ++num_rows;
+ source_state = SourceState::FINISHED;
+ }
+ } else {
+ source_state = SourceState::FINISHED;
+ }
+ }
+
+ {
+ SCOPED_TIMER(_serialize_data_timer);
+ for (size_t i = 0; i < _aggregate_evaluators.size(); ++i) {
+ value_data_types[i] =
+ _aggregate_evaluators[i]->function()->get_serialized_type();
+ if (mem_reuse) {
+ value_columns[i] =
+ std::move(*block->get_by_position(i + key_size).column)
+ .mutate();
+ } else {
+ value_columns[i] =
+ _aggregate_evaluators[i]->function()->create_serialize_column();
+ }
+ _aggregate_evaluators[i]->function()->serialize_to_column(
+ _values, p._offsets_of_aggregate_states[i], value_columns[i],
+ num_rows);
+ }
+ }
+ },
+ _agg_data->method_variant);
+
+ if (!mem_reuse) {
+ vectorized::ColumnsWithTypeAndName columns_with_schema;
+ for (int i = 0; i < key_size; ++i) {
+ columns_with_schema.emplace_back(std::move(key_columns[i]),
+ _probe_expr_ctxs[i]->root()->data_type(),
+ _probe_expr_ctxs[i]->root()->expr_name());
+ }
+ for (int i = 0; i < agg_size; ++i) {
+ columns_with_schema.emplace_back(std::move(value_columns[i]), value_data_types[i], "");
+ }
+ *block = vectorized::Block(columns_with_schema);
+ }
+
+ return Status::OK();
+}
+
+void StreamingAggLocalState::make_nullable_output_key(vectorized::Block* block) {
+ if (block->rows() != 0) {
+ for (auto cid : _parent->cast<StreamingAggOperatorX>()._make_nullable_keys) {
+ block->get_by_position(cid).column = make_nullable(block->get_by_position(cid).column);
+ block->get_by_position(cid).type = make_nullable(block->get_by_position(cid).type);
+ }
+ }
+}
+
+Status StreamingAggLocalState::_get_without_key_result(RuntimeState* state,
+ vectorized::Block* block,
+ SourceState& source_state) {
+ DCHECK(_agg_data->without_key != nullptr);
+ block->clear();
+
+ auto& p = _parent->cast<StreamingAggOperatorX>();
+ *block = vectorized::VectorizedUtils::create_empty_columnswithtypename(p._row_descriptor);
+ int agg_size = _aggregate_evaluators.size();
+
+ vectorized::MutableColumns columns(agg_size);
+ std::vector<vectorized::DataTypePtr> data_types(agg_size);
+ for (int i = 0; i < _aggregate_evaluators.size(); ++i) {
+ data_types[i] = _aggregate_evaluators[i]->function()->get_return_type();
+ columns[i] = data_types[i]->create_column();
+ }
+
+ for (int i = 0; i < _aggregate_evaluators.size(); ++i) {
+ auto column = columns[i].get();
+ _aggregate_evaluators[i]->insert_result_info(
+ _agg_data->without_key + p._offsets_of_aggregate_states[i], column);
+ }
+
+ const auto& block_schema = block->get_columns_with_type_and_name();
+ DCHECK_EQ(block_schema.size(), columns.size());
+ for (int i = 0; i < block_schema.size(); ++i) {
+ const auto column_type = block_schema[i].type;
+ if (!column_type->equals(*data_types[i])) {
+ if (!vectorized::is_array(remove_nullable(column_type))) {
+ if (!column_type->is_nullable() || data_types[i]->is_nullable() ||
+ !remove_nullable(column_type)->equals(*data_types[i])) {
+ return Status::InternalError(
+ "node id = {}, column_type not match data_types, column_type={}, "
+ "data_types={}",
+ _parent->node_id(), column_type->get_name(), data_types[i]->get_name());
+ }
+ }
+
+ if (column_type->is_nullable() && !data_types[i]->is_nullable()) {
+ vectorized::ColumnPtr ptr = std::move(columns[i]);
+ // unless `count`, other aggregate function dispose empty set should be null
+ // so here check the children row return
+ ptr = make_nullable(ptr, _input_num_rows == 0);
+ columns[i] = ptr->assume_mutable();
+ }
+ }
+ }
+
+ block->set_columns(std::move(columns));
+ source_state = SourceState::FINISHED;
+ return Status::OK();
+}
+
+void StreamingAggLocalState::_emplace_into_hash_table(vectorized::AggregateDataPtr* places,
+ vectorized::ColumnRawPtrs& key_columns,
+ const size_t num_rows) {
+ std::visit(
+ [&](auto&& agg_method) -> void {
+ SCOPED_TIMER(_hash_table_compute_timer);
+ using HashMethodType = std::decay_t<decltype(agg_method)>;
+ using AggState = typename HashMethodType::State;
+ AggState state(key_columns);
+ agg_method.init_serialized_keys(key_columns, num_rows);
+
+ auto creator = [this](const auto& ctor, auto& key, auto& origin) {
+ HashMethodType::try_presis_key_and_origin(key, origin, *_agg_arena_pool);
+ auto mapped = _aggregate_data_container->append_data(origin);
+ auto st = _create_agg_status(mapped);
+ if (!st) {
+ throw Exception(st.code(), st.to_string());
+ }
+ ctor(key, mapped);
+ };
+
+ auto creator_for_null_key = [&](auto& mapped) {
+ mapped = _agg_arena_pool->aligned_alloc(
+ Base::_parent->template cast<StreamingAggOperatorX>()
+ ._total_size_of_aggregate_states,
+ Base::_parent->template cast<StreamingAggOperatorX>()
+ ._align_aggregate_states);
+ auto st = _create_agg_status(mapped);
+ if (!st) {
+ throw Exception(st.code(), st.to_string());
+ }
+ };
+
+ SCOPED_TIMER(_hash_table_emplace_timer);
+ for (size_t i = 0; i < num_rows; ++i) {
+ places[i] = agg_method.lazy_emplace(state, i, creator, creator_for_null_key);
+ }
+
+ COUNTER_UPDATE(_hash_table_input_counter, num_rows);
+ },
+ _agg_data->method_variant);
+}
+
+StreamingAggOperatorX::StreamingAggOperatorX(ObjectPool* pool, int operator_id,
+ const TPlanNode& tnode, const DescriptorTbl& descs)
+ : StatefulOperatorX<StreamingAggLocalState>(pool, tnode, operator_id, descs),
+ _intermediate_tuple_id(tnode.agg_node.intermediate_tuple_id),
+ _intermediate_tuple_desc(nullptr),
+ _output_tuple_id(tnode.agg_node.output_tuple_id),
+ _output_tuple_desc(nullptr),
+ _needs_finalize(tnode.agg_node.need_finalize),
+ _is_merge(false),
+ _is_first_phase(tnode.agg_node.__isset.is_first_phase && tnode.agg_node.is_first_phase),
+ _have_conjuncts(tnode.__isset.vconjunct && !tnode.vconjunct.nodes.empty()) {}
+
+Status StreamingAggOperatorX::init(const TPlanNode& tnode, RuntimeState* state) {
+ RETURN_IF_ERROR(StatefulOperatorX<StreamingAggLocalState>::init(tnode, state));
+ // ignore return status for now , so we need to introduce ExecNode::init()
+ RETURN_IF_ERROR(
+ vectorized::VExpr::create_expr_trees(tnode.agg_node.grouping_exprs, _probe_expr_ctxs));
+
+ // init aggregate functions
+ _aggregate_evaluators.reserve(tnode.agg_node.aggregate_functions.size());
+ // In case of : `select * from (select GoodEvent from hits union select CounterID from hits) as h limit 10;`
+ // only union with limit: we can short circuit query the pipeline exec engine.
+ _can_short_circuit =
+ tnode.agg_node.aggregate_functions.empty() && state->enable_pipeline_exec();
+
+ TSortInfo dummy;
+ for (int i = 0; i < tnode.agg_node.aggregate_functions.size(); ++i) {
+ vectorized::AggFnEvaluator* evaluator = nullptr;
+ RETURN_IF_ERROR(vectorized::AggFnEvaluator::create(
+ _pool, tnode.agg_node.aggregate_functions[i],
+ tnode.agg_node.__isset.agg_sort_infos ? tnode.agg_node.agg_sort_infos[i] : dummy,
+ &evaluator));
+ _aggregate_evaluators.push_back(evaluator);
+ }
+
+ const auto& agg_functions = tnode.agg_node.aggregate_functions;
+ _external_agg_bytes_threshold = state->external_agg_bytes_threshold();
+
+ if (_external_agg_bytes_threshold > 0) {
+ _spill_partition_count_bits = 4;
+ if (state->query_options().__isset.external_agg_partition_bits) {
+ _spill_partition_count_bits = state->query_options().external_agg_partition_bits;
+ }
+ }
+
+ _is_merge = std::any_of(agg_functions.cbegin(), agg_functions.cend(),
+ [](const auto& e) { return e.nodes[0].agg_expr.is_merge_agg; });
+ _op_name = "STREAMING_AGGREGATION_OPERATOR";
+ return Status::OK();
+}
+
+Status StreamingAggOperatorX::prepare(RuntimeState* state) {
+ RETURN_IF_ERROR(StatefulOperatorX<StreamingAggLocalState>::prepare(state));
+ _intermediate_tuple_desc = state->desc_tbl().get_tuple_descriptor(_intermediate_tuple_id);
+ _output_tuple_desc = state->desc_tbl().get_tuple_descriptor(_output_tuple_id);
+ DCHECK_EQ(_intermediate_tuple_desc->slots().size(), _output_tuple_desc->slots().size());
+ RETURN_IF_ERROR(vectorized::VExpr::prepare(_probe_expr_ctxs, state, _child_x->row_desc()));
+
+ int j = _probe_expr_ctxs.size();
+ for (int i = 0; i < j; ++i) {
+ auto nullable_output = _output_tuple_desc->slots()[i]->is_nullable();
+ auto nullable_input = _probe_expr_ctxs[i]->root()->is_nullable();
+ if (nullable_output != nullable_input) {
+ DCHECK(nullable_output);
+ _make_nullable_keys.emplace_back(i);
+ }
+ }
+ for (int i = 0; i < _aggregate_evaluators.size(); ++i, ++j) {
+ SlotDescriptor* intermediate_slot_desc = _intermediate_tuple_desc->slots()[j];
+ SlotDescriptor* output_slot_desc = _output_tuple_desc->slots()[j];
+ RETURN_IF_ERROR(_aggregate_evaluators[i]->prepare(
+ state, _child_x->row_desc(), intermediate_slot_desc, output_slot_desc));
+ }
+
+ _offsets_of_aggregate_states.resize(_aggregate_evaluators.size());
+
+ for (size_t i = 0; i < _aggregate_evaluators.size(); ++i) {
+ _offsets_of_aggregate_states[i] = _total_size_of_aggregate_states;
+
+ const auto& agg_function = _aggregate_evaluators[i]->function();
+ // aggreate states are aligned based on maximum requirement
+ _align_aggregate_states = std::max(_align_aggregate_states, agg_function->align_of_data());
+ _total_size_of_aggregate_states += agg_function->size_of_data();
+
+ // If not the last aggregate_state, we need pad it so that next aggregate_state will be aligned.
+ if (i + 1 < _aggregate_evaluators.size()) {
+ size_t alignment_of_next_state =
+ _aggregate_evaluators[i + 1]->function()->align_of_data();
+ if ((alignment_of_next_state & (alignment_of_next_state - 1)) != 0) {
+ return Status::RuntimeError("Logical error: align_of_data is not 2^N");
+ }
+
+ /// Extend total_size to next alignment requirement
+ /// Add padding by rounding up 'total_size_of_aggregate_states' to be a multiplier of alignment_of_next_state.
+ _total_size_of_aggregate_states =
+ (_total_size_of_aggregate_states + alignment_of_next_state - 1) /
+ alignment_of_next_state * alignment_of_next_state;
+ }
+ }
+
+ return Status::OK();
+}
+
+Status StreamingAggOperatorX::open(RuntimeState* state) {
+ RETURN_IF_ERROR(StatefulOperatorX<StreamingAggLocalState>::open(state));
+ RETURN_IF_ERROR(vectorized::VExpr::open(_probe_expr_ctxs, state));
+
+ for (int i = 0; i < _aggregate_evaluators.size(); ++i) {
+ RETURN_IF_ERROR(_aggregate_evaluators[i]->open(state));
+ _aggregate_evaluators[i]->set_version(state->be_exec_version());
+ }
+
+ return Status::OK();
+}
+
+Status StreamingAggLocalState::close(RuntimeState* state) {
+ if (_closed) {
+ return Status::OK();
+ }
+ SCOPED_TIMER(Base::exec_time_counter());
+ SCOPED_TIMER(Base::_close_timer);
+ if (Base::_closed) {
+ return Status::OK();
+ }
+ _pre_aggregated_block->clear();
+ vectorized::PODArray<vectorized::AggregateDataPtr> tmp_places;
+ _places.swap(tmp_places);
+
+ std::vector<char> tmp_deserialize_buffer;
+ _deserialize_buffer.swap(tmp_deserialize_buffer);
+ Base::_mem_tracker->release(_mem_usage_record.used_in_state + _mem_usage_record.used_in_arena);
+
+ /// _hash_table_size_counter may be null if prepare failed.
+ if (_hash_table_size_counter) {
+ std::visit(
+ [&](auto&& agg_method) {
+ COUNTER_SET(_hash_table_size_counter, int64_t(agg_method.hash_table->size()));
+ },
+ _agg_data->method_variant);
+ }
+ return Base::close(state);
+}
+
+Status StreamingAggOperatorX::pull(RuntimeState* state, vectorized::Block* block,
+ SourceState& source_state) const {
+ auto& local_state = get_local_state(state);
+ if (!local_state._pre_aggregated_block->empty()) {
+ local_state._pre_aggregated_block->swap(*block);
+ } else {
+ RETURN_IF_ERROR(
+ local_state._executor->get_result(&local_state, state, block, source_state));
+ local_state.make_nullable_output_key(block);
+ // dispose the having clause, should not be execute in prestreaming agg
+ RETURN_IF_ERROR(
+ vectorized::VExprContext::filter_block(_conjuncts, block, block->columns()));
+ }
+ local_state.reached_limit(block, source_state);
+
+ return Status::OK();
+}
+
+Status StreamingAggOperatorX::push(RuntimeState* state, vectorized::Block* in_block,
+ SourceState source_state) const {
+ auto& local_state = get_local_state(state);
+ local_state._input_num_rows += in_block->rows();
+ if (in_block->rows() > 0) {
+ RETURN_IF_ERROR(local_state.do_pre_agg(in_block, local_state._pre_aggregated_block.get()));
+ }
+ in_block->clear_column_data(_child_x->row_desc().num_materialized_slots());
+ return Status::OK();
+}
+
+bool StreamingAggOperatorX::need_more_input_data(RuntimeState* state) const {
+ auto& local_state = get_local_state(state);
+ return local_state._pre_aggregated_block->empty() &&
+ local_state._child_source_state != SourceState::FINISHED;
+}
+
+} // namespace doris::pipeline
diff --git a/be/src/pipeline/exec/streaming_aggregation_operator.h b/be/src/pipeline/exec/streaming_aggregation_operator.h
new file mode 100644
index 00000000000..213e63459d5
--- /dev/null
+++ b/be/src/pipeline/exec/streaming_aggregation_operator.h
@@ -0,0 +1,231 @@
+// 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.
+
+#pragma once
+
+#include <stdint.h>
+
+#include <memory>
+
+#include "common/status.h"
+#include "pipeline/pipeline_x/operator.h"
+#include "util/runtime_profile.h"
+#include "vec/core/block.h"
+
+namespace doris {
+class RuntimeState;
+
+namespace pipeline {
+
+class StreamingAggOperatorX;
+
+class StreamingAggLocalState final : public PipelineXLocalState<FakeDependency> {
+public:
+ using Parent = StreamingAggOperatorX;
+ using Base = PipelineXLocalState<FakeDependency>;
+ ENABLE_FACTORY_CREATOR(StreamingAggLocalState);
+ StreamingAggLocalState(RuntimeState* state, OperatorXBase* parent);
+ ~StreamingAggLocalState() override = default;
+
+ Status init(RuntimeState* state, LocalStateInfo& info) override;
+ Status close(RuntimeState* state) override;
+ Status do_pre_agg(vectorized::Block* input_block, vectorized::Block* output_block);
+ void make_nullable_output_key(vectorized::Block* block);
+
+private:
+ friend class StreamingAggOperatorX;
+ template <typename LocalStateType>
+ friend class StatefulOperatorX;
+
+ size_t _memory_usage() const;
+ Status _pre_agg_with_serialized_key(doris::vectorized::Block* in_block,
+ doris::vectorized::Block* out_block);
+ bool _should_expand_preagg_hash_tables();
+ void _make_nullable_output_key(vectorized::Block* block);
+ Status _execute_without_key(vectorized::Block* block);
+ Status _merge_without_key(vectorized::Block* block);
+ void _update_memusage_without_key();
+ Status _execute_with_serialized_key(vectorized::Block* block);
+ Status _merge_with_serialized_key(vectorized::Block* block);
+ void _update_memusage_with_serialized_key();
+ void _init_hash_method(const vectorized::VExprContextSPtrs& probe_exprs);
+ Status _get_without_key_result(RuntimeState* state, vectorized::Block* block,
+ SourceState& source_state);
+ Status _serialize_without_key(RuntimeState* state, vectorized::Block* block,
+ SourceState& source_state);
+ Status _get_with_serialized_key_result(RuntimeState* state, vectorized::Block* block,
+ SourceState& source_state);
+ Status _serialize_with_serialized_key_result(RuntimeState* state, vectorized::Block* block,
+ SourceState& source_state);
+
+ template <bool limit, bool for_spill = false>
+ Status _merge_with_serialized_key_helper(vectorized::Block* block);
+ template <bool limit>
+ Status _execute_with_serialized_key_helper(vectorized::Block* block);
+ void _find_in_hash_table(vectorized::AggregateDataPtr* places,
+ vectorized::ColumnRawPtrs& key_columns, size_t num_rows);
+ int _get_slot_column_id(const vectorized::AggFnEvaluator* evaluator);
+ void _emplace_into_hash_table(vectorized::AggregateDataPtr* places,
+ vectorized::ColumnRawPtrs& key_columns, const size_t num_rows);
+ Status _create_agg_status(vectorized::AggregateDataPtr data);
+ size_t _get_hash_table_size();
+
+ RuntimeProfile::Counter* _queue_byte_size_counter = nullptr;
+ RuntimeProfile::Counter* _queue_size_counter = nullptr;
+ RuntimeProfile::Counter* _streaming_agg_timer = nullptr;
+ RuntimeProfile::Counter* _hash_table_compute_timer = nullptr;
+ RuntimeProfile::Counter* _hash_table_emplace_timer = nullptr;
+ RuntimeProfile::Counter* _hash_table_input_counter = nullptr;
+ RuntimeProfile::Counter* _build_timer = nullptr;
+ RuntimeProfile::Counter* _expr_timer = nullptr;
+ RuntimeProfile::Counter* _build_table_convert_timer = nullptr;
+ RuntimeProfile::Counter* _serialize_key_timer = nullptr;
+ RuntimeProfile::Counter* _merge_timer = nullptr;
+ RuntimeProfile::Counter* _serialize_data_timer = nullptr;
+ RuntimeProfile::Counter* _deserialize_data_timer = nullptr;
+ RuntimeProfile::Counter* _max_row_size_counter = nullptr;
+ RuntimeProfile::Counter* _hash_table_memory_usage = nullptr;
+ RuntimeProfile::HighWaterMarkCounter* _serialize_key_arena_memory_usage = nullptr;
+ RuntimeProfile::Counter* _hash_table_size_counter = nullptr;
+ RuntimeProfile::Counter* _get_results_timer = nullptr;
+ RuntimeProfile::Counter* _serialize_result_timer = nullptr;
+ RuntimeProfile::Counter* _hash_table_iterate_timer = nullptr;
+ RuntimeProfile::Counter* _insert_keys_to_column_timer = nullptr;
+
+ bool _should_expand_hash_table = true;
+ int64_t _cur_num_rows_returned = 0;
+ std::unique_ptr<vectorized::Arena> _agg_arena_pool = nullptr;
+ vectorized::AggregatedDataVariantsUPtr _agg_data = nullptr;
+ std::vector<vectorized::AggFnEvaluator*> _aggregate_evaluators;
+ // group by k1,k2
+ vectorized::VExprContextSPtrs _probe_expr_ctxs;
+ std::unique_ptr<vectorized::Arena> _agg_profile_arena = nullptr;
+ std::unique_ptr<vectorized::AggregateDataContainer> _aggregate_data_container = nullptr;
+ bool _should_limit_output = false;
+ bool _reach_limit = false;
+ size_t _input_num_rows = 0;
+
+ vectorized::PODArray<vectorized::AggregateDataPtr> _places;
+ std::vector<char> _deserialize_buffer;
+
+ struct ExecutorBase {
+ virtual Status execute(StreamingAggLocalState* local_state, vectorized::Block* block) = 0;
+ virtual void update_memusage(StreamingAggLocalState* local_state) = 0;
+ virtual Status get_result(StreamingAggLocalState* local_state, RuntimeState* state,
+ vectorized::Block* block, SourceState& source_state) = 0;
+ virtual ~ExecutorBase() = default;
+ };
+ template <bool WithoutKey, bool NeedToMerge, bool NeedFinalize>
+ struct Executor final : public ExecutorBase {
+ Status get_result(StreamingAggLocalState* local_state, RuntimeState* state,
+ vectorized::Block* block, SourceState& source_state) override {
+ if constexpr (WithoutKey) {
+ if constexpr (NeedFinalize) {
+ return local_state->_get_without_key_result(state, block, source_state);
+ } else {
+ return local_state->_serialize_without_key(state, block, source_state);
+ }
+ } else {
+ if constexpr (NeedFinalize) {
+ return local_state->_get_with_serialized_key_result(state, block, source_state);
+ } else {
+ return local_state->_serialize_with_serialized_key_result(state, block,
+ source_state);
+ }
+ }
+ }
+
+ Status execute(StreamingAggLocalState* 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(StreamingAggLocalState* local_state) override {
+ if constexpr (WithoutKey) {
+ local_state->_update_memusage_without_key();
+ } else {
+ local_state->_update_memusage_with_serialized_key();
+ }
+ }
+ };
+ std::unique_ptr<ExecutorBase> _executor = nullptr;
+
+ struct MemoryRecord {
+ MemoryRecord() : used_in_arena(0), used_in_state(0) {}
+ int64_t used_in_arena;
+ int64_t used_in_state;
+ };
+ MemoryRecord _mem_usage_record;
+ std::unique_ptr<vectorized::Block> _child_block = nullptr;
+ SourceState _child_source_state;
+ std::unique_ptr<vectorized::Block> _pre_aggregated_block = nullptr;
+ std::vector<vectorized::AggregateDataPtr> _values;
+};
+
+class StreamingAggOperatorX final : public StatefulOperatorX<StreamingAggLocalState> {
+public:
+ StreamingAggOperatorX(ObjectPool* pool, int operator_id, const TPlanNode& tnode,
+ const DescriptorTbl& descs);
+ ~StreamingAggOperatorX() override = default;
+ Status init(const TPlanNode& tnode, RuntimeState* state) override;
+ Status prepare(RuntimeState* state) override;
+ Status open(RuntimeState* state) override;
+ Status pull(RuntimeState* state, vectorized::Block* block,
+ SourceState& source_state) const override;
+ Status push(RuntimeState* state, vectorized::Block* input_block,
+ SourceState source_state) const override;
+ bool need_more_input_data(RuntimeState* state) const override;
+
+private:
+ friend class StreamingAggLocalState;
+ // may be we don't have to know the tuple id
+ TupleId _intermediate_tuple_id;
+ TupleDescriptor* _intermediate_tuple_desc = nullptr;
+
+ TupleId _output_tuple_id;
+ TupleDescriptor* _output_tuple_desc = nullptr;
+ bool _needs_finalize;
+ bool _is_merge;
+ const bool _is_first_phase;
+ size_t _align_aggregate_states = 1;
+ /// The offset to the n-th aggregate function in a row of aggregate functions.
+ vectorized::Sizes _offsets_of_aggregate_states;
+ /// The total size of the row from the aggregate functions.
+ size_t _total_size_of_aggregate_states = 0;
+ size_t _external_agg_bytes_threshold;
+ // group by k1,k2
+ vectorized::VExprContextSPtrs _probe_expr_ctxs;
+ std::vector<vectorized::AggFnEvaluator*> _aggregate_evaluators;
+ bool _can_short_circuit = false;
+ std::vector<size_t> _make_nullable_keys;
+ size_t _spill_partition_count_bits;
+ bool _have_conjuncts;
+};
+
+} // namespace pipeline
+} // namespace doris
diff --git a/be/src/pipeline/exec/streaming_aggregation_sink_operator.cpp b/be/src/pipeline/exec/streaming_aggregation_sink_operator.cpp
index 41d46e2f3ad..5805b209f4f 100644
--- a/be/src/pipeline/exec/streaming_aggregation_sink_operator.cpp
+++ b/be/src/pipeline/exec/streaming_aggregation_sink_operator.cpp
@@ -90,302 +90,4 @@ StreamingAggSinkOperatorBuilder::StreamingAggSinkOperatorBuilder(int32_t id, Exe
OperatorPtr StreamingAggSinkOperatorBuilder::build_operator() {
return std::make_shared<StreamingAggSinkOperator>(this, _node, _data_queue);
}
-
-/// The minimum reduction factor (input rows divided by output rows) to grow hash tables
-/// in a streaming preaggregation, given that the hash tables are currently the given
-/// size or above. The sizes roughly correspond to hash table sizes where the bucket
-/// arrays will fit in a cache level. Intuitively, we don't want the working set of the
-/// aggregation to expand to the next level of cache unless we're reducing the input
-/// enough to outweigh the increased memory latency we'll incur for each hash table
-/// lookup.
-///
-/// Note that the current reduction achieved is not always a good estimate of the
-/// final reduction. It may be biased either way depending on the ordering of the
-/// input. If the input order is random, we will underestimate the final reduction
-/// factor because the probability of a row having the same key as a previous row
-/// increases as more input is processed. If the input order is correlated with the
-/// key, skew may bias the estimate. If high cardinality keys appear first, we
-/// may overestimate and if low cardinality keys appear first, we underestimate.
-/// To estimate the eventual reduction achieved, we estimate the final reduction
-/// using the planner's estimated input cardinality and the assumption that input
-/// is in a random order. This means that we assume that the reduction factor will
-/// increase over time.
-struct StreamingHtMinReductionEntry {
- // Use 'streaming_ht_min_reduction' if the total size of hash table bucket directories in
- // bytes is greater than this threshold.
- int min_ht_mem;
- // The minimum reduction factor to expand the hash tables.
- double streaming_ht_min_reduction;
-};
-
-// TODO: experimentally tune these values and also programmatically get the cache size
-// of the machine that we're running on.
-static constexpr StreamingHtMinReductionEntry STREAMING_HT_MIN_REDUCTION[] = {
- // Expand up to L2 cache always.
- {0, 0.0},
- // Expand into L3 cache if we look like we're getting some reduction.
- // At present, The L2 cache is generally 1024k or more
- {1024 * 1024, 1.1},
- // Expand into main memory if we're getting a significant reduction.
- // The L3 cache is generally 16MB or more
- {16 * 1024 * 1024, 2.0},
-};
-
-static constexpr int STREAMING_HT_MIN_REDUCTION_SIZE =
- sizeof(STREAMING_HT_MIN_REDUCTION) / sizeof(STREAMING_HT_MIN_REDUCTION[0]);
-
-StreamingAggSinkLocalState::StreamingAggSinkLocalState(DataSinkOperatorXBase* parent,
- RuntimeState* state)
- : Base(parent, state),
- _queue_byte_size_counter(nullptr),
- _queue_size_counter(nullptr),
- _streaming_agg_timer(nullptr) {}
-
-Status StreamingAggSinkLocalState::init(RuntimeState* state, LocalSinkStateInfo& info) {
- RETURN_IF_ERROR(Base::init(state, info));
- _queue_byte_size_counter = ADD_COUNTER(profile(), "MaxSizeInBlockQueue", TUnit::BYTES);
- _queue_size_counter = ADD_COUNTER(profile(), "MaxSizeOfBlockQueue", TUnit::UNIT);
- _streaming_agg_timer = ADD_TIMER(profile(), "StreamingAggTime");
- return Status::OK();
-}
-
-Status StreamingAggSinkLocalState::do_pre_agg(vectorized::Block* input_block,
- vectorized::Block* output_block) {
- RETURN_IF_ERROR(_pre_agg_with_serialized_key(input_block, output_block));
-
- // pre stream agg need use _num_row_return to decide whether to do pre stream agg
- _num_rows_returned += output_block->rows();
- _make_nullable_output_key(output_block);
- // COUNTER_SET(_rows_returned_counter, _num_rows_returned);
- _executor->update_memusage(this);
- return Status::OK();
-}
-
-bool StreamingAggSinkLocalState::_should_expand_preagg_hash_tables() {
- if (!_should_expand_hash_table) {
- return false;
- }
-
- return std::visit(
- [&](auto&& agg_method) -> bool {
- auto& hash_tbl = *agg_method.hash_table;
- auto [ht_mem, ht_rows] =
- std::pair {hash_tbl.get_buffer_size_in_bytes(), hash_tbl.size()};
-
- // Need some rows in tables to have valid statistics.
- if (ht_rows == 0) {
- return true;
- }
-
- // Find the appropriate reduction factor in our table for the current hash table sizes.
- int cache_level = 0;
- while (cache_level + 1 < STREAMING_HT_MIN_REDUCTION_SIZE &&
- ht_mem >= STREAMING_HT_MIN_REDUCTION[cache_level + 1].min_ht_mem) {
- ++cache_level;
- }
-
- // Compare the number of rows in the hash table with the number of input rows that
- // were aggregated into it. Exclude passed through rows from this calculation since
- // they were not in hash tables.
- const int64_t input_rows = _shared_state->input_num_rows;
- const int64_t aggregated_input_rows = input_rows - _num_rows_returned;
- // TODO chenhao
- // const int64_t expected_input_rows = estimated_input_cardinality_ - num_rows_returned_;
- double current_reduction = static_cast<double>(aggregated_input_rows) / ht_rows;
-
- // TODO: workaround for IMPALA-2490: subplan node rows_returned counter may be
- // inaccurate, which could lead to a divide by zero below.
- if (aggregated_input_rows <= 0) {
- return true;
- }
-
- // Extrapolate the current reduction factor (r) using the formula
- // R = 1 + (N / n) * (r - 1), where R is the reduction factor over the full input data
- // set, N is the number of input rows, excluding passed-through rows, and n is the
- // number of rows inserted or merged into the hash tables. This is a very rough
- // approximation but is good enough to be useful.
- // TODO: consider collecting more statistics to better estimate reduction.
- // double estimated_reduction = aggregated_input_rows >= expected_input_rows
- // ? current_reduction
- // : 1 + (expected_input_rows / aggregated_input_rows) * (current_reduction - 1);
- double min_reduction =
- STREAMING_HT_MIN_REDUCTION[cache_level].streaming_ht_min_reduction;
-
- // COUNTER_SET(preagg_estimated_reduction_, estimated_reduction);
- // COUNTER_SET(preagg_streaming_ht_min_reduction_, min_reduction);
- // return estimated_reduction > min_reduction;
- _should_expand_hash_table = current_reduction > min_reduction;
- return _should_expand_hash_table;
- },
- _agg_data->method_variant);
-}
-
-Status StreamingAggSinkLocalState::_pre_agg_with_serialized_key(
- doris::vectorized::Block* in_block, doris::vectorized::Block* out_block) {
- SCOPED_TIMER(_build_timer);
- DCHECK(!_shared_state->probe_expr_ctxs.empty());
-
- size_t key_size = _shared_state->probe_expr_ctxs.size();
- vectorized::ColumnRawPtrs key_columns(key_size);
- {
- SCOPED_TIMER(_expr_timer);
- for (size_t i = 0; i < key_size; ++i) {
- int result_column_id = -1;
- RETURN_IF_ERROR(
- _shared_state->probe_expr_ctxs[i]->execute(in_block, &result_column_id));
- in_block->get_by_position(result_column_id).column =
- in_block->get_by_position(result_column_id)
- .column->convert_to_full_column_if_const();
- key_columns[i] = in_block->get_by_position(result_column_id).column.get();
- }
- }
-
- int rows = in_block->rows();
- _places.resize(rows);
-
- // Stop expanding hash tables if we're not reducing the input sufficiently. As our
- // hash tables expand out of each level of cache hierarchy, every hash table lookup
- // will take longer. We also may not be able to expand hash tables because of memory
- // pressure. In either case we should always use the remaining space in the hash table
- // to avoid wasting memory.
- // But for fixed hash map, it never need to expand
- bool ret_flag = false;
- RETURN_IF_ERROR(std::visit(
- [&](auto&& agg_method) -> Status {
- if (auto& hash_tbl = *agg_method.hash_table;
- hash_tbl.add_elem_size_overflow(rows)) {
- /// If too much memory is used during the pre-aggregation stage,
- /// it is better to output the data directly without performing further aggregation.
- const bool used_too_much_memory =
- (_parent->cast<StreamingAggSinkOperatorX>()
- ._external_agg_bytes_threshold > 0 &&
- _memory_usage() > _parent->cast<StreamingAggSinkOperatorX>()
- ._external_agg_bytes_threshold);
- // do not try to do agg, just init and serialize directly return the out_block
- if (!_should_expand_preagg_hash_tables() || used_too_much_memory) {
- SCOPED_TIMER(_streaming_agg_timer);
- ret_flag = true;
-
- // will serialize value data to string column.
- // non-nullable column(id in `_make_nullable_keys`)
- // will be converted to nullable.
- bool mem_reuse =
- _shared_state->make_nullable_keys.empty() && out_block->mem_reuse();
-
- std::vector<vectorized::DataTypePtr> data_types;
- vectorized::MutableColumns value_columns;
- for (int i = 0; i < _shared_state->aggregate_evaluators.size(); ++i) {
- auto data_type = _shared_state->aggregate_evaluators[i]
- ->function()
- ->get_serialized_type();
- if (mem_reuse) {
- value_columns.emplace_back(
- std::move(*out_block->get_by_position(i + key_size).column)
- .mutate());
- } else {
- // slot type of value it should always be string type
- value_columns.emplace_back(_shared_state->aggregate_evaluators[i]
- ->function()
- ->create_serialize_column());
- }
- data_types.emplace_back(data_type);
- }
-
- for (int i = 0; i != _shared_state->aggregate_evaluators.size(); ++i) {
- SCOPED_TIMER(_serialize_data_timer);
- RETURN_IF_ERROR(_shared_state->aggregate_evaluators[i]
- ->streaming_agg_serialize_to_column(
- in_block, value_columns[i], rows,
- _agg_arena_pool));
- }
-
- if (!mem_reuse) {
- vectorized::ColumnsWithTypeAndName columns_with_schema;
- for (int i = 0; i < key_size; ++i) {
- columns_with_schema.emplace_back(
- key_columns[i]->clone_resized(rows),
- _shared_state->probe_expr_ctxs[i]->root()->data_type(),
- _shared_state->probe_expr_ctxs[i]->root()->expr_name());
- }
- for (int i = 0; i < value_columns.size(); ++i) {
- columns_with_schema.emplace_back(std::move(value_columns[i]),
- data_types[i], "");
- }
- out_block->swap(vectorized::Block(columns_with_schema));
- } else {
- for (int i = 0; i < key_size; ++i) {
- std::move(*out_block->get_by_position(i).column)
- .mutate()
- ->insert_range_from(*key_columns[i], 0, rows);
- }
- }
- }
- }
- return Status::OK();
- },
- _agg_data->method_variant));
-
- if (!ret_flag) {
- RETURN_IF_CATCH_EXCEPTION(_emplace_into_hash_table(_places.data(), key_columns, rows));
-
- for (int i = 0; i < _shared_state->aggregate_evaluators.size(); ++i) {
- RETURN_IF_ERROR(_shared_state->aggregate_evaluators[i]->execute_batch_add(
- in_block, _shared_state->offsets_of_aggregate_states[i], _places.data(),
- _agg_arena_pool, _should_expand_hash_table));
- }
- }
-
- return Status::OK();
-}
-
-StreamingAggSinkOperatorX::StreamingAggSinkOperatorX(ObjectPool* pool, int operator_id,
- const TPlanNode& tnode,
- const DescriptorTbl& descs)
- : AggSinkOperatorX<StreamingAggSinkLocalState>(pool, operator_id, tnode, descs, true) {}
-
-Status StreamingAggSinkOperatorX::init(const TPlanNode& tnode, RuntimeState* state) {
- RETURN_IF_ERROR(AggSinkOperatorX<StreamingAggSinkLocalState>::init(tnode, state));
- _name = "STREAMING_AGGREGATION_SINK_OPERATOR";
- return Status::OK();
-}
-
-Status StreamingAggSinkOperatorX::sink(RuntimeState* state, vectorized::Block* in_block,
- SourceState source_state) {
- auto& local_state = get_local_state(state);
- SCOPED_TIMER(local_state.exec_time_counter());
- COUNTER_UPDATE(local_state.rows_input_counter(), (int64_t)in_block->rows());
- local_state._shared_state->input_num_rows += in_block->rows();
- Status ret = Status::OK();
- if (in_block && in_block->rows() > 0) {
- auto block_from_ctx = local_state._shared_state->data_queue->get_free_block();
- RETURN_IF_ERROR(local_state.do_pre_agg(in_block, block_from_ctx.get()));
- if (block_from_ctx->rows() == 0) {
- local_state._shared_state->data_queue->push_free_block(std::move(block_from_ctx));
- } else {
- local_state._shared_state->data_queue->push_block(std::move(block_from_ctx));
- }
- }
-
- if (UNLIKELY(source_state == SourceState::FINISHED)) {
- local_state._shared_state->data_queue->set_finish();
- }
- return Status::OK();
-}
-
-Status StreamingAggSinkLocalState::close(RuntimeState* state, Status exec_status) {
- if (_closed) {
- return Status::OK();
- }
- if (_shared_state->data_queue && !_shared_state->data_queue->is_finish()) {
- // finish should be set, if not set here means error.
- _shared_state->data_queue->set_canceled();
- }
- if (_shared_state->data_queue) {
- COUNTER_SET(_queue_size_counter, _shared_state->data_queue->max_size_of_queue());
- COUNTER_SET(_queue_byte_size_counter, _shared_state->data_queue->max_bytes_in_queue());
- }
- _preagg_block.clear();
- return Base::close(state, exec_status);
-}
-
} // namespace doris::pipeline
diff --git a/be/src/pipeline/exec/streaming_aggregation_sink_operator.h b/be/src/pipeline/exec/streaming_aggregation_sink_operator.h
index bd2640517cf..99e94e2d585 100644
--- a/be/src/pipeline/exec/streaming_aggregation_sink_operator.h
+++ b/be/src/pipeline/exec/streaming_aggregation_sink_operator.h
@@ -72,66 +72,5 @@ private:
std::shared_ptr<DataQueue> _data_queue;
};
-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<StreamingAggSinkDependency, StreamingAggSinkLocalState> {
-public:
- using Parent = StreamingAggSinkOperatorX;
- using Base = AggSinkLocalState<StreamingAggSinkDependency, StreamingAggSinkLocalState>;
- ENABLE_FACTORY_CREATOR(StreamingAggSinkLocalState);
- StreamingAggSinkLocalState(DataSinkOperatorXBase* parent, RuntimeState* state);
- ~StreamingAggSinkLocalState() override = default;
-
- Status init(RuntimeState* state, LocalSinkStateInfo& info) override;
- Status close(RuntimeState* state, Status exec_status) override;
- Status do_pre_agg(vectorized::Block* input_block, vectorized::Block* output_block);
-
-private:
- friend class StreamingAggSinkOperatorX;
-
- Status _pre_agg_with_serialized_key(doris::vectorized::Block* in_block,
- doris::vectorized::Block* out_block);
- bool _should_expand_preagg_hash_tables();
- void _make_nullable_output_key(vectorized::Block* block) {
- if (block->rows() != 0) {
- auto& shared_state = *Base ::_shared_state;
- for (auto cid : shared_state.make_nullable_keys) {
- block->get_by_position(cid).column =
- make_nullable(block->get_by_position(cid).column);
- block->get_by_position(cid).type = make_nullable(block->get_by_position(cid).type);
- }
- }
- }
-
- RuntimeProfile::Counter* _queue_byte_size_counter = nullptr;
- RuntimeProfile::Counter* _queue_size_counter = nullptr;
- RuntimeProfile::Counter* _streaming_agg_timer = nullptr;
-
- bool _should_expand_hash_table = true;
- int64_t _num_rows_returned = 0;
-};
-
-class StreamingAggSinkOperatorX final : public AggSinkOperatorX<StreamingAggSinkLocalState> {
-public:
- StreamingAggSinkOperatorX(ObjectPool* pool, int operator_id, const TPlanNode& tnode,
- const DescriptorTbl& descs);
- ~StreamingAggSinkOperatorX() override = default;
- Status init(const TPlanNode& tnode, RuntimeState* state) override;
- Status sink(RuntimeState* state, vectorized::Block* in_block,
- SourceState source_state) override;
- DataDistribution required_data_distribution() const override {
- return DataSinkOperatorX<StreamingAggSinkLocalState>::required_data_distribution();
- }
-};
-
} // namespace pipeline
} // namespace doris
diff --git a/be/src/pipeline/exec/streaming_aggregation_source_operator.cpp b/be/src/pipeline/exec/streaming_aggregation_source_operator.cpp
index 44991e2e14f..a6585720835 100644
--- a/be/src/pipeline/exec/streaming_aggregation_source_operator.cpp
+++ b/be/src/pipeline/exec/streaming_aggregation_source_operator.cpp
@@ -73,44 +73,5 @@ OperatorPtr StreamingAggSourceOperatorBuilder::build_operator() {
return std::make_shared<StreamingAggSourceOperator>(this, _node, _data_queue);
}
-StreamingAggSourceOperatorX::StreamingAggSourceOperatorX(ObjectPool* pool, const TPlanNode& tnode,
- int operator_id,
- const DescriptorTbl& descs)
- : Base(pool, tnode, operator_id, descs, true) {}
-
-Status StreamingAggSourceOperatorX::get_block(RuntimeState* state, vectorized::Block* block,
- SourceState& source_state) {
- auto& local_state = get_local_state(state);
- SCOPED_TIMER(local_state.exec_time_counter());
- if (!local_state._shared_state->data_queue->data_exhausted()) {
- std::unique_ptr<vectorized::Block> agg_block = nullptr;
- DCHECK(local_state._dependency->is_blocked_by() == nullptr);
- RETURN_IF_ERROR(local_state._shared_state->data_queue->get_block_from_queue(&agg_block));
-
- if (local_state._shared_state->data_queue->data_exhausted()) {
- RETURN_IF_ERROR(Base::get_block(state, block, source_state));
- } else if (agg_block) {
- block->swap(*agg_block);
- agg_block->clear_column_data(row_desc().num_materialized_slots());
- local_state._shared_state->data_queue->push_free_block(std::move(agg_block));
- } else if (local_state._shared_state->data_queue->is_all_finish()) {
- source_state = SourceState::FINISHED;
- } else {
- return Status::InternalError("Something wrong in StreamingAggSource: {}",
- Base::debug_string(0));
- }
- } else {
- RETURN_IF_ERROR(Base::get_block(state, block, source_state));
- }
-
- return Status::OK();
-}
-
-Status StreamingAggSourceOperatorX::init(const TPlanNode& tnode, RuntimeState* state) {
- RETURN_IF_ERROR(Base::init(tnode, state));
- _op_name = "STREAMING_AGGREGATION_OPERATOR";
- return Status::OK();
-}
-
} // namespace pipeline
} // namespace doris
diff --git a/be/src/pipeline/exec/streaming_aggregation_source_operator.h b/be/src/pipeline/exec/streaming_aggregation_source_operator.h
index 51e3c9ed972..89dbaab058a 100644
--- a/be/src/pipeline/exec/streaming_aggregation_source_operator.h
+++ b/be/src/pipeline/exec/streaming_aggregation_source_operator.h
@@ -60,18 +60,5 @@ private:
std::shared_ptr<DataQueue> _data_queue;
};
-class StreamingAggSourceOperatorX final : public AggSourceOperatorX {
-public:
- using Base = AggSourceOperatorX;
- StreamingAggSourceOperatorX(ObjectPool* pool, const TPlanNode& tnode, int operator_id,
- const DescriptorTbl& descs);
- ~StreamingAggSourceOperatorX() = default;
-
- Status init(const TPlanNode& tnode, RuntimeState* state) override;
-
- Status get_block(RuntimeState* state, vectorized::Block* block,
- SourceState& source_state) override;
-};
-
} // namespace pipeline
} // namespace doris
diff --git a/be/src/pipeline/pipeline_x/dependency.h b/be/src/pipeline/pipeline_x/dependency.h
index b2ca808e209..326185c9d2b 100644
--- a/be/src/pipeline/pipeline_x/dependency.h
+++ b/be/src/pipeline/pipeline_x/dependency.h
@@ -341,7 +341,6 @@ 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, 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,8 +401,7 @@ public:
struct UnionSharedState : public BasicSharedState {
public:
- UnionSharedState(int child_count = 1)
- : data_queue(child_count, false), _child_count(child_count) {};
+ UnionSharedState(int child_count = 1) : data_queue(child_count), _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/local_exchange/local_exchange_source_operator.h b/be/src/pipeline/pipeline_x/local_exchange/local_exchange_source_operator.h
index 63d71bbe08b..b62ec1a8fb9 100644
--- a/be/src/pipeline/pipeline_x/local_exchange/local_exchange_source_operator.h
+++ b/be/src/pipeline/pipeline_x/local_exchange/local_exchange_source_operator.h
@@ -78,7 +78,7 @@ public:
return _child_x->intermediate_row_desc();
}
RowDescriptor& row_descriptor() override { return _child_x->row_descriptor(); }
- const RowDescriptor& row_desc() override { return _child_x->row_desc(); }
+ const RowDescriptor& row_desc() const override { return _child_x->row_desc(); }
Status get_block(RuntimeState* state, vectorized::Block* block,
SourceState& source_state) override;
diff --git a/be/src/pipeline/pipeline_x/operator.cpp b/be/src/pipeline/pipeline_x/operator.cpp
index 86c7f885792..890c4298bb0 100644
--- a/be/src/pipeline/pipeline_x/operator.cpp
+++ b/be/src/pipeline/pipeline_x/operator.cpp
@@ -29,7 +29,7 @@
#include "pipeline/exec/analytic_source_operator.h"
#include "pipeline/exec/assert_num_rows_operator.h"
#include "pipeline/exec/datagen_operator.h"
-#include "pipeline/exec/distinct_streaming_aggregation_sink_operator.h"
+#include "pipeline/exec/distinct_streaming_aggregation_operator.h"
#include "pipeline/exec/empty_set_operator.h"
#include "pipeline/exec/es_scan_operator.h"
#include "pipeline/exec/exchange_sink_operator.h"
@@ -59,8 +59,7 @@
#include "pipeline/exec/set_source_operator.h"
#include "pipeline/exec/sort_sink_operator.h"
#include "pipeline/exec/sort_source_operator.h"
-#include "pipeline/exec/streaming_aggregation_sink_operator.h"
-#include "pipeline/exec/streaming_aggregation_source_operator.h"
+#include "pipeline/exec/streaming_aggregation_operator.h"
#include "pipeline/exec/table_function_operator.h"
#include "pipeline/exec/union_sink_operator.h"
#include "pipeline/exec/union_source_operator.h"
@@ -72,10 +71,6 @@
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;
@@ -601,9 +596,7 @@ DECLARE_OPERATOR_X(OlapTableSinkV2LocalState)
DECLARE_OPERATOR_X(AnalyticSinkLocalState)
DECLARE_OPERATOR_X(SortSinkLocalState)
DECLARE_OPERATOR_X(LocalExchangeSinkLocalState)
-DECLARE_OPERATOR_X(BlockingAggSinkLocalState)
-DECLARE_OPERATOR_X(StreamingAggSinkLocalState)
-DECLARE_OPERATOR_X(DistinctStreamingAggSinkLocalState)
+DECLARE_OPERATOR_X(AggSinkLocalState)
DECLARE_OPERATOR_X(ExchangeSinkLocalState)
DECLARE_OPERATOR_X(NestedLoopJoinBuildSinkLocalState)
DECLARE_OPERATOR_X(UnionSinkLocalState)
@@ -648,6 +641,8 @@ template class StreamingOperatorX<SelectLocalState>;
template class StatefulOperatorX<HashJoinProbeLocalState>;
template class StatefulOperatorX<RepeatLocalState>;
+template class StatefulOperatorX<StreamingAggLocalState>;
+template class StatefulOperatorX<DistinctStreamingAggLocalState>;
template class StatefulOperatorX<NestedLoopJoinProbeLocalState>;
template class StatefulOperatorX<TableFunctionLocalState>;
@@ -656,8 +651,6 @@ 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>;
diff --git a/be/src/pipeline/pipeline_x/operator.h b/be/src/pipeline/pipeline_x/operator.h
index 61afe40e126..848ed5cd001 100644
--- a/be/src/pipeline/pipeline_x/operator.h
+++ b/be/src/pipeline/pipeline_x/operator.h
@@ -278,7 +278,7 @@ public:
[[nodiscard]] int64_t limit() const { return _limit; }
- [[nodiscard]] virtual const RowDescriptor& row_desc() override {
+ [[nodiscard]] const RowDescriptor& row_desc() const override {
return _output_row_descriptor ? *_output_row_descriptor : _row_descriptor;
}
diff --git a/be/src/pipeline/pipeline_x/pipeline_x_fragment_context.cpp b/be/src/pipeline/pipeline_x/pipeline_x_fragment_context.cpp
index 67c5da97063..38b7b7457fe 100644
--- a/be/src/pipeline/pipeline_x/pipeline_x_fragment_context.cpp
+++ b/be/src/pipeline/pipeline_x/pipeline_x_fragment_context.cpp
@@ -44,8 +44,7 @@
#include "pipeline/exec/assert_num_rows_operator.h"
#include "pipeline/exec/data_queue.h"
#include "pipeline/exec/datagen_operator.h"
-#include "pipeline/exec/distinct_streaming_aggregation_sink_operator.h"
-#include "pipeline/exec/distinct_streaming_aggregation_source_operator.h"
+#include "pipeline/exec/distinct_streaming_aggregation_operator.h"
#include "pipeline/exec/empty_set_operator.h"
#include "pipeline/exec/es_scan_operator.h"
#include "pipeline/exec/exchange_sink_operator.h"
@@ -76,8 +75,7 @@
#include "pipeline/exec/set_source_operator.h"
#include "pipeline/exec/sort_sink_operator.h"
#include "pipeline/exec/sort_source_operator.h"
-#include "pipeline/exec/streaming_aggregation_sink_operator.h"
-#include "pipeline/exec/streaming_aggregation_source_operator.h"
+#include "pipeline/exec/streaming_aggregation_operator.h"
#include "pipeline/exec/table_function_operator.h"
#include "pipeline/exec/union_sink_operator.h"
#include "pipeline/exec/union_source_operator.h"
@@ -950,39 +948,12 @@ Status PipelineXFragmentContext::_create_operator(ObjectPool* pool, const TPlanN
}
case TPlanNodeType::AGGREGATION_NODE: {
if (tnode.agg_node.aggregate_functions.empty()) {
- op.reset(new DistinctStreamingAggSourceOperatorX(pool, tnode, next_operator_id(),
- descs));
+ op.reset(new DistinctStreamingAggOperatorX(pool, next_operator_id(), tnode, descs));
RETURN_IF_ERROR(cur_pipe->add_operator(op));
-
- const auto downstream_pipeline_id = cur_pipe->id();
- if (_dag.find(downstream_pipeline_id) == _dag.end()) {
- _dag.insert({downstream_pipeline_id, {}});
- }
- cur_pipe = add_pipeline(cur_pipe);
- _dag[downstream_pipeline_id].push_back(cur_pipe->id());
- DataSinkOperatorXPtr sink;
- sink.reset(new DistinctStreamingAggSinkOperatorX(pool, next_sink_operator_id(), tnode,
- descs));
- sink->set_dests_id({op->operator_id()});
- RETURN_IF_ERROR(cur_pipe->set_sink(sink));
- RETURN_IF_ERROR(cur_pipe->sink_x()->init(tnode, _runtime_state.get()));
} else if (tnode.agg_node.__isset.use_streaming_preaggregation &&
tnode.agg_node.use_streaming_preaggregation) {
- op.reset(new StreamingAggSourceOperatorX(pool, tnode, next_operator_id(), descs));
+ op.reset(new StreamingAggOperatorX(pool, next_operator_id(), tnode, descs));
RETURN_IF_ERROR(cur_pipe->add_operator(op));
-
- const auto downstream_pipeline_id = cur_pipe->id();
- if (_dag.find(downstream_pipeline_id) == _dag.end()) {
- _dag.insert({downstream_pipeline_id, {}});
- }
- cur_pipe = add_pipeline(cur_pipe);
- _dag[downstream_pipeline_id].push_back(cur_pipe->id());
- DataSinkOperatorXPtr sink;
- sink.reset(new StreamingAggSinkOperatorX(pool, next_sink_operator_id(), tnode, descs));
- sink->set_dests_id({op->operator_id()});
- RETURN_IF_ERROR(cur_pipe->set_sink(sink));
- RETURN_IF_ERROR(cur_pipe->sink_x()->init(tnode, _runtime_state.get()));
-
} else {
op.reset(new AggSourceOperatorX(pool, tnode, next_operator_id(), descs));
RETURN_IF_ERROR(cur_pipe->add_operator(op));
@@ -995,7 +966,7 @@ Status PipelineXFragmentContext::_create_operator(ObjectPool* pool, const TPlanN
_dag[downstream_pipeline_id].push_back(cur_pipe->id());
DataSinkOperatorXPtr sink;
- sink.reset(new AggSinkOperatorX<>(pool, next_sink_operator_id(), tnode, descs));
+ sink.reset(new AggSinkOperatorX(pool, next_sink_operator_id(), tnode, descs));
sink->set_dests_id({op->operator_id()});
RETURN_IF_ERROR(cur_pipe->set_sink(sink));
RETURN_IF_ERROR(cur_pipe->sink_x()->init(tnode, _runtime_state.get()));
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org