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

@@ -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<Ordering> {
+impl PartialOrd for TermsWithSegmentOrd {
+    fn partial_cmp(&self, other: &TermsWithSegmentOrd) -> Option<Ordering> {
         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<TermsWithSegmentOrd<'a>>,
-    term_streams_with_segment: Vec<TermsWithSegmentOrd<'a>>,
+pub struct TermMerger {
+    heap: BinaryHeap<TermsWithSegmentOrd>,
+    term_streams_with_segment: Vec<TermsWithSegmentOrd>,
 }
 
-impl<'a> TermMerger<'a> {
+impl TermMerger {
     /// Stream of merged term dictionary
-    pub fn new(term_streams_with_segment: Vec<TermsWithSegmentOrd<'a>>) -> TermMerger<'a> {
+    pub fn new(term_streams_with_segment: Vec<TermsWithSegmentOrd>) -> TermMerger {
         TermMerger {
             heap: BinaryHeap::new(),
             term_streams_with_segment,
         }
     }
 
-    pub(crate) fn matching_segments<'b: 'a>(
-        &'b self,
-    ) -> impl 'b + Iterator<Item = (usize, TermOrdinal)> {
+    pub(crate) fn matching_segments(&self) -> impl Iterator<Item = (usize, TermOrdinal)> {
         self.term_streams_with_segment
             .iter()
             .map(|heap_item| (heap_item.segment_ord, heap_item.terms.term_ord()))

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<RangeSSTable>,
     column_data: &FileSlice,
     format_version: Version,
 ) -> io::Result<Vec<DynamicColumnHandle>> {

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<VoidSSTable>;
 
 pub use common::DateTime;
 

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<TermStreamer<'_>> {
-        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")]

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<W> = sstable::Writer<W, TermInfoValueWriter>;
 
 /// `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<TermSSTable, A>,
+    // 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<TermSSTable, A>) -> 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<TermSSTable, A>;
+
+    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<sstable::Streamer<TermSSTable, A>> for TermStreamer<'a, A>
+where
+    A: Automaton,
+    A::State: Clone,
+{
+    fn from(inner: sstable::Streamer<TermSSTable, A>) -> 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<sstable::StreamerBuilder<'a, TermSSTable, A>> 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<T: AsRef<[u8]>>(self, bound: T) -> Self {
+        TermStreamerBuilder(self.0.ge(bound))
+    }
+
+    /// Limit the range to terms strictly greater than the bound
+    pub fn gt<T: AsRef<[u8]>>(self, bound: T) -> Self {
+        TermStreamerBuilder(self.0.gt(bound))
+    }
+
+    /// Limit the range to terms lesser or equal to the bound
+    pub fn le<T: AsRef<[u8]>>(self, bound: T) -> Self {
+        TermStreamerBuilder(self.0.le(bound))
+    }
+
+    /// Limit the range to terms strictly lesser than the bound
+    pub fn lt<T: AsRef<[u8]>>(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<TermStreamer<'a, A>> {
+        Ok(TermStreamer::new(self.0.into_stream()?))
+    }
+
+    /// Async version of [`TermStreamerBuilder::into_stream`].
+    pub async fn into_stream_async(self) -> io::Result<TermStreamer<'a, A>> {
+        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<TermStreamer<'a, A>> {
+        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;

sstable/src/dictionary.rs

@@ -613,7 +613,7 @@ impl<TSSTable: SSTable> Dictionary<TSSTable> {
     }
 
     /// A stream of all the sorted terms.
-    pub fn stream(&self) -> io::Result<Streamer<'_, TSSTable>> {
+    pub fn stream(&self) -> io::Result<Streamer<TSSTable>> {
         self.range().into_stream()
     }
 

sstable/src/streamer.rs

@@ -128,7 +128,7 @@ where
     fn into_stream_given_delta_reader(
         self,
         delta_reader: DeltaReader<<TSSTable as SSTable>::ValueReader>,
-    ) -> io::Result<Streamer<'a, TSSTable, A>> {
+    ) -> io::Result<Streamer<TSSTable, A>> {
         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<Streamer<'a, TSSTable, A>> {
+    pub async fn into_stream_async(self) -> io::Result<Streamer<TSSTable, A>> {
         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<Streamer<'a, TSSTable, A>> {
+    ) -> io::Result<Streamer<TSSTable, A>> {
         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<Streamer<'a, TSSTable, A>> {
+    pub fn into_stream(self) -> io::Result<Streamer<TSSTable, A>> {
         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<TSSTable, A = AlwaysMatch>
 where
     A: Automaton,
     A::State: Clone,
@@ -202,13 +201,11 @@ where
     lower_bound: Bound<Vec<u8>>,
     upper_bound: Bound<Vec<u8>>,
     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<usize>,
 }
 
-impl<TSSTable> Streamer<'_, TSSTable, AlwaysMatch>
+impl<TSSTable> Streamer<TSSTable, AlwaysMatch>
 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<TSSTable, A> Streamer<'_, TSSTable, A>
+impl<TSSTable, A> Streamer<TSSTable, A>
 where
     A: Automaton,
     A::State: Clone,