be/src/exec/pipeline/rec_cte_shared_state.cpp

Source
// 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/pipeline/rec_cte_shared_state.h"

#include <gen_cpp/internal_service.pb.h>

#include "exec/common/distinct_agg_utils.h"
#include "exec/common/template_helpers.hpp"
#include "runtime/exec_env.h"
#include "runtime/query_context.h"
#include "runtime/runtime_state.h"
#include "util/brpc_client_cache.h"
#include "util/uid_util.h"

namespace doris {

RecCTESharedState::RecCTESharedState() = default;

RecCTESharedState::~RecCTESharedState() = default;

Status RecCTESharedState::emplace_block(RuntimeState* state, Block&& block) {
    if (agg_data) {
        auto num_rows = uint32_t(block.rows());
        ColumnRawPtrs raw_columns;
        std::vector<ColumnPtr> columns = block.get_columns_and_convert();
        for (auto& col : columns) {
            raw_columns.push_back(col.get());
        }

        std::visit(Overload {[&](std::monostate& arg) -> void {
                                 throw doris::Exception(ErrorCode::INTERNAL_ERROR,
                                                        "uninited hash table");
                             },
                             [&](auto& agg_method) -> void {
                                 SCOPED_TIMER(hash_table_compute_timer);
                                 using HashMethodType = std::decay_t<decltype(agg_method)>;
                                 using AggState = typename HashMethodType::State;

                                 AggState agg_state(raw_columns);
                                 agg_method.init_serialized_keys(raw_columns, num_rows);
                                 distinct_row.clear();

                                 size_t row = 0;
                                 auto creator = [&](const auto& ctor, auto& key, auto& origin) {
                                     HashMethodType::try_presis_key(key, origin, arena);
                                     ctor(key);
                                     distinct_row.push_back(row);
                                 };
                                 auto creator_for_null_key = [&]() { distinct_row.push_back(row); };

                                 SCOPED_TIMER(hash_table_emplace_timer);
                                 lazy_emplace_batch_void(agg_method, agg_state, num_rows, creator,
                                                         creator_for_null_key,
                                                         [&](uint32_t r) { row = r; });
                                 COUNTER_UPDATE(hash_table_input_counter, num_rows);
                             }},
                   agg_data->method_variant);

        if (distinct_row.size() == block.rows()) {
            blocks.emplace_back(std::move(block));
        } else if (!distinct_row.empty()) {
            auto distinct_block = MutableBlock(block.clone_empty());
            RETURN_IF_ERROR(block.append_to_block_by_selector(&distinct_block, distinct_row));
            blocks.emplace_back(distinct_block.to_block());
        }
    } else {
        blocks.emplace_back(std::move(block));
    }
    return Status::OK();
}

PTransmitRecCTEBlockParams RecCTESharedState::build_basic_param(RuntimeState* state,
                                                                const TRecCTETarget& target) const {
    PTransmitRecCTEBlockParams request;
    request.set_node_id(target.node_id);
    request.mutable_query_id()->CopyFrom(UniqueId(state->query_id()).to_proto());
    request.mutable_fragment_instance_id()->CopyFrom(
            UniqueId(target.fragment_instance_id).to_proto());
    return request;
}

Status RecCTESharedState::send_data_to_targets(RuntimeState* state, size_t round_offset) const {
    if (targets.size() == 0) {
        return Status::OK();
    }
    int send_multi_blocks_byte_size = state->query_options().exchange_multi_blocks_byte_size;
    int block_number_per_target =
            int(blocks.size() - round_offset + targets.size() - 1) / targets.size();
    for (auto target : targets) {
        auto stub = state->get_query_ctx()->exec_env()->brpc_internal_client_cache()->get_client(
                target.addr);
        if (!stub) {
            return Status::InternalError(fmt::format("Get rpc stub failed, host={}, port={}",
                                                     target.addr.hostname, target.addr.port));
        }

        // send blocks
        int step = block_number_per_target;
        while (round_offset < blocks.size() && step > 0) {
            PTransmitRecCTEBlockParams request = build_basic_param(state, target);
            auto current_bytes = 0;
            while (round_offset < blocks.size() && step > 0 &&
                   current_bytes < send_multi_blocks_byte_size) {
                auto* pblock = request.add_blocks();
                size_t uncompressed_bytes = 0;
                size_t compressed_bytes = 0;
                int64_t compress_time;
                RETURN_IF_ERROR(blocks[round_offset].serialize(
                        state->be_exec_version(), pblock, &uncompressed_bytes, &compressed_bytes,
                        &compress_time, state->fragement_transmission_compression_type()));
                round_offset++;
                step--;
                current_bytes += compressed_bytes;
            }
            request.set_eos(false);

            PTransmitRecCTEBlockResult result;
            brpc::Controller controller;
            controller.set_timeout_ms(
                    get_execution_rpc_timeout_ms(state->get_query_ctx()->execution_timeout()));

            stub->transmit_rec_cte_block(&controller, &request, &result, brpc::DoNothing());
            brpc::Join(controller.call_id());
            RETURN_IF_ERROR(Status::create(result.status()));
        }

        // send eos
        {
            PTransmitRecCTEBlockParams request = build_basic_param(state, target);
            request.set_eos(true);

            PTransmitRecCTEBlockResult result;
            brpc::Controller controller;
            stub->transmit_rec_cte_block(&controller, &request, &result, brpc::DoNothing());
            brpc::Join(controller.call_id());
            RETURN_IF_ERROR(Status::create(result.status()));
        }
    }
    return Status::OK();
}

} // namespace doris

