Merge changes (#47, #48, #71, #74)

eeeebbbbrrrr · rebasedming · commit 0f8e51d406e0 · 2026-02-10T17:04:09.000-05:00
* perf: remove general overhead during segment merging (#47) * changes to make merging work (#48) * fix: Use smaller buffers during merging (#71) * chore: Use smaller merge buffers. (#74)
diff --git a/common/src/buffered_file_slice.rs b/common/src/buffered_file_slice.rs
@@ -0,0 +1,106 @@
+use std::cell::RefCell;
+use std::cmp::min;
+use std::io;
+use std::ops::Range;
+
+use super::file_slice::FileSlice;
+use super::{HasLen, OwnedBytes};
+
+const DEFAULT_BUFFER_MAX_SIZE: usize = 512 * 1024; // 512K
+
+/// A buffered reader for a FileSlice.
+///
+/// Reads the underlying `FileSlice` in large, sequential chunks to amortize
+/// the cost of `read_bytes` calls, while keeping peak memory usage under control.
+///
+/// TODO: Rather than wrapping a `FileSlice` in buffering, it will usually be better to adjust a
+/// `FileHandle` to directly handle buffering itself.
+/// TODO: See: https://github.com/paradedb/paradedb/issues/3374
+pub struct BufferedFileSlice {
+    file_slice: FileSlice,
+    buffer: RefCell<OwnedBytes>,
+    buffer_range: RefCell<Range<u64>>,
+    buffer_max_size: usize,
+}
+
+impl BufferedFileSlice {
+    /// Creates a new `BufferedFileSlice`.
+    ///
+    /// The `buffer_max_size` is the amount of data that will be read from the
+    /// `FileSlice` on a buffer miss.
+    pub fn new(file_slice: FileSlice, buffer_max_size: usize) -> Self {
+        Self {
+            file_slice,
+            buffer: RefCell::new(OwnedBytes::empty()),
+            buffer_range: RefCell::new(0..0),
+            buffer_max_size,
+        }
+    }
+
+    /// Creates a new `BufferedFileSlice` with a default buffer max size.
+    pub fn new_with_default_buffer_size(file_slice: FileSlice) -> Self {
+        Self::new(file_slice, DEFAULT_BUFFER_MAX_SIZE)
+    }
+
+    /// Creates an empty `BufferedFileSlice`.
+    pub fn empty() -> Self {
+        Self::new(FileSlice::empty(), 0)
+    }
+
+    /// Returns an `OwnedBytes` corresponding to the given `required_range`.
+    ///
+    /// If the requested range is not in the buffer, this will trigger a read
+    /// from the underlying `FileSlice`.
+    ///
+    /// If the requested range is larger than the buffer_max_size, it will be read directly from the
+    /// source without buffering.
+    ///
+    /// # Errors
+    ///
+    /// Returns an `io::Error` if the underlying read fails or the range is
+    /// out of bounds.
+    pub fn get_bytes(&self, required_range: Range<u64>) -> io::Result<OwnedBytes> {
+        let buffer_range = self.buffer_range.borrow();
+
+        // Cache miss condition: the required range is not fully contained in the current buffer.
+        if required_range.start < buffer_range.start || required_range.end > buffer_range.end {
+            drop(buffer_range); // release borrow before mutating
+
+            if required_range.end > self.file_slice.len() as u64 {
+                return Err(io::Error::new(
+                    io::ErrorKind::UnexpectedEof,
+                    "Requested range extends beyond the end of the file slice.",
+                ));
+            }
+
+            if (required_range.end - required_range.start) as usize > self.buffer_max_size {
+                // This read is larger than our buffer max size.
+                // Read it directly and bypass the buffer to avoid churning.
+                return self
+                    .file_slice
+                    .read_bytes_slice(required_range.start as usize..required_range.end as usize);
+            }
+
+            let new_buffer_start = required_range.start;
+            let new_buffer_end = min(
+                new_buffer_start + self.buffer_max_size as u64,
+                self.file_slice.len() as u64,
+            );
+            let read_range = new_buffer_start..new_buffer_end;
+
+            let new_buffer = self
+                .file_slice
+                .read_bytes_slice(read_range.start as usize..read_range.end as usize)?;
+
+            self.buffer.replace(new_buffer);
+            self.buffer_range.replace(read_range);
+        }
+
+        // Now the data is guaranteed to be in the buffer.
+        let buffer = self.buffer.borrow();
+        let buffer_range = self.buffer_range.borrow();
+        let local_start = (required_range.start - buffer_range.start) as usize;
+        let local_end = (required_range.end - buffer_range.start) as usize;
+        Ok(buffer.slice(local_start..local_end))
+    }
+}
diff --git a/common/src/lib.rs b/common/src/lib.rs
@@ -6,6 +6,7 @@ pub use byteorder::LittleEndian as Endianness;
 
 mod bitset;
 pub mod bounds;
+pub mod buffered_file_slice;
 mod byte_count;
 mod datetime;
 pub mod file_slice;
diff --git a/runtests.sh b/runtests.sh
@@ -1,3 +1,3 @@
 #! /bin/bash
 
-cargo +stable nextest run --features mmap,stopwords,lz4-compression,zstd-compression,failpoints --verbose --workspace
+cargo +stable nextest run --features quickwit,mmap,stopwords,lz4-compression,zstd-compression,failpoints --verbose --workspace
diff --git a/src/directory/mod.rs b/src/directory/mod.rs
@@ -18,6 +18,7 @@ mod composite_file;
 use std::io::BufWriter;
 use std::path::PathBuf;
 
+pub use common::buffered_file_slice::BufferedFileSlice;
 pub use common::file_slice::{FileHandle, FileSlice};
 pub use common::{AntiCallToken, OwnedBytes, TerminatingWrite};
 
diff --git a/src/index/merge_optimized_inverted_index_reader.rs b/src/index/merge_optimized_inverted_index_reader.rs
@@ -1,8 +1,6 @@
 use std::io;
 
-use common::OwnedBytes;
-
-use crate::directory::FileSlice;
+use crate::directory::{BufferedFileSlice, FileSlice};
 use crate::positions::PositionReader;
 use crate::postings::{BlockSegmentPostings, SegmentPostings, TermInfo};
 use crate::schema::IndexRecordOption;
@@ -11,16 +9,17 @@ use crate::termdict::TermDictionary;
 /// The inverted index reader is in charge of accessing
 /// the inverted index associated with a specific field.
 ///
-/// This is optimized for merging in that it full reads
-/// the postings and positions files into memory when opened.
-/// This eliminates all disk I/O to these files during merging.
+/// This is optimized for merging in that it uses a buffered reader
+/// for the postings and positions files.
+/// This eliminates most disk I/O to these files during merging, without
+/// reading the entire file into memory at once.
 ///
 /// NB:  This is a copy/paste from [`InvertedIndexReader`] and trimmed
 /// down to only include the methods required by the merge process.
 pub(crate) struct MergeOptimizedInvertedIndexReader {
     termdict: TermDictionary,
-    postings_bytes: OwnedBytes,
-    positions_bytes: OwnedBytes,
+    postings_reader: BufferedFileSlice,
+    positions_reader: BufferedFileSlice,
     record_option: IndexRecordOption,
 }
 
@@ -34,8 +33,8 @@ impl MergeOptimizedInvertedIndexReader {
         let (_, postings_body) = postings_file_slice.split(8);
         Ok(MergeOptimizedInvertedIndexReader {
             termdict,
-            postings_bytes: postings_body.read_bytes()?,
-            positions_bytes: positions_file_slice.read_bytes()?,
+            postings_reader: BufferedFileSlice::new_with_default_buffer_size(postings_body),
+            positions_reader: BufferedFileSlice::new_with_default_buffer_size(positions_file_slice),
             record_option,
         })
     }
@@ -45,8 +44,8 @@ impl MergeOptimizedInvertedIndexReader {
     pub fn empty(record_option: IndexRecordOption) -> MergeOptimizedInvertedIndexReader {
         MergeOptimizedInvertedIndexReader {
             termdict: TermDictionary::empty(),
-            postings_bytes: FileSlice::empty().read_bytes().unwrap(),
-            positions_bytes: FileSlice::empty().read_bytes().unwrap(),
+            postings_reader: BufferedFileSlice::empty(),
+            positions_reader: BufferedFileSlice::empty(),
             record_option,
         }
     }
@@ -65,7 +64,9 @@ impl MergeOptimizedInvertedIndexReader {
         term_info: &TermInfo,
         requested_option: IndexRecordOption,
     ) -> io::Result<BlockSegmentPostings> {
-        let postings_data = self.postings_bytes.slice(term_info.postings_range.clone());
+        let postings_data = self.postings_reader.get_bytes(
+            term_info.postings_range.start as u64..term_info.postings_range.end as u64,
+        )?;
         BlockSegmentPostings::open(
             term_info.doc_freq,
             postings_data,
@@ -88,9 +89,9 @@ impl MergeOptimizedInvertedIndexReader {
         let block_postings = self.read_block_postings_from_terminfo(term_info, option)?;
         let position_reader = {
             if option.has_positions() {
-                let positions_data = self
-                    .positions_bytes
-                    .slice(term_info.positions_range.clone());
+                let positions_data = self.positions_reader.get_bytes(
+                    term_info.positions_range.start as u64..term_info.positions_range.end as u64,
+                )?;
                 let position_reader = PositionReader::open(positions_data)?;
                 Some(position_reader)
             } else {
diff --git a/src/indexer/mod.rs b/src/indexer/mod.rs
@@ -18,13 +18,13 @@ mod log_merge_policy;
 mod merge_index_test;
 mod merge_operation;
 pub(crate) mod merge_policy;
-pub(crate) mod merger;
+pub mod merger;
 pub(crate) mod operation;
 pub(crate) mod prepared_commit;
 mod segment_entry;
 mod segment_manager;
 mod segment_register;
-pub(crate) mod segment_serializer;
+pub mod segment_serializer;
 pub(crate) mod segment_updater;
 pub(crate) mod segment_writer;
 pub(crate) mod single_segment_index_writer;
diff --git a/src/positions/mod.rs b/src/positions/mod.rs
@@ -28,6 +28,7 @@
 //! * *VIntPosDeltas* := *VIntPosDelta*^(*P* % 128).
 //!
 //! The skip widths encoded separately makes it easy and fast to rapidly skip over n positions.
+
 mod reader;
 mod serializer;
 
diff --git a/src/positions/serializer.rs b/src/positions/serializer.rs
@@ -45,13 +45,109 @@ impl<W: io::Write> PositionSerializer<W> {
 
     /// Writes all of the given positions delta.
     pub fn write_positions_delta(&mut self, mut positions_delta: &[u32]) {
-        while !positions_delta.is_empty() {
-            let remaining_block_len = self.remaining_block_len();
-            let num_to_write = remaining_block_len.min(positions_delta.len());
-            self.block.extend(&positions_delta[..num_to_write]);
-            positions_delta = &positions_delta[num_to_write..];
-            if self.remaining_block_len() == 0 {
-                self.flush_block();
+        match positions_delta.len() {
+            0 => {}
+            1 => {
+                if self.remaining_block_len() == 0 {
+                    self.flush_block();
+                }
+                self.block.push(positions_delta[0]);
+            }
+            2 => {
+                let rem = self.remaining_block_len();
+                if rem < 2 {
+                    if rem == 1 {
+                        self.block.push(positions_delta[0]);
+                        self.flush_block();
+                        self.block.push(positions_delta[1]);
+                    } else {
+                        self.flush_block();
+                        self.block.push(positions_delta[0]);
+                        self.block.push(positions_delta[1]);
+                    }
+                } else {
+                    self.block.push(positions_delta[0]);
+                    self.block.push(positions_delta[1]);
+                }
+            }
+            3 => {
+                let rem = self.remaining_block_len();
+                match rem {
+                    3.. => {
+                        self.block.push(positions_delta[0]);
+                        self.block.push(positions_delta[1]);
+                        self.block.push(positions_delta[2]);
+                    }
+                    2 => {
+                        self.block.push(positions_delta[0]);
+                        self.block.push(positions_delta[1]);
+                        self.flush_block();
+                        self.block.push(positions_delta[2]);
+                    }
+                    1 => {
+                        self.block.push(positions_delta[0]);
+                        self.flush_block();
+                        self.block.push(positions_delta[1]);
+                        self.block.push(positions_delta[2]);
+                    }
+                    0 => {
+                        self.flush_block();
+                        self.block.push(positions_delta[0]);
+                        self.block.push(positions_delta[1]);
+                        self.block.push(positions_delta[2]);
+                    }
+                }
+            }
+            4 => {
+                let rem = self.remaining_block_len();
+                match rem {
+                    4.. => {
+                        self.block.push(positions_delta[0]);
+                        self.block.push(positions_delta[1]);
+                        self.block.push(positions_delta[2]);
+                        self.block.push(positions_delta[3]);
+                    }
+                    3 => {
+                        self.block.push(positions_delta[0]);
+                        self.block.push(positions_delta[1]);
+                        self.block.push(positions_delta[2]);
+                        self.flush_block();
+                        self.block.push(positions_delta[3]);
+                    }
+                    2 => {
+                        self.block.push(positions_delta[0]);
+                        self.block.push(positions_delta[1]);
+                        self.flush_block();
+                        self.block.push(positions_delta[2]);
+                        self.block.push(positions_delta[3]);
+                    }
+                    1 => {
+                        self.block.push(positions_delta[0]);
+                        self.flush_block();
+                        self.block.push(positions_delta[1]);
+                        self.block.push(positions_delta[2]);
+                        self.block.push(positions_delta[3]);
+                    }
+                    0 => {
+                        self.flush_block();
+                        self.block.push(positions_delta[0]);
+                        self.block.push(positions_delta[1]);
+                        self.block.push(positions_delta[2]);
+                        self.block.push(positions_delta[3]);
+                    }
+                }
+            }
+            _ => {
+                while !positions_delta.is_empty() {
+                    let remaining_block_len = self.remaining_block_len();
+                    let num_to_write = remaining_block_len.min(positions_delta.len());
+                    self.block
+                        .extend_from_slice(&positions_delta[..num_to_write]);
+                    positions_delta = &positions_delta[num_to_write..];
+                    if self.remaining_block_len() == 0 {
+                        self.flush_block();
+                    }
+                }
             }
         }
     }
diff --git a/src/postings/block_segment_postings.rs b/src/postings/block_segment_postings.rs
@@ -10,7 +10,7 @@ use crate::query::Bm25Weight;
 use crate::schema::IndexRecordOption;
 use crate::{DocId, Score, TERMINATED};
 
-fn max_score<I: Iterator<Item = Score>>(mut it: I) -> Option<Score> {
+pub(crate) fn max_score<I: Iterator<Item = Score>>(mut it: I) -> Option<Score> {
     it.next().map(|first| it.fold(first, Score::max))
 }
 
@@ -33,7 +33,7 @@ pub struct BlockSegmentPostings {
     skip_reader: SkipReader,
 }
 
-fn decode_bitpacked_block(
+pub(crate) fn decode_bitpacked_block(
     doc_decoder: &mut BlockDecoder,
     freq_decoder_opt: Option<&mut BlockDecoder>,
     data: &[u8],
@@ -53,7 +53,7 @@ fn decode_bitpacked_block(
     }
 }
 
-fn decode_vint_block(
+pub(crate) fn decode_vint_block(
     doc_decoder: &mut BlockDecoder,
     freq_decoder_opt: Option<&mut BlockDecoder>,
     data: &[u8],
diff --git a/src/postings/mod.rs b/src/postings/mod.rs
@@ -4,7 +4,8 @@ mod block_search;
 
 pub(crate) use self::block_search::branchless_binary_search;
 
-mod block_segment_postings;
+pub(crate) mod block_segment_postings;
+
 pub(crate) mod compression;
 mod indexing_context;
 mod json_postings_writer;
@@ -15,7 +16,7 @@ mod postings_writer;
 mod recorder;
 mod segment_postings;
 mod serializer;
-mod skip;
+pub(crate) mod skip;
 mod term_info;
 
 pub(crate) use loaded_postings::LoadedPostings;
diff --git a/src/postings/skip.rs b/src/postings/skip.rs

Original file line number	Diff line number	Diff line change
`@@ -1,3 +1,3 @@`
`1`	`1`	`#! /bin/bash`
`2`	`2`
`3`		`-cargo +stable nextest run --features mmap,stopwords,lz4-compression,zstd-compression,failpoints --verbose --workspace`
	`3`	`+cargo +stable nextest run --features quickwit,mmap,stopwords,lz4-compression,zstd-compression,failpoints --verbose --workspace`
Original file line number	Diff line number	Diff line change
`@@ -10,7 +10,7 @@ use crate::query::Bm25Weight;`
`10`	`10`	`use crate::schema::IndexRecordOption;`
`11`	`11`	`use crate::{DocId, Score, TERMINATED};`
`12`	`12`
`13`		`-fn max_score<I: Iterator<Item = Score>>(mut it: I) -> Option<Score> {`
	`13`	`+pub(crate) fn max_score<I: Iterator<Item = Score>>(mut it: I) -> Option<Score> {`
`14`	`14`	`it.next().map(\|first\| it.fold(first, Score::max))`
`15`	`15`	`}`
`16`	`16`
`@@ -33,7 +33,7 @@ pub struct BlockSegmentPostings {`
`33`	`33`	`skip_reader: SkipReader,`
`34`	`34`	`}`
`35`	`35`
`36`		`-fn decode_bitpacked_block(`
	`36`	`+pub(crate) fn decode_bitpacked_block(`
`37`	`37`	`doc_decoder: &mut BlockDecoder,`
`38`	`38`	`freq_decoder_opt: Option<&mut BlockDecoder>,`
`39`	`39`	`data: &[u8],`
`@@ -53,7 +53,7 @@ fn decode_bitpacked_block(`
`53`	`53`	`}`
`54`	`54`	`}`
`55`	`55`
`56`		`-fn decode_vint_block(`
	`56`	`+pub(crate) fn decode_vint_block(`
`57`	`57`	`doc_decoder: &mut BlockDecoder,`
`58`	`58`	`freq_decoder_opt: Option<&mut BlockDecoder>,`
`59`	`59`	`data: &[u8],`