Coverage Report

Created: 2026-08-07 14:29

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/cloud/cloud_rowset_builder.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 "cloud/cloud_rowset_builder.h"
19
20
#include "cloud/cloud_meta_mgr.h"
21
#include "cloud/cloud_storage_engine.h"
22
#include "cloud/cloud_tablet.h"
23
#include "cloud/cloud_tablet_mgr.h"
24
#include "storage/rowset/group_rowset_writer.h"
25
#include "storage/rowset/rowset_factory.h"
26
#include "storage/rowset/rowset_writer_context.h"
27
#include "storage/storage_policy.h"
28
#include "storage/tablet_info.h"
29
30
namespace doris {
31
using namespace ErrorCode;
32
33
CloudRowsetBuilder::CloudRowsetBuilder(CloudStorageEngine& engine, const WriteRequest& req,
34
                                       RuntimeProfile* profile)
35
266k
        : BaseRowsetBuilder(req, profile), _engine(engine) {}
36
37
CloudGroupRowsetBuilder::CloudGroupRowsetBuilder(CloudStorageEngine& engine,
38
                                                 const WriteRequest& group_build_req,
39
                                                 const WriteRequest& sub_data_req,
40
                                                 const WriteRequest& sub_row_binlog_req,
41
                                                 RuntimeProfile* profile)
42
1
        : CloudRowsetBuilder(engine, group_build_req, profile) {
43
1
    DCHECK(group_build_req.write_req_type == WriteRequestType::GROUP &&
44
1
           sub_data_req.write_req_type == WriteRequestType::DATA &&
45
1
           sub_row_binlog_req.write_req_type == WriteRequestType::ROW_BINLOG);
46
1
    _data_builder = std::make_shared<CloudRowsetBuilder>(engine, sub_data_req, profile);
47
1
    _row_binlog_builder = std::make_shared<CloudRowsetBuilder>(engine, sub_row_binlog_req, profile);
48
1
}
49
50
266k
CloudRowsetBuilder::~CloudRowsetBuilder() {
51
    // Clear file cache immediately when load fails
52
266k
    if (_is_init && _rowset != nullptr && _rowset->rowset_meta()->rowset_state() == PREPARED) {
53
0
        _rowset->clear_cache();
54
0
    }
55
266k
}
56
57
2
Status CloudRowsetBuilder::init() {
58
2
    _tablet = DORIS_TRY(_engine.get_tablet(_req.tablet_id));
59
60
2
    std::shared_ptr<MowContext> mow_context;
61
2
    if (_tablet->enable_unique_key_merge_on_write() && is_data_builder()) {
62
1
        if (config::cloud_mow_sync_rowsets_when_load_txn_begin) {
63
0
            auto st = std::static_pointer_cast<CloudTablet>(_tablet)->sync_rowsets();
64
            // sync_rowsets will return INVALID_TABLET_STATE when tablet is under alter
65
0
            if (!st.ok() && !st.is<ErrorCode::INVALID_TABLET_STATE>()) {
66
0
                return st;
67
0
            }
68
0
        }
69
1
        RETURN_IF_ERROR(init_mow_context(mow_context));
70
1
    } else if (_req.write_req_type == WriteRequestType::ROW_BINLOG) {
71
        // Row binlog tablets use txn_delete_bitmap_cache for local make-visible.
72
        // The real binlog delete bitmap is derived when the base tablet calculates delete bitmap.
73
1
        _delete_bitmap = std::make_shared<DeleteBitmap>(_req.tablet_id);
74
1
    }
75
2
    RETURN_IF_ERROR(check_tablet_version_count());
76
77
2
    using namespace std::chrono;
78
2
    std::static_pointer_cast<CloudTablet>(_tablet)->last_load_time_ms =
79
2
            duration_cast<milliseconds>(system_clock::now().time_since_epoch()).count();
80
81
    // build tablet schema in request level
82
2
    RETURN_IF_ERROR(_build_current_tablet_schema(_req.index_id, _req.table_schema_param.get(),
83
2
                                                 *_tablet->tablet_schema()));
84
85
2
    RowsetWriterContext context;
86
2
    context.txn_id = _req.txn_id;
87
2
    context.txn_expiration = _req.txn_expiration;
88
2
    context.load_id = _req.load_id;
89
2
    context.db_id = _req.table_schema_param->db_id();
90
2
    context.table_id = _req.table_schema_param->table_id();
91
2
    context.rowset_state = PREPARED;
92
2
    context.segments_overlap = OVERLAPPING;
93
2
    context.tablet_schema = _tablet_schema;
94
2
    context.newest_write_timestamp = UnixSeconds();
95
2
    context.tablet_id = _req.tablet_id;
96
2
    context.tablet_schema_hash = _req.schema_hash;
97
2
    context.index_id = _req.index_id;
98
2
    context.tablet = _tablet;
99
2
    if (_req.write_req_type == WriteRequestType::ROW_BINLOG || !_attach_rowset_ids.empty()) {
100
2
        context.enable_segcompaction = false;
101
2
    }
102
2
    context.write_type = DataWriteType::TYPE_DIRECT;
103
2
    context.mow_context = mow_context;
104
2
    context.write_file_cache = _req.write_file_cache;
105
2
    context.partial_update_info = _partial_update_info;
106
2
    context.write_binlog_opt().enable = _req.write_req_type == WriteRequestType::ROW_BINLOG;
107
2
    context.file_cache_ttl_sec = _tablet->ttl_seconds();
108
2
    context.storage_resource = _engine.get_storage_resource(_req.storage_vault_id);
109
2
    if (!context.storage_resource) {
110
0
        return Status::InternalError("vault id not found, maybe not sync, vault id {}",
111
0
                                     _req.storage_vault_id);
112
0
    }
113
114
2
    _rowset_writer = DORIS_TRY(_tablet->create_rowset_writer(context, false));
115
2
    _rowset_id = context.rowset_id;
116
117
2
    _calc_delete_bitmap_token = _engine.calc_delete_bitmap_executor()->create_token();
118
119
2
    if (!_skip_writing_rowset_metadata) {
120
0
        RETURN_IF_ERROR(_engine.meta_mgr().prepare_rowset(*_rowset_writer->rowset_meta(), "",
121
0
                                                          _tablet->table_id()));
122
0
    }
123
124
2
    _is_init = true;
125
2
    return Status::OK();
126
2
}
127
128
1
Status CloudGroupRowsetBuilder::init() {
129
1
    RETURN_IF_ERROR(_row_binlog_builder->init());
130
1
    RETURN_IF_ERROR(
131
1
            _data_builder->attach_pending_rs_guard_to_txn(_row_binlog_builder->rowset_id()));
132
1
    RETURN_IF_ERROR(_data_builder->init());
133
1
    _tablet = _data_builder->tablet_sptr();
134
135
1
    std::unique_ptr<GroupRowsetWriter> group_writer;
136
1
    RETURN_IF_ERROR(RowsetFactory::create_empty_group_rowset_writer(&group_writer));
137
1
    group_writer->set_data_writer(_data_builder->rowset_writer());
138
1
    group_writer->set_row_binlog_writer(_row_binlog_builder->rowset_writer());
139
140
1
    {
141
1
        const auto& data_ctx = _data_builder->rowset_writer()->context();
142
1
        auto& binlog_ctx =
143
1
                const_cast<RowsetWriterContext&>(_row_binlog_builder->rowset_writer()->context());
144
1
        auto& cfg = binlog_ctx.write_binlog_opt().write_binlog_config();
145
1
        cfg.source.tablet_schema = data_ctx.tablet_schema;
146
1
        cfg.source.partial_update_info = data_ctx.partial_update_info;
147
1
        cfg.source.mow_context = data_ctx.mow_context;
148
1
        cfg.source.is_transient_rowset_writer = data_ctx.is_transient_rowset_writer;
149
1
        cfg.source.source_write_type = data_ctx.write_type;
150
1
        cfg.source.base_tablet = _data_builder->tablet_sptr();
151
1
    }
152
153
1
    _rowset_writer = std::move(group_writer);
154
1
    _is_init = true;
155
1
    return Status::OK();
156
1
}
157
158
1
Status CloudGroupRowsetBuilder::build_rowset() {
159
1
    RETURN_IF_ERROR(_row_binlog_builder->build_rowset());
160
1
    return _data_builder->build_rowset();
161
1
}
162
163
0
Status CloudGroupRowsetBuilder::submit_calc_delete_bitmap_task() {
164
0
    return _data_builder->submit_calc_delete_bitmap_task();
165
0
}
166
167
0
Status CloudGroupRowsetBuilder::wait_calc_delete_bitmap() {
168
0
    return _data_builder->wait_calc_delete_bitmap();
169
0
}
170
171
0
void CloudGroupRowsetBuilder::update_tablet_stats() {
172
0
    _data_builder->update_tablet_stats();
173
0
    _row_binlog_builder->update_tablet_stats();
174
0
}
175
176
0
Status CloudGroupRowsetBuilder::commit_rowset(const std::string& job_id, int64_t table_id) {
177
0
    return _engine.meta_mgr().commit_rowset(*_data_builder->rowset_meta(), job_id, table_id,
178
0
                                            nullptr, _row_binlog_builder->rowset_meta().get());
179
0
}
180
181
0
Status CloudGroupRowsetBuilder::set_txn_related_info() {
182
0
    RowBinlogTxnInfo attach_row_binlog;
183
0
    attach_row_binlog.rowset = _row_binlog_builder->rowset();
184
0
    attach_row_binlog.tablet = _row_binlog_builder->tablet_sptr();
185
0
    if (_data_builder->tablet()->enable_unique_key_merge_on_write()) {
186
0
        attach_row_binlog.delete_bitmap =
187
0
                std::make_shared<DeleteBitmap>(_row_binlog_builder->tablet()->tablet_id());
188
0
    }
189
0
    RETURN_IF_ERROR(_data_builder->attach_row_binlog_to_txn(attach_row_binlog));
190
0
    RETURN_IF_ERROR(_data_builder->set_txn_related_info());
191
0
    return _row_binlog_builder->set_txn_related_info();
192
0
}
193
194
1
void CloudGroupRowsetBuilder::set_skip_writing_rowset_metadata(bool skip) {
195
1
    _data_builder->set_skip_writing_rowset_metadata(skip);
196
1
    _row_binlog_builder->set_skip_writing_rowset_metadata(skip);
197
1
}
198
199
173k
Status CloudRowsetBuilder::check_tablet_version_count() {
200
173k
    int64_t version_count = cloud_tablet()->fetch_add_approximate_num_rowsets(0);
201
173k
    DBUG_EXECUTE_IF("RowsetBuilder.check_tablet_version_count.too_many_version",
202
173k
                    { version_count = INT_MAX; });
203
    // TODO(plat1ko): load backoff algorithm
204
173k
    int32_t max_version_config = cloud_tablet()->max_version_config();
205
173k
    if (version_count > max_version_config) {
206
0
        return Status::Error<TOO_MANY_VERSION>(
207
0
                "failed to init rowset builder. version count: {}, exceed limit: {}, "
208
0
                "tablet: {}. Please reduce the frequency of loading data or adjust the "
209
0
                "max_tablet_version_num or time_series_max_tablet_version_numin be.conf to a "
210
0
                "larger value.",
211
0
                version_count, max_version_config, _tablet->tablet_id());
212
0
    }
213
173k
    return Status::OK();
214
173k
}
215
216
174k
void CloudRowsetBuilder::update_tablet_stats() {
217
174k
    auto* tablet = cloud_tablet();
218
174k
    DCHECK(tablet);
219
174k
    DCHECK(_rowset);
220
174k
    tablet->fetch_add_approximate_num_rowsets(1);
221
174k
    tablet->fetch_add_approximate_num_segments(_rowset->num_segments());
222
174k
    tablet->fetch_add_approximate_num_rows(_rowset->num_rows());
223
174k
    tablet->fetch_add_approximate_data_size(_rowset->total_disk_size());
224
174k
    tablet->fetch_add_approximate_cumu_num_rowsets(1);
225
174k
    tablet->fetch_add_approximate_cumu_num_deltas(_rowset->num_segments());
226
174k
    tablet->write_count.fetch_add(1, std::memory_order_relaxed);
227
174k
}
228
229
521k
CloudTablet* CloudRowsetBuilder::cloud_tablet() {
230
521k
    return static_cast<CloudTablet*>(_tablet.get());
231
521k
}
232
233
87.9k
const RowsetMetaSharedPtr& CloudRowsetBuilder::rowset_meta() {
234
87.9k
    return _rowset_writer->rowset_meta();
235
87.9k
}
236
237
0
Status CloudRowsetBuilder::commit_rowset(const std::string& job_id, int64_t table_id) {
238
0
    return _engine.meta_mgr().commit_rowset(*rowset_meta(), job_id, table_id);
239
0
}
240
241
0
Status CloudRowsetBuilder::set_txn_related_info() {
242
0
    if (_tablet->enable_unique_key_merge_on_write() || _tablet->is_row_binlog_tablet()) {
243
        // For empty rowsets when skip_writing_empty_rowset_metadata=true,
244
        // store only a lightweight marker instead of full rowset info.
245
        // This allows CalcDeleteBitmapTask to detect and skip gracefully,
246
        // while using minimal memory (~16 bytes per entry).
247
0
        if (_skip_writing_rowset_metadata) {
248
0
            _engine.txn_delete_bitmap_cache().mark_empty_rowset(_req.txn_id, _tablet->tablet_id(),
249
0
                                                                _req.txn_expiration);
250
0
            return Status::OK();
251
0
        }
252
0
        if (config::enable_merge_on_write_correctness_check &&
253
0
            _tablet->enable_unique_key_merge_on_write() && _rowset->num_rows() != 0) {
254
0
            auto st = _tablet->check_delete_bitmap_correctness(
255
0
                    _delete_bitmap, _rowset->end_version() - 1, _req.txn_id, *_rowset_ids);
256
0
            if (!st.ok()) {
257
0
                LOG(WARNING) << fmt::format(
258
0
                        "[tablet_id:{}][txn_id:{}][load_id:{}][partition_id:{}] "
259
0
                        "delete bitmap correctness check failed in commit phase!",
260
0
                        _req.tablet_id, _req.txn_id, UniqueId(_req.load_id).to_string(),
261
0
                        _req.partition_id);
262
0
                return st;
263
0
            }
264
0
        }
265
0
        _engine.txn_delete_bitmap_cache().set_tablet_txn_info(
266
0
                _req.txn_id, _tablet->tablet_id(), _delete_bitmap, *_rowset_ids, _rowset,
267
0
                _req.txn_expiration, _partial_update_info, _attach_row_binlog);
268
0
    } else {
269
        // TSO-enabled rowsets must become visible from MS rowset meta.
270
0
        if (config::enable_cloud_make_rs_visible_on_be && !_tablet_schema->is_tso_enabled()) {
271
0
            if (_skip_writing_rowset_metadata) {
272
0
                _engine.committed_rs_mgr().mark_empty_rowset(_req.txn_id, _tablet->tablet_id(),
273
0
                                                             _req.txn_expiration);
274
0
            } else {
275
0
                _engine.meta_mgr().cache_committed_rowset(rowset_meta(), _req.txn_expiration);
276
0
            }
277
0
        }
278
0
    }
279
0
    return Status::OK();
280
0
}
281
} // namespace doris