Coverage Report

Created: 2026-03-25 18:05

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/storage/tablet_info.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 "storage/tablet_info.h"
19
20
#include <butil/logging.h>
21
#include <gen_cpp/Descriptors_types.h>
22
#include <gen_cpp/Exprs_types.h>
23
#include <gen_cpp/Partitions_types.h>
24
#include <gen_cpp/Types_types.h>
25
#include <gen_cpp/descriptors.pb.h>
26
#include <gen_cpp/olap_file.pb.h>
27
#include <glog/logging.h>
28
29
#include <algorithm>
30
#include <cstddef>
31
#include <cstdint>
32
#include <memory>
33
#include <ostream>
34
#include <string>
35
#include <tuple>
36
37
#include "common/exception.h"
38
#include "common/logging.h"
39
#include "common/status.h"
40
#include "core/column/column.h"
41
#include "core/data_type/data_type.h"
42
#include "core/data_type/data_type_factory.hpp"
43
#include "core/data_type/define_primitive_type.h"
44
#include "core/data_type/primitive_type.h"
45
#include "core/value/large_int_value.h"
46
#include "runtime/descriptors.h"
47
#include "runtime/memory/mem_tracker.h"
48
#include "storage/tablet/tablet_schema.h"
49
#include "util/raw_value.h"
50
#include "util/string_parser.hpp"
51
#include "util/string_util.h"
52
// NOLINTNEXTLINE(unused-includes)
53
#include "core/value/vdatetime_value.h"
54
#include "exprs/function/cast/cast_to_date_or_datetime_impl.hpp"
55
#include "exprs/function/cast/cast_to_datetimev2_impl.hpp"
56
#include "exprs/function/cast/cast_to_datev2_impl.hpp"
57
#include "exprs/function/cast/cast_to_timestamptz.h"
58
#include "exprs/vexpr_context.h" // IWYU pragma: keep
59
#include "exprs/vliteral.h"
60
61
namespace doris {
62
#include "common/compile_check_begin.h"
63
64
32
void OlapTableIndexSchema::to_protobuf(POlapTableIndexSchema* pindex) const {
65
32
    pindex->set_id(index_id);
66
32
    pindex->set_schema_hash(schema_hash);
67
32
    for (auto* slot : slots) {
68
0
        pindex->add_columns(slot->col_name());
69
0
    }
70
32
    for (auto* column : columns) {
71
0
        column->to_schema_pb(pindex->add_columns_desc());
72
0
    }
73
32
    for (auto* index : indexes) {
74
0
        index->to_schema_pb(pindex->add_indexes_desc());
75
0
    }
76
32
}
77
78
bool VOlapTablePartKeyComparator::operator()(const BlockRowWithIndicator& lhs,
79
104
                                             const BlockRowWithIndicator& rhs) const {
80
104
    Block* l_block = std::get<0>(lhs);
81
104
    Block* r_block = std::get<0>(rhs);
82
104
    int32_t l_row = std::get<1>(lhs);
83
104
    int32_t r_row = std::get<1>(rhs);
84
104
    bool l_use_new = std::get<2>(lhs);
85
104
    bool r_use_new = std::get<2>(rhs);
86
87
104
    VLOG_TRACE << '\n' << l_block->dump_data() << '\n' << r_block->dump_data();
88
89
104
    if (l_row == -1) {
90
0
        return false;
91
104
    } else if (r_row == -1) {
92
0
        return true;
93
0
    }
94
95
104
    if (_param_locs.empty()) { // no transform, use origin column
96
79
        for (auto slot_loc : _slot_locs) {
97
79
            auto res = l_block->get_by_position(slot_loc).column->compare_at(
98
79
                    l_row, r_row, *r_block->get_by_position(slot_loc).column, -1);
99
79
            if (res != 0) {
100
75
                return res < 0;
101
75
            }
102
79
        }
103
79
    } else { // use transformed column to compare
104
25
        DCHECK(_slot_locs.size() == _param_locs.size())
105
0
                << _slot_locs.size() << ' ' << _param_locs.size();
106
107
25
        const std::vector<uint16_t>* l_index = l_use_new ? &_param_locs : &_slot_locs;
108
25
        const std::vector<uint16_t>* r_index = r_use_new ? &_param_locs : &_slot_locs;
109
110
25
        for (int i = 0; i < _slot_locs.size(); i++) {
111
25
            ColumnPtr l_col = l_block->get_by_position((*l_index)[i]).column;
112
25
            ColumnPtr r_col = r_block->get_by_position((*r_index)[i]).column;
113
114
25
            auto res = l_col->compare_at(l_row, r_row, *r_col, -1);
115
25
            if (res != 0) {
116
25
                return res < 0;
117
25
            }
118
25
        }
119
25
    }
120
121
    // equal, return false
122
4
    return false;
123
104
}
124
125
14
Status OlapTableSchemaParam::init(const POlapTableSchemaParam& pschema) {
126
14
    _db_id = pschema.db_id();
127
14
    _table_id = pschema.table_id();
128
14
    _version = pschema.version();
129
14
    if (pschema.has_unique_key_update_mode()) {
130
14
        _unique_key_update_mode = pschema.unique_key_update_mode();
131
14
        if (pschema.has_sequence_map_col_unique_id()) {
132
14
            _sequence_map_col_uid = pschema.sequence_map_col_unique_id();
133
14
        }
134
14
    } else {
135
        // for backward compatibility
136
0
        if (pschema.has_partial_update() && pschema.partial_update()) {
137
0
            _unique_key_update_mode = UniqueKeyUpdateModePB::UPDATE_FIXED_COLUMNS;
138
0
        } else {
139
0
            _unique_key_update_mode = UniqueKeyUpdateModePB::UPSERT;
140
0
        }
141
0
    }
142
14
    _is_strict_mode = pschema.is_strict_mode();
143
14
    if (_unique_key_update_mode == UniqueKeyUpdateModePB::UPDATE_FIXED_COLUMNS) {
144
0
        _auto_increment_column = pschema.auto_increment_column();
145
0
        if (!_auto_increment_column.empty() && pschema.auto_increment_column_unique_id() == -1) {
146
0
            return Status::InternalError(
147
0
                    "Auto increment column id is not set in FE. Maybe FE is an older version "
148
0
                    "different from BE.");
149
0
        }
150
0
        _auto_increment_column_unique_id = pschema.auto_increment_column_unique_id();
151
0
    }
152
14
    if (_unique_key_update_mode != UniqueKeyUpdateModePB::UPSERT) {
153
0
        if (pschema.has_partial_update_new_key_policy()) {
154
0
            _partial_update_new_row_policy = pschema.partial_update_new_key_policy();
155
0
        }
156
0
    }
157
14
    _timestamp_ms = pschema.timestamp_ms();
158
14
    if (pschema.has_nano_seconds()) {
159
14
        _nano_seconds = pschema.nano_seconds();
160
14
    }
161
14
    _timezone = pschema.timezone();
162
163
14
    for (const auto& col : pschema.partial_update_input_columns()) {
164
0
        _partial_update_input_columns.insert(col);
165
0
    }
166
14
    std::unordered_map<std::string, SlotDescriptor*> slots_map;
167
168
14
    _tuple_desc = _obj_pool.add(new TupleDescriptor(pschema.tuple_desc()));
169
170
84
    for (const auto& p_slot_desc : pschema.slot_descs()) {
171
84
        auto* slot_desc = _obj_pool.add(new SlotDescriptor(p_slot_desc));
172
84
        _tuple_desc->add_slot(slot_desc);
173
84
        std::string data_type;
174
84
        EnumToString(TPrimitiveType, to_thrift(slot_desc->col_type()), data_type);
175
84
        std::string is_null_str = slot_desc->is_nullable() ? "true" : "false";
176
84
        std::string data_type_str =
177
84
                std::to_string(int64_t(TabletColumn::get_field_type_by_string(data_type)));
178
84
        slots_map.emplace(to_lower(slot_desc->col_name()) + "+" + data_type_str + is_null_str,
179
84
                          slot_desc);
180
84
    }
181
182
28
    for (const auto& p_index : pschema.indexes()) {
183
28
        auto* index = _obj_pool.add(new OlapTableIndexSchema());
184
28
        index->index_id = p_index.id();
185
28
        index->schema_hash = p_index.schema_hash();
186
28
        for (const auto& pcolumn_desc : p_index.columns_desc()) {
187
0
            if (_unique_key_update_mode != UniqueKeyUpdateModePB::UPDATE_FIXED_COLUMNS ||
188
0
                _partial_update_input_columns.contains(pcolumn_desc.name())) {
189
0
                std::string is_null_str = pcolumn_desc.is_nullable() ? "true" : "false";
190
0
                std::string data_type_str = std::to_string(
191
0
                        int64_t(TabletColumn::get_field_type_by_string(pcolumn_desc.type())));
192
0
                auto it = slots_map.find(to_lower(pcolumn_desc.name()) + "+" + data_type_str +
193
0
                                         is_null_str);
194
0
                if (it == std::end(slots_map)) {
195
0
                    std::string keys {};
196
0
                    for (const auto& [key, _] : slots_map) {
197
0
                        keys += fmt::format("{},", key);
198
0
                    }
199
0
                    LOG_EVERY_SECOND(WARNING) << fmt::format(
200
0
                            "[OlapTableSchemaParam::init(const POlapTableSchemaParam& pschema)]: "
201
0
                            "unknown index column, column={}, type={}, data_type_str={}, "
202
0
                            "is_null_str={}, slots_map.keys()=[{}], {}\npschema={}",
203
0
                            pcolumn_desc.name(), pcolumn_desc.type(), data_type_str, is_null_str,
204
0
                            keys, debug_string(), pschema.ShortDebugString());
205
206
0
                    return Status::InternalError("unknown index column, column={}, type={}",
207
0
                                                 pcolumn_desc.name(), pcolumn_desc.type());
208
0
                }
209
0
                index->slots.emplace_back(it->second);
210
0
            }
211
0
            TabletColumn* tc = _obj_pool.add(new TabletColumn());
212
0
            tc->init_from_pb(pcolumn_desc);
213
0
            index->columns.emplace_back(tc);
214
0
        }
215
28
        for (const auto& pindex_desc : p_index.indexes_desc()) {
216
0
            TabletIndex* ti = _obj_pool.add(new TabletIndex());
217
0
            ti->init_from_pb(pindex_desc);
218
0
            index->indexes.emplace_back(ti);
219
0
        }
220
28
        _indexes.emplace_back(index);
221
28
    }
222
223
14
    std::sort(_indexes.begin(), _indexes.end(),
224
28
              [](const OlapTableIndexSchema* lhs, const OlapTableIndexSchema* rhs) {
225
28
                  return lhs->index_id < rhs->index_id;
226
28
              });
227
14
    return Status::OK();
228
14
}
229
230
25
Status OlapTableSchemaParam::init_unique_key_update_mode(const TOlapTableSchemaParam& tschema) {
231
25
    if (tschema.__isset.unique_key_update_mode) {
232
0
        switch (tschema.unique_key_update_mode) {
233
0
        case doris::TUniqueKeyUpdateMode::UPSERT: {
234
0
            _unique_key_update_mode = UniqueKeyUpdateModePB::UPSERT;
235
0
            break;
236
0
        }
237
0
        case doris::TUniqueKeyUpdateMode::UPDATE_FIXED_COLUMNS: {
238
0
            _unique_key_update_mode = UniqueKeyUpdateModePB::UPDATE_FIXED_COLUMNS;
239
0
            break;
240
0
        }
241
0
        case doris::TUniqueKeyUpdateMode::UPDATE_FLEXIBLE_COLUMNS: {
242
0
            _unique_key_update_mode = UniqueKeyUpdateModePB::UPDATE_FLEXIBLE_COLUMNS;
243
0
            break;
244
0
        }
245
0
        default: {
246
0
            return Status::InternalError(
247
0
                    "Unknown unique_key_update_mode: {}, should be one of "
248
0
                    "UPSERT/UPDATE_FIXED_COLUMNS/UPDATE_FLEXIBLE_COLUMNS",
249
0
                    tschema.unique_key_update_mode);
250
0
        }
251
0
        }
252
0
        if (tschema.__isset.sequence_map_col_unique_id) {
253
0
            _sequence_map_col_uid = tschema.sequence_map_col_unique_id;
254
0
        }
255
25
    } else {
256
        // for backward compatibility
257
25
        if (tschema.__isset.is_partial_update && tschema.is_partial_update) {
258
0
            _unique_key_update_mode = UniqueKeyUpdateModePB::UPDATE_FIXED_COLUMNS;
259
25
        } else {
260
25
            _unique_key_update_mode = UniqueKeyUpdateModePB::UPSERT;
261
25
        }
262
25
    }
263
25
    return Status::OK();
264
25
}
265
266
25
Status OlapTableSchemaParam::init(const TOlapTableSchemaParam& tschema) {
267
25
    _db_id = tschema.db_id;
268
25
    _table_id = tschema.table_id;
269
25
    _version = tschema.version;
270
25
    RETURN_IF_ERROR(init_unique_key_update_mode(tschema));
271
25
    if (tschema.__isset.is_strict_mode) {
272
25
        _is_strict_mode = tschema.is_strict_mode;
273
25
    }
274
25
    if (_unique_key_update_mode == UniqueKeyUpdateModePB::UPDATE_FIXED_COLUMNS) {
275
0
        _auto_increment_column = tschema.auto_increment_column;
276
0
        if (!_auto_increment_column.empty() && tschema.auto_increment_column_unique_id == -1) {
277
0
            return Status::InternalError(
278
0
                    "Auto increment column id is not set in FE. Maybe FE is an older version "
279
0
                    "different from BE.");
280
0
        }
281
0
        _auto_increment_column_unique_id = tschema.auto_increment_column_unique_id;
282
0
    }
283
284
25
    if (_unique_key_update_mode != UniqueKeyUpdateModePB::UPSERT) {
285
0
        if (tschema.__isset.partial_update_new_key_policy) {
286
0
            switch (tschema.partial_update_new_key_policy) {
287
0
            case doris::TPartialUpdateNewRowPolicy::APPEND: {
288
0
                _partial_update_new_row_policy = PartialUpdateNewRowPolicyPB::APPEND;
289
0
                break;
290
0
            }
291
0
            case doris::TPartialUpdateNewRowPolicy::ERROR: {
292
0
                _partial_update_new_row_policy = PartialUpdateNewRowPolicyPB::ERROR;
293
0
                break;
294
0
            }
295
0
            default: {
296
0
                return Status::InvalidArgument(
297
0
                        "Unknown partial_update_new_key_behavior: {}, should be one of "
298
0
                        "'APPEND' or 'ERROR'",
299
0
                        tschema.partial_update_new_key_policy);
300
0
            }
301
0
            }
302
0
        }
303
0
    }
304
305
25
    for (const auto& tcolumn : tschema.partial_update_input_columns) {
306
0
        _partial_update_input_columns.insert(tcolumn);
307
0
    }
308
25
    std::unordered_map<std::string, SlotDescriptor*> slots_map;
309
25
    _tuple_desc = _obj_pool.add(new TupleDescriptor(tschema.tuple_desc));
310
105
    for (const auto& t_slot_desc : tschema.slot_descs) {
311
105
        auto* slot_desc = _obj_pool.add(new SlotDescriptor(t_slot_desc));
312
105
        _tuple_desc->add_slot(slot_desc);
313
105
        std::string is_null_str = slot_desc->is_nullable() ? "true" : "false";
314
105
        std::string data_type_str = std::to_string(int64_t(slot_desc->col_type()));
315
105
        slots_map.emplace(to_lower(slot_desc->col_name()) + "+" + data_type_str + is_null_str,
316
105
                          slot_desc);
317
105
    }
318
319
41
    for (const auto& t_index : tschema.indexes) {
320
41
        std::unordered_map<std::string, int32_t> index_slots_map;
321
41
        auto* index = _obj_pool.add(new OlapTableIndexSchema());
322
41
        index->index_id = t_index.id;
323
41
        index->schema_hash = t_index.schema_hash;
324
41
        for (const auto& tcolumn_desc : t_index.columns_desc) {
325
0
            if (_unique_key_update_mode != UniqueKeyUpdateModePB::UPDATE_FIXED_COLUMNS ||
326
0
                _partial_update_input_columns.contains(tcolumn_desc.column_name)) {
327
0
                std::string is_null_str = tcolumn_desc.is_allow_null ? "true" : "false";
328
0
                std::string data_type_str =
329
0
                        std::to_string(int64_t(thrift_to_type(tcolumn_desc.column_type.type)));
330
0
                auto it = slots_map.find(to_lower(tcolumn_desc.column_name) + "+" + data_type_str +
331
0
                                         is_null_str);
332
0
                if (it == slots_map.end()) {
333
0
                    std::stringstream ss;
334
0
                    ss << tschema;
335
0
                    std::string keys {};
336
0
                    for (const auto& [key, _] : slots_map) {
337
0
                        keys += fmt::format("{},", key);
338
0
                    }
339
0
                    LOG_EVERY_SECOND(WARNING) << fmt::format(
340
0
                            "[OlapTableSchemaParam::init(const TOlapTableSchemaParam& tschema)]: "
341
0
                            "unknown index column, column={}, type={}, data_type_str={}, "
342
0
                            "is_null_str={}, slots_map.keys()=[{}], {}\ntschema={}",
343
0
                            tcolumn_desc.column_name, tcolumn_desc.column_type.type, data_type_str,
344
0
                            is_null_str, keys, debug_string(), ss.str());
345
0
                    return Status::InternalError("unknown index column, column={}, type={}",
346
0
                                                 tcolumn_desc.column_name,
347
0
                                                 tcolumn_desc.column_type.type);
348
0
                }
349
0
                index->slots.emplace_back(it->second);
350
0
            }
351
0
            index_slots_map.emplace(to_lower(tcolumn_desc.column_name), tcolumn_desc.col_unique_id);
352
0
            TabletColumn* tc = _obj_pool.add(new TabletColumn());
353
0
            tc->init_from_thrift(tcolumn_desc);
354
0
            index->columns.emplace_back(tc);
355
0
        }
356
41
        if (t_index.__isset.indexes_desc) {
357
0
            for (const auto& tindex_desc : t_index.indexes_desc) {
358
0
                std::vector<int32_t> column_unique_ids(tindex_desc.columns.size());
359
0
                for (size_t i = 0; i < tindex_desc.columns.size(); i++) {
360
0
                    auto it = index_slots_map.find(to_lower(tindex_desc.columns[i]));
361
0
                    if (it != index_slots_map.end()) {
362
0
                        column_unique_ids[i] = it->second;
363
0
                    }
364
0
                }
365
0
                TabletIndex* ti = _obj_pool.add(new TabletIndex());
366
0
                ti->init_from_thrift(tindex_desc, column_unique_ids);
367
0
                index->indexes.emplace_back(ti);
368
0
            }
369
0
        }
370
41
        if (t_index.__isset.where_clause) {
371
0
            RETURN_IF_ERROR(VExpr::create_expr_tree(t_index.where_clause, index->where_clause));
372
0
        }
373
41
        _indexes.emplace_back(index);
374
41
    }
375
376
25
    std::sort(_indexes.begin(), _indexes.end(),
377
32
              [](const OlapTableIndexSchema* lhs, const OlapTableIndexSchema* rhs) {
378
32
                  return lhs->index_id < rhs->index_id;
379
32
              });
380
25
    return Status::OK();
381
25
}
382
383
16
void OlapTableSchemaParam::to_protobuf(POlapTableSchemaParam* pschema) const {
384
16
    pschema->set_db_id(_db_id);
385
16
    pschema->set_table_id(_table_id);
386
16
    pschema->set_version(_version);
387
16
    pschema->set_unique_key_update_mode(_unique_key_update_mode);
388
16
    if (_unique_key_update_mode == UniqueKeyUpdateModePB::UPDATE_FIXED_COLUMNS) {
389
        // for backward compatibility
390
0
        pschema->set_partial_update(true);
391
0
    }
392
16
    pschema->set_partial_update_new_key_policy(_partial_update_new_row_policy);
393
16
    pschema->set_is_strict_mode(_is_strict_mode);
394
16
    pschema->set_auto_increment_column(_auto_increment_column);
395
16
    pschema->set_auto_increment_column_unique_id(_auto_increment_column_unique_id);
396
16
    pschema->set_timestamp_ms(_timestamp_ms);
397
16
    pschema->set_timezone(_timezone);
398
16
    pschema->set_nano_seconds(_nano_seconds);
399
16
    pschema->set_sequence_map_col_unique_id(_sequence_map_col_uid);
400
16
    for (auto col : _partial_update_input_columns) {
401
0
        *pschema->add_partial_update_input_columns() = col;
402
0
    }
403
16
    _tuple_desc->to_protobuf(pschema->mutable_tuple_desc());
404
96
    for (auto* slot : _tuple_desc->slots()) {
405
96
        slot->to_protobuf(pschema->add_slot_descs());
406
96
    }
407
32
    for (auto* index : _indexes) {
408
32
        index->to_protobuf(pschema->add_indexes());
409
32
    }
410
16
}
411
412
0
std::string OlapTableSchemaParam::debug_string() const {
413
0
    std::stringstream ss;
414
0
    ss << "tuple_desc=" << _tuple_desc->debug_string();
415
0
    return ss.str();
416
0
}
417
418
VOlapTablePartitionParam::VOlapTablePartitionParam(std::shared_ptr<OlapTableSchemaParam>& schema,
419
                                                   const TOlapTablePartitionParam& t_param)
420
9
        : _schema(schema),
421
9
          _t_param(t_param),
422
9
          _slots(_schema->tuple_desc()->slots()),
423
9
          _mem_tracker(std::make_unique<MemTracker>("OlapTablePartitionParam")),
424
9
          _part_type(t_param.partition_type) {
425
9
    if (t_param.__isset.enable_automatic_partition && t_param.enable_automatic_partition) {
426
2
        _is_auto_partition = true;
427
2
        auto size = t_param.partition_function_exprs.size();
428
2
        _part_func_ctx.resize(size);
429
2
        _partition_function.resize(size);
430
2
        DCHECK((t_param.partition_type == TPartitionType::RANGE_PARTITIONED && size == 1) ||
431
0
               (t_param.partition_type == TPartitionType::LIST_PARTITIONED && size >= 1))
432
0
                << "now support only 1 partition column for auto range partitions. "
433
0
                << t_param.partition_type << " " << size;
434
4
        for (int i = 0; i < size; ++i) {
435
2
            Status st =
436
2
                    VExpr::create_expr_tree(t_param.partition_function_exprs[i], _part_func_ctx[i]);
437
2
            if (!st.ok()) {
438
0
                throw Exception(Status::InternalError("Partition function expr is not valid"),
439
0
                                "Partition function expr is not valid");
440
0
            }
441
2
            _partition_function[i] = _part_func_ctx[i]->root();
442
2
        }
443
2
    }
444
445
9
    if (t_param.__isset.enable_auto_detect_overwrite && t_param.enable_auto_detect_overwrite) {
446
1
        _is_auto_detect_overwrite = true;
447
1
        DCHECK(t_param.__isset.overwrite_group_id);
448
1
        _overwrite_group_id = t_param.overwrite_group_id;
449
1
    }
450
451
9
    if (t_param.__isset.master_address) {
452
0
        _master_address = std::make_shared<TNetworkAddress>(t_param.master_address);
453
0
    }
454
455
9
    if (_is_auto_partition) {
456
        // the nullable mode depends on partition_exprs. not column slots. so use them.
457
2
        DCHECK(_partition_function.size() <= _slots.size())
458
0
                << _partition_function.size() << ", " << _slots.size();
459
460
        // suppose (k0, [k1], [k2]), so get [k1, 0], [k2, 1]
461
2
        std::map<std::string, int> partition_slots_map; // name to idx in part_exprs
462
4
        for (size_t i = 0; i < t_param.partition_columns.size(); i++) {
463
2
            partition_slots_map.emplace(t_param.partition_columns[i], i);
464
2
        }
465
466
        // here we rely on the same order and number of the _part_funcs and _slots in the prefix
467
        // _part_block contains all slots of table.
468
2
        for (auto* slot : _slots) {
469
            // try to replace with partition expr.
470
2
            if (auto it = partition_slots_map.find(slot->col_name());
471
2
                it != partition_slots_map.end()) { // it's a partition column slot
472
2
                auto& expr_type = _partition_function[it->second]->data_type();
473
2
                _partition_block.insert({expr_type->create_column(), expr_type, slot->col_name()});
474
2
            } else {
475
0
                _partition_block.insert({slot->get_empty_mutable_column(),
476
0
                                         slot->get_data_type_ptr(), slot->col_name()});
477
0
            }
478
2
        }
479
2
        VLOG_TRACE << _partition_block.dump_structure();
480
7
    } else {
481
        // we insert all. but not all will be used. it will controlled by _partition_slot_locs
482
7
        for (auto* slot : _slots) {
483
7
            _partition_block.insert({slot->get_empty_mutable_column(), slot->get_data_type_ptr(),
484
7
                                     slot->col_name()});
485
7
        }
486
7
    }
487
9
}
488
489
9
VOlapTablePartitionParam::~VOlapTablePartitionParam() {
490
9
    _mem_tracker->release(_mem_usage);
491
9
}
492
493
9
Status VOlapTablePartitionParam::init() {
494
9
    std::vector<std::string> slot_column_names;
495
9
    for (auto* slot_desc : _schema->tuple_desc()->slots()) {
496
9
        slot_column_names.emplace_back(slot_desc->col_name());
497
9
    }
498
499
9
    auto find_slot_locs = [&slot_column_names](const std::string& slot_name,
500
9
                                               std::vector<uint16_t>& locs,
501
17
                                               const std::string& column_type) {
502
17
        auto it = std::find(slot_column_names.begin(), slot_column_names.end(), slot_name);
503
17
        if (it == slot_column_names.end()) {
504
0
            return Status::InternalError("{} column not found, column ={}", column_type, slot_name);
505
0
        }
506
17
        locs.emplace_back(it - slot_column_names.begin());
507
17
        return Status::OK();
508
17
    };
509
510
    // here we find the partition columns. others maybe non-partition columns/special columns.
511
9
    if (_t_param.__isset.partition_columns) {
512
9
        for (auto& part_col : _t_param.partition_columns) {
513
9
            RETURN_IF_ERROR(find_slot_locs(part_col, _partition_slot_locs, "partition"));
514
9
        }
515
9
    }
516
517
9
    _partitions_map = std::make_unique<
518
9
            std::map<BlockRowWithIndicator, VOlapTablePartition*, VOlapTablePartKeyComparator>>(
519
9
            VOlapTablePartKeyComparator(_partition_slot_locs, _transformed_slot_locs));
520
9
    if (_t_param.__isset.distributed_columns) {
521
8
        for (auto& col : _t_param.distributed_columns) {
522
8
            RETURN_IF_ERROR(find_slot_locs(col, _distributed_slot_locs, "distributed"));
523
8
        }
524
8
    }
525
526
    // for both auto/non-auto partition table.
527
9
    _is_in_partition = _part_type == TPartitionType::type::LIST_PARTITIONED;
528
529
    // initial partitions. if meet dummy partitions only for open BE nodes, not generate key of them for finding
530
17
    for (const auto& t_part : _t_param.partitions) {
531
17
        VOlapTablePartition* part = nullptr;
532
17
        RETURN_IF_ERROR(generate_partition_from(t_part, part));
533
17
        _partitions.emplace_back(part);
534
535
17
        if (!_t_param.partitions_is_fake) {
536
17
            if (_is_in_partition) {
537
0
                for (auto& in_key : part->in_keys) {
538
0
                    _partitions_map->emplace(std::tuple {in_key.first, in_key.second, false}, part);
539
0
                }
540
17
            } else {
541
17
                _partitions_map->emplace(
542
17
                        std::tuple {part->end_key.first, part->end_key.second, false}, part);
543
17
            }
544
17
        }
545
17
    }
546
547
9
    _mem_usage = _partition_block.allocated_bytes();
548
9
    _mem_tracker->consume(_mem_usage);
549
9
    return Status::OK();
550
9
}
551
552
bool VOlapTablePartitionParam::_part_contains(VOlapTablePartition* part,
553
27
                                              BlockRowWithIndicator key) const {
554
27
    VOlapTablePartKeyComparator comparator(_partition_slot_locs, _transformed_slot_locs);
555
    // we have used upper_bound to find to ensure key < part.right and this part is closest(right - key is min)
556
    // now we only have to check (key >= part.left). the comparator(a,b) means a < b, so we use anti
557
27
    return part->start_key.second == -1 /* spj: start_key.second == -1 means only single partition*/
558
27
           || !comparator(key, std::tuple {part->start_key.first, part->start_key.second, false});
559
27
}
560
561
// insert value into _partition_block's column
562
// NOLINTBEGIN(readability-function-size)
563
38
static Status _create_partition_key(const TExprNode& t_expr, BlockRow* part_key, uint16_t pos) {
564
38
    auto column = std::move(*part_key->first->get_by_position(pos).column).mutate();
565
38
    switch (t_expr.node_type) {
566
0
    case TExprNodeType::DATE_LITERAL: {
567
0
        auto primitive_type =
568
0
                DataTypeFactory::instance().create_data_type(t_expr.type)->get_primitive_type();
569
0
        if (primitive_type == TYPE_DATEV2) {
570
0
            DateV2Value<DateV2ValueType> dt;
571
0
            CastParameters params;
572
0
            if (!CastToDateV2::from_string_strict_mode<DatelikeParseMode::STRICT>(
573
0
                        {t_expr.date_literal.value.c_str(), t_expr.date_literal.value.size()}, dt,
574
0
                        nullptr, params)) {
575
0
                std::stringstream ss;
576
0
                ss << "invalid date literal in partition column, date=" << t_expr.date_literal;
577
0
                return Status::InternalError(ss.str());
578
0
            }
579
0
            column->insert_data(reinterpret_cast<const char*>(&dt), 0);
580
0
        } else if (primitive_type == TYPE_DATETIMEV2) {
581
0
            DateV2Value<DateTimeV2ValueType> dt;
582
0
            const int32_t scale =
583
0
                    t_expr.type.types.empty() ? -1 : t_expr.type.types.front().scalar_type.scale;
584
0
            CastParameters params;
585
0
            if (!CastToDatetimeV2::from_string_strict_mode<DatelikeParseMode::STRICT>(
586
0
                        {t_expr.date_literal.value.c_str(), t_expr.date_literal.value.size()}, dt,
587
0
                        nullptr, scale, params)) {
588
0
                std::stringstream ss;
589
0
                ss << "invalid date literal in partition column, date=" << t_expr.date_literal;
590
0
                return Status::InternalError(ss.str());
591
0
            }
592
0
            column->insert_data(reinterpret_cast<const char*>(&dt), 0);
593
0
        } else if (primitive_type == TYPE_TIMESTAMPTZ) {
594
0
            TimestampTzValue res;
595
0
            CastParameters params {.status = Status::OK(), .is_strict = true};
596
0
            const int32_t scale =
597
0
                    t_expr.type.types.empty() ? -1 : t_expr.type.types.front().scalar_type.scale;
598
0
            if (!CastToTimestampTz::from_string(
599
0
                        {t_expr.date_literal.value.c_str(), t_expr.date_literal.value.size()}, res,
600
0
                        params, nullptr, scale)) [[unlikely]] {
601
0
                std::stringstream ss;
602
0
                ss << "invalid timestamptz literal in partition column, value="
603
0
                   << t_expr.date_literal;
604
0
                return Status::InternalError(ss.str());
605
0
            }
606
0
            column->insert_data(reinterpret_cast<const char*>(&res), 0);
607
0
        } else {
608
0
            VecDateTimeValue dt;
609
0
            CastParameters params;
610
0
            if (!CastToDateOrDatetime::from_string_strict_mode<DatelikeParseMode::STRICT,
611
0
                                                               DatelikeTargetType::DATE_TIME>(
612
0
                        {t_expr.date_literal.value.c_str(), t_expr.date_literal.value.size()}, dt,
613
0
                        nullptr, params)) {
614
0
                std::stringstream ss;
615
0
                ss << "invalid date literal in partition column, date=" << t_expr.date_literal;
616
0
                return Status::InternalError(ss.str());
617
0
            }
618
0
            if (primitive_type == TYPE_DATE) {
619
0
                dt.cast_to_date();
620
0
            }
621
0
            column->insert_data(reinterpret_cast<const char*>(&dt), 0);
622
0
        }
623
0
        break;
624
0
    }
625
38
    case TExprNodeType::INT_LITERAL: {
626
38
        switch (t_expr.type.types[0].scalar_type.type) {
627
0
        case TPrimitiveType::TINYINT: {
628
0
            auto value = cast_set<int8_t>(t_expr.int_literal.value);
629
0
            column->insert_data(reinterpret_cast<const char*>(&value), 0);
630
0
            break;
631
0
        }
632
0
        case TPrimitiveType::SMALLINT: {
633
0
            auto value = cast_set<int16_t>(t_expr.int_literal.value);
634
0
            column->insert_data(reinterpret_cast<const char*>(&value), 0);
635
0
            break;
636
0
        }
637
38
        case TPrimitiveType::INT: {
638
38
            auto value = cast_set<int32_t>(t_expr.int_literal.value);
639
38
            column->insert_data(reinterpret_cast<const char*>(&value), 0);
640
38
            break;
641
0
        }
642
0
        default:
643
0
            int64_t value = t_expr.int_literal.value;
644
0
            column->insert_data(reinterpret_cast<const char*>(&value), 0);
645
38
        }
646
38
        break;
647
38
    }
648
38
    case TExprNodeType::LARGE_INT_LITERAL: {
649
0
        StringParser::ParseResult parse_result = StringParser::PARSE_SUCCESS;
650
0
        auto value = StringParser::string_to_int<__int128>(t_expr.large_int_literal.value.c_str(),
651
0
                                                           t_expr.large_int_literal.value.size(),
652
0
                                                           &parse_result);
653
0
        if (parse_result != StringParser::PARSE_SUCCESS) {
654
0
            value = MAX_INT128;
655
0
        }
656
0
        column->insert_data(reinterpret_cast<const char*>(&value), 0);
657
0
        break;
658
38
    }
659
0
    case TExprNodeType::STRING_LITERAL: {
660
0
        size_t len = t_expr.string_literal.value.size();
661
0
        const char* str_val = t_expr.string_literal.value.c_str();
662
0
        column->insert_data(str_val, len);
663
0
        break;
664
38
    }
665
0
    case TExprNodeType::BOOL_LITERAL: {
666
0
        column->insert_data(reinterpret_cast<const char*>(&t_expr.bool_literal.value), 0);
667
0
        break;
668
38
    }
669
0
    case TExprNodeType::NULL_LITERAL: {
670
        // insert a null literal
671
0
        if (!column->is_nullable()) {
672
            // https://github.com/apache/doris/pull/39449 have forbid this cause. always add this check as protective measures
673
0
            return Status::InternalError("The column {} is not null, can't insert into NULL value.",
674
0
                                         part_key->first->get_by_position(pos).name);
675
0
        }
676
0
        column->insert_data(nullptr, 0);
677
0
        break;
678
0
    }
679
0
    default: {
680
0
        return Status::InternalError("unsupported partition column node type, type={}",
681
0
                                     t_expr.node_type);
682
0
    }
683
38
    }
684
38
    part_key->second = cast_set<int32_t>(column->size() - 1);
685
38
    return Status::OK();
686
38
}
687
// NOLINTEND(readability-function-size)
688
689
Status VOlapTablePartitionParam::_create_partition_keys(const std::vector<TExprNode>& t_exprs,
690
38
                                                        BlockRow* part_key) {
691
76
    for (int i = 0; i < t_exprs.size(); i++) {
692
38
        RETURN_IF_ERROR(_create_partition_key(t_exprs[i], part_key, _partition_slot_locs[i]));
693
38
    }
694
38
    return Status::OK();
695
38
}
696
697
Status VOlapTablePartitionParam::generate_partition_from(const TOlapTablePartition& t_part,
698
17
                                                         VOlapTablePartition*& part_result) {
699
17
    DCHECK(part_result == nullptr);
700
    // here we set the default value of partition bounds first! if it doesn't have some key, it will be -1.
701
17
    part_result = _obj_pool.add(new VOlapTablePartition(&_partition_block));
702
17
    part_result->id = t_part.id;
703
17
    part_result->is_mutable = t_part.is_mutable;
704
    // only load_to_single_tablet = true will set load_tablet_idx
705
17
    if (t_part.__isset.load_tablet_idx) {
706
1
        part_result->load_tablet_idx = t_part.load_tablet_idx;
707
1
    }
708
709
17
    if (_is_in_partition) {
710
0
        for (const auto& keys : t_part.in_keys) {
711
0
            RETURN_IF_ERROR(_create_partition_keys(
712
0
                    keys, &part_result->in_keys.emplace_back(&_partition_block, -1)));
713
0
        }
714
0
        if (t_part.__isset.is_default_partition && t_part.is_default_partition &&
715
0
            _default_partition == nullptr) {
716
0
            _default_partition = part_result;
717
0
        }
718
17
    } else { // range
719
17
        if (t_part.__isset.start_keys) {
720
17
            RETURN_IF_ERROR(_create_partition_keys(t_part.start_keys, &part_result->start_key));
721
17
        }
722
        // we generate the right bound but not insert into partition map
723
17
        if (t_part.__isset.end_keys) {
724
17
            RETURN_IF_ERROR(_create_partition_keys(t_part.end_keys, &part_result->end_key));
725
17
        }
726
17
    }
727
728
17
    part_result->num_buckets = t_part.num_buckets;
729
17
    auto num_indexes = _schema->indexes().size();
730
17
    if (t_part.indexes.size() != num_indexes) {
731
0
        return Status::InternalError(
732
0
                "number of partition's index is not equal with schema's"
733
0
                ", num_part_indexes={}, num_schema_indexes={}",
734
0
                t_part.indexes.size(), num_indexes);
735
0
    }
736
17
    part_result->indexes = t_part.indexes;
737
17
    std::sort(part_result->indexes.begin(), part_result->indexes.end(),
738
17
              [](const OlapTableIndexTablets& lhs, const OlapTableIndexTablets& rhs) {
739
0
                  return lhs.index_id < rhs.index_id;
740
0
              });
741
    // check index
742
34
    for (int j = 0; j < num_indexes; ++j) {
743
17
        if (part_result->indexes[j].index_id != _schema->indexes()[j]->index_id) {
744
0
            return Status::InternalError(
745
0
                    "partition's index is not equal with schema's"
746
0
                    ", part_index={}, schema_index={}",
747
0
                    part_result->indexes[j].index_id, _schema->indexes()[j]->index_id);
748
0
        }
749
17
    }
750
17
    if (t_part.__isset.total_replica_num) {
751
0
        part_result->total_replica_num = t_part.total_replica_num;
752
0
    }
753
17
    if (t_part.__isset.load_required_replica_num) {
754
0
        part_result->load_required_replica_num = t_part.load_required_replica_num;
755
0
    }
756
17
    if (t_part.__isset.tablet_version_gap_backends) {
757
0
        for (const auto& [tablet_id, backend_ids] : t_part.tablet_version_gap_backends) {
758
0
            auto& gap_set = part_result->tablet_version_gap_backends[tablet_id];
759
0
            for (auto backend_id : backend_ids) {
760
0
                gap_set.insert(backend_id);
761
0
            }
762
0
        }
763
0
    }
764
17
    return Status::OK();
765
17
}
766
767
Status VOlapTablePartitionParam::add_partitions(
768
2
        const std::vector<TOlapTablePartition>& partitions) {
769
2
    for (const auto& t_part : partitions) {
770
2
        auto* part = _obj_pool.add(new VOlapTablePartition(&_partition_block));
771
2
        part->id = t_part.id;
772
2
        part->is_mutable = t_part.is_mutable;
773
774
        // we dont pass right keys when it's MAX_VALUE. so there's possibility we only have start_key but not end_key
775
        // range partition
776
2
        if (t_part.__isset.start_keys) {
777
2
            RETURN_IF_ERROR(_create_partition_keys(t_part.start_keys, &part->start_key));
778
2
        }
779
2
        if (t_part.__isset.end_keys) {
780
2
            RETURN_IF_ERROR(_create_partition_keys(t_part.end_keys, &part->end_key));
781
2
        }
782
        // list partition - we only set 1 value in 1 partition for new created ones
783
2
        if (t_part.__isset.in_keys) {
784
0
            for (const auto& keys : t_part.in_keys) {
785
0
                RETURN_IF_ERROR(_create_partition_keys(
786
0
                        keys, &part->in_keys.emplace_back(&_partition_block, -1)));
787
0
            }
788
0
            if (t_part.__isset.is_default_partition && t_part.is_default_partition) {
789
0
                _default_partition = part;
790
0
            }
791
0
        }
792
793
2
        part->num_buckets = t_part.num_buckets;
794
2
        auto num_indexes = _schema->indexes().size();
795
2
        if (t_part.indexes.size() != num_indexes) {
796
0
            return Status::InternalError(
797
0
                    "number of partition's index is not equal with schema's"
798
0
                    ", num_part_indexes={}, num_schema_indexes={}",
799
0
                    t_part.indexes.size(), num_indexes);
800
0
        }
801
2
        part->indexes = t_part.indexes;
802
2
        std::sort(part->indexes.begin(), part->indexes.end(),
803
2
                  [](const OlapTableIndexTablets& lhs, const OlapTableIndexTablets& rhs) {
804
0
                      return lhs.index_id < rhs.index_id;
805
0
                  });
806
        // check index
807
4
        for (int j = 0; j < num_indexes; ++j) {
808
2
            if (part->indexes[j].index_id != _schema->indexes()[j]->index_id) {
809
0
                return Status::InternalError(
810
0
                        "partition's index is not equal with schema's"
811
0
                        ", part_index={}, schema_index={}",
812
0
                        part->indexes[j].index_id, _schema->indexes()[j]->index_id);
813
0
            }
814
2
        }
815
2
        _partitions.emplace_back(part);
816
        // after _creating_partiton_keys
817
2
        if (_is_in_partition) {
818
0
            for (auto& in_key : part->in_keys) {
819
0
                _partitions_map->emplace(std::tuple {in_key.first, in_key.second, false}, part);
820
0
            }
821
2
        } else {
822
2
            _partitions_map->emplace(std::tuple {part->end_key.first, part->end_key.second, false},
823
2
                                     part);
824
2
        }
825
2
    }
826
827
2
    return Status::OK();
828
2
}
829
830
Status VOlapTablePartitionParam::replace_partitions(
831
        std::vector<int64_t>& old_partition_ids,
832
1
        const std::vector<TOlapTablePartition>& new_partitions) {
833
    // remove old replaced partitions
834
1
    DCHECK(old_partition_ids.size() == new_partitions.size());
835
836
    // init and add new partitions. insert into _partitions
837
3
    for (int i = 0; i < new_partitions.size(); i++) {
838
2
        const auto& t_part = new_partitions[i];
839
        // pair old_partition_ids and new_partitions one by one. TODO: sort to opt performance
840
2
        VOlapTablePartition* old_part = nullptr;
841
2
        auto old_part_id = old_partition_ids[i];
842
2
        if (auto it = std::find_if(
843
2
                    _partitions.begin(), _partitions.end(),
844
3
                    [=](const VOlapTablePartition* lhs) { return lhs->id == old_part_id; });
845
2
            it != _partitions.end()) {
846
2
            old_part = *it;
847
2
        } else {
848
0
            return Status::InternalError("Cannot find old tablet {} in replacing", old_part_id);
849
0
        }
850
851
2
        auto* part = _obj_pool.add(new VOlapTablePartition(&_partition_block));
852
2
        part->id = t_part.id;
853
2
        part->is_mutable = t_part.is_mutable;
854
855
        /// just substitute directly. no need to remove and reinsert keys.
856
        // range partition
857
2
        part->start_key = std::move(old_part->start_key);
858
2
        part->end_key = std::move(old_part->end_key);
859
        // list partition
860
2
        part->in_keys = std::move(old_part->in_keys);
861
2
        if (t_part.__isset.is_default_partition && t_part.is_default_partition) {
862
0
            _default_partition = part;
863
0
        }
864
865
2
        part->num_buckets = t_part.num_buckets;
866
2
        auto num_indexes = _schema->indexes().size();
867
2
        if (t_part.indexes.size() != num_indexes) {
868
0
            return Status::InternalError(
869
0
                    "number of partition's index is not equal with schema's"
870
0
                    ", num_part_indexes={}, num_schema_indexes={}",
871
0
                    t_part.indexes.size(), num_indexes);
872
0
        }
873
2
        part->indexes = t_part.indexes;
874
2
        std::sort(part->indexes.begin(), part->indexes.end(),
875
2
                  [](const OlapTableIndexTablets& lhs, const OlapTableIndexTablets& rhs) {
876
0
                      return lhs.index_id < rhs.index_id;
877
0
                  });
878
        // check index
879
4
        for (int j = 0; j < num_indexes; ++j) {
880
2
            if (part->indexes[j].index_id != _schema->indexes()[j]->index_id) {
881
0
                return Status::InternalError(
882
0
                        "partition's index is not equal with schema's"
883
0
                        ", part_index={}, schema_index={}",
884
0
                        part->indexes[j].index_id, _schema->indexes()[j]->index_id);
885
0
            }
886
2
        }
887
888
        // add new partitions with new id.
889
2
        _partitions.emplace_back(part);
890
2
        VLOG_NOTICE << "params add new partition " << part->id;
891
892
        // replace items in _partition_maps
893
2
        if (_is_in_partition) {
894
0
            for (auto& in_key : part->in_keys) {
895
0
                (*_partitions_map)[std::tuple {in_key.first, in_key.second, false}] = part;
896
0
            }
897
2
        } else {
898
2
            (*_partitions_map)[std::tuple {part->end_key.first, part->end_key.second, false}] =
899
2
                    part;
900
2
        }
901
2
    }
902
    // remove old partitions by id
903
1
    std::ranges::sort(old_partition_ids);
904
5
    for (auto it = _partitions.begin(); it != _partitions.end();) {
905
4
        if (std::ranges::binary_search(old_partition_ids, (*it)->id)) {
906
2
            it = _partitions.erase(it);
907
2
        } else {
908
2
            it++;
909
2
        }
910
4
    }
911
912
1
    return Status::OK();
913
1
}
914
#include "common/compile_check_end.h"
915
916
} // namespace doris