be/src/storage/index/index_compound_reader.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_compound_reader.h" |
19 | | |
20 | | #include <CLucene/clucene-config.h> |
21 | | #include <CLucene/debug/error.h> |
22 | | #include <CLucene/debug/mem.h> |
23 | | #include <CLucene/store/RAMDirectory.h> |
24 | | #include <CLucene/util/Misc.h> |
25 | | |
26 | | #include <cstdio> |
27 | | #include <cstring> |
28 | | #include <cwchar> |
29 | | #include <memory> |
30 | | #include <utility> |
31 | | |
32 | | #include "storage/index/index_fs_directory.h" |
33 | | #include "storage/tablet/tablet_schema.h" |
34 | | #include "util/debug_points.h" |
35 | | |
36 | | namespace doris::io { |
37 | | class FileWriter; |
38 | | } // namespace doris::io |
39 | | |
40 | 186 | #define BUFFER_LENGTH 16384 |
41 | 3 | #define CL_MAX_PATH 4096 |
42 | | |
43 | | using FileWriterPtr = std::unique_ptr<doris::io::FileWriter>; |
44 | | |
45 | | namespace doris::segment_v2 { |
46 | | /** Implementation of an IndexInput that reads from a portion of the |
47 | | * compound file. |
48 | | */ |
49 | | class CSIndexInput : public lucene::store::BufferedIndexInput { |
50 | | private: |
51 | | CL_NS(store)::IndexInput* base; |
52 | | std::string file_name; |
53 | | int64_t fileOffset; |
54 | | int64_t _length; |
55 | | const io::IOContext* _io_ctx = nullptr; |
56 | | |
57 | | protected: |
58 | | void readInternal(uint8_t* /*b*/, const int32_t /*len*/) override; |
59 | 6.43M | void seekInternal(const int64_t /*pos*/) override {} |
60 | | |
61 | | public: |
62 | | CSIndexInput(CL_NS(store)::IndexInput* base, const std::string& file_name, |
63 | | const int64_t fileOffset, const int64_t length, |
64 | | const int32_t read_buffer_size = CL_NS(store)::BufferedIndexInput::BUFFER_SIZE); |
65 | | CSIndexInput(const CSIndexInput& clone); |
66 | | ~CSIndexInput() override; |
67 | | void close() override; |
68 | | lucene::store::IndexInput* clone() const override; |
69 | 6.29M | int64_t length() const override { return _length; } |
70 | 1 | const char* getDirectoryType() const override { return DorisCompoundReader::getClassName(); } |
71 | 17 | const char* getObjectName() const override { return getClassName(); } |
72 | 17 | static const char* getClassName() { return "CSIndexInput"; } |
73 | | void setIoContext(const void* io_ctx) override; |
74 | | }; |
75 | | |
76 | | CSIndexInput::CSIndexInput(CL_NS(store)::IndexInput* base, const std::string& file_name, |
77 | | const int64_t fileOffset, const int64_t length, |
78 | | const int32_t read_buffer_size) |
79 | 14.6k | : BufferedIndexInput(read_buffer_size) { |
80 | 14.6k | this->base = base; |
81 | 14.6k | this->file_name = file_name; |
82 | 14.6k | this->fileOffset = fileOffset; |
83 | 14.6k | this->_length = length; |
84 | 14.6k | } |
85 | | |
86 | 6.14M | void CSIndexInput::readInternal(uint8_t* b, const int32_t len) { |
87 | 6.14M | std::lock_guard wlock(((DorisFSDirectory::FSIndexInput*)base)->_this_lock); |
88 | | |
89 | 6.14M | auto start = getFilePointer(); |
90 | 6.14M | if (start + len > _length) { |
91 | 0 | _CLTHROWA(CL_ERR_IO, "read past EOF"); |
92 | 0 | } |
93 | | |
94 | 6.14M | if (_io_ctx) { |
95 | 576 | base->setIoContext(_io_ctx); |
96 | 576 | } |
97 | | |
98 | 6.14M | DBUG_EXECUTE_IF("CSIndexInput.readInternal", { |
99 | 6.14M | for (const auto& entry : InvertedIndexDescriptor::index_file_info_map) { |
100 | 6.14M | if (file_name.find(entry.first) != std::string::npos) { |
101 | 6.14M | if (!static_cast<const io::IOContext*>(base->getIoContext())->is_index_data) { |
102 | 6.14M | _CLTHROWA(CL_ERR_IO, |
103 | 6.14M | "The 'is_index_data' flag should be true for inverted index meta " |
104 | 6.14M | "files."); |
105 | 6.14M | } |
106 | 6.14M | } |
107 | 6.14M | } |
108 | 6.14M | for (const auto& entry : InvertedIndexDescriptor::normal_file_info_map) { |
109 | 6.14M | if (file_name.find(entry.first) != std::string::npos) { |
110 | 6.14M | if (static_cast<const io::IOContext*>(base->getIoContext())->is_index_data) { |
111 | 6.14M | _CLTHROWA(CL_ERR_IO, |
112 | 6.14M | "The 'is_index_data' flag should be false for non-meta inverted " |
113 | 6.14M | "index files."); |
114 | 6.14M | } |
115 | 6.14M | } |
116 | 6.14M | } |
117 | 6.14M | }); |
118 | | |
119 | 6.14M | base->seek(fileOffset + start); |
120 | 6.14M | bool read_from_buffer = true; |
121 | 6.14M | base->readBytes(b, len, read_from_buffer); |
122 | | |
123 | 6.14M | if (_io_ctx) { |
124 | 576 | base->setIoContext(nullptr); |
125 | 576 | } |
126 | 6.14M | } |
127 | | |
128 | 6.38M | CSIndexInput::~CSIndexInput() = default; |
129 | | |
130 | 6.36M | lucene::store::IndexInput* CSIndexInput::clone() const { |
131 | 6.36M | return _CLNEW CSIndexInput(*this); |
132 | 6.36M | } |
133 | | |
134 | 6.36M | CSIndexInput::CSIndexInput(const CSIndexInput& clone) : BufferedIndexInput(clone) { |
135 | 6.36M | this->base = clone.base; |
136 | 6.36M | this->file_name = clone.file_name; |
137 | 6.36M | this->fileOffset = clone.fileOffset; |
138 | 6.36M | this->_length = clone._length; |
139 | 6.36M | } |
140 | | |
141 | 18.9k | void CSIndexInput::close() {} |
142 | | |
143 | 13.5M | void CSIndexInput::setIoContext(const void* io_ctx) { |
144 | 13.5M | _io_ctx = static_cast<const io::IOContext*>(io_ctx); |
145 | 13.5M | } |
146 | | |
147 | | DorisCompoundReader::DorisCompoundReader(CL_NS(store)::IndexInput* stream, |
148 | | const EntriesType& entries_clone, int32_t read_buffer_size, |
149 | | const io::IOContext* io_ctx) |
150 | 4.89k | : _stream(stream), |
151 | 4.89k | _entries(std::make_unique<EntriesType>()), |
152 | 4.89k | _read_buffer_size(read_buffer_size) { |
153 | | // After stream clone, the io_ctx needs to be reconfigured. |
154 | 4.89k | initialize(io_ctx); |
155 | | |
156 | 42.8k | for (const auto& e : entries_clone) { |
157 | 42.8k | const auto& origin_entry = e.second; |
158 | 42.8k | auto entry = std::make_unique<ReaderFileEntry>(); |
159 | 42.8k | entry->file_name = origin_entry->file_name; |
160 | 42.8k | entry->offset = origin_entry->offset; |
161 | 42.8k | entry->length = origin_entry->length; |
162 | 42.8k | (*_entries)[e.first] = std::move(entry); |
163 | 42.8k | } |
164 | 4.89k | }; |
165 | | |
166 | | DorisCompoundReader::DorisCompoundReader(CL_NS(store)::IndexInput* stream, int32_t read_buffer_size, |
167 | | const io::IOContext* io_ctx) |
168 | 43 | : _ram_dir(std::make_unique<lucene::store::RAMDirectory>()), |
169 | 43 | _stream(stream), |
170 | 43 | _entries(std::make_unique<EntriesType>()), |
171 | 43 | _read_buffer_size(read_buffer_size) { |
172 | | // After stream clone, the io_ctx needs to be reconfigured. |
173 | 43 | initialize(io_ctx); |
174 | | |
175 | 43 | try { |
176 | 43 | int32_t count = _stream->readVInt(); |
177 | 43 | uint8_t buffer[BUFFER_LENGTH]; |
178 | 242 | for (int32_t i = 0; i < count; i++) { |
179 | 199 | auto entry = std::make_unique<ReaderFileEntry>(); |
180 | | // Read the string length first |
181 | 199 | int32_t string_length = stream->readVInt(); |
182 | | // Allocate appropriate buffer for the string |
183 | 199 | std::wstring tid; |
184 | 199 | tid.resize(string_length); |
185 | | // Read the string characters directly |
186 | 199 | stream->readChars(tid.data(), 0, string_length); |
187 | 199 | std::string file_name_str(tid.begin(), tid.end()); |
188 | 199 | entry->file_name = file_name_str; |
189 | 199 | entry->offset = stream->readLong(); |
190 | 199 | entry->length = stream->readLong(); |
191 | 199 | VLOG_DEBUG << "string_length:" << string_length << " file_name:" << entry->file_name |
192 | 0 | << " offset:" << entry->offset << " length:" << entry->length; |
193 | 199 | DBUG_EXECUTE_IF("construct_DorisCompoundReader_failed", { |
194 | 199 | CLuceneError err; |
195 | 199 | err.set(CL_ERR_IO, "construct_DorisCompoundReader_failed"); |
196 | 199 | throw err; |
197 | 199 | }) |
198 | | // read header file data |
199 | 199 | if (entry->offset < 0) { |
200 | | //if offset is -1, it means it's size is lower than DorisFSDirectory::MAX_HEADER_DATA_SIZE, which is 128k. |
201 | 186 | _copyFile(entry->file_name.c_str(), static_cast<int32_t>(entry->length), buffer, |
202 | 186 | BUFFER_LENGTH); |
203 | 186 | } |
204 | 199 | _entries->emplace(std::move(file_name_str), std::move(entry)); |
205 | 199 | } |
206 | 43 | } catch (...) { |
207 | 0 | try { |
208 | 0 | if (_stream != nullptr) { |
209 | 0 | _stream->close(); |
210 | 0 | _CLDELETE(_stream) |
211 | 0 | } |
212 | 0 | if (_entries != nullptr) { |
213 | 0 | _entries->clear(); |
214 | 0 | } |
215 | 0 | if (_ram_dir) { |
216 | 0 | _ram_dir->close(); |
217 | 0 | } |
218 | 0 | } catch (CLuceneError& err) { |
219 | 0 | if (err.number() != CL_ERR_IO) { |
220 | 0 | throw err; |
221 | 0 | } |
222 | 0 | } |
223 | 0 | throw; |
224 | 0 | } |
225 | 43 | } |
226 | | |
227 | | void DorisCompoundReader::_copyFile(const char* file, int32_t file_length, uint8_t* buffer, |
228 | 188 | int32_t buffer_length) { |
229 | 188 | std::unique_ptr<lucene::store::IndexOutput> output(_ram_dir->createOutput(file)); |
230 | 188 | int64_t start_ptr = output->getFilePointer(); |
231 | 188 | auto remainder = file_length; |
232 | 188 | auto chunk = buffer_length; |
233 | 188 | auto batch_len = file_length < chunk ? file_length : chunk; |
234 | | |
235 | 377 | while (remainder > 0) { |
236 | 189 | auto len = remainder < batch_len ? remainder : batch_len; |
237 | 189 | _stream->readBytes(buffer, len); |
238 | 189 | output->writeBytes(buffer, len); |
239 | 189 | remainder -= len; |
240 | 189 | } |
241 | 188 | if (remainder != 0) { |
242 | 0 | TCHAR buf[CL_MAX_PATH + 100]; |
243 | 0 | swprintf(buf, CL_MAX_PATH + 100, |
244 | 0 | _T("Non-zero remainder length after copying") |
245 | 0 | _T(": %d (id: %s, length: %d, buffer size: %d)"), |
246 | 0 | (int)remainder, file, (int)file_length, (int)chunk); |
247 | 0 | _CLTHROWT(CL_ERR_IO, buf); |
248 | 0 | } |
249 | | |
250 | 188 | int64_t end_ptr = output->getFilePointer(); |
251 | 188 | int64_t diff = end_ptr - start_ptr; |
252 | 188 | if (diff != file_length) { |
253 | 0 | TCHAR buf[100]; |
254 | 0 | swprintf(buf, 100, |
255 | 0 | _T("Difference in the output file offsets %d ") |
256 | 0 | _T("does not match the original file length %d"), |
257 | 0 | (int)diff, (int)file_length); |
258 | 0 | _CLTHROWA(CL_ERR_IO, buf); |
259 | 0 | } |
260 | 188 | output->close(); |
261 | 188 | } |
262 | | |
263 | 4.93k | DorisCompoundReader::~DorisCompoundReader() { |
264 | 4.93k | if (!_closed) { |
265 | 3.35k | try { |
266 | 3.35k | close(); |
267 | 3.35k | } catch (CLuceneError& err) { |
268 | 0 | LOG(ERROR) << "DorisCompoundReader finalize error:" << err.what(); |
269 | 0 | } |
270 | 3.35k | } |
271 | 4.93k | } |
272 | | |
273 | 5 | const char* DorisCompoundReader::getClassName() { |
274 | 5 | return "DorisCompoundReader"; |
275 | 5 | } |
276 | 4 | const char* DorisCompoundReader::getObjectName() const { |
277 | 4 | return getClassName(); |
278 | 4 | } |
279 | | |
280 | 3.15k | bool DorisCompoundReader::list(std::vector<std::string>* names) const { |
281 | 3.15k | if (_closed || _entries == nullptr) { |
282 | 1 | _CLTHROWA(CL_ERR_IO, "DorisCompoundReader is already closed"); |
283 | 1 | } |
284 | 28.9k | for (const auto& entry : *_entries) { |
285 | 28.9k | names->push_back(entry.first); |
286 | 28.9k | } |
287 | 3.14k | return true; |
288 | 3.15k | } |
289 | | |
290 | 1.21k | bool DorisCompoundReader::fileExists(const char* name) const { |
291 | 1.21k | if (_closed || _entries == nullptr) { |
292 | 1 | _CLTHROWA(CL_ERR_IO, "DorisCompoundReader is already closed"); |
293 | 1 | } |
294 | 1.21k | return _entries->find(std::string(name)) != _entries->end(); |
295 | 1.21k | } |
296 | | |
297 | 1 | int64_t DorisCompoundReader::fileModified(const char* name) const { |
298 | 1 | return 0; |
299 | 1 | } |
300 | | |
301 | 26 | int64_t DorisCompoundReader::fileLength(const char* name) const { |
302 | 26 | if (_closed || _entries == nullptr) { |
303 | 1 | _CLTHROWA(CL_ERR_IO, "DorisCompoundReader is already closed"); |
304 | 1 | } |
305 | 25 | auto it = _entries->find(std::string(name)); |
306 | 25 | if (it == _entries->end()) { |
307 | 1 | char buf[CL_MAX_PATH + 30]; |
308 | 1 | strcpy(buf, "File "); |
309 | 1 | strncat(buf, name, CL_MAX_PATH); |
310 | 1 | strcat(buf, " does not exist"); |
311 | 1 | _CLTHROWA(CL_ERR_IO, buf); |
312 | 1 | } |
313 | 24 | return it->second->length; |
314 | 25 | } |
315 | | |
316 | | bool DorisCompoundReader::openInput(const char* name, |
317 | | std::unique_ptr<lucene::store::IndexInput>& ret, |
318 | 37 | CLuceneError& error, int32_t bufferSize) { |
319 | 37 | if (_closed || _entries == nullptr) { |
320 | 0 | error.set(CL_ERR_IO, "DorisCompoundReader is already closed"); |
321 | 0 | return false; |
322 | 0 | } |
323 | 37 | lucene::store::IndexInput* tmp; |
324 | 37 | bool success = openInput(name, tmp, error, bufferSize); |
325 | 37 | if (success) { |
326 | 36 | ret.reset(tmp); |
327 | 36 | } |
328 | 37 | return success; |
329 | 37 | } |
330 | | |
331 | | bool DorisCompoundReader::openInput(const char* name, lucene::store::IndexInput*& ret, |
332 | 14.7k | CLuceneError& error, int32_t bufferSize) { |
333 | 14.7k | if (_stream == nullptr) { |
334 | 1 | error.set(CL_ERR_IO, "Stream closed"); |
335 | 1 | return false; |
336 | 1 | } |
337 | | |
338 | 14.7k | auto it = _entries->find(std::string(name)); |
339 | 14.7k | if (it == _entries->end()) { |
340 | 2 | char buf[CL_MAX_PATH + 26]; |
341 | 2 | snprintf(buf, CL_MAX_PATH + 26, "No sub-file with id %s found", name); |
342 | 2 | error.set(CL_ERR_IO, buf); |
343 | 2 | return false; |
344 | 2 | } |
345 | | |
346 | 14.7k | const auto& entry = it->second; |
347 | | |
348 | | // If file is in RAM, just return. |
349 | 14.7k | if (_ram_dir && _ram_dir->fileExists(name)) { |
350 | 147 | return _ram_dir->openInput(name, ret, error, bufferSize); |
351 | 147 | } |
352 | | |
353 | 14.6k | if (bufferSize < 1) { |
354 | 6.95k | bufferSize = _read_buffer_size; |
355 | 6.95k | } |
356 | | |
357 | 14.6k | ret = _CLNEW CSIndexInput(_stream, entry->file_name, entry->offset, entry->length, bufferSize); |
358 | 14.6k | return true; |
359 | 14.7k | } |
360 | | |
361 | 4.93k | void DorisCompoundReader::close() { |
362 | 4.93k | std::lock_guard<std::mutex> wlock(_this_lock); |
363 | 4.93k | if (_stream != nullptr) { |
364 | 4.93k | _stream->close(); |
365 | 4.93k | _CLDELETE(_stream) |
366 | 4.93k | } |
367 | 4.93k | if (_entries != nullptr) { |
368 | | // The life cycle of _entries should be consistent with that of the DorisCompoundReader. |
369 | | // DO NOT DELETE _entries here, it will be deleted in the destructor |
370 | | // When directory is closed, all _entries are cleared. But the directory may be called in other places. |
371 | | // If we delete the _entries object here, it will cause core dump. |
372 | 4.93k | _entries->clear(); |
373 | 4.93k | } |
374 | 4.93k | if (_ram_dir) { |
375 | 44 | _ram_dir->close(); |
376 | 44 | } |
377 | 4.93k | _closed = true; |
378 | 4.93k | } |
379 | | |
380 | 1 | bool DorisCompoundReader::doDeleteFile(const char* /*name*/) { |
381 | 1 | _CLTHROWA(CL_ERR_UnsupportedOperation, |
382 | 1 | "UnsupportedOperationException: DorisCompoundReader::doDeleteFile"); |
383 | 1 | } |
384 | | |
385 | 1 | void DorisCompoundReader::renameFile(const char* /*from*/, const char* /*to*/) { |
386 | 1 | _CLTHROWA(CL_ERR_UnsupportedOperation, |
387 | 1 | "UnsupportedOperationException: DorisCompoundReader::renameFile"); |
388 | 1 | } |
389 | | |
390 | 1 | void DorisCompoundReader::touchFile(const char* /*name*/) { |
391 | 1 | _CLTHROWA(CL_ERR_UnsupportedOperation, |
392 | 1 | "UnsupportedOperationException: DorisCompoundReader::touchFile"); |
393 | 1 | } |
394 | | |
395 | 1 | lucene::store::IndexOutput* DorisCompoundReader::createOutput(const char* /*name*/) { |
396 | 1 | _CLTHROWA(CL_ERR_UnsupportedOperation, |
397 | 1 | "UnsupportedOperationException: DorisCompoundReader::createOutput"); |
398 | 1 | } |
399 | | |
400 | 1 | std::string DorisCompoundReader::toString() const { |
401 | 1 | return "DorisCompoundReader@"; |
402 | 1 | } |
403 | | |
404 | 208 | CL_NS(store)::IndexInput* DorisCompoundReader::getDorisIndexInput() { |
405 | 208 | return _stream; |
406 | 208 | } |
407 | | |
408 | 4.93k | void DorisCompoundReader::initialize(const io::IOContext* io_ctx) { |
409 | 4.93k | _stream->setIoContext(io_ctx); |
410 | 4.93k | _stream->setIdxFileCache(true); |
411 | 4.93k | } |
412 | | |
413 | | } // namespace doris::segment_v2 |