Coverage Report

Created: 2026-06-29 14:06

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/exec/sink/vrow_distribution.cpp
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/sink/vrow_distribution.h"
19
20
#include <gen_cpp/FrontendService.h>
21
#include <gen_cpp/FrontendService_types.h>
22
#include <glog/logging.h>
23
24
#include <cstdint>
25
#include <memory>
26
#include <string>
27
28
#include "common/cast_set.h"
29
#include "common/logging.h"
30
#include "common/status.h"
31
#include "core/assert_cast.h"
32
#include "core/column/column.h"
33
#include "core/column/column_const.h"
34
#include "core/column/column_nullable.h"
35
#include "core/column/column_vector.h"
36
#include "core/data_type/data_type.h"
37
#include "exec/sink/writer/vtablet_writer.h"
38
#include "runtime/exec_env.h"
39
#include "runtime/query_context.h"
40
#include "runtime/runtime_state.h"
41
#include "service/backend_options.h"
42
#include "util/client_cache.h"
43
#include "util/debug_points.h"
44
#include "util/thrift_rpc_helper.h"
45
46
namespace doris {
47
48
13
std::pair<VExprContextSPtrs, VExprSPtrs> VRowDistribution::_get_partition_function() {
49
13
    return {_vpartition->get_part_func_ctx(), _vpartition->get_partition_function()};
50
13
}
51
52
Status VRowDistribution::_save_missing_values(
53
        const Block& input_block,
54
        std::vector<std::vector<std::string>>& col_strs, // non-const ref for move
55
        int col_size, Block* block, const std::vector<uint32_t>& filter,
56
2
        const std::vector<const NullMap*>& col_null_maps) {
57
    // de-duplication for new partitions but save all rows.
58
2
    RETURN_IF_ERROR(
59
2
            _batching_block->add_rows(&input_block, filter.data(), filter.data() + filter.size()));
60
2
    std::vector<TNullableStringLiteral> cur_row_values;
61
6
    for (int row = 0; row < col_strs[0].size(); ++row) {
62
4
        cur_row_values.clear();
63
8
        for (int col = 0; col < col_size; ++col) {
64
4
            TNullableStringLiteral node;
65
            // OlapTableBlockConvertor::_validate_data() materializes destination slots so won't be const.
66
4
            const auto* null_map = col_null_maps[col]; // null map for this col
67
4
            node.__set_is_null((null_map && (*null_map)[filter[row]])
68
4
                                       ? true
69
4
                                       : node.is_null); // if not, dont change(default false)
70
4
            if (!node.is_null) {
71
4
                node.__set_value(col_strs[col][row]);
72
4
            }
73
4
            cur_row_values.push_back(node);
74
4
        }
75
4
        if (!_deduper.contains(cur_row_values)) {
76
2
            _deduper.insert(cur_row_values);
77
2
            _partitions_need_create.emplace_back(cur_row_values);
78
2
        }
79
4
    }
80
81
    // to avoid too large mem use
82
2
    if (_batching_block->rows() > _batch_size) {
83
0
        _deal_batched = true;
84
0
    }
85
2
    _batching_rows = _batching_block->rows();
86
2
    VLOG_NOTICE << "pushed some batching lines, now numbers = " << _batching_rows;
87
88
2
    return Status::OK();
89
2
}
90
91
2
void VRowDistribution::clear_batching_stats() {
92
2
    _partitions_need_create.clear();
93
2
    _batching_rows = 0;
94
2
    _batching_bytes = 0;
95
2
}
96
97
2
Status VRowDistribution::automatic_create_partition() {
98
2
    MonotonicStopWatch timer;
99
2
    if (_state->enable_profile() && _state->profile_level() >= 2) {
100
0
        timer.start();
101
0
    }
102
103
2
    SCOPED_TIMER(_add_partition_request_timer);
104
2
    TCreatePartitionRequest request;
105
2
    TCreatePartitionResult result;
106
2
    bool injected = false;
107
2
    std::string be_endpoint = BackendOptions::get_be_endpoint();
108
2
    request.__set_txn_id(_txn_id);
109
2
    request.__set_db_id(_vpartition->db_id());
110
2
    request.__set_table_id(_vpartition->table_id());
111
2
    request.__set_partitionValues(_partitions_need_create);
112
2
    request.__set_be_endpoint(be_endpoint);
113
2
    request.__set_write_single_replica(_write_single_replica);
114
2
    request.__set_load_to_single_tablet(_tablet_finder->is_find_tablet_every_sink());
115
2
    request.__set_enable_adaptive_random_bucket(_tablet_finder->is_adaptive_random_bucket());
116
2
    if (_state && _state->get_query_ctx()) {
117
        // Pass query_id to FE so it can determine if this is a multi-instance load by checking Coordinator
118
2
        request.__set_query_id(_state->get_query_ctx()->query_id());
119
2
    }
120
121
2
    DBUG_EXECUTE_IF("VRowDistribution.automatic_create_partition.inject_result", {
122
2
        DBUG_RUN_CALLBACK(&request, &result);
123
2
        injected = true;
124
2
    });
125
126
2
    VLOG_NOTICE << "automatic partition rpc begin request " << request;
127
2
    if (!injected) {
128
0
        std::shared_ptr<TNetworkAddress> master_addr;
129
0
        if (_vpartition->get_master_address() == nullptr) {
130
0
            auto* cluster_info = ExecEnv::GetInstance()->cluster_info();
131
0
            if (cluster_info == nullptr) {
132
0
                return Status::InternalError("cluster_info is null");
133
0
            }
134
0
            master_addr = std::make_shared<TNetworkAddress>(cluster_info->master_fe_addr);
135
0
        } else {
136
0
            master_addr = _vpartition->get_master_address();
137
0
        }
138
0
        int time_out = _state->execution_timeout() * 1000;
139
0
        RETURN_IF_ERROR(ThriftRpcHelper::rpc<FrontendServiceClient>(
140
0
                master_addr->hostname, master_addr->port,
141
0
                [&request, &result](FrontendServiceConnection& client) {
142
0
                    client->createPartition(result, request);
143
0
                },
144
0
                time_out));
145
0
    }
146
147
2
    Status status(Status::create(result.status));
148
2
    VLOG_NOTICE << "automatic partition rpc end response " << result;
149
2
    if (result.status.status_code == TStatusCode::OK) {
150
        // Add new partitions before incremental open because adaptive random bucket builds
151
        // sender/receiver routing params from _vpartition.
152
2
        RETURN_IF_ERROR(_vpartition->add_partitions(result.partitions));
153
2
        for (const auto& part : result.partitions) {
154
2
            _new_partition_ids.insert(part.id);
155
2
            VLOG_TRACE << "record new id: " << part.id;
156
2
        }
157
2
        RETURN_IF_ERROR(_create_partition_callback(_caller, &result));
158
2
    }
159
160
    // Record this request's elapsed time
161
2
    if (_state->enable_profile() && _state->profile_level() >= 2) {
162
0
        int64_t elapsed_ns = timer.elapsed_time();
163
0
        _add_partition_request_times.push_back(elapsed_ns);
164
0
    }
165
2
    return status;
166
2
}
167
168
// for reuse the same create callback of create-partition
169
1
static TCreatePartitionResult cast_as_create_result(const TReplacePartitionResult& arg) {
170
1
    TCreatePartitionResult result;
171
1
    result.status = arg.status;
172
1
    result.nodes = arg.nodes;
173
1
    result.partitions = arg.partitions;
174
1
    result.tablets = arg.tablets;
175
1
    result.slave_tablets = arg.slave_tablets;
176
1
    return result;
177
1
}
178
179
// use _partitions and replace them
180
2
Status VRowDistribution::_replace_overwriting_partition() {
181
2
    SCOPED_TIMER(_add_partition_request_timer); // also for replace_partition
182
2
    TReplacePartitionRequest request;
183
2
    TReplacePartitionResult result;
184
2
    bool injected = false;
185
2
    request.__set_overwrite_group_id(_vpartition->get_overwrite_group_id());
186
2
    request.__set_db_id(_vpartition->db_id());
187
2
    request.__set_table_id(_vpartition->table_id());
188
2
    request.__set_write_single_replica(_write_single_replica);
189
190
    // only request for partitions not recorded for replacement
191
2
    std::set<int64_t> id_deduper;
192
3
    for (const auto* part : _partitions) {
193
3
        if (part != nullptr) {
194
3
            if (_new_partition_ids.contains(part->id)) {
195
                // this is a new partition. dont replace again.
196
1
                VLOG_TRACE << "skip new partition: " << part->id;
197
2
            } else {
198
                // request for replacement
199
2
                id_deduper.insert(part->id);
200
2
            }
201
3
        } else if (_missing_map.empty()) {
202
            // no origin partition. and not allow to create.
203
0
            return Status::InvalidArgument(
204
0
                    "Cannot found origin partitions in auto detect overwriting, stop "
205
0
                    "processing");
206
0
        } // else: part is null and _missing_map is not empty. dealed outside using auto-partition way. nothing to do here.
207
3
    }
208
2
    if (id_deduper.empty()) {
209
1
        return Status::OK(); // no need to request
210
1
    }
211
    // de-duplicate. there's no check in FE
212
1
    std::vector<int64_t> request_part_ids(id_deduper.begin(), id_deduper.end());
213
214
1
    request.__set_partition_ids(request_part_ids);
215
216
1
    std::string be_endpoint = BackendOptions::get_be_endpoint();
217
1
    request.__set_be_endpoint(be_endpoint);
218
1
    request.__set_load_to_single_tablet(_tablet_finder->is_find_tablet_every_sink());
219
1
    request.__set_enable_adaptive_random_bucket(_tablet_finder->is_adaptive_random_bucket());
220
1
    if (_state && _state->get_query_ctx()) {
221
        // Pass query_id to FE so it can determine if this is a multi-instance load by checking Coordinator
222
1
        request.__set_query_id(_state->get_query_ctx()->query_id());
223
1
    }
224
225
1
    DBUG_EXECUTE_IF("VRowDistribution.replace_overwriting_partition.inject_result", {
226
1
        DBUG_RUN_CALLBACK(&request, &result);
227
1
        injected = true;
228
1
    });
229
230
1
    VLOG_NOTICE << "auto detect replace partition request: " << request;
231
1
    if (!injected) {
232
0
        std::shared_ptr<TNetworkAddress> master_addr;
233
0
        if (_vpartition->get_master_address() == nullptr) {
234
0
            auto* cluster_info = ExecEnv::GetInstance()->cluster_info();
235
0
            if (cluster_info == nullptr) {
236
0
                return Status::InternalError("cluster_info is null");
237
0
            }
238
0
            master_addr = std::make_shared<TNetworkAddress>(cluster_info->master_fe_addr);
239
0
        } else {
240
0
            master_addr = _vpartition->get_master_address();
241
0
        }
242
0
        int time_out = _state->execution_timeout() * 1000;
243
0
        RETURN_IF_ERROR(ThriftRpcHelper::rpc<FrontendServiceClient>(
244
0
                master_addr->hostname, master_addr->port,
245
0
                [&request, &result](FrontendServiceConnection& client) {
246
0
                    client->replacePartition(result, request);
247
0
                },
248
0
                time_out));
249
0
    }
250
251
1
    Status status(Status::create(result.status));
252
1
    VLOG_NOTICE << "auto detect replace partition result: " << result;
253
1
    if (result.status.status_code == TStatusCode::OK) {
254
        // record new partitions
255
2
        for (const auto& part : result.partitions) {
256
2
            _new_partition_ids.insert(part.id);
257
2
            VLOG_TRACE << "record new id: " << part.id;
258
2
        }
259
        // replace data in _partitions
260
        // Adaptive random bucket builds sender/receiver routing params from _vpartition during
261
        // incremental open, so the replacement must be visible first.
262
1
        RETURN_IF_ERROR(_vpartition->replace_partitions(request_part_ids, result.partitions));
263
        // Reuse the function as the args' structure are same. It adds nodes/locations.
264
1
        auto result_as_create = cast_as_create_result(result);
265
1
        RETURN_IF_ERROR(_create_partition_callback(_caller, &result_as_create));
266
1
    }
267
268
1
    return status;
269
1
}
270
271
void VRowDistribution::_get_tablet_ids(Block* block, int32_t index_idx,
272
9
                                       std::vector<int64_t>& tablet_ids) {
273
9
    tablet_ids.resize(block->rows());
274
26
    for (int row_idx = 0; row_idx < block->rows(); row_idx++) {
275
17
        if (_skip[row_idx]) {
276
6
            continue;
277
6
        }
278
11
        auto& partition = _partitions[row_idx];
279
11
        auto& tablet_index = _tablet_indexes[row_idx];
280
11
        auto& index = partition->indexes[index_idx];
281
282
11
        auto tablet_id = index.tablets[tablet_index];
283
11
        tablet_ids[row_idx] = tablet_id;
284
11
    }
285
9
}
286
287
7
void VRowDistribution::_filter_block_by_skip(Block* block, RowPartTabletIds& row_part_tablet_id) {
288
7
    auto& row_ids = row_part_tablet_id.row_ids;
289
7
    auto& partition_ids = row_part_tablet_id.partition_ids;
290
7
    auto& tablet_ids = row_part_tablet_id.tablet_ids;
291
292
7
    auto rows = block->rows();
293
    // row count of a block should not exceed UINT32_MAX
294
7
    auto rows_uint32 = cast_set<uint32_t>(rows);
295
19
    for (uint32_t i = 0; i < rows_uint32; i++) {
296
12
        if (!_skip[i]) {
297
6
            row_ids.emplace_back(i);
298
6
            partition_ids.emplace_back(_partitions[i]->id);
299
6
            if (!_tablet_finder->is_adaptive_random_bucket()) {
300
6
                tablet_ids.emplace_back(_tablet_ids[i]);
301
6
            }
302
6
        }
303
12
    }
304
7
}
305
306
Status VRowDistribution::_filter_block_by_skip_and_where_clause(
307
2
        Block* block, const VExprContextSPtr& where_clause, RowPartTabletIds& row_part_tablet_id) {
308
    // TODO
309
    //SCOPED_RAW_TIMER(&_stat.where_clause_ns);
310
2
    ColumnPtr filter_column;
311
2
    RETURN_IF_ERROR(where_clause->execute(block, filter_column));
312
313
2
    auto& row_ids = row_part_tablet_id.row_ids;
314
2
    auto& partition_ids = row_part_tablet_id.partition_ids;
315
2
    auto& tablet_ids = row_part_tablet_id.tablet_ids;
316
2
    if (const auto* nullable_column = check_and_get_column<ColumnNullable>(*filter_column)) {
317
0
        auto rows = block->rows();
318
        // row count of a block should not exceed UINT32_MAX
319
0
        auto rows_uint32 = cast_set<uint32_t>(rows);
320
0
        for (uint32_t i = 0; i < rows_uint32; i++) {
321
0
            if (nullable_column->get_bool_inline(i) && !_skip[i]) {
322
0
                row_ids.emplace_back(i);
323
0
                partition_ids.emplace_back(_partitions[i]->id);
324
0
                if (!_tablet_finder->is_adaptive_random_bucket()) {
325
0
                    tablet_ids.emplace_back(_tablet_ids[i]);
326
0
                }
327
0
            }
328
0
        }
329
2
    } else if (const auto* const_column = check_and_get_column<ColumnConst>(*filter_column)) {
330
1
        bool ret = const_column->get_bool(0);
331
1
        if (!ret) {
332
1
            return Status::OK();
333
1
        }
334
        // should we optimize?
335
0
        _filter_block_by_skip(block, row_part_tablet_id);
336
1
    } else {
337
1
        const auto& filter = assert_cast<const ColumnUInt8&>(*filter_column).get_data();
338
1
        auto rows = block->rows();
339
        // row count of a block should not exceed UINT32_MAX
340
1
        auto rows_uint32 = cast_set<uint32_t>(rows);
341
4
        for (uint32_t i = 0; i < rows_uint32; i++) {
342
3
            if (filter[i] != 0 && !_skip[i]) {
343
2
                row_ids.emplace_back(i);
344
2
                partition_ids.emplace_back(_partitions[i]->id);
345
2
                if (!_tablet_finder->is_adaptive_random_bucket()) {
346
2
                    tablet_ids.emplace_back(_tablet_ids[i]);
347
2
                }
348
2
            }
349
3
        }
350
1
    }
351
352
1
    return Status::OK();
353
2
}
354
355
Status VRowDistribution::_filter_block(Block* block,
356
9
                                       std::vector<RowPartTabletIds>& row_part_tablet_ids) {
357
18
    for (int i = 0; i < _schema->indexes().size(); i++) {
358
9
        if (!_tablet_finder->is_adaptive_random_bucket()) {
359
9
            _get_tablet_ids(block, i, _tablet_ids);
360
9
        }
361
9
        auto& where_clause = _schema->indexes()[i]->where_clause;
362
9
        if (where_clause != nullptr) {
363
2
            RETURN_IF_ERROR(_filter_block_by_skip_and_where_clause(block, where_clause,
364
2
                                                                   row_part_tablet_ids[i]));
365
7
        } else {
366
7
            _filter_block_by_skip(block, row_part_tablet_ids[i]);
367
7
        }
368
9
    }
369
9
    return Status::OK();
370
9
}
371
372
Status VRowDistribution::_generate_rows_distribution_for_non_auto_partition(
373
5
        Block* block, bool has_filtered_rows, std::vector<RowPartTabletIds>& row_part_tablet_ids) {
374
5
    int num_rows = cast_set<int>(block->rows());
375
376
5
    RETURN_IF_ERROR(_tablet_finder->find_tablets(_state, block, num_rows, _partitions,
377
5
                                                 _tablet_indexes, _skip));
378
5
    if (has_filtered_rows) {
379
0
        for (int i = 0; i < num_rows; i++) {
380
0
            _skip[i] = _skip[i] || _block_convertor->filter_map()[i];
381
0
        }
382
0
    }
383
5
    RETURN_IF_ERROR(_filter_block(block, row_part_tablet_ids));
384
5
    return Status::OK();
385
5
}
386
387
Status VRowDistribution::_deal_missing_map(const Block& input_block, Block* block,
388
                                           const std::vector<uint16_t>& partition_cols_idx,
389
2
                                           int64_t& rows_stat_val) {
390
    // for missing partition keys, calc the missing partition and save in _partitions_need_create
391
2
    auto [part_ctxs, part_exprs] = _get_partition_function();
392
2
    int part_col_num = cast_set<int>(part_exprs.size());
393
    // the two vectors are in column-first-order
394
2
    std::vector<std::vector<std::string>> col_strs;
395
2
    std::vector<const NullMap*> col_null_maps;
396
2
    col_strs.resize(part_col_num);
397
2
    col_null_maps.reserve(part_col_num);
398
399
2
    auto format_options = DataTypeSerDe::get_default_format_options();
400
2
    format_options.timezone = &_state->timezone_obj();
401
402
4
    for (int i = 0; i < part_col_num; ++i) {
403
2
        auto return_type = part_exprs[i]->data_type();
404
        // expose the data column. the return type would be nullable
405
2
        const auto& [range_left_col, col_const] =
406
2
                unpack_if_const(block->get_by_position(partition_cols_idx[i]).column);
407
2
        if (range_left_col->is_nullable()) {
408
0
            col_null_maps.push_back(&(
409
0
                    assert_cast<const ColumnNullable*>(range_left_col.get())->get_null_map_data()));
410
2
        } else {
411
2
            col_null_maps.push_back(nullptr);
412
2
        }
413
4
        for (auto row : _missing_map) {
414
4
            col_strs[i].push_back(return_type->to_string(
415
4
                    *range_left_col, index_check_const(row, col_const), format_options));
416
4
        }
417
2
    }
418
419
    // calc the end value and save them. in the end of sending, we will create partitions for them and deal them.
420
    // NOTE: must save old batching stats before calling _save_missing_values(),
421
    // because _save_missing_values() will update _batching_rows internally.
422
2
    size_t old_bt_rows = _batching_rows;
423
2
    size_t old_bt_bytes = _batching_bytes;
424
425
2
    RETURN_IF_ERROR(_save_missing_values(input_block, col_strs, part_col_num, block, _missing_map,
426
2
                                         col_null_maps));
427
428
2
    size_t new_bt_rows = _batching_block->rows();
429
2
    size_t new_bt_bytes = _batching_block->bytes();
430
2
    rows_stat_val -= new_bt_rows - old_bt_rows;
431
2
    _state->update_num_rows_load_total(old_bt_rows - new_bt_rows);
432
2
    _state->update_num_bytes_load_total(old_bt_bytes - new_bt_bytes);
433
434
2
    return Status::OK();
435
2
}
436
437
Status VRowDistribution::_generate_rows_distribution_for_auto_partition(
438
        const Block& input_block, Block* block, const std::vector<uint16_t>& partition_cols_idx,
439
        bool has_filtered_rows, std::vector<RowPartTabletIds>& row_part_tablet_ids,
440
2
        int64_t& rows_stat_val) {
441
2
    int num_rows = cast_set<int>(block->rows());
442
2
    std::vector<uint16_t> partition_keys = _vpartition->get_partition_keys();
443
444
2
    auto& partition_col = block->get_by_position(partition_keys[0]);
445
2
    _missing_map.clear();
446
2
    _missing_map.reserve(partition_col.column->size());
447
448
2
    RETURN_IF_ERROR(_tablet_finder->find_tablets(_state, block, num_rows, _partitions,
449
2
                                                 _tablet_indexes, _skip, &_missing_map));
450
451
    // the missing vals for auto partition are also skipped.
452
2
    if (has_filtered_rows) {
453
0
        for (int i = 0; i < num_rows; i++) {
454
0
            _skip[i] = _skip[i] || _block_convertor->filter_map()[i];
455
0
        }
456
0
    }
457
2
    RETURN_IF_ERROR(_filter_block(block, row_part_tablet_ids));
458
459
2
    if (!_missing_map.empty()) {
460
2
        RETURN_IF_ERROR(_deal_missing_map(input_block, block, partition_cols_idx,
461
2
                                          rows_stat_val)); // send input block to save
462
2
    }
463
2
    return Status::OK();
464
2
}
465
466
Status VRowDistribution::_generate_rows_distribution_for_auto_overwrite(
467
        const Block& input_block, Block* block, const std::vector<uint16_t>& partition_cols_idx,
468
        bool has_filtered_rows, std::vector<RowPartTabletIds>& row_part_tablet_ids,
469
2
        int64_t& rows_stat_val) {
470
2
    int num_rows = cast_set<int>(block->rows());
471
472
    // for non-auto-partition situation, goes into two 'else' branch. just find the origin partitions, replace them by rpc,
473
    //  and find the new partitions to use.
474
    // for auto-partition's, find and save origins in _partitions and replace them. at meanwhile save the missing values for auto
475
    //  partition. then we find partition again to get replaced partitions in _partitions. this time _missing_map is ignored cuz
476
    //  we already saved missing values.
477
2
    if (_vpartition->is_auto_partition() &&
478
2
        _state->query_options().enable_auto_create_when_overwrite) {
479
        // allow auto create partition for missing rows.
480
0
        std::vector<uint16_t> partition_keys = _vpartition->get_partition_keys();
481
0
        auto partition_col = block->get_by_position(partition_keys[0]);
482
0
        _missing_map.clear();
483
0
        _missing_map.reserve(partition_col.column->size());
484
485
0
        RETURN_IF_ERROR(_tablet_finder->find_tablets(_state, block, num_rows, _partitions,
486
0
                                                     _tablet_indexes, _skip, &_missing_map));
487
488
        // allow and really need to create during auto-detect-overwriting.
489
0
        if (!_missing_map.empty()) {
490
0
            RETURN_IF_ERROR(
491
0
                    _deal_missing_map(input_block, block, partition_cols_idx, rows_stat_val));
492
0
        }
493
2
    } else {
494
2
        RETURN_IF_ERROR(_tablet_finder->find_tablets(_state, block, num_rows, _partitions,
495
2
                                                     _tablet_indexes, _skip));
496
2
    }
497
2
    RETURN_IF_ERROR(_replace_overwriting_partition());
498
499
    // regenerate locations for new partitions & tablets
500
2
    _reset_find_tablets(num_rows);
501
2
    if (_vpartition->is_auto_partition() &&
502
2
        _state->query_options().enable_auto_create_when_overwrite) {
503
        // here _missing_map is just a placeholder
504
0
        RETURN_IF_ERROR(_tablet_finder->find_tablets(_state, block, num_rows, _partitions,
505
0
                                                     _tablet_indexes, _skip, &_missing_map));
506
0
        if (VLOG_TRACE_IS_ON) {
507
0
            std::string tmp;
508
0
            for (auto v : _missing_map) {
509
0
                tmp += std::to_string(v).append(", ");
510
0
            }
511
0
            VLOG_TRACE << "Trace missing map of " << this << ':' << tmp;
512
0
        }
513
2
    } else {
514
2
        RETURN_IF_ERROR(_tablet_finder->find_tablets(_state, block, num_rows, _partitions,
515
2
                                                     _tablet_indexes, _skip));
516
2
    }
517
2
    if (has_filtered_rows) {
518
0
        for (int i = 0; i < num_rows; i++) {
519
0
            _skip[i] = _skip[i] || _block_convertor->filter_map()[i];
520
0
        }
521
0
    }
522
2
    RETURN_IF_ERROR(_filter_block(block, row_part_tablet_ids));
523
2
    return Status::OK();
524
2
}
525
526
void VRowDistribution::_reset_row_part_tablet_ids(
527
9
        std::vector<RowPartTabletIds>& row_part_tablet_ids, int64_t rows) {
528
9
    row_part_tablet_ids.resize(_schema->indexes().size());
529
9
    for (auto& row_part_tablet_id : row_part_tablet_ids) {
530
9
        auto& row_ids = row_part_tablet_id.row_ids;
531
9
        auto& partition_ids = row_part_tablet_id.partition_ids;
532
9
        auto& tablet_ids = row_part_tablet_id.tablet_ids;
533
534
9
        row_ids.clear();
535
9
        partition_ids.clear();
536
9
        tablet_ids.clear();
537
        // This is important for performance.
538
9
        row_ids.reserve(rows);
539
9
        partition_ids.reserve(rows);
540
9
        if (!_tablet_finder->is_adaptive_random_bucket()) {
541
9
            tablet_ids.reserve(rows);
542
9
        }
543
9
    }
544
9
}
545
546
Status VRowDistribution::generate_rows_distribution(
547
        Block& input_block, std::shared_ptr<Block>& block,
548
9
        std::vector<RowPartTabletIds>& row_part_tablet_ids, int64_t& rows_stat_val) {
549
9
    auto input_rows = input_block.rows();
550
9
    _reset_row_part_tablet_ids(row_part_tablet_ids, input_rows);
551
552
    // we store the batching block with value of `input_block`. so just do all of these again.
553
9
    bool has_filtered_rows = false;
554
9
    RETURN_IF_ERROR(_block_convertor->validate_and_convert_block(
555
9
            _state, &input_block, block, *_vec_output_expr_ctxs, input_rows, has_filtered_rows));
556
557
    // batching block rows which need new partitions. deal together at finish.
558
9
    if (!_batching_block) [[unlikely]] {
559
8
        std::unique_ptr<Block> tmp_block = input_block.create_same_struct_block(0);
560
8
        _batching_block = MutableBlock::create_unique(std::move(*tmp_block));
561
8
    }
562
563
9
    auto num_rows = block->rows();
564
9
    _reset_find_tablets(num_rows);
565
566
    // if there's projection of partition calc, we need to calc it first.
567
9
    auto [part_ctxs, part_funcs] = _get_partition_function();
568
9
    std::vector<uint16_t> partition_cols_idx;
569
9
    if (_vpartition->is_projection_partition()) {
570
        // calc the start value of missing partition ranges.
571
2
        auto func_size = part_funcs.size();
572
4
        for (int i = 0; i < func_size; ++i) {
573
2
            int result_idx = -1;
574
            // we just calc left range here. leave right to FE to avoid dup calc.
575
2
            RETURN_IF_ERROR(part_funcs[i]->execute(part_ctxs[i].get(), block.get(), &result_idx));
576
577
2
            VLOG_DEBUG << "Partition-calculated block:\n" << block->dump_data(0, 1);
578
2
            DCHECK(result_idx != -1);
579
580
2
            partition_cols_idx.push_back(cast_set<uint16_t>(result_idx));
581
2
        }
582
583
        // change the column to compare to transformed.
584
2
        _vpartition->set_transformed_slots(partition_cols_idx);
585
2
    }
586
587
9
    Status st = Status::OK();
588
9
    if (_vpartition->is_auto_detect_overwrite() && !_deal_batched) {
589
        // when overwrite, no auto create partition allowed.
590
2
        st = _generate_rows_distribution_for_auto_overwrite(input_block, block.get(),
591
2
                                                            partition_cols_idx, has_filtered_rows,
592
2
                                                            row_part_tablet_ids, rows_stat_val);
593
7
    } else if (_vpartition->is_auto_partition() && !_deal_batched) {
594
2
        st = _generate_rows_distribution_for_auto_partition(input_block, block.get(),
595
2
                                                            partition_cols_idx, has_filtered_rows,
596
2
                                                            row_part_tablet_ids, rows_stat_val);
597
5
    } else { // not auto partition
598
5
        st = _generate_rows_distribution_for_non_auto_partition(block.get(), has_filtered_rows,
599
5
                                                                row_part_tablet_ids);
600
5
    }
601
602
9
    return st;
603
9
}
604
605
// reuse vars for find_tablets
606
11
void VRowDistribution::_reset_find_tablets(int64_t rows) {
607
11
    _tablet_finder->filter_bitmap().Reset(rows);
608
11
    _partitions.assign(rows, nullptr);
609
11
    _skip.assign(rows, false);
610
11
    _tablet_indexes.assign(rows, 0);
611
11
}
612
613
} // namespace doris