Coverage Report

Created: 2026-07-15 14:44

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/storage/segment/column_reader_cache.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/segment/column_reader_cache.h"
19
20
#include "storage/segment/column_meta_accessor.h"
21
#include "storage/segment/segment.h"
22
#include "storage/segment/variant/variant_column_reader.h"
23
24
bvar::Adder<int64_t> g_segment_column_reader_cache_count("segment_column_cache_count");
25
bvar::Adder<int64_t> g_segment_column_cache_hit_count("segment_column_cache_hit_count");
26
bvar::Adder<int64_t> g_segment_column_cache_miss_count("segment_column_cache_miss_count");
27
bvar::Adder<int64_t> g_segment_column_cache_evict_count("segment_column_cache_evict_count");
28
29
namespace doris::segment_v2 {
30
31
ColumnReaderCache::ColumnReaderCache(
32
        ColumnMetaAccessor* accessor, TabletSchemaSPtr tablet_schema,
33
        io::FileReaderSPtr file_reader, uint64_t num_rows,
34
        std::function<Status(std::shared_ptr<SegmentFooterPB>&, OlapReaderStatistics*,
35
                             const io::IOContext*)>
36
                get_footer_cb)
37
3.66k
        : _accessor(accessor),
38
3.66k
          _tablet_schema(std::move(tablet_schema)),
39
3.66k
          _file_reader(std::move(file_reader)),
40
3.66k
          _num_rows(num_rows),
41
3.66k
          _get_footer_cb(std::move(get_footer_cb)) {}
42
43
3.66k
ColumnReaderCache::~ColumnReaderCache() {
44
3.66k
    g_segment_column_reader_cache_count << -_cache_map.size();
45
3.66k
}
46
47
14.1k
std::shared_ptr<ColumnReader> ColumnReaderCache::_lookup(const ColumnReaderCacheKey& key) {
48
14.1k
    std::lock_guard<std::mutex> lock(_cache_mutex);
49
14.1k
    auto it = _cache_map.find(key);
50
14.1k
    if (it == _cache_map.end()) {
51
8.57k
        g_segment_column_cache_miss_count << 1;
52
8.57k
        return nullptr;
53
8.57k
    }
54
    // Move the accessed node to the front of the linked list
55
5.56k
    _lru_list.splice(_lru_list.begin(), _lru_list, it->second);
56
5.56k
    DCHECK_EQ(it->second->key.first, key.first);
57
5.56k
    g_segment_column_cache_hit_count << 1;
58
5.56k
    return it->second->reader;
59
14.1k
}
60
61
void ColumnReaderCache::_insert_locked_nocheck(const ColumnReaderCacheKey& key,
62
8.54k
                                               const std::shared_ptr<ColumnReader>& reader) {
63
    // If capacity exceeded, remove least recently used (tail)
64
8.54k
    if (_cache_map.size() >= config::max_segment_partial_column_cache_size) {
65
278
        g_segment_column_reader_cache_count << -1;
66
278
        g_segment_column_cache_evict_count << 1;
67
278
        auto last_it = _lru_list.end();
68
278
        --last_it;
69
278
        _cache_map.erase(last_it->key);
70
278
        _lru_list.pop_back();
71
278
    }
72
8.54k
    g_segment_column_reader_cache_count << 1;
73
8.54k
    _lru_list.push_front(CacheNode {
74
8.54k
            .key = key, .reader = reader, .last_access = std::chrono::steady_clock::now()});
75
8.54k
    _cache_map[key] = _lru_list.begin();
76
8.54k
}
77
78
void ColumnReaderCache::_insert_direct(const ColumnReaderCacheKey& key,
79
8.54k
                                       const std::shared_ptr<ColumnReader>& column_reader) {
80
8.54k
    std::lock_guard<std::mutex> lock(_cache_mutex);
81
8.54k
    _insert_locked_nocheck(key, column_reader);
82
8.54k
}
83
84
std::map<int32_t, std::shared_ptr<ColumnReader>> ColumnReaderCache::get_available_readers(
85
71
        bool include_subcolumns) {
86
71
    std::lock_guard<std::mutex> lock(_cache_mutex);
87
71
    std::map<int32_t, std::shared_ptr<ColumnReader>> readers;
88
120
    for (const auto& node : _lru_list) {
89
120
        if (include_subcolumns || node.key.second.empty()) {
90
76
            readers.insert({node.key.first, node.reader});
91
76
        }
92
120
    }
93
71
    return readers;
94
71
}
95
96
Status ColumnReaderCache::get_column_reader(int32_t col_uid,
97
                                            std::shared_ptr<ColumnReader>* column_reader,
98
                                            OlapReaderStatistics* stats,
99
                                            const io::IOContext* source_io_ctx,
100
13.1k
                                            std::optional<Field> const_value) {
101
    // Attempt to find in cache
102
13.1k
    if (auto cached = _lookup({col_uid, {}})) {
103
5.08k
        *column_reader = cached;
104
5.08k
        return Status::OK();
105
5.08k
    }
106
    // Load footer once under cache mutex (not thread-safe otherwise)
107
8.06k
    std::shared_ptr<SegmentFooterPB> footer_pb_shared;
108
8.06k
    {
109
8.06k
        std::lock_guard<std::mutex> lock(_cache_mutex);
110
8.06k
        RETURN_IF_ERROR(_get_footer_cb(footer_pb_shared, stats, source_io_ctx));
111
8.06k
    }
112
113
    // Lookup column meta by uid via ColumnMetaAccessor. If not initialized or not found, return NOT_FOUND.
114
8.05k
    ColumnMetaPB meta;
115
8.05k
    Status st_meta = _accessor->get_column_meta_by_uid(*footer_pb_shared, col_uid, &meta, stats,
116
8.05k
                                                       source_io_ctx);
117
8.05k
    if (st_meta.is<ErrorCode::NOT_FOUND>()) {
118
1
        *column_reader = nullptr;
119
1
        return st_meta;
120
1
    }
121
8.05k
    RETURN_IF_ERROR(st_meta);
122
123
8.05k
    ColumnReaderOptions opts {.kept_in_memory = _tablet_schema->is_in_memory(),
124
8.05k
                              .be_exec_version = _be_exec_version,
125
8.05k
                              .tablet_schema = _tablet_schema};
126
8.05k
    opts.const_value = std::move(const_value);
127
128
8.05k
    std::shared_ptr<ColumnReader> reader;
129
8.05k
    if ((FieldType)meta.type() == FieldType::OLAP_FIELD_TYPE_VARIANT) {
130
        // Variant root columns require VariantColumnReader, which encapsulates
131
        // subcolumn layout, sparse columns and external meta.
132
700
        std::unique_ptr<VariantColumnReader> variant_reader(new VariantColumnReader());
133
700
        RETURN_IF_ERROR(variant_reader->init(opts, _accessor, footer_pb_shared, col_uid, _num_rows,
134
700
                                             _file_reader, stats, source_io_ctx));
135
700
        reader.reset(variant_reader.release());
136
700
        VLOG_DEBUG << "insert cache (variant): uid=" << col_uid << " col_id=" << meta.column_id();
137
7.35k
    } else {
138
        // For non-variant columns, we can create reader directly from ColumnMetaPB.
139
7.35k
        VLOG_DEBUG << "insert cache: uid=" << col_uid << " col_id=" << meta.column_id();
140
7.35k
        RETURN_IF_ERROR(ColumnReader::create(opts, meta, _num_rows, _file_reader, &reader));
141
7.35k
    }
142
143
8.05k
    _insert_direct({col_uid, {}}, reader);
144
8.05k
    *column_reader = std::move(reader);
145
8.05k
    return Status::OK();
146
8.05k
}
147
148
Status ColumnReaderCache::get_path_column_reader(int32_t col_uid, PathInData relative_path,
149
                                                 std::shared_ptr<ColumnReader>* column_reader,
150
                                                 OlapReaderStatistics* stats,
151
                                                 const SubcolumnColumnMetaInfo::Node* node_hint,
152
1.00k
                                                 const io::IOContext* source_io_ctx) {
153
    // Attempt to find in cache first
154
1.00k
    if (auto cached = _lookup({col_uid, relative_path})) {
155
486
        *column_reader = cached;
156
486
        return Status::OK();
157
486
    }
158
159
517
    if (!_accessor->has_column_uid(col_uid)) {
160
0
        *column_reader = nullptr;
161
0
        return Status::Error<ErrorCode::NOT_FOUND, false>("column not found in segment, col_uid={}",
162
0
                                                          col_uid);
163
0
    }
164
165
    // Load footer once under cache mutex (not thread-safe otherwise)
166
517
    std::shared_ptr<SegmentFooterPB> footer_pb_shared;
167
517
    {
168
517
        std::lock_guard<std::mutex> lock(_cache_mutex);
169
517
        RETURN_IF_ERROR(_get_footer_cb(footer_pb_shared, stats, source_io_ctx));
170
517
    }
171
172
    // Ensure variant root reader is available in cache.
173
517
    ColumnReaderOptions opts {.kept_in_memory = _tablet_schema->is_in_memory(),
174
517
                              .be_exec_version = _be_exec_version,
175
517
                              .tablet_schema = _tablet_schema};
176
517
    std::shared_ptr<ColumnReader> variant_column_reader;
177
517
    RETURN_IF_ERROR(get_column_reader(col_uid, &variant_column_reader, stats, source_io_ctx));
178
179
517
    if (relative_path.empty()) {
180
0
        *column_reader = std::move(variant_column_reader);
181
0
        return Status::OK();
182
0
    }
183
184
    // Delegate path-level reader creation to VariantColumnReader, which hides
185
    // inline vs external meta details.
186
517
    std::shared_ptr<ColumnReader> path_reader;
187
517
    auto* vreader = static_cast<VariantColumnReader*>(variant_column_reader.get());
188
517
    Status st = vreader->create_path_reader(relative_path, opts, _accessor, *footer_pb_shared,
189
517
                                            _file_reader, _num_rows, &path_reader, stats,
190
517
                                            source_io_ctx);
191
517
    if (st.is<ErrorCode::NOT_FOUND>()) {
192
26
        *column_reader = nullptr;
193
26
        return st;
194
26
    }
195
491
    RETURN_IF_ERROR(st);
196
197
    // Cache and return
198
491
    _insert_direct({col_uid, relative_path}, path_reader);
199
491
    *column_reader = std::move(path_reader);
200
491
    return Status::OK();
201
491
}
202
203
} // namespace doris::segment_v2