Coverage Report

Created: 2026-08-26 22:18

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/storage/index/index_file_writer.h
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
#pragma once
19
20
// CLucene is third-party code and is not clean under -Wconversion (which
21
// -Wshorten-64-to-32 belongs to). Whether its first expansion lands inside
22
// someone else's suppressed region depends on include order, so suppress it
23
// deliberately here (same pattern as inverted_index_common_impl.h).
24
#ifdef __clang__
25
#pragma clang diagnostic push
26
#pragma clang diagnostic ignored "-Wconversion"
27
#endif
28
#include <CLucene.h> // IWYU pragma: keep
29
#include <CLucene/store/IndexInput.h>
30
#ifdef __clang__
31
#pragma clang diagnostic pop
32
#endif
33
#include <gen_cpp/olap_common.pb.h>
34
#include <gen_cpp/olap_file.pb.h>
35
36
#include <optional>
37
#include <string>
38
#include <utility>
39
#include <vector>
40
41
#include "common/be_mock_util.h"
42
#include "io/fs/file_system.h"
43
#include "io/fs/file_writer.h"
44
#include "io/fs/local_file_system.h"
45
#include "storage/index/index_storage_format.h"
46
#include "storage/index/inverted/common_grams/common_grams_segment_metadata.h"
47
#include "storage/index/inverted/inverted_index_common.h"
48
#include "storage/index/inverted/inverted_index_compound_reader.h"
49
#include "storage/index/inverted/inverted_index_searcher.h"
50
#include "storage/index/snii/format/format_constants.h"
51
#include "storage/index/snii/snii_doris_adapter.h"
52
#include "storage/index/snii/writer/snii_compound_writer.h"
53
54
namespace doris::snii::writer {
55
class MemoryReporter;
56
class SpimiTermBuffer;
57
class SniiCompoundWriter;
58
} // namespace doris::snii::writer
59
60
namespace doris {
61
class TabletIndex;
62
63
namespace segment_v2 {
64
class DorisFSDirectory;
65
namespace snii_doris {
66
class DorisSniiFileWriter;
67
} // namespace snii_doris
68
69
using InvertedIndexDirectoryMap =
70
        std::map<std::pair<int64_t, std::string>, std::shared_ptr<lucene::store::Directory>>;
71
72
class IndexFileWriter;
73
using IndexFileWriterPtr = std::unique_ptr<IndexFileWriter>;
74
75
class IndexFileWriter {
76
public:
77
    IndexFileWriter(io::FileSystemSPtr fs, std::string index_path_prefix, std::string rowset_id,
78
                    int64_t seg_id, InvertedIndexStorageFormatPB storage_format,
79
                    io::FileWriterPtr file_writer = nullptr, bool can_use_ram_dir = true,
80
                    int64_t tablet_id = -1);
81
712
    virtual ~IndexFileWriter() = default;
82
83
    MOCK_FUNCTION Result<std::shared_ptr<DorisFSDirectory>> open(const TabletIndex* index_meta);
84
    // The directory an ANN index is built into. Separate from open() because the
85
    // two formats stage ANN output in different places: V1/V2 hand faiss the same
86
    // CLucene filesystem directory every other index gets, while SNII hands it a
87
    // memory-backed staging directory whose bytes begin_close() seals into a blob
88
    // logical index. Callers only ever write through it, so the return type is
89
    // the lucene::store::Directory base -- widening open() itself would push that
90
    // base type onto the CLucene inverted writer and index_tool, which genuinely
91
    // need the DorisFSDirectory subclass.
92
    Result<std::shared_ptr<lucene::store::Directory>> open_ann_directory(
93
            const TabletIndex* index_meta);
94
    // SNII only: drops the staging directory of one ANN index whose serialization
95
    // failed. Its sub-files unlink themselves once their last owner is gone, and
96
    // the producer releases its own reference alongside this call -- without that
97
    // a failed save keeps an ANN-sized file and its descriptor on the temp
98
    // filesystem until this writer is destroyed, which for a rowset build is not
99
    // until every other segment has been written. V1/V2 keep their directory: its
100
    // files ARE the index output, and begin_close() is what removes them.
101
    void discard_ann_staging_directory(const TabletIndex* index_meta);
102
    // SNII only: drops the staging of EVERY index on this writer because the
103
    // segment they belong to is being abandoned. Layered above the per-index
104
    // discard, not a duplicate of it: that one fires the instant one ANN
105
    // serialization fails, while this one covers a segment that failed AFTER its
106
    // indexes staged successfully, when nothing else will ever seal them.
107
    void abandon_snii_staging();
108
    // Write-path facts for one SNII index flush.
109
    struct SniiAddIndexOptions {
110
        // This flush serves a stream/broker load (DataWriteType::TYPE_DIRECT):
111
        // the prx region compresses at snii_prx_zstd_level_direct_load;
112
        // compaction / schema change / ADD INDEX keep snii_prx_zstd_level.
113
        bool is_direct_load = false;
114
        // Present only for a CommonGrams writer that has a complete immutable
115
        // capability identity. These are semantic BM25 inputs; physical TTF is
116
        // still derived from every emitted unigram and gram posting.
117
        std::vector<uint8_t> encoded_norms;
118
        std::optional<inverted_index::CommonGramsSegmentMetadata> common_grams_metadata;
119
        snii::format::CommonGramsPostingPolicy common_grams_posting_policy =
120
                snii::format::CommonGramsPostingPolicy::kNone;
121
    };
122
    Status add_snii_index(const TabletIndex* index_meta, uint32_t doc_count,
123
                          std::vector<uint32_t> null_docids,
124
                          doris::snii::writer::SpimiTermBuffer* const term_buffer,
125
                          doris::snii::format::IndexConfig index_config,
126
                          SniiAddIndexOptions options,
127
                          doris::snii::writer::MemoryReporter* const mem_reporter);
128
    // T2.2 compaction index merge fast path: begins a STREAMED SNII index
129
    // session on this compound. Unlike add_snii_index (which drains a SPIMI
130
    // term buffer), the caller pushes pre-merged, lexicographically sorted
131
    // terms through *session and seals the index with (*session)->finish().
132
    // Write parameters resolve through the SAME helper as add_snii_index
133
    // (write_freq / zstd levels / dict block size), always at the COMPACTION
134
    // prx tier (a merge is never a direct load). CommonGrams T3 callers transfer
135
    // a precharged destination norm vector and a validated static metadata seed;
136
    // the streamed session late-binds semantic token_count before finish. Only ONE
137
    // session may be active per compound at a time, and begin_close() with an
138
    // unfinished session fails instead of sealing a half-fed container. The
139
    // handle is owned by this writer and valid until it is destroyed.
140
    Status add_snii_index_streamed(
141
            const TabletIndex* index_meta, uint32_t doc_count,
142
            doris::snii::writer::TrackedNullDocids null_docids,
143
            doris::snii::format::IndexConfig index_config,
144
            std::shared_ptr<doris::snii::writer::MemoryReporter> mem_reporter,
145
            doris::snii::writer::SniiStreamedIndexSession** session);
146
    Status add_snii_index_streamed(
147
            const TabletIndex* index_meta, uint32_t doc_count,
148
            doris::snii::writer::TrackedNullDocids null_docids,
149
            doris::snii::writer::TrackedEncodedNorms encoded_norms,
150
            std::optional<inverted_index::CommonGramsSegmentMetadata> common_grams_metadata,
151
            doris::snii::format::CommonGramsPostingPolicy common_grams_posting_policy,
152
            doris::snii::format::IndexConfig index_config,
153
            std::shared_ptr<doris::snii::writer::MemoryReporter> mem_reporter,
154
            doris::snii::writer::SniiStreamedIndexSession** session);
155
    // Registers one opaque BLOB logical index (a numeric BKD, an ANN graph, ...)
156
    // on this SNII compound. Unlike add_snii_index it feeds the writer no terms:
157
    // the sub-file bytes are pulled through the BlobFileSource callbacks at
158
    // finish(), which is what lets the container -- not the producer -- decide
159
    // cold/hot placement. Registration writes no byte, so a rejected call leaves
160
    // the writer clean.
161
    Status add_snii_blob_index(const TabletIndex* index_meta,
162
                               doris::snii::format::LogicalIndexKind kind,
163
                               std::vector<doris::snii::writer::BlobFileSource> cold_files,
164
                               std::vector<doris::snii::writer::BlobFileSource> hot_files);
165
    void retain_snii_memory_reporter(
166
            std::unique_ptr<doris::snii::writer::MemoryReporter> mem_reporter);
167
    // SNII only, BUILD INDEX rewrite: copies the source container's valid
168
    // physical prefix and registers the inherited metadata groups so begin_close
169
    // re-emits them without decoding a posting. Must precede every
170
    // add_snii_index on this writer (the copied prefix owns the container
171
    // front).
172
    Status inherit_snii(const doris::snii::reader::SniiRewriteSnapshot& snapshot,
173
                        doris::snii::io::FileReader* source);
174
    Status delete_index(const TabletIndex* index_meta);
175
    Status initialize(InvertedIndexDirectoryMap& indices_dirs);
176
    Status add_into_searcher_cache();
177
    // Begin the close process. This mainly triggers the asynchronous close operation of
178
    // _idx_v2_writer by calling close(true), which starts the close process but returns
179
    // immediately without waiting for completion.
180
    Status begin_close();
181
    // Finish the close process. This waits for the close operation to complete by calling
182
    // _idx_v2_writer->close(false), which blocks until the close is fully done.
183
    Status finish_close();
184
297
    const InvertedIndexFileInfo* get_index_file_info() const {
185
297
        DCHECK(_closed) << debug_string();
186
297
        return &_file_info;
187
297
    }
188
421
    int64_t get_index_file_total_size() const {
189
421
        DCHECK(_closed) << debug_string();
190
421
        return _total_file_size;
191
421
    }
192
0
    const io::FileSystemSPtr& get_fs() const { return _fs; }
193
7.53k
    InvertedIndexStorageFormatPB get_storage_format() const { return _storage_format; }
194
364
    void set_file_writer_opts(const io::FileWriterOptions& opts) { _opts = opts; }
195
    std::vector<std::string> get_index_file_names() const;
196
    std::string debug_string() const;
197
198
    // Get internal file writer (for merge file index collection)
199
0
    io::FileWriter* get_file_writer() const { return _idx_v2_writer.get(); }
200
201
private:
202
    Status _insert_directory_into_map(int64_t index_id, const std::string& index_suffix,
203
                                      std::shared_ptr<lucene::store::Directory> dir);
204
    // SNII only: registers a memory-backed staging directory for one ANN index,
205
    // together with the metadata begin_close() needs to seal it.
206
    Result<std::shared_ptr<lucene::store::Directory>> _open_snii_ann_staging_directory(
207
            const TabletIndex* index_meta);
208
    virtual Result<std::unique_ptr<IndexSearcherBuilder>> _construct_index_searcher_builder(
209
            const DorisCompoundReader* dir);
210
    // SNII only: turns every ANN staging directory into a blob logical index in
211
    // the container. Runs once, from begin_close(), before the compound writer is
212
    // sealed. Registration copies no byte -- the staged buffers are pulled by
213
    // finish() through the blob sources.
214
    Status _seal_snii_blob_directories();
215
    // Drops the staging directories once the container owns their bytes, or once
216
    // sealing has failed and they are dead either way. Only the SNII path needs
217
    // this: the V1/V2 branch of begin_close() releases its own directories
218
    // inline.
219
    void _release_snii_blob_directories();
220
221
    // Member variables...
222
    InvertedIndexDirectoryMap _indices_dirs;
223
    // SNII only: the index metadata behind each entry of _indices_dirs. Owned a
224
    // copy rather than borrowed, because the harvest happens in begin_close(),
225
    // long after open() returned. Held by shared_ptr so this header keeps
226
    // TabletIndex incomplete -- it is included nearly everywhere.
227
    std::map<std::pair<int64_t, std::string>, std::shared_ptr<TabletIndex>> _snii_blob_dir_metas;
228
    const io::FileSystemSPtr _fs;
229
    std::string _index_path_prefix;
230
    std::string _rowset_id;
231
    int64_t _seg_id;
232
    InvertedIndexStorageFormatPB _storage_format;
233
    std::string _tmp_dir;
234
    const std::shared_ptr<io::LocalFileSystem>& _local_fs;
235
236
    // write to disk or stream
237
    io::FileWriterPtr _idx_v2_writer = nullptr;
238
    io::FileWriterOptions _opts;
239
240
    // v1: all file size
241
    // v2: file size
242
    int64_t _total_file_size = 0;
243
    InvertedIndexFileInfo _file_info;
244
245
    // only once
246
    bool _closed = false;
247
    bool _can_use_ram_dir = true;
248
249
    IndexStorageFormatPtr _index_storage_format;
250
    int64_t _tablet_id = -1;
251
    std::unique_ptr<snii_doris::DorisSniiFileWriter> _snii_file_writer;
252
    std::vector<std::shared_ptr<doris::snii::writer::MemoryReporter>> _snii_memory_reporters;
253
    std::unique_ptr<doris::snii::writer::SniiCompoundWriter> _snii_compound_writer;
254
    size_t _snii_index_count = 0;
255
256
    friend class IndexStorageFormatV1;
257
    friend class IndexStorageFormatV2;
258
    friend class IndexFileWriterTest;
259
};
260
261
} // namespace segment_v2
262
} // namespace doris