Sd merge hard ceiling

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 <N> [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<WriterDoc> {
         .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<WriterDoc> {
+    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::<Vec<_>>()
+                .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 <N> [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<String> = 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);
 

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<u32>` 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<Vec<(usize, Vec<u32>)>> {
+    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<u32> = 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<Vec<(usize, Vec<u32>)>> {
+    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<String> = 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<u32>)> = Vec::new();
+        for_each_posting(&frq, &prx, ti, |doc, pos| got.push((doc, pos.to_vec()))).unwrap();
+        assert_eq!(got, expected);
+    }
 }

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,83 @@ 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<u32>)> =
+            s.positions_all(&field, &term).into_iter().collect();
+        expected.sort_by_key(|(d, _)| *d);
+
+        let mut got: Vec<(usize, Vec<u32>)> = 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);
+    }
+
+    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<usize> = 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<usize> = 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<usize> = 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