Coverage Report

Created: 2026-08-14 19:57

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/storage/task/index_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 "storage/task/index_builder.h"
19
20
#include <mutex>
21
22
#include "common/logging.h"
23
#include "common/status.h"
24
#include "storage/index/index_file_reader.h"
25
#include "storage/index/index_file_writer.h"
26
#include "storage/index/index_writer.h" // IndexColumnWriter, complete type for member calls
27
#include "storage/index/inverted/inverted_index_desc.h"
28
#include "storage/index/inverted/inverted_index_fs_directory.h"
29
#include "storage/olap_define.h"
30
#include "storage/rowset/beta_rowset.h"
31
#include "storage/rowset/rowset_writer_context.h"
32
#include "storage/schema.h"
33
#include "storage/segment/segment_loader.h"
34
#include "storage/storage_engine.h"
35
#include "storage/tablet/tablet_schema.h"
36
#include "util/debug_points.h"
37
#include "util/trace.h"
38
39
namespace doris {
40
41
IndexBuilder::IndexBuilder(StorageEngine& engine, TabletSharedPtr tablet,
42
                           const std::vector<TColumn>& columns,
43
                           const std::vector<doris::TOlapTableIndex>& alter_inverted_indexes,
44
                           bool is_drop_op)
45
25
        : _engine(engine),
46
25
          _tablet(std::move(tablet)),
47
25
          _columns(columns),
48
25
          _alter_inverted_indexes(alter_inverted_indexes),
49
25
          _is_drop_op(is_drop_op) {
50
25
    _olap_data_convertor = std::make_unique<OlapBlockDataConvertor>();
51
25
}
52
53
25
IndexBuilder::~IndexBuilder() {
54
25
    _olap_data_convertor.reset();
55
25
    _index_column_writers.clear();
56
25
}
57
58
25
Status IndexBuilder::init() {
59
27
    for (auto inverted_index : _alter_inverted_indexes) {
60
27
        _alter_index_ids.insert(inverted_index.index_id);
61
27
    }
62
25
    return Status::OK();
63
25
}
64
65
21
Status IndexBuilder::update_inverted_index_info() {
66
    // just do link files
67
21
    LOG(INFO) << "begin to update_inverted_index_info, tablet=" << _tablet->tablet_id()
68
21
              << ", is_drop_op=" << _is_drop_op;
69
    // index ids that will not be linked
70
21
    std::set<int64_t> without_index_uids;
71
21
    _output_rowsets.reserve(_input_rowsets.size());
72
21
    _pending_rs_guards.reserve(_input_rowsets.size());
73
22
    for (auto&& input_rowset : _input_rowsets) {
74
22
        bool is_local_rowset = input_rowset->is_local();
75
22
        DBUG_EXECUTE_IF("IndexBuilder::update_inverted_index_info_is_local_rowset",
76
22
                        { is_local_rowset = false; })
77
22
        if (!is_local_rowset) [[unlikely]] {
78
            // DCHECK(false) << _tablet->tablet_id() << ' ' << input_rowset->rowset_id();
79
0
            return Status::InternalError("should be local rowset. tablet_id={} rowset_id={}",
80
0
                                         _tablet->tablet_id(),
81
0
                                         input_rowset->rowset_id().to_string());
82
0
        }
83
84
22
        TabletSchemaSPtr output_rs_tablet_schema = std::make_shared<TabletSchema>();
85
22
        const auto& input_rs_tablet_schema = input_rowset->tablet_schema();
86
22
        output_rs_tablet_schema->copy_from(*input_rs_tablet_schema);
87
22
        int64_t total_index_size = 0;
88
22
        auto* beta_rowset = reinterpret_cast<BetaRowset*>(input_rowset.get());
89
22
        auto size_st = beta_rowset->get_inverted_index_size(&total_index_size);
90
22
        DBUG_EXECUTE_IF("IndexBuilder::update_inverted_index_info_size_st_not_ok", {
91
22
            size_st = Status::Error<ErrorCode::INIT_FAILED>("debug point: get fs failed");
92
22
        })
93
22
        if (!size_st.ok() && !size_st.is<ErrorCode::INVERTED_INDEX_FILE_NOT_FOUND>() &&
94
22
            !size_st.is<ErrorCode::NOT_FOUND>()) {
95
0
            return size_st;
96
0
        }
97
22
        auto num_segments = input_rowset->num_segments();
98
22
        size_t drop_index_size = 0;
99
100
22
        if (_is_drop_op) {
101
6
            for (const auto& t_inverted_index : _alter_inverted_indexes) {
102
6
                DCHECK_EQ(t_inverted_index.columns.size(), 1);
103
6
                auto column_name = t_inverted_index.columns[0];
104
6
                auto column_idx = output_rs_tablet_schema->field_index(column_name);
105
6
                if (column_idx < 0) {
106
0
                    if (!t_inverted_index.column_unique_ids.empty()) {
107
0
                        auto column_unique_id = t_inverted_index.column_unique_ids[0];
108
0
                        column_idx = output_rs_tablet_schema->field_index(column_unique_id);
109
0
                    }
110
0
                    if (column_idx < 0) {
111
0
                        LOG(WARNING) << "referenced column was missing. "
112
0
                                     << "[column=" << column_name
113
0
                                     << " referenced_column=" << column_idx << "]";
114
0
                        continue;
115
0
                    }
116
0
                }
117
6
                auto column = output_rs_tablet_schema->column(column_idx);
118
119
                // inverted index
120
6
                auto index_metas = output_rs_tablet_schema->inverted_indexs(column);
121
6
                for (const auto& index_meta : index_metas) {
122
                    // Only drop the index that matches the requested index_id,
123
                    // not all indexes on this column
124
6
                    if (index_meta->index_id() != t_inverted_index.index_id) {
125
1
                        continue;
126
1
                    }
127
5
                    if (output_rs_tablet_schema->get_inverted_index_storage_format() ==
128
5
                        InvertedIndexStorageFormatPB::V1) {
129
1
                        const auto& fs = io::global_local_filesystem();
130
131
2
                        for (int seg_id = 0; seg_id < num_segments; seg_id++) {
132
1
                            auto seg_path = local_segment_path(
133
1
                                    _tablet->tablet_path(), input_rowset->rowset_id().to_string(),
134
1
                                    seg_id);
135
1
                            auto index_path = InvertedIndexDescriptor::get_index_file_path_v1(
136
1
                                    InvertedIndexDescriptor::get_index_file_path_prefix(seg_path),
137
1
                                    index_meta->index_id(), index_meta->get_index_suffix());
138
1
                            int64_t index_size = 0;
139
1
                            RETURN_IF_ERROR(fs->file_size(index_path, &index_size));
140
1
                            VLOG_DEBUG << "inverted index file:" << index_path
141
0
                                       << " size:" << index_size;
142
1
                            drop_index_size += index_size;
143
1
                        }
144
1
                    }
145
5
                    _dropped_inverted_indexes.push_back(*index_meta);
146
                    // ATTN: DO NOT REMOVE INDEX AFTER OUTPUT_ROWSET_WRITER CREATED.
147
                    // remove dropped index_meta from output rowset tablet schema
148
5
                    output_rs_tablet_schema->remove_index(index_meta->index_id());
149
5
                }
150
151
                // ann index
152
6
                const auto* ann_index = output_rs_tablet_schema->ann_index(column);
153
6
                if (!ann_index) {
154
5
                    continue;
155
5
                }
156
                // Only drop the ann index that matches the requested index_id
157
1
                if (ann_index->index_id() != t_inverted_index.index_id) {
158
0
                    continue;
159
0
                }
160
1
                DCHECK(output_rs_tablet_schema->get_inverted_index_storage_format() !=
161
1
                       InvertedIndexStorageFormatPB::V1);
162
1
                _dropped_inverted_indexes.push_back(*ann_index);
163
                // ATTN: DO NOT REMOVE INDEX AFTER OUTPUT_ROWSET_WRITER CREATED.
164
                // remove dropped index_meta from output rowset tablet schema
165
1
                output_rs_tablet_schema->remove_index(ann_index->index_id());
166
1
            }
167
168
6
            DBUG_EXECUTE_IF("index_builder.update_inverted_index_info.drop_index", {
169
6
                auto indexes_count = DebugPoints::instance()->get_debug_param_or_default<int32_t>(
170
6
                        "index_builder.update_inverted_index_info.drop_index", "indexes_count", 0);
171
6
                if (indexes_count < 0) {
172
6
                    return Status::Error<ErrorCode::INTERNAL_ERROR>(
173
6
                            "indexes count cannot be negative");
174
6
                }
175
6
                auto indexes_size = output_rs_tablet_schema->inverted_indexes().size();
176
6
                if (indexes_count != indexes_size) {
177
6
                    return Status::Error<ErrorCode::INTERNAL_ERROR>(
178
6
                            "indexes count not equal to expected");
179
6
                }
180
6
            })
181
16
        } else {
182
            // base on input rowset's tablet_schema to build
183
            // output rowset's tablet_schema which only add
184
            // the indexes specified in this build index request
185
18
            for (auto t_inverted_index : _alter_inverted_indexes) {
186
18
                TabletIndex index;
187
18
                index.init_from_thrift(t_inverted_index, *input_rs_tablet_schema);
188
18
                auto column_uid = index.col_unique_ids()[0];
189
18
                if (column_uid < 0) {
190
3
                    LOG(WARNING) << "referenced column was missing. "
191
3
                                 << "[column=" << t_inverted_index.columns[0]
192
3
                                 << " referenced_column=" << column_uid << "]";
193
3
                    continue;
194
3
                }
195
15
                const TabletColumn& col = output_rs_tablet_schema->column_by_uid(column_uid);
196
197
                // inverted index
198
15
                auto exist_indexs = output_rs_tablet_schema->inverted_indexs(col);
199
15
                for (const auto& exist_index : exist_indexs) {
200
0
                    if (exist_index->index_id() != index.index_id()) {
201
0
                        if (exist_index->is_same_except_id(&index)) {
202
0
                            LOG(WARNING) << fmt::format(
203
0
                                    "column: {} has a exist inverted index, but the index id not "
204
0
                                    "equal "
205
0
                                    "request's index id, , exist index id: {}, request's index id: "
206
0
                                    "{}, "
207
0
                                    "remove exist index in new output_rs_tablet_schema",
208
0
                                    column_uid, exist_index->index_id(), index.index_id());
209
0
                            without_index_uids.insert(exist_index->index_id());
210
0
                            output_rs_tablet_schema->remove_index(exist_index->index_id());
211
0
                        }
212
0
                    }
213
0
                }
214
215
                // ann index
216
15
                const auto* exist_index = output_rs_tablet_schema->ann_index(col);
217
15
                if (exist_index && exist_index->index_id() != index.index_id()) {
218
0
                    if (exist_index->is_same_except_id(&index)) {
219
0
                        LOG(WARNING) << fmt::format(
220
0
                                "column: {} has a exist ann index, but the index id not "
221
0
                                "equal request's index id, , exist index id: {}, request's index "
222
0
                                "id: {}, remove exist index in new output_rs_tablet_schema",
223
0
                                column_uid, exist_index->index_id(), index.index_id());
224
0
                        without_index_uids.insert(exist_index->index_id());
225
0
                        output_rs_tablet_schema->remove_index(exist_index->index_id());
226
0
                    }
227
0
                }
228
229
15
                output_rs_tablet_schema->append_index(std::move(index));
230
15
            }
231
16
        }
232
        // construct input rowset reader
233
22
        RowsetReaderSharedPtr input_rs_reader;
234
22
        RETURN_IF_ERROR(input_rowset->create_reader(&input_rs_reader));
235
        // construct output rowset writer
236
22
        RowsetWriterContext context;
237
22
        context.version = input_rs_reader->version();
238
22
        context.rowset_state = VISIBLE;
239
22
        context.segments_overlap = input_rowset->rowset_meta()->segments_overlap();
240
22
        context.tablet_schema = output_rs_tablet_schema;
241
22
        context.newest_write_timestamp = input_rs_reader->newest_write_timestamp();
242
22
        auto output_rs_writer = DORIS_TRY(_tablet->create_rowset_writer(context, false));
243
22
        _pending_rs_guards.push_back(_engine.add_pending_rowset(context));
244
245
        // if without_index_uids is not empty, copy _alter_index_ids to it
246
        // else just use _alter_index_ids to avoid copy
247
22
        if (!without_index_uids.empty()) {
248
0
            without_index_uids.insert(_alter_index_ids.begin(), _alter_index_ids.end());
249
0
        }
250
251
        // build output rowset
252
22
        RETURN_IF_ERROR(input_rowset->link_files_to(
253
22
                _tablet->tablet_path(), output_rs_writer->rowset_id(), 0,
254
22
                without_index_uids.empty() ? &_alter_index_ids : &without_index_uids));
255
256
22
        auto input_rowset_meta = input_rowset->rowset_meta();
257
22
        RowsetMetaSharedPtr rowset_meta = std::make_shared<RowsetMeta>();
258
22
        rowset_meta->set_num_rows(input_rowset_meta->num_rows());
259
22
        if (output_rs_tablet_schema->get_inverted_index_storage_format() ==
260
22
            InvertedIndexStorageFormatPB::V1) {
261
4
            if (_is_drop_op) {
262
1
                VLOG_DEBUG << "data_disk_size:" << input_rowset_meta->data_disk_size()
263
0
                           << " total_disk_size:" << input_rowset_meta->total_disk_size()
264
0
                           << " index_disk_size:" << input_rowset_meta->index_disk_size()
265
0
                           << " drop_index_size:" << drop_index_size;
266
1
                rowset_meta->set_total_disk_size(input_rowset_meta->total_disk_size() -
267
1
                                                 drop_index_size);
268
1
                rowset_meta->set_data_disk_size(input_rowset_meta->data_disk_size());
269
1
                rowset_meta->set_index_disk_size(input_rowset_meta->index_disk_size() -
270
1
                                                 drop_index_size);
271
3
            } else {
272
3
                rowset_meta->set_total_disk_size(input_rowset_meta->total_disk_size());
273
3
                rowset_meta->set_data_disk_size(input_rowset_meta->data_disk_size());
274
3
                rowset_meta->set_index_disk_size(input_rowset_meta->index_disk_size());
275
3
            }
276
18
        } else {
277
39
            for (int seg_id = 0; seg_id < num_segments; seg_id++) {
278
21
                auto seg_path = DORIS_TRY(input_rowset->segment_path(seg_id));
279
21
                auto idx_file_reader = std::make_unique<IndexFileReader>(
280
21
                        context.fs(),
281
21
                        std::string {InvertedIndexDescriptor::get_index_file_path_prefix(seg_path)},
282
21
                        output_rs_tablet_schema->get_inverted_index_storage_format(),
283
21
                        InvertedIndexFileInfo(), _tablet->tablet_id());
284
21
                auto st = idx_file_reader->init();
285
21
                DBUG_EXECUTE_IF(
286
21
                        "IndexBuilder::update_inverted_index_info_index_file_reader_init_not_ok", {
287
21
                            st = Status::Error<ErrorCode::INIT_FAILED>(
288
21
                                    "debug point: reader init error");
289
21
                        })
290
21
                if (!st.ok() && !st.is<ErrorCode::INVERTED_INDEX_FILE_NOT_FOUND>()) {
291
0
                    return st;
292
0
                }
293
21
                _index_file_readers.emplace(
294
21
                        std::make_pair(output_rs_writer->rowset_id().to_string(), seg_id),
295
21
                        std::move(idx_file_reader));
296
21
            }
297
18
            rowset_meta->set_total_disk_size(input_rowset_meta->total_disk_size() -
298
18
                                             total_index_size);
299
18
            rowset_meta->set_data_disk_size(input_rowset_meta->data_disk_size());
300
18
            rowset_meta->set_index_disk_size(input_rowset_meta->index_disk_size() -
301
18
                                             total_index_size);
302
18
        }
303
22
        rowset_meta->set_empty(input_rowset_meta->empty());
304
22
        rowset_meta->set_num_segments(input_rowset_meta->num_segments());
305
22
        rowset_meta->set_segments_overlap(input_rowset_meta->segments_overlap());
306
22
        rowset_meta->set_rowset_state(input_rowset_meta->rowset_state());
307
22
        std::vector<KeyBoundsPB> key_bounds;
308
22
        RETURN_IF_ERROR(input_rowset->get_segments_key_bounds(&key_bounds));
309
22
        rowset_meta->set_segments_key_bounds_truncated(
310
22
                input_rowset_meta->is_segments_key_bounds_truncated());
311
        // preserve aggregated layout via the setter so the aggregated flag is not
312
        // clobbered by set_segments_key_bounds's default reset path.
313
22
        rowset_meta->set_segments_key_bounds(
314
22
                key_bounds, input_rowset_meta->is_segments_key_bounds_aggregated());
315
22
        std::vector<uint32_t> num_segment_rows;
316
22
        input_rowset_meta->get_num_segment_rows(&num_segment_rows);
317
22
        rowset_meta->set_num_segment_rows(num_segment_rows);
318
22
        auto output_rowset = output_rs_writer->manual_build(rowset_meta);
319
22
        if (input_rowset_meta->has_delete_predicate()) {
320
0
            output_rowset->rowset_meta()->set_delete_predicate(
321
0
                    input_rowset_meta->delete_predicate());
322
0
        }
323
22
        _output_rowsets.push_back(output_rowset);
324
22
    }
325
326
21
    return Status::OK();
327
21
}
328
329
Status IndexBuilder::handle_single_rowset(RowsetMetaSharedPtr output_rowset_meta,
330
20
                                          std::vector<segment_v2::SegmentSharedPtr>& segments) {
331
20
    bool is_local_rowset = output_rowset_meta->is_local();
332
20
    DBUG_EXECUTE_IF("IndexBuilder::handle_single_rowset_is_local_rowset",
333
20
                    { is_local_rowset = false; })
334
20
    if (!is_local_rowset) [[unlikely]] {
335
        // DCHECK(false) << _tablet->tablet_id() << ' ' << output_rowset_meta->rowset_id();
336
0
        return Status::InternalError("should be local rowset. tablet_id={} rowset_id={}",
337
0
                                     _tablet->tablet_id(),
338
0
                                     output_rowset_meta->rowset_id().to_string());
339
0
    }
340
341
20
    if (_is_drop_op) {
342
6
        const auto& output_rs_tablet_schema = output_rowset_meta->tablet_schema();
343
6
        if (output_rs_tablet_schema->get_inverted_index_storage_format() !=
344
6
            InvertedIndexStorageFormatPB::V1) {
345
5
            const auto& fs = output_rowset_meta->fs();
346
347
5
            const auto& output_rowset_schema = output_rowset_meta->tablet_schema();
348
5
            size_t inverted_index_size = 0;
349
5
            for (auto& seg_ptr : segments) {
350
5
                auto idx_file_reader_iter = _index_file_readers.find(
351
5
                        std::make_pair(output_rowset_meta->rowset_id().to_string(), seg_ptr->id()));
352
5
                DBUG_EXECUTE_IF("IndexBuilder::handle_single_rowset_can_not_find_reader_drop_op",
353
5
                                { idx_file_reader_iter = _index_file_readers.end(); })
354
5
                if (idx_file_reader_iter == _index_file_readers.end()) {
355
0
                    LOG(ERROR) << "idx_file_reader_iter" << output_rowset_meta->rowset_id() << ":"
356
0
                               << seg_ptr->id() << " cannot be found";
357
0
                    continue;
358
0
                }
359
5
                auto dirs = DORIS_TRY(idx_file_reader_iter->second->get_all_directories());
360
361
5
                std::string index_path_prefix {
362
5
                        InvertedIndexDescriptor::get_index_file_path_prefix(local_segment_path(
363
5
                                _tablet->tablet_path(), output_rowset_meta->rowset_id().to_string(),
364
5
                                seg_ptr->id()))};
365
366
5
                std::string index_path =
367
5
                        InvertedIndexDescriptor::get_index_file_path_v2(index_path_prefix);
368
5
                io::FileWriterPtr file_writer;
369
5
                Status st = fs->create_file(index_path, &file_writer);
370
5
                if (!st.ok()) {
371
0
                    LOG(WARNING) << "failed to create writable file. path=" << index_path
372
0
                                 << ", err: " << st;
373
0
                    return st;
374
0
                }
375
5
                auto index_file_writer = std::make_unique<IndexFileWriter>(
376
5
                        fs, std::move(index_path_prefix),
377
5
                        output_rowset_meta->rowset_id().to_string(), seg_ptr->id(),
378
5
                        output_rowset_schema->get_inverted_index_storage_format(),
379
5
                        std::move(file_writer), true /* can_use_ram_dir */, _tablet->tablet_id());
380
5
                RETURN_IF_ERROR(index_file_writer->initialize(dirs));
381
                // create inverted index writer
382
5
                for (auto& index_meta : _dropped_inverted_indexes) {
383
5
                    RETURN_IF_ERROR(index_file_writer->delete_index(&index_meta));
384
5
                }
385
5
                _index_file_writers.emplace(seg_ptr->id(), std::move(index_file_writer));
386
5
            }
387
5
            for (auto&& [seg_id, index_file_writer] : _index_file_writers) {
388
5
                auto st = index_file_writer->begin_close();
389
5
                if (!st.ok()) {
390
0
                    LOG(ERROR) << "close index_file_writer error:" << st;
391
0
                    return st;
392
0
                }
393
5
                inverted_index_size += index_file_writer->get_index_file_total_size();
394
5
            }
395
5
            for (auto&& [seg_id, index_file_writer] : _index_file_writers) {
396
5
                auto st = index_file_writer->finish_close();
397
5
                if (!st.ok()) {
398
0
                    LOG(ERROR) << "wait close index_file_writer error:" << st;
399
0
                    return st;
400
0
                }
401
5
            }
402
5
            _index_file_writers.clear();
403
5
            output_rowset_meta->set_data_disk_size(output_rowset_meta->data_disk_size());
404
5
            output_rowset_meta->set_total_disk_size(output_rowset_meta->total_disk_size() +
405
5
                                                    inverted_index_size);
406
5
            output_rowset_meta->set_index_disk_size(output_rowset_meta->index_disk_size() +
407
5
                                                    inverted_index_size);
408
5
        }
409
6
        LOG(INFO) << "all row nums. source_rows=" << output_rowset_meta->num_rows();
410
6
        return Status::OK();
411
14
    } else {
412
        // create inverted or ann index writer
413
14
        const auto& fs = output_rowset_meta->fs();
414
14
        auto output_rowset_schema = output_rowset_meta->tablet_schema();
415
14
        size_t inverted_index_size = 0;
416
17
        for (auto& seg_ptr : segments) {
417
17
            std::string index_path_prefix {
418
17
                    InvertedIndexDescriptor::get_index_file_path_prefix(local_segment_path(
419
17
                            _tablet->tablet_path(), output_rowset_meta->rowset_id().to_string(),
420
17
                            seg_ptr->id()))};
421
17
            std::vector<ColumnId> return_columns;
422
17
            std::vector<std::pair<int64_t, int64_t>> inverted_index_writer_signs;
423
17
            _olap_data_convertor->reserve(_alter_inverted_indexes.size());
424
425
17
            std::unique_ptr<IndexFileWriter> index_file_writer = nullptr;
426
17
            if (output_rowset_schema->get_inverted_index_storage_format() >=
427
17
                InvertedIndexStorageFormatPB::V2) {
428
14
                auto idx_file_reader_iter = _index_file_readers.find(
429
14
                        std::make_pair(output_rowset_meta->rowset_id().to_string(), seg_ptr->id()));
430
14
                DBUG_EXECUTE_IF("IndexBuilder::handle_single_rowset_can_not_find_reader",
431
14
                                { idx_file_reader_iter = _index_file_readers.end(); })
432
14
                if (idx_file_reader_iter == _index_file_readers.end()) {
433
0
                    LOG(ERROR) << "idx_file_reader_iter" << output_rowset_meta->rowset_id() << ":"
434
0
                               << seg_ptr->id() << " cannot be found";
435
0
                    continue;
436
0
                }
437
14
                std::string index_path =
438
14
                        InvertedIndexDescriptor::get_index_file_path_v2(index_path_prefix);
439
14
                io::FileWriterPtr file_writer;
440
14
                Status st = fs->create_file(index_path, &file_writer);
441
14
                if (!st.ok()) {
442
0
                    LOG(WARNING) << "failed to create writable file. path=" << index_path
443
0
                                 << ", err: " << st;
444
0
                    return st;
445
0
                }
446
14
                auto dirs = DORIS_TRY(idx_file_reader_iter->second->get_all_directories());
447
14
                index_file_writer = std::make_unique<IndexFileWriter>(
448
14
                        fs, index_path_prefix, output_rowset_meta->rowset_id().to_string(),
449
14
                        seg_ptr->id(), output_rowset_schema->get_inverted_index_storage_format(),
450
14
                        std::move(file_writer), true /* can_use_ram_dir */, _tablet->tablet_id());
451
14
                RETURN_IF_ERROR(index_file_writer->initialize(dirs));
452
14
            } else {
453
3
                index_file_writer = std::make_unique<IndexFileWriter>(
454
3
                        fs, index_path_prefix, output_rowset_meta->rowset_id().to_string(),
455
3
                        seg_ptr->id(), output_rowset_schema->get_inverted_index_storage_format(),
456
3
                        nullptr, true /* can_use_ram_dir */, _tablet->tablet_id());
457
3
            }
458
            // create inverted index writer, or ann index writer
459
21
            for (auto inverted_index : _alter_inverted_indexes) {
460
21
                DCHECK(inverted_index.index_type == TIndexType::INVERTED ||
461
21
                       inverted_index.index_type == TIndexType::ANN);
462
21
                DCHECK_EQ(inverted_index.columns.size(), 1);
463
21
                auto index_id = inverted_index.index_id;
464
21
                auto column_name = inverted_index.columns[0];
465
21
                auto column_idx = output_rowset_schema->field_index(column_name);
466
21
                if (column_idx < 0) {
467
4
                    if (inverted_index.__isset.column_unique_ids &&
468
4
                        !inverted_index.column_unique_ids.empty()) {
469
1
                        column_idx = output_rowset_schema->field_index(
470
1
                                inverted_index.column_unique_ids[0]);
471
1
                    }
472
4
                    if (column_idx < 0) {
473
3
                        LOG(WARNING) << "referenced column was missing. "
474
3
                                     << "[column=" << column_name
475
3
                                     << " referenced_column=" << column_idx << "]";
476
3
                        continue;
477
3
                    }
478
4
                }
479
18
                auto column = output_rowset_schema->column(column_idx);
480
                // variant column is not support for building index
481
18
                auto is_support_inverted_index =
482
18
                        IndexColumnWriter::check_support_inverted_index(column);
483
18
                auto is_support_ann_index = IndexColumnWriter::check_support_ann_index(column);
484
18
                DBUG_EXECUTE_IF("IndexBuilder::handle_single_rowset_support_inverted_index",
485
18
                                { is_support_inverted_index = false; })
486
18
                if (!is_support_inverted_index && !is_support_ann_index) {
487
0
                    continue;
488
0
                }
489
18
                DCHECK(output_rowset_schema->has_inverted_index_with_index_id(index_id));
490
18
                _olap_data_convertor->add_column_data_convertor(column);
491
18
                return_columns.emplace_back(column_idx);
492
493
18
                if (inverted_index.index_type == TIndexType::INVERTED) {
494
                    // inverted index
495
17
                    auto index_metas = output_rowset_schema->inverted_indexs(column);
496
17
                    for (const auto& index_meta : index_metas) {
497
17
                        if (index_meta->index_id() != index_id) {
498
0
                            continue;
499
0
                        }
500
17
                        std::unique_ptr<segment_v2::IndexColumnWriter> inverted_index_builder;
501
17
                        try {
502
17
                            RETURN_IF_ERROR(segment_v2::IndexColumnWriter::create(
503
17
                                    &column, &inverted_index_builder, index_file_writer.get(),
504
17
                                    index_meta));
505
17
                            DBUG_EXECUTE_IF(
506
17
                                    "IndexBuilder::handle_single_rowset_index_column_writer_create_"
507
17
                                    "error",
508
17
                                    {
509
17
                                        _CLTHROWA(CL_ERR_IO,
510
17
                                                  "debug point: "
511
17
                                                  "handle_single_rowset_index_column_writer_create_"
512
17
                                                  "error");
513
17
                                    })
514
17
                        } catch (const std::exception& e) {
515
0
                            return Status::Error<ErrorCode::INVERTED_INDEX_CLUCENE_ERROR>(
516
0
                                    "CLuceneError occurred: {}", e.what());
517
0
                        }
518
519
17
                        if (inverted_index_builder) {
520
17
                            auto writer_sign = std::make_pair(seg_ptr->id(), index_id);
521
17
                            _index_column_writers.insert(
522
17
                                    std::make_pair(writer_sign, std::move(inverted_index_builder)));
523
17
                            inverted_index_writer_signs.emplace_back(writer_sign);
524
17
                        }
525
17
                    }
526
17
                } else if (inverted_index.index_type == TIndexType::ANN) {
527
                    // ann index
528
1
                    const auto* index_meta = output_rowset_schema->ann_index(column);
529
1
                    if (index_meta && index_meta->index_id() == index_id) {
530
1
                        std::unique_ptr<segment_v2::IndexColumnWriter> index_writer;
531
1
                        try {
532
1
                            RETURN_IF_ERROR(segment_v2::IndexColumnWriter::create(
533
1
                                    &column, &index_writer, index_file_writer.get(), index_meta));
534
1
                            DBUG_EXECUTE_IF(
535
1
                                    "IndexBuilder::handle_single_rowset_index_column_writer_create_"
536
1
                                    "error",
537
1
                                    {
538
1
                                        _CLTHROWA(CL_ERR_IO,
539
1
                                                  "debug point: "
540
1
                                                  "handle_single_rowset_index_column_writer_create_"
541
1
                                                  "error");
542
1
                                    })
543
1
                        } catch (const std::exception& e) {
544
0
                            return Status::Error<ErrorCode::INVERTED_INDEX_CLUCENE_ERROR>(
545
0
                                    "CLuceneError occurred: {}", e.what());
546
0
                        }
547
548
1
                        if (index_writer) {
549
1
                            auto writer_sign = std::make_pair(seg_ptr->id(), index_id);
550
1
                            _index_column_writers.insert(
551
1
                                    std::make_pair(writer_sign, std::move(index_writer)));
552
1
                            inverted_index_writer_signs.emplace_back(writer_sign);
553
1
                        }
554
1
                    }
555
1
                }
556
18
            }
557
558
            // DO NOT forget index_file_writer for the segment, otherwise, original inverted index will be deleted.
559
17
            _index_file_writers.emplace(seg_ptr->id(), std::move(index_file_writer));
560
17
            if (return_columns.empty()) {
561
                // no columns to read
562
3
                continue;
563
3
            }
564
            // create iterator for each segment
565
14
            StorageReadOptions read_options;
566
14
            OlapReaderStatistics stats;
567
14
            read_options.stats = &stats;
568
14
            read_options.tablet_schema = output_rowset_schema;
569
14
            auto schema = std::make_shared<ReadSchema>(
570
14
                    project_columns_by_ordinal(output_rowset_schema->columns(), return_columns));
571
14
            std::unique_ptr<RowwiseIterator> iter;
572
14
            auto res = seg_ptr->new_iterator(schema, read_options, &iter);
573
14
            DBUG_EXECUTE_IF("IndexBuilder::handle_single_rowset_create_iterator_error", {
574
14
                res = Status::Error<ErrorCode::INTERNAL_ERROR>(
575
14
                        "debug point: handle_single_rowset_create_iterator_error");
576
14
            })
577
14
            if (!res.ok()) {
578
0
                LOG(WARNING) << "failed to create iterator[" << seg_ptr->id()
579
0
                             << "]: " << res.to_string();
580
0
                return Status::Error<ErrorCode::ROWSET_READER_INIT>(res.to_string());
581
0
            }
582
583
14
            auto block = Block::create_unique(schema->create_read_block());
584
28
            while (true) {
585
28
                auto status = iter->next_batch(block.get());
586
28
                DBUG_EXECUTE_IF("IndexBuilder::handle_single_rowset_iterator_next_batch_error", {
587
28
                    status = Status::Error<ErrorCode::SCHEMA_CHANGE_INFO_INVALID>(
588
28
                            "next_batch fault injection");
589
28
                });
590
28
                if (!status.ok()) {
591
14
                    if (status.is<ErrorCode::END_OF_FILE>()) {
592
14
                        break;
593
14
                    }
594
14
                    LOG(WARNING)
595
0
                            << "failed to read next block when schema change for inverted index."
596
0
                            << ", err=" << status.to_string();
597
0
                    return status;
598
14
                }
599
600
                // write inverted index data, or ann index data
601
14
                status = _write_inverted_index_data(output_rowset_schema, iter->data_id(),
602
14
                                                    block.get());
603
14
                DBUG_EXECUTE_IF(
604
14
                        "IndexBuilder::handle_single_rowset_write_inverted_index_data_error", {
605
14
                            status = Status::Error<ErrorCode::INTERNAL_ERROR>(
606
14
                                    "debug point: "
607
14
                                    "handle_single_rowset_write_inverted_index_data_error");
608
14
                        })
609
14
                if (!status.ok()) {
610
0
                    return Status::Error<ErrorCode::SCHEMA_CHANGE_INFO_INVALID>(
611
0
                            "failed to write block.");
612
0
                }
613
14
                block->clear_column_data();
614
14
            }
615
616
            // finish write inverted index, flush data to compound file
617
18
            for (auto& writer_sign : inverted_index_writer_signs) {
618
18
                try {
619
18
                    if (_index_column_writers[writer_sign]) {
620
18
                        RETURN_IF_ERROR(_index_column_writers[writer_sign]->finish());
621
18
                    }
622
18
                    DBUG_EXECUTE_IF("IndexBuilder::handle_single_rowset_index_build_finish_error", {
623
18
                        _CLTHROWA(CL_ERR_IO,
624
18
                                  "debug point: handle_single_rowset_index_build_finish_error");
625
18
                    })
626
18
                } catch (const std::exception& e) {
627
0
                    return Status::Error<ErrorCode::INVERTED_INDEX_CLUCENE_ERROR>(
628
0
                            "CLuceneError occurred: {}", e.what());
629
0
                }
630
18
            }
631
632
14
            _olap_data_convertor->reset();
633
14
        }
634
17
        for (auto&& [seg_id, index_file_writer] : _index_file_writers) {
635
17
            auto st = index_file_writer->begin_close();
636
17
            DBUG_EXECUTE_IF("IndexBuilder::handle_single_rowset_file_writer_close_error", {
637
17
                st = Status::Error<ErrorCode::INVERTED_INDEX_CLUCENE_ERROR>(
638
17
                        "debug point: handle_single_rowset_file_writer_close_error");
639
17
            })
640
17
            if (!st.ok()) {
641
0
                LOG(ERROR) << "close index_file_writer error:" << st;
642
0
                return st;
643
0
            }
644
17
            inverted_index_size += index_file_writer->get_index_file_total_size();
645
17
        }
646
17
        for (auto&& [seg_id, index_file_writer] : _index_file_writers) {
647
17
            auto st = index_file_writer->finish_close();
648
17
            if (!st.ok()) {
649
0
                LOG(ERROR) << "wait close index_file_writer error:" << st;
650
0
                return st;
651
0
            }
652
17
        }
653
14
        _index_column_writers.clear();
654
14
        _index_file_writers.clear();
655
14
        output_rowset_meta->set_data_disk_size(output_rowset_meta->data_disk_size());
656
14
        output_rowset_meta->set_total_disk_size(output_rowset_meta->total_disk_size() +
657
14
                                                inverted_index_size);
658
14
        output_rowset_meta->set_index_disk_size(output_rowset_meta->index_disk_size() +
659
14
                                                inverted_index_size);
660
14
        LOG(INFO) << "all row nums. source_rows=" << output_rowset_meta->num_rows();
661
14
    }
662
663
14
    return Status::OK();
664
20
}
665
666
Status IndexBuilder::_write_inverted_index_data(TabletSchemaSPtr tablet_schema, int64_t segment_idx,
667
14
                                                Block* block) {
668
14
    VLOG_DEBUG << "begin to write inverted/ann index";
669
    // converter block data
670
14
    _olap_data_convertor->set_source_content(block, 0, block->rows());
671
32
    for (auto i = 0; i < _alter_inverted_indexes.size(); ++i) {
672
18
        auto inverted_index = _alter_inverted_indexes[i];
673
18
        auto index_id = inverted_index.index_id;
674
18
        auto column_name = inverted_index.columns[0];
675
18
        auto column_idx = tablet_schema->field_index(column_name);
676
18
        DBUG_EXECUTE_IF("IndexBuilder::_write_inverted_index_data_column_idx_is_negative",
677
18
                        { column_idx = -1; })
678
18
        if (column_idx < 0) {
679
1
            if (!inverted_index.column_unique_ids.empty()) {
680
1
                auto column_unique_id = inverted_index.column_unique_ids[0];
681
1
                column_idx = tablet_schema->field_index(column_unique_id);
682
1
            }
683
1
            if (column_idx < 0) {
684
0
                LOG(WARNING) << "referenced column was missing. "
685
0
                             << "[column=" << column_name << " referenced_column=" << column_idx
686
0
                             << "]";
687
0
                continue;
688
0
            }
689
1
        }
690
18
        const auto& column = tablet_schema->column(column_idx);
691
18
        auto writer_sign = std::make_pair(segment_idx, index_id);
692
18
        auto converted_result = _olap_data_convertor->convert_column_data(i);
693
18
        DBUG_EXECUTE_IF("IndexBuilder::_write_inverted_index_data_convert_column_data_error", {
694
18
            converted_result.first = Status::Error<ErrorCode::INTERNAL_ERROR>(
695
18
                    "debug point: _write_inverted_index_data_convert_column_data_error");
696
18
        })
697
18
        if (converted_result.first != Status::OK()) {
698
0
            LOG(WARNING) << "failed to convert block, errcode: " << converted_result.first;
699
0
            return converted_result.first;
700
0
        }
701
18
        const auto* ptr = (const uint8_t*)converted_result.second->get_data();
702
18
        const auto* null_map = converted_result.second->get_nullmap();
703
18
        if (null_map) {
704
0
            RETURN_IF_ERROR(_add_nullable(column_name, writer_sign, &column, null_map, &ptr,
705
0
                                          block->rows()));
706
18
        } else {
707
18
            RETURN_IF_ERROR(_add_data(column_name, writer_sign, &column, &ptr, block->rows()));
708
18
        }
709
18
    }
710
14
    _olap_data_convertor->clear_source_content();
711
712
14
    return Status::OK();
713
14
}
714
715
Status IndexBuilder::_add_nullable(const std::string& column_name,
716
                                   const std::pair<int64_t, int64_t>& index_writer_sign,
717
                                   const TabletColumn* column, const uint8_t* null_map,
718
0
                                   const uint8_t** ptr, size_t num_rows) {
719
    // TODO: need to process null data for inverted index
720
0
    if (column->type() == FieldType::OLAP_FIELD_TYPE_ARRAY) {
721
0
        DCHECK(column->get_subtype_count() == 1);
722
        // [size, offset_ptr, item_data_ptr, item_nullmap_ptr]
723
0
        const auto* data_ptr = reinterpret_cast<const uint64_t*>(*ptr);
724
        // total number length
725
0
        auto offset_data = *(data_ptr + 1);
726
0
        const auto* offsets_ptr = (const uint8_t*)offset_data;
727
0
        try {
728
0
            auto data = *(data_ptr + 2);
729
0
            auto nested_null_map = *(data_ptr + 3);
730
0
            RETURN_IF_ERROR(_index_column_writers[index_writer_sign]->add_array_values(
731
0
                    field_type_size(column->get_sub_column(0).type()),
732
0
                    reinterpret_cast<const void*>(data),
733
0
                    reinterpret_cast<const uint8_t*>(nested_null_map), offsets_ptr, num_rows));
734
0
            DBUG_EXECUTE_IF("IndexBuilder::_add_nullable_add_array_values_error", {
735
0
                _CLTHROWA(CL_ERR_IO, "debug point: _add_nullable_add_array_values_error");
736
0
            })
737
0
            RETURN_IF_ERROR(
738
0
                    _index_column_writers[index_writer_sign]->add_array_nulls(null_map, num_rows));
739
0
        } catch (const std::exception& e) {
740
0
            return Status::Error<ErrorCode::INVERTED_INDEX_CLUCENE_ERROR>(
741
0
                    "CLuceneError occurred: {}", e.what());
742
0
        }
743
744
0
        return Status::OK();
745
0
    }
746
0
    size_t offset = 0;
747
0
    auto next_run_step = [&]() {
748
0
        size_t step = 1;
749
0
        for (auto i = offset + 1; i < num_rows; ++i) {
750
0
            if (null_map[offset] == null_map[i]) {
751
0
                step++;
752
0
            } else {
753
0
                break;
754
0
            }
755
0
        }
756
0
        return step;
757
0
    };
758
0
    try {
759
0
        do {
760
0
            auto step = next_run_step();
761
0
            if (null_map[offset]) {
762
0
                RETURN_IF_ERROR(_index_column_writers[index_writer_sign]->add_nulls(
763
0
                        static_cast<uint32_t>(step)));
764
0
            } else {
765
0
                RETURN_IF_ERROR(_index_column_writers[index_writer_sign]->add_values(column_name,
766
0
                                                                                     *ptr, step));
767
0
            }
768
0
            *ptr += field_type_size(column->type()) * step;
769
0
            offset += step;
770
0
            DBUG_EXECUTE_IF("IndexBuilder::_add_nullable_throw_exception",
771
0
                            { _CLTHROWA(CL_ERR_IO, "debug point: _add_nullable_throw_exception"); })
772
0
        } while (offset < num_rows);
773
0
    } catch (const std::exception& e) {
774
0
        return Status::Error<ErrorCode::INVERTED_INDEX_CLUCENE_ERROR>("CLuceneError occurred: {}",
775
0
                                                                      e.what());
776
0
    }
777
778
0
    return Status::OK();
779
0
}
780
781
Status IndexBuilder::_add_data(const std::string& column_name,
782
                               const std::pair<int64_t, int64_t>& index_writer_sign,
783
18
                               const TabletColumn* column, const uint8_t** ptr, size_t num_rows) {
784
18
    try {
785
18
        if (column->type() == FieldType::OLAP_FIELD_TYPE_ARRAY) {
786
2
            DCHECK(column->get_subtype_count() == 1);
787
            // [size, offset_ptr, item_data_ptr, item_nullmap_ptr]
788
2
            const auto* data_ptr = reinterpret_cast<const uint64_t*>(*ptr);
789
            // total number length
790
2
            auto element_cnt = size_t((unsigned long)(*data_ptr));
791
2
            auto offset_data = *(data_ptr + 1);
792
2
            const auto* offsets_ptr = (const uint8_t*)offset_data;
793
2
            if (element_cnt > 0) {
794
2
                auto data = *(data_ptr + 2);
795
2
                auto nested_null_map = *(data_ptr + 3);
796
2
                RETURN_IF_ERROR(_index_column_writers[index_writer_sign]->add_array_values(
797
2
                        field_type_size(column->get_sub_column(0).type()),
798
2
                        reinterpret_cast<const void*>(data),
799
2
                        reinterpret_cast<const uint8_t*>(nested_null_map), offsets_ptr, num_rows));
800
2
            }
801
16
        } else {
802
16
            RETURN_IF_ERROR(_index_column_writers[index_writer_sign]->add_values(column_name, *ptr,
803
16
                                                                                 num_rows));
804
16
        }
805
18
        DBUG_EXECUTE_IF("IndexBuilder::_add_data_throw_exception",
806
18
                        { _CLTHROWA(CL_ERR_IO, "debug point: _add_data_throw_exception"); })
807
18
    } catch (const std::exception& e) {
808
0
        return Status::Error<ErrorCode::INVERTED_INDEX_CLUCENE_ERROR>("CLuceneError occurred: {}",
809
0
                                                                      e.what());
810
0
    }
811
812
18
    return Status::OK();
813
18
}
814
815
20
Status IndexBuilder::handle_inverted_index_data() {
816
20
    LOG(INFO) << "begin to handle_inverted_index_data";
817
20
    DCHECK(_input_rowsets.size() == _output_rowsets.size());
818
21
    for (auto& _output_rowset : _output_rowsets) {
819
21
        SegmentCacheHandle segment_cache_handle;
820
21
        RETURN_IF_ERROR(SegmentLoader::instance()->load_segments(
821
21
                std::static_pointer_cast<BetaRowset>(_output_rowset), &segment_cache_handle));
822
21
        auto output_rowset_meta = _output_rowset->rowset_meta();
823
21
        auto& segments = segment_cache_handle.get_segments();
824
21
        RETURN_IF_ERROR(handle_single_rowset(output_rowset_meta, segments));
825
21
    }
826
19
    return Status::OK();
827
20
}
828
829
24
Status IndexBuilder::do_build_inverted_index() {
830
24
    LOG(INFO) << "begin to do_build_inverted_index, tablet=" << _tablet->tablet_id()
831
24
              << ", is_drop_op=" << _is_drop_op;
832
24
    DBUG_EXECUTE_IF("IndexBuilder::do_build_inverted_index_alter_inverted_indexes_empty",
833
24
                    { _alter_inverted_indexes.clear(); })
834
24
    if (_alter_inverted_indexes.empty()) {
835
0
        return Status::OK();
836
0
    }
837
838
24
    static constexpr long TRY_LOCK_TIMEOUT = 30;
839
24
    std::unique_lock schema_change_lock(_tablet->get_schema_change_lock(), std::defer_lock);
840
24
    bool owns_lock = schema_change_lock.try_lock_for(std::chrono::seconds(TRY_LOCK_TIMEOUT));
841
842
24
    if (!owns_lock) {
843
0
        return Status::ObtainLockFailed(
844
0
                "try schema_change_lock failed. There might be schema change or cooldown running "
845
0
                "on "
846
0
                "tablet={} ",
847
0
                _tablet->tablet_id());
848
0
    }
849
    // Check executing serially with compaction task.
850
24
    std::unique_lock<std::mutex> base_compaction_lock(_tablet->get_base_compaction_lock(),
851
24
                                                      std::try_to_lock);
852
24
    if (!base_compaction_lock.owns_lock()) {
853
0
        return Status::ObtainLockFailed("try base_compaction_lock failed. tablet={} ",
854
0
                                        _tablet->tablet_id());
855
0
    }
856
24
    std::unique_lock<std::mutex> cumu_compaction_lock(_tablet->get_cumulative_compaction_lock(),
857
24
                                                      std::try_to_lock);
858
24
    if (!cumu_compaction_lock.owns_lock()) {
859
0
        return Status::ObtainLockFailed("try cumu_compaction_lock failed. tablet={}",
860
0
                                        _tablet->tablet_id());
861
0
    }
862
863
24
    std::unique_lock<std::mutex> cold_compaction_lock(_tablet->get_cold_compaction_lock(),
864
24
                                                      std::try_to_lock);
865
24
    if (!cold_compaction_lock.owns_lock()) {
866
0
        return Status::ObtainLockFailed("try cold_compaction_lock failed. tablet={}",
867
0
                                        _tablet->tablet_id());
868
0
    }
869
870
24
    std::unique_lock<std::mutex> build_inverted_index_lock(_tablet->get_build_inverted_index_lock(),
871
24
                                                           std::try_to_lock);
872
24
    if (!build_inverted_index_lock.owns_lock()) {
873
0
        return Status::ObtainLockFailed("failed to obtain build inverted index lock. tablet={}",
874
0
                                        _tablet->tablet_id());
875
0
    }
876
877
24
    std::shared_lock migration_rlock(_tablet->get_migration_lock(), std::try_to_lock);
878
24
    if (!migration_rlock.owns_lock()) {
879
0
        return Status::ObtainLockFailed("got migration_rlock failed. tablet={}",
880
0
                                        _tablet->tablet_id());
881
0
    }
882
883
24
    DCHECK(!_alter_index_ids.empty());
884
24
    _input_rowsets =
885
24
            _tablet->pick_candidate_rowsets_to_build_inverted_index(_alter_index_ids, _is_drop_op);
886
24
    if (_input_rowsets.empty()) {
887
1
        LOG(INFO) << "_input_rowsets is empty";
888
1
        return Status::OK();
889
1
    }
890
891
23
    auto st = update_inverted_index_info();
892
23
    if (!st.ok()) {
893
3
        LOG(WARNING) << "failed to update_inverted_index_info. "
894
3
                     << "tablet=" << _tablet->tablet_id() << ", error=" << st;
895
3
        gc_output_rowset();
896
3
        return st;
897
3
    }
898
899
    // create inverted index file for output rowset
900
20
    st = handle_inverted_index_data();
901
20
    if (!st.ok()) {
902
1
        LOG(WARNING) << "failed to handle_inverted_index_data. "
903
1
                     << "tablet=" << _tablet->tablet_id() << ", error=" << st;
904
1
        gc_output_rowset();
905
1
        return st;
906
1
    }
907
908
    // modify rowsets in memory
909
19
    st = modify_rowsets();
910
19
    DBUG_EXECUTE_IF("IndexBuilder::do_build_inverted_index_modify_rowsets_status_error", {
911
19
        st = Status::Error<ErrorCode::DELETE_VERSION_ERROR>(
912
19
                "debug point: do_build_inverted_index_modify_rowsets_status_error");
913
19
    })
914
19
    if (!st.ok()) {
915
0
        LOG(WARNING) << "failed to modify rowsets in memory. "
916
0
                     << "tablet=" << _tablet->tablet_id() << ", error=" << st;
917
0
        gc_output_rowset();
918
0
        return st;
919
0
    }
920
19
    return Status::OK();
921
19
}
922
923
19
Status IndexBuilder::modify_rowsets(const Merger::Statistics* stats) {
924
19
    DCHECK(std::ranges::all_of(
925
19
            _output_rowsets.begin(), _output_rowsets.end(), [&engine = _engine](auto&& rs) {
926
19
                if (engine.check_rowset_id_in_unused_rowsets(rs->rowset_id())) {
927
19
                    LOG(ERROR) << "output rowset: " << rs->rowset_id() << " in unused rowsets";
928
19
                    return false;
929
19
                }
930
19
                return true;
931
19
            }));
932
933
19
    if (_tablet->keys_type() == KeysType::UNIQUE_KEYS &&
934
19
        _tablet->enable_unique_key_merge_on_write()) {
935
1
        std::lock_guard<std::mutex> rowset_update_wlock(_tablet->get_rowset_update_lock());
936
1
        std::lock_guard meta_wlock(_tablet->get_header_lock());
937
1
        SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD);
938
1
        DeleteBitmapPtr delete_bitmap = std::make_shared<DeleteBitmap>(_tablet->tablet_id());
939
2
        for (auto i = 0; i < _input_rowsets.size(); ++i) {
940
1
            RowsetId input_rowset_id = _input_rowsets[i]->rowset_id();
941
1
            RowsetId output_rowset_id = _output_rowsets[i]->rowset_id();
942
1
            for (const auto& [k, v] : _tablet->tablet_meta()->delete_bitmap().delete_bitmap) {
943
0
                RowsetId rs_id = std::get<0>(k);
944
0
                if (rs_id == input_rowset_id) {
945
0
                    DeleteBitmap::BitmapKey output_rs_key = {output_rowset_id, std::get<1>(k),
946
0
                                                             std::get<2>(k)};
947
0
                    auto res = delete_bitmap->set(output_rs_key, v);
948
0
                    DCHECK(res > 0) << "delete_bitmap set failed, res=" << res;
949
0
                }
950
0
            }
951
1
        }
952
1
        _tablet->tablet_meta()->delete_bitmap().merge(*delete_bitmap);
953
954
        // modify_rowsets will remove the delete_bitmap for input rowsets,
955
        // should call it after merge delete_bitmap
956
1
        RETURN_IF_ERROR(_tablet->modify_rowsets(_output_rowsets, _input_rowsets, true));
957
18
    } else {
958
18
        std::lock_guard wrlock(_tablet->get_header_lock());
959
18
        RETURN_IF_ERROR(_tablet->modify_rowsets(_output_rowsets, _input_rowsets, true));
960
18
    }
961
962
#ifndef BE_TEST
963
    {
964
        std::shared_lock rlock(_tablet->get_header_lock());
965
        _tablet->save_meta();
966
    }
967
#endif
968
19
    return Status::OK();
969
19
}
970
971
4
void IndexBuilder::gc_output_rowset() {
972
4
    for (auto&& output_rowset : _output_rowsets) {
973
2
        auto is_local_rowset = output_rowset->is_local();
974
2
        DBUG_EXECUTE_IF("IndexBuilder::gc_output_rowset_is_local_rowset",
975
2
                        { is_local_rowset = false; })
976
2
        if (!is_local_rowset) {
977
0
            _tablet->record_unused_remote_rowset(output_rowset->rowset_id(),
978
0
                                                 output_rowset->rowset_meta()->resource_id(),
979
0
                                                 output_rowset->num_segments());
980
0
            return;
981
0
        }
982
2
        _engine.add_unused_rowset(output_rowset);
983
2
    }
984
4
}
985
986
} // namespace doris