/root/doris/be/src/vec/runtime/partitioner.h

Source (jump to first uncovered line)
// 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 "util/runtime_profile.h"
#include "vec/exprs/vexpr.h"
#include "vec/exprs/vexpr_context.h"

namespace doris {

namespace vectorized {

struct ChannelField {
    const void* channel_id;
    const uint32_t len;

    template <typename T>
    const T* get() const {
        CHECK_EQ(sizeof(T), len) << " sizeof(T): " << sizeof(T) << " len: " << len;
        return reinterpret_cast<const T*>(channel_id);
    }
};

class PartitionerBase {
public:
    PartitionerBase(size_t partition_count) : _partition_count(partition_count) {}
    virtual ~PartitionerBase() = default;

    virtual Status init(const std::vector<TExpr>& texprs) = 0;

    virtual Status prepare(RuntimeState* state, const RowDescriptor& row_desc) = 0;

    virtual Status open(RuntimeState* state) = 0;

    virtual Status do_partitioning(RuntimeState* state, Block* block) const = 0;

    virtual ChannelField get_channel_ids() const = 0;

    virtual Status clone(RuntimeState* state, std::unique_ptr<PartitionerBase>& partitioner) = 0;

protected:
    const size_t _partition_count;
};

template <typename HashValueType, typename ChannelIds>
class Partitioner : public PartitionerBase {
public:
    Partitioner(int partition_count) : PartitionerBase(partition_count) {}
    ~Partitioner() override = default;

    Status init(const std::vector<TExpr>& texprs) override {
        return VExpr::create_expr_trees(texprs, _partition_expr_ctxs);
    }

    Status prepare(RuntimeState* state, const RowDescriptor& row_desc) override {
        return VExpr::prepare(_partition_expr_ctxs, state, row_desc);
    }

    Status open(RuntimeState* state) override { return VExpr::open(_partition_expr_ctxs, state); }

    Status do_partitioning(RuntimeState* state, Block* block) const override;

    ChannelField get_channel_ids() const override {
        return {_hash_vals.data(), sizeof(HashValueType)};
    }

protected:
    Status _get_partition_column_result(Block* block, std::vector<int>& result) const {
        int counter = 0;
        for (auto ctx : _partition_expr_ctxs) {
            RETURN_IF_ERROR(ctx->execute(block, &result[counter++]));
        }
        return Status::OK();
    }

    virtual void _do_hash(const ColumnPtr& column, HashValueType* __restrict result,
                          int idx) const = 0;

    VExprContextSPtrs _partition_expr_ctxs;
    mutable std::vector<HashValueType> _hash_vals;
};

template <typename ChannelIds>
class XXHashPartitioner final : public Partitioner<uint64_t, ChannelIds> {
public:
    using Base = Partitioner<uint64_t, ChannelIds>;
    XXHashPartitioner(int partition_count) : Partitioner<uint64_t, ChannelIds>(partition_count) {}
    ~XXHashPartitioner() override = default;

    Status clone(RuntimeState* state, std::unique_ptr<PartitionerBase>& partitioner) override;

private:
    void _do_hash(const ColumnPtr& column, uint64_t* __restrict result, int idx) const override;
};

template <typename ChannelIds>
class Crc32HashPartitioner final : public Partitioner<uint32_t, ChannelIds> {
public:
    using Base = Partitioner<uint32_t, ChannelIds>;
    Crc32HashPartitioner(int partition_count)
            : Partitioner<uint32_t, ChannelIds>(partition_count) {}
    ~Crc32HashPartitioner() override = default;

    Status clone(RuntimeState* state, std::unique_ptr<PartitionerBase>& partitioner) override;

private:
    void _do_hash(const ColumnPtr& column, uint32_t* __restrict result, int idx) const override;
};

struct SpillPartitionChannelIds {
    template <typename HashValueType>
    HashValueType operator()(HashValueType l, size_t r) {
        return ((l >> 16) | (l << 16)) % r;
    }
};

} // namespace vectorized
} // namespace doris

