From a1447cc9c2ecb2831c37df1255e162aaca4b2722 Mon Sep 17 00:00:00 2001 From: trinity-1686a Date: Thu, 19 Dec 2024 17:30:05 +0100 Subject: [PATCH] remove breaking change in sstable public api --- columnar/src/columnar/reader/mod.rs | 2 +- src/index/inverted_index_reader.rs | 4 +++- sstable/src/streamer.rs | 12 +++++++++--- 3 files changed, 13 insertions(+), 5 deletions(-) diff --git a/columnar/src/columnar/reader/mod.rs b/columnar/src/columnar/reader/mod.rs index 832fdb7da..f850e4f6f 100644 --- a/columnar/src/columnar/reader/mod.rs +++ b/columnar/src/columnar/reader/mod.rs @@ -170,7 +170,7 @@ impl ColumnarReader { ) -> io::Result> { let stream = self .stream_for_column_range(column_name) - .into_stream_async(0) + .into_stream_async() .await?; read_all_columns_in_stream(stream, &self.column_data, self.format_version) } diff --git a/src/index/inverted_index_reader.rs b/src/index/inverted_index_reader.rs index 5781f9df3..d3f8f4640 100644 --- a/src/index/inverted_index_reader.rs +++ b/src/index/inverted_index_reader.rs @@ -253,7 +253,9 @@ impl InvertedIndexReader { range_builder }; - let mut stream = range_builder.into_stream_async(merge_holes_under).await?; + let mut stream = range_builder + .into_stream_async_merging_holes(merge_holes_under) + .await?; let iter = std::iter::from_fn(move || stream.next().map(|(_k, v)| v.clone())); diff --git a/sstable/src/streamer.rs b/sstable/src/streamer.rs index e1c34a538..de727cec4 100644 --- a/sstable/src/streamer.rs +++ b/sstable/src/streamer.rs @@ -137,11 +137,17 @@ where } /// See `into_stream(..)` - pub async fn into_stream_async( + pub async fn into_stream_async(self) -> io::Result> { + self.into_stream_async_merging_holes(0).await + } + + /// Same as `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 appart. + pub async fn into_stream_async_merging_holes( self, - merge_holes_undex: usize, + merge_holes_under: usize, ) -> io::Result> { - let delta_reader = self.delta_reader_async(merge_holes_undex).await?; + let delta_reader = self.delta_reader_async(merge_holes_under).await?; self.into_stream_given_delta_reader(delta_reader) }