Coverage Report

Created: 2026-08-06 13:04

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