Coverage Report

Created: 2026-08-31 02:43

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/exec/sink/vrow_distribution.h
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
#pragma once
19
20
// IWYU pragma: no_include <bits/chrono.h>
21
#include <fmt/format.h>
22
#include <gen_cpp/FrontendService.h>
23
#include <gen_cpp/FrontendService_types.h>
24
#include <gen_cpp/PaloInternalService_types.h>
25
#include <glog/logging.h>
26
27
#include <cstdint>
28
#include <functional>
29
#include <string>
30
#include <unordered_set>
31
#include <vector>
32
33
#include "common/status.h"
34
#include "core/block/block.h"
35
#include "core/custom_allocator.h"
36
#include "exec/sink/vtablet_block_convertor.h"
37
#include "exec/sink/vtablet_finder.h"
38
#include "exprs/vexpr_context.h"
39
#include "exprs/vexpr_fwd.h"
40
#include "runtime/runtime_profile.h"
41
#include "runtime/runtime_state.h"
42
#include "storage/tablet_info.h"
43
44
namespace doris {
45
46
class IndexChannel;
47
class VNodeChannel;
48
49
// <row_idx, partition_id, tablet_id>
50
class RowPartTabletIds {
51
public:
52
    DorisVector<uint32_t> row_ids;
53
    DorisVector<int64_t> partition_ids;
54
    DorisVector<int64_t> tablet_ids;
55
56
0
    std::string debug_string() const {
57
0
        std::string value;
58
0
        value.reserve(row_ids.size() * 15);
59
0
        for (int i = 0; i < row_ids.size(); i++) {
60
0
            if (i < tablet_ids.size()) {
61
0
                value.append(
62
0
                        fmt::format("[{}, {}, {}]", row_ids[i], partition_ids[i], tablet_ids[i]));
63
0
            } else {
64
0
                value.append(fmt::format("[{}, {}]", row_ids[i], partition_ids[i]));
65
0
            }
66
0
        }
67
0
        return value;
68
0
    }
69
};
70
71
// void* for caller
72
using CreatePartitionCallback = Status (*)(void*, TCreatePartitionResult*);
73
74
class VRowDistribution {
75
public:
76
    // only used to pass parameters for VRowDistribution
77
    struct VRowDistributionContext {
78
        RuntimeState* state = nullptr;
79
        OlapTableBlockConvertor* block_convertor = nullptr;
80
        OlapTabletFinder* tablet_finder = nullptr;
81
        VOlapTablePartitionParam* vpartition = nullptr;
82
        RuntimeProfile::Counter* add_partition_request_timer = nullptr;
83
        int64_t txn_id = -1;
84
        ObjectPool* pool = nullptr;
85
        OlapTableLocationParam* location = nullptr;
86
        const VExprContextSPtrs* vec_output_expr_ctxs = nullptr;
87
        std::shared_ptr<OlapTableSchemaParam> schema;
88
        void* caller = nullptr;
89
        CreatePartitionCallback create_partition_callback;
90
    };
91
    friend class VTabletWriter;
92
    friend class VTabletWriterV2;
93
94
24
    VRowDistribution() = default;
95
24
    virtual ~VRowDistribution() = default;
96
97
8
    void init(VRowDistributionContext ctx) {
98
8
        _state = ctx.state;
99
8
        _batch_size = std::max(_state->batch_size(), 8192);
100
8
        _block_convertor = ctx.block_convertor;
101
8
        _tablet_finder = ctx.tablet_finder;
102
8
        _vpartition = ctx.vpartition;
103
8
        _add_partition_request_timer = ctx.add_partition_request_timer;
104
8
        _txn_id = ctx.txn_id;
105
8
        _pool = ctx.pool;
106
8
        _location = ctx.location;
107
8
        _vec_output_expr_ctxs = ctx.vec_output_expr_ctxs;
108
8
        _schema = ctx.schema;
109
8
        _caller = ctx.caller;
110
8
        _create_partition_callback = ctx.create_partition_callback;
111
8
    }
112
113
0
    void output_profile_info(RuntimeProfile* profile) {
114
0
        if (!_add_partition_request_times.empty()) {
115
0
            std::stringstream ss;
116
0
            ss << "[";
117
0
            for (size_t i = 0; i < _add_partition_request_times.size(); ++i) {
118
0
                if (i > 0) {
119
0
                    ss << ", ";
120
0
                }
121
0
                ss << PrettyPrinter::print(_add_partition_request_times[i], TUnit::TIME_NS);
122
0
            }
123
0
            ss << "]";
124
0
            profile->add_info_string("AddPartitionRequestTimeList", ss.str());
125
0
        }
126
0
    }
127
128
8
    Status open(RowDescriptor* output_row_desc) {
129
8
        if (_vpartition->is_auto_partition()) {
130
2
            auto [part_ctxs, part_funcs] = _get_partition_function();
131
2
            for (auto part_ctx : part_ctxs) {
132
2
                RETURN_IF_ERROR(part_ctx->prepare(_state, *output_row_desc));
133
2
                part_ctx->set_auto_partition_boundary_context();
134
2
                RETURN_IF_ERROR(part_ctx->open(_state));
135
2
            }
136
2
        }
137
8
        for (const auto& index : _schema->indexes()) {
138
8
            auto& where_clause = index->where_clause;
139
8
            if (where_clause != nullptr) {
140
0
                RETURN_IF_ERROR(where_clause->prepare(_state, *output_row_desc));
141
0
                RETURN_IF_ERROR(where_clause->open(_state));
142
0
            }
143
8
        }
144
8
        return Status::OK();
145
8
    }
146
147
    // auto partition
148
    // mv where clause
149
    // v1 needs index->node->row_ids - tabletids
150
    // v2 needs index,tablet->rowids
151
    Status generate_rows_distribution(Block& input_block, std::shared_ptr<Block>& block,
152
                                      std::vector<RowPartTabletIds>& row_part_tablet_ids,
153
                                      int64_t& rows_stat_val);
154
    // have 2 ways remind to deal batching block:
155
    // 1. in row_distribution, _batching_rows reaches the threshold, this class set _deal_batched = true.
156
    // 2. in caller, after last block and before close, set _deal_batched = true.
157
3
    bool need_deal_batching() const { return _deal_batched && _batching_rows > 0; }
158
    // create partitions when need for auto-partition table using #_partitions_need_create.
159
    Status automatic_create_partition();
160
    void clear_batching_stats();
161
3
    const std::vector<bool>& get_skipped() const { return _skip; } // skipped in last round
162
163
    // for auto partition
164
    std::unique_ptr<MutableBlock> _batching_block; // same structure with input_block
165
    bool _deal_batched = false; // If true, send batched block before any block's append.
166
167
private:
168
    std::pair<VExprContextSPtrs, VExprSPtrs> _get_partition_function();
169
170
    Status _save_missing_values(const Block& input_block,
171
                                std::vector<std::vector<std::string>>& col_strs, int col_size,
172
                                Block* block, const std::vector<uint32_t>& filter,
173
                                const std::vector<const NullMap*>& col_null_maps);
174
175
    void _get_tablet_ids(Block* block, int32_t index_idx, std::vector<int64_t>& tablet_ids);
176
177
    void _filter_block_by_skip(Block* block, RowPartTabletIds& row_part_tablet_id);
178
179
    Status _filter_block_by_skip_and_where_clause(Block* block,
180
                                                  const VExprContextSPtr& where_clause,
181
                                                  RowPartTabletIds& row_part_tablet_id);
182
183
    Status _filter_block(Block* block, std::vector<RowPartTabletIds>& row_part_tablet_ids);
184
185
    Status _generate_rows_distribution_for_auto_partition(
186
            const Block& input_block, Block* block, const std::vector<uint16_t>& partition_col_idx,
187
            bool has_filtered_rows, std::vector<RowPartTabletIds>& row_part_tablet_ids,
188
            int64_t& rows_stat_val);
189
    // the whole process to deal missing rows. will call _save_missing_values
190
    Status _deal_missing_map(const Block& input_block, Block* block,
191
                             const std::vector<uint16_t>& partition_cols_idx,
192
                             int64_t& rows_stat_val);
193
194
    Status _generate_rows_distribution_for_non_auto_partition(
195
            Block* block, bool has_filtered_rows,
196
            std::vector<RowPartTabletIds>& row_part_tablet_ids);
197
198
    Status _generate_rows_distribution_for_auto_overwrite(
199
            const Block& input_block, Block* block, const std::vector<uint16_t>& partition_cols_idx,
200
            bool has_filtered_rows, std::vector<RowPartTabletIds>& row_part_tablet_ids,
201
            int64_t& rows_stat_val);
202
    Status _replace_overwriting_partition();
203
204
    void _reset_row_part_tablet_ids(std::vector<RowPartTabletIds>& row_part_tablet_ids,
205
                                    int64_t rows);
206
    void _reset_find_tablets(int64_t rows);
207
208
    struct NullableStringListHash {
209
6
        std::size_t _hash(const TNullableStringLiteral& arg) const {
210
6
            if (arg.is_null) {
211
0
                return 0;
212
0
            }
213
6
            return std::hash<std::string>()(arg.value);
214
6
        }
215
6
        std::size_t operator()(const std::vector<TNullableStringLiteral>& arg) const {
216
6
            std::size_t result = 0;
217
6
            for (const auto& v : arg) {
218
6
                result = (result << 1) ^ _hash(v);
219
6
            }
220
6
            return result;
221
6
        }
222
    };
223
224
    RuntimeState* _state = nullptr;
225
    int _batch_size = 0;
226
227
    // for auto partitions
228
    std::vector<std::vector<TNullableStringLiteral>> _partitions_need_create;
229
    size_t _batching_rows = 0, _batching_bytes = 0;
230
    std::unordered_set<std::vector<TNullableStringLiteral>, NullableStringListHash> _deduper;
231
232
    OlapTableBlockConvertor* _block_convertor = nullptr;
233
    OlapTabletFinder* _tablet_finder = nullptr;
234
    VOlapTablePartitionParam* _vpartition = nullptr;
235
    RuntimeProfile::Counter* _add_partition_request_timer = nullptr;
236
    int64_t _txn_id = -1;
237
    ObjectPool* _pool = nullptr;
238
    OlapTableLocationParam* _location = nullptr;
239
240
    // Record each auto-partition request time for detailed profiling
241
    std::vector<int64_t> _add_partition_request_times;
242
    // int64_t _number_output_rows = 0;
243
    const VExprContextSPtrs* _vec_output_expr_ctxs = nullptr;
244
    // generally it's writer's on_partitions_created
245
    CreatePartitionCallback _create_partition_callback = nullptr;
246
    void* _caller = nullptr;
247
    std::shared_ptr<OlapTableSchemaParam> _schema;
248
249
    // reuse for find_tablet. save partitions found by find_tablets
250
    std::vector<VOlapTablePartition*> _partitions;
251
    std::vector<bool> _skip;
252
    std::vector<uint32_t> _tablet_indexes;
253
    std::vector<int64_t> _tablet_ids;
254
    std::vector<uint32_t> _missing_map; // indice of missing values in partition_col
255
    // for auto detect overwrite partition
256
    std::set<int64_t> _new_partition_ids; // if contains, not to replace it again.
257
};
258
259
} // namespace doris