From 0673cc06ca684ffebf56e5570474bfecba889d59 Mon Sep 17 00:00:00 2001 From: trinity-1686a Date: Wed, 7 Oct 2026 15:55:59 +0200 Subject: [PATCH] remove lifetime from sstable steamer this can allow removing one call to termdict.search() in warm_postings_automaton --- columnar/src/columnar/merge/term_merger.rs | 30 +++-- columnar/src/columnar/reader/mod.rs | 2 +- columnar/src/lib.rs | 2 +- src/termdict/mod.rs | 27 ++++- src/termdict/sstable_termdict/mod.rs | 135 ++++++++++++++++++++- sstable/src/dictionary.rs | 2 +- sstable/src/streamer.rs | 18 ++- 7 files changed, 180 insertions(+), 36 deletions(-) diff --git a/columnar/src/columnar/merge/term_merger.rs b/columnar/src/columnar/merge/term_merger.rs index 710af485a..9136c98bd 100644 --- a/columnar/src/columnar/merge/term_merger.rs +++ b/columnar/src/columnar/merge/term_merger.rs @@ -6,27 +6,27 @@ use sstable::TermOrdinal; use crate::Streamer; /// The terms of a column with the ordinal of the segment. -pub struct TermsWithSegmentOrd<'a> { - pub terms: Streamer<'a>, +pub struct TermsWithSegmentOrd { + pub terms: Streamer, pub segment_ord: usize, } -impl PartialEq for TermsWithSegmentOrd<'_> { +impl PartialEq for TermsWithSegmentOrd { fn eq(&self, other: &Self) -> bool { self.segment_ord == other.segment_ord } } -impl Eq for TermsWithSegmentOrd<'_> {} +impl Eq for TermsWithSegmentOrd {} -impl<'a> PartialOrd for TermsWithSegmentOrd<'a> { - fn partial_cmp(&self, other: &TermsWithSegmentOrd<'a>) -> Option { +impl PartialOrd for TermsWithSegmentOrd { + fn partial_cmp(&self, other: &TermsWithSegmentOrd) -> Option { Some(self.cmp(other)) } } -impl<'a> Ord for TermsWithSegmentOrd<'a> { - fn cmp(&self, other: &TermsWithSegmentOrd<'a>) -> Ordering { +impl Ord for TermsWithSegmentOrd { + fn cmp(&self, other: &TermsWithSegmentOrd) -> Ordering { (&other.terms.key(), &other.segment_ord).cmp(&(&self.terms.key(), &self.segment_ord)) } } @@ -37,23 +37,21 @@ impl<'a> Ord for TermsWithSegmentOrd<'a> { /// The item yield is actually a pair with /// - the term /// - a slice with the ordinal of the segments containing the terms. -pub struct TermMerger<'a> { - heap: BinaryHeap>, - term_streams_with_segment: Vec>, +pub struct TermMerger { + heap: BinaryHeap, + term_streams_with_segment: Vec, } -impl<'a> TermMerger<'a> { +impl TermMerger { /// Stream of merged term dictionary - pub fn new(term_streams_with_segment: Vec>) -> TermMerger<'a> { + pub fn new(term_streams_with_segment: Vec) -> TermMerger { TermMerger { heap: BinaryHeap::new(), term_streams_with_segment, } } - pub(crate) fn matching_segments<'b: 'a>( - &'b self, - ) -> impl 'b + Iterator { + pub(crate) fn matching_segments(&self) -> impl Iterator { self.term_streams_with_segment .iter() .map(|heap_item| (heap_item.segment_ord, heap_item.terms.term_ord())) diff --git a/columnar/src/columnar/reader/mod.rs b/columnar/src/columnar/reader/mod.rs index 198f659de..1a93396d7 100644 --- a/columnar/src/columnar/reader/mod.rs +++ b/columnar/src/columnar/reader/mod.rs @@ -53,7 +53,7 @@ impl fmt::Debug for ColumnarReader { /// It takes a stream from the column sstable and return the list of /// `DynamicColumn` available in it. fn read_all_columns_in_stream( - mut stream: sstable::Streamer<'_, RangeSSTable>, + mut stream: sstable::Streamer, column_data: &FileSlice, format_version: Version, ) -> io::Result> { diff --git a/columnar/src/lib.rs b/columnar/src/lib.rs index 1da8d9604..5555c886a 100644 --- a/columnar/src/lib.rs +++ b/columnar/src/lib.rs @@ -59,7 +59,7 @@ pub struct RowAddr { } pub use sstable::{Dictionary, TermOrdHit}; -pub type Streamer<'a> = sstable::Streamer<'a, VoidSSTable>; +pub type Streamer = sstable::Streamer; pub use common::DateTime; diff --git a/src/termdict/mod.rs b/src/termdict/mod.rs index 4000b08d4..cb7cfe377 100644 --- a/src/termdict/mod.rs +++ b/src/termdict/mod.rs @@ -141,19 +141,40 @@ impl TermDictionary { /// Returns a range builder, to stream all of the terms /// within an interval. pub fn range(&self) -> TermStreamerBuilder<'_> { - self.0.range() + #[cfg(feature = "quickwit")] + { + self.0.range().into() + } + #[cfg(not(feature = "quickwit"))] + { + self.0.range() + } } /// A stream of all the sorted terms. pub fn stream(&self) -> io::Result> { - self.0.stream() + #[cfg(feature = "quickwit")] + { + self.0.stream().map(Into::into) + } + #[cfg(not(feature = "quickwit"))] + { + self.0.stream() + } } /// Returns a search builder, to stream all of the terms /// within the Automaton pub fn search<'a, A: Automaton + 'a>(&'a self, automaton: A) -> TermStreamerBuilder<'a, A> where A::State: Clone { - self.0.search(automaton) + #[cfg(feature = "quickwit")] + { + self.0.search(automaton).into() + } + #[cfg(not(feature = "quickwit"))] + { + self.0.search(automaton) + } } #[cfg(feature = "quickwit")] diff --git a/src/termdict/sstable_termdict/mod.rs b/src/termdict/sstable_termdict/mod.rs index cc82eba3c..b9d200546 100644 --- a/src/termdict/sstable_termdict/mod.rs +++ b/src/termdict/sstable_termdict/mod.rs @@ -8,6 +8,7 @@ use common::VInt; use sstable::value::{ValueReader, ValueWriter}; use sstable::SSTable; use tantivy_fst::automaton::AlwaysMatch; +use tantivy_fst::Automaton; pub use self::merger::TermMerger; use crate::postings::TermInfo; @@ -25,14 +26,142 @@ pub type TermDictionaryBuilder = sstable::Writer; /// `TermStreamer` acts as a cursor over a range of terms of a segment. /// Terms are guaranteed to be sorted. -pub type TermStreamer<'a, A = AlwaysMatch> = sstable::Streamer<'a, TermSSTable, A>; +pub struct TermStreamer<'a, A = AlwaysMatch> +where + A: Automaton, + A::State: Clone, +{ + inner: sstable::Streamer, + // needed to share the same interface as FST-based term dictionary + _phantom: std::marker::PhantomData<&'a ()>, +} + +impl<'a, A> TermStreamer<'a, A> +where + A: Automaton, + A::State: Clone, +{ + pub(crate) fn new(inner: sstable::Streamer) -> Self { + TermStreamer { + inner, + _phantom: std::marker::PhantomData, + } + } +} + +impl<'a, A> std::ops::Deref for TermStreamer<'a, A> +where + A: Automaton, + A::State: Clone, +{ + type Target = sstable::Streamer; + + fn deref(&self) -> &Self::Target { + &self.inner + } +} + +impl<'a, A> std::ops::DerefMut for TermStreamer<'a, A> +where + A: Automaton, + A::State: Clone, +{ + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.inner + } +} + +impl<'a, A> From> for TermStreamer<'a, A> +where + A: Automaton, + A::State: Clone, +{ + fn from(inner: sstable::Streamer) -> Self { + TermStreamer::new(inner) + } +} + +/// `TermStreamerBuilder` is a helper object used to define a range of terms that should be +/// streamed. +pub struct TermStreamerBuilder<'a, A = AlwaysMatch>(sstable::StreamerBuilder<'a, TermSSTable, A>) +where + A: Automaton, + A::State: Clone; + +impl<'a, A> From> for TermStreamerBuilder<'a, A> +where + A: Automaton, + A::State: Clone, +{ + fn from(inner: sstable::StreamerBuilder<'a, TermSSTable, A>) -> Self { + TermStreamerBuilder(inner) + } +} + +impl<'a, A> TermStreamerBuilder<'a, A> +where + A: Automaton, + A::State: Clone, +{ + /// Limit the range to terms greater or equal to the bound + pub fn ge>(self, bound: T) -> Self { + TermStreamerBuilder(self.0.ge(bound)) + } + + /// Limit the range to terms strictly greater than the bound + pub fn gt>(self, bound: T) -> Self { + TermStreamerBuilder(self.0.gt(bound)) + } + + /// Limit the range to terms lesser or equal to the bound + pub fn le>(self, bound: T) -> Self { + TermStreamerBuilder(self.0.le(bound)) + } + + /// Limit the range to terms strictly lesser than the bound + pub fn lt>(self, bound: T) -> Self { + TermStreamerBuilder(self.0.lt(bound)) + } + + /// Load no more data than what's required to get `limit` + /// matching entries. + /// + /// The resulting [`TermStreamer`] can still return marginally + /// more than `limit` elements. + pub fn limit(self, limit: u64) -> Self { + TermStreamerBuilder(self.0.limit(limit)) + } + + /// Creates the stream corresponding to the range + /// of terms defined using the `TermStreamerBuilder`. + pub fn into_stream(self) -> io::Result> { + Ok(TermStreamer::new(self.0.into_stream()?)) + } + + /// Async version of [`TermStreamerBuilder::into_stream`]. + pub async fn into_stream_async(self) -> io::Result> { + Ok(TermStreamer::new(self.0.into_stream_async().await?)) + } + + /// Same as [`TermStreamerBuilder::into_stream_async`], but tries to issue a + /// single io operation when requesting blocks that are not consecutive, + /// but also less than `merge_holes_under_bytes` bytes apart. + pub async fn into_stream_async_merging_holes( + self, + merge_holes_under_bytes: usize, + ) -> io::Result> { + Ok(TermStreamer::new( + self.0 + .into_stream_async_merging_holes(merge_holes_under_bytes) + .await?, + )) + } +} /// SSTable used to store TermInfo objects. #[derive(Clone)] pub struct TermSSTable; -pub type TermStreamerBuilder<'a, A = AlwaysMatch> = sstable::StreamerBuilder<'a, TermSSTable, A>; - impl SSTable for TermSSTable { type Value = TermInfo; type ValueReader = TermInfoValueReader; diff --git a/sstable/src/dictionary.rs b/sstable/src/dictionary.rs index 69b57053f..8122f78c5 100644 --- a/sstable/src/dictionary.rs +++ b/sstable/src/dictionary.rs @@ -613,7 +613,7 @@ impl Dictionary { } /// A stream of all the sorted terms. - pub fn stream(&self) -> io::Result> { + pub fn stream(&self) -> io::Result> { self.range().into_stream() } diff --git a/sstable/src/streamer.rs b/sstable/src/streamer.rs index 9e5d11541..30ef4f652 100644 --- a/sstable/src/streamer.rs +++ b/sstable/src/streamer.rs @@ -128,7 +128,7 @@ where fn into_stream_given_delta_reader( self, delta_reader: DeltaReader<::ValueReader>, - ) -> io::Result> { + ) -> io::Result> { let start_state = self.automaton.start(); let start_key = bound_as_byte_slice(&self.lower); @@ -159,12 +159,11 @@ where lower_bound: self.lower, upper_bound: self.upper, upper_bound_comparator: DeltaKeyComparator::new(), - _lifetime: std::marker::PhantomData, }) } /// See `into_stream(..)` - pub async fn into_stream_async(self) -> io::Result> { + pub async fn into_stream_async(self) -> io::Result> { self.into_stream_async_merging_holes(0).await } @@ -173,14 +172,14 @@ where pub async fn into_stream_async_merging_holes( self, merge_holes_under_bytes: usize, - ) -> io::Result> { + ) -> io::Result> { let delta_reader = self.delta_reader_async(merge_holes_under_bytes).await?; self.into_stream_given_delta_reader(delta_reader) } /// Creates the stream corresponding to the range /// of terms defined using the `StreamerBuilder`. - pub fn into_stream(self) -> io::Result> { + pub fn into_stream(self) -> io::Result> { let delta_reader = self.delta_reader()?; self.into_stream_given_delta_reader(delta_reader) } @@ -188,7 +187,7 @@ where /// `Streamer` acts as a cursor over a range of terms of a segment. /// Terms are guaranteed to be sorted. -pub struct Streamer<'a, TSSTable, A = AlwaysMatch> +pub struct Streamer where A: Automaton, A::State: Clone, @@ -202,13 +201,11 @@ where lower_bound: Bound>, upper_bound: Bound>, upper_bound_comparator: DeltaKeyComparator, - // this field is used to please the type-interface of a dictionary in tantivy - _lifetime: std::marker::PhantomData<&'a ()>, lower_bound_reached: bool, always_match_at: Option, } -impl Streamer<'_, TSSTable, AlwaysMatch> +impl Streamer where TSSTable: SSTable { pub fn empty() -> Self { @@ -223,12 +220,11 @@ where TSSTable: SSTable lower_bound: Bound::Unbounded, upper_bound: Bound::Unbounded, upper_bound_comparator: DeltaKeyComparator::new(), - _lifetime: std::marker::PhantomData, } } } -impl Streamer<'_, TSSTable, A> +impl Streamer where A: Automaton, A::State: Clone,