github-actions[bot] commented on code in PR #68758: URL: https://github.com/apache/doris/pull/68758#discussion_r4257047318
########## be/src/storage/index/inverted/reader/text_index_reader.cpp: ########## @@ -0,0 +1,676 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "storage/index/inverted/reader/text_index_reader.h" + +#include <algorithm> +#include <map> +#include <memory> +#include <roaring/roaring.hh> +#include <set> +#include <string> +#include <string_view> +#include <type_traits> + +#include "common/config.h" +#include "common/exception.h" +#include "common/logging.h" +#include "common/status.h" +#include "core/data_type/primitive_type.h" +#include "runtime/runtime_profile.h" +#include "runtime/runtime_state.h" +#include "storage/index/index_file_reader.h" +#include "storage/index/index_reader_helper.h" +#include "storage/index/inverted/analysis/analyzer/analyzer.h" +#include "storage/index/inverted/analysis/inverted_index_parser.h" +#include "storage/index/inverted/common/inverted_index_cache.h" +#include "storage/index/inverted/common/inverted_index_query_type.h" +#include "storage/index/inverted/common/single_flight.h" +#include "storage/index/inverted/common/string_helper.h" +#include "storage/index/inverted/inverted_index_iterator.h" +#include "storage/index/inverted/query/engine/all_query/all_query.h" +#include "storage/index/inverted/query/engine/bit_set_query/bit_set_query.h" +#include "storage/index/inverted/query/engine/boolean_query/boolean_query_builder.h" +#include "storage/index/inverted/query/engine/boolean_query/listed_terms.h" +#include "storage/index/inverted/query/engine/boolean_query/operator.h" +#include "storage/index/inverted/query/engine/collect/doc_set_collector.h" +#include "storage/index/inverted/query/engine/complete_null_bitmap.h" +#include "storage/index/inverted/query/engine/expand_query/expand_query.h" +#include "storage/index/inverted/query/engine/phrase_prefix_query/phrase_prefix_query.h" +#include "storage/index/inverted/query/engine/phrase_query/multi_phrase_query.h" +#include "storage/index/inverted/query/engine/phrase_query/phrase_query.h" +#include "storage/index/inverted/query/engine/term_query/term_query.h" +#include "storage/index/inverted/query/exec/docid_set_ops.h" +#include "storage/index/inverted/query/exec/roaring_docid_sink.h" +#include "storage/index/inverted/query/exec/term_pattern.h" +#include "storage/index/inverted/query/logical/match_lowering.h" +#include "storage/index/inverted/query/scoring/bm25_similarity.h" +#include "storage/olap_common.h" +#include "util/defer_op.h" + +namespace doris::segment_v2 { + +namespace logical = index_query::logical; +namespace query_v2 = inverted_index::query_v2; + +namespace { + +namespace logical = index_query::logical; +namespace query_v2 = inverted_index::query_v2; + +query_v2::QueryPtr term_query(const IndexQueryContextPtr& context, const std::wstring& field, + const std::string& term) { + return std::make_shared<query_v2::TermQuery>(context, field, term); +} + +// One term is queried as itself; several form the boolean the set asks for. SEARCH counts a +// threshold before a set reaches here. +query_v2::QueryPtr term_set_query(const IndexQueryContextPtr& context, const std::wstring& field, + const std::string& binding_key, const logical::TermSet& set) { + DORIS_CHECK(set.min_should_match == 0); + if (set.terms.size() == 1) { + return term_query(context, field, set.terms.front()); + } + auto builder = query_v2::create_operator_boolean_query_builder( + set.require_all ? query_v2::OperatorType::OP_AND : query_v2::OperatorType::OP_OR); + for (const auto& term : set.terms) { + builder->add(term_query(context, field, term), binding_key); + } + return builder->build(); +} + +Status phrase_query(const IndexQueryContextPtr& context, const std::wstring& field, + const logical::Phrase& phrase, const roaring::Roaring* candidates, + query_v2::QueryPtr* out) { + const bool single_terms = std::ranges::all_of( + phrase.slots, [](const TermInfo& slot) { return slot.is_single_term(); }); + if (phrase.prefix) { + if (!single_terms) { + return Status::Error<ErrorCode::INVERTED_INDEX_NOT_SUPPORTED>( + "a phrase prefix with several terms at one position is not supported"); + } + *out = std::make_shared<query_v2::PhrasePrefixQuery>(context, field, phrase.slots, + candidates, phrase.suffix); + return Status::OK(); + } + const index_query::PhraseQueryOptions options { + .slop = static_cast<uint32_t>(phrase.slop), + .ordered = phrase.ordered, + .candidates = candidates, + .candidate_rows_consumed = + candidates == nullptr ? nullptr : &context->candidate_rows_consumed}; + if (single_terms) { + *out = std::make_shared<query_v2::PhraseQuery>(context, field, phrase.slots, options); + } else { + *out = std::make_shared<query_v2::MultiPhraseQuery>(context, field, phrase.slots, options); + } + return Status::OK(); +} + +index_query::TermPatternKind pattern_kind(logical::ExpandKind kind) { + switch (kind) { + case logical::ExpandKind::kPrefix: + return index_query::TermPatternKind::kPrefix; + case logical::ExpandKind::kRegexp: + return index_query::TermPatternKind::kRegexp; + case logical::ExpandKind::kContains: + return index_query::TermPatternKind::kContains; + case logical::ExpandKind::kWildcard: + default: + return index_query::TermPatternKind::kWildcard; + } +} + +} // namespace + +Status TextIndexReader::open_source(const IndexQueryContextPtr& context, const std::wstring& field, + std::unique_ptr<OpenedIndex>* opened, + index_query::IndexSourcePtr* source) { + RETURN_IF_ERROR(_open_index(context, opened)); + *source = _bind_source(context, field, **opened); + return Status::OK(); +} + +Status run_leaf(const IndexQueryContextPtr& context, const std::wstring& field, + const logical::Node& leaf, const roaring::Roaring* candidates, bool scoring, + index_query::IndexSourcePtr source, uint32_t doc_count, + const std::shared_ptr<roaring::Roaring>& result) { + SCOPED_RAW_TIMER(&context->stats->inverted_index_searcher_search_timer); + if (const auto* expand = leaf.as<logical::Expand>(); expand != nullptr && !scoring) { + SCOPED_RAW_TIMER(&context->stats->inverted_index_searcher_search_exec_timer); + if (doc_count == 0) { + return Status::OK(); + } + const auto kind = pattern_kind(expand->kind); + index_query::TermPattern pattern; + THROW_IF_ERROR(index_query::TermPattern::create(kind, expand->pattern, &pattern)); + THROW_IF_ERROR(query_v2::collect_expanded_rows( + *source, pattern, + index_query::expansion_limit(kind, index_query::max_expansions(*context)), nullptr, + result.get())); + return Status::OK(); + } + std::span<const std::string> terms; + if (const auto* term = leaf.as<logical::Term>(); term != nullptr) { + terms = std::span(&term->term, 1); + } else if (const auto* set = leaf.as<logical::TermSet>(); + set != nullptr && (!set->require_all || set->terms.size() == 1) && + set->min_should_match == 0) { + terms = set->terms; + } + if (!terms.empty() && (!scoring || terms.size() > 1)) { + SCOPED_RAW_TIMER(&context->stats->inverted_index_searcher_search_exec_timer); + if (!scoring) { + index_query::RoaringDocIdSink sink(*result); + return source->collect_terms(terms, sink); + } + query_v2::ListedTerms listed(std::move(source), nullptr); + for (size_t i = 0; i < terms.size(); ++i) { + auto similarity = std::make_shared<BM25Similarity>(); + similarity->for_one_term(context, field, + inverted_index::StringHelper::to_wstring(terms[i])); + listed.add(i, terms[i], std::move(similarity)); + } + const auto scorer = listed.scored_disjunction(); + if (context->collection_similarity != nullptr) { + *result |= query_v2::collect_scored_rows(scorer, 0, *context->collection_similarity); + } else { + query_v2::collect_true_rows(scorer, result.get()); + } + return Status::OK(); + } + query_v2::WeightPtr weight; + { + SCOPED_RAW_TIMER(&context->stats->inverted_index_searcher_search_init_timer); + query_v2::QueryPtr query; + RETURN_IF_ERROR(plan_query(leaf, context, field, "", candidates, &query)); + weight = query->weight(scoring); + } + SCOPED_RAW_TIMER(&context->stats->inverted_index_searcher_search_exec_timer); + query_v2::QueryExecutionContext exec_ctx; + exec_ctx.segment_num_rows = doc_count; + exec_ctx.field_sources.emplace(field, std::move(source)); + query_v2::collect_multi_segment_doc_set(weight, exec_ctx, "", result, + context->collection_similarity, scoring); + return Status::OK(); +} + +Status plan_query(const logical::Node& leaf, const IndexQueryContextPtr& context, + const std::wstring& field, const std::string& binding_key, + const roaring::Roaring* candidates, query_v2::QueryPtr* out) { + if (const auto* term = leaf.as<logical::Term>()) { + *out = term_query(context, field, term->term); + } else if (const auto* set = leaf.as<logical::TermSet>()) { + *out = term_set_query(context, field, binding_key, *set); + } else if (const auto* phrase = leaf.as<logical::Phrase>()) { + return phrase_query(context, field, *phrase, candidates, out); + } else if (const auto* expand = leaf.as<logical::Expand>()) { + *out = std::make_shared<query_v2::ExpandQuery>(context, field, pattern_kind(expand->kind), + expand->pattern); + } else if (leaf.as<logical::Exists>() != nullptr) { + *out = std::make_shared<query_v2::AllQuery>(field, /*nullable=*/true); + } else if (leaf.as<logical::Empty>() != nullptr) { + *out = std::make_shared<query_v2::BitSetQuery>(roaring::Roaring()); Review Comment: [P1] Preserve field NULLs when analysis produces an empty leaf On a nullable SNII index with parser=standard, support_phrase=true and default stopwords, `search('NOT body:"the"', '{"mode":"standard"}')` over foo and NULL now includes the NULL row, even with scoring and result caching disabled. The PHRASE analyzes to zero tokens, so lower_analyzed_leaf emits logical::Empty, and this branch builds a BitSetQuery without the field NULL bitmap. NOT therefore sees FALSE instead of UNKNOWN. The removed native SNII MATCH_PHRASE path returned an empty result and its SEARCH leaf still read and retained the field NULL bitmap, excluding that row. This is a separate compiler instance from the existing absent-dictionary-term finding: no TermWeight or term cursor is created, so fixing its EmptyScorer branch does not fix this case. Compile a field-bound nullable empty query here and cover analyzer-empty ALL/PHRASE leaves under standard-mode NOT. ########## be/src/storage/index/inverted/query/engine/boolean_query/listed_terms.cpp: ########## @@ -0,0 +1,304 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "storage/index/inverted/query/engine/boolean_query/listed_terms.h" + +#include <parallel_hashmap/phmap.h> + +#include <algorithm> +#include <bit> +#include <optional> +#include <span> +#include <utility> + +#include "common/check.h" +#include "common/exception.h" +#include "storage/index/inverted/query/engine/postings/listed_walk.h" +#include "storage/index/inverted/query/engine/scored_rows_scorer.h" +#include "storage/index/inverted/query/exec/block_doc_set.h" +#include "storage/index/inverted/query/exec/cursor_chained_postings.h" +#include "storage/index/inverted/query/exec/roaring_docid_sink.h" +#include "storage/index/inverted/query/exec/term_waves.h" +#include "storage/index/inverted/query/phrase/position_span.h" + +namespace doris::segment_v2::inverted_index::query_v2 { + +ListedTerms::ListedTerms(index_query::IndexSourcePtr source, + std::shared_ptr<roaring::Roaring> nulls) + : _source(std::move(source)), _nulls(std::move(nulls)) {} + +void ListedTerms::add(size_t clause, std::string term, + index_query::ScoringContextPtr<float> similarity) { + DCHECK(_clauses.empty() || _clauses.back() < clause); + _clauses.push_back(clause); + _terms.push_back(std::move(term)); + _similarities.push_back(std::move(similarity)); +} + +bool ListedTerms::holds(size_t clause) const { + return std::ranges::binary_search(_clauses, clause); +} + +void ListedTerms::open(bool conjunctive, bool scoring) { + if (!conjunctive) { + return; + } + for (const std::string& term : _terms) { + bool held = false; + THROW_IF_ERROR(_source->may_hold(term, &held)); + if (!held) { + _cursors.resize(_terms.size()); + return; + } + } + // A scored conjunction reads the positions of the rows it lists. + THROW_IF_ERROR( + _source->open_terms(_terms, /*positions=*/scoring, /*scoring=*/false, &_cursors)); +} + +bool ListedTerms::has_absent_term() const { + return std::ranges::any_of(_cursors, [](const auto& cursor) { return cursor == nullptr; }); +} + +uint64_t ListedTerms::cheapest_doc_freq() const { + uint64_t cheapest = UINT64_MAX; + for (const auto& cursor : _cursors) { + cheapest = std::min<uint64_t>(cheapest, cursor == nullptr ? 0 : cursor->doc_freq()); + } + return cheapest; +} + +index_query::TruthSet ListedTerms::conjunction(const std::vector<uint32_t>* candidates) { + index_query::TruthSet result; + if (has_absent_term()) { + return result; + } + std::vector<uint32_t> docs; + THROW_IF_ERROR(index_query::chain_cursors(_cursors, candidates, &docs)); + result.true_rows.addMany(docs.size(), docs.data()); + if (_nulls != nullptr) { + result.null_rows = *_nulls; + } + return result; +} + +index_query::TruthSet ListedTerms::disjunction() { + index_query::TruthSet result; + bool any_present = false; + index_query::RoaringDocIdSink sink(result.true_rows); + THROW_IF_ERROR(_source->collect_terms(_terms, sink, &any_present)); + if (any_present && _nulls != nullptr) { Review Comment: [P1] Mask field NULL postings before Boolean composition A valid nullable ARRAY can retain nested values under its outer NULL map (for example, after inserting an array_concat result); ArrayColumnWriter writes those values before recording the outer NULLs. For an SNII field whose NULL row still has an `alpha` posting, `search('arr:alpha OR other:omega', '{"mode":"standard"}')` returns that row even when `other` does not match it. This leaf supplies the row in both true_rows and null_rows, and TruthSet::union_with then removes it from null_rows, so the final mask cannot exclude it. The removed native SNII SEARCH leaf subtracted field NULLs before composing the Boolean query. Both terms can be present in the index, and this reproduces with caching disabled, so it is distinct from the existing absent-term and cache-publication findings. Make every nullable leaf's TRUE/UNKNOWN sets disjoint before Boolean composition, including the nonlisted scorer path, and cover the existing array-null-payload fixture through SEARCH. ########## be/src/exprs/function/search/function_search.cpp: ########## @@ -0,0 +1,634 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "exprs/function/search/function_search.h" + +#include <CLucene/config/repl_wchar.h> +#include <CLucene/debug/error.h> +#include <fmt/format.h> +#include <gen_cpp/Exprs_types.h> +#include <glog/logging.h> + +#include <algorithm> +#include <iterator> +#include <memory> +#include <optional> +#include <roaring/roaring.hh> +#include <string> +#include <unordered_map> +#include <utility> +#include <vector> + +#include "common/config.h" +#include "common/exception.h" +#include "common/status.h" +#include "core/block/columns_with_type_and_name.h" +#include "exprs/function/search/function_search_leaf.h" +#include "exprs/function/search/search_lowering.h" +#include "exprs/function/search/variant_inverted_index_search.h" +#include "exprs/function/simple_function_factory.h" +#include "exprs/vexpr_context.h" +#include "runtime/runtime_profile.h" +#include "storage/index/index_file_reader.h" +#include "storage/index/index_query_context.h" +#include "storage/index/inverted/analysis/analyzer/analyzer.h" +#include "storage/index/inverted/inverted_index_iterator.h" +#include "storage/index/inverted/inverted_index_reader.h" +#include "storage/index/inverted/query/engine/all_query/all_query.h" +#include "storage/index/inverted/query/engine/bit_set_query/bit_set_query.h" +#include "storage/index/inverted/query/engine/boolean_query/boolean_query_builder.h" +#include "storage/index/inverted/query/engine/boolean_query/operator.h" +#include "storage/index/inverted/query/engine/collect/doc_set_collector.h" +#include "storage/index/inverted/query/engine/collect/top_k_collector.h" +#include "storage/index/inverted/query/exec/term_pattern.h" +#include "storage/index/inverted/query/logical/node.h" +#include "storage/index/inverted/reader/collection_statistics.h" +#include "storage/olap_common.h" +#include "storage/segment/variant/nested_group_provider.h" +#include "util/thrift_util.h" + +namespace doris { + +namespace query_v2 = segment_v2::inverted_index::query_v2; + +// Serializes the whole TSearchParam, so every field of the DSL takes part in the signature, and +// adds the limit on the terms PREFIX, WILDCARD and REGEXP clauses expand to. +std::string FunctionSearch::dsl_cache_signature(const TSearchParam& search_param, + const IndexQueryContext* context) { + ThriftSerializer ser(false, 1024); + TSearchParam copy = search_param; + std::string sig; + auto st = ser.serialize(©, &sig); + if (UNLIKELY(!st.ok())) { + LOG(WARNING) << "build_dsl_signature: Thrift serialization failed: " << st.to_string() + << ", caching disabled for this query"; + return ""; + } + sig += "|max_expansions="; + sig += std::to_string( + index_query::max_expansions(context != nullptr ? context->runtime_state : nullptr)); + return sig; +} + +// Extract segment path prefix from the first available inverted index iterator. +// All fields in the same segment share the same path prefix. +static std::string extract_segment_prefix( + const std::unordered_map<std::string, IndexIterator*>& iterators) { + for (const auto& [field_name, iter] : iterators) { + auto* inv_iter = dynamic_cast<InvertedIndexIterator*>(iter); + if (!inv_iter) continue; + // Try fulltext reader first, then string type + for (auto type : + {InvertedIndexReaderType::FULLTEXT, InvertedIndexReaderType::STRING_TYPE}) { + IndexReaderType reader_type = type; + auto reader = inv_iter->get_reader(reader_type); + if (!reader) continue; + auto inv_reader = std::dynamic_pointer_cast<InvertedIndexReader>(reader); + if (!inv_reader) continue; + auto file_reader = inv_reader->get_index_file_reader(); + if (!file_reader) continue; + return file_reader->get_index_path_prefix(); + } + } + VLOG_DEBUG << "extract_segment_prefix: no suitable inverted index reader found across " + << iterators.size() << " iterators, caching disabled for this query"; + return ""; +} + +namespace { + +bool is_nested_group_search_supported() { + auto provider = segment_v2::create_nested_group_read_provider(); + return provider != nullptr && provider->should_enable_nested_group_read_path(); +} + +namespace logical = index_query::logical; + +// Lowering's view of the resolver: what index a field binds to and how that +// index analyzes values. +class SearchFieldCatalog final : public logical::FieldCatalog { +public: + SearchFieldCatalog(FieldReaderResolver& resolver, std::shared_ptr<IndexQueryContext> context) + : _resolver(resolver), _context(std::move(context)) {} + + Status resolve(const std::string& field, InvertedIndexQueryType query_type, + logical::FieldProps* out) override { + FieldReaderBinding binding; + RETURN_IF_ERROR(_resolver.resolve(field, query_type, &binding)); + if (!binding.is_bound()) { + LOG(INFO) << "search: No inverted index for field '" << field + << "' in this segment, query_type=" << static_cast<int>(query_type) + << ", returning UNKNOWN bitmap"; + *out = logical::FieldProps {}; + return Status::OK(); + } + const bool scalar = binding.inverted_reader != nullptr && + binding.inverted_reader->type() == InvertedIndexReaderType::BKD; + const bool analyzed = !scalar && inverted_index::InvertedIndexAnalyzer::should_analyzer( + binding.index_properties); + *out = logical::FieldProps {.bound = true, + .direct_index = scalar, + .analyzed = analyzed, + .binding = binding.binding_key}; + return Status::OK(); + } + + Status analyze(const logical::FieldProps& props, const std::string& value, + std::vector<logical::Token>* out) override { + const FieldReaderBinding* binding = _resolver.find_binding(props.binding); + if (binding == nullptr) { + return Status::InternalError("search: no binding '{}' to analyze with", props.binding); + } + int64_t unused_timer = 0; + SCOPED_RAW_TIMER(_context != nullptr && _context->stats != nullptr + ? &_context->stats->inverted_index_analyzer_timer + : &unused_timer); + InvertedIndexAnalyzerCtxSPtr analyzer_ctx; + RETURN_IF_ERROR(_resolver.analyzer_context_for(props.binding, &analyzer_ctx)); + return inverted_index::InvertedIndexAnalyzer::analyze(value, analyzer_ctx.get(), + binding->index_properties, out); + } + + Status normalize(const logical::FieldProps& props, const std::string& value, + std::string* out) override { + const FieldReaderBinding* binding = _resolver.find_binding(props.binding); + if (binding == nullptr) { + return Status::InternalError("search: no binding '{}' to normalize with", + props.binding); + } + try { + *out = inverted_index::InvertedIndexAnalyzer::normalize(value, + binding->index_properties); + } catch (const CLuceneError& e) { + return Status::Error<ErrorCode::INVERTED_INDEX_ANALYZER_ERROR>( + "search: normalizing '{}' failed: {}", value, e.what()); + } catch (const Exception& e) { + return Status::Error<ErrorCode::INVERTED_INDEX_ANALYZER_ERROR>( + "search: normalizing '{}' failed: {}", value, e.what()); + } + return Status::OK(); + } + +private: + FieldReaderResolver& _resolver; + std::shared_ptr<IndexQueryContext> _context; +}; + +query_v2::Occur to_query_occur(logical::Occur occur) { + switch (occur) { + case logical::Occur::kShould: + return query_v2::Occur::SHOULD; + case logical::Occur::kMustNot: + return query_v2::Occur::MUST_NOT; + case logical::Occur::kMust: + default: + return query_v2::Occur::MUST; + } +} + +Status compile_node(const logical::Node& node, const SearchLeafContext& ctx, + FieldReaderResolver& resolver, query_v2::QueryPtr* out, + std::string* binding_key); + +using Clauses = std::vector<std::pair<logical::Occur, logical::NodePtr>>; + +bool has_required_clause(const Clauses& clauses) { + return std::ranges::any_of( + clauses, [](const auto& clause) { return clause.first == logical::Occur::kMust; }); +} + +// Without scoring, optional clauses beside a required one change neither the rows nor their +// UNKNOWN state, so they are left out. +Clauses matching_clauses(const logical::Bool& boolean, bool scoring) { + if (scoring || boolean.op != logical::BoolOp::kOccur || boolean.min_should_match != 0 || + !has_required_clause(boolean.clauses)) { + return boolean.clauses; + } + Clauses kept; + std::ranges::copy_if(boolean.clauses, std::back_inserter(kept), [](const auto& clause) { + return clause.first != logical::Occur::kShould; + }); + return kept; +} + +// AND, OR and NOT ignore the per-clause occur; OCCUR keeps it and the threshold. A Boolean +// left with one clause after dropping clauses that only score is that clause. +Status compile_bool(const logical::Bool& boolean, const SearchLeafContext& ctx, + FieldReaderResolver& resolver, query_v2::QueryPtr* out) { + const Clauses clauses = matching_clauses(boolean, ctx.scoring); + if (clauses.size() == 1 && boolean.clauses.size() > 1) { + return compile_node(*clauses.front().second, ctx, resolver, out, nullptr); + } + std::vector<query_v2::QueryPtr> queries(clauses.size()); + std::vector<std::string> keys(clauses.size()); + for (size_t i = 0; i < clauses.size(); ++i) { + RETURN_IF_ERROR(compile_node(*clauses[i].second, ctx, resolver, &queries[i], &keys[i])); + } + if (boolean.op == logical::BoolOp::kOccur) { + auto builder = query_v2::create_occur_boolean_query_builder(); + builder->set_minimum_number_should_match(boolean.min_should_match); + for (size_t i = 0; i < clauses.size(); ++i) { + builder->add(queries[i], to_query_occur(clauses[i].first), std::move(keys[i])); + } + *out = builder->build(); + return Status::OK(); + } + query_v2::OperatorType op = query_v2::OperatorType::OP_AND; + if (boolean.op == logical::BoolOp::kOr) { + op = query_v2::OperatorType::OP_OR; + } else if (boolean.op == logical::BoolOp::kNot) { + op = query_v2::OperatorType::OP_NOT; + } + auto builder = query_v2::create_operator_boolean_query_builder(op); + for (size_t i = 0; i < clauses.size(); ++i) { + builder->add(queries[i], std::move(keys[i])); + } + *out = builder->build(); + return Status::OK(); +} + +// A threshold counts the set's terms above the field's compiler, so every format +// answers it the same way: each term is its own leaf on the field. +Status compile_term_threshold(const logical::TermSet& set, const SearchLeafContext& ctx, + SearchLeafCompiler& compiler, query_v2::QueryPtr* out) { + auto builder = query_v2::create_occur_boolean_query_builder(); + builder->set_minimum_number_should_match(set.min_should_match); + const query_v2::Occur occur = set.require_all ? query_v2::Occur::MUST : query_v2::Occur::SHOULD; + for (const auto& term : set.terms) { + const auto leaf = logical::make_node(logical::TermSet { + .field = set.field, .terms = {term}, .require_all = false, .min_should_match = 0}); + query_v2::QueryPtr term_query; + RETURN_IF_ERROR(compiler.compile(*leaf, ctx, &term_query)); + builder->add(term_query, occur, set.field.binding); + } + *out = builder->build(); + return Status::OK(); +} + +// A leaf goes to the compiler of the index it was bound to; its binding key and +// the leaf mapper follow it. A term threshold stays one leaf for the mapper. +Status compile_node(const logical::Node& node, const SearchLeafContext& ctx, + FieldReaderResolver& resolver, query_v2::QueryPtr* out, + std::string* binding_key) { + *out = nullptr; + if (binding_key != nullptr) { + binding_key->clear(); + } + if (node.as<logical::All>() != nullptr) { + *out = std::make_shared<query_v2::AllQuery>(); + return Status::OK(); + } + if (const auto* boolean = node.as<logical::Bool>()) { + return compile_bool(*boolean, ctx, resolver, out); + } + const logical::FieldRef* field = node.field(); + DCHECK(field != nullptr); + if (binding_key != nullptr) { + *binding_key = field->binding; + } + if (node.as<logical::Unknown>() != nullptr) { + *out = make_unknown_leaf_query(ctx.num_rows); + return resolver.map_leaf_query(field->name, out); + } + const FieldReaderBinding* binding = resolver.find_binding(field->binding); + if (binding == nullptr || binding->leaf_compiler == nullptr) { + return Status::InternalError("search: field '{}' has no compiler for binding '{}'", + field->name, field->binding); + } + const auto* set = node.as<logical::TermSet>(); + if (set != nullptr && set->min_should_match > 0) { + RETURN_IF_ERROR(compile_term_threshold(*set, ctx, *binding->leaf_compiler, out)); + } else { + RETURN_IF_ERROR(binding->leaf_compiler->compile(node, ctx, out)); + } + return resolver.map_leaf_query(field->name, out); +} + +} // namespace + +Status FunctionSearch::execute_impl(FunctionContext* /*context*/, Block& /*block*/, + const ColumnNumbers& /*arguments*/, uint32_t /*result*/, + size_t /*input_rows_count*/) const { + return Status::RuntimeError("only inverted index queries are supported"); +} + +// Enhanced implementation: Handle new parameter structure (DSL + SlotReferences) +Status FunctionSearch::evaluate_inverted_index( + const ColumnsWithTypeAndName& arguments, + const std::vector<IndexFieldNameAndTypePair>& data_type_with_names, + std::vector<IndexIterator*> iterators, uint32_t num_rows, + const InvertedIndexAnalyzerCtx* /*analyzer_ctx*/, + InvertedIndexResultBitmap& bitmap_result) const { + return Status::OK(); +} + +Status FunctionSearch::evaluate_inverted_index_with_search_param( + const TSearchParam& search_param, + const std::unordered_map<std::string, IndexFieldNameAndTypePair>& data_type_with_names, + std::unordered_map<std::string, IndexIterator*> iterators, uint32_t num_rows, + InvertedIndexResultBitmap& bitmap_result, bool enable_cache) const { + return evaluate_inverted_index_with_search_param(search_param, data_type_with_names, + std::move(iterators), num_rows, bitmap_result, + enable_cache, nullptr, nullptr); +} + +// A lazy leaf reads its index while the result is listed; what it throws is returned like the +// errors the evaluation returns. +Status FunctionSearch::evaluate_inverted_index_with_search_param( + const TSearchParam& search_param, + const std::unordered_map<std::string, IndexFieldNameAndTypePair>& data_type_with_names, + std::unordered_map<std::string, IndexIterator*> iterators, uint32_t num_rows, + InvertedIndexResultBitmap& bitmap_result, bool enable_cache, + const IndexExecContext* index_exec_ctx, const TabletColumn* nested_column, + const std::shared_ptr<IndexQueryContext>& index_query_context) const { + // VSearchExpr enters here directly, outside IFunction::execute() and its exception boundary, + // so this is where a failure inside the search, such as an analyzer whose first token + // stream throws, has to become a Status. + try { + RETURN_IF_ERROR_OR_CATCH_EXCEPTION(_evaluate_search_param( + search_param, data_type_with_names, std::move(iterators), num_rows, bitmap_result, + enable_cache, index_exec_ctx, nested_column, index_query_context)); + } catch (const CLuceneError& e) { + return Status::Error<ErrorCode::INVERTED_INDEX_ANALYZER_ERROR>("search failed: {}", + e.what()); + } + return Status::OK(); +} + +// The evaluation keeps its result cache, the tree's compile and the collection in one path. +// NOLINTNEXTLINE(readability-function-cognitive-complexity,readability-function-size) +Status FunctionSearch::_evaluate_search_param( + const TSearchParam& search_param, + const std::unordered_map<std::string, IndexFieldNameAndTypePair>& data_type_with_names, + std::unordered_map<std::string, IndexIterator*> iterators, uint32_t num_rows, + InvertedIndexResultBitmap& bitmap_result, bool enable_cache, + const IndexExecContext* index_exec_ctx, const TabletColumn* nested_column, + const std::shared_ptr<IndexQueryContext>& index_query_context) const { + const bool is_nested_query = search_param.root.clause_type == "NESTED"; + if (is_nested_query && !is_nested_group_search_supported()) { + return Status::NotSupported( + "NESTED query requires NestedGroup support, which is unavailable in this build"); + } + + if (!is_nested_query && (iterators.empty() || data_type_with_names.empty())) { + LOG(INFO) << "No indexed columns or iterators available, returning empty result, dsl:" + << search_param.original_dsl; + bitmap_result = InvertedIndexResultBitmap(std::make_shared<roaring::Roaring>(), + std::make_shared<roaring::Roaring>()); + return Status::OK(); + } + + // Track overall query time (equivalent to inverted_index_query_timer in MATCH path). + // Must be declared before the DSL cache lookup so that cache-hit fast paths are + // also covered by the timer. + int64_t query_timer_dummy = 0; + OlapReaderStatistics* outer_stats = index_query_context ? index_query_context->stats : nullptr; + SCOPED_RAW_TIMER(outer_stats ? &outer_stats->inverted_index_query_timer : &query_timer_dummy); + + // DSL result cache only stores bitmap/null bitmap. It does not store BM25 scores, + // so score() queries must execute scorers to populate CollectionSimilarity. + const bool enable_scoring = + index_query_context != nullptr && index_query_context->collection_similarity != nullptr; + auto* dsl_cache = + enable_cache && !enable_scoring ? InvertedIndexQueryCache::instance() : nullptr; + std::string seg_prefix; + std::string dsl_sig; + InvertedIndexQueryCache::CacheKey dsl_cache_key; + bool cache_usable = false; + if (dsl_cache) { + seg_prefix = extract_segment_prefix(iterators); + dsl_sig = dsl_cache_signature(search_param, index_query_context.get()); + if (!seg_prefix.empty() && !dsl_sig.empty()) { + dsl_cache_key = InvertedIndexQueryCache::CacheKey { + seg_prefix, "__search_dsl__", InvertedIndexQueryType::SEARCH_DSL_QUERY, + dsl_sig}; + cache_usable = true; + InvertedIndexQueryCacheHandle dsl_cache_handle; + bool dsl_hit = false; + { + int64_t lookup_dummy = 0; + SCOPED_RAW_TIMER(outer_stats ? &outer_stats->inverted_index_lookup_timer + : &lookup_dummy); + dsl_hit = dsl_cache->lookup(dsl_cache_key, &dsl_cache_handle); + } + if (dsl_hit) { + auto cached_bitmap = dsl_cache_handle.get_bitmap(); + if (cached_bitmap) { + if (outer_stats) { + outer_stats->inverted_index_query_cache_hit++; + } + // Also retrieve cached null bitmap for three-valued SQL logic + // (needed by compound operators NOT, OR, AND in VCompoundPred) + auto null_cache_key = InvertedIndexQueryCache::CacheKey { + seg_prefix, "__search_dsl__", InvertedIndexQueryType::SEARCH_DSL_QUERY, + dsl_sig + "__null"}; + InvertedIndexQueryCacheHandle null_cache_handle; + std::shared_ptr<roaring::Roaring> null_bitmap; + if (dsl_cache->lookup(null_cache_key, &null_cache_handle)) { + null_bitmap = null_cache_handle.get_bitmap(); + } + if (!null_bitmap) { + null_bitmap = std::make_shared<roaring::Roaring>(); + } + bitmap_result = + InvertedIndexResultBitmap(cached_bitmap, std::move(null_bitmap)); + return Status::OK(); + } + } + if (outer_stats) { + outer_stats->inverted_index_query_cache_miss++; + } + } + } + + std::shared_ptr<IndexQueryContext> context; + if (index_query_context) { + context = index_query_context; + } else { + context = std::make_shared<IndexQueryContext>(); + context->collection_statistics = std::make_shared<CollectionStatistics>(); + context->collection_similarity = std::make_shared<CollectionSimilarity>(); + } + + const auto* effective_data_type_with_names = &data_type_with_names; + + // Pass field_bindings to resolver for variant subcolumn detection + FieldReaderResolver resolver(*effective_data_type_with_names, iterators, context, + search_param.field_bindings); + + if (is_nested_query) { + std::shared_ptr<roaring::Roaring> row_bitmap; + VariantNestedSearchEvaluator nested_evaluator(*this); + RETURN_IF_ERROR(nested_evaluator.evaluate(search_param, search_param.root, context, + resolver, num_rows, index_exec_ctx, nested_column, + row_bitmap)); + bitmap_result = InvertedIndexResultBitmap(std::move(row_bitmap), + std::make_shared<roaring::Roaring>()); + bitmap_result.mask_out_null(); + return Status::OK(); + } + + // Extract default_operator from TSearchParam (default: "or") + std::string default_operator = "or"; + if (search_param.__isset.default_operator && !search_param.default_operator.empty()) { + default_operator = search_param.default_operator; + } + // Extract minimum_should_match from TSearchParam (-1 means not set) + int32_t minimum_should_match = -1; + if (search_param.__isset.minimum_should_match) { + minimum_should_match = search_param.minimum_should_match; + } + + auto* stats = context->stats; + int64_t dummy_timer = 0; + SCOPED_RAW_TIMER(stats ? &stats->inverted_index_searcher_search_timer : &dummy_timer); + + query_v2::QueryPtr root_query; + std::string root_binding_key; + { + int64_t init_dummy = 0; + SCOPED_RAW_TIMER(stats ? &stats->inverted_index_searcher_search_init_timer : &init_dummy); + RETURN_IF_ERROR(build_query_recursive(search_param.root, context, resolver, &root_query, + &root_binding_key, default_operator, + minimum_should_match, num_rows, enable_scoring)); + } + if (root_query == nullptr) { + LOG(INFO) << "search: Query tree resolved to empty query, dsl:" + << search_param.original_dsl; + bitmap_result = InvertedIndexResultBitmap(std::make_shared<roaring::Roaring>(), + std::make_shared<roaring::Roaring>()); + return Status::OK(); + } + + VariantSearchNullBitmapAdapter null_resolver(resolver); Review Comment: [P1] Propagate SNII NULL-bitmap read failures Routing SNII through this shared NULL adapter changes a failed NULL-section read into a successful query: VariantSearchNullBitmapAdapter::bitmap_for logs a non-OK read_null_bitmap Status and returns nullptr, which the scorers interpret as no NULL rows. On a cold NULL cache for a scalar SNII field containing foo, bar and NULL, a failed remote read or CRC check confined to the NULL section makes `search('NOT body:foo', '{"mode":"standard"}')` return bar plus the NULL row with Status OK. Opening the source validates its metadata without reading that NULL payload, and VSearch has no earlier NULL filtering. The removed native SNII SEARCH leaf propagated read_null_bitmap failures with RETURN_IF_ERROR. This is distinct from the existing absent-term/cache findings and the ARRAY overlap issue: the term exists, the field is scalar, and the NULL read itself fails. Propagate that error through the existing evaluation exception/Status boundary and cover an isolated failing SNII NULL-section re ad through SEARCH. ########## be/src/storage/index/inverted/snii/reader/snii_postings_cursor.cpp: ########## @@ -0,0 +1,848 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "storage/index/inverted/snii/reader/snii_postings_cursor.h" + +#include <algorithm> +#include <utility> + +#include "common/check.h" +#include "storage/index/inverted/snii/encoding/byte_source.h" +#include "storage/index/inverted/snii/format/frq_pod.h" +#include "storage/index/inverted/snii/format/prx_decode_stats.h" +#include "storage/index/inverted/snii/format/prx_frame.h" +#include "storage/index/inverted/snii/format/prx_pod.h" +#include "storage/index/inverted/snii/reader/windowed_posting.h" + +namespace doris::snii::reader { + +using format::DictEntryEnc; +using format::DictEntryKind; +using format::FrqPreludeReader; +using format::WindowMeta; + +namespace { + +Status posting_corrupted(const char* message) { + return Status::Error<ErrorCode::INVERTED_INDEX_FILE_CORRUPTED, false>(message); +} + +} // namespace + +io::BatchRangeFetcher& SniiReadWave::batch() { + if (_open == nullptr) { + _open = std::make_unique<io::BatchRangeFetcher>(_reader); + } + return *_open; +} + +void SniiReadWave::after_fetch(const void* owner, std::function<Status(const Round&)> completion) { + _completions.push_back({.owner = owner, .run = std::move(completion)}); +} + +void SniiReadWave::drop(const void* owner) { + std::erase_if(_completions, + [owner](const Completion& completion) { return completion.owner == owner; }); + // Every read on the open round came with a completion, so a round none waits for is dropped. + if (_completions.empty()) { + _open.reset(); + } +} + +Status SniiReadWave::fetch() { + if (_open == nullptr) { + DORIS_CHECK(_completions.empty()); + return Status::OK(); + } + std::unique_ptr<io::BatchRangeFetcher> open = std::move(_open); + std::vector<Completion> completions; + completions.swap(_completions); + if (open->pending() > 0) { + RETURN_IF_ERROR(open->fetch()); + } + const Round round(std::move(open)); + ++_fetched_rounds; + std::erase_if(_fetched, [](const auto& fetched) { return fetched.expired(); }); + _fetched.push_back(round); + _peak_held_bytes = std::max(_peak_held_bytes, held_bytes()); + for (const Completion& completion : completions) { + RETURN_IF_ERROR(completion.run(round)); + } + return Status::OK(); +} + +uint64_t SniiReadWave::held_bytes() const { + uint64_t bytes = 0; + for (const auto& fetched : _fetched) { + if (const Round round = fetched.lock()) { + bytes += round->fetched_bytes(); + } + } + return bytes; +} + +SniiPostingsCursor::SniiPostingsCursor(const LogicalIndexReader& idx, format::DictEntry entry, + uint64_t frq_base, uint64_t prx_base, bool positions, + bool scoring, const format::NormsPodReader* norms, + SniiReadWave* wave, format::PrxDecodeStats* prx_stats) + : _idx(idx), + _entry(std::move(entry)), + _frq_base(frq_base), + _prx_base(prx_base), + _positions_wanted(positions), + _scoring(scoring), + _norms(norms), + _wave(wave), + _prx_stats(prx_stats), + _kind(_posting_kind(_entry)) { + if (_wave == nullptr) { + _own_wave = std::make_unique<SniiReadWave>(_idx.reader()); + _wave = _own_wave.get(); + } +} + +SniiPostingsCursor::~SniiPostingsCursor() { + _wave->drop(this); +} + +SniiPostingsCursor::Kind SniiPostingsCursor::_posting_kind(const format::DictEntry& entry) { + if (entry.kind == DictEntryKind::kInline) { + return Kind::kInline; + } + return entry.enc == DictEntryEnc::kWindowed ? Kind::kWindowed : Kind::kSlim; +} + +void SniiPostingsCursor::set_prelude(std::shared_ptr<const FrqPreludeReader> prelude) { + _prelude = std::move(prelude); +} + +Status SniiPostingsCursor::prepare() { + if (_kind == Kind::kWindowed) { + return _read_prelude(); + } + return _kind == Kind::kSlim ? open() : Status::OK(); +} + +Status SniiPostingsCursor::open() { + if (_opened) { + return Status::OK(); + } + switch (_kind) { + case Kind::kInline: { + if (_entry.dd_meta.disk_len > _entry.frq_bytes.size()) { + return posting_corrupted("snii postings: inline dd region exceeds frq bytes"); + } + _windows.assign(1, WindowBytes {}); + _windows[0].dd = + Slice(_entry.frq_bytes.data(), static_cast<size_t>(_entry.dd_meta.disk_len)); + _windows[0].dd_available = true; + _windows[0].prx = Slice(_entry.prx_bytes); + _windows[0].prx_available = true; + _window_count = 1; + break; + } + case Kind::kSlim: + RETURN_IF_ERROR(_open_slim()); + break; + case Kind::kWindowed: + RETURN_IF_ERROR(_read_prelude()); + RETURN_IF_ERROR(_read_span(_wants_prx())); + break; + } + _opened = true; + return _fetch_own(); +} + +// Opens the span unless a prefetch chose the windows, and fetches the wave when this cursor's +// bytes are still on it. +Status SniiPostingsCursor::_ensure_ready() { + if (!_opened) { + RETURN_IF_ERROR(open()); + } + if (_wave->pending()) { + RETURN_IF_ERROR(_wave->fetch()); + } + return Status::OK(); +} + +Status SniiPostingsCursor::_fetch_own() { + return _own_wave == nullptr ? Status::OK() : _wave->fetch(); +} + +Status SniiPostingsCursor::prefetch(const std::vector<uint32_t>* candidates, bool positions) { + if (_kind != Kind::kWindowed) { + // One round holds the whole posting. + return open(); + } + // A scoring cursor decodes every block it lands on, so its frames come with the windows. + const bool prx = positions ? _wants_prx() : _scores_from_prx(); + RETURN_IF_ERROR(_read_prelude()); + if (candidates == nullptr) { + RETURN_IF_ERROR(_read_span(prx)); + } else { + if (_prelude == nullptr) { + // The windows are known only once the prelude arrives. + RETURN_IF_ERROR(_wave->fetch()); + } + std::vector<uint32_t> windows; + _prelude->select_covering_windows(*candidates, &windows); + RETURN_IF_ERROR(_read_windows(windows, prx)); + } + _opened = true; + return _fetch_own(); +} + +Status SniiPostingsCursor::rewind() { + DORIS_CHECK(!_streaming); + _next_window = 0; + _current_window = kNoWindow; + _single_decoded = false; + _bound_known = false; + _positions_decoded = false; + _doc_open = false; + return Status::OK(); +} + +// Counts the frames read for positions, from their headers; a window whose positions were not +// read is left out. +Status SniiPostingsCursor::positions_per_doc(uint64_t* out) { + uint64_t work = 0; + uint64_t docs = 0; + for (uint32_t window = 0; window < _window_count; ++window) { + if (!_windows[window].prx_available) { + continue; + } + RETURN_IF_ERROR(format::add_prx_frames_position_work(_windows[window].prx, &work)); + if (_kind != Kind::kWindowed) { + docs += _entry.df; + continue; + } + WindowMeta meta; + RETURN_IF_ERROR(_prelude->window(window, &meta)); + docs += meta.doc_count; + } + *out = docs == 0 ? 0 : work / docs; + return Status::OK(); +} + +// A block whose positions were already decoded whole serves them as they are. +Status SniiPostingsCursor::stream_positions(std::span<const uint32_t> ordinals) { + if (!_positions_wanted) { + return Status::NotSupported("This posting type does not support positions"); + } + DORIS_CHECK(_current_window != kNoWindow); + DORIS_CHECK(!_streaming); + DORIS_CHECK(!ordinals.empty()); + if (_positions_decoded) { + return Status::OK(); + } + RETURN_IF_ERROR(_ensure_prx(_current_window)); + _stream_context = {.stats = _prx_stats}; + RETURN_IF_ERROR(_stream.reset(_windows[_current_window].prx, static_cast<uint32_t>(_doc_count), + ordinals, &_stream_context)); + _stream_last = ordinals.back(); + _streaming = true; + return Status::OK(); +} + +Status SniiPostingsCursor::_open_slim() { + uint64_t dd_off = 0; + uint64_t dd_len = 0; + RETURN_IF_ERROR(_idx.resolve_frq_window(_entry, _frq_base, &dd_off, &dd_len)); + _windows.assign(1, WindowBytes {}); + _window_count = 1; + std::vector<Piece> pieces {{.offset = dd_off, .length = dd_len, .window = 0, .prx = false}}; + if (_wants_prx()) { + uint64_t prx_off = 0; + uint64_t prx_len = 0; + RETURN_IF_ERROR(_idx.resolve_prx_window(_entry, _prx_base, &prx_off, &prx_len)); + pieces.push_back({.offset = prx_off, .length = prx_len, .window = 0, .prx = true}); + } + return _read_pieces(std::move(pieces)); +} + +Status SniiPostingsCursor::_read_prelude() { + if (!_windows.empty() || _prelude_pending) { + return Status::OK(); + } + const auto adopt = [this](std::shared_ptr<const FrqPreludeReader> prelude) -> Status { + if (_wants_prx() && !prelude->has_prx()) { + return posting_corrupted("snii postings: positions requested but prelude has none"); + } + _prelude = std::move(prelude); + _window_count = _prelude->n_windows(); + _windows.assign(_window_count, WindowBytes {}); + _prelude_pending = false; + return Status::OK(); + }; + if (_prelude != nullptr) { + return adopt(_prelude); + } + uint64_t prelude_abs = 0; + RETURN_IF_ERROR(prelude_abs_offset(_idx, _entry, _frq_base, &prelude_abs)); + const size_t handle = _wave->batch().add(prelude_abs, _entry.prelude_len); + _prelude_pending = true; + _wave->after_fetch(this, [adopt, handle](const SniiReadWave::Round& round) { + auto prelude = std::make_shared<FrqPreludeReader>(); + RETURN_IF_ERROR(FrqPreludeReader::open(round->get(handle), prelude.get())); + return adopt(std::move(prelude)); + }); + return Status::OK(); +} + +// Reads the whole dd-block and, when asked, the PRX region next to it, in one round. +Status SniiPostingsCursor::_read_span(bool prx) { + const bool need_dd = !_span_dd_read; + const bool need_prx = prx && !_span_prx_read; + if (!need_dd && !need_prx) { + return Status::OK(); + } + _span_dd_read = true; + _span_prx_read = _span_prx_read || need_prx; + if (_prelude == nullptr) { + return _read_regions(need_dd, need_prx); + } + std::vector<Piece> pieces; + for (uint32_t w = 0; w < _window_count; ++w) { + WindowAbsRange range; + RETURN_IF_ERROR(windowed_window_range(_idx, _entry, _frq_base, _prx_base, *_prelude, w, + need_prx, &range)); + if (need_dd && !_windows[w].dd_available) { + pieces.push_back({.offset = range.dd_off, .length = range.dd_len, .window = w}); + } + if (need_prx && !_windows[w].prx_available) { + pieces.push_back( + {.offset = range.prx_off, .length = range.prx_len, .window = w, .prx = true}); + } + } + return _read_pieces(std::move(pieces)); +} + +Status SniiPostingsCursor::_read_regions(bool dd, bool prx) { + DCHECK(_prelude_pending); + uint64_t dd_off = 0; + uint64_t dd_len = 0; + RETURN_IF_ERROR(_idx.resolve_frq_window(_entry, _frq_base, &dd_off, &dd_len)); + uint64_t prx_off = 0; + uint64_t prx_len = 0; + if (prx) { + RETURN_IF_ERROR(_idx.resolve_prx_window(_entry, _prx_base, &prx_off, &prx_len)); + } + io::BatchRangeFetcher& batch = _wave->batch(); + const size_t dd_handle = dd ? batch.add(dd_off, dd_len) : 0; + const size_t prx_handle = prx ? batch.add(prx_off, prx_len) : 0; + // The prelude's completion, registered before this one, has parsed it by now. + const auto slice = [this, dd, prx, dd_handle, prx_handle, dd_off, + prx_off](const SniiReadWave::Round& round) -> Status { + _keep(round); + const io::BatchRangeFetcher& fetched = *round; + for (uint32_t w = 0; w < _window_count; ++w) { + WindowAbsRange range; + RETURN_IF_ERROR(windowed_window_range(_idx, _entry, _frq_base, _prx_base, *_prelude, w, + prx, &range)); + WindowBytes& window = _windows[w]; + if (dd) { + window.dd = + fetched.get(dd_handle).subslice(static_cast<size_t>(range.dd_off - dd_off), + static_cast<size_t>(range.dd_len)); + window.dd_available = true; + } + if (prx) { + window.prx = fetched.get(prx_handle) + .subslice(static_cast<size_t>(range.prx_off - prx_off), + static_cast<size_t>(range.prx_len)); + window.prx_available = true; + } + } + return Status::OK(); + }; + _wave->after_fetch(this, slice); + return Status::OK(); +} + +void SniiPostingsCursor::_keep(const SniiReadWave::Round& round) { + if (_rounds.empty() || _rounds.back() != round) { + _rounds.push_back(round); + } +} + +// Reads the docids of `windows` not yet read, and their PRX frames when `prx`, in one round. +Status SniiPostingsCursor::_read_windows(const std::vector<uint32_t>& windows, bool prx) { + std::vector<Piece> pieces; + for (const uint32_t w : windows) { + WindowMeta meta; + RETURN_IF_ERROR(_prelude->window(w, &meta)); + bool dense = false; + RETURN_IF_ERROR(is_dense_full_window(meta, w, &dense)); + WindowBytes& bytes = _windows[w]; + if (dense) { + // A full window lists its range; no read holds its docids. + bytes.dd_available = true; + } + const bool need_dd = !bytes.dd_available; + const bool need_prx = prx && !bytes.prx_available; + if (!need_dd && !need_prx) { + continue; + } + WindowAbsRange range; + RETURN_IF_ERROR(windowed_window_range(_idx, _entry, _frq_base, _prx_base, *_prelude, w, + need_prx, &range)); + if (need_dd) { + pieces.push_back({.offset = range.dd_off, .length = range.dd_len, .window = w}); + } + if (need_prx) { + pieces.push_back( + {.offset = range.prx_off, .length = range.prx_len, .window = w, .prx = true}); + } + } + return _read_pieces(std::move(pieces)); +} + +Status SniiPostingsCursor::_read_pieces(std::vector<Piece> pieces) { + if (pieces.empty()) { + return Status::OK(); + } + std::ranges::sort(pieces, {}, &Piece::offset); + // Pieces within the same-term gap read as one range; each keeps its run and offset in it. + struct Run { + uint64_t offset = 0; + uint64_t end = 0; + size_t handle = 0; + }; + std::vector<Run> runs; + std::vector<size_t> run_of(pieces.size()); + for (size_t i = 0; i < pieces.size(); ++i) { + const Piece& piece = pieces[i]; + if (runs.empty() || piece.offset > runs.back().end + kSameTermCoalesceGap) { + runs.push_back({.offset = piece.offset, .end = piece.offset + piece.length}); + } else { + runs.back().end = std::max(runs.back().end, piece.offset + piece.length); Review Comment: [P2] Preserve sparse window selection when coalescing position reads `prefetch(candidates)` selects only covering windows, but `_read_pieces` always merges gaps up to 16 KiB. A scored SNII conjunction uses this path to fetch a common term's positions only for the intersection rows. With 1 KiB PRX windows and one selected window every sixteen windows, the 15 KiB discarded gaps all merge into one nearly full-posting read, instead of reading approximately one sixteenth of its position bytes. The removed accumulate_windowed_candidate_scores deliberately used a zero coalescing gap for sparse selections to preserve that bound; the new cursor test avoids nonadjacent windows because they would read the whole block. Pass the sparse-selection policy through to these reads, retaining the wider gap for full scans, and add a metered nonadjacent-window case. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