Line	Count	Source
1		// Licensed to the Apache Software Foundation (ASF) under one
2		// or more contributor license agreements. See the NOTICE file
3		// distributed with this work for additional information
4		// regarding copyright ownership. The ASF licenses this file
5		// to you under the Apache License, Version 2.0 (the
6		// "License"); you may not use this file except in compliance
7		// with the License. You may obtain a copy of the License at
8		//
9		// http://www.apache.org/licenses/LICENSE-2.0
10		//
11		// Unless required by applicable law or agreed to in writing,
12		// software distributed under the License is distributed on an
13		// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14		// KIND, either express or implied. See the License for the
15		// specific language governing permissions and limitations
16		// under the License.
17
18		#include "exec/pipeline/rec_cte_shared_state.h"
19
20		#include <gen_cpp/internal_service.pb.h>
21
22		#include "exec/common/distinct_agg_utils.h"
23		#include "exec/common/template_helpers.hpp"
24		#include "runtime/exec_env.h"
25		#include "runtime/query_context.h"
26		#include "runtime/runtime_state.h"
27		#include "util/brpc_client_cache.h"
28		#include "util/uid_util.h"
29
30		namespace doris {
31
32	0	RecCTESharedState::RecCTESharedState() = default;
33
34	0	RecCTESharedState::~RecCTESharedState() = default;
35
36	0	Status RecCTESharedState::emplace_block(RuntimeState* state, Block&& block) {
37	0	if (agg_data) {
38	0	auto num_rows = uint32_t(block.rows());
39	0	ColumnRawPtrs raw_columns;
40	0	std::vector<ColumnPtr> columns = block.get_columns_and_convert();
41	0	for (auto& col : columns) {
42	0	raw_columns.push_back(col.get());
43	0	}
44
45	0	std::visit(Overload {[&](std::monostate& arg) -> void {
46	0	throw doris::Exception(ErrorCode::INTERNAL_ERROR,
47	0	"uninited hash table");
48	0	},
49	0	[&](auto& agg_method) -> void {
50	0	SCOPED_TIMER(hash_table_compute_timer);
51	0	using HashMethodType = std::decay_t<decltype(agg_method)>;
52	0	using AggState = typename HashMethodType::State;
53
54	0	AggState agg_state(raw_columns);
55	0	agg_method.init_serialized_keys(raw_columns, num_rows);
56	0	distinct_row.clear();
57
58	0	size_t row = 0;
59	0	auto creator = [&](const auto& ctor, auto& key, auto& origin) {
60	0	HashMethodType::try_presis_key(key, origin, arena);
61	0	ctor(key);
62	0	distinct_row.push_back(row);
63	0	}; Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_16MethodSerializedI9PHHashSetINS_9StringRefE11DefaultHashIS9_vEEEEEEvRT_ENKUlRKSE_RT0_RT1_E_clIN5phmap4priv12raw_hash_setINSP_17FlatHashSetPolicyIS9_EESB_NS_7EqualToIS9_EENS_10Allocator_IS9_EEE11constructorES9_S9_EEDaSH_SJ_SL_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodOneNumberIh21SmallFixedSizeHashSetIhEEEEEvRT_ENKUlRKSB_RT0_RT1_E_clIZNS9_12lazy_emplaceIRhRSJ_EEvOSB_RPvmOSF_EUlSC_E_hhEEDaSE_SG_SI_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodOneNumberIt9PHHashSetIt9HashCRC32ItEEEEEEvRT_ENKUlRKSD_RT0_RT1_E_clIN5phmap4priv12raw_hash_setINSO_17FlatHashSetPolicyItEESA_NS_7EqualToItEENS_10Allocator_ItEEE11constructorEttEEDaSG_SI_SK_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodOneNumberIj9PHHashSetIj9HashCRC32IjEEEEEEvRT_ENKUlRKSD_RT0_RT1_E_clIN5phmap4priv12raw_hash_setINSO_17FlatHashSetPolicyIjEESA_NS_7EqualToIjEENS_10Allocator_IjEEE11constructorEjjEEDaSG_SI_SK_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodOneNumberIm9PHHashSetIm9HashCRC32ImEEEEEEvRT_ENKUlRKSD_RT0_RT1_E_clIN5phmap4priv12raw_hash_setINSO_17FlatHashSetPolicyImEESA_NS_7EqualToImEENS_10Allocator_ImEEE11constructorEmmEEDaSG_SI_SK_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_19MethodStringNoCacheI9PHHashSetINS_9StringRefE11DefaultHashIS9_vEEEEEEvRT_ENKUlRKSE_RT0_RT1_E_clIN5phmap4priv12raw_hash_setINSP_17FlatHashSetPolicyIS9_EESB_NS_7EqualToIS9_EENS_10Allocator_IS9_EEE11constructorES9_S9_EEDaSH_SJ_SL_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodOneNumberIN4wide7integerILm128EjEE9PHHashSetISA_9HashCRC32ISA_EEEEEEvRT_ENKUlRKSG_RT0_RT1_E_clIN5phmap4priv12raw_hash_setINSR_17FlatHashSetPolicyISA_EESD_NS_7EqualToISA_EENS_10Allocator_ISA_EEE11constructorESA_SA_EEDaSJ_SL_SN_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodOneNumberIN4wide7integerILm256EjEE9PHHashSetISA_9HashCRC32ISA_EEEEEEvRT_ENKUlRKSG_RT0_RT1_E_clIN5phmap4priv12raw_hash_setINSR_17FlatHashSetPolicyISA_EESD_NS_7EqualToISA_EENS_10Allocator_ISA_EEE11constructorESA_SA_EEDaSJ_SL_SN_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodOneNumberIj9PHHashSetIj14HashMixWrapperIj9HashCRC32IjEEEEEEEvRT_ENKUlRKSF_RT0_RT1_E_clIN5phmap4priv12raw_hash_setINSQ_17FlatHashSetPolicyIjEESC_NS_7EqualToIjEENS_10Allocator_IjEEE11constructorEjjEEDaSI_SK_SM_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodOneNumberIm9PHHashSetIm14HashMixWrapperIm9HashCRC32ImEEEEEEEvRT_ENKUlRKSF_RT0_RT1_E_clIN5phmap4priv12raw_hash_setINSQ_17FlatHashSetPolicyImEESC_NS_7EqualToImEENS_10Allocator_ImEEE11constructorEmmEEDaSI_SK_SM_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_26MethodSingleNullableColumnINS_15MethodOneNumberIhNS_15DataWithNullKeyI21SmallFixedSizeHashSetIhEEEEEEEEEvRT_ENKUlRKSF_RT0_RT1_E_clIZNSB_12lazy_emplaceIRhRSN_EEvOSF_RPvmOSJ_EUlSG_E_hhEEDaSI_SK_SM_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_26MethodSingleNullableColumnINS_15MethodOneNumberItNS_15DataWithNullKeyI9PHHashSetIt9HashCRC32ItEEEEEEEEEEvRT_ENKUlRKSH_RT0_RT1_E_clIN5phmap4priv12raw_hash_setINSS_17FlatHashSetPolicyItEESC_NS_7EqualToItEENS_10Allocator_ItEEE11constructorEttEEDaSK_SM_SO_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_26MethodSingleNullableColumnINS_15MethodOneNumberIjNS_15DataWithNullKeyI9PHHashSetIj9HashCRC32IjEEEEEEEEEEvRT_ENKUlRKSH_RT0_RT1_E_clIN5phmap4priv12raw_hash_setINSS_17FlatHashSetPolicyIjEESC_NS_7EqualToIjEENS_10Allocator_IjEEE11constructorEjjEEDaSK_SM_SO_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_26MethodSingleNullableColumnINS_15MethodOneNumberImNS_15DataWithNullKeyI9PHHashSetIm9HashCRC32ImEEEEEEEEEEvRT_ENKUlRKSH_RT0_RT1_E_clIN5phmap4priv12raw_hash_setINSS_17FlatHashSetPolicyImEESC_NS_7EqualToImEENS_10Allocator_ImEEE11constructorEmmEEDaSK_SM_SO_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_26MethodSingleNullableColumnINS_15MethodOneNumberIjNS_15DataWithNullKeyI9PHHashSetIj14HashMixWrapperIj9HashCRC32IjEEEEEEEEEEEvRT_ENKUlRKSJ_RT0_RT1_E_clIN5phmap4priv12raw_hash_setINSU_17FlatHashSetPolicyIjEESE_NS_7EqualToIjEENS_10Allocator_IjEEE11constructorEjjEEDaSM_SO_SQ_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_26MethodSingleNullableColumnINS_15MethodOneNumberImNS_15DataWithNullKeyI9PHHashSetIm14HashMixWrapperIm9HashCRC32ImEEEEEEEEEEEvRT_ENKUlRKSJ_RT0_RT1_E_clIN5phmap4priv12raw_hash_setINSU_17FlatHashSetPolicyImEESE_NS_7EqualToImEENS_10Allocator_ImEEE11constructorEmmEEDaSM_SO_SQ_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_26MethodSingleNullableColumnINS_15MethodOneNumberIN4wide7integerILm128EjEENS_15DataWithNullKeyI9PHHashSetISB_9HashCRC32ISB_EEEEEEEEEEvRT_ENKUlRKSK_RT0_RT1_E_clIN5phmap4priv12raw_hash_setINSV_17FlatHashSetPolicyISB_EESF_NS_7EqualToISB_EENS_10Allocator_ISB_EEE11constructorESB_SB_EEDaSN_SP_SR_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_26MethodSingleNullableColumnINS_15MethodOneNumberIN4wide7integerILm256EjEENS_15DataWithNullKeyI9PHHashSetISB_9HashCRC32ISB_EEEEEEEEEEvRT_ENKUlRKSK_RT0_RT1_E_clIN5phmap4priv12raw_hash_setINSV_17FlatHashSetPolicyISB_EESF_NS_7EqualToISB_EENS_10Allocator_ISB_EEE11constructorESB_SB_EEDaSN_SP_SR_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_26MethodSingleNullableColumnINS_19MethodStringNoCacheINS_15DataWithNullKeyI9PHHashSetINS_9StringRefE11DefaultHashISB_vEEEEEEEEEEvRT_ENKUlRKSI_RT0_RT1_E_clIN5phmap4priv12raw_hash_setINST_17FlatHashSetPolicyISB_EESD_NS_7EqualToISB_EENS_10Allocator_ISB_EEE11constructorESB_SB_EEDaSL_SN_SP_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodKeysFixedI9PHHashSetIm9HashCRC32ImEEEEEEvRT_ENKUlRKSD_RT0_RT1_E_clIN5phmap4priv12raw_hash_setINSO_17FlatHashSetPolicyImEESA_NS_7EqualToImEENS_10Allocator_ImEEE11constructorEmmEEDaSG_SI_SK_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodKeysFixedI9PHHashSetINS_6UInt72E9HashCRC32IS9_EEEEEEvRT_ENKUlRKSE_RT0_RT1_E_clIN5phmap4priv12raw_hash_setINSP_17FlatHashSetPolicyIS9_EESB_NS_7EqualToIS9_EENS_10Allocator_IS9_EEE11constructorES9_S9_EEDaSH_SJ_SL_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodKeysFixedI9PHHashSetINS_6UInt96E9HashCRC32IS9_EEEEEEvRT_ENKUlRKSE_RT0_RT1_E_clIN5phmap4priv12raw_hash_setINSP_17FlatHashSetPolicyIS9_EESB_NS_7EqualToIS9_EENS_10Allocator_IS9_EEE11constructorES9_S9_EEDaSH_SJ_SL_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodKeysFixedI9PHHashSetINS_7UInt104E9HashCRC32IS9_EEEEEEvRT_ENKUlRKSE_RT0_RT1_E_clIN5phmap4priv12raw_hash_setINSP_17FlatHashSetPolicyIS9_EESB_NS_7EqualToIS9_EENS_10Allocator_IS9_EEE11constructorES9_S9_EEDaSH_SJ_SL_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodKeysFixedI9PHHashSetIN4wide7integerILm128EjEE9HashCRC32ISB_EEEEEEvRT_ENKUlRKSG_RT0_RT1_E_clIN5phmap4priv12raw_hash_setINSR_17FlatHashSetPolicyISB_EESD_NS_7EqualToISB_EENS_10Allocator_ISB_EEE11constructorESB_SB_EEDaSJ_SL_SN_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodKeysFixedI9PHHashSetINS_7UInt136E9HashCRC32IS9_EEEEEEvRT_ENKUlRKSE_RT0_RT1_E_clIN5phmap4priv12raw_hash_setINSP_17FlatHashSetPolicyIS9_EESB_NS_7EqualToIS9_EENS_10Allocator_IS9_EEE11constructorES9_S9_EEDaSH_SJ_SL_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodKeysFixedI9PHHashSetIN4wide7integerILm256EjEE9HashCRC32ISB_EEEEEEvRT_ENKUlRKSG_RT0_RT1_E_clIN5phmap4priv12raw_hash_setINSR_17FlatHashSetPolicyISB_EESD_NS_7EqualToISB_EENS_10Allocator_ISB_EEE11constructorESB_SB_EEDaSJ_SL_SN_
64	0	auto creator_for_null_key = [&]() { distinct_row.push_back(row); }; Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_26MethodSingleNullableColumnINS_15MethodOneNumberIhNS_15DataWithNullKeyI21SmallFixedSizeHashSetIhEEEEEEEEEvRT_ENKUlvE_clEv Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_26MethodSingleNullableColumnINS_15MethodOneNumberItNS_15DataWithNullKeyI9PHHashSetIt9HashCRC32ItEEEEEEEEEEvRT_ENKUlvE_clEv Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_26MethodSingleNullableColumnINS_15MethodOneNumberIjNS_15DataWithNullKeyI9PHHashSetIj9HashCRC32IjEEEEEEEEEEvRT_ENKUlvE_clEv Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_26MethodSingleNullableColumnINS_15MethodOneNumberImNS_15DataWithNullKeyI9PHHashSetIm9HashCRC32ImEEEEEEEEEEvRT_ENKUlvE_clEv Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_26MethodSingleNullableColumnINS_15MethodOneNumberIjNS_15DataWithNullKeyI9PHHashSetIj14HashMixWrapperIj9HashCRC32IjEEEEEEEEEEEvRT_ENKUlvE_clEv Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_26MethodSingleNullableColumnINS_15MethodOneNumberImNS_15DataWithNullKeyI9PHHashSetIm14HashMixWrapperIm9HashCRC32ImEEEEEEEEEEEvRT_ENKUlvE_clEv Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_26MethodSingleNullableColumnINS_15MethodOneNumberIN4wide7integerILm128EjEENS_15DataWithNullKeyI9PHHashSetISB_9HashCRC32ISB_EEEEEEEEEEvRT_ENKUlvE_clEv Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_26MethodSingleNullableColumnINS_15MethodOneNumberIN4wide7integerILm256EjEENS_15DataWithNullKeyI9PHHashSetISB_9HashCRC32ISB_EEEEEEEEEEvRT_ENKUlvE_clEv Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_26MethodSingleNullableColumnINS_19MethodStringNoCacheINS_15DataWithNullKeyI9PHHashSetINS_9StringRefE11DefaultHashISB_vEEEEEEEEEEvRT_ENKUlvE_clEv
65
66	0	SCOPED_TIMER(hash_table_emplace_timer);
67	0	lazy_emplace_batch_void(agg_method, agg_state, num_rows, creator,
68	0	creator_for_null_key,
69	0	[&](uint32_t r) { row = r; }); Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_16MethodSerializedI9PHHashSetINS_9StringRefE11DefaultHashIS9_vEEEEEEvRT_ENKUljE_clEj Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodOneNumberIh21SmallFixedSizeHashSetIhEEEEEvRT_ENKUljE_clEj Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodOneNumberIt9PHHashSetIt9HashCRC32ItEEEEEEvRT_ENKUljE_clEj Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodOneNumberIj9PHHashSetIj9HashCRC32IjEEEEEEvRT_ENKUljE_clEj Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodOneNumberIm9PHHashSetIm9HashCRC32ImEEEEEEvRT_ENKUljE_clEj Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_19MethodStringNoCacheI9PHHashSetINS_9StringRefE11DefaultHashIS9_vEEEEEEvRT_ENKUljE_clEj Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodOneNumberIN4wide7integerILm128EjEE9PHHashSetISA_9HashCRC32ISA_EEEEEEvRT_ENKUljE_clEj Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodOneNumberIN4wide7integerILm256EjEE9PHHashSetISA_9HashCRC32ISA_EEEEEEvRT_ENKUljE_clEj Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodOneNumberIj9PHHashSetIj14HashMixWrapperIj9HashCRC32IjEEEEEEEvRT_ENKUljE_clEj Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodOneNumberIm9PHHashSetIm14HashMixWrapperIm9HashCRC32ImEEEEEEEvRT_ENKUljE_clEj Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_26MethodSingleNullableColumnINS_15MethodOneNumberIhNS_15DataWithNullKeyI21SmallFixedSizeHashSetIhEEEEEEEEEvRT_ENKUljE_clEj Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_26MethodSingleNullableColumnINS_15MethodOneNumberItNS_15DataWithNullKeyI9PHHashSetIt9HashCRC32ItEEEEEEEEEEvRT_ENKUljE_clEj Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_26MethodSingleNullableColumnINS_15MethodOneNumberIjNS_15DataWithNullKeyI9PHHashSetIj9HashCRC32IjEEEEEEEEEEvRT_ENKUljE_clEj Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_26MethodSingleNullableColumnINS_15MethodOneNumberImNS_15DataWithNullKeyI9PHHashSetIm9HashCRC32ImEEEEEEEEEEvRT_ENKUljE_clEj Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_26MethodSingleNullableColumnINS_15MethodOneNumberIjNS_15DataWithNullKeyI9PHHashSetIj14HashMixWrapperIj9HashCRC32IjEEEEEEEEEEEvRT_ENKUljE_clEj Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_26MethodSingleNullableColumnINS_15MethodOneNumberImNS_15DataWithNullKeyI9PHHashSetIm14HashMixWrapperIm9HashCRC32ImEEEEEEEEEEEvRT_ENKUljE_clEj Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_26MethodSingleNullableColumnINS_15MethodOneNumberIN4wide7integerILm128EjEENS_15DataWithNullKeyI9PHHashSetISB_9HashCRC32ISB_EEEEEEEEEEvRT_ENKUljE_clEj Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_26MethodSingleNullableColumnINS_15MethodOneNumberIN4wide7integerILm256EjEENS_15DataWithNullKeyI9PHHashSetISB_9HashCRC32ISB_EEEEEEEEEEvRT_ENKUljE_clEj Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_26MethodSingleNullableColumnINS_19MethodStringNoCacheINS_15DataWithNullKeyI9PHHashSetINS_9StringRefE11DefaultHashISB_vEEEEEEEEEEvRT_ENKUljE_clEj Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodKeysFixedI9PHHashSetIm9HashCRC32ImEEEEEEvRT_ENKUljE_clEj Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodKeysFixedI9PHHashSetINS_6UInt72E9HashCRC32IS9_EEEEEEvRT_ENKUljE_clEj Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodKeysFixedI9PHHashSetINS_6UInt96E9HashCRC32IS9_EEEEEEvRT_ENKUljE_clEj Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodKeysFixedI9PHHashSetINS_7UInt104E9HashCRC32IS9_EEEEEEvRT_ENKUljE_clEj Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodKeysFixedI9PHHashSetIN4wide7integerILm128EjEE9HashCRC32ISB_EEEEEEvRT_ENKUljE_clEj Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodKeysFixedI9PHHashSetINS_7UInt136E9HashCRC32IS9_EEEEEEvRT_ENKUljE_clEj Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodKeysFixedI9PHHashSetIN4wide7integerILm256EjEE9HashCRC32ISB_EEEEEEvRT_ENKUljE_clEj
70	0	COUNTER_UPDATE(hash_table_input_counter, num_rows);
71	0	}}, Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_16MethodSerializedI9PHHashSetINS_9StringRefE11DefaultHashIS9_vEEEEEEvRT_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodOneNumberIh21SmallFixedSizeHashSetIhEEEEEvRT_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodOneNumberIt9PHHashSetIt9HashCRC32ItEEEEEEvRT_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodOneNumberIj9PHHashSetIj9HashCRC32IjEEEEEEvRT_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodOneNumberIm9PHHashSetIm9HashCRC32ImEEEEEEvRT_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_19MethodStringNoCacheI9PHHashSetINS_9StringRefE11DefaultHashIS9_vEEEEEEvRT_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodOneNumberIN4wide7integerILm128EjEE9PHHashSetISA_9HashCRC32ISA_EEEEEEvRT_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodOneNumberIN4wide7integerILm256EjEE9PHHashSetISA_9HashCRC32ISA_EEEEEEvRT_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodOneNumberIj9PHHashSetIj14HashMixWrapperIj9HashCRC32IjEEEEEEEvRT_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodOneNumberIm9PHHashSetIm14HashMixWrapperIm9HashCRC32ImEEEEEEEvRT_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_26MethodSingleNullableColumnINS_15MethodOneNumberIhNS_15DataWithNullKeyI21SmallFixedSizeHashSetIhEEEEEEEEEvRT_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_26MethodSingleNullableColumnINS_15MethodOneNumberItNS_15DataWithNullKeyI9PHHashSetIt9HashCRC32ItEEEEEEEEEEvRT_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_26MethodSingleNullableColumnINS_15MethodOneNumberIjNS_15DataWithNullKeyI9PHHashSetIj9HashCRC32IjEEEEEEEEEEvRT_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_26MethodSingleNullableColumnINS_15MethodOneNumberImNS_15DataWithNullKeyI9PHHashSetIm9HashCRC32ImEEEEEEEEEEvRT_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_26MethodSingleNullableColumnINS_15MethodOneNumberIjNS_15DataWithNullKeyI9PHHashSetIj14HashMixWrapperIj9HashCRC32IjEEEEEEEEEEEvRT_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_26MethodSingleNullableColumnINS_15MethodOneNumberImNS_15DataWithNullKeyI9PHHashSetIm14HashMixWrapperIm9HashCRC32ImEEEEEEEEEEEvRT_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_26MethodSingleNullableColumnINS_15MethodOneNumberIN4wide7integerILm128EjEENS_15DataWithNullKeyI9PHHashSetISB_9HashCRC32ISB_EEEEEEEEEEvRT_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_26MethodSingleNullableColumnINS_15MethodOneNumberIN4wide7integerILm256EjEENS_15DataWithNullKeyI9PHHashSetISB_9HashCRC32ISB_EEEEEEEEEEvRT_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_26MethodSingleNullableColumnINS_19MethodStringNoCacheINS_15DataWithNullKeyI9PHHashSetINS_9StringRefE11DefaultHashISB_vEEEEEEEEEEvRT_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodKeysFixedI9PHHashSetIm9HashCRC32ImEEEEEEvRT_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodKeysFixedI9PHHashSetINS_6UInt72E9HashCRC32IS9_EEEEEEvRT_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodKeysFixedI9PHHashSetINS_6UInt96E9HashCRC32IS9_EEEEEEvRT_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodKeysFixedI9PHHashSetINS_7UInt104E9HashCRC32IS9_EEEEEEvRT_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodKeysFixedI9PHHashSetIN4wide7integerILm128EjEE9HashCRC32ISB_EEEEEEvRT_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodKeysFixedI9PHHashSetINS_7UInt136E9HashCRC32IS9_EEEEEEvRT_ Unexecuted instantiation: rec_cte_shared_state.cpp:_ZZN5doris17RecCTESharedState13emplace_blockEPNS_12RuntimeStateEONS_5BlockEENK3$_1clINS_15MethodKeysFixedI9PHHashSetIN4wide7integerILm256EjEE9HashCRC32ISB_EEEEEEvRT_
72	0	agg_data->method_variant);
73
74	0	if (distinct_row.size() == block.rows()) {
75	0	blocks.emplace_back(std::move(block));
76	0	} else if (!distinct_row.empty()) {
77	0	auto distinct_block = MutableBlock(block.clone_empty());
78	0	RETURN_IF_ERROR(block.append_to_block_by_selector(&distinct_block, distinct_row));
79	0	blocks.emplace_back(distinct_block.to_block());
80	0	}
81	0	} else {
82	0	blocks.emplace_back(std::move(block));
83	0	}
84	0	return Status::OK();
85	0	}
86
87		PTransmitRecCTEBlockParams RecCTESharedState::build_basic_param(RuntimeState* state,
88	0	const TRecCTETarget& target) const {
89	0	PTransmitRecCTEBlockParams request;
90	0	request.set_node_id(target.node_id);
91	0	request.mutable_query_id()->CopyFrom(UniqueId(state->query_id()).to_proto());
92	0	request.mutable_fragment_instance_id()->CopyFrom(
93	0	UniqueId(target.fragment_instance_id).to_proto());
94	0	return request;
95	0	}
96
97	0	Status RecCTESharedState::send_data_to_targets(RuntimeState* state, size_t round_offset) const {
98	0	if (targets.size() == 0) {
99	0	return Status::OK();
100	0	}
101	0	int send_multi_blocks_byte_size = state->query_options().exchange_multi_blocks_byte_size;
102	0	int block_number_per_target =
103	0	int(blocks.size() - round_offset + targets.size() - 1) / targets.size();
104	0	for (auto target : targets) {
105	0	auto stub = state->get_query_ctx()->exec_env()->brpc_internal_client_cache()->get_client(
106	0	target.addr);
107	0	if (!stub) {
108	0	return Status::InternalError(fmt::format("Get rpc stub failed, host={}, port={}",
109	0	target.addr.hostname, target.addr.port));
110	0	}
111
112		// send blocks
113	0	int step = block_number_per_target;
114	0	while (round_offset < blocks.size() && step > 0) {
115	0	PTransmitRecCTEBlockParams request = build_basic_param(state, target);
116	0	auto current_bytes = 0;
117	0	while (round_offset < blocks.size() && step > 0 &&
118	0	current_bytes < send_multi_blocks_byte_size) {
119	0	auto* pblock = request.add_blocks();
120	0	size_t uncompressed_bytes = 0;
121	0	size_t compressed_bytes = 0;
122	0	int64_t compress_time;
123	0	RETURN_IF_ERROR(blocks[round_offset].serialize(
124	0	state->be_exec_version(), pblock, &uncompressed_bytes, &compressed_bytes,
125	0	&compress_time, state->fragement_transmission_compression_type()));
126	0	round_offset++;
127	0	step--;
128	0	current_bytes += compressed_bytes;
129	0	}
130	0	request.set_eos(false);
131
132	0	PTransmitRecCTEBlockResult result;
133	0	brpc::Controller controller;
134	0	controller.set_timeout_ms(
135	0	get_execution_rpc_timeout_ms(state->get_query_ctx()->execution_timeout()));
136
137	0	stub->transmit_rec_cte_block(&controller, &request, &result, brpc::DoNothing());
138	0	brpc::Join(controller.call_id());
139	0	RETURN_IF_ERROR(Status::create(result.status()));
140	0	}
141
142		// send eos
143	0	{
144	0	PTransmitRecCTEBlockParams request = build_basic_param(state, target);
145	0	request.set_eos(true);
146
147	0	PTransmitRecCTEBlockResult result;
148	0	brpc::Controller controller;
149	0	stub->transmit_rec_cte_block(&controller, &request, &result, brpc::DoNothing());
150	0	brpc::Join(controller.call_id());
151	0	RETURN_IF_ERROR(Status::create(result.status()));
152	0	}
153	0	}
154	0	return Status::OK();
155	0	}
156
157		} // namespace doris

Coverage Report

Created: 2026-08-14 13:56