/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 {
class MemTracker;

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

Coverage Report

Created: 2024-11-21 14:46