Coverage Report

Created: 2025-03-13 18:46

/root/doris/be/src/olap/rowset_builder.cpp
Line
Count
Source (jump to first uncovered line)
1
// Licensed to the Apache Software Foundation (ASF) under one
2
// or more contributor license agreements.  See the NOTICE file
3
// distributed with this work for additional information
4
// regarding copyright ownership.  The ASF licenses this file
5
// to you under the Apache License, Version 2.0 (the
6
// "License"); you may not use this file except in compliance
7
// with the License.  You may obtain a copy of the License at
8
//
9
//   http://www.apache.org/licenses/LICENSE-2.0
10
//
11
// Unless required by applicable law or agreed to in writing,
12
// software distributed under the License is distributed on an
13
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14
// KIND, either express or implied.  See the License for the
15
// specific language governing permissions and limitations
16
// under the License.
17
18
#include "olap/rowset_builder.h"
19
20
#include <brpc/controller.h>
21
#include <fmt/format.h>
22
23
#include <filesystem>
24
#include <memory>
25
#include <ostream>
26
#include <string>
27
#include <utility>
28
29
// IWYU pragma: no_include <opentelemetry/common/threadlocal.h>
30
#include "common/compiler_util.h" // IWYU pragma: keep
31
#include "common/config.h"
32
#include "common/status.h"
33
#include "exec/tablet_info.h"
34
#include "gutil/strings/numbers.h"
35
#include "io/fs/file_system.h"
36
#include "io/fs/file_writer.h" // IWYU pragma: keep
37
#include "olap/calc_delete_bitmap_executor.h"
38
#include "olap/olap_define.h"
39
#include "olap/partial_update_info.h"
40
#include "olap/rowset/beta_rowset.h"
41
#include "olap/rowset/beta_rowset_writer.h"
42
#include "olap/rowset/pending_rowset_helper.h"
43
#include "olap/rowset/rowset_meta.h"
44
#include "olap/rowset/rowset_meta_manager.h"
45
#include "olap/rowset/rowset_writer.h"
46
#include "olap/rowset/rowset_writer_context.h"
47
#include "olap/schema_change.h"
48
#include "olap/storage_engine.h"
49
#include "olap/tablet_manager.h"
50
#include "olap/tablet_meta.h"
51
#include "olap/tablet_schema.h"
52
#include "olap/txn_manager.h"
53
#include "runtime/memory/global_memory_arbitrator.h"
54
#include "util/brpc_client_cache.h"
55
#include "util/brpc_closure.h"
56
#include "util/debug_points.h"
57
#include "util/mem_info.h"
58
#include "util/stopwatch.hpp"
59
#include "util/time.h"
60
#include "util/trace.h"
61
#include "vec/common/schema_util.h"
62
#include "vec/core/block.h"
63
64
namespace doris {
65
using namespace ErrorCode;
66
67
BaseRowsetBuilder::BaseRowsetBuilder(const WriteRequest& req, RuntimeProfile* profile)
68
28
        : _req(req), _tablet_schema(std::make_shared<TabletSchema>()) {
69
28
    _init_profile(profile);
70
28
}
71
72
RowsetBuilder::RowsetBuilder(StorageEngine& engine, const WriteRequest& req,
73
                             RuntimeProfile* profile)
74
28
        : BaseRowsetBuilder(req, profile), _engine(engine) {}
75
76
28
void BaseRowsetBuilder::_init_profile(RuntimeProfile* profile) {
77
28
    _profile = profile->create_child(fmt::format("RowsetBuilder {}", _req.tablet_id), true, true);
78
28
    _build_rowset_timer = ADD_TIMER(_profile, "BuildRowsetTime");
79
28
    _submit_delete_bitmap_timer = ADD_TIMER(_profile, "DeleteBitmapSubmitTime");
80
28
    _wait_delete_bitmap_timer = ADD_TIMER(_profile, "DeleteBitmapWaitTime");
81
28
}
82
83
0
void RowsetBuilder::_init_profile(RuntimeProfile* profile) {
84
0
    BaseRowsetBuilder::_init_profile(profile);
85
0
    _commit_txn_timer = ADD_TIMER(_profile, "CommitTxnTime");
86
0
}
87
88
28
BaseRowsetBuilder::~BaseRowsetBuilder() {
89
28
    if (!_is_init) {
90
1
        return;
91
1
    }
92
93
27
    if (_calc_delete_bitmap_token != nullptr) {
94
27
        _calc_delete_bitmap_token->cancel();
95
27
    }
96
27
}
97
98
28
RowsetBuilder::~RowsetBuilder() {
99
28
    if (_is_init && !_is_committed) {
100
5
        _garbage_collection();
101
5
    }
102
28
}
103
104
228
Tablet* RowsetBuilder::tablet() {
105
228
    return static_cast<Tablet*>(_tablet.get());
106
228
}
107
108
0
TabletSharedPtr RowsetBuilder::tablet_sptr() {
109
0
    return std::static_pointer_cast<Tablet>(_tablet);
110
0
}
111
112
5
void RowsetBuilder::_garbage_collection() {
113
5
    Status rollback_status;
114
5
    TxnManager* txn_mgr = _engine.txn_manager();
115
5
    if (tablet() != nullptr) {
116
5
        rollback_status = txn_mgr->rollback_txn(_req.partition_id, *tablet(), _req.txn_id);
117
5
    }
118
    // has to check rollback status, because the rowset maybe committed in this thread and
119
    // published in another thread, then rollback will fail.
120
    // when rollback failed should not delete rowset
121
5
    if (rollback_status.ok()) {
122
5
        _engine.add_unused_rowset(_rowset);
123
5
    }
124
5
}
125
126
3
Status BaseRowsetBuilder::init_mow_context(std::shared_ptr<MowContext>& mow_context) {
127
3
    std::lock_guard<std::shared_mutex> lck(tablet()->get_header_lock());
128
3
    _max_version_in_flush_phase = tablet()->max_version_unlocked();
129
3
    std::vector<RowsetSharedPtr> rowset_ptrs;
130
    // tablet is under alter process. The delete bitmap will be calculated after conversion.
131
3
    if (tablet()->tablet_state() == TABLET_NOTREADY) {
132
        // Disable 'partial_update' when the tablet is undergoing a 'schema changing process'
133
0
        if (_req.table_schema_param->is_partial_update()) {
134
0
            return Status::InternalError(
135
0
                    "Unable to do 'partial_update' when "
136
0
                    "the tablet is undergoing a 'schema changing process'");
137
0
        }
138
0
        _rowset_ids.clear();
139
3
    } else {
140
3
        RETURN_IF_ERROR(
141
3
                tablet()->get_all_rs_id_unlocked(_max_version_in_flush_phase, &_rowset_ids));
142
3
        rowset_ptrs = tablet()->get_rowset_by_ids(&_rowset_ids);
143
3
    }
144
3
    _delete_bitmap = std::make_shared<DeleteBitmap>(tablet()->tablet_id());
145
3
    mow_context = std::make_shared<MowContext>(_max_version_in_flush_phase, _req.txn_id,
146
3
                                               _rowset_ids, rowset_ptrs, _delete_bitmap);
147
3
    return Status::OK();
148
3
}
149
150
27
Status RowsetBuilder::check_tablet_version_count() {
151
27
    bool injection = false;
152
27
    DBUG_EXECUTE_IF("RowsetBuilder.check_tablet_version_count.too_many_version",
153
27
                    { injection = true; });
154
27
    if (injection) {
155
        // do not return if injection
156
27
    } else if (!_tablet->exceed_version_limit(config::max_tablet_version_num - 100) ||
157
27
               GlobalMemoryArbitrator::is_exceed_soft_mem_limit(GB_EXCHANGE_BYTE)) {
158
27
        return Status::OK();
159
27
    }
160
    //trigger compaction
161
0
    auto st = _engine.submit_compaction_task(tablet_sptr(), CompactionType::CUMULATIVE_COMPACTION,
162
0
                                             true);
163
0
    if (!st.ok()) [[unlikely]] {
164
0
        LOG(WARNING) << "failed to trigger compaction, tablet_id=" << _tablet->tablet_id() << " : "
165
0
                     << st;
166
0
    }
167
0
    int version_count = tablet()->version_count();
168
0
    DBUG_EXECUTE_IF("RowsetBuilder.check_tablet_version_count.too_many_version",
169
0
                    { version_count = INT_MAX; });
170
0
    if (version_count > config::max_tablet_version_num) {
171
0
        return Status::Error<TOO_MANY_VERSION>(
172
0
                "failed to init rowset builder. version count: {}, exceed limit: {}, "
173
0
                "tablet: {}. Please reduce the frequency of loading data or adjust the "
174
0
                "max_tablet_version_num in be.conf to a larger value.",
175
0
                version_count, config::max_tablet_version_num, _tablet->tablet_id());
176
0
    }
177
0
    return Status::OK();
178
0
}
179
180
27
Status RowsetBuilder::prepare_txn() {
181
27
    std::shared_lock base_migration_lock(tablet()->get_migration_lock(), std::defer_lock);
182
27
    if (!base_migration_lock.try_lock_for(
183
27
                std::chrono::milliseconds(config::migration_lock_timeout_ms))) {
184
0
        return Status::Error<TRY_LOCK_FAILED>("try_lock migration lock failed after {}ms",
185
0
                                              config::migration_lock_timeout_ms);
186
0
    }
187
27
    std::lock_guard<std::mutex> push_lock(tablet()->get_push_lock());
188
27
    return _engine.txn_manager()->prepare_txn(_req.partition_id, *tablet(), _req.txn_id,
189
27
                                              _req.load_id);
190
27
}
191
192
28
Status RowsetBuilder::init() {
193
28
    _tablet = DORIS_TRY(_engine.get_tablet(_req.tablet_id));
194
27
    std::shared_ptr<MowContext> mow_context;
195
27
    if (_tablet->enable_unique_key_merge_on_write()) {
196
3
        RETURN_IF_ERROR(init_mow_context(mow_context));
197
3
    }
198
199
27
    if (!config::disable_auto_compaction &&
200
27
        !_tablet->tablet_meta()->tablet_schema()->disable_auto_compaction()) {
201
27
        RETURN_IF_ERROR(check_tablet_version_count());
202
27
    }
203
204
27
    int version_count = tablet()->version_count() + tablet()->stale_version_count();
205
27
    if (tablet()->avg_rs_meta_serialize_size() * version_count >
206
27
        config::tablet_meta_serialize_size_limit) {
207
0
        return Status::Error<TOO_MANY_VERSION>(
208
0
                "failed to init rowset builder. meta serialize size : {}, exceed limit: {}, "
209
0
                "tablet: {}. Please reduce the frequency of loading data or adjust the "
210
0
                "max_tablet_version_num in be.conf to a larger value.",
211
0
                tablet()->avg_rs_meta_serialize_size() * version_count,
212
0
                config::tablet_meta_serialize_size_limit, _tablet->tablet_id());
213
0
    }
214
215
27
    RETURN_IF_ERROR(prepare_txn());
216
217
27
    DBUG_EXECUTE_IF("BaseRowsetBuilder::init.check_partial_update_column_num", {
218
27
        if (_req.table_schema_param->partial_update_input_columns().size() !=
219
27
            dp->param<int>("column_num")) {
220
27
            return Status::InternalError("partial update input column num wrong!");
221
27
        };
222
27
    })
223
    // build tablet schema in request level
224
27
    RETURN_IF_ERROR(_build_current_tablet_schema(_req.index_id, _req.table_schema_param.get(),
225
27
                                                 *_tablet->tablet_schema()));
226
27
    RowsetWriterContext context;
227
27
    context.txn_id = _req.txn_id;
228
27
    context.load_id = _req.load_id;
229
27
    context.rowset_state = PREPARED;
230
27
    context.segments_overlap = OVERLAPPING;
231
27
    context.tablet_schema = _tablet_schema;
232
27
    context.newest_write_timestamp = UnixSeconds();
233
27
    context.tablet_id = _req.tablet_id;
234
27
    context.index_id = _req.index_id;
235
27
    context.tablet = _tablet;
236
27
    context.write_type = DataWriteType::TYPE_DIRECT;
237
27
    context.mow_context = mow_context;
238
27
    context.write_file_cache = _req.write_file_cache;
239
27
    context.partial_update_info = _partial_update_info;
240
27
    _rowset_writer = DORIS_TRY(_tablet->create_rowset_writer(context, false));
241
27
    _pending_rs_guard = _engine.pending_local_rowsets().add(context.rowset_id);
242
243
27
    _calc_delete_bitmap_token = _engine.calc_delete_bitmap_executor()->create_token();
244
245
27
    _is_init = true;
246
27
    return Status::OK();
247
27
}
248
249
22
Status BaseRowsetBuilder::build_rowset() {
250
22
    std::lock_guard<std::mutex> l(_lock);
251
22
    DCHECK(_is_init) << "rowset builder is supposed be to initialized before "
252
0
                        "build_rowset() being called";
253
254
22
    SCOPED_TIMER(_build_rowset_timer);
255
    // use rowset meta manager to save meta
256
22
    RETURN_NOT_OK_STATUS_WITH_WARN(_rowset_writer->build(_rowset), "fail to build rowset");
257
22
    return Status::OK();
258
22
}
259
260
16
Status BaseRowsetBuilder::submit_calc_delete_bitmap_task() {
261
16
    if (!_tablet->enable_unique_key_merge_on_write()) {
262
13
        return Status::OK();
263
13
    }
264
3
    std::lock_guard<std::mutex> l(_lock);
265
3
    SCOPED_TIMER(_submit_delete_bitmap_timer);
266
3
    if (_partial_update_info && _partial_update_info->is_flexible_partial_update()) {
267
0
        if (_rowset->num_segments() > 1) {
268
            // in flexible partial update, when there are more one segment in one load,
269
            // we need to do alignment process for same keys between segments, we haven't
270
            // implemented it yet and just report an error when encouter this situation
271
0
            return Status::NotSupported(
272
0
                    "too large input data in flexible partial update, Please "
273
0
                    "reduce the amount of data imported in a single load.");
274
0
        }
275
0
    }
276
277
3
    auto* beta_rowset = reinterpret_cast<BetaRowset*>(_rowset.get());
278
3
    std::vector<segment_v2::SegmentSharedPtr> segments;
279
3
    RETURN_IF_ERROR(beta_rowset->load_segments(&segments));
280
3
    if (segments.size() > 1) {
281
        // calculate delete bitmap between segments
282
0
        RETURN_IF_ERROR(_tablet->calc_delete_bitmap_between_segments(_rowset->rowset_id(), segments,
283
0
                                                                     _delete_bitmap));
284
0
    }
285
286
    // tablet is under alter process. The delete bitmap will be calculated after conversion.
287
3
    if (_tablet->tablet_state() == TABLET_NOTREADY) {
288
0
        LOG(INFO) << "tablet is under alter process, delete bitmap will be calculated later, "
289
0
                     "tablet_id: "
290
0
                  << _tablet->tablet_id() << " txn_id: " << _req.txn_id;
291
0
        return Status::OK();
292
0
    }
293
294
    // For partial update, we need to fill in the entire row of data, during the calculation
295
    // of the delete bitmap. This operation is resource-intensive, and we need to minimize
296
    // the number of times it occurs. Therefore, we skip this operation here.
297
3
    if (_partial_update_info->is_partial_update()) {
298
        // for partial update, the delete bitmap calculation is done while append_block()
299
        // we print it's summarize logs here before commit.
300
0
        LOG(INFO) << fmt::format(
301
0
                "{} calc delete bitmap summary before commit: tablet({}), txn_id({}), "
302
0
                "rowset_ids({}), cur max_version({}), bitmap num({}), bitmap_cardinality({}), num "
303
0
                "rows updated({}), num rows new added({}), num rows deleted({}), total rows({})",
304
0
                _partial_update_info->partial_update_mode_str(), tablet()->tablet_id(), _req.txn_id,
305
0
                _rowset_ids.size(), rowset_writer()->context().mow_context->max_version,
306
0
                _delete_bitmap->get_delete_bitmap_count(), _delete_bitmap->cardinality(),
307
0
                rowset_writer()->num_rows_updated(), rowset_writer()->num_rows_new_added(),
308
0
                rowset_writer()->num_rows_deleted(), rowset_writer()->num_rows());
309
0
        return Status::OK();
310
0
    }
311
312
3
    LOG(INFO) << "submit calc delete bitmap task to executor, tablet_id: " << tablet()->tablet_id()
313
3
              << ", txn_id: " << _req.txn_id;
314
3
    return BaseTablet::commit_phase_update_delete_bitmap(_tablet, _rowset, _rowset_ids,
315
3
                                                         _delete_bitmap, segments, _req.txn_id,
316
3
                                                         _calc_delete_bitmap_token.get(), nullptr);
317
3
}
318
319
16
Status BaseRowsetBuilder::wait_calc_delete_bitmap() {
320
16
    if (!_tablet->enable_unique_key_merge_on_write() || _partial_update_info->is_partial_update()) {
321
13
        return Status::OK();
322
13
    }
323
3
    std::lock_guard<std::mutex> l(_lock);
324
3
    SCOPED_TIMER(_wait_delete_bitmap_timer);
325
3
    RETURN_IF_ERROR(_calc_delete_bitmap_token->wait());
326
3
    return Status::OK();
327
3
}
328
329
22
Status RowsetBuilder::commit_txn() {
330
22
    if (tablet()->enable_unique_key_merge_on_write() &&
331
22
        config::enable_merge_on_write_correctness_check && _rowset->num_rows() != 0 &&
332
22
        tablet()->tablet_state() != TABLET_NOTREADY) {
333
3
        auto st = tablet()->check_delete_bitmap_correctness(
334
3
                _delete_bitmap, _rowset->end_version() - 1, _req.txn_id, _rowset_ids);
335
3
        if (!st.ok()) {
336
0
            LOG(WARNING) << fmt::format(
337
0
                    "[tablet_id:{}][txn_id:{}][load_id:{}][partition_id:{}] "
338
0
                    "delete bitmap correctness check failed in commit phase!",
339
0
                    _req.tablet_id, _req.txn_id, UniqueId(_req.load_id).to_string(),
340
0
                    _req.partition_id);
341
0
            return st;
342
0
        }
343
3
    }
344
22
    std::lock_guard<std::mutex> l(_lock);
345
22
    SCOPED_TIMER(_commit_txn_timer);
346
347
22
    const RowsetWriterContext& rw_ctx = _rowset_writer->context();
348
22
    if (rw_ctx.tablet_schema->num_variant_columns() > 0 && _rowset->num_rows() > 0) {
349
        // Need to merge schema with `rw_ctx.merged_tablet_schema` in prior,
350
        // merged schema keeps the newest merged schema for the rowset, which is updated and merged
351
        // during flushing segments.
352
0
        if (rw_ctx.merged_tablet_schema != nullptr) {
353
0
            RETURN_IF_ERROR(tablet()->update_by_least_common_schema(rw_ctx.merged_tablet_schema));
354
0
        } else {
355
            // We should merge rowset schema further, in case that the merged_tablet_schema maybe null
356
            // when enable_memtable_on_sink_node is true, the merged_tablet_schema will not be passed to
357
            // the destination backend.
358
            // update tablet schema when meet variant columns, before commit_txn
359
            // Eg. rowset schema:       A(int),    B(float),  C(int), D(int)
360
            // _tabelt->tablet_schema:  A(bigint), B(double)
361
            //  => update_schema:       A(bigint), B(double), C(int), D(int)
362
0
            RETURN_IF_ERROR(tablet()->update_by_least_common_schema(rw_ctx.tablet_schema));
363
0
        }
364
0
    }
365
366
    // Transfer ownership of `PendingRowsetGuard` to `TxnManager`
367
22
    Status res = _engine.txn_manager()->commit_txn(
368
22
            _req.partition_id, *tablet(), _req.txn_id, _req.load_id, _rowset,
369
22
            std::move(_pending_rs_guard), false, _partial_update_info);
370
371
22
    if (!res && !res.is<PUSH_TRANSACTION_ALREADY_EXIST>()) {
372
0
        LOG(WARNING) << "Failed to commit txn: " << _req.txn_id
373
0
                     << " for rowset: " << _rowset->rowset_id();
374
0
        return res;
375
0
    }
376
22
    if (_tablet->enable_unique_key_merge_on_write()) {
377
3
        _engine.txn_manager()->set_txn_related_delete_bitmap(
378
3
                _req.partition_id, _req.txn_id, tablet()->tablet_id(), tablet()->tablet_uid(), true,
379
3
                _delete_bitmap, _rowset_ids, _partial_update_info);
380
3
    }
381
382
22
    _is_committed = true;
383
22
    return Status::OK();
384
22
}
385
386
0
Status BaseRowsetBuilder::cancel() {
387
0
    std::lock_guard<std::mutex> l(_lock);
388
0
    if (_is_cancelled) {
389
0
        return Status::OK();
390
0
    }
391
0
    if (_calc_delete_bitmap_token != nullptr) {
392
0
        _calc_delete_bitmap_token->cancel();
393
0
    }
394
0
    _is_cancelled = true;
395
0
    return Status::OK();
396
0
}
397
398
Status BaseRowsetBuilder::_build_current_tablet_schema(
399
        int64_t index_id, const OlapTableSchemaParam* table_schema_param,
400
27
        const TabletSchema& ori_tablet_schema) {
401
    // find the right index id
402
27
    int i = 0;
403
27
    auto indexes = table_schema_param->indexes();
404
27
    for (; i < indexes.size(); i++) {
405
13
        if (indexes[i]->index_id == index_id) {
406
13
            break;
407
13
        }
408
13
    }
409
27
    if (!indexes.empty() && !indexes[i]->columns.empty() &&
410
27
        indexes[i]->columns[0]->unique_id() >= 0) {
411
0
        _tablet_schema->shawdow_copy_without_columns(ori_tablet_schema);
412
0
        _tablet_schema->build_current_tablet_schema(index_id, table_schema_param->version(),
413
0
                                                    indexes[i], ori_tablet_schema);
414
27
    } else {
415
27
        _tablet_schema->copy_from(ori_tablet_schema);
416
27
    }
417
27
    if (_tablet_schema->schema_version() > ori_tablet_schema.schema_version()) {
418
        // After schema change, should include extracted column
419
        // For example: a table has two columns, k and v
420
        // After adding a column v2, the schema version increases, max_version_schema needs to be updated.
421
        // _tablet_schema includes k, v, and v2
422
        // if v is a variant, need to add the columns decomposed from the v to the _tablet_schema.
423
0
        if (_tablet_schema->num_variant_columns() > 0) {
424
0
            TabletSchemaSPtr max_version_schema = std::make_shared<TabletSchema>();
425
0
            max_version_schema->copy_from(*_tablet_schema);
426
0
            max_version_schema->copy_extracted_columns(ori_tablet_schema);
427
0
            _tablet->update_max_version_schema(max_version_schema);
428
0
        } else {
429
0
            _tablet->update_max_version_schema(_tablet_schema);
430
0
        }
431
0
    }
432
433
27
    _tablet_schema->set_table_id(table_schema_param->table_id());
434
27
    _tablet_schema->set_db_id(table_schema_param->db_id());
435
27
    if (table_schema_param->is_partial_update()) {
436
0
        _tablet_schema->set_auto_increment_column(table_schema_param->auto_increment_coulumn());
437
0
    }
438
    // set partial update columns info
439
27
    _partial_update_info = std::make_shared<PartialUpdateInfo>();
440
27
    RETURN_IF_ERROR(_partial_update_info->init(
441
27
            tablet()->tablet_id(), _req.txn_id, *_tablet_schema,
442
27
            table_schema_param->unique_key_update_mode(),
443
27
            table_schema_param->partial_update_input_columns(),
444
27
            table_schema_param->is_strict_mode(), table_schema_param->timestamp_ms(),
445
27
            table_schema_param->nano_seconds(), table_schema_param->timezone(),
446
27
            table_schema_param->auto_increment_coulumn(),
447
27
            table_schema_param->sequence_map_col_uid(), _max_version_in_flush_phase));
448
27
    return Status::OK();
449
27
}
450
451
} // namespace doris