Coverage Report

Created: 2025-10-17 00:26

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
/root/doris/be/src/olap/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 "collection_statistics.h"
19
20
#include "common/exception.h"
21
#include "olap/rowset/rowset.h"
22
#include "olap/rowset/rowset_reader.h"
23
#include "olap/rowset/segment_v2/index_file_reader.h"
24
#include "olap/rowset/segment_v2/index_reader_helper.h"
25
#include "olap/rowset/segment_v2/inverted_index/analyzer/analyzer.h"
26
#include "olap/rowset/segment_v2/inverted_index/util/string_helper.h"
27
#include "vec/exprs/vexpr.h"
28
#include "vec/exprs/vexpr_context.h"
29
#include "vec/exprs/vliteral.h"
30
#include "vec/exprs/vslot_ref.h"
31
32
namespace doris {
33
#include "common/compile_check_begin.h"
34
35
Status CollectionStatistics::collect(
36
        RuntimeState* state, const std::vector<RowSetSplits>& rs_splits,
37
        const TabletSchemaSPtr& tablet_schema,
38
10
        const vectorized::VExprContextSPtrs& common_expr_ctxs_push_down) {
39
10
    std::unordered_map<std::wstring, CollectInfo> collect_infos;
40
10
    RETURN_IF_ERROR(
41
10
            extract_collect_info(state, common_expr_ctxs_push_down, tablet_schema, &collect_infos));
42
43
10
    for (const auto& rs_split : rs_splits) {
44
5
        const auto& rs_reader = rs_split.rs_reader;
45
5
        auto rowset = rs_reader->rowset();
46
5
        auto rowset_meta = rowset->rowset_meta();
47
48
5
        auto num_segments = rowset->num_segments();
49
7
        for (int32_t seg_id = 0; seg_id < num_segments; ++seg_id) {
50
2
            auto seg_path = DORIS_TRY(rowset->segment_path(seg_id));
51
2
            auto status = process_segment(seg_path, rowset_meta->fs(), tablet_schema.get(),
52
2
                                          collect_infos);
53
2
            if (!status.ok()) {
54
0
                if (status.code() == ErrorCode::INVERTED_INDEX_FILE_NOT_FOUND ||
55
0
                    status.code() == ErrorCode::INVERTED_INDEX_BYPASS) {
56
0
                    LOG(ERROR) << "Index statistics collection failed: " << status.to_string();
57
0
                } else {
58
0
                    return status;
59
0
                }
60
0
            }
61
2
        }
62
5
    }
63
64
10
#ifndef NDEBUG
65
10
    LOG(INFO) << "term_num_docs: " << _total_num_docs;
66
10
    for (const auto& [ws_field_name, num_tokens] : _total_num_tokens) {
67
0
        LOG(INFO) << "field_name: " << StringHelper::to_string(ws_field_name)
68
0
                  << ", num_tokens: " << num_tokens;
69
0
        for (const auto& [term, doc_freq] : _term_doc_freqs.at(ws_field_name)) {
70
0
            LOG(INFO) << "term: " << StringHelper::to_string(term) << ", doc_freq: " << doc_freq;
71
0
        }
72
0
    }
73
10
    LOG(INFO) << "--------------------------------";
74
10
#endif
75
76
10
    return Status::OK();
77
10
}
78
79
14
vectorized::VSlotRef* find_slot_ref(const vectorized::VExprSPtr& expr) {
80
14
    if (!expr) return nullptr;
81
13
    auto cur = vectorized::VExpr::expr_without_cast(expr);
82
13
    if (cur->node_type() == TExprNodeType::SLOT_REF) {
83
12
        return static_cast<vectorized::VSlotRef*>(cur.get());
84
12
    }
85
1
    for (auto& ch : cur->children()) {
86
1
        if (auto* s = find_slot_ref(ch)) return s;
87
1
    }
88
0
    return nullptr;
89
1
}
90
91
Status handle_match_pred(RuntimeState* state, const TabletSchemaSPtr& tablet_schema,
92
                         const vectorized::VExprSPtr& expr,
93
9
                         std::unordered_map<std::wstring, CollectInfo>* collect_infos) {
94
9
    auto* left_slot_ref = find_slot_ref(expr->children()[0]);
95
9
    if (left_slot_ref == nullptr) {
96
0
        return Status::Error<ErrorCode::INVERTED_INDEX_NOT_SUPPORTED>(
97
0
                "Index statistics collection failed: Cannot find slot reference in match predicate "
98
0
                "left expression");
99
0
    }
100
9
    auto* right_literal = static_cast<vectorized::VLiteral*>(expr->children()[1].get());
101
9
    DCHECK(right_literal != nullptr);
102
103
9
    const auto* sd = state->desc_tbl().get_slot_descriptor(left_slot_ref->slot_id());
104
9
    if (sd == nullptr) {
105
0
        return Status::Error<ErrorCode::INVERTED_INDEX_NOT_SUPPORTED>(
106
0
                "Index statistics collection failed: Cannot find slot descriptor for slot_id={}",
107
0
                left_slot_ref->slot_id());
108
0
    }
109
9
    int32_t col_idx = tablet_schema->field_index(left_slot_ref->column_name());
110
9
    if (col_idx == -1) {
111
0
        return Status::Error<ErrorCode::INVERTED_INDEX_NOT_SUPPORTED>(
112
0
                "Index statistics collection failed: Cannot find column index for column={}",
113
0
                left_slot_ref->column_name());
114
0
    }
115
116
9
    const auto& column = tablet_schema->column(col_idx);
117
9
    auto index_metas = tablet_schema->inverted_indexs(sd->col_unique_id(), column.suffix_path());
118
#ifndef BE_TEST
119
    if (index_metas.empty()) {
120
        return Status::Error<ErrorCode::INVERTED_INDEX_NOT_SUPPORTED>(
121
                "Index statistics collection failed: Score query is not supported without inverted "
122
                "index for column={}",
123
                left_slot_ref->column_name());
124
    }
125
#endif
126
9
    for (const auto* index_meta : index_metas) {
127
0
        if (!InvertedIndexAnalyzer::should_analyzer(index_meta->properties())) {
128
0
            continue;
129
0
        }
130
0
        if (!segment_v2::IndexReaderHelper::is_need_similarity_score(expr->op(), index_meta)) {
131
0
            continue;
132
0
        }
133
134
0
        auto term_infos = InvertedIndexAnalyzer::get_analyse_result(right_literal->value(),
135
0
                                                                    index_meta->properties());
136
137
0
        std::string field_name = std::to_string(index_meta->col_unique_ids()[0]);
138
0
        if (!column.suffix_path().empty()) {
139
0
            field_name += "." + column.suffix_path();
140
0
        }
141
0
        std::wstring ws_field_name = StringHelper::to_wstring(field_name);
142
0
        auto iter = collect_infos->find(ws_field_name);
143
0
        if (iter == collect_infos->end()) {
144
0
            CollectInfo collect_info;
145
0
            collect_info.term_infos.insert(term_infos.begin(), term_infos.end());
146
0
            collect_info.index_meta = index_meta;
147
0
            (*collect_infos)[ws_field_name] = std::move(collect_info);
148
0
        } else {
149
0
            iter->second.term_infos.insert(term_infos.begin(), term_infos.end());
150
0
        }
151
0
    }
152
9
    return Status::OK();
153
9
}
154
155
Status CollectionStatistics::extract_collect_info(
156
        RuntimeState* state, const vectorized::VExprContextSPtrs& common_expr_ctxs_push_down,
157
        const TabletSchemaSPtr& tablet_schema,
158
10
        std::unordered_map<std::wstring, CollectInfo>* collect_infos) {
159
10
    for (const auto& root_expr_ctx : common_expr_ctxs_push_down) {
160
10
        const auto& root_expr = root_expr_ctx->root();
161
10
        if (root_expr == nullptr) {
162
0
            continue;
163
0
        }
164
165
10
        std::stack<vectorized::VExprSPtr> stack;
166
10
        stack.emplace(root_expr);
167
168
24
        while (!stack.empty()) {
169
14
            const auto& expr = stack.top();
170
14
            stack.pop();
171
172
14
            if (expr->node_type() == TExprNodeType::MATCH_PRED) {
173
9
                RETURN_IF_ERROR(handle_match_pred(state, tablet_schema, expr, collect_infos));
174
9
            }
175
176
14
            const auto& children = expr->children();
177
37
            for (int32_t i = static_cast<int32_t>(children.size()) - 1; i >= 0; --i) {
178
23
                if (!children[i]->children().empty()) {
179
4
                    stack.emplace(children[i]);
180
4
                }
181
23
            }
182
14
        }
183
10
    }
184
10
    return Status::OK();
185
10
}
186
187
Status CollectionStatistics::process_segment(
188
        const std::string& seg_path, const io::FileSystemSPtr& fs,
189
        const TabletSchema* tablet_schema,
190
2
        const std::unordered_map<std::wstring, CollectInfo>& collect_infos) {
191
2
    auto idx_file_reader = std::make_unique<IndexFileReader>(
192
2
            fs, std::string {InvertedIndexDescriptor::get_index_file_path_prefix(seg_path)},
193
2
            tablet_schema->get_inverted_index_storage_format());
194
2
    RETURN_IF_ERROR(idx_file_reader->init());
195
196
2
    int32_t total_seg_num_docs = 0;
197
2
    for (const auto& [ws_field_name, collect_info] : collect_infos) {
198
0
#ifdef BE_TEST
199
0
        auto compound_reader = DORIS_TRY(idx_file_reader->open(collect_info.index_meta, nullptr));
200
0
        auto* reader = lucene::index::IndexReader::open(compound_reader.get());
201
0
        auto index_searcher = std::make_shared<lucene::search::IndexSearcher>(reader, true);
202
203
0
        auto* index_reader = index_searcher->getReader();
204
#else
205
        InvertedIndexCacheHandle inverted_index_cache_handle;
206
        auto index_file_key = idx_file_reader->get_index_file_cache_key(collect_info.index_meta);
207
        InvertedIndexSearcherCache::CacheKey searcher_cache_key(index_file_key);
208
        if (!InvertedIndexSearcherCache::instance()->lookup(searcher_cache_key,
209
                                                            &inverted_index_cache_handle)) {
210
            auto compound_reader =
211
                    DORIS_TRY(idx_file_reader->open(collect_info.index_meta, nullptr));
212
            auto* reader = lucene::index::IndexReader::open(compound_reader.get());
213
            size_t reader_size = reader->getTermInfosRAMUsed();
214
            auto index_searcher = std::make_shared<lucene::search::IndexSearcher>(reader, true);
215
            auto* cache_value = new InvertedIndexSearcherCache::CacheValue(
216
                    std::move(index_searcher), reader_size, UnixMillis());
217
            InvertedIndexSearcherCache::instance()->insert(searcher_cache_key, cache_value,
218
                                                           &inverted_index_cache_handle);
219
        }
220
221
        auto searcher_variant = inverted_index_cache_handle.get_index_searcher();
222
        auto index_searcher = std::get<FulltextIndexSearcherPtr>(searcher_variant);
223
        auto* index_reader = index_searcher->getReader();
224
#endif
225
226
0
        total_seg_num_docs = std::max(total_seg_num_docs, index_reader->maxDoc());
227
0
        _total_num_tokens[ws_field_name] +=
228
0
                index_reader->sumTotalTermFreq(ws_field_name.c_str()).value_or(0);
229
230
0
        for (const auto& term_info : collect_info.term_infos) {
231
0
            auto iter = TermIterator::create(nullptr, false, index_reader, ws_field_name,
232
0
                                             term_info.get_single_term());
233
0
            _term_doc_freqs[ws_field_name][iter->term()] += iter->doc_freq();
234
0
        }
235
0
    }
236
2
    _total_num_docs += total_seg_num_docs;
237
2
    return Status::OK();
238
2
}
239
240
uint64_t CollectionStatistics::get_term_doc_freq_by_col(const std::wstring& lucene_col_name,
241
9
                                                        const std::wstring& term) {
242
9
    if (!_term_doc_freqs.contains(lucene_col_name)) {
243
1
        throw Exception(ErrorCode::INVERTED_INDEX_CLUCENE_ERROR,
244
1
                        "Index statistics collection failed: Not such column {}",
245
1
                        StringHelper::to_string(lucene_col_name));
246
1
    }
247
248
8
    if (!_term_doc_freqs[lucene_col_name].contains(term)) {
249
0
        throw Exception(ErrorCode::INVERTED_INDEX_CLUCENE_ERROR,
250
0
                        "Index statistics collection failed: Not such term {}",
251
0
                        StringHelper::to_string(term));
252
0
    }
253
254
8
    return _term_doc_freqs[lucene_col_name][term];
255
8
}
256
257
8
uint64_t CollectionStatistics::get_total_term_cnt_by_col(const std::wstring& lucene_col_name) {
258
8
    if (!_total_num_tokens.contains(lucene_col_name)) {
259
2
        throw Exception(ErrorCode::INVERTED_INDEX_CLUCENE_ERROR,
260
2
                        "Index statistics collection failed: Not such column {}",
261
2
                        StringHelper::to_string(lucene_col_name));
262
2
    }
263
264
6
    return _total_num_tokens[lucene_col_name];
265
8
}
266
267
14
uint64_t CollectionStatistics::get_doc_num() const {
268
14
    if (_total_num_docs == 0) {
269
3
        throw Exception(
270
3
                ErrorCode::INVERTED_INDEX_CLUCENE_ERROR,
271
3
                "Index statistics collection failed: No data available for SimilarityCollector");
272
3
    }
273
274
11
    return _total_num_docs;
275
14
}
276
277
7
float CollectionStatistics::get_or_calculate_avg_dl(const std::wstring& lucene_col_name) {
278
7
    auto iter = _avg_dl_by_col.find(lucene_col_name);
279
7
    if (iter != _avg_dl_by_col.end()) {
280
2
        return iter->second;
281
2
    }
282
283
5
    const uint64_t total_term_cnt = get_total_term_cnt_by_col(lucene_col_name);
284
5
    const uint64_t total_doc_cnt = get_doc_num();
285
5
    float avg_dl = total_doc_cnt > 0 ? float((double)total_term_cnt / (double)total_doc_cnt) : 0.0F;
286
5
    _avg_dl_by_col[lucene_col_name] = avg_dl;
287
5
    return avg_dl;
288
7
}
289
290
float CollectionStatistics::get_or_calculate_idf(const std::wstring& lucene_col_name,
291
8
                                                 const std::wstring& term) {
292
8
    auto iter = _idf_by_col_term.find(lucene_col_name);
293
8
    if (iter != _idf_by_col_term.end()) {
294
3
        auto term_iter = iter->second.find(term);
295
3
        if (term_iter != iter->second.end()) {
296
1
            return term_iter->second;
297
1
        }
298
3
    }
299
300
7
    const uint64_t doc_num = get_doc_num();
301
7
    const uint64_t doc_freq = get_term_doc_freq_by_col(lucene_col_name, term);
302
7
    auto idf = (float)std::log(1 + ((double)doc_num - (double)doc_freq + (double)0.5) /
303
7
                                           ((double)doc_freq + (double)0.5));
304
7
    _idf_by_col_term[lucene_col_name][term] = idf;
305
7
    return idf;
306
8
}
307
308
#include "common/compile_check_end.h"
309
} // namespace doris