1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
|
#include <Storages/MergeTree/MergeTreeIndexReader.h>
#include <Interpreters/Context.h>
#include <Storages/MergeTree/LoadedMergeTreeDataPartInfoForReader.h>
namespace
{
using namespace DB;
std::unique_ptr<MergeTreeReaderStream> makeIndexReader(
const std::string & extension,
MergeTreeIndexPtr index,
MergeTreeData::DataPartPtr part,
size_t marks_count,
const MarkRanges & all_mark_ranges,
MarkCache * mark_cache,
UncompressedCache * uncompressed_cache,
MergeTreeReaderSettings settings)
{
auto context = part->storage.getContext();
auto * load_marks_threadpool = settings.read_settings.load_marks_asynchronously ? &context->getLoadMarksThreadpool() : nullptr;
return std::make_unique<MergeTreeReaderStream>(
std::make_shared<LoadedMergeTreeDataPartInfoForReader>(part, std::make_shared<AlterConversions>()),
index->getFileName(), extension, marks_count,
all_mark_ranges,
std::move(settings), mark_cache, uncompressed_cache,
part->getFileSizeOrZero(index->getFileName() + extension),
&part->index_granularity_info,
ReadBufferFromFileBase::ProfileCallback{}, CLOCK_MONOTONIC_COARSE, false, load_marks_threadpool);
}
}
namespace DB
{
MergeTreeIndexReader::MergeTreeIndexReader(
MergeTreeIndexPtr index_,
MergeTreeData::DataPartPtr part_,
size_t marks_count_,
const MarkRanges & all_mark_ranges_,
MarkCache * mark_cache,
UncompressedCache * uncompressed_cache,
MergeTreeReaderSettings settings)
: index(index_)
{
auto index_format = index->getDeserializedFormat(part_->getDataPartStorage(), index->getFileName());
stream = makeIndexReader(
index_format.extension,
index_,
part_,
marks_count_,
all_mark_ranges_,
mark_cache,
uncompressed_cache,
std::move(settings));
version = index_format.version;
stream->adjustRightMark(getLastMark(all_mark_ranges_));
stream->seekToStart();
}
MergeTreeIndexReader::~MergeTreeIndexReader() = default;
void MergeTreeIndexReader::seek(size_t mark)
{
stream->seekToMark(mark);
}
MergeTreeIndexGranulePtr MergeTreeIndexReader::read()
{
auto granule = index->createIndexGranule();
granule->deserializeBinary(*stream->getDataBuffer(), version);
return granule;
}
}
|
