Coverage Report

Created: 2026-10-16 17:31

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
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