Coverage Report

Created: 2026-08-02 13:57

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/storage/index/index_file_writer.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/index/index_file_writer.h"
19
20
#include <glog/logging.h>
21
22
#include <atomic>
23
#include <filesystem>
24
25
#include "common/status.h"
26
#include "io/fs/packed_file_writer.h"
27
#include "io/fs/s3_file_writer.h"
28
#include "io/fs/stream_sink_file_writer.h"
29
#include "storage/index/ann/ann_index_files.h"
30
#include "storage/index/index_file_reader.h"
31
#include "storage/index/index_storage_format_v1.h"
32
#include "storage/index/index_storage_format_v2.h"
33
#include "storage/index/index_writer.h"
34
#include "storage/index/inverted/inverted_index_compound_reader.h"
35
#include "storage/index/inverted/inverted_index_desc.h"
36
#include "storage/index/inverted/inverted_index_fs_directory.h"
37
#include "storage/index/inverted/inverted_index_reader.h"
38
#include "storage/tablet/tablet_schema.h"
39
40
namespace doris::segment_v2 {
41
42
IndexFileWriter::IndexFileWriter(io::FileSystemSPtr fs, std::string index_path_prefix,
43
                                 std::string rowset_id, int64_t seg_id,
44
                                 InvertedIndexStorageFormatPB storage_format,
45
                                 io::FileWriterPtr file_writer, bool can_use_ram_dir,
46
                                 int64_t tablet_id)
47
531
        : _fs(std::move(fs)),
48
531
          _index_path_prefix(std::move(index_path_prefix)),
49
531
          _rowset_id(std::move(rowset_id)),
50
531
          _seg_id(seg_id),
51
531
          _storage_format(storage_format),
52
531
          _local_fs(io::global_local_filesystem()),
53
531
          _idx_v2_writer(std::move(file_writer)),
54
531
          _can_use_ram_dir(can_use_ram_dir),
55
531
          _tablet_id(tablet_id) {
56
531
    auto tmp_file_dir = ExecEnv::GetInstance()->get_tmp_file_dirs()->get_tmp_file_dir();
57
531
    _tmp_dir = tmp_file_dir.native();
58
531
    if (_storage_format == InvertedIndexStorageFormatPB::V1) {
59
35
        _index_storage_format = std::make_unique<IndexStorageFormatV1>(this);
60
496
    } else {
61
496
        _index_storage_format = std::make_unique<IndexStorageFormatV2>(this);
62
496
    }
63
531
}
64
65
20
Status IndexFileWriter::initialize(InvertedIndexDirectoryMap& indices_dirs) {
66
20
    _indices_dirs = std::move(indices_dirs);
67
20
    return Status::OK();
68
20
}
69
70
Status IndexFileWriter::_insert_directory_into_map(int64_t index_id,
71
                                                   const std::string& index_suffix,
72
2.85k
                                                   std::shared_ptr<DorisFSDirectory> dir) {
73
2.85k
    auto key = std::make_pair(index_id, index_suffix);
74
2.85k
    auto [it, inserted] = _indices_dirs.emplace(key, std::move(dir));
75
2.85k
    if (!inserted) {
76
1
        LOG(ERROR) << "IndexFileWriter::open attempted to insert a duplicate key: (" << key.first
77
1
                   << ", " << key.second << ")";
78
1
        LOG(ERROR) << "Directories already in map: ";
79
1
        for (const auto& entry : _indices_dirs) {
80
1
            LOG(ERROR) << "Key: (" << entry.first.first << ", " << entry.first.second << ")";
81
1
        }
82
1
        return Status::InternalError("IndexFileWriter::open attempted to insert a duplicate dir");
83
1
    }
84
2.85k
    return Status::OK();
85
2.85k
}
86
87
2.84k
Result<std::shared_ptr<DorisFSDirectory>> IndexFileWriter::open(const TabletIndex* index_meta) {
88
2.84k
    auto local_fs_index_path = InvertedIndexDescriptor::get_temporary_index_path(
89
2.84k
            _tmp_dir, _rowset_id, _seg_id, index_meta->index_id(), index_meta->get_index_suffix());
90
2.84k
    auto dir = std::shared_ptr<DorisFSDirectory>(DorisFSDirectoryFactory::getDirectory(
91
2.84k
            _local_fs, local_fs_index_path.c_str(), _can_use_ram_dir));
92
2.84k
    auto st =
93
2.84k
            _insert_directory_into_map(index_meta->index_id(), index_meta->get_index_suffix(), dir);
94
2.84k
    if (!st.ok()) {
95
0
        return ResultError(st);
96
0
    }
97
98
2.84k
    return dir;
99
2.84k
}
100
101
11
Status IndexFileWriter::delete_index(const TabletIndex* index_meta) {
102
11
    DBUG_EXECUTE_IF("IndexFileWriter::delete_index_index_meta_nullptr", { index_meta = nullptr; });
103
11
    if (!index_meta) {
104
1
        return Status::Error<ErrorCode::INVALID_ARGUMENT>("Index metadata is null.");
105
1
    }
106
107
10
    auto index_id = index_meta->index_id();
108
10
    const auto& index_suffix = index_meta->get_index_suffix();
109
110
    // Check if the specified index exists
111
10
    auto index_it = _indices_dirs.find(std::make_pair(index_id, index_suffix));
112
10
    DBUG_EXECUTE_IF("IndexFileWriter::delete_index_indices_dirs_reach_end",
113
10
                    { index_it = _indices_dirs.end(); })
114
10
    if (index_it == _indices_dirs.end()) {
115
5
        std::ostringstream errMsg;
116
5
        errMsg << "No inverted index with id " << index_id << " and suffix " << index_suffix
117
5
               << " found.";
118
5
        LOG(WARNING) << errMsg.str();
119
5
        return Status::OK();
120
5
    }
121
122
5
    _indices_dirs.erase(index_it);
123
5
    return Status::OK();
124
10
}
125
126
8
Status IndexFileWriter::add_into_searcher_cache() {
127
8
    auto index_file_reader = std::make_unique<IndexFileReader>(
128
8
            _fs, _index_path_prefix, _storage_format, InvertedIndexFileInfo(), _tablet_id);
129
8
    auto st = index_file_reader->init();
130
8
    if (!st.ok()) {
131
1
        if (dynamic_cast<io::StreamSinkFileWriter*>(_idx_v2_writer.get()) != nullptr) {
132
            // StreamSinkFileWriter not found file is normal.
133
0
            return Status::OK();
134
0
        }
135
1
        if (dynamic_cast<io::PackedFileWriter*>(_idx_v2_writer.get()) != nullptr) {
136
            // PackedFileWriter: file may be merged, skip cache for now.
137
            // The cache will be populated on first read.
138
0
            return Status::OK();
139
0
        }
140
1
        LOG(WARNING) << "IndexFileWriter::add_into_searcher_cache for " << _index_path_prefix
141
1
                     << ", error " << st.msg();
142
1
        return st;
143
1
    }
144
7
    for (const auto& entry : _indices_dirs) {
145
7
        auto index_meta = entry.first;
146
7
        auto dir = DORIS_TRY(index_file_reader->_open(index_meta.first, index_meta.second));
147
7
        std::vector<std::string> file_names;
148
7
        dir->list(&file_names);
149
        // Skip ANN indexes – they use FAISS files (ann.faiss, ann.ivfdata) instead of
150
        // CLucene segments, so building an inverted-index searcher would fail.
151
        // HNSW/IVF produces 1 file (ann.faiss); IVF_ON_DISK produces 2 (ann.faiss + ann.ivfdata).
152
7
        bool is_ann_index =
153
7
                std::any_of(file_names.begin(), file_names.end(), [](const std::string& f) {
154
7
                    return f == faiss_index_fila_name || f == faiss_ivfdata_file_name;
155
7
                });
156
7
        if (is_ann_index) {
157
0
            continue;
158
0
        }
159
7
        auto index_file_key = InvertedIndexDescriptor::get_index_file_cache_key(
160
7
                _index_path_prefix, index_meta.first, index_meta.second);
161
7
        InvertedIndexSearcherCache::CacheKey searcher_cache_key(index_file_key);
162
7
        InvertedIndexCacheHandle inverted_index_cache_handle;
163
7
        if (InvertedIndexSearcherCache::instance()->lookup(searcher_cache_key,
164
7
                                                           &inverted_index_cache_handle)) {
165
1
            st = InvertedIndexSearcherCache::instance()->erase(searcher_cache_key.index_file_path);
166
1
            if (!st.ok()) {
167
0
                LOG(WARNING) << "IndexFileWriter::add_into_searcher_cache for "
168
0
                             << _index_path_prefix << ", error " << st.msg();
169
0
            }
170
1
        }
171
7
        IndexSearcherPtr searcher;
172
7
        size_t reader_size = 0;
173
7
        auto index_searcher_builder = DORIS_TRY(_construct_index_searcher_builder(dir.get()));
174
7
        RETURN_IF_ERROR(InvertedIndexReader::create_index_searcher(
175
7
                index_searcher_builder.get(), dir.get(), &searcher, reader_size));
176
7
        auto* cache_value = new InvertedIndexSearcherCache::CacheValue(std::move(searcher),
177
7
                                                                       reader_size, UnixMillis());
178
7
        InvertedIndexSearcherCache::instance()->insert(searcher_cache_key, cache_value);
179
7
    }
180
7
    return Status::OK();
181
7
}
182
183
Result<std::unique_ptr<IndexSearcherBuilder>> IndexFileWriter::_construct_index_searcher_builder(
184
0
        const DorisCompoundReader* dir) {
185
0
    std::vector<std::string> files;
186
0
    dir->list(&files);
187
0
    auto reader_type = InvertedIndexReaderType::FULLTEXT;
188
0
    bool found_bkd = std::any_of(files.begin(), files.end(), [](const std::string& file) {
189
0
        return file == InvertedIndexDescriptor::get_temporary_bkd_index_data_file_name();
190
0
    });
191
0
    if (found_bkd) {
192
0
        reader_type = InvertedIndexReaderType::BKD;
193
0
    }
194
0
    return IndexSearcherBuilder::create_index_searcher_builder(reader_type);
195
0
}
196
197
476
Status IndexFileWriter::begin_close() {
198
476
    DCHECK(!_closed) << debug_string();
199
476
    _closed = true;
200
476
    if (_indices_dirs.empty()) {
201
        // An empty file must still be created even if there are no indexes to write
202
14
        if (dynamic_cast<io::StreamSinkFileWriter*>(_idx_v2_writer.get()) != nullptr ||
203
14
            dynamic_cast<io::S3FileWriter*>(_idx_v2_writer.get()) != nullptr ||
204
14
            dynamic_cast<io::PackedFileWriter*>(_idx_v2_writer.get()) != nullptr) {
205
2
            return _idx_v2_writer->close(true);
206
2
        }
207
12
        return Status::OK();
208
14
    }
209
462
    DBUG_EXECUTE_IF("inverted_index_storage_format_must_be_v2", {
210
462
        if (_storage_format != InvertedIndexStorageFormatPB::V2) {
211
462
            return Status::Error<ErrorCode::INVERTED_INDEX_CLUCENE_ERROR>(
212
462
                    "IndexFileWriter::close fault injection:inverted index storage format "
213
462
                    "must be v2");
214
462
        }
215
462
    })
216
462
    try {
217
462
        RETURN_IF_ERROR(_index_storage_format->write());
218
2.83k
        for (const auto& entry : _indices_dirs) {
219
2.83k
            const auto& dir = entry.second;
220
            // delete index path, which contains separated inverted index files
221
2.83k
            if (std::strcmp(dir->getObjectName(), "DorisFSDirectory") == 0) {
222
90
                auto* compound_dir = static_cast<DorisFSDirectory*>(dir.get());
223
90
                compound_dir->deleteDirectory();
224
90
            }
225
2.83k
        }
226
455
    } catch (CLuceneError& err) {
227
0
        if (_storage_format == InvertedIndexStorageFormatPB::V1) {
228
0
            return Status::Error<ErrorCode::INVERTED_INDEX_CLUCENE_ERROR>(
229
0
                    "CLuceneError occur when close, error msg: {}", err.what());
230
0
        } else {
231
0
            return Status::Error<ErrorCode::INVERTED_INDEX_CLUCENE_ERROR>(
232
0
                    "CLuceneError occur when close idx file {}, error msg: {}",
233
0
                    InvertedIndexDescriptor::get_index_file_path_v2(_index_path_prefix),
234
0
                    err.what());
235
0
        }
236
0
    }
237
455
    return Status::OK();
238
462
}
239
240
461
Status IndexFileWriter::finish_close() {
241
461
    DCHECK(_closed) << debug_string();
242
461
    if (_indices_dirs.empty()) {
243
        // An empty file must still be created even if there are no indexes to write
244
13
        if (dynamic_cast<io::StreamSinkFileWriter*>(_idx_v2_writer.get()) != nullptr ||
245
13
            dynamic_cast<io::S3FileWriter*>(_idx_v2_writer.get()) != nullptr ||
246
13
            dynamic_cast<io::PackedFileWriter*>(_idx_v2_writer.get()) != nullptr) {
247
2
            return _idx_v2_writer->close(false);
248
2
        }
249
11
        return Status::OK();
250
13
    }
251
448
    if (_idx_v2_writer != nullptr && _idx_v2_writer->state() != io::FileWriter::State::CLOSED) {
252
426
        RETURN_IF_ERROR(_idx_v2_writer->close(false));
253
426
    }
254
255
448
    Status st = Status::OK();
256
448
    if (config::enable_write_index_searcher_cache) {
257
7
        st = add_into_searcher_cache();
258
7
    }
259
448
    _indices_dirs.clear();
260
448
    return st;
261
448
}
262
263
3
std::vector<std::string> IndexFileWriter::get_index_file_names() const {
264
3
    std::vector<std::string> file_names;
265
3
    if (_storage_format == InvertedIndexStorageFormatPB::V1) {
266
2
        if (_closed && _file_info.index_info_size() > 0) {
267
1
            for (const auto& index_info : _file_info.index_info()) {
268
1
                file_names.emplace_back(InvertedIndexDescriptor::get_index_file_name_v1(
269
1
                        _rowset_id, _seg_id, index_info.index_id(), index_info.index_suffix()));
270
1
            }
271
1
        } else {
272
2
            for (const auto& [index_info, _] : _indices_dirs) {
273
2
                file_names.emplace_back(InvertedIndexDescriptor::get_index_file_name_v1(
274
2
                        _rowset_id, _seg_id, index_info.first, index_info.second));
275
2
            }
276
1
        }
277
2
    } else {
278
1
        file_names.emplace_back(
279
1
                InvertedIndexDescriptor::get_index_file_name_v2(_rowset_id, _seg_id));
280
1
    }
281
3
    return file_names;
282
3
}
283
284
0
std::string IndexFileWriter::debug_string() const {
285
0
    std::stringstream indices_dirs;
286
0
    for (const auto& [index, dir] : _indices_dirs) {
287
0
        indices_dirs << "index id is: " << index.first << " , index suffix is: " << index.second
288
0
                     << " , index dir is: " << dir->toString();
289
0
    }
290
0
    return fmt::format(
291
0
            "inverted index file writer debug string: index storage format is: {}, index path "
292
0
            "prefix is: {}, rowset id is: {}, seg id is: {}, closed is: {}, total file size "
293
0
            "is: {}, index dirs is: {}",
294
0
            _storage_format, _index_path_prefix, _rowset_id, _seg_id, _closed, _total_file_size,
295
0
            indices_dirs.str());
296
0
}
297
298
} // namespace doris::segment_v2