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 | 699 | 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 | | // Write-path facts for one SNII index flush. |
95 | | struct SniiAddIndexOptions { |
96 | | // This flush serves a stream/broker load (DataWriteType::TYPE_DIRECT): |
97 | | // the prx region compresses at snii_prx_zstd_level_direct_load; |
98 | | // compaction / schema change / ADD INDEX keep snii_prx_zstd_level. |
99 | | bool is_direct_load = false; |
100 | | // Present only for a CommonGrams writer that has a complete immutable |
101 | | // capability identity. These are semantic BM25 inputs; physical TTF is |
102 | | // still derived from every emitted unigram and gram posting. |
103 | | std::vector<uint8_t> encoded_norms; |
104 | | std::optional<inverted_index::CommonGramsSegmentMetadata> common_grams_metadata; |
105 | | snii::format::CommonGramsPostingPolicy common_grams_posting_policy = |
106 | | snii::format::CommonGramsPostingPolicy::kNone; |
107 | | }; |
108 | | Status add_snii_index(const TabletIndex* index_meta, uint32_t doc_count, |
109 | | std::vector<uint32_t> null_docids, |
110 | | doris::snii::writer::SpimiTermBuffer* const term_buffer, |
111 | | doris::snii::format::IndexConfig index_config, |
112 | | SniiAddIndexOptions options, |
113 | | doris::snii::writer::MemoryReporter* const mem_reporter); |
114 | | // T2.2 compaction index merge fast path: begins a STREAMED SNII index |
115 | | // session on this compound. Unlike add_snii_index (which drains a SPIMI |
116 | | // term buffer), the caller pushes pre-merged, lexicographically sorted |
117 | | // terms through *session and seals the index with (*session)->finish(). |
118 | | // Write parameters resolve through the SAME helper as add_snii_index |
119 | | // (write_freq / zstd levels / dict block size), always at the COMPACTION |
120 | | // prx tier (a merge is never a direct load). CommonGrams T3 callers transfer |
121 | | // a precharged destination norm vector and a validated static metadata seed; |
122 | | // the streamed session late-binds semantic token_count before finish. Only ONE |
123 | | // session may be active per compound at a time, and begin_close() with an |
124 | | // unfinished session fails instead of sealing a half-fed container. The |
125 | | // handle is owned by this writer and valid until it is destroyed. |
126 | | Status add_snii_index_streamed( |
127 | | const TabletIndex* index_meta, uint32_t doc_count, |
128 | | doris::snii::writer::TrackedNullDocids null_docids, |
129 | | doris::snii::format::IndexConfig index_config, |
130 | | std::shared_ptr<doris::snii::writer::MemoryReporter> mem_reporter, |
131 | | doris::snii::writer::SniiStreamedIndexSession** session); |
132 | | Status add_snii_index_streamed( |
133 | | const TabletIndex* index_meta, uint32_t doc_count, |
134 | | doris::snii::writer::TrackedNullDocids null_docids, |
135 | | doris::snii::writer::TrackedEncodedNorms encoded_norms, |
136 | | std::optional<inverted_index::CommonGramsSegmentMetadata> common_grams_metadata, |
137 | | doris::snii::format::CommonGramsPostingPolicy common_grams_posting_policy, |
138 | | doris::snii::format::IndexConfig index_config, |
139 | | std::shared_ptr<doris::snii::writer::MemoryReporter> mem_reporter, |
140 | | doris::snii::writer::SniiStreamedIndexSession** session); |
141 | | // Registers one opaque BLOB logical index (a numeric BKD, an ANN graph, ...) |
142 | | // on this SNII compound. Unlike add_snii_index it feeds the writer no terms: |
143 | | // the sub-file bytes are pulled through the BlobFileSource callbacks at |
144 | | // finish(), which is what lets the container -- not the producer -- decide |
145 | | // cold/hot placement. Registration writes no byte, so a rejected call leaves |
146 | | // the writer clean. |
147 | | Status add_snii_blob_index(const TabletIndex* index_meta, |
148 | | doris::snii::format::LogicalIndexKind kind, |
149 | | std::vector<doris::snii::writer::BlobFileSource> cold_files, |
150 | | std::vector<doris::snii::writer::BlobFileSource> hot_files); |
151 | | void retain_snii_memory_reporter( |
152 | | std::unique_ptr<doris::snii::writer::MemoryReporter> mem_reporter); |
153 | | // SNII only, BUILD INDEX rewrite: copies the source container's valid |
154 | | // physical prefix and registers the inherited metadata groups so begin_close |
155 | | // re-emits them without decoding a posting. Must precede every |
156 | | // add_snii_index on this writer (the copied prefix owns the container |
157 | | // front). |
158 | | Status inherit_snii(const doris::snii::reader::SniiRewriteSnapshot& snapshot, |
159 | | doris::snii::io::FileReader* source); |
160 | | Status delete_index(const TabletIndex* index_meta); |
161 | | Status initialize(InvertedIndexDirectoryMap& indices_dirs); |
162 | | Status add_into_searcher_cache(); |
163 | | // Begin the close process. This mainly triggers the asynchronous close operation of |
164 | | // _idx_v2_writer by calling close(true), which starts the close process but returns |
165 | | // immediately without waiting for completion. |
166 | | Status begin_close(); |
167 | | // Finish the close process. This waits for the close operation to complete by calling |
168 | | // _idx_v2_writer->close(false), which blocks until the close is fully done. |
169 | | Status finish_close(); |
170 | 297 | const InvertedIndexFileInfo* get_index_file_info() const { |
171 | 297 | DCHECK(_closed) << debug_string(); |
172 | 297 | return &_file_info; |
173 | 297 | } |
174 | 421 | int64_t get_index_file_total_size() const { |
175 | 421 | DCHECK(_closed) << debug_string(); |
176 | 421 | return _total_file_size; |
177 | 421 | } |
178 | 0 | const io::FileSystemSPtr& get_fs() const { return _fs; } |
179 | 7.53k | InvertedIndexStorageFormatPB get_storage_format() const { return _storage_format; } |
180 | 364 | void set_file_writer_opts(const io::FileWriterOptions& opts) { _opts = opts; } |
181 | | std::vector<std::string> get_index_file_names() const; |
182 | | std::string debug_string() const; |
183 | | |
184 | | // Get internal file writer (for merge file index collection) |
185 | 0 | io::FileWriter* get_file_writer() const { return _idx_v2_writer.get(); } |
186 | | |
187 | | private: |
188 | | Status _insert_directory_into_map(int64_t index_id, const std::string& index_suffix, |
189 | | std::shared_ptr<lucene::store::Directory> dir); |
190 | | // SNII only: registers a memory-backed staging directory for one ANN index, |
191 | | // together with the metadata begin_close() needs to seal it. |
192 | | Result<std::shared_ptr<lucene::store::Directory>> _open_snii_ann_staging_directory( |
193 | | const TabletIndex* index_meta); |
194 | | virtual Result<std::unique_ptr<IndexSearcherBuilder>> _construct_index_searcher_builder( |
195 | | const DorisCompoundReader* dir); |
196 | | // SNII only: turns every ANN staging directory into a blob logical index in |
197 | | // the container. Runs once, from begin_close(), before the compound writer is |
198 | | // sealed. Registration copies no byte -- the staged buffers are pulled by |
199 | | // finish() through the blob sources. |
200 | | Status _seal_snii_blob_directories(); |
201 | | // Drops the staging directories once the container owns their bytes, or once |
202 | | // sealing has failed and they are dead either way. Only the SNII path needs |
203 | | // this: the V1/V2 branch of begin_close() releases its own directories |
204 | | // inline. |
205 | | void _release_snii_blob_directories(); |
206 | | |
207 | | // Member variables... |
208 | | InvertedIndexDirectoryMap _indices_dirs; |
209 | | // SNII only: the index metadata behind each entry of _indices_dirs. Owned a |
210 | | // copy rather than borrowed, because the harvest happens in begin_close(), |
211 | | // long after open() returned. Held by shared_ptr so this header keeps |
212 | | // TabletIndex incomplete -- it is included nearly everywhere. |
213 | | std::map<std::pair<int64_t, std::string>, std::shared_ptr<TabletIndex>> _snii_blob_dir_metas; |
214 | | const io::FileSystemSPtr _fs; |
215 | | std::string _index_path_prefix; |
216 | | std::string _rowset_id; |
217 | | int64_t _seg_id; |
218 | | InvertedIndexStorageFormatPB _storage_format; |
219 | | std::string _tmp_dir; |
220 | | const std::shared_ptr<io::LocalFileSystem>& _local_fs; |
221 | | |
222 | | // write to disk or stream |
223 | | io::FileWriterPtr _idx_v2_writer = nullptr; |
224 | | io::FileWriterOptions _opts; |
225 | | |
226 | | // v1: all file size |
227 | | // v2: file size |
228 | | int64_t _total_file_size = 0; |
229 | | InvertedIndexFileInfo _file_info; |
230 | | |
231 | | // only once |
232 | | bool _closed = false; |
233 | | bool _can_use_ram_dir = true; |
234 | | |
235 | | IndexStorageFormatPtr _index_storage_format; |
236 | | int64_t _tablet_id = -1; |
237 | | std::unique_ptr<snii_doris::DorisSniiFileWriter> _snii_file_writer; |
238 | | std::vector<std::shared_ptr<doris::snii::writer::MemoryReporter>> _snii_memory_reporters; |
239 | | std::unique_ptr<doris::snii::writer::SniiCompoundWriter> _snii_compound_writer; |
240 | | size_t _snii_index_count = 0; |
241 | | |
242 | | friend class IndexStorageFormatV1; |
243 | | friend class IndexStorageFormatV2; |
244 | | friend class IndexFileWriterTest; |
245 | | }; |
246 | | |
247 | | } // namespace segment_v2 |
248 | | } // namespace doris |