This is an automated email from the ASF dual-hosted git repository. Mryange pushed a commit to branch groupjoin in repository https://gitbox.apache.org/repos/asf/doris.git
commit 8eb13c229e8386486d18b6a60942a661e60a2bc7 Author: Mryange <[email protected]> AuthorDate: Mon Jun 22 10:29:16 2026 +0800 be part --- be/src/exec/common/groupjoin_utils.h | 133 +++++++ be/src/exec/operator/groupjoin_build_sink.cpp | 265 +++++++++++++ be/src/exec/operator/groupjoin_build_sink.h | 104 +++++ be/src/exec/operator/groupjoin_operator_utils.cpp | 415 +++++++++++++++++++ be/src/exec/operator/groupjoin_operator_utils.h | 82 ++++ be/src/exec/operator/groupjoin_probe_operator.cpp | 251 ++++++++++++ be/src/exec/operator/groupjoin_probe_operator.h | 109 +++++ be/src/exec/operator/operator.cpp | 7 + be/src/exec/pipeline/dependency.cpp | 5 + be/src/exec/pipeline/dependency.h | 27 ++ be/src/exec/pipeline/pipeline_fragment_context.cpp | 22 ++ .../runtime_filter_producer_helper_groupjoin.h | 94 +++++ be/src/exprs/aggregate/aggregate_function.h | 39 +- be/src/exprs/aggregate/aggregate_function_count.h | 51 +++ .../exprs/aggregate/aggregate_function_min_max.h | 13 + .../exprs/aggregate/aggregate_function_null_v2.h | 63 +++ be/src/exprs/aggregate/aggregate_function_sum.h | 40 ++ be/src/exprs/vectorized_agg_fn.cpp | 9 + be/src/exprs/vectorized_agg_fn.h | 4 + .../operator/groupjoin_operator_utils_test.cpp | 154 ++++++++ .../aggregate/aggregate_function_repeat_test.cpp | 438 +++++++++++++++++++++ 21 files changed, 2324 insertions(+), 1 deletion(-) diff --git a/be/src/exec/common/groupjoin_utils.h b/be/src/exec/common/groupjoin_utils.h new file mode 100644 index 00000000000..c257122aed5 --- /dev/null +++ b/be/src/exec/common/groupjoin_utils.h @@ -0,0 +1,133 @@ +// 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 <memory> +#include <variant> +#include <vector> + +#include "common/exception.h" +#include "exec/common/columns_hashing.h" +#include "exec/common/hash_table/hash_crc32_return32.h" +#include "exec/common/hash_table/hash_key_type.h" +#include "exec/common/hash_table/hash_map_context.h" +#include "exec/common/hash_table/hash_map_util.h" +#include "exec/common/hash_table/ph_hash_map.h" +#include "exec/common/hash_table/string_hash_map.h" +#include "exprs/aggregate/aggregate_function.h" + +namespace doris { + +struct GroupJoinEntry { + uint64_t build_count = 0; + uint64_t probe_count = 0; + AggregateDataPtr agg_states = nullptr; +}; + +template <typename T> +using GroupJoinData = PHHashMap<T, GroupJoinEntry*, HashCRC32<T>>; + +using GroupJoinDataWithStringKey = PHHashMap<StringRef, GroupJoinEntry*>; +using GroupJoinDataWithShortStringKey = StringHashMap<GroupJoinEntry*>; + +template <class T> +using GroupJoinPrimaryHashTableContext = MethodOneNumber<T, GroupJoinData<T>>; + +template <class Key> +using GroupJoinFixedKeyHashTableContext = MethodKeysFixed<GroupJoinData<Key>>; + +using GroupJoinSerializedHashTableContext = MethodSerialized<GroupJoinDataWithStringKey>; +using GroupJoinMethodOneString = MethodStringNoCache<GroupJoinDataWithShortStringKey>; + +using GroupJoinMethodVariants = std::variant< + std::monostate, GroupJoinSerializedHashTableContext, + GroupJoinPrimaryHashTableContext<UInt8>, GroupJoinPrimaryHashTableContext<UInt16>, + GroupJoinPrimaryHashTableContext<UInt32>, GroupJoinPrimaryHashTableContext<UInt64>, + GroupJoinPrimaryHashTableContext<UInt128>, GroupJoinPrimaryHashTableContext<UInt256>, + GroupJoinFixedKeyHashTableContext<UInt64>, GroupJoinFixedKeyHashTableContext<UInt72>, + GroupJoinFixedKeyHashTableContext<UInt96>, GroupJoinFixedKeyHashTableContext<UInt104>, + GroupJoinFixedKeyHashTableContext<UInt128>, GroupJoinFixedKeyHashTableContext<UInt136>, + GroupJoinFixedKeyHashTableContext<UInt256>, GroupJoinMethodOneString>; + +struct GroupJoinDataVariants + : public DataVariants<GroupJoinMethodVariants, MethodSingleNullableColumn, MethodOneNumber, + DataWithNullKey> { + void init(const std::vector<DataTypePtr>& data_types, HashKeyType type) { + switch (type) { + case HashKeyType::serialized: + method_variant.emplace<GroupJoinSerializedHashTableContext>(); + break; + case HashKeyType::int8_key: + method_variant.emplace<GroupJoinPrimaryHashTableContext<UInt8>>(); + break; + case HashKeyType::int16_key: + method_variant.emplace<GroupJoinPrimaryHashTableContext<UInt16>>(); + break; + case HashKeyType::int32_key: + method_variant.emplace<GroupJoinPrimaryHashTableContext<UInt32>>(); + break; + case HashKeyType::int64_key: + method_variant.emplace<GroupJoinPrimaryHashTableContext<UInt64>>(); + break; + case HashKeyType::int128_key: + method_variant.emplace<GroupJoinPrimaryHashTableContext<UInt128>>(); + break; + case HashKeyType::int256_key: + method_variant.emplace<GroupJoinPrimaryHashTableContext<UInt256>>(); + break; + case HashKeyType::string_key: + method_variant.emplace<GroupJoinMethodOneString>(); + break; + case HashKeyType::fixed64: + method_variant.emplace<GroupJoinFixedKeyHashTableContext<UInt64>>( + get_key_sizes(data_types)); + break; + case HashKeyType::fixed72: + method_variant.emplace<GroupJoinFixedKeyHashTableContext<UInt72>>( + get_key_sizes(data_types)); + break; + case HashKeyType::fixed96: + method_variant.emplace<GroupJoinFixedKeyHashTableContext<UInt96>>( + get_key_sizes(data_types)); + break; + case HashKeyType::fixed104: + method_variant.emplace<GroupJoinFixedKeyHashTableContext<UInt104>>( + get_key_sizes(data_types)); + break; + case HashKeyType::fixed128: + method_variant.emplace<GroupJoinFixedKeyHashTableContext<UInt128>>( + get_key_sizes(data_types)); + break; + case HashKeyType::fixed136: + method_variant.emplace<GroupJoinFixedKeyHashTableContext<UInt136>>( + get_key_sizes(data_types)); + break; + case HashKeyType::fixed256: + method_variant.emplace<GroupJoinFixedKeyHashTableContext<UInt256>>( + get_key_sizes(data_types)); + break; + default: + throw Exception(ErrorCode::INTERNAL_ERROR, + "GroupJoinDataVariants meet invalid key type, type={}", type); + } + } +}; + +using GroupJoinDataVariantsUPtr = std::unique_ptr<GroupJoinDataVariants>; + +} // namespace doris diff --git a/be/src/exec/operator/groupjoin_build_sink.cpp b/be/src/exec/operator/groupjoin_build_sink.cpp new file mode 100644 index 00000000000..1ab9c9b11a3 --- /dev/null +++ b/be/src/exec/operator/groupjoin_build_sink.cpp @@ -0,0 +1,265 @@ +// 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 "exec/operator/groupjoin_build_sink.h" + +#include <variant> + +#include "common/cast_set.h" +#include "core/data_type/data_type_nullable.h" +#include "exec/common/hash_table/hash_map_util.h" +#include "exec/common/util.hpp" +#include "exec/operator/groupjoin_operator_utils.h" +#include "exprs/vectorized_agg_fn.h" +#include "exprs/vexpr.h" +#include "runtime/descriptors.h" +#include "runtime/runtime_state.h" + +namespace doris { + +GroupJoinBuildSinkLocalState::GroupJoinBuildSinkLocalState(DataSinkOperatorXBase* parent, + RuntimeState* state) + : Base(parent, state) { + _finish_dependency = std::make_shared<CountedFinishDependency>( + parent->operator_id(), parent->node_id(), parent->get_name() + "_FINISH_DEPENDENCY"); +} + +Status GroupJoinBuildSinkLocalState::init(RuntimeState* state, LocalSinkStateInfo& info) { + RETURN_IF_ERROR(Base::init(state, info)); + auto& p = _parent->cast<GroupJoinBuildSinkOperatorX>(); + _build_expr_ctxs.resize(p._build_expr_ctxs.size()); + for (size_t i = 0; i < _build_expr_ctxs.size(); ++i) { + RETURN_IF_ERROR(p._build_expr_ctxs[i]->clone(state, _build_expr_ctxs[i])); + } + _aggregate_evaluators.reserve(p._aggregate_evaluators.size()); + for (auto* evaluator : p._aggregate_evaluators) { + _aggregate_evaluators.push_back(evaluator->clone(state, p._pool)); + } + RETURN_IF_ERROR(groupjoin::register_agg_state_layout( + _shared_state, p._aggregate_sides, p._sizes_of_aggregate_states, + p._aligns_of_aggregate_states, p._aggregate_indices, p._aggregate_evaluators)); + DCHECK(std::holds_alternative<std::monostate>(_shared_state->data_variants->method_variant)); + std::vector<DataTypePtr> data_types; + data_types.reserve(_build_expr_ctxs.size()); + for (const auto& ctx : _build_expr_ctxs) { + data_types.emplace_back(remove_nullable(ctx->root()->data_type())); + } + RETURN_IF_ERROR(init_hash_method<GroupJoinDataVariants>(_shared_state->data_variants.get(), + data_types, true)); + if (!p._runtime_filter_descs.empty()) { + _runtime_filter_producer_helper = std::make_shared<RuntimeFilterProducerHelperGroupJoin>(); + RETURN_IF_ERROR(_runtime_filter_producer_helper->init( + state, _build_expr_ctxs, p._runtime_filter_descs, p._child->row_desc())); + } + _dependency->set_ready(); + return Status::OK(); +} + +Status GroupJoinBuildSinkLocalState::terminate(RuntimeState* state) { + if (_terminated) { + return Status::OK(); + } + if (_runtime_filter_producer_helper) { + RETURN_IF_ERROR(_runtime_filter_producer_helper->skip_process(state)); + } + return Base::terminate(state); +} + +Status GroupJoinBuildSinkLocalState::close(RuntimeState* state, Status exec_status) { + if (_closed) { + return Status::OK(); + } + try { + if (!_terminated && _runtime_filter_producer_helper && !state->is_cancelled()) { + if (_runtime_filter_size_sent && exec_status.ok()) { + RETURN_IF_ERROR(_runtime_filter_producer_helper->build_and_publish(state)); + } else { + RETURN_IF_ERROR(_runtime_filter_producer_helper->skip_process(state)); + } + } + } catch (Exception& e) { + return Status::InternalError("GroupJoin runtime filter process meet error: {}", + e.to_string()); + } + if (_runtime_filter_producer_helper) { + _runtime_filter_producer_helper->collect_realtime_profile(custom_profile()); + } + return Base::close(state, exec_status); +} + +Status GroupJoinBuildSinkLocalState::_append_runtime_filter_columns(Block* block) { + auto& p = _parent->cast<GroupJoinBuildSinkOperatorX>(); + if (p._runtime_filter_descs.empty()) { + return Status::OK(); + } + RETURN_IF_ERROR(_runtime_filter_producer_helper->append_block(block)); + return Status::OK(); +} + +GroupJoinBuildSinkOperatorX::GroupJoinBuildSinkOperatorX(ObjectPool* pool, int operator_id, + int dest_id, const TPlanNode& tnode, + const DescriptorTbl& descs) + : Base(operator_id, tnode, dest_id), + _join_distribution(tnode.group_join_node.__isset.dist_type + ? tnode.group_join_node.dist_type + : TJoinDistributionType::NONE), + _pool(pool), + _partition_exprs(tnode.__isset.distribute_expr_lists ? tnode.distribute_expr_lists[0] + : std::vector<TExpr> {}), + _runtime_filter_descs(tnode.runtime_filters) {} + +Status GroupJoinBuildSinkOperatorX::init(const TPlanNode& tnode, RuntimeState* state) { + RETURN_IF_ERROR(Base::init(tnode, state)); + RETURN_IF_ERROR(groupjoin::validate_group_join_node(tnode)); + for (const auto& eq_join_conjunct : tnode.group_join_node.eq_join_conjuncts) { + VExprContextSPtr build_ctx; + RETURN_IF_ERROR(VExpr::create_expr_tree(eq_join_conjunct.right, build_ctx)); + { + VExprContextSPtr probe_ctx; + RETURN_IF_ERROR(VExpr::create_expr_tree(eq_join_conjunct.left, probe_ctx)); + auto build_side_expr_type = build_ctx->root()->data_type(); + auto probe_side_expr_type = probe_ctx->root()->data_type(); + if (!make_nullable(build_side_expr_type) + ->equals(*make_nullable(probe_side_expr_type))) { + return Status::InternalError( + "GroupJoin build side type {}, not match probe side type {}, node={}", + build_side_expr_type->get_name(), probe_side_expr_type->get_name(), + debug_string(0)); + } + } + _build_expr_ctxs.push_back(build_ctx); + } + const auto& group_join_node = tnode.group_join_node; + _aggregate_sides.resize(group_join_node.aggregate_functions.size()); + _sizes_of_aggregate_states.assign(group_join_node.aggregate_functions.size(), 0); + _aligns_of_aggregate_states.assign(group_join_node.aggregate_functions.size(), 1); + _output_tuple_id = group_join_node.output_tuple_id; + + TSortInfo dummy; + for (int i = 0; i < group_join_node.aggregate_functions.size(); ++i) { + const auto& aggregate_function = group_join_node.aggregate_functions[i]; + _aggregate_sides[i] = aggregate_function.input_side; + if (aggregate_function.input_side != TGroupJoinAggSide::BUILD) { + continue; + } + AggFnEvaluator* evaluator = nullptr; + RETURN_IF_ERROR(AggFnEvaluator::create(_pool, aggregate_function.aggregate_function, dummy, + false, false, &evaluator)); + _aggregate_evaluators.push_back(evaluator); + _aggregate_indices.push_back(i); + } + return Status::OK(); +} + +Status GroupJoinBuildSinkOperatorX::prepare(RuntimeState* state) { + RETURN_IF_ERROR(Base::prepare(state)); + RETURN_IF_ERROR(VExpr::prepare(_build_expr_ctxs, state, _child->row_desc())); + RETURN_IF_ERROR(VExpr::open(_build_expr_ctxs, state)); + _output_tuple_desc = state->desc_tbl().get_tuple_descriptor(_output_tuple_id); + DCHECK(_output_tuple_desc != nullptr); + const auto key_size = _build_expr_ctxs.size(); + for (size_t i = 0; i < _aggregate_evaluators.size(); ++i) { + const auto agg_idx = _aggregate_indices[i]; + SlotDescriptor* output_slot_desc = _output_tuple_desc->slots()[key_size + agg_idx]; + RETURN_IF_ERROR(_aggregate_evaluators[i]->prepare(state, _child->row_desc(), + output_slot_desc, output_slot_desc)); + _aggregate_evaluators[i]->set_version(state->be_exec_version()); + _sizes_of_aggregate_states[agg_idx] = _aggregate_evaluators[i]->function()->size_of_data(); + _aligns_of_aggregate_states[agg_idx] = + _aggregate_evaluators[i]->function()->align_of_data(); + } + for (auto* evaluator : _aggregate_evaluators) { + RETURN_IF_ERROR(evaluator->open(state)); + } + return Status::OK(); +} + +Status GroupJoinBuildSinkOperatorX::sink_impl(RuntimeState* state, Block* in_block, bool eos) { + auto& local_state = get_local_state(state); + SCOPED_TIMER(local_state.exec_time_counter()); + COUNTER_UPDATE(local_state.rows_input_counter(), static_cast<int64_t>(in_block->rows())); + if (in_block->rows() > 0) { + const auto rows = cast_set<uint32_t>(in_block->rows()); + RETURN_IF_ERROR(groupjoin::do_evaluate(*in_block, local_state._build_expr_ctxs, + local_state._key_columns_holder)); + RETURN_IF_ERROR(groupjoin::extract_key_columns( + in_block->rows(), local_state._key_columns_holder, + local_state._build_key_not_nullable_columns, local_state._null_map_column)); + const uint8_t* null_map = local_state._null_map_column + ? local_state._null_map_column->get_data().data() + : nullptr; + local_state._places.resize(rows); + RETURN_IF_ERROR(groupjoin::add_build_counts_by_key( + local_state._shared_state, *local_state._shared_state->arena, + local_state._build_key_not_nullable_columns, rows, null_map, _aggregate_indices, + local_state._places.data())); + for (size_t i = 0; i < local_state._aggregate_evaluators.size(); ++i) { + const auto offset = + local_state._shared_state->offsets_of_aggregate_states[_aggregate_indices[i]]; + // If there is no nullable join key, every build row has a valid place because build + // always creates/fetches a hash-table entry. If null_map exists, rows with NULL join + // keys are skipped for normal inner equal join and keep places[row] as nullptr. + if (null_map == nullptr) { + RETURN_IF_ERROR(local_state._aggregate_evaluators[i]->execute_batch_add( + in_block, offset, local_state._places.data(), + *local_state._shared_state->arena)); + } else { + RETURN_IF_ERROR(local_state._aggregate_evaluators[i]->execute_batch_add_selected( + in_block, offset, local_state._places.data(), + *local_state._shared_state->arena)); + } + } + RETURN_IF_ERROR(local_state._append_runtime_filter_columns(in_block)); + } + if (eos) { + if (local_state._runtime_filter_producer_helper) { + RETURN_IF_ERROR(local_state._runtime_filter_producer_helper->send_filter_size( + state, local_state._runtime_filter_producer_helper->build_rows(), + local_state._finish_dependency)); + local_state._runtime_filter_size_sent = true; + } + local_state._dependency->set_ready_to_read(); + } + return Status::OK(); +} + +DataDistribution GroupJoinBuildSinkOperatorX::required_data_distribution( + RuntimeState* state) const { + // Keep the same distribution rule as hash join's non-broadcast path. GroupJoin currently + // supports only inner equi join, so broadcast/null-aware special branches are not needed. + return _join_distribution == TJoinDistributionType::BUCKET_SHUFFLE || + _join_distribution == TJoinDistributionType::COLOCATE + ? DataDistribution(ExchangeType::BUCKET_HASH_SHUFFLE, _partition_exprs) + : DataDistribution(ExchangeType::HASH_SHUFFLE, _partition_exprs); +} + +bool GroupJoinBuildSinkOperatorX::is_shuffled_operator() const { + return _join_distribution == TJoinDistributionType::PARTITIONED || + _join_distribution == TJoinDistributionType::BUCKET_SHUFFLE || + _join_distribution == TJoinDistributionType::COLOCATE; +} + +bool GroupJoinBuildSinkOperatorX::is_colocated_operator() const { + return _join_distribution == TJoinDistributionType::BUCKET_SHUFFLE || + _join_distribution == TJoinDistributionType::COLOCATE; +} + +bool GroupJoinBuildSinkOperatorX::followed_by_shuffled_operator() const { + return (is_shuffled_operator() && !is_colocated_operator()) || _followed_by_shuffled_operator; +} + +} // namespace doris diff --git a/be/src/exec/operator/groupjoin_build_sink.h b/be/src/exec/operator/groupjoin_build_sink.h new file mode 100644 index 00000000000..9eb2a4579ea --- /dev/null +++ b/be/src/exec/operator/groupjoin_build_sink.h @@ -0,0 +1,104 @@ +// 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 <gen_cpp/PlanNodes_types.h> + +#include <memory> +#include <vector> + +#include "core/block/block.h" +#include "core/column/column_vector.h" +#include "exec/common/groupjoin_utils.h" +#include "exec/operator/operator.h" +#include "exec/pipeline/dependency.h" +#include "exec/runtime_filter/runtime_filter_producer_helper_groupjoin.h" + +namespace doris { + +class AggFnEvaluator; +class GroupJoinBuildSinkOperatorX; + +class GroupJoinBuildSinkLocalState final : public PipelineXSinkLocalState<GroupJoinSharedState> { +public: + ENABLE_FACTORY_CREATOR(GroupJoinBuildSinkLocalState); + using Base = PipelineXSinkLocalState<GroupJoinSharedState>; + using Parent = GroupJoinBuildSinkOperatorX; + + GroupJoinBuildSinkLocalState(DataSinkOperatorXBase* parent, RuntimeState* state); + ~GroupJoinBuildSinkLocalState() override = default; + + Status init(RuntimeState* state, LocalSinkStateInfo& info) override; + Status terminate(RuntimeState* state) override; + Status close(RuntimeState* state, Status exec_status) override; + + Dependency* finishdependency() override { return _finish_dependency.get(); } + +private: + friend class GroupJoinBuildSinkOperatorX; + + Status _append_runtime_filter_columns(Block* block); + + VExprContextSPtrs _build_expr_ctxs; + std::vector<AggFnEvaluator*> _aggregate_evaluators; + ColumnRawPtrs _build_key_not_nullable_columns; + std::vector<ColumnPtr> _key_columns_holder; + std::vector<AggregateDataPtr> _places; + ColumnUInt8::MutablePtr _null_map_column; + std::shared_ptr<RuntimeFilterProducerHelperGroupJoin> _runtime_filter_producer_helper; + std::shared_ptr<CountedFinishDependency> _finish_dependency; + bool _runtime_filter_size_sent = false; +}; + +class GroupJoinBuildSinkOperatorX final : public DataSinkOperatorX<GroupJoinBuildSinkLocalState> { +public: + using Base = DataSinkOperatorX<GroupJoinBuildSinkLocalState>; + + GroupJoinBuildSinkOperatorX(ObjectPool* pool, int operator_id, int dest_id, + const TPlanNode& tnode, const DescriptorTbl& descs); + + Status init(const TDataSink& tsink) override { + return Status::InternalError("{} should not init with TDataSink", _name); + } + Status init(const TPlanNode& tnode, RuntimeState* state) override; + Status prepare(RuntimeState* state) override; + Status sink_impl(RuntimeState* state, Block* in_block, bool eos) override; + + DataDistribution required_data_distribution(RuntimeState* state) const override; + bool is_shuffled_operator() const override; + bool is_colocated_operator() const override; + bool followed_by_shuffled_operator() const override; + +private: + friend class GroupJoinBuildSinkLocalState; + + const TJoinDistributionType::type _join_distribution; + ObjectPool* _pool = nullptr; + std::vector<TExpr> _partition_exprs; + VExprContextSPtrs _build_expr_ctxs; + std::vector<AggFnEvaluator*> _aggregate_evaluators; + std::vector<int> _aggregate_indices; + std::vector<TGroupJoinAggSide::type> _aggregate_sides; + const std::vector<TRuntimeFilterDesc> _runtime_filter_descs; + Sizes _sizes_of_aggregate_states; + Sizes _aligns_of_aggregate_states; + TupleId _output_tuple_id; + TupleDescriptor* _output_tuple_desc = nullptr; +}; + +} // namespace doris diff --git a/be/src/exec/operator/groupjoin_operator_utils.cpp b/be/src/exec/operator/groupjoin_operator_utils.cpp new file mode 100644 index 00000000000..53253ac74d3 --- /dev/null +++ b/be/src/exec/operator/groupjoin_operator_utils.cpp @@ -0,0 +1,415 @@ +// 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 "exec/operator/groupjoin_operator_utils.h" + +#include <gen_cpp/Opcodes_types.h> + +#include <algorithm> +#include <new> +#include <variant> + +#include "common/cast_set.h" +#include "common/exception.h" +#include "core/arena.h" +#include "core/column/column_nullable.h" +#include "exec/common/hash_table/hash_map_context.h" +#include "exec/common/hash_table/hash_map_util.h" +#include "exec/common/template_helpers.hpp" +#include "exec/common/util.hpp" +#include "exec/operator/groupjoin_probe_operator.h" +#include "exec/pipeline/dependency.h" +#include "exprs/vectorized_agg_fn.h" +#include "exprs/vexpr.h" + +namespace doris::groupjoin { + +Status validate_group_join_node(const TPlanNode& tnode) { + if (!tnode.__isset.group_join_node) { + return Status::InternalError("GroupJoin node is not set"); + } + + const auto& group_join_node = tnode.group_join_node; + if (group_join_node.join_op != TJoinOp::INNER_JOIN) { + return Status::InternalError("GroupJoin only supports inner join now"); + } + if (group_join_node.__isset.dist_type && + group_join_node.dist_type == TJoinDistributionType::BROADCAST) { + return Status::InternalError("GroupJoin does not support broadcast join now"); + } + if (group_join_node.agg_output_mode != TGroupJoinAggOutputMode::FINAL_RESULT) { + return Status::InternalError( + "GroupJoin only supports final-result aggregate output mode now: {}", + group_join_node.agg_output_mode); + } + if (group_join_node.aggregate_functions.empty()) { + return Status::InternalError("GroupJoin requires at least one aggregate function"); + } + for (const auto& eq_join_conjunct : group_join_node.eq_join_conjuncts) { + if (eq_join_conjunct.__isset.opcode && + eq_join_conjunct.opcode == TExprOpcode::EQ_FOR_NULL) { + return Status::InternalError("GroupJoin does not support null-safe equal join now"); + } + } + return Status::OK(); +} + +Status do_evaluate(const Block& block, VExprContextSPtrs& exprs, + std::vector<ColumnPtr>& key_columns_holder) { + key_columns_holder.resize(exprs.size()); + for (size_t i = 0; i < exprs.size(); ++i) { + RETURN_IF_ERROR(exprs[i]->execute(&block, key_columns_holder[i])); + key_columns_holder[i] = key_columns_holder[i]->convert_to_full_column_if_const(); + } + return Status::OK(); +} + +Status extract_key_columns(size_t rows, const std::vector<ColumnPtr>& key_columns_holder, + ColumnRawPtrs& key_not_nullable_columns, + ColumnUInt8::MutablePtr& null_map_column) { + key_not_nullable_columns.resize(key_columns_holder.size()); + null_map_column.reset(); + + for (size_t i = 0; i < key_columns_holder.size(); ++i) { + const auto* column = key_columns_holder[i].get(); + if (const auto* nullable = check_and_get_column<ColumnNullable>(*column); nullable) { + const auto& col_nested = nullable->get_nested_column(); + const auto& col_nullmap = nullable->get_null_map_data(); + if (!null_map_column) { + null_map_column = ColumnUInt8::create(); + null_map_column->get_data().assign(rows, uint8_t {0}); + } + VectorizedUtils::update_null_map(null_map_column->get_data(), col_nullmap); + key_not_nullable_columns[i] = &col_nested; + } else { + key_not_nullable_columns[i] = column; + } + } + return Status::OK(); +} + +Status register_agg_state_layout(GroupJoinSharedState* shared_state, + const std::vector<TGroupJoinAggSide::type>& aggregate_sides, + const Sizes& sizes_of_aggregate_states, + const Sizes& aligns_of_aggregate_states, + const std::vector<int>& aggregate_indices, + const std::vector<AggFnEvaluator*>& aggregate_evaluators) { + if (shared_state->aggregate_sides.empty()) { + shared_state->aggregate_sides = aggregate_sides; + shared_state->sizes_of_aggregate_states.assign(aggregate_sides.size(), 0); + shared_state->aligns_of_aggregate_states.assign(aggregate_sides.size(), 1); + shared_state->offsets_of_aggregate_states.assign(aggregate_sides.size(), 0); + shared_state->aggregate_evaluators.assign(aggregate_sides.size(), nullptr); + } else if (shared_state->aggregate_sides != aggregate_sides) { + return Status::InternalError("GroupJoin aggregate sides are inconsistent"); + } + + if (sizes_of_aggregate_states.size() != shared_state->aggregate_sides.size() || + aligns_of_aggregate_states.size() != shared_state->aggregate_sides.size()) { + return Status::InternalError("GroupJoin aggregate state layout size is inconsistent"); + } + if (aggregate_indices.size() != aggregate_evaluators.size()) { + return Status::InternalError("GroupJoin aggregate evaluator index size is inconsistent"); + } + + for (size_t i = 0; i < sizes_of_aggregate_states.size(); ++i) { + if (sizes_of_aggregate_states[i] == 0) { + continue; + } + if (shared_state->sizes_of_aggregate_states[i] != 0 && + shared_state->sizes_of_aggregate_states[i] != sizes_of_aggregate_states[i]) { + return Status::InternalError("GroupJoin aggregate state size is inconsistent"); + } + if (shared_state->aligns_of_aggregate_states[i] != 1 && + shared_state->aligns_of_aggregate_states[i] != aligns_of_aggregate_states[i]) { + return Status::InternalError("GroupJoin aggregate state align is inconsistent"); + } + shared_state->sizes_of_aggregate_states[i] = sizes_of_aggregate_states[i]; + shared_state->aligns_of_aggregate_states[i] = aligns_of_aggregate_states[i]; + } + for (size_t i = 0; i < aggregate_indices.size(); ++i) { + const auto agg_idx = aggregate_indices[i]; + DCHECK_GE(agg_idx, 0); + DCHECK_LT(agg_idx, shared_state->aggregate_evaluators.size()); + if (shared_state->aggregate_evaluators[agg_idx] != nullptr && + shared_state->aggregate_evaluators[agg_idx] != aggregate_evaluators[i]) { + return Status::InternalError("GroupJoin aggregate evaluator is inconsistent"); + } + shared_state->aggregate_evaluators[agg_idx] = aggregate_evaluators[i]; + } + + bool ready = true; + for (size_t size : shared_state->sizes_of_aggregate_states) { + ready &= size != 0; + } + for (auto* evaluator : shared_state->aggregate_evaluators) { + ready &= evaluator != nullptr; + } + if (!ready || shared_state->agg_layout_ready) { + return Status::OK(); + } + + shared_state->total_size_of_aggregate_states = 0; + shared_state->align_aggregate_states = 1; + for (size_t i = 0; i < shared_state->sizes_of_aggregate_states.size(); ++i) { + shared_state->offsets_of_aggregate_states[i] = shared_state->total_size_of_aggregate_states; + const auto align = shared_state->aligns_of_aggregate_states[i]; + if ((align & (align - 1)) != 0) { + return Status::RuntimeError("Logical error: GroupJoin align_of_data is not 2^N"); + } + shared_state->align_aggregate_states = + std::max(shared_state->align_aggregate_states, align); + shared_state->total_size_of_aggregate_states += shared_state->sizes_of_aggregate_states[i]; + if (i + 1 < shared_state->sizes_of_aggregate_states.size()) { + const auto next_align = shared_state->aligns_of_aggregate_states[i + 1]; + shared_state->total_size_of_aggregate_states = + (shared_state->total_size_of_aggregate_states + next_align - 1) / next_align * + next_align; + } + } + shared_state->agg_layout_ready = true; + return Status::OK(); +} + +void create_all_agg_states(GroupJoinSharedState* shared_state, AggregateDataPtr data) { + for (size_t i = 0; i < shared_state->aggregate_evaluators.size(); ++i) { + try { + shared_state->aggregate_evaluators[i]->create( + data + shared_state->offsets_of_aggregate_states[i]); + } catch (...) { + for (size_t j = 0; j < i; ++j) { + shared_state->aggregate_evaluators[j]->destroy( + data + shared_state->offsets_of_aggregate_states[j]); + } + throw; + } + } +} + +void destroy_all_agg_states(GroupJoinSharedState* shared_state, AggregateDataPtr data) { + if (data == nullptr) { + return; + } + for (size_t i = 0; i < shared_state->aggregate_evaluators.size(); ++i) { + shared_state->aggregate_evaluators[i]->destroy( + data + shared_state->offsets_of_aggregate_states[i]); + } +} + +void ensure_entry_agg_states(GroupJoinEntry* entry, GroupJoinSharedState* shared_state, + Arena& arena) { + DCHECK(!shared_state->aggregate_evaluators.empty()) + << "ensure_entry_agg_states is called only when this side has aggregate evaluators"; + DCHECK(shared_state->agg_layout_ready) + << "aggregate state layout must be registered before creating entry states"; + if (entry->agg_states != nullptr) { + return; + } + auto* agg_states = arena.aligned_alloc(shared_state->total_size_of_aggregate_states, + shared_state->align_aggregate_states); + create_all_agg_states(shared_state, agg_states); + entry->agg_states = agg_states; +} + +Status add_build_counts_by_key(GroupJoinSharedState* shared_state, Arena& arena, + ColumnRawPtrs& key_not_nullable_columns, uint32_t num_rows, + const uint8_t* null_map, const std::vector<int>& aggregate_indices, + AggregateDataPtr* places) { + std::fill(places, places + num_rows, nullptr); + return std::visit( + Overload {[&](std::monostate&) -> Status { + throw doris::Exception(ErrorCode::INTERNAL_ERROR, "uninited hash table"); + }, + [&](auto& hash_method) -> Status { + using HashMethodType = std::decay_t<decltype(hash_method)>; + using State = typename HashMethodType::State; + State state(key_not_nullable_columns); + hash_method.init_serialized_keys(key_not_nullable_columns, num_rows, + null_map); + + auto creator = [&](const auto& ctor, auto& key, auto& origin) { + HashMethodType::try_presis_key_and_origin(key, origin, arena); + auto* mapped = new (arena.alloc<GroupJoinEntry>()) GroupJoinEntry(); + ctor(key, mapped); + }; + auto creator_for_null_key = [](auto&) { + throw doris::Exception( + ErrorCode::INTERNAL_ERROR, + "GroupJoin key columns should not contain nullable columns"); + }; + + auto result_handler = [&](uint32_t row, auto& mapped) { + ++mapped->build_count; + // Only create aggregate states when the build side + // has aggregate functions to update. Otherwise this + // side only maintains row counts. + if (!aggregate_indices.empty()) { + ensure_entry_agg_states(mapped, shared_state, arena); + } + places[row] = mapped->agg_states; + }; + if (null_map == nullptr) { + lazy_emplace_batch(hash_method, state, num_rows, creator, + creator_for_null_key, result_handler); + } else { + for (uint32_t row = 0; row < num_rows; ++row) { + // For normal inner equal join, any row with a NULL join key can + // never match the probe side. Skip it before lazy_emplace so the + // build hash map does not contain NULL-key entries. + if (null_map[row]) { + continue; + } + auto& mapped = *hash_method.lazy_emplace(state, row, creator, + creator_for_null_key); + result_handler(row, mapped); + } + } + return Status::OK(); + }}, + shared_state->data_variants->method_variant); +} + +Status update_probe_counts(GroupJoinSharedState* shared_state, Arena& arena, + ColumnRawPtrs& key_not_nullable_columns, uint32_t num_rows, + const uint8_t* null_map, const std::vector<int>& aggregate_indices, + AggregateDataPtr* places, int64_t& matched_rows, + uint32_t& matched_probe_rows) { + std::fill(places, places + num_rows, nullptr); + matched_rows = 0; + matched_probe_rows = 0; + return std::visit( + Overload {[&](std::monostate&) -> Status { + throw doris::Exception(ErrorCode::INTERNAL_ERROR, "uninited hash table"); + }, + [&](auto& hash_method) -> Status { + using HashMethodType = std::decay_t<decltype(hash_method)>; + using State = typename HashMethodType::State; + State state(key_not_nullable_columns); + hash_method.init_serialized_keys(key_not_nullable_columns, num_rows, + null_map); + find_batch(hash_method, state, num_rows, [&](uint32_t row, auto& result) { + // For normal inner equal join, any row with a NULL join key can never + // match the build side. + if ((null_map != nullptr && null_map[row]) || !result.is_found()) { + return; + } + auto* mapped = result.get_mapped(); + ++mapped->probe_count; + ++matched_probe_rows; + matched_rows += cast_set<int64_t>(mapped->build_count); + // Only create aggregate states when the probe side has aggregate + // functions to update. Otherwise this side only maintains row counts. + if (!aggregate_indices.empty()) { + ensure_entry_agg_states(mapped, shared_state, arena); + } + places[row] = mapped->agg_states; + }); + return Status::OK(); + }}, + shared_state->data_variants->method_variant); +} + +Status drain_groupjoin_result(GroupJoinSharedState* shared_state, size_t batch_size, + GroupJoinProbeLocalState& local_state, MutableColumns& key_columns, + MutableColumns& value_columns, bool& output_eos) { + local_state._output_arena.clear(); + const size_t agg_size = shared_state->aggregate_evaluators.size(); + return std::visit( + Overload {[&](std::monostate&) -> Status { + throw doris::Exception(ErrorCode::INTERNAL_ERROR, "uninited hash table"); + }, + [&](auto& hash_method) -> Status { + if (!shared_state->drain_inited) { + hash_method.init_iterator(); + shared_state->drain_inited = true; + } + using HashMethodType = std::decay_t<decltype(hash_method)>; + using KeyType = typename HashMethodType::Key; + std::vector<KeyType> keys(batch_size); + if (local_state._values.size() < batch_size) { + local_state._values.resize(batch_size); + } + if (local_state._entries.size() < batch_size) { + local_state._entries.resize(batch_size); + } + if (local_state._repeats.size() < batch_size) { + local_state._repeats.resize(batch_size); + } + + uint32_t num_rows = 0; + auto& iter = hash_method.begin; + while (iter != hash_method.end && num_rows < batch_size) { + auto* entry = iter.get_second(); + // Inner join only outputs groups that have valid rows on both sides. + // Build-side NULL rows may create an entry before being skipped, so + // probe_count alone is not sufficient. + if (entry->build_count > 0 && entry->probe_count > 0) { + keys[num_rows] = iter.get_first(); + local_state._entries[num_rows] = entry; + local_state._values[num_rows] = entry->agg_states; + ++num_rows; + } + ++iter; + } + + hash_method.insert_keys_into_columns(keys, key_columns, num_rows); + for (size_t i = 0; i < agg_size; ++i) { + for (uint32_t row = 0; row < num_rows; ++row) { + local_state._repeats[row] = + shared_state->aggregate_sides[i] == + TGroupJoinAggSide::BUILD + ? local_state._entries[row]->probe_count + : local_state._entries[row]->build_count; + } + shared_state->aggregate_evaluators[i]->insert_result_info_repeat_vec( + local_state._values, + shared_state->offsets_of_aggregate_states[i], + local_state._repeats, value_columns[i].get(), num_rows, + local_state._output_arena); + } + + output_eos = iter == hash_method.end; + return Status::OK(); + }}, + shared_state->data_variants->method_variant); +} + +void destroy_entry_agg_states(GroupJoinSharedState* shared_state, GroupJoinEntry* entry) { + if (entry == nullptr || entry->agg_states == nullptr) { + return; + } + destroy_all_agg_states(shared_state, entry->agg_states); + entry->agg_states = nullptr; +} + +void destroy_agg_states(GroupJoinSharedState* shared_state) { + if (shared_state == nullptr || shared_state->data_variants == nullptr) { + return; + } + std::visit(Overload {[&](std::monostate&) -> void {}, + [&](auto& hash_method) -> void { + auto& hash_table = *hash_method.hash_table; + auto iter = hash_table.begin(); + while (iter != hash_table.end()) { + destroy_entry_agg_states(shared_state, iter.get_second()); + ++iter; + } + }}, + shared_state->data_variants->method_variant); +} + +} // namespace doris::groupjoin diff --git a/be/src/exec/operator/groupjoin_operator_utils.h b/be/src/exec/operator/groupjoin_operator_utils.h new file mode 100644 index 00000000000..4463ceb3bad --- /dev/null +++ b/be/src/exec/operator/groupjoin_operator_utils.h @@ -0,0 +1,82 @@ +// 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 <gen_cpp/PlanNodes_types.h> + +#include <vector> + +#include "common/status.h" +#include "core/block/block.h" +#include "core/column/column_vector.h" +#include "exec/common/groupjoin_utils.h" +#include "exprs/vexpr_fwd.h" + +namespace doris { + +class Arena; +class AggFnEvaluator; +class GroupJoinProbeLocalState; +struct GroupJoinSharedState; +namespace groupjoin { + +Status validate_group_join_node(const TPlanNode& tnode); + +Status do_evaluate(const Block& block, VExprContextSPtrs& exprs, + std::vector<ColumnPtr>& key_columns_holder); + +// Hash map batch APIs take ColumnRawPtrs as key input. This helper converts evaluated +// key expression columns to raw column pointers and builds the combined null map. +// GroupJoin currently supports only normal equal join keys. Therefore nullable keys are +// split into nested columns and the combined null map is used to skip rows with NULL keys: +// for a normal inner equal join, any row containing a NULL join key can never match. +// Null-safe equal needs a different path that keeps nullable keys encoded in the hash key. +Status extract_key_columns(size_t rows, const std::vector<ColumnPtr>& key_columns_holder, + ColumnRawPtrs& key_not_nullable_columns, + ColumnUInt8::MutablePtr& null_map_column); + +Status register_agg_state_layout(GroupJoinSharedState* shared_state, + const std::vector<TGroupJoinAggSide::type>& aggregate_sides, + const Sizes& sizes_of_aggregate_states, + const Sizes& aligns_of_aggregate_states, + const std::vector<int>& aggregate_indices, + const std::vector<AggFnEvaluator*>& aggregate_evaluators); + +Status add_build_counts_by_key(GroupJoinSharedState* shared_state, Arena& arena, + ColumnRawPtrs& key_not_nullable_columns, uint32_t num_rows, + const uint8_t* null_map, const std::vector<int>& aggregate_indices, + AggregateDataPtr* places); + +Status update_probe_counts(GroupJoinSharedState* shared_state, Arena& arena, + ColumnRawPtrs& key_not_nullable_columns, uint32_t num_rows, + const uint8_t* null_map, const std::vector<int>& aggregate_indices, + AggregateDataPtr* places, int64_t& matched_rows, + uint32_t& matched_probe_rows); + +void create_all_agg_states(GroupJoinSharedState* shared_state, AggregateDataPtr data); + +void destroy_all_agg_states(GroupJoinSharedState* shared_state, AggregateDataPtr data); + +Status drain_groupjoin_result(GroupJoinSharedState* shared_state, size_t batch_size, + GroupJoinProbeLocalState& local_state, MutableColumns& key_columns, + MutableColumns& value_columns, bool& output_eos); + +void destroy_agg_states(GroupJoinSharedState* shared_state); + +} // namespace groupjoin +} // namespace doris diff --git a/be/src/exec/operator/groupjoin_probe_operator.cpp b/be/src/exec/operator/groupjoin_probe_operator.cpp new file mode 100644 index 00000000000..9a80da4bcb6 --- /dev/null +++ b/be/src/exec/operator/groupjoin_probe_operator.cpp @@ -0,0 +1,251 @@ +// 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 "exec/operator/groupjoin_probe_operator.h" + +#include <algorithm> + +#include "common/cast_set.h" +#include "core/column/column_nullable.h" +#include "core/data_type/data_type_nullable.h" +#include "exec/common/util.hpp" +#include "exec/operator/groupjoin_operator_utils.h" +#include "exprs/vectorized_agg_fn.h" +#include "exprs/vexpr.h" +#include "runtime/descriptors.h" +#include "runtime/runtime_state.h" + +namespace doris { + +Status GroupJoinProbeLocalState::init(RuntimeState* state, LocalStateInfo& info) { + RETURN_IF_ERROR(Base::init(state, info)); + auto& p = _parent->cast<GroupJoinProbeOperatorX>(); + _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])); + } + _aggregate_evaluators.reserve(p._aggregate_evaluators.size()); + for (auto* evaluator : p._aggregate_evaluators) { + _aggregate_evaluators.push_back(evaluator->clone(state, p._pool)); + } + RETURN_IF_ERROR(groupjoin::register_agg_state_layout( + _shared_state, p._aggregate_sides, p._sizes_of_aggregate_states, + p._aligns_of_aggregate_states, p._aggregate_indices, p._aggregate_evaluators)); + return Status::OK(); +} + +GroupJoinProbeOperatorX::GroupJoinProbeOperatorX(ObjectPool* pool, const TPlanNode& tnode, + int operator_id, const DescriptorTbl& descs) + : Base(pool, tnode, operator_id, descs), + _join_distribution(tnode.group_join_node.__isset.dist_type + ? tnode.group_join_node.dist_type + : TJoinDistributionType::NONE), + _partition_exprs(tnode.__isset.distribute_expr_lists ? tnode.distribute_expr_lists[0] + : std::vector<TExpr> {}) { + _op_name = "GROUP_JOIN_PROBE_OPERATOR"; +} + +Status GroupJoinProbeOperatorX::init(const TPlanNode& tnode, RuntimeState* state) { + RETURN_IF_ERROR(Base::init(tnode, state)); + RETURN_IF_ERROR(groupjoin::validate_group_join_node(tnode)); + for (const auto& eq_join_conjunct : tnode.group_join_node.eq_join_conjuncts) { + VExprContextSPtr ctx; + RETURN_IF_ERROR(VExpr::create_expr_tree(eq_join_conjunct.left, ctx)); + _probe_expr_ctxs.push_back(ctx); + } + const auto& group_join_node = tnode.group_join_node; + _aggregate_sides.resize(group_join_node.aggregate_functions.size()); + _sizes_of_aggregate_states.assign(group_join_node.aggregate_functions.size(), 0); + _aligns_of_aggregate_states.assign(group_join_node.aggregate_functions.size(), 1); + _output_tuple_id = group_join_node.output_tuple_id; + + TSortInfo dummy; + for (int i = 0; i < group_join_node.aggregate_functions.size(); ++i) { + const auto& aggregate_function = group_join_node.aggregate_functions[i]; + _aggregate_sides[i] = aggregate_function.input_side; + if (aggregate_function.input_side != TGroupJoinAggSide::PROBE) { + continue; + } + AggFnEvaluator* evaluator = nullptr; + RETURN_IF_ERROR(AggFnEvaluator::create(_pool, aggregate_function.aggregate_function, dummy, + false, false, &evaluator)); + _aggregate_evaluators.push_back(evaluator); + _aggregate_indices.push_back(i); + } + return Status::OK(); +} + +Status GroupJoinProbeOperatorX::prepare(RuntimeState* state) { + RETURN_IF_ERROR(Base::prepare(state)); + RETURN_IF_ERROR(VExpr::prepare(_probe_expr_ctxs, state, _child->row_desc())); + RETURN_IF_ERROR(VExpr::open(_probe_expr_ctxs, state)); + _output_tuple_desc = state->desc_tbl().get_tuple_descriptor(_output_tuple_id); + DCHECK(_output_tuple_desc != nullptr); + const auto key_size = _probe_expr_ctxs.size(); + for (size_t i = 0; i < key_size; ++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 (size_t i = 0; i < _aggregate_evaluators.size(); ++i) { + const auto agg_idx = _aggregate_indices[i]; + SlotDescriptor* output_slot_desc = _output_tuple_desc->slots()[key_size + agg_idx]; + RETURN_IF_ERROR(_aggregate_evaluators[i]->prepare(state, _child->row_desc(), + output_slot_desc, output_slot_desc)); + _aggregate_evaluators[i]->set_version(state->be_exec_version()); + _sizes_of_aggregate_states[agg_idx] = _aggregate_evaluators[i]->function()->size_of_data(); + _aligns_of_aggregate_states[agg_idx] = + _aggregate_evaluators[i]->function()->align_of_data(); + } + for (auto* evaluator : _aggregate_evaluators) { + RETURN_IF_ERROR(evaluator->open(state)); + } + return Status::OK(); +} + +Status GroupJoinProbeOperatorX::push(RuntimeState* state, Block* input_block, bool eos) const { + auto& local_state = get_local_state(state); + if (input_block->rows() > 0) { + const auto rows = cast_set<uint32_t>(input_block->rows()); + RETURN_IF_ERROR(groupjoin::do_evaluate(*input_block, local_state._probe_expr_ctxs, + local_state._key_columns_holder)); + RETURN_IF_ERROR(groupjoin::extract_key_columns( + input_block->rows(), local_state._key_columns_holder, + local_state._probe_key_not_nullable_columns, local_state._null_map_column)); + const uint8_t* null_map = local_state._null_map_column + ? local_state._null_map_column->get_data().data() + : nullptr; + local_state._places.resize(rows); + int64_t matched_rows = 0; + uint32_t matched_probe_rows = 0; + RETURN_IF_ERROR(groupjoin::update_probe_counts( + local_state._shared_state, *local_state._shared_state->arena, + local_state._probe_key_not_nullable_columns, rows, null_map, _aggregate_indices, + local_state._places.data(), matched_rows, matched_probe_rows)); + local_state._shared_state->total_match_count += matched_rows; + for (size_t i = 0; i < local_state._aggregate_evaluators.size(); ++i) { + const auto offset = + local_state._shared_state->offsets_of_aggregate_states[_aggregate_indices[i]]; + // Probe rows can have nullptr places not only because of NULL join keys, but also + // because an inner join probe key may not exist in the build hash table. + if (matched_probe_rows == rows) { + RETURN_IF_ERROR(local_state._aggregate_evaluators[i]->execute_batch_add( + input_block, offset, local_state._places.data(), + *local_state._shared_state->arena)); + } else { + RETURN_IF_ERROR(local_state._aggregate_evaluators[i]->execute_batch_add_selected( + input_block, offset, local_state._places.data(), + *local_state._shared_state->arena)); + } + } + } + if (eos) { + local_state._shared_state->probe_eos = true; + } + return Status::OK(); +} + +Status GroupJoinProbeOperatorX::pull(RuntimeState* state, Block* output_block, bool* eos) const { + auto& local_state = get_local_state(state); + auto* shared_state = local_state._shared_state; + if (shared_state->result_emitted) { + output_block->clear_column_data(); + *eos = true; + return Status::OK(); + } + + auto columns_with_schema = VectorizedUtils::create_columns_with_type_and_name(row_desc()); + const size_t key_size = local_state._probe_expr_ctxs.size(); + const size_t agg_size = shared_state->aggregate_evaluators.size(); + DCHECK_EQ(columns_with_schema.size(), key_size + agg_size); + + MutableColumns key_columns; + key_columns.reserve(key_size); + for (size_t i = 0; i < key_size; ++i) { + const auto output_key_need_nullable = + std::ranges::find(_make_nullable_keys, i) != _make_nullable_keys.end(); + key_columns.emplace_back((output_key_need_nullable + ? remove_nullable(columns_with_schema[i].type) + : columns_with_schema[i].type) + ->create_column()); + } + + MutableColumns value_columns; + value_columns.reserve(agg_size); + for (size_t i = 0; i < agg_size; ++i) { + DCHECK(shared_state->aggregate_evaluators[i] != nullptr); + value_columns.emplace_back(columns_with_schema[key_size + i].type->create_column()); + } + + bool output_eos = false; + RETURN_IF_ERROR(groupjoin::drain_groupjoin_result(shared_state, state->batch_size(), + local_state, key_columns, value_columns, + output_eos)); + + for (size_t i = 0; i < key_size; ++i) { + columns_with_schema[i].column = std::move(key_columns[i]); + } + for (size_t i = 0; i < agg_size; ++i) { + columns_with_schema[key_size + i].column = std::move(value_columns[i]); + } + *output_block = Block(std::move(columns_with_schema)); + if (output_block->rows() != 0) { + for (auto cid : _make_nullable_keys) { + output_block->get_by_position(cid).column = + make_nullable(output_block->get_by_position(cid).column); + output_block->get_by_position(cid).type = + make_nullable(output_block->get_by_position(cid).type); + } + } + shared_state->result_emitted = output_eos; + *eos = output_eos; + return Status::OK(); +} + +bool GroupJoinProbeOperatorX::need_more_input_data(RuntimeState* state) const { + auto& local_state = get_local_state(state); + return !local_state._shared_state->probe_eos; +} + +DataDistribution GroupJoinProbeOperatorX::required_data_distribution(RuntimeState* state) const { + // Keep the same distribution rule as hash join's non-broadcast path. GroupJoin currently + // supports only inner equi join, so broadcast/null-aware special branches are not needed. + return _join_distribution == TJoinDistributionType::BUCKET_SHUFFLE || + _join_distribution == TJoinDistributionType::COLOCATE + ? DataDistribution(ExchangeType::BUCKET_HASH_SHUFFLE, _partition_exprs) + : DataDistribution(ExchangeType::HASH_SHUFFLE, _partition_exprs); +} + +bool GroupJoinProbeOperatorX::is_shuffled_operator() const { + return _join_distribution == TJoinDistributionType::PARTITIONED || + _join_distribution == TJoinDistributionType::BUCKET_SHUFFLE || + _join_distribution == TJoinDistributionType::COLOCATE; +} + +bool GroupJoinProbeOperatorX::is_colocated_operator() const { + return _join_distribution == TJoinDistributionType::BUCKET_SHUFFLE || + _join_distribution == TJoinDistributionType::COLOCATE; +} + +bool GroupJoinProbeOperatorX::followed_by_shuffled_operator() const { + return (is_shuffled_operator() && !is_colocated_operator()) || _followed_by_shuffled_operator; +} + +} // namespace doris diff --git a/be/src/exec/operator/groupjoin_probe_operator.h b/be/src/exec/operator/groupjoin_probe_operator.h new file mode 100644 index 00000000000..78a1c0156f7 --- /dev/null +++ b/be/src/exec/operator/groupjoin_probe_operator.h @@ -0,0 +1,109 @@ +// 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 <gen_cpp/PlanNodes_types.h> + +#include <memory> +#include <vector> + +#include "core/arena.h" +#include "core/column/column_vector.h" +#include "exec/common/groupjoin_utils.h" +#include "exec/operator/operator.h" +#include "exec/pipeline/dependency.h" + +namespace doris { + +class AggFnEvaluator; +class GroupJoinProbeLocalState; +class GroupJoinProbeOperatorX; +namespace groupjoin { +Status drain_groupjoin_result(GroupJoinSharedState* shared_state, size_t batch_size, + GroupJoinProbeLocalState& local_state, MutableColumns& key_columns, + MutableColumns& value_columns, bool& output_eos); +} + +class GroupJoinProbeLocalState final : public PipelineXLocalState<GroupJoinSharedState> { +public: + ENABLE_FACTORY_CREATOR(GroupJoinProbeLocalState); + using Base = PipelineXLocalState<GroupJoinSharedState>; + using Parent = GroupJoinProbeOperatorX; + + GroupJoinProbeLocalState(RuntimeState* state, OperatorXBase* parent) : Base(state, parent) {} + ~GroupJoinProbeLocalState() override = default; + + Status init(RuntimeState* state, LocalStateInfo& info) override; + +private: + friend class GroupJoinProbeOperatorX; + friend Status groupjoin::drain_groupjoin_result(GroupJoinSharedState*, size_t, + GroupJoinProbeLocalState&, MutableColumns&, + MutableColumns&, bool&); + template <typename LocalStateType> + friend class StatefulOperatorX; + + std::unique_ptr<Block> _child_block = Block::create_unique(); + bool _child_eos = false; + VExprContextSPtrs _probe_expr_ctxs; + std::vector<AggFnEvaluator*> _aggregate_evaluators; + ColumnRawPtrs _probe_key_not_nullable_columns; + std::vector<ColumnPtr> _key_columns_holder; + std::vector<AggregateDataPtr> _places; + std::vector<AggregateDataPtr> _values; + std::vector<GroupJoinEntry*> _entries; + std::vector<uint64_t> _repeats; + Arena _output_arena; + ColumnUInt8::MutablePtr _null_map_column; +}; + +class GroupJoinProbeOperatorX final : public StatefulOperatorX<GroupJoinProbeLocalState> { +public: + using Base = StatefulOperatorX<GroupJoinProbeLocalState>; + + GroupJoinProbeOperatorX(ObjectPool* pool, const TPlanNode& tnode, int operator_id, + const DescriptorTbl& descs); + + Status init(const TPlanNode& tnode, RuntimeState* state) override; + Status prepare(RuntimeState* state) override; + Status push(RuntimeState* state, Block* input_block, bool eos) const override; + Status pull(RuntimeState* state, Block* output_block, bool* eos) const override; + bool need_more_input_data(RuntimeState* state) const override; + + DataDistribution required_data_distribution(RuntimeState* state) const override; + bool is_shuffled_operator() const override; + bool is_colocated_operator() const override; + bool followed_by_shuffled_operator() const override; + +private: + friend class GroupJoinProbeLocalState; + + const TJoinDistributionType::type _join_distribution; + std::vector<TExpr> _partition_exprs; + VExprContextSPtrs _probe_expr_ctxs; + std::vector<AggFnEvaluator*> _aggregate_evaluators; + std::vector<int> _aggregate_indices; + std::vector<TGroupJoinAggSide::type> _aggregate_sides; + Sizes _sizes_of_aggregate_states; + Sizes _aligns_of_aggregate_states; + std::vector<size_t> _make_nullable_keys; + TupleId _output_tuple_id; + TupleDescriptor* _output_tuple_desc = nullptr; +}; + +} // namespace doris diff --git a/be/src/exec/operator/operator.cpp b/be/src/exec/operator/operator.cpp index 796cc116969..4ca59c893cf 100644 --- a/be/src/exec/operator/operator.cpp +++ b/be/src/exec/operator/operator.cpp @@ -42,6 +42,8 @@ #include "exec/operator/file_scan_operator.h" #include "exec/operator/group_commit_block_sink_operator.h" #include "exec/operator/group_commit_scan_operator.h" +#include "exec/operator/groupjoin_build_sink.h" +#include "exec/operator/groupjoin_probe_operator.h" #include "exec/operator/hashjoin_build_sink.h" #include "exec/operator/hashjoin_probe_operator.h" #include "exec/operator/hive_table_sink_operator.h" @@ -858,6 +860,7 @@ DECLARE_OPERATOR(LocalExchangeSinkLocalState) DECLARE_OPERATOR(AggSinkLocalState) DECLARE_OPERATOR(BucketedAggSinkLocalState) DECLARE_OPERATOR(PartitionedAggSinkLocalState) +DECLARE_OPERATOR(GroupJoinBuildSinkLocalState) DECLARE_OPERATOR(ExchangeSinkLocalState) DECLARE_OPERATOR(NestedLoopJoinBuildSinkLocalState) DECLARE_OPERATOR(UnionSinkLocalState) @@ -908,6 +911,7 @@ DECLARE_OPERATOR(PartitionedHashJoinProbeLocalState) DECLARE_OPERATOR(CacheSourceLocalState) DECLARE_OPERATOR(RecCTESourceLocalState) DECLARE_OPERATOR(RecCTEScanLocalState) +DECLARE_OPERATOR(GroupJoinProbeLocalState) #ifdef BE_TEST DECLARE_OPERATOR(MockLocalState) @@ -926,6 +930,7 @@ template class StatefulOperatorX<StreamingAggLocalState>; template class StatefulOperatorX<DistinctStreamingAggLocalState>; template class StatefulOperatorX<NestedLoopJoinProbeLocalState>; template class StatefulOperatorX<TableFunctionLocalState>; +template class StatefulOperatorX<GroupJoinProbeLocalState>; template class PipelineXSinkLocalState<HashJoinSharedState>; template class PipelineXSinkLocalState<PartitionedHashJoinSharedState>; @@ -935,6 +940,7 @@ template class PipelineXSinkLocalState<NestedLoopJoinSharedState>; template class PipelineXSinkLocalState<AnalyticSharedState>; template class PipelineXSinkLocalState<AggSharedState>; template class PipelineXSinkLocalState<BucketedAggSharedState>; +template class PipelineXSinkLocalState<GroupJoinSharedState>; template class PipelineXSinkLocalState<PartitionedAggSharedState>; template class PipelineXSinkLocalState<FakeSharedState>; template class PipelineXSinkLocalState<UnionSharedState>; @@ -954,6 +960,7 @@ template class PipelineXLocalState<NestedLoopJoinSharedState>; template class PipelineXLocalState<AnalyticSharedState>; template class PipelineXLocalState<AggSharedState>; template class PipelineXLocalState<BucketedAggSharedState>; +template class PipelineXLocalState<GroupJoinSharedState>; template class PipelineXLocalState<PartitionedAggSharedState>; template class PipelineXLocalState<FakeSharedState>; template class PipelineXLocalState<UnionSharedState>; diff --git a/be/src/exec/pipeline/dependency.cpp b/be/src/exec/pipeline/dependency.cpp index f808e6b10e8..bcbd73ea178 100644 --- a/be/src/exec/pipeline/dependency.cpp +++ b/be/src/exec/pipeline/dependency.cpp @@ -26,6 +26,7 @@ #include "exec/common/join_utils.h" #include "exec/common/set_utils.h" #include "exec/common/template_helpers.hpp" +#include "exec/operator/groupjoin_operator_utils.h" #include "exec/operator/multi_cast_data_streamer.h" #include "exec/pipeline/pipeline_fragment_context.h" #include "exec/pipeline/pipeline_task.h" @@ -496,6 +497,10 @@ void BucketedAggSharedState::_destroy_agg_status(AggregateDataPtr data) { LocalExchangeSharedState::~LocalExchangeSharedState() = default; +GroupJoinSharedState::~GroupJoinSharedState() { + groupjoin::destroy_agg_states(this); +} + Status SetSharedState::update_build_not_ignore_null(const VExprContextSPtrs& ctxs) { if (ctxs.size() > build_not_ignore_null.size()) { return Status::InternalError("build_not_ignore_null not initialized"); diff --git a/be/src/exec/pipeline/dependency.h b/be/src/exec/pipeline/dependency.h index 53f9ed9281b..dbb7c8b9f91 100644 --- a/be/src/exec/pipeline/dependency.h +++ b/be/src/exec/pipeline/dependency.h @@ -22,7 +22,9 @@ #include <sys/_types/_u_int.h> #endif +#include <concurrentqueue.h> #include <gen_cpp/Partitions_types.h> +#include <gen_cpp/PlanNodes_types.h> #include <gen_cpp/internal_service.pb.h> #include <sqltypes.h> @@ -46,7 +48,11 @@ #include "core/arena.h" #include "core/block/block.h" #include "core/types.h" +#include "exec/common/agg_utils.h" +#include "exec/common/groupjoin_utils.h" #include "exec/common/join_op_utils.h" +#include "exec/common/join_utils.h" +#include "exec/common/set_utils.h" #include "exec/operator/data_queue.h" #include "exec/sort/partition_sorter.h" #include "exec/sort/sorter.h" @@ -676,6 +682,27 @@ struct HashJoinSharedState : public JoinSharedState { std::vector<uint32_t> asof_build_row_to_bucket; }; +struct GroupJoinSharedState : public BasicSharedState { + ENABLE_FACTORY_CREATOR(GroupJoinSharedState) + GroupJoinSharedState() { data_variants = std::make_unique<GroupJoinDataVariants>(); } + ~GroupJoinSharedState() override; + + bool probe_eos = false; + bool result_emitted = false; + GroupJoinDataVariantsUPtr data_variants = nullptr; + std::shared_ptr<Arena> arena = std::make_shared<Arena>(); + int64_t total_match_count = 0; + bool drain_inited = false; + std::vector<TGroupJoinAggSide::type> aggregate_sides; + std::vector<AggFnEvaluator*> aggregate_evaluators; + Sizes offsets_of_aggregate_states; + Sizes sizes_of_aggregate_states; + Sizes aligns_of_aggregate_states; + size_t total_size_of_aggregate_states = 0; + size_t align_aggregate_states = 1; + bool agg_layout_ready = false; +}; + struct PartitionedHashJoinSharedState : public HashJoinSharedState, public std::enable_shared_from_this<PartitionedHashJoinSharedState> { diff --git a/be/src/exec/pipeline/pipeline_fragment_context.cpp b/be/src/exec/pipeline/pipeline_fragment_context.cpp index 2fa064c8a68..9e79f185a4e 100644 --- a/be/src/exec/pipeline/pipeline_fragment_context.cpp +++ b/be/src/exec/pipeline/pipeline_fragment_context.cpp @@ -68,6 +68,8 @@ #include "exec/operator/file_scan_operator.h" #include "exec/operator/group_commit_block_sink_operator.h" #include "exec/operator/group_commit_scan_operator.h" +#include "exec/operator/groupjoin_build_sink.h" +#include "exec/operator/groupjoin_probe_operator.h" #include "exec/operator/hashjoin_build_sink.h" #include "exec/operator/hashjoin_probe_operator.h" #include "exec/operator/hive_table_sink_operator.h" @@ -1766,6 +1768,26 @@ Status PipelineFragmentContext::_create_operator(ObjectPool* pool, const TPlanNo } break; } + case TPlanNodeType::GROUP_JOIN_NODE: { + op = std::make_shared<GroupJoinProbeOperatorX>(pool, tnode, next_operator_id(), descs); + RETURN_IF_ERROR(cur_pipe->add_operator(op, _parallel_instances)); + + const auto downstream_pipeline_id = cur_pipe->id(); + if (!_dag.contains(downstream_pipeline_id)) { + _dag.insert({downstream_pipeline_id, {}}); + } + PipelinePtr build_side_pipe = add_pipeline(cur_pipe); + _dag[downstream_pipeline_id].push_back(build_side_pipe->id()); + + sink_ops.push_back(std::make_shared<GroupJoinBuildSinkOperatorX>( + pool, next_sink_operator_id(), op->operator_id(), tnode, descs)); + RETURN_IF_ERROR(build_side_pipe->set_sink(sink_ops.back())); + RETURN_IF_ERROR(build_side_pipe->sink()->init(tnode, _runtime_state.get())); + + _pipeline_parent_map.push(op->node_id(), cur_pipe); + _pipeline_parent_map.push(op->node_id(), build_side_pipe); + break; + } case TPlanNodeType::HASH_JOIN_NODE: { const auto is_broadcast_join = tnode.hash_join_node.__isset.is_broadcast_join && tnode.hash_join_node.is_broadcast_join; diff --git a/be/src/exec/runtime_filter/runtime_filter_producer_helper_groupjoin.h b/be/src/exec/runtime_filter/runtime_filter_producer_helper_groupjoin.h new file mode 100644 index 00000000000..2080d2220a4 --- /dev/null +++ b/be/src/exec/runtime_filter/runtime_filter_producer_helper_groupjoin.h @@ -0,0 +1,94 @@ +// 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 <vector> + +#include "common/status.h" +#include "core/block/block.h" +#include "exec/runtime_filter/runtime_filter_producer_helper.h" +#include "exprs/vexpr.h" +#include "exprs/vexpr_context.h" +#include "runtime/runtime_state.h" + +namespace doris { + +// This helper is used by GroupJoin build sink. GroupJoin does not keep a full build block, so +// this helper caches only runtime-filter source columns and replays them after RF size is ready. +class RuntimeFilterProducerHelperGroupJoin final : public RuntimeFilterProducerHelper { +public: + ~RuntimeFilterProducerHelperGroupJoin() override = default; + + RuntimeFilterProducerHelperGroupJoin() : RuntimeFilterProducerHelper(true, false) {} + + Status append_block(Block* block) { + if (_skip_runtime_filters_process) { + return Status::OK(); + } + if (block->rows() == 0) { + return Status::OK(); + } + + std::vector<ColumnPtr> filter_columns; + filter_columns.reserve(_filter_expr_contexts.size()); + for (auto& ctx : _filter_expr_contexts) { + ColumnPtr column; + RETURN_IF_ERROR(ctx->execute(block, column)); + column = column->convert_to_full_column_if_const(); + filter_columns.emplace_back(std::move(column)); + } + + _build_rows += block->rows(); + _cached_filter_columns.emplace_back(std::move(filter_columns)); + return Status::OK(); + } + + uint64_t build_rows() const { return _build_rows; } + + Status build_and_publish(RuntimeState* state) { + if (_skip_runtime_filters_process) { + return Status::OK(); + } + + RETURN_IF_ERROR(_init_filters(state, _build_rows)); + for (const auto& filter_columns : _cached_filter_columns) { + RETURN_IF_ERROR(_insert_columns(filter_columns)); + } + + for (const auto& filter : _producers) { + filter->set_wrapper_state_and_ready_to_publish(RuntimeFilterWrapper::State::READY); + } + RETURN_IF_ERROR(_publish(state)); + return Status::OK(); + } + +private: + Status _insert_columns(const std::vector<ColumnPtr>& filter_columns) { + DCHECK_EQ(filter_columns.size(), _producers.size()); + SCOPED_TIMER(_runtime_filter_compute_timer.get()); + for (size_t i = 0; i < _producers.size(); ++i) { + RETURN_IF_ERROR(_producers[i]->insert(filter_columns[i], 0)); + } + return Status::OK(); + } + + std::vector<std::vector<ColumnPtr>> _cached_filter_columns; + uint64_t _build_rows = 0; +}; + +} // namespace doris diff --git a/be/src/exprs/aggregate/aggregate_function.h b/be/src/exprs/aggregate/aggregate_function.h index ac780b7c025..f931148e342 100644 --- a/be/src/exprs/aggregate/aggregate_function.h +++ b/be/src/exprs/aggregate/aggregate_function.h @@ -28,6 +28,7 @@ #include "common/exception.h" #include "common/status.h" +#include "core/arena.h" #include "core/assert_cast.h" #include "core/column/column_complex.h" #include "core/column/column_fixed_length_object.h" @@ -194,13 +195,22 @@ public: /// Inserts results into a column. // todo: Consider whether this passes a ConstAggregateDataPtr // Hot path. Callers must validate `to` with check_result_column_type() before calling - // insert_result_into()/insert_result_into_vec()/insert_result_into_range(). + // insert_result_into()/insert_result_into_vec()/insert_result_into_range()/ + // insert_result_into_repeat()/insert_result_into_repeat_vec(). virtual void insert_result_into(ConstAggregateDataPtr __restrict place, IColumn& to) const = 0; virtual void insert_result_into_vec(const std::vector<AggregateDataPtr>& places, const size_t offset, IColumn& to, const size_t num_rows) const = 0; + virtual void insert_result_into_repeat(ConstAggregateDataPtr place, uint64_t repeat, + IColumn& to, Arena& arena) const = 0; + + virtual void insert_result_into_repeat_vec(const std::vector<AggregateDataPtr>& places, + const size_t offset, + const std::vector<uint64_t>& repeats, IColumn& to, + const size_t num_rows, Arena& arena) const = 0; + /** Contains a loop with calls to "add" function. You can collect arguments into array "places" * and do a single call to "add_batch" for devirtualization and inlining. * This function distributes inputs row to their corresponding aggregation states, @@ -494,6 +504,33 @@ public: } } + void insert_result_into_repeat(ConstAggregateDataPtr place, uint64_t repeat, IColumn& to, + Arena& arena) const override { + const Derived* derived = assert_cast<const Derived*>(this); + auto* output_state = arena.aligned_alloc(derived->size_of_data(), derived->align_of_data()); + derived->create(output_state); + try { + for (uint64_t i = 0; i < repeat; ++i) { + derived->merge(output_state, place, arena); + } + derived->insert_result_into(output_state, to); + } catch (...) { + derived->destroy(output_state); + throw; + } + derived->destroy(output_state); + } + + void insert_result_into_repeat_vec(const std::vector<AggregateDataPtr>& places, + const size_t offset, const std::vector<uint64_t>& repeats, + IColumn& to, const size_t num_rows, + Arena& arena) const override { + const Derived* derived = assert_cast<const Derived*>(this); + for (size_t i = 0; i != num_rows; ++i) { + derived->insert_result_into_repeat(places[i] + offset, repeats[i], to, arena); + } + } + void serialize_vec(const std::vector<AggregateDataPtr>& places, size_t offset, BufferWritable& buf, const size_t num_rows) const override { const Derived* derived = assert_cast<const Derived*>(this); diff --git a/be/src/exprs/aggregate/aggregate_function_count.h b/be/src/exprs/aggregate/aggregate_function_count.h index fd0610158bc..61950739ab8 100644 --- a/be/src/exprs/aggregate/aggregate_function_count.h +++ b/be/src/exprs/aggregate/aggregate_function_count.h @@ -96,6 +96,23 @@ public: data(place).count); } + void insert_result_into_repeat(ConstAggregateDataPtr place, uint64_t repeat, IColumn& to, + Arena&) const override { + auto& column = assert_cast<ColumnInt64&>(to); + column.get_data().push_back(data(place).count * repeat); + } + + void insert_result_into_repeat_vec(const std::vector<AggregateDataPtr>& places, + const size_t offset, const std::vector<uint64_t>& repeats, + IColumn& to, const size_t num_rows, Arena&) const override { + auto& column = assert_cast<ColumnInt64&>(to); + auto& column_data = column.get_data(); + column_data.reserve(column_data.size() + num_rows); + for (size_t i = 0; i != num_rows; ++i) { + column_data.push_back(data(places[i] + offset).count * repeats[i]); + } + } + void serialize_to_column(const std::vector<AggregateDataPtr>& places, size_t offset, MutableColumnPtr& dst, const size_t num_rows) const override { auto& col = assert_cast<ColumnFixedLengthObject&>(*dst); @@ -259,6 +276,40 @@ public: IAggregateFunction::check_result_column_type(to); } + void insert_result_into_repeat(ConstAggregateDataPtr place, uint64_t repeat, IColumn& to, + Arena&) const override { + if (is_column_nullable(to)) { + auto& nullable_column = assert_cast<ColumnNullable&>(to); + nullable_column.get_null_map_data().push_back(0); + assert_cast<ColumnInt64&>(nullable_column.get_nested_column()) + .get_data() + .push_back(data(place).count * repeat); + } else { + assert_cast<ColumnInt64&>(to).get_data().push_back(data(place).count * repeat); + } + } + + void insert_result_into_repeat_vec(const std::vector<AggregateDataPtr>& places, + const size_t offset, const std::vector<uint64_t>& repeats, + IColumn& to, const size_t num_rows, Arena&) const override { + if (is_column_nullable(to)) { + auto& nullable_column = assert_cast<ColumnNullable&>(to); + nullable_column.get_null_map_column().insert_many_vals(0, num_rows); + auto& nested_data = + assert_cast<ColumnInt64&>(nullable_column.get_nested_column()).get_data(); + nested_data.reserve(nested_data.size() + num_rows); + for (size_t i = 0; i != num_rows; ++i) { + nested_data.push_back(data(places[i] + offset).count * repeats[i]); + } + } else { + auto& column_data = assert_cast<ColumnInt64&>(to).get_data(); + column_data.reserve(column_data.size() + num_rows); + for (size_t i = 0; i != num_rows; ++i) { + column_data.push_back(data(places[i] + offset).count * repeats[i]); + } + } + } + void serialize_to_column(const std::vector<AggregateDataPtr>& places, size_t offset, MutableColumnPtr& dst, const size_t num_rows) const override { auto& col = assert_cast<ColumnFixedLengthObject&>(*dst); diff --git a/be/src/exprs/aggregate/aggregate_function_min_max.h b/be/src/exprs/aggregate/aggregate_function_min_max.h index 452ded73d98..04f2f270b1d 100644 --- a/be/src/exprs/aggregate/aggregate_function_min_max.h +++ b/be/src/exprs/aggregate/aggregate_function_min_max.h @@ -843,6 +843,19 @@ public: } } + void insert_result_into_repeat(ConstAggregateDataPtr place, uint64_t, IColumn& to, + Arena&) const override { + this->data(place).insert_result_into(to); + } + + void insert_result_into_repeat_vec(const std::vector<AggregateDataPtr>& places, + const size_t offset, const std::vector<uint64_t>&, + IColumn& to, const size_t num_rows, Arena&) const override { + for (size_t i = 0; i != num_rows; ++i) { + this->data(places[i] + offset).insert_result_into(to); + } + } + void serialize_to_column(const std::vector<AggregateDataPtr>& places, size_t offset, MutableColumnPtr& dst, const size_t num_rows) const override { if constexpr (Data::IsFixedLength) { diff --git a/be/src/exprs/aggregate/aggregate_function_null_v2.h b/be/src/exprs/aggregate/aggregate_function_null_v2.h index cb9d7dd5316..327658fa88d 100644 --- a/be/src/exprs/aggregate/aggregate_function_null_v2.h +++ b/be/src/exprs/aggregate/aggregate_function_null_v2.h @@ -396,6 +396,69 @@ public: nested_function->check_result_column_type(to); } } + + void insert_result_into_repeat(ConstAggregateDataPtr place, uint64_t repeat, IColumn& to, + Arena& arena) const override { + if constexpr (result_is_nullable) { + auto& to_concrete = assert_cast<ColumnNullable&>(to); + auto& nested_column = to_concrete.get_nested_column(); + auto& null_map = to_concrete.get_null_map_data(); + if (get_flag(place)) { + nested_function->insert_result_into_repeat(nested_place(place), repeat, + nested_column, arena); + null_map.push_back(0); + } else { + to_concrete.insert_default(); + } + } else { + nested_function->insert_result_into_repeat(nested_place(place), repeat, to, arena); + } + } + + void insert_result_into_repeat_vec(const std::vector<AggregateDataPtr>& places, + const size_t offset, const std::vector<uint64_t>& repeats, + IColumn& to, const size_t num_rows, + Arena& arena) const override { + if constexpr (result_is_nullable) { + auto& to_concrete = assert_cast<ColumnNullable&>(to); + auto& nested_column = to_concrete.get_nested_column(); + + std::vector<AggregateDataPtr> nested_places; + std::vector<uint64_t> nested_repeats; + nested_places.reserve(num_rows); + nested_repeats.reserve(num_rows); + + // Nullable result columns are stored as a nested value column plus a null map. + // The nested aggregate function only writes value rows, so the wrapper has to + // flush each contiguous non-NULL run before inserting a NULL row. This preserves + // row order and keeps the nested column size aligned with the null map size. + auto flush_nested = [&]() { + if (nested_places.empty()) { + return; + } + nested_function->insert_result_into_repeat_vec(nested_places, offset + prefix_size, + nested_repeats, nested_column, + nested_places.size(), arena); + to_concrete.get_null_map_column().insert_many_vals(0, nested_places.size()); + nested_places.clear(); + nested_repeats.clear(); + }; + + for (size_t i = 0; i != num_rows; ++i) { + if (get_flag(places[i] + offset)) { + nested_places.push_back(places[i]); + nested_repeats.push_back(repeats[i]); + } else { + flush_nested(); + to_concrete.insert_default(); + } + } + flush_nested(); + } else { + nested_function->insert_result_into_repeat_vec(places, offset + prefix_size, repeats, + to, num_rows, arena); + } + } }; template <typename NestFuction, bool result_is_nullable> diff --git a/be/src/exprs/aggregate/aggregate_function_sum.h b/be/src/exprs/aggregate/aggregate_function_sum.h index 8eddb893ed9..643244409af 100644 --- a/be/src/exprs/aggregate/aggregate_function_sum.h +++ b/be/src/exprs/aggregate/aggregate_function_sum.h @@ -80,6 +80,8 @@ class AggregateFunctionSum<T, TResult, Data> final : public IAggregateFunctionDataHelper<Data, AggregateFunctionSum<T, TResult, Data>>, UnaryExpression, NullableAggregateFunction { + using Base = IAggregateFunctionDataHelper<Data, AggregateFunctionSum<T, TResult, Data>>; + public: using ResultDataType = typename PrimitiveTypeTraits<TResult>::DataType; using ColVecType = typename PrimitiveTypeTraits<T>::ColumnType; @@ -131,6 +133,44 @@ public: column.get_data().push_back(this->data(place).get()); } + void insert_result_into_repeat(ConstAggregateDataPtr place, uint64_t repeat, IColumn& to, + Arena& arena) const override { + if constexpr (is_decimalv2(TResult)) { + Base::insert_result_into_repeat(place, repeat, to, arena); + } else { + auto& column = assert_cast<ColVecResult&>(to); + auto result = this->data(place).get(); + if constexpr (is_decimalv3(TResult)) { + result.value *= typename decltype(result)::NativeType(repeat); + } else { + result *= static_cast<decltype(result)>(repeat); + } + column.get_data().push_back(result); + } + } + + void insert_result_into_repeat_vec(const std::vector<AggregateDataPtr>& places, + const size_t offset, const std::vector<uint64_t>& repeats, + IColumn& to, const size_t num_rows, + Arena& arena) const override { + if constexpr (is_decimalv2(TResult)) { + Base::insert_result_into_repeat_vec(places, offset, repeats, to, num_rows, arena); + } else { + auto& column = assert_cast<ColVecResult&>(to); + auto& column_data = column.get_data(); + column_data.reserve(column_data.size() + num_rows); + for (size_t i = 0; i != num_rows; ++i) { + auto result = this->data(places[i] + offset).get(); + if constexpr (is_decimalv3(TResult)) { + result.value *= typename decltype(result)::NativeType(repeats[i]); + } else { + result *= static_cast<decltype(result)>(repeats[i]); + } + column_data.push_back(result); + } + } + } + void serialize_to_column(const std::vector<AggregateDataPtr>& places, size_t offset, MutableColumnPtr& dst, const size_t num_rows) const override { auto& col = assert_cast<ColumnFixedLengthObject&>(*dst); diff --git a/be/src/exprs/vectorized_agg_fn.cpp b/be/src/exprs/vectorized_agg_fn.cpp index d12339d63f6..0f6bb18ca65 100644 --- a/be/src/exprs/vectorized_agg_fn.cpp +++ b/be/src/exprs/vectorized_agg_fn.cpp @@ -353,6 +353,15 @@ void AggFnEvaluator::insert_result_info_range(ConstAggregateDataPtr place, IColu _function->insert_result_into_range(place, *column, start, end); } +void AggFnEvaluator::insert_result_info_repeat_vec(const std::vector<AggregateDataPtr>& places, + size_t offset, + const std::vector<uint64_t>& repeats, + IColumn* column, const size_t num_rows, + Arena& arena) { + _function->check_result_column_type(*column); + _function->insert_result_into_repeat_vec(places, offset, repeats, *column, num_rows, arena); +} + void AggFnEvaluator::reset(AggregateDataPtr place) { _function->reset(place); } diff --git a/be/src/exprs/vectorized_agg_fn.h b/be/src/exprs/vectorized_agg_fn.h index caf8369dee2..a1b6b4b0247 100644 --- a/be/src/exprs/vectorized_agg_fn.h +++ b/be/src/exprs/vectorized_agg_fn.h @@ -104,6 +104,10 @@ public: void insert_result_info_range(ConstAggregateDataPtr place, IColumn* column, size_t start, size_t end); + void insert_result_info_repeat_vec(const std::vector<AggregateDataPtr>& places, size_t offset, + const std::vector<uint64_t>& repeats, IColumn* column, + const size_t num_rows, Arena& arena); + void reset(AggregateDataPtr place); DataTypePtr& data_type() { return _data_type; } diff --git a/be/test/exec/operator/groupjoin_operator_utils_test.cpp b/be/test/exec/operator/groupjoin_operator_utils_test.cpp new file mode 100644 index 00000000000..754af7d43a6 --- /dev/null +++ b/be/test/exec/operator/groupjoin_operator_utils_test.cpp @@ -0,0 +1,154 @@ +// 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 "exec/operator/groupjoin_operator_utils.h" + +#include <gtest/gtest.h> + +#include <memory> +#include <variant> +#include <vector> + +#include "core/assert_cast.h" +#include "core/column/column_nullable.h" +#include "core/data_type/data_type_number.h" +#include "exec/common/hash_table/hash_map_util.h" +#include "exec/common/template_helpers.hpp" +#include "exec/pipeline/dependency.h" +#include "testutil/column_helper.h" + +namespace doris { +namespace { + +#define ASSERT_OK(stmt) \ + do { \ + const auto& status = (stmt); \ + ASSERT_TRUE(status.ok()) << status; \ + } while (false) + +Status init_int32_groupjoin_state(GroupJoinSharedState* shared_state) { + return init_hash_method<GroupJoinDataVariants>(shared_state->data_variants.get(), + {std::make_shared<DataTypeInt32>()}, true); +} + +size_t count_hash_table_entries(GroupJoinSharedState* shared_state) { + return std::visit( + Overload {[&](std::monostate&) -> size_t { return 0; }, + [&](auto& hash_method) -> size_t { + size_t count = 0; + hash_method.hash_table->for_each_mapped([&](auto&) { ++count; }); + return count; + }}, + shared_state->data_variants->method_variant); +} + +TEST(GroupJoinOperatorUtilsTest, ExtractKeyColumnsBuildsCombinedNullMap) { + auto key0 = ColumnHelper::create_nullable_column<DataTypeInt32>({1, 2, 3, 4}, {0, 1, 0, 0}); + auto key1 = ColumnHelper::create_nullable_column<DataTypeInt32>({10, 20, 30, 40}, {0, 0, 1, 0}); + std::vector<ColumnPtr> key_columns_holder {key0, key1}; + + ColumnRawPtrs key_not_nullable_columns; + ColumnUInt8::MutablePtr null_map_column; + ASSERT_OK(groupjoin::extract_key_columns(4, key_columns_holder, key_not_nullable_columns, + null_map_column)); + + ASSERT_TRUE(null_map_column.get() != nullptr); + ASSERT_EQ(key_not_nullable_columns.size(), 2); + EXPECT_EQ(key_not_nullable_columns[0], + &assert_cast<const ColumnNullable&>(*key0).get_nested_column()); + EXPECT_EQ(key_not_nullable_columns[1], + &assert_cast<const ColumnNullable&>(*key1).get_nested_column()); + + const auto& null_map = null_map_column->get_data(); + ASSERT_EQ(null_map.size(), 4); + EXPECT_EQ(null_map[0], 0); + EXPECT_EQ(null_map[1], 1); + EXPECT_EQ(null_map[2], 1); + EXPECT_EQ(null_map[3], 0); +} + +TEST(GroupJoinOperatorUtilsTest, AddBuildCountsWithoutNullMapUsesBatchPath) { + GroupJoinSharedState shared_state; + ASSERT_OK(init_int32_groupjoin_state(&shared_state)); + + auto build_key = ColumnHelper::create_column<DataTypeInt32>({1, 2, 1, 3}); + ColumnRawPtrs build_key_columns {build_key.get()}; + std::vector<AggregateDataPtr> build_places(build_key->size()); + std::vector<int> aggregate_indices; + ASSERT_OK(groupjoin::add_build_counts_by_key(&shared_state, *shared_state.arena, + build_key_columns, + static_cast<uint32_t>(build_key->size()), nullptr, + aggregate_indices, build_places.data())); + + EXPECT_EQ(count_hash_table_entries(&shared_state), 3); + + auto probe_key = ColumnHelper::create_column<DataTypeInt32>({1, 2, 3, 4}); + ColumnRawPtrs probe_key_columns {probe_key.get()}; + std::vector<AggregateDataPtr> probe_places(probe_key->size()); + int64_t matched_rows = 0; + uint32_t matched_probe_rows = 0; + ASSERT_OK(groupjoin::update_probe_counts(&shared_state, *shared_state.arena, probe_key_columns, + static_cast<uint32_t>(probe_key->size()), nullptr, + aggregate_indices, probe_places.data(), matched_rows, + matched_probe_rows)); + + EXPECT_EQ(matched_probe_rows, 3); + EXPECT_EQ(matched_rows, 4); +} + +TEST(GroupJoinOperatorUtilsTest, AddBuildCountsSkipsNullKeyRowsBeforeEmplace) { + GroupJoinSharedState shared_state; + ASSERT_OK(init_int32_groupjoin_state(&shared_state)); + + auto build_key = ColumnHelper::create_nullable_column<DataTypeInt32>({1, 0, 1, 2, 0, 2}, + {0, 1, 0, 0, 1, 0}); + std::vector<ColumnPtr> key_columns_holder {build_key}; + ColumnRawPtrs build_key_not_nullable_columns; + ColumnUInt8::MutablePtr null_map_column; + ASSERT_OK(groupjoin::extract_key_columns(build_key->size(), key_columns_holder, + build_key_not_nullable_columns, null_map_column)); + ASSERT_TRUE(null_map_column.get() != nullptr); + + std::vector<AggregateDataPtr> build_places(build_key->size()); + std::vector<int> aggregate_indices; + ASSERT_OK(groupjoin::add_build_counts_by_key( + &shared_state, *shared_state.arena, build_key_not_nullable_columns, + static_cast<uint32_t>(build_key->size()), null_map_column->get_data().data(), + aggregate_indices, build_places.data())); + + EXPECT_EQ(count_hash_table_entries(&shared_state), 2); + // This side has no aggregate function in this case, so all places stay nullptr. + // The important contract here is that NULL-key rows are skipped before emplace. + EXPECT_EQ(build_places[1], nullptr); + EXPECT_EQ(build_places[4], nullptr); + + auto probe_key = ColumnHelper::create_column<DataTypeInt32>({0, 1, 2}); + ColumnRawPtrs probe_key_columns {probe_key.get()}; + std::vector<AggregateDataPtr> probe_places(probe_key->size()); + int64_t matched_rows = 0; + uint32_t matched_probe_rows = 0; + ASSERT_OK(groupjoin::update_probe_counts(&shared_state, *shared_state.arena, probe_key_columns, + static_cast<uint32_t>(probe_key->size()), nullptr, + aggregate_indices, probe_places.data(), matched_rows, + matched_probe_rows)); + + EXPECT_EQ(matched_probe_rows, 2); + EXPECT_EQ(matched_rows, 4); +} + +} // namespace +} // namespace doris diff --git a/be/test/exprs/aggregate/aggregate_function_repeat_test.cpp b/be/test/exprs/aggregate/aggregate_function_repeat_test.cpp new file mode 100644 index 00000000000..ba561420fc4 --- /dev/null +++ b/be/test/exprs/aggregate/aggregate_function_repeat_test.cpp @@ -0,0 +1,438 @@ +// 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 <gtest/gtest.h> + +#include <memory> +#include <vector> + +#include "common/exception.h" +#include "core/assert_cast.h" +#include "core/column/column_nullable.h" +#include "core/column/column_vector.h" +#include "core/data_type/data_type_nullable.h" +#include "core/data_type/data_type_number.h" +#include "exprs/aggregate/aggregate_function.h" +#include "exprs/aggregate/aggregate_function_simple_factory.h" +#include "testutil/column_helper.h" + +namespace doris { +namespace { + +AggregateFunctionPtr get_agg_function(const std::string& name, const DataTypes& argument_types, + bool result_is_nullable, AggregateFunctionAttr attr = {}) { + auto function = AggregateFunctionSimpleFactory::instance().get(name, argument_types, nullptr, + result_is_nullable, -1, attr); + if (function == nullptr) { + throw Exception(ErrorCode::INTERNAL_ERROR, "failed to create aggregate function {}", name); + } + return function; +} + +AggregateDataPtr create_state(const AggregateFunctionPtr& function, Arena& arena) { + auto* place = arena.aligned_alloc(function->size_of_data(), function->align_of_data()); + function->create(place); + return place; +} + +void destroy_states(const AggregateFunctionPtr& function, + const std::vector<AggregateDataPtr>& places) { + for (auto* place : places) { + function->destroy(place); + } +} + +void destroy_states(const AggregateFunctionPtr& function, + const std::vector<AggregateDataPtr>& places, size_t offset) { + for (auto* place : places) { + function->destroy(place + offset); + } +} + +TEST(AggregateFunctionRepeatTest, CountRepeatOutput) { + Arena arena; + auto function = get_agg_function("count", {}, false); + std::vector<AggregateDataPtr> places {create_state(function, arena), + create_state(function, arena)}; + + function->add(places[0], nullptr, 0, arena); + function->add(places[0], nullptr, 0, arena); + for (int i = 0; i < 5; ++i) { + function->add(places[1], nullptr, 0, arena); + } + + ColumnInt64 single_result; + function->insert_result_into_repeat(places[0], 3, single_result, arena); + ASSERT_EQ(single_result.size(), 1); + EXPECT_EQ(single_result.get_element(0), 6); + + ColumnInt64 result; + std::vector<uint64_t> repeats {3, 4}; + function->insert_result_into_repeat_vec(places, 0, repeats, result, places.size(), arena); + + ASSERT_EQ(result.size(), 2); + EXPECT_EQ(result.get_element(0), 6); + EXPECT_EQ(result.get_element(1), 20); + destroy_states(function, places); +} + +TEST(AggregateFunctionRepeatTest, CountRepeatZeroOutput) { + Arena arena; + auto function = get_agg_function("count", {}, false); + std::vector<AggregateDataPtr> places {create_state(function, arena)}; + + function->add(places[0], nullptr, 0, arena); + function->add(places[0], nullptr, 0, arena); + function->add(places[0], nullptr, 0, arena); + + ColumnInt64 single_result; + function->insert_result_into_repeat(places[0], 0, single_result, arena); + ASSERT_EQ(single_result.size(), 1); + EXPECT_EQ(single_result.get_element(0), 0); + + ColumnInt64 result; + std::vector<uint64_t> repeats {0}; + function->insert_result_into_repeat_vec(places, 0, repeats, result, places.size(), arena); + ASSERT_EQ(result.size(), 1); + EXPECT_EQ(result.get_element(0), 0); + + destroy_states(function, places); +} + +TEST(AggregateFunctionRepeatTest, CountNullableRepeatOutput) { + Arena arena; + auto nullable_type = std::make_shared<DataTypeNullable>(std::make_shared<DataTypeInt32>()); + auto function = get_agg_function("count", {nullable_type}, false); + std::vector<AggregateDataPtr> places {create_state(function, arena), + create_state(function, arena)}; + + auto input = + ColumnHelper::create_nullable_column<DataTypeInt32>({1, 2, 3, 4, 5}, {0, 1, 0, 0, 1}); + const IColumn* columns[] = {input.get()}; + function->add(places[0], columns, 0, arena); + function->add(places[0], columns, 1, arena); + function->add(places[0], columns, 2, arena); + function->add(places[1], columns, 3, arena); + function->add(places[1], columns, 4, arena); + + ColumnInt64 result; + std::vector<uint64_t> repeats {5, 7}; + function->insert_result_into_repeat_vec(places, 0, repeats, result, places.size(), arena); + + ASSERT_EQ(result.size(), 2); + EXPECT_EQ(result.get_element(0), 10); + EXPECT_EQ(result.get_element(1), 7); + destroy_states(function, places); +} + +TEST(AggregateFunctionRepeatTest, RepeatOutputHonorsStateOffset) { + Arena arena; + auto function = get_agg_function("sum", {std::make_shared<DataTypeInt32>()}, true); + const size_t offset = function->align_of_data(); + std::vector<AggregateDataPtr> places { + arena.aligned_alloc(offset + function->size_of_data(), function->align_of_data()), + arena.aligned_alloc(offset + function->size_of_data(), function->align_of_data())}; + for (auto* place : places) { + function->create(place + offset); + } + + auto input = ColumnHelper::create_column<DataTypeInt32>({2, 3, 5, 7}); + const IColumn* columns[] = {input.get()}; + function->add(places[0] + offset, columns, 0, arena); + function->add(places[0] + offset, columns, 1, arena); + function->add(places[1] + offset, columns, 2, arena); + function->add(places[1] + offset, columns, 3, arena); + + ColumnInt64 result; + std::vector<uint64_t> repeats {4, 6}; + function->insert_result_into_repeat_vec(places, offset, repeats, result, places.size(), arena); + + ASSERT_EQ(result.size(), 2); + EXPECT_EQ(result.get_element(0), 20); + EXPECT_EQ(result.get_element(1), 72); + destroy_states(function, places, offset); +} + +TEST(AggregateFunctionRepeatTest, SumRepeatOutput) { + Arena arena; + auto function = get_agg_function("sum", {std::make_shared<DataTypeInt32>()}, true); + std::vector<AggregateDataPtr> places {create_state(function, arena), + create_state(function, arena)}; + + auto input = ColumnHelper::create_column<DataTypeInt32>({1, 2, 4, 6}); + const IColumn* columns[] = {input.get()}; + function->add(places[0], columns, 0, arena); + function->add(places[0], columns, 1, arena); + function->add(places[1], columns, 2, arena); + function->add(places[1], columns, 3, arena); + + ColumnInt64 result; + std::vector<uint64_t> repeats {3, 2}; + function->insert_result_into_repeat_vec(places, 0, repeats, result, places.size(), arena); + + ASSERT_EQ(result.size(), 2); + EXPECT_EQ(result.get_element(0), 9); + EXPECT_EQ(result.get_element(1), 20); + destroy_states(function, places); +} + +TEST(AggregateFunctionRepeatTest, MinMaxRepeatOutput) { + Arena arena; + auto min_function = get_agg_function("min", {std::make_shared<DataTypeInt32>()}, true); + auto max_function = get_agg_function("max", {std::make_shared<DataTypeInt32>()}, true); + + auto input = ColumnHelper::create_column<DataTypeInt32>({4, 2, 1, 7}); + const IColumn* columns[] = {input.get()}; + std::vector<uint64_t> repeats {10, 20}; + + std::vector<AggregateDataPtr> min_places {create_state(min_function, arena), + create_state(min_function, arena)}; + min_function->add(min_places[0], columns, 0, arena); + min_function->add(min_places[0], columns, 1, arena); + min_function->add(min_places[1], columns, 2, arena); + min_function->add(min_places[1], columns, 3, arena); + ColumnInt32 min_result; + min_function->insert_result_into_repeat_vec(min_places, 0, repeats, min_result, + min_places.size(), arena); + + ASSERT_EQ(min_result.size(), 2); + EXPECT_EQ(min_result.get_element(0), 2); + EXPECT_EQ(min_result.get_element(1), 1); + destroy_states(min_function, min_places); + + std::vector<AggregateDataPtr> max_places {create_state(max_function, arena), + create_state(max_function, arena)}; + max_function->add(max_places[0], columns, 0, arena); + max_function->add(max_places[0], columns, 1, arena); + max_function->add(max_places[1], columns, 2, arena); + max_function->add(max_places[1], columns, 3, arena); + ColumnInt32 max_result; + max_function->insert_result_into_repeat_vec(max_places, 0, repeats, max_result, + max_places.size(), arena); + + ASSERT_EQ(max_result.size(), 2); + EXPECT_EQ(max_result.get_element(0), 4); + EXPECT_EQ(max_result.get_element(1), 7); + destroy_states(max_function, max_places); +} + +TEST(AggregateFunctionRepeatTest, NullableV2SumRepeatOutput) { + Arena arena; + AggregateFunctionAttr attr; + attr.enable_aggregate_function_null_v2 = true; + auto nullable_type = std::make_shared<DataTypeNullable>(std::make_shared<DataTypeInt32>()); + auto function = get_agg_function("sum", {nullable_type}, true, attr); + std::vector<AggregateDataPtr> places {create_state(function, arena), + create_state(function, arena), + create_state(function, arena)}; + + auto input = ColumnHelper::create_nullable_column<DataTypeInt32>({5, 0, 3}, {0, 1, 0}); + const IColumn* columns[] = {input.get()}; + function->add(places[0], columns, 0, arena); + function->add(places[1], columns, 1, arena); + function->add(places[2], columns, 2, arena); + + auto nested_result = ColumnInt64::create(); + auto null_map = ColumnUInt8::create(); + auto result = ColumnNullable::create(std::move(nested_result), std::move(null_map)); + std::vector<uint64_t> repeats {4, 7, 6}; + function->insert_result_into_repeat_vec(places, 0, repeats, *result, places.size(), arena); + + ASSERT_EQ(result->size(), 3); + EXPECT_FALSE(result->is_null_at(0)); + EXPECT_TRUE(result->is_null_at(1)); + EXPECT_FALSE(result->is_null_at(2)); + EXPECT_EQ(assert_cast<const ColumnInt64&>(result->get_nested_column()).get_element(0), 20); + EXPECT_EQ(assert_cast<const ColumnInt64&>(result->get_nested_column()).get_element(2), 18); + + auto single_nested_result = ColumnInt64::create(); + auto single_null_map = ColumnUInt8::create(); + auto single_result = + ColumnNullable::create(std::move(single_nested_result), std::move(single_null_map)); + function->insert_result_into_repeat(places[0], 4, *single_result, arena); + function->insert_result_into_repeat(places[1], 7, *single_result, arena); + function->insert_result_into_repeat(places[2], 6, *single_result, arena); + ASSERT_EQ(single_result->size(), 3); + EXPECT_FALSE(single_result->is_null_at(0)); + EXPECT_TRUE(single_result->is_null_at(1)); + EXPECT_FALSE(single_result->is_null_at(2)); + EXPECT_EQ(assert_cast<const ColumnInt64&>(single_result->get_nested_column()).get_element(0), + 20); + EXPECT_EQ(assert_cast<const ColumnInt64&>(single_result->get_nested_column()).get_element(2), + 18); + + destroy_states(function, places); +} + +TEST(AggregateFunctionRepeatTest, NullableV2RepeatOutputAppendsAfterExistingRows) { + Arena arena; + AggregateFunctionAttr attr; + attr.enable_aggregate_function_null_v2 = true; + auto nullable_type = std::make_shared<DataTypeNullable>(std::make_shared<DataTypeInt32>()); + auto function = get_agg_function("sum", {nullable_type}, true, attr); + std::vector<AggregateDataPtr> places {create_state(function, arena), + create_state(function, arena), + create_state(function, arena)}; + + auto input = ColumnHelper::create_nullable_column<DataTypeInt32>({5, 0, 3}, {0, 1, 0}); + const IColumn* columns[] = {input.get()}; + function->add(places[0], columns, 0, arena); + function->add(places[1], columns, 1, arena); + function->add(places[2], columns, 2, arena); + + auto nested_result = ColumnInt64::create(); + auto null_map = ColumnUInt8::create(); + auto result = ColumnNullable::create(std::move(nested_result), std::move(null_map)); + result->insert_default(); + + std::vector<uint64_t> repeats {4, 7, 6}; + function->insert_result_into_repeat_vec(places, 0, repeats, *result, places.size(), arena); + + ASSERT_EQ(result->size(), 4); + EXPECT_TRUE(result->is_null_at(0)); + EXPECT_FALSE(result->is_null_at(1)); + EXPECT_TRUE(result->is_null_at(2)); + EXPECT_FALSE(result->is_null_at(3)); + EXPECT_EQ(result->get_nested_column().size(), result->get_null_map_column().size()); + EXPECT_EQ(assert_cast<const ColumnInt64&>(result->get_nested_column()).get_element(1), 20); + EXPECT_EQ(assert_cast<const ColumnInt64&>(result->get_nested_column()).get_element(3), 18); + + destroy_states(function, places); +} + +TEST(AggregateFunctionRepeatTest, NullableV2SumRepeatOutputAllValidRun) { + Arena arena; + AggregateFunctionAttr attr; + attr.enable_aggregate_function_null_v2 = true; + auto nullable_type = std::make_shared<DataTypeNullable>(std::make_shared<DataTypeInt32>()); + auto function = get_agg_function("sum", {nullable_type}, true, attr); + std::vector<AggregateDataPtr> places {create_state(function, arena), + create_state(function, arena), + create_state(function, arena)}; + + auto input = ColumnHelper::create_nullable_column<DataTypeInt32>({2, 4, 6}, {0, 0, 0}); + const IColumn* columns[] = {input.get()}; + function->add(places[0], columns, 0, arena); + function->add(places[1], columns, 1, arena); + function->add(places[2], columns, 2, arena); + + auto nested_result = ColumnInt64::create(); + auto null_map = ColumnUInt8::create(); + auto result = ColumnNullable::create(std::move(nested_result), std::move(null_map)); + std::vector<uint64_t> repeats {3, 5, 7}; + function->insert_result_into_repeat_vec(places, 0, repeats, *result, places.size(), arena); + + ASSERT_EQ(result->size(), 3); + EXPECT_FALSE(result->is_null_at(0)); + EXPECT_FALSE(result->is_null_at(1)); + EXPECT_FALSE(result->is_null_at(2)); + const auto& nested_column = assert_cast<const ColumnInt64&>(result->get_nested_column()); + EXPECT_EQ(nested_column.get_element(0), 6); + EXPECT_EQ(nested_column.get_element(1), 20); + EXPECT_EQ(nested_column.get_element(2), 42); + + destroy_states(function, places); +} + +TEST(AggregateFunctionRepeatTest, NullableV2SumRepeatOutputAllNull) { + Arena arena; + AggregateFunctionAttr attr; + attr.enable_aggregate_function_null_v2 = true; + auto nullable_type = std::make_shared<DataTypeNullable>(std::make_shared<DataTypeInt32>()); + auto function = get_agg_function("sum", {nullable_type}, true, attr); + std::vector<AggregateDataPtr> places {create_state(function, arena), + create_state(function, arena)}; + + auto input = ColumnHelper::create_nullable_column<DataTypeInt32>({0, 0}, {1, 1}); + const IColumn* columns[] = {input.get()}; + function->add(places[0], columns, 0, arena); + function->add(places[1], columns, 1, arena); + + auto nested_result = ColumnInt64::create(); + auto null_map = ColumnUInt8::create(); + auto result = ColumnNullable::create(std::move(nested_result), std::move(null_map)); + std::vector<uint64_t> repeats {3, 5}; + function->insert_result_into_repeat_vec(places, 0, repeats, *result, places.size(), arena); + + ASSERT_EQ(result->size(), 2); + EXPECT_TRUE(result->is_null_at(0)); + EXPECT_TRUE(result->is_null_at(1)); + EXPECT_EQ(result->get_nested_column().size(), result->get_null_map_column().size()); + + destroy_states(function, places); +} + +TEST(AggregateFunctionRepeatTest, NullableV2MinMaxRepeatOutput) { + Arena arena; + AggregateFunctionAttr attr; + attr.enable_aggregate_function_null_v2 = true; + auto nullable_type = std::make_shared<DataTypeNullable>(std::make_shared<DataTypeInt32>()); + auto min_function = get_agg_function("min", {nullable_type}, true, attr); + auto max_function = get_agg_function("max", {nullable_type}, true, attr); + + auto input = + ColumnHelper::create_nullable_column<DataTypeInt32>({5, 0, 9, 1, 0}, {0, 1, 0, 0, 1}); + const IColumn* columns[] = {input.get()}; + std::vector<uint64_t> repeats {3, 7, 11}; + + std::vector<AggregateDataPtr> min_places {create_state(min_function, arena), + create_state(min_function, arena), + create_state(min_function, arena)}; + min_function->add(min_places[0], columns, 0, arena); + min_function->add(min_places[1], columns, 1, arena); + min_function->add(min_places[2], columns, 2, arena); + min_function->add(min_places[2], columns, 3, arena); + + auto min_nested_result = ColumnInt32::create(); + auto min_null_map = ColumnUInt8::create(); + auto min_result = ColumnNullable::create(std::move(min_nested_result), std::move(min_null_map)); + min_function->insert_result_into_repeat_vec(min_places, 0, repeats, *min_result, + min_places.size(), arena); + + ASSERT_EQ(min_result->size(), 3); + EXPECT_FALSE(min_result->is_null_at(0)); + EXPECT_TRUE(min_result->is_null_at(1)); + EXPECT_FALSE(min_result->is_null_at(2)); + EXPECT_EQ(assert_cast<const ColumnInt32&>(min_result->get_nested_column()).get_element(0), 5); + EXPECT_EQ(assert_cast<const ColumnInt32&>(min_result->get_nested_column()).get_element(2), 1); + destroy_states(min_function, min_places); + + std::vector<AggregateDataPtr> max_places {create_state(max_function, arena), + create_state(max_function, arena), + create_state(max_function, arena)}; + max_function->add(max_places[0], columns, 0, arena); + max_function->add(max_places[1], columns, 4, arena); + max_function->add(max_places[2], columns, 2, arena); + max_function->add(max_places[2], columns, 3, arena); + + auto max_nested_result = ColumnInt32::create(); + auto max_null_map = ColumnUInt8::create(); + auto max_result = ColumnNullable::create(std::move(max_nested_result), std::move(max_null_map)); + max_function->insert_result_into_repeat_vec(max_places, 0, repeats, *max_result, + max_places.size(), arena); + + ASSERT_EQ(max_result->size(), 3); + EXPECT_FALSE(max_result->is_null_at(0)); + EXPECT_TRUE(max_result->is_null_at(1)); + EXPECT_FALSE(max_result->is_null_at(2)); + EXPECT_EQ(assert_cast<const ColumnInt32&>(max_result->get_nested_column()).get_element(0), 5); + EXPECT_EQ(assert_cast<const ColumnInt32&>(max_result->get_nested_column()).get_element(2), 9); + destroy_states(max_function, max_places); +} + +} // namespace +} // namespace doris --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