Line	Count	Source (jump to first uncovered line)
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		#pragma once
19
20		#include "util/runtime_profile.h"
21		#include "vec/exprs/vexpr.h"
22		#include "vec/exprs/vexpr_context.h"
23
24		namespace doris {
25
26		namespace vectorized {
27
28		struct ChannelField {
29		const void* channel_id;
30		const uint32_t len;
31
32		template <typename T>
33	0	const T* get() const {
34	0	CHECK_EQ(sizeof(T), len) << " sizeof(T): " << sizeof(T) << " len: " << len;
35	0	return reinterpret_cast<const T*>(channel_id);
36	0	}
37		};
38
39		class PartitionerBase {
40		public:
41	0	PartitionerBase(size_t partition_count) : _partition_count(partition_count) {}
42	0	virtual ~PartitionerBase() = default;
43
44		virtual Status init(const std::vector<TExpr>& texprs) = 0;
45
46		virtual Status prepare(RuntimeState* state, const RowDescriptor& row_desc) = 0;
47
48		virtual Status open(RuntimeState* state) = 0;
49
50		virtual Status do_partitioning(RuntimeState* state, Block* block) const = 0;
51
52		virtual ChannelField get_channel_ids() const = 0;
53
54		virtual Status clone(RuntimeState* state, std::unique_ptr<PartitionerBase>& partitioner) = 0;
55
56		protected:
57		const size_t _partition_count;
58		};
59
60		template <typename HashValueType, typename ChannelIds>
61		class Partitioner : public PartitionerBase {
62		public:
63	0	Partitioner(int partition_count) : PartitionerBase(partition_count) {} Unexecuted instantiation: _ZN5doris10vectorized11PartitionerIjNS0_17ShuffleChannelIdsEEC2Ei Unexecuted instantiation: _ZN5doris10vectorized11PartitionerImNS_8pipeline23LocalExchangeChannelIdsEEC2Ei Unexecuted instantiation: _ZN5doris10vectorized11PartitionerImNS0_17ShuffleChannelIdsEEC2Ei Unexecuted instantiation: _ZN5doris10vectorized11PartitionerIjNS0_24SpillPartitionChannelIdsEEC2Ei
64	0	~Partitioner() override = default; Unexecuted instantiation: _ZN5doris10vectorized11PartitionerIjNS0_17ShuffleChannelIdsEED2Ev Unexecuted instantiation: _ZN5doris10vectorized11PartitionerImNS_8pipeline23LocalExchangeChannelIdsEED2Ev Unexecuted instantiation: _ZN5doris10vectorized11PartitionerImNS0_17ShuffleChannelIdsEED2Ev Unexecuted instantiation: _ZN5doris10vectorized11PartitionerIjNS0_24SpillPartitionChannelIdsEED2Ev
65
66	0	Status init(const std::vector<TExpr>& texprs) override {
67	0	return VExpr::create_expr_trees(texprs, _partition_expr_ctxs);
68	0	} Unexecuted instantiation: _ZN5doris10vectorized11PartitionerIjNS0_17ShuffleChannelIdsEE4initERKSt6vectorINS_5TExprESaIS5_EE Unexecuted instantiation: _ZN5doris10vectorized11PartitionerImNS_8pipeline23LocalExchangeChannelIdsEE4initERKSt6vectorINS_5TExprESaIS6_EE Unexecuted instantiation: _ZN5doris10vectorized11PartitionerImNS0_17ShuffleChannelIdsEE4initERKSt6vectorINS_5TExprESaIS5_EE Unexecuted instantiation: _ZN5doris10vectorized11PartitionerIjNS0_24SpillPartitionChannelIdsEE4initERKSt6vectorINS_5TExprESaIS5_EE
69
70	0	Status prepare(RuntimeState* state, const RowDescriptor& row_desc) override {
71	0	return VExpr::prepare(_partition_expr_ctxs, state, row_desc);
72	0	} Unexecuted instantiation: _ZN5doris10vectorized11PartitionerIjNS0_17ShuffleChannelIdsEE7prepareEPNS_12RuntimeStateERKNS_13RowDescriptorE Unexecuted instantiation: _ZN5doris10vectorized11PartitionerImNS_8pipeline23LocalExchangeChannelIdsEE7prepareEPNS_12RuntimeStateERKNS_13RowDescriptorE Unexecuted instantiation: _ZN5doris10vectorized11PartitionerImNS0_17ShuffleChannelIdsEE7prepareEPNS_12RuntimeStateERKNS_13RowDescriptorE Unexecuted instantiation: _ZN5doris10vectorized11PartitionerIjNS0_24SpillPartitionChannelIdsEE7prepareEPNS_12RuntimeStateERKNS_13RowDescriptorE
73
74	0	Status open(RuntimeState* state) override { return VExpr::open(_partition_expr_ctxs, state); } Unexecuted instantiation: _ZN5doris10vectorized11PartitionerIjNS0_17ShuffleChannelIdsEE4openEPNS_12RuntimeStateE Unexecuted instantiation: _ZN5doris10vectorized11PartitionerImNS_8pipeline23LocalExchangeChannelIdsEE4openEPNS_12RuntimeStateE Unexecuted instantiation: _ZN5doris10vectorized11PartitionerImNS0_17ShuffleChannelIdsEE4openEPNS_12RuntimeStateE Unexecuted instantiation: _ZN5doris10vectorized11PartitionerIjNS0_24SpillPartitionChannelIdsEE4openEPNS_12RuntimeStateE
75
76		Status do_partitioning(RuntimeState* state, Block* block) const override;
77
78	0	ChannelField get_channel_ids() const override {
79	0	return {_hash_vals.data(), sizeof(HashValueType)};
80	0	} Unexecuted instantiation: _ZNK5doris10vectorized11PartitionerIjNS0_17ShuffleChannelIdsEE15get_channel_idsEv Unexecuted instantiation: _ZNK5doris10vectorized11PartitionerImNS_8pipeline23LocalExchangeChannelIdsEE15get_channel_idsEv Unexecuted instantiation: _ZNK5doris10vectorized11PartitionerImNS0_17ShuffleChannelIdsEE15get_channel_idsEv Unexecuted instantiation: _ZNK5doris10vectorized11PartitionerIjNS0_24SpillPartitionChannelIdsEE15get_channel_idsEv
81
82		protected:
83	0	Status _get_partition_column_result(Block* block, std::vector<int>& result) const {
84	0	int counter = 0;
85	0	for (auto ctx : _partition_expr_ctxs) {
86	0	RETURN_IF_ERROR(ctx->execute(block, &result[counter++]));
87	0	}
88	0	return Status::OK();
89	0	} Unexecuted instantiation: _ZNK5doris10vectorized11PartitionerImNS_8pipeline23LocalExchangeChannelIdsEE28_get_partition_column_resultEPNS0_5BlockERSt6vectorIiSaIiEE Unexecuted instantiation: _ZNK5doris10vectorized11PartitionerImNS0_17ShuffleChannelIdsEE28_get_partition_column_resultEPNS0_5BlockERSt6vectorIiSaIiEE Unexecuted instantiation: _ZNK5doris10vectorized11PartitionerIjNS0_17ShuffleChannelIdsEE28_get_partition_column_resultEPNS0_5BlockERSt6vectorIiSaIiEE Unexecuted instantiation: _ZNK5doris10vectorized11PartitionerIjNS0_24SpillPartitionChannelIdsEE28_get_partition_column_resultEPNS0_5BlockERSt6vectorIiSaIiEE
90
91		virtual void _do_hash(const ColumnPtr& column, HashValueType* __restrict result,
92		int idx) const = 0;
93
94		VExprContextSPtrs _partition_expr_ctxs;
95		mutable std::vector<HashValueType> _hash_vals;
96		};
97
98		template <typename ChannelIds>
99		class XXHashPartitioner final : public Partitioner<uint64_t, ChannelIds> {
100		public:
101		using Base = Partitioner<uint64_t, ChannelIds>;
102	0	XXHashPartitioner(int partition_count) : Partitioner<uint64_t, ChannelIds>(partition_count) {} Unexecuted instantiation: _ZN5doris10vectorized17XXHashPartitionerINS_8pipeline23LocalExchangeChannelIdsEEC2Ei Unexecuted instantiation: _ZN5doris10vectorized17XXHashPartitionerINS0_17ShuffleChannelIdsEEC2Ei
103	0	~XXHashPartitioner() override = default; Unexecuted instantiation: _ZN5doris10vectorized17XXHashPartitionerINS_8pipeline23LocalExchangeChannelIdsEED2Ev Unexecuted instantiation: _ZN5doris10vectorized17XXHashPartitionerINS0_17ShuffleChannelIdsEED2Ev
104
105		Status clone(RuntimeState* state, std::unique_ptr<PartitionerBase>& partitioner) override;
106
107		private:
108		void _do_hash(const ColumnPtr& column, uint64_t* __restrict result, int idx) const override;
109		};
110
111		template <typename ChannelIds>
112		class Crc32HashPartitioner final : public Partitioner<uint32_t, ChannelIds> {
113		public:
114		using Base = Partitioner<uint32_t, ChannelIds>;
115		Crc32HashPartitioner(int partition_count)
116	0	: Partitioner<uint32_t, ChannelIds>(partition_count) {} Unexecuted instantiation: _ZN5doris10vectorized20Crc32HashPartitionerINS0_17ShuffleChannelIdsEEC2Ei Unexecuted instantiation: _ZN5doris10vectorized20Crc32HashPartitionerINS0_24SpillPartitionChannelIdsEEC2Ei
117	0	~Crc32HashPartitioner() override = default; Unexecuted instantiation: _ZN5doris10vectorized20Crc32HashPartitionerINS0_17ShuffleChannelIdsEED2Ev Unexecuted instantiation: _ZN5doris10vectorized20Crc32HashPartitionerINS0_24SpillPartitionChannelIdsEED2Ev
118
119		Status clone(RuntimeState* state, std::unique_ptr<PartitionerBase>& partitioner) override;
120
121		private:
122		void _do_hash(const ColumnPtr& column, uint32_t* __restrict result, int idx) const override;
123		};
124
125		struct SpillPartitionChannelIds {
126		template <typename HashValueType>
127	0	HashValueType operator()(HashValueType l, size_t r) {
128	0	return ((l >> 16) \| (l << 16)) % r;
129	0	}
130		};
131
132		} // namespace vectorized
133		} // namespace doris

Coverage Report

Created: 2025-04-16 14:10