/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 |