Coverage Report

Created: 2026-04-14 13:01

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/storage/compaction/collection_statistics.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/compaction/collection_statistics.h"
19
20
#include <set>
21
#include <sstream>
22
23
#include "common/exception.h"
24
#include "exprs/vexpr.h"
25
#include "exprs/vexpr_context.h"
26
#include "exprs/vliteral.h"
27
#include "exprs/vslot_ref.h"
28
#include "storage/index/index_file_reader.h"
29
#include "storage/index/index_reader_helper.h"
30
#include "storage/index/inverted/analyzer/analyzer.h"
31
#include "storage/index/inverted/util/string_helper.h"
32
#include "storage/index/inverted/util/term_iterator.h"
33
#include "storage/rowset/rowset.h"
34
#include "storage/rowset/rowset_reader.h"
35
#include "util/uid_util.h"
36
37
namespace doris {
38
39
Status CollectionStatistics::collect(RuntimeState* state,
40
                                     const std::vector<RowSetSplits>& rs_splits,
41
                                     const TabletSchemaSPtr& tablet_schema,
42
                                     const VExprContextSPtrs& common_expr_ctxs_push_down,
43
24
                                     io::IOContext* io_ctx) {
44
24
    std::unordered_map<std::wstring, CollectInfo> collect_infos;
45
24
    RETURN_IF_ERROR(
46
24
            extract_collect_info(state, common_expr_ctxs_push_down, tablet_schema, &collect_infos));
47
24
    if (collect_infos.empty()) {
48
10
        LOG(WARNING) << "Index statistics collection: no collect info extracted.";
49
10
        return Status::OK();
50
10
    }
51
52
42
    for (const auto& rs_split : rs_splits) {
53
42
        const auto& rs_reader = rs_split.rs_reader;
54
42
        auto rowset = rs_reader->rowset();
55
42
        auto num_segments = rowset->num_segments();
56
70
        for (int32_t seg_id = 0; seg_id < num_segments; ++seg_id) {
57
28
            auto status =
58
28
                    process_segment(rowset, seg_id, tablet_schema.get(), collect_infos, io_ctx);
59
28
            if (!status.ok()) {
60
0
                if (status.code() == ErrorCode::INVERTED_INDEX_FILE_NOT_FOUND ||
61
0
                    status.code() == ErrorCode::INVERTED_INDEX_BYPASS) {
62
0
                    LOG(ERROR) << "Index statistics collection failed: " << status.to_string();
63
0
                } else {
64
0
                    return status;
65
0
                }
66
0
            }
67
28
        }
68
42
    }
69
70
    // Build a single-line log with query_id, tablet_ids, and per-field term statistics
71
14
    if (VLOG_IS_ON(1)) {
72
0
        std::set<int64_t> tablet_ids;
73
0
        for (const auto& rs_split : rs_splits) {
74
0
            if (rs_split.rs_reader && rs_split.rs_reader->rowset()) {
75
0
                tablet_ids.insert(rs_split.rs_reader->rowset()->rowset_meta()->tablet_id());
76
0
            }
77
0
        }
78
79
0
        std::ostringstream oss;
80
0
        oss << "CollectionStatistics: query_id=" << print_id(state->query_id());
81
82
0
        oss << ", tablet_ids=[";
83
0
        bool first_tablet = true;
84
0
        for (int64_t tid : tablet_ids) {
85
0
            if (!first_tablet) oss << ",";
86
0
            oss << tid;
87
0
            first_tablet = false;
88
0
        }
89
0
        oss << "]";
90
91
0
        oss << ", total_num_docs=" << _total_num_docs;
92
93
0
        for (const auto& [ws_field_name, num_tokens] : _total_num_tokens) {
94
0
            oss << ", {field=" << StringHelper::to_string(ws_field_name)
95
0
                << ", num_tokens=" << num_tokens << ", terms=[";
96
97
0
            bool first_term = true;
98
0
            for (const auto& [term, doc_freq] : _term_doc_freqs.at(ws_field_name)) {
99
0
                if (!first_term) oss << ", ";
100
0
                oss << "(" << StringHelper::to_string(term) << ":" << doc_freq << ")";
101
0
                first_term = false;
102
0
            }
103
0
            oss << "]}";
104
0
        }
105
106
0
        VLOG(1) << oss.str();
107
0
    }
108
109
14
    return Status::OK();
110
14
}
111
112
Status CollectionStatistics::extract_collect_info(
113
        RuntimeState* state, const VExprContextSPtrs& common_expr_ctxs_push_down,
114
24
        const TabletSchemaSPtr& tablet_schema, CollectInfoMap* collect_infos) {
115
24
    DCHECK(collect_infos != nullptr);
116
117
24
    std::unordered_map<TExprNodeType::type, PredicateCollectorPtr> collectors;
118
24
    collectors[TExprNodeType::MATCH_PRED] = std::make_unique<MatchPredicateCollector>();
119
24
    collectors[TExprNodeType::SEARCH_EXPR] = std::make_unique<SearchPredicateCollector>();
120
121
24
    for (const auto& root_expr_ctx : common_expr_ctxs_push_down) {
122
24
        const auto& root_expr = root_expr_ctx->root();
123
24
        if (root_expr == nullptr) {
124
0
            continue;
125
0
        }
126
127
24
        std::stack<VExprSPtr> stack;
128
24
        stack.emplace(root_expr);
129
130
99
        while (!stack.empty()) {
131
75
            auto expr = stack.top();
132
75
            stack.pop();
133
134
75
            if (!expr) {
135
0
                continue;
136
0
            }
137
138
75
            auto collector_it = collectors.find(expr->node_type());
139
75
            if (collector_it != collectors.end()) {
140
23
                RETURN_IF_ERROR(
141
23
                        collector_it->second->collect(state, tablet_schema, expr, collect_infos));
142
23
            }
143
144
75
            const auto& children = expr->children();
145
75
            for (const auto& child : children) {
146
51
                stack.push(child);
147
51
            }
148
75
        }
149
24
    }
150
151
24
    LOG(INFO) << "Extracted collect info for " << collect_infos->size() << " fields";
152
153
24
    return Status::OK();
154
24
}
155
156
Status CollectionStatistics::process_segment(const RowsetSharedPtr& rowset, int32_t seg_id,
157
                                             const TabletSchema* tablet_schema,
158
                                             const CollectInfoMap& collect_infos,
159
28
                                             io::IOContext* io_ctx) {
160
28
    auto seg_path = DORIS_TRY(rowset->segment_path(seg_id));
161
28
    auto rowset_meta = rowset->rowset_meta();
162
163
28
    auto idx_file_reader = std::make_unique<IndexFileReader>(
164
28
            rowset_meta->fs(),
165
28
            std::string {InvertedIndexDescriptor::get_index_file_path_prefix(seg_path)},
166
28
            tablet_schema->get_inverted_index_storage_format(),
167
28
            rowset_meta->inverted_index_file_info(seg_id), rowset_meta->tablet_id());
168
28
    RETURN_IF_ERROR(idx_file_reader->init(config::inverted_index_read_buffer_size, io_ctx));
169
170
28
    int32_t total_seg_num_docs = 0;
171
172
28
    for (const auto& [ws_field_name, collect_info] : collect_infos) {
173
28
        lucene::search::IndexSearcher* index_searcher = nullptr;
174
28
        lucene::index::IndexReader* index_reader = nullptr;
175
176
#ifdef BE_TEST
177
        auto compound_reader = DORIS_TRY(idx_file_reader->open(collect_info.index_meta, io_ctx));
178
        auto* reader = lucene::index::IndexReader::open(compound_reader.get());
179
        auto searcher_ptr = std::make_shared<lucene::search::IndexSearcher>(reader, true);
180
        index_searcher = searcher_ptr.get();
181
        index_reader = index_searcher->getReader();
182
#else
183
28
        InvertedIndexCacheHandle inverted_index_cache_handle;
184
28
        auto index_file_key = idx_file_reader->get_index_file_cache_key(collect_info.index_meta);
185
28
        InvertedIndexSearcherCache::CacheKey searcher_cache_key(index_file_key);
186
187
28
        if (!InvertedIndexSearcherCache::instance()->lookup(searcher_cache_key,
188
28
                                                            &inverted_index_cache_handle)) {
189
0
            auto compound_reader =
190
0
                    DORIS_TRY(idx_file_reader->open(collect_info.index_meta, io_ctx));
191
0
            auto* reader = lucene::index::IndexReader::open(compound_reader.get());
192
0
            size_t reader_size = reader->getTermInfosRAMUsed();
193
0
            auto searcher_ptr = std::make_shared<lucene::search::IndexSearcher>(reader, true);
194
0
            auto* cache_value = new InvertedIndexSearcherCache::CacheValue(
195
0
                    std::move(searcher_ptr), reader_size, UnixMillis());
196
0
            InvertedIndexSearcherCache::instance()->insert(searcher_cache_key, cache_value,
197
0
                                                           &inverted_index_cache_handle);
198
0
        }
199
200
28
        auto searcher_variant = inverted_index_cache_handle.get_index_searcher();
201
28
        auto index_searcher_ptr = std::get<FulltextIndexSearcherPtr>(searcher_variant);
202
28
        index_searcher = index_searcher_ptr.get();
203
28
        index_reader = index_searcher->getReader();
204
28
#endif
205
28
        total_seg_num_docs = std::max(total_seg_num_docs, index_reader->maxDoc());
206
207
28
        _total_num_tokens[ws_field_name] +=
208
28
                index_reader->sumTotalTermFreq(ws_field_name.c_str()).value_or(0);
209
210
41
        for (const auto& term_info : collect_info.term_infos) {
211
41
            auto iter = TermIterator::create(io_ctx, false, index_reader, ws_field_name,
212
41
                                             term_info.get_single_term());
213
41
            _term_doc_freqs[ws_field_name][iter->term()] += iter->doc_freq();
214
41
        }
215
28
    }
216
217
28
    _total_num_docs += total_seg_num_docs;
218
219
28
    return Status::OK();
220
28
}
221
222
uint64_t CollectionStatistics::get_term_doc_freq_by_col(const std::wstring& lucene_col_name,
223
48
                                                        const std::wstring& term) {
224
48
    if (!_term_doc_freqs.contains(lucene_col_name)) {
225
1
        throw Exception(ErrorCode::INVERTED_INDEX_CLUCENE_ERROR,
226
1
                        "Index statistics collection failed: Not such column {}",
227
1
                        StringHelper::to_string(lucene_col_name));
228
1
    }
229
230
47
    if (!_term_doc_freqs[lucene_col_name].contains(term)) {
231
0
        throw Exception(ErrorCode::INVERTED_INDEX_CLUCENE_ERROR,
232
0
                        "Index statistics collection failed: Not such term {}",
233
0
                        StringHelper::to_string(term));
234
0
    }
235
236
47
    return _term_doc_freqs[lucene_col_name][term];
237
47
}
238
239
27
uint64_t CollectionStatistics::get_total_term_cnt_by_col(const std::wstring& lucene_col_name) {
240
27
    if (!_total_num_tokens.contains(lucene_col_name)) {
241
2
        throw Exception(ErrorCode::INVERTED_INDEX_CLUCENE_ERROR,
242
2
                        "Index statistics collection failed: Not such column {}",
243
2
                        StringHelper::to_string(lucene_col_name));
244
2
    }
245
246
25
    return _total_num_tokens[lucene_col_name];
247
27
}
248
249
72
uint64_t CollectionStatistics::get_doc_num() const {
250
72
    if (_total_num_docs == 0) {
251
3
        throw Exception(
252
3
                ErrorCode::INVERTED_INDEX_CLUCENE_ERROR,
253
3
                "Index statistics collection failed: No data available for SimilarityCollector");
254
3
    }
255
256
69
    return _total_num_docs;
257
72
}
258
259
51
float CollectionStatistics::get_or_calculate_avg_dl(const std::wstring& lucene_col_name) {
260
51
    auto iter = _avg_dl_by_col.find(lucene_col_name);
261
51
    if (iter != _avg_dl_by_col.end()) {
262
27
        return iter->second;
263
27
    }
264
265
24
    const uint64_t total_term_cnt = get_total_term_cnt_by_col(lucene_col_name);
266
24
    const uint64_t total_doc_cnt = get_doc_num();
267
24
    float avg_dl = total_doc_cnt > 0 ? float((double)total_term_cnt / (double)total_doc_cnt) : 0.0F;
268
24
    _avg_dl_by_col[lucene_col_name] = avg_dl;
269
24
    return avg_dl;
270
51
}
271
272
float CollectionStatistics::get_or_calculate_idf(const std::wstring& lucene_col_name,
273
63
                                                 const std::wstring& term) {
274
63
    auto iter = _idf_by_col_term.find(lucene_col_name);
275
63
    if (iter != _idf_by_col_term.end()) {
276
39
        auto term_iter = iter->second.find(term);
277
39
        if (term_iter != iter->second.end()) {
278
17
            return term_iter->second;
279
17
        }
280
39
    }
281
282
46
    const uint64_t doc_num = get_doc_num();
283
46
    const uint64_t doc_freq = get_term_doc_freq_by_col(lucene_col_name, term);
284
46
    auto idf = (float)std::log(1 + ((double)doc_num - (double)doc_freq + (double)0.5) /
285
46
                                           ((double)doc_freq + (double)0.5));
286
46
    _idf_by_col_term[lucene_col_name][term] = idf;
287
46
    return idf;
288
63
}
289
290
} // namespace doris