Coverage Report

Created: 2026-07-07 22:21

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