airborne12 commented on code in PR #68758: URL: https://github.com/apache/doris/pull/68758#discussion_r4257615773
########## 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: Confirmed and fixed in 1d781d703ae. `SniiPostingsCursor::_read_windows` now reads a selection of some windows without coalescing, so the round holds only the docids and frames of those windows; a read of every window keeps `kSameTermCoalesceGap`, as the removed `accumulate_windowed_candidate_scores` did. `SniiPostingsCursor.CandidatesInDistantWindowsReadOnlyTheirWindows` prefetches candidates in the first and last windows of a windowed posting with positions and expects the pair to read exactly the bytes of the two single-window reads, less one shared prelude. Before the fix every one of these reads spanned most of that posting, so the pair did not match (3228 bytes against the expected 3664). -- 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]
