From ceeb993b4d8e5e56d11d973bae46cf24f5e83b0b Mon Sep 17 00:00:00 2001 From: Federico Martinez Date: Sat, 11 Jul 2026 15:05:44 -0300 Subject: [PATCH 1/6] zsl: add lazy per-posting decoder (for_each_posting / for_each_live_posting) --- sdsearch-core/src/zsl/postings.rs | 61 ++++++++++++++++++++++++------- sdsearch-core/src/zsl/segment.rs | 38 ++++++++++++++++++- 2 files changed, 85 insertions(+), 14 deletions(-) diff --git a/sdsearch-core/src/zsl/postings.rs b/sdsearch-core/src/zsl/postings.rs index 91572cc..8c6baf3 100644 --- a/sdsearch-core/src/zsl/postings.rs +++ b/sdsearch-core/src/zsl/postings.rs @@ -64,20 +64,19 @@ pub fn read_positions( Ok(Vec::new()) } -/// Decodes ALL positions of a term in a single pass: doc -> positions. -/// Avoids re-walking `.frq`/`.prx` from the pointer for each doc (which made phrase O(C·docFreq)). -pub fn read_all_positions( +/// Decodes a term's postings one doc at a time, invoking `f(doc, &positions)` in ascending +/// doc order. A scratch `Vec` is reused across docs so peak extra RAM is one doc's +/// positions, not the whole term. +pub fn for_each_posting( frq: &[u8], prx: &[u8], info: &TermInfo, -) -> std::io::Result)>> { + mut f: impl FnMut(usize, &[u32]), +) -> std::io::Result<()> { let mut fpos = info.freq_pointer as usize; let mut ppos = info.prox_pointer as usize; let mut prev = 0usize; - let mut out = Vec::with_capacity(checked_capacity( - info.doc_freq as usize, - frq.len().saturating_sub(fpos), - )); + let mut positions: Vec = Vec::new(); for _ in 0..info.doc_freq { let v = read_vint(frq, &mut fpos)?; let doc_delta = (v >> 1) as usize; @@ -87,18 +86,30 @@ pub fn read_all_positions( read_vint(frq, &mut fpos)? as u32 }; let d = prev + doc_delta; - let mut positions = Vec::with_capacity(checked_capacity( - freq as usize, - prx.len().saturating_sub(ppos), - )); + positions.clear(); let mut prev_pos = 0u32; for _ in 0..freq { prev_pos += read_vint(prx, &mut ppos)? as u32; positions.push(prev_pos); } - out.push((d, positions)); + f(d, &positions); prev = d; } + Ok(()) +} + +/// Decodes ALL positions of a term in a single pass: doc -> positions. +/// Avoids re-walking `.frq`/`.prx` from the pointer for each doc (which made phrase O(C·docFreq)). +pub fn read_all_positions( + frq: &[u8], + prx: &[u8], + info: &TermInfo, +) -> std::io::Result)>> { + let mut out = Vec::with_capacity(checked_capacity( + info.doc_freq as usize, + frq.len().saturating_sub(info.freq_pointer as usize), + )); + for_each_posting(frq, prx, info, |d, pos| out.push((d, pos.to_vec())))?; Ok(out) } @@ -152,4 +163,28 @@ mod tests { }; assert!(read_freqs(&[], &info).is_err()); } + + #[test] + fn for_each_posting_matches_read_all_positions() { + let cf = cfs(); + let sub = |ext: &str| { + cf.sub(&cf.names().into_iter().find(|n| n.ends_with(ext)).unwrap()) + .unwrap() + .to_vec() + }; + let names: Vec = read_field_infos(&sub(".fnm")) + .unwrap() + .into_iter() + .map(|f| f.name) + .collect(); + let dict = TermDict::read(&sub(".tis"), &names).unwrap(); + let ti = dict.info("title", "new").unwrap(); + let frq = sub(".frq"); + let prx = sub(".prx"); + + let expected = read_all_positions(&frq, &prx, ti).unwrap(); + let mut got: Vec<(usize, Vec)> = Vec::new(); + for_each_posting(&frq, &prx, ti, |doc, pos| got.push((doc, pos.to_vec()))).unwrap(); + assert_eq!(got, expected); + } } diff --git a/sdsearch-core/src/zsl/segment.rs b/sdsearch-core/src/zsl/segment.rs index 96a1fa9..1ba19d6 100644 --- a/sdsearch-core/src/zsl/segment.rs +++ b/sdsearch-core/src/zsl/segment.rs @@ -4,7 +4,7 @@ use crate::zsl::cfs::CompoundFile; use crate::zsl::deletes::DeletedDocs; use crate::zsl::fields::{read_field_infos, FieldInfo}; use crate::zsl::norms::{approx_field_len, read_norms}; -use crate::zsl::postings::{read_all_positions, read_freqs, read_positions}; +use crate::zsl::postings::{for_each_posting, read_all_positions, read_freqs, read_positions}; use crate::zsl::stored::{read_stored_fields, read_stored_raw, StoredRaw}; use crate::zsl::terms::{TermCursor, TermDict}; use std::collections::HashMap; @@ -58,6 +58,27 @@ impl ZslSegment { self.dict.iter_terms() } + /// Streams a term's LIVE postings (deletes filtered) one local doc at a time, ascending. + /// No-op if the term or `.prx` is absent (mirrors `positions_all`'s degradation). + pub fn for_each_live_posting(&self, field: &str, term: &str, mut f: impl FnMut(usize, &[u32])) { + if self.prx_name.is_empty() { + return; + } + let Some(ti) = self.dict.info(field, term) else { + return; + }; + let (Some(frq), Some(prx)) = (self.cfs.sub(&self.frq_name), self.cfs.sub(&self.prx_name)) + else { + return; + }; + // degrade: a corrupt tail stops iteration rather than panicking across FFI + let _ = for_each_posting(frq, prx, ti, |d, pos| { + if !self.deletes.is_deleted(d) { + f(d, pos); + } + }); + } + /// lazy cursor over every `(field, term)` pair in ZSL canonical order /// (field names ascending, terms ascending within each field), without /// materializing a `Vec` of all terms like `all_terms` does. Used by the @@ -371,6 +392,21 @@ mod tests { assert_eq!(got, expected); } + #[test] + fn for_each_live_posting_matches_positions_all() { + let s = seg_kb(); + let (field, term) = s.all_terms().into_iter().next().unwrap(); + let mut expected: Vec<(usize, Vec)> = + s.positions_all(&field, &term).into_iter().collect(); + expected.sort_by_key(|(d, _)| *d); + + let mut got: Vec<(usize, Vec)> = Vec::new(); + s.for_each_live_posting(&field, &term, |d, pos| got.push((d, pos.to_vec()))); + got.sort_by_key(|(d, _)| *d); + + assert_eq!(got, expected); + } + #[test] fn merge_accessors_expose_fields_deletes_norms_terms_stored() { let s = seg(); // incidents fixture, 4 docs, no deletes From c333d557c33cd5e6d9d9859a96925a9b27226b7c Mon Sep 17 00:00:00 2001 From: Federico Martinez Date: Sat, 11 Jul 2026 15:12:50 -0300 Subject: [PATCH 2/6] test(zsl): exercise the delete-filter branch of for_each_live_posting MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit for_each_live_posting_matches_positions_all ran on the KB fixture segment "_2", which has no deletions, so a dropped or inverted delete filter would not have been caught. Add a test on the multiseg fixture's segment "_1" (real .del, local doc 1 genuinely deleted): it fetches title:backup's postings via the delete-agnostic read_freqs, confirms the deleted doc is present there but absent from for_each_live_posting's output, and asserts the two counts actually differ so the assertion can't be vacuous. Verified by temporarily inverting the filter in for_each_live_posting (dropping the `!`) — both this test and the existing for_each_live_posting_matches_positions_all failed as expected — then reverting. --- sdsearch-core/src/zsl/segment.rs | 62 ++++++++++++++++++++++++++++++++ 1 file changed, 62 insertions(+) diff --git a/sdsearch-core/src/zsl/segment.rs b/sdsearch-core/src/zsl/segment.rs index 1ba19d6..1741cce 100644 --- a/sdsearch-core/src/zsl/segment.rs +++ b/sdsearch-core/src/zsl/segment.rs @@ -407,6 +407,68 @@ mod tests { assert_eq!(got, expected); } + fn seg_multiseg_1() -> ZslSegment { + let dir = PathBuf::from(concat!( + env!("CARGO_MANIFEST_DIR"), + "/tests/fixtures/zsl_index_multiseg" + )); + // segment "_1" is the fixture's 2nd commit batch ("gamma vpn tutorial" local doc 0, + // "delta backup notes" local doc 1); its "_1_1.del" (del_gen 1) marks local doc 1 + // deleted. Unlike the KB fixture (no deletes at all), this segment lets a test + // actually exercise the delete-filter branch of `for_each_live_posting`. + ZslSegment::open_named(&dir, "_1", 1).unwrap() + } + + #[test] + fn for_each_live_posting_excludes_genuinely_deleted_doc() { + let s = seg_multiseg_1(); + + // sanity: the fixture really has a deletion here, or the rest of this test is vacuous. + assert!( + s.is_deleted(1), + "fixture local doc 1 (\"delta backup notes\") must be deleted" + ); + + // "backup" only appears in the deleted doc. Fetch its postings straight from the + // .frq via read_freqs (delete-AGNOSTIC — no `deletes.is_deleted` filtering), so this + // is an independent source from `positions_all` (which is itself delete-filtered) + // and can actually prove `for_each_live_posting` removes something. + let field = "title"; + let term = "backup"; + let ti = s + .dict + .info(field, term) + .expect("fixture must index title:backup"); + let raw = read_freqs(s.cfs.sub(&s.frq_name).unwrap(), ti).unwrap(); + let raw_docs: Vec = raw.iter().map(|(d, _)| *d).collect(); + assert!( + raw_docs.contains(&1), + "delete-agnostic postings must include the deleted doc: {raw_docs:?}" + ); + + let mut live: Vec = Vec::new(); + s.for_each_live_posting(field, term, |d, _pos| live.push(d)); + + // the chosen term must have a genuine gap between delete-agnostic and live counts, + // otherwise a dropped/inverted filter would not be caught by this test. + assert!( + raw_docs.len() > live.len(), + "term must have delete-agnostic count > live count to be non-vacuous: raw={raw_docs:?} live={live:?}" + ); + assert!( + !live.contains(&1), + "for_each_live_posting must exclude the deleted doc: live={live:?}" + ); + assert!(live.iter().all(|d| !s.is_deleted(*d))); + + // must also agree with positions_all's doc set (both are delete-filtered). + let mut expected: Vec = s.positions_all(field, term).into_keys().collect(); + expected.sort(); + let mut got = live.clone(); + got.sort(); + assert_eq!(got, expected); + } + #[test] fn merge_accessors_expose_fields_deletes_norms_terms_stored() { let s = seg(); // incidents fixture, 4 docs, no deletes From d581acae8330766abb69d354c5b513d3f90fc860 Mon Sep 17 00:00:00 2001 From: Federico Martinez Date: Sat, 11 Jul 2026 15:19:25 -0300 Subject: [PATCH 3/6] zsl: add stream-within-term writer API (begin_term/add_posting/end_term) Lets the merge write a term's postings one doc at a time instead of collecting them into a Vec first. Kept add_term as an independent oracle (does not call the new methods) so the byte-identity test comparing the two paths stays meaningful; only the low-level .tis/.tii tail encoding is shared via a new dump_tis_entry helper. --- sdsearch-core/src/zsl/writer/terms.rs | 238 ++++++++++++++++++++++++-- 1 file changed, 222 insertions(+), 16 deletions(-) diff --git a/sdsearch-core/src/zsl/writer/terms.rs b/sdsearch-core/src/zsl/writer/terms.rs index 8a9390b..18e4129 100644 --- a/sdsearch-core/src/zsl/writer/terms.rs +++ b/sdsearch-core/src/zsl/writer/terms.rs @@ -52,6 +52,16 @@ where idx_prev: Option, last_index_position: u64, term_count: u64, + + // per-term streaming state for `begin_term`/`add_posting`/`end_term`. + cur_field: usize, + cur_text: String, + cur_freq_ptr: u64, + cur_prox_ptr: u64, + cur_doc_freq: u32, + cur_prev_doc: usize, + frq_scratch: Vec, + prx_scratch: Vec, } impl TermDictStreamWriter @@ -92,11 +102,27 @@ where idx_prev: None, last_index_position: 24, term_count: 0, + + cur_field: 0, + cur_text: String::new(), + cur_freq_ptr: 0, + cur_prox_ptr: 0, + cur_doc_freq: 0, + cur_prev_doc: 0, + frq_scratch: Vec::new(), + prx_scratch: Vec::new(), }) } /// Appends one term's dictionary entry (`.tis`, and `.tii` every `INDEX_INTERVAL`th /// term) plus its postings (`.frq`/`.prx`). Terms must arrive in ZSL canonical order. + /// + /// Kept independent of `begin_term`/`add_posting`/`end_term`: it computes its own + /// `freq_ptr`/`prox_ptr`/`doc_freq` and writes postings in one shot via + /// `write_term_postings`, only sharing the `.tis`/`.tii` tail encoding (`dump_tis_entry`) + /// with the streaming API. This keeps it a valid independent oracle for the + /// byte-identity test in `stream_within_term_matches_add_term_byte_for_byte` — see + /// the module docs and the plan's Global Constraints (commit d93b16d). pub fn add_term(&mut self, term: &TermPostings) -> io::Result<()> { // `write_term_postings` only depends on `term.docs` (it resets its own doc/position // deltas per call), so writing into fresh local buffers reproduces the exact same @@ -110,25 +136,106 @@ where let doc_freq = term.doc_freq(); + self.dump_tis_entry(term.field_num, &term.text, doc_freq, freq_ptr, prox_ptr)?; + + self.frq.write_all(&frq_buf)?; + self.frq_len += frq_buf.len() as u64; + self.prx.write_all(&prx_buf)?; + self.prx_len += prx_buf.len() as u64; + + Ok(()) + } + + /// Starts a new term for the stream-within-term API: records the `.frq`/`.prx` + /// pointers this term will start at (the running lengths *before* any posting of + /// this term is written) and resets the per-term posting state. Writes nothing. + pub fn begin_term(&mut self, field_num: usize, text: &str) -> io::Result<()> { + self.cur_field = field_num; + self.cur_text = text.to_string(); + self.cur_freq_ptr = self.frq_len; + self.cur_prox_ptr = self.prx_len; + self.cur_doc_freq = 0; + self.cur_prev_doc = 0; + Ok(()) + } + + /// Appends one posting (`new_doc_id`, ascending, plus its ascending positions) to + /// the current term started by `begin_term`. Encodes exactly like + /// `write_term_postings`'s per-doc loop: `docDelta*2 (+1 if freq==1)`, else + /// `docDelta*2` followed by `VInt(freq)`; positions as per-doc deltas in `.prx`. + pub fn add_posting(&mut self, new_doc_id: usize, positions: &[u32]) -> io::Result<()> { + let doc_delta = (new_doc_id - self.cur_prev_doc) * 2; + if positions.len() > 1 { + write_vint(&mut self.frq_scratch, doc_delta as u64); + write_vint(&mut self.frq_scratch, positions.len() as u64); + } else { + write_vint(&mut self.frq_scratch, (doc_delta + 1) as u64); + } + + let mut prev_pos = 0u32; + for &pos in positions { + write_vint(&mut self.prx_scratch, (pos - prev_pos) as u64); + prev_pos = pos; + } + + self.frq.write_all(&self.frq_scratch)?; + self.frq_len += self.frq_scratch.len() as u64; + self.prx.write_all(&self.prx_scratch)?; + self.prx_len += self.prx_scratch.len() as u64; + self.frq_scratch.clear(); + self.prx_scratch.clear(); + + self.cur_prev_doc = new_doc_id; + self.cur_doc_freq += 1; + Ok(()) + } + + /// Closes the term started by `begin_term`. If no posting was added (`doc_freq == + /// 0`), the term is dropped: no `.tis` entry, no `.tii` sample, no `.frq`/`.prx` + /// bytes (matching the merge's "skip empty terms" behavior). Otherwise writes the + /// `.tis` entry and the `.tii` sample exactly as `add_term`'s tail does, via the + /// shared `dump_tis_entry`. + pub fn end_term(&mut self) -> io::Result<()> { + if self.cur_doc_freq == 0 { + return Ok(()); + } + let field_num = self.cur_field; + let text = std::mem::take(&mut self.cur_text); + let doc_freq = self.cur_doc_freq; + let freq_ptr = self.cur_freq_ptr; + let prox_ptr = self.cur_prox_ptr; + self.dump_tis_entry(field_num, &text, doc_freq, freq_ptr, prox_ptr) + } + + /// Shared `.tis`/`.tii` tail: writes the term-dict entry (prefix-shared vs the + /// previous term of the same field), advances the prev-term/term-count state, and + /// samples into `.tii` every `INDEX_INTERVAL`th term. Used by both `add_term` and + /// `end_term` — this is shared low-level ENCODING, not one path built on top of the + /// other: each caller independently computes its own `freq_ptr`/`prox_ptr`/`doc_freq` + /// via a different code path (`write_term_postings` vs `add_posting`'s running state). + fn dump_tis_entry( + &mut self, + field_num: usize, + text: &str, + doc_freq: u32, + freq_ptr: u64, + prox_ptr: u64, + ) -> io::Result<()> { let mut tis_entry = Vec::new(); dump_entry( &mut tis_entry, self.prev .as_ref() .map(|(t, f, fp, pp)| (t.as_str(), *f, *fp, *pp)), - term, + field_num, + text, doc_freq, freq_ptr, prox_ptr, ); self.tis.write_all(&tis_entry)?; self.tis_len += tis_entry.len() as u64; - self.prev = Some((term.text.clone(), term.field_num, freq_ptr, prox_ptr)); - - self.frq.write_all(&frq_buf)?; - self.frq_len += frq_buf.len() as u64; - self.prx.write_all(&prx_buf)?; - self.prx_len += prx_buf.len() as u64; + self.prev = Some((text.to_string(), field_num, freq_ptr, prox_ptr)); self.term_count += 1; // sample every indexInterval terms @@ -139,7 +246,8 @@ where self.idx_prev .as_ref() .map(|(t, f, fp, pp)| (t.as_str(), *f, *fp, *pp)), - term, + field_num, + text, doc_freq, freq_ptr, prox_ptr, @@ -148,15 +256,14 @@ where write_vint(&mut tii_entry, index_position - self.last_index_position); self.last_index_position = index_position; self.tii.write_all(&tii_entry)?; - self.idx_prev = Some((term.text.clone(), term.field_num, freq_ptr, prox_ptr)); + self.idx_prev = Some((text.to_string(), field_num, freq_ptr, prox_ptr)); } Ok(()) } - /// Back-patches the 8-byte term counts at offset 4 in `.tis`/`.tii` and flushes all - /// four sinks. - pub fn finish(mut self) -> io::Result<()> { + /// Back-patches the 8-byte term counts at offset 4 in `.tis`/`.tii`. + fn backpatch_term_counts(&mut self) -> io::Result<()> { let term_count = self.term_count; self.tis.seek(SeekFrom::Start(4))?; self.tis.write_all(&term_count.to_be_bytes())?; @@ -164,7 +271,13 @@ where let tii_count = (term_count - term_count % INDEX_INTERVAL) / INDEX_INTERVAL + 1; self.tii.seek(SeekFrom::Start(4))?; self.tii.write_all(&tii_count.to_be_bytes())?; + Ok(()) + } + /// Back-patches the 8-byte term counts at offset 4 in `.tis`/`.tii` and flushes all + /// four sinks. + pub fn finish(mut self) -> io::Result<()> { + self.backpatch_term_counts()?; self.tis.flush()?; self.tii.flush()?; self.frq.flush()?; @@ -212,18 +325,19 @@ fn write_header(out: &mut Vec) { fn dump_entry( out: &mut Vec, prev: Option<(&str, usize, u64, u64)>, - term: &TermPostings, + field_num: usize, + text: &str, doc_freq: u32, freq_ptr: u64, prox_ptr: u64, ) { let (prefix_chars, prefix_bytes) = match prev { - Some((ptext, pfield, ..)) if pfield == term.field_num => common_prefix(ptext, &term.text), + Some((ptext, pfield, ..)) if pfield == field_num => common_prefix(ptext, text), _ => (0, 0), }; write_vint(out, prefix_chars as u64); - write_modified_utf8(out, &term.text[prefix_bytes..]); - write_vint(out, term.field_num as u64); + write_modified_utf8(out, &text[prefix_bytes..]); + write_vint(out, field_num as u64); write_vint(out, doc_freq as u64); match prev { Some((_, _, pf, pp)) => { @@ -254,6 +368,34 @@ fn common_prefix(a: &str, b: &str) -> (usize, usize) { (chars, bytes) } +/// Test-only: a `TermDictStreamWriter` over owned in-memory buffers, and the plain +/// `(tis, tii, frq, prx)` byte tuple it produces. +#[cfg(test)] +type TestStreamWriter = + TermDictStreamWriter>, io::Cursor>, Vec, Vec>; +#[cfg(test)] +type TestDictBuffers = (Vec, Vec, Vec, Vec); + +#[cfg(test)] +impl TestStreamWriter { + /// Test-only: like `finish`, but returns the four finished buffers instead of + /// discarding them, so a test can compare bytes across two independently-driven + /// writers. + fn finish_into_buffers(mut self) -> io::Result { + self.backpatch_term_counts()?; + self.tis.flush()?; + self.tii.flush()?; + self.frq.flush()?; + self.prx.flush()?; + Ok(( + self.tis.into_inner(), + self.tii.into_inner(), + self.frq, + self.prx, + )) + } +} + #[cfg(test)] mod tests { use super::*; @@ -516,6 +658,70 @@ mod tests { (chars, bytes) } + /// Test-only writer that owns its four sinks as in-memory buffers, so a test can + /// build one, drive it, and pull the finished bytes back out without juggling + /// external `Cursor`/`Vec` borrows. + fn new_stream_writer_over_cursors() -> TestStreamWriter { + TermDictStreamWriter::new( + io::Cursor::new(Vec::new()), + io::Cursor::new(Vec::new()), + Vec::new(), + Vec::new(), + ) + .unwrap() + } + + /// Feeds the SAME terms through `add_term` (the existing, unmodified oracle — it + /// still computes postings in one shot via `write_term_postings`) and through + /// `begin_term`/`add_posting`/`end_term` (the new streaming API, one doc at a time) + /// into two INDEPENDENT writers/sinks, then asserts all four output buffers match + /// byte-for-byte. `add_term` is not rerouted through the new methods (see module + /// docs / Global Constraints), so this is a genuine two-implementation comparison, + /// not the subject compared against itself. + #[test] + fn stream_within_term_matches_add_term_byte_for_byte() { + let terms = multi_field_sample_terms(); + assert!( + terms.len() > 128, + "must exceed indexInterval to test .tii sampling + patch" + ); + + let mut a = new_stream_writer_over_cursors(); + for t in &terms { + a.add_term(t).unwrap(); + } + let (a_tis, a_tii, a_frq, a_prx) = a.finish_into_buffers().unwrap(); + + let mut b = new_stream_writer_over_cursors(); + for t in &terms { + b.begin_term(t.field_num, &t.text).unwrap(); + for (doc, pos) in &t.docs { + b.add_posting(*doc, pos).unwrap(); + } + b.end_term().unwrap(); + } + let (b_tis, b_tii, b_frq, b_prx) = b.finish_into_buffers().unwrap(); + + assert_eq!(a_tis, b_tis, "tis mismatch"); + assert_eq!(a_tii, b_tii, "tii mismatch"); + assert_eq!(a_frq, b_frq, "frq mismatch"); + assert_eq!(a_prx, b_prx, "prx mismatch"); + } + + #[test] + fn zero_posting_term_writes_nothing() { + let mut w = new_stream_writer_over_cursors(); + w.begin_term(0, "ghost").unwrap(); + w.end_term().unwrap(); // no add_posting: doc_freq stays 0, term is dropped + + let (tis, tii, frq, prx) = w.finish_into_buffers().unwrap(); + // must be identical to a writer that never saw a term at all (only headers). + let empty = new_stream_writer_over_cursors() + .finish_into_buffers() + .unwrap(); + assert_eq!((tis, tii, frq, prx), empty); + } + #[test] fn stream_writer_matches_batch_writer_byte_for_byte() { let terms = multi_field_sample_terms(); From 3a71de4f5a0ab629eb009c98b40f473f26141c9b Mon Sep 17 00:00:00 2001 From: Federico Martinez Date: Sat, 11 Jul 2026 15:26:51 -0300 Subject: [PATCH 4/6] zsl: stream postings per-doc in the merge (bounds the near-stopword term) Rewires phase 3 of merge_segments_streaming to use the Task 2 begin_term/add_posting/end_term API on top of the Task 1 for_each_live_posting lazy decoder, instead of gathering the whole term's positions into a TermPostings before writing it. Contributing segments are walked in ascending index, and for_each_live_posting yields ascending local docs within each, so postings arrive in ascending new-id order with no per-term gather/sort needed. Adds the streaming_matches_batch_with_a_term_in_every_doc differential test (a term present in every doc across segments) as the hot-term regression guard, and a `hot` scenario in optimize_bench.rs to empirically measure the per-term heap spike. --- sdsearch-core/examples/optimize_bench.rs | 98 ++++++++++++++++++++++++ sdsearch-core/src/zsl/writer/merge.rs | 78 ++++++++++++++----- 2 files changed, 157 insertions(+), 19 deletions(-) diff --git a/sdsearch-core/examples/optimize_bench.rs b/sdsearch-core/examples/optimize_bench.rs index 8924f3c..1c1c378 100644 --- a/sdsearch-core/examples/optimize_bench.rs +++ b/sdsearch-core/examples/optimize_bench.rs @@ -16,6 +16,10 @@ //! N number of docs to add on top of the base (default 2000) //! cap max_buffered_docs while building (default 1000 → ceil(N/cap) flushed segments) //! +//! cargo run -p sdsearch-core --release --example optimize_bench -- hot [cap] +//! same shape, but every doc shares one "stopword" token (doc_freq == N), to check that +//! `heap_peak_kb` no longer spikes proportionally to N for a term present in every doc. +//! //! Prints ONE JSON line: {"n":..,"cap":..,"segments_before":..,"doc_count":.., //! "optimize_ms":..,"heap_peak_kb":..,"vmhwm_kb":..} @@ -155,6 +159,55 @@ fn gen_docs(n: usize) -> Vec { .collect() } +/// N deterministic docs whose `body` starts with the SAME token ("stopword") repeated once, +/// plus the same per-doc unique tokens as `gen_docs` — so "stopword" gets `doc_freq == n` (a +/// synthetic near-stopword term), letting `--hot-term` isolate the per-term posting-list spike +/// the streaming merge (Task 3) is meant to bound. +fn gen_docs_hot(n: usize) -> Vec { + const POOL: &[&str] = &[ + "printer", "network", "vpn", "login", "email", "server", "crash", "slow", "reset", + "password", "access", "error", "update", "install", "config", "backup", "restore", + "timeout", "license", "upgrade", "firewall", "router", "disk", "memory", "cpu", + ]; + let np = POOL.len(); + (0..n) + .map(|i| { + let title = format!( + "ticket {i} stopword {} {} issue{i}", + POOL[i % np], + POOL[(i * 3) % np] + ); + let body: String = (0..40) + .map(|j| POOL[(i * 7 + j * 5) % np]) + .collect::>() + .join(" "); + let body = format!("stopword {body} ref{i}"); + WriterDoc { + fields: vec![ + WriterField { + name: "title".into(), + value: title, + kind: FieldKind::Text, + stored: true, + }, + WriterField { + name: "body".into(), + value: body, + kind: FieldKind::Text, + stored: true, + }, + WriterField { + name: "id".into(), + value: format!("REC-{i}"), + kind: FieldKind::Keyword, + stored: true, + }, + ], + } + }) + .collect() +} + /// copies an arbitrary index dir to `dst` (skips lock files and `.sti`). Overwrites `dst`. fn copy_index(src: &Path, dst: &Path) { if dst.is_dir() { @@ -218,12 +271,57 @@ fn run_existing(args: &[String]) { std::fs::remove_dir_all(scratch).ok(); } +/// `optimize_bench hot [cap]`: same shape as the default run, but every doc's `title` and +/// `body` carry the shared "stopword" token (so it has `doc_freq == N`), to empirically check +/// that the streaming merge's `heap_peak_kb` no longer spikes proportionally to N for a term +/// that is in every doc (the near-stopword case the streaming per-posting merge bounds). +fn run_hot(args: &[String]) { + let n: usize = args.get(2).and_then(|s| s.parse().ok()).unwrap_or(2000); + let cap: usize = args.get(3).and_then(|s| s.parse().ok()).unwrap_or(1000); + + let dir = copy_kb_base(n, cap); + + { + let opts = WriterOpts { + max_buffered_docs: cap, + ..WriterOpts::default() + }; + let mut w = IndexWriter::open(&dir, opts).expect("open (build) failed"); + for d in gen_docs_hot(n) { + w.add_document(d).expect("add_document failed"); + } + w.commit().expect("commit failed"); + } + + let segments_before = sdsearch_core::zsl::segments::read_segment_infos(&dir) + .expect("read segment infos") + .len(); + + reset_peak(); + let t0 = std::time::Instant::now(); + let w = IndexWriter::open(&dir, WriterOpts::default()).expect("open (optimize) failed"); + let report = w.optimize().expect("optimize failed"); + let optimize_ms = t0.elapsed().as_secs_f64() * 1000.0; + let heap_peak_kb = PEAK.load(Ordering::Relaxed) as u64 / 1024; + + println!( + "{{\"mode\":\"hot\",\"n\":{},\"cap\":{},\"segments_before\":{},\"doc_count\":{},\"optimize_ms\":{:.2},\"heap_peak_kb\":{},\"vmhwm_kb\":{}}}", + n, cap, segments_before, report.doc_count, optimize_ms, heap_peak_kb, vmhwm_kb() + ); + + std::fs::remove_dir_all(&dir).ok(); +} + fn main() { let args: Vec = std::env::args().collect(); if args.get(1).map(String::as_str) == Some("existing") { run_existing(&args); return; } + if args.get(1).map(String::as_str) == Some("hot") { + run_hot(&args); + return; + } let n: usize = args.get(1).and_then(|s| s.parse().ok()).unwrap_or(2000); let cap: usize = args.get(2).and_then(|s| s.parse().ok()).unwrap_or(1000); diff --git a/sdsearch-core/src/zsl/writer/merge.rs b/sdsearch-core/src/zsl/writer/merge.rs index 4f5badc..8c5e13d 100644 --- a/sdsearch-core/src/zsl/writer/merge.rs +++ b/sdsearch-core/src/zsl/writer/merge.rs @@ -380,31 +380,34 @@ fn merge_streaming_inner( // Contributing segments in ASCENDING index (+ locals ascending within each) => new // doc-ids come out ascending, matching merge_segments' BTreeMap ordering - // (dense renumbering assigns lower ids to earlier segments). + // (dense renumbering assigns lower ids to earlier segments). `for_each_live_posting` + // yields ascending LOCAL doc within a segment, which maps to ascending NEW id (dense + // renumbering), and segment ranges are globally ordered — so postings arrive in + // ascending new-id order across the whole term without any gather/sort, bounding peak + // memory to one posting's positions rather than the whole term's doc list. contributing.sort_unstable(); let mf = field_index[&field]; - let mut docs: Vec<(usize, Vec)> = Vec::new(); + dict_writer.begin_term(mf, &term)?; for &si in &contributing { - // positions_all ALREADY filters deletes. It may return an unordered HashMap, so sort - // locals ascending before remapping so the new ids come out ascending. - let mut locals: Vec<(usize, Vec)> = - segs[si].positions_all(&field, &term).into_iter().collect(); - locals.sort_by_key(|(local, _)| *local); - for (local, positions) in locals { - if let Some(new_id) = doc_maps[si][local] { - docs.push((new_id, positions)); + let map = &doc_maps[si]; + // `for_each_live_posting`'s callback is `FnMut` and can't propagate `?`, so capture + // any `add_posting` error here and check it after the segment finishes iterating. + let mut posting_err: Option = None; + segs[si].for_each_live_posting(&field, &term, |local, positions| { + if posting_err.is_some() { + return; + } + if let Some(new_id) = map[local] { + if let Err(e) = dict_writer.add_posting(new_id, positions) { + posting_err = Some(e); + } } + }); + if let Some(e) = posting_err { + return Err(e); } } - // terms with no live docs are dropped (== ZSL / merge_segments). - if docs.is_empty() { - continue; - } - dict_writer.add_term(&TermPostings { - field_num: mf, - text: term, - docs, - })?; + dict_writer.end_term()?; // drops the term if 0 live postings (== ZSL / merge_segments) } dict_writer.finish()?; // back-patch term counts + close the temp .frq/.prx @@ -830,6 +833,43 @@ mod tests { std::fs::remove_dir_all(&dir).ok(); } + #[test] + fn streaming_matches_batch_with_a_term_in_every_doc() { + // multi-segment base (cap=2 → 2 flushed segments) where EVERY added doc's `title` + // contains "allofthem" (so it has a posting in every live doc across every segment, + // i.e. doc_freq == total live docs — the shape the old gather would have held fully + // resident), plus a per-doc unique token so postings differ doc to doc. + let dir = temp_kb_full(); + let opts = WriterOpts { + max_buffered_docs: 2, + ..WriterOpts::default() + }; + let mut w = IndexWriter::open(&dir, opts).unwrap(); + for i in 0..6 { + w.add_document(WriterDoc { + fields: vec![WriterField::text("title", &format!("allofthem unique{i}"))], + }) + .unwrap(); + } + w.commit().unwrap(); + + let infos = read_segment_infos(&dir).unwrap(); + assert!( + infos.len() >= 2, + "expected multi-segment base, got {}", + infos.len() + ); + + let before = ZslIndex::open(&dir).unwrap(); + // KB fixture's 20 docs do NOT contain "allofthem", so doc_freq == exactly the 6 added. + assert_eq!(before.doc_freq("title", "allofthem"), 6); + drop(before); + + let refs: Vec<(String, i64)> = infos.iter().map(|s| (s.name.clone(), s.del_gen)).collect(); + assert_streaming_byte_identical(&dir, &refs); + std::fs::remove_dir_all(&dir).ok(); + } + #[test] fn merge_segments_collapses_multiseg_with_delete() { let dir = temp_kb_full(); From 5aebbf2633f38d8bacc43590061eb6b528259ff4 Mon Sep 17 00:00:00 2001 From: Federico Martinez Date: Sat, 11 Jul 2026 15:40:45 -0300 Subject: [PATCH 5/6] zsl: buffer the merge's temp .fdt/.frq/.prx sinks (perf fix) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit merge_segments_streaming's per-doc/per-posting streaming (3a71de4) writes through raw File handles, so add_posting/add_doc issue one unbuffered write_all syscall per posting/doc — for a term with doc_freq N that's N syscalls instead of one, regressing optimize_ms on hot-term merges. Wrap the temp .fdt/.frq/.prx File handles in a fixed-size BufWriter (8 KiB, independent of doc_freq) before handing them to StoredStreamWriter / TermDictStreamWriter. The .tis/.tii sinks stay untouched (in-RAM Cursors, need Seek for the term-count back-patch, written per-term not per-posting). finish() on both writers already calls flush() on every sink generically, so the buffered bytes are guaranteed on disk before write_cfs_streaming reads the temp files back for CFS assembly. --- sdsearch-core/src/zsl/writer/merge.rs | 18 ++++++++++++++---- 1 file changed, 14 insertions(+), 4 deletions(-) diff --git a/sdsearch-core/src/zsl/writer/merge.rs b/sdsearch-core/src/zsl/writer/merge.rs index 8c5e13d..93d499b 100644 --- a/sdsearch-core/src/zsl/writer/merge.rs +++ b/sdsearch-core/src/zsl/writer/merge.rs @@ -19,7 +19,7 @@ use crate::zsl::segment::ZslSegment; use std::cmp::Reverse; use std::collections::{BTreeMap, BinaryHeap, HashMap}; use std::fs::File; -use std::io; +use std::io::{self, BufWriter}; use std::path::Path; /// merge result: the bytes of the merged `.cfs` + the number of live docs. @@ -259,8 +259,13 @@ fn merge_streaming_inner( let mut norm_cols: Vec> = vec![Vec::new(); fields.len()]; let mut next_id = 0usize; + // `.fdt` is append-only and written one doc at a time (`add_doc`): wrap it in a `BufWriter` + // so per-doc writes coalesce into few syscalls instead of one `write_all` per doc. The + // buffer is a small constant (default 8 KiB), independent of segment size; `finish()` below + // flushes it to disk BEFORE the temp file is read back during CFS assembly. let mut fdx_buf: Vec = Vec::new(); - let mut stored_writer = stored::StoredStreamWriter::new(File::create(fdt_tmp)?, &mut fdx_buf); + let mut stored_writer = + stored::StoredStreamWriter::new(BufWriter::new(File::create(fdt_tmp)?), &mut fdx_buf); for seg in &segs { let local_fields = seg.field_infos(); @@ -338,11 +343,16 @@ fn merge_streaming_inner( // .frq/.prx are the big blocks, streamed to temp files. let mut tis_buf = io::Cursor::new(Vec::new()); let mut tii_buf = io::Cursor::new(Vec::new()); + // `.frq`/`.prx` are append-only and written one posting at a time (`add_posting`): same + // `BufWriter` treatment as `.fdt` above, for the same reason (per-term-size N postings would + // otherwise be N unbuffered `write_all` syscalls). `.tis`/`.tii` stay as in-RAM `Cursor`s + // (Write+Seek, needed for the term-count back-patch) — they are written per-term, not + // per-posting, so they were never the bottleneck. let mut dict_writer = terms::TermDictStreamWriter::new( &mut tis_buf, &mut tii_buf, - File::create(frq_tmp)?, - File::create(prx_tmp)?, + BufWriter::new(File::create(frq_tmp)?), + BufWriter::new(File::create(prx_tmp)?), )?; // one cursor per segment; each yields (field, term) ascending. A min-heap (BinaryHeap is a From 4ebedf42896289e67c89a128ddfb6832b7c15db3 Mon Sep 17 00:00:00 2001 From: Federico Martinez Date: Sat, 11 Jul 2026 15:47:35 -0300 Subject: [PATCH 6/6] zsl: harden term dict stream writer against invariant violations add_posting now debug-asserts that doc ids arrive strictly ascending within a term (the sort that used to guarantee this was removed in the merge rewrite, so the underflow in doc_delta is now reachable from a caller bug rather than provably impossible). end_term resets cur_doc_freq to 0 before writing its entry, so a stray second end_term call without an intervening begin_term is a no-op instead of emitting a duplicate .tis/.tii entry. Both are internal-invariant hardening with no effect on correct usage; byte output is unchanged (stream_within_term_matches_add_term_byte_for_byte and all merge streaming_matches_batch_* tests stay green). --- sdsearch-core/src/zsl/writer/terms.rs | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/sdsearch-core/src/zsl/writer/terms.rs b/sdsearch-core/src/zsl/writer/terms.rs index 18e4129..d031576 100644 --- a/sdsearch-core/src/zsl/writer/terms.rs +++ b/sdsearch-core/src/zsl/writer/terms.rs @@ -164,6 +164,11 @@ where /// `write_term_postings`'s per-doc loop: `docDelta*2 (+1 if freq==1)`, else /// `docDelta*2` followed by `VInt(freq)`; positions as per-doc deltas in `.prx`. pub fn add_posting(&mut self, new_doc_id: usize, positions: &[u32]) -> io::Result<()> { + debug_assert!( + self.cur_doc_freq == 0 || new_doc_id > self.cur_prev_doc, + "add_posting: doc ids must be strictly ascending within a term (got {new_doc_id} after {})", + self.cur_prev_doc + ); let doc_delta = (new_doc_id - self.cur_prev_doc) * 2; if positions.len() > 1 { write_vint(&mut self.frq_scratch, doc_delta as u64); @@ -204,6 +209,10 @@ where let doc_freq = self.cur_doc_freq; let freq_ptr = self.cur_freq_ptr; let prox_ptr = self.cur_prox_ptr; + // Reset before writing so a stray second `end_term` (without an intervening + // `begin_term`) sees `cur_doc_freq == 0` and is dropped above, instead of + // re-emitting a duplicate `.tis`/`.tii` entry for the same term. + self.cur_doc_freq = 0; self.dump_tis_entry(field_num, &text, doc_freq, freq_ptr, prox_ptr) } @@ -708,6 +717,16 @@ mod tests { assert_eq!(a_prx, b_prx, "prx mismatch"); } + #[test] + #[should_panic(expected = "doc ids must be strictly ascending")] + #[cfg(debug_assertions)] + fn add_posting_panics_on_non_ascending_doc_id() { + let mut w = new_stream_writer_over_cursors(); + w.begin_term(0, "term").unwrap(); + w.add_posting(5, &[0]).unwrap(); + w.add_posting(3, &[0]).unwrap(); // not ascending: must panic + } + #[test] fn zero_posting_term_writes_nothing() { let mut w = new_stream_writer_over_cursors();