diff --git a/.gitignore b/.gitignore index 4b4a6df..9f99bcf 100644 --- a/.gitignore +++ b/.gitignore @@ -5,3 +5,5 @@ /.remember/ /.claude/ /coverage +/benches/results.jsonl +/benches/run_full.log diff --git a/README.md b/README.md index ffef394..e36c3c7 100644 --- a/README.md +++ b/README.md @@ -107,6 +107,11 @@ Order-of-magnitude numbers from local benchmarking against the legacy PHP engine These are single-machine, order-of-magnitude measurements, not precise benchmarks — treat them as a shape-of-the-win indicator, not a guarantee. +For a reproducible, self-contained benchmark suite — rebuild / churn / search across +1k–500k-document indexes, comparing the PHP extension against Zend Search Lucene on an +identical deterministic corpus, with per-operation Rust heap and process RSS — see +[`benches/`](benches/README.md) (`./benches/run.sh`). + ## Continuous integration The pipeline in `.github/workflows/ci.yml` runs on every push/PR: diff --git a/benches/README.md b/benches/README.md new file mode 100644 index 0000000..e057c26 --- /dev/null +++ b/benches/README.md @@ -0,0 +1,138 @@ +# sdsearch benchmarks + +A reproducible benchmark suite that measures the sdsearch engine on three workloads across +index sizes of **1k / 10k / 50k / 100k / 500k** documents, and compares **our PHP extension +against Zend Search Lucene** on the same corpus and the same queries. + +```bash +# full run (needs the Zend oracle for the comparison — see Prerequisites) +ZEND_LUCENE_PATH=/path/to/zend-framework-1/library ./benches/run.sh + +# fast smoke run (1k + 10k only) +./benches/run.sh --quick +``` + +Output: `benches/results.jsonl` (one JSON line per measurement) and `benches/RESULTS.md` +(rendered tables with cross-engine speedups). `results.jsonl` is machine-specific and +git-ignored; regenerate it locally. + +## What is measured + +Three workloads, each as a function of index size N: + +| Workload | What it does | +|-----------|--------------| +| `rebuild` | Build a **production-shaped** index from scratch: add N docs on top of the committed KB base, then `optimize()` to a single segment (as the host does per batch). | +| `churn` | On an optimized N-doc index, delete the first 1% of docs (by global id) and add 1% fresh docs, then commit (incremental update). | +| `search` | Query three term classes at two realistic paging depths (top-20, top-100) over the **optimized** index and report p50/p95 latency; the true hit count per class is measured separately. | + +Every index the benchmark searches or churns is `optimize()`d to a single segment first — the shape +a production deployment uses — so query cost reflects the real deployment, not a many-segment +work-in-progress index. + +Three engines: + +- **`native`** — the Rust core directly (`sdsearch-core/examples/bench_engine.rs`). This is the + only engine that can report the **exact Rust heap** high-water mark of an operation. +- **`sdsearch`** — our PHP extension (`SdSearch\Writer` / `SdSearch\Engine`), i.e. exactly what a + PHP application uses. +- **`zend`** — `Zend_Search_Lucene`, the pure-PHP engine sdsearch replaces. + +## Memory: three different numbers, on purpose + +- `heap_peak_kb` (**native only**): peak heap bytes attributed to the measured op alone (a + tracking global allocator, reset to steady-state right before the op). PHP's + `memory_get_peak_usage()` **cannot** see this, because the Rust heap lives outside the Zend + memory manager — that is why the native bench exists. +- `rss_peak_kb` / `rss_peak_mb`: process peak resident memory (`VmHWM` from + `/proc/self/status`). It includes both engines' real footprint (the Rust heap shows up here), + so **RSS is the number to compare across engines**. +- `php_peak_mb` (PHP engines): `memory_get_peak_usage(true)`. Meaningful for `zend`, but it + **understates `sdsearch`** (Rust heap invisible), so it is kept in `results.jsonl` for + reference and left out of the report tables. + +Each measurement runs in a **fresh process** so peak-RSS is not cross-contaminated. For `churn` +and `search`, the PHP engines run against a **pre-built index** (an unmeasured `build` step), so +the build's memory does not leak into the churn/search RSS. The native bench builds in-process +but isolates the op's heap with the allocator reset, so its `heap_peak_kb` stays clean (its +`rss_peak_kb` for churn/search may still include the build and is context-only). + +**Native runs in two passes.** The tracking allocator's atomic bookkeeping on every alloc/free +taxes the hot path, so a heap-tracked run's wall time is not comparable to the extension's (which +uses the plain system allocator). The runner therefore measures each native `rebuild`/`churn` +twice: a **time pass** (`BENCH_TRACK_HEAP=0`, no atomics → honest `ms`, the engine floor without +the FFI/JSON boundary) and a **heap pass** (`BENCH_TRACK_HEAP=1` → accurate `heap`, `ms` ignored). +The report takes `ms`/`rss` from the time pass and `heap` from the heap pass. Absolute times are +also sensitive to machine load (a busy laptop / thermal throttling adds noise, especially at +small N where a run is only milliseconds); the **relative** ordering across engines is the stable +signal — run on an idle, AC-powered machine for clean absolute numbers. + +## Corpus and queries (deterministic) + +The corpus is fully deterministic — no RNG — so runs are reproducible and the two engines index +byte-identical documents. Doc `i` has a `title` (~5 tokens), a `body` (~40 tokens drawn from a +fixed 25-word pool), and an `id` keyword. The generator is defined **identically** in +`sdsearch-core/examples/bench_engine.rs` (`gen_docs`) and `tools/bench_compare.php` (`gen_one`); +keep them in sync or the engines stop comparing the same work. + +Three tokens are planted to fix the search result classes at exact, size-independent doc +frequencies (distinct 3-char prefixes so the fuzzy query path, `prefix_len = 3`, can never +cross-match one class to another): + +| Class | Token | Appears in | Query result | +|--------|---------------|------------|--------------| +| `many` | `widetoken` | every doc | ~N hits | +| `few` | `sparsetoken` | 5 docs | 5 hits | +| `none` | `absenttoken` | never | 0 hits | + +Searches use the same **fuzzy + wildcard + parser** boolean on all three engines (mirroring the +engine's `text_subquery` and the host app's query builder), so hit counts match — the report's +`hits` column doubles as a correctness gate: it must be identical across engines. Each query +**reopens the index**, exactly as `SdSearch\Engine::search` and a fresh `Zend_Search_Lucene::open` +do per request; at larger N the index-open cost dominates the low-hit classes. The native column +uses the same `search_index` path, so it measures the engine's real per-request cost minus the +FFI/JSON boundary. Zend's `find()` has no top-K limit (it always scores and returns everything), +so its top-20 and top-100 latencies coincide. + +## Prerequisites + +- A Rust toolchain (builds `examples/bench_engine` and the PHP extension). +- PHP 8.x CLI with the `sdsearch` extension available. If it is not already enabled in `php.ini`, + the runner loads `target/release/libsdsearch.so` with `-d extension=…` automatically. +- **`ZEND_LUCENE_PATH`** pointing at a Zend Search Lucene `library/` root, for the `zend` + comparison. Without it, `zend` runs are recorded as `skipped` and the suite still reports + `native` + `sdsearch`. + +`./benches/run.sh` builds everything in release by default (pass `--no-build` to skip). + +## Zend at large sizes + +Zend indexes in pure PHP and is orders of magnitude slower, so large rebuilds are guarded: + +- `zend` is **skipped for sizes above `BENCH_ZEND_MAX_DOCS` (default 100000)**. Set + `BENCH_ZEND_MAX_DOCS=500000` to attempt the 500k point. +- Every `zend` run has a wall-clock budget `BENCH_ZEND_TIMEOUT` (default 900s); over-budget runs + are recorded as `skipped`, never fatal. +- For sizes at/above `BENCH_ZEND_ULIMIT_FROM` (default 500000) an address-space cap + `BENCH_ZEND_MEM_MB` (MB, `0` = off) is applied; an over-cap run fails its malloc and is skipped. + +`native` and `sdsearch` always run the full 1k–500k range. + +## Flags + +``` +--quick sizes = 1000 10000 only +--sizes "1000 50000" explicit size list +--workloads "rebuild" subset of {rebuild churn search} +--engines "native zend" subset of {native sdsearch zend} +--iters N sampled search iterations (default 50) +--no-build skip the release build +--append append to results.jsonl instead of truncating +``` + +## Files + +- `benches/run.sh` — orchestrator (matrix, guards, collection, report). +- `tools/bench_compare.php` — one measurement (engine × workload × size) → one JSON line. +- `sdsearch-core/examples/bench_engine.rs` — the native bench (exact heap). +- `tools/bench_report.php` — renders `RESULTS.md` from `results.jsonl`. diff --git a/benches/RESULTS.md b/benches/RESULTS.md new file mode 100644 index 0000000..3bb73a6 --- /dev/null +++ b/benches/RESULTS.md @@ -0,0 +1,80 @@ +# sdsearch benchmarks + +Generated by `benches/run.sh` → `tools/bench_report.php`. Reproducible: deterministic corpus, fixed iterations, warm-up discarded. See [README.md](README.md) for the corpus and query definitions. + +| | | +|---|---| +| Date | 2026-07-12 05:27 | +| Commit | `6c3c2b7` | +| Host | Linux 7.0.0-27-generic | +| CPU | 13th Gen Intel(R) Core(TM) i7-1355U | +| PHP | 8.4.23 | + +**Engines.** `native` = the Rust core (examples/bench_engine) — reports the EXACT Rust heap. `sdsearch` = our PHP extension. `zend` = Zend_Search_Lucene (pure PHP). + +**Memory columns.** `heap` (native only) = peak Rust heap attributed to the op (`memory_get_peak_usage()` cannot see this). `rss` = process peak RSS (VmHWM), the cross-engine-comparable footprint. `php_peak` (PHP engines) omitted from tables because it understates sdsearch (Rust heap is invisible to it); it is in results.jsonl for reference. + +## rebuild — Rebuild (build the whole index from the KB base) + +| docs | native ms | native heap MB | native rss MB | sdsearch ms | sdsearch rss MB | zend ms | zend rss MB | speedup vs zend | doc_count | +|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:| +| 1,000 | 20.07 | 3.2 | 8.8 | 20.09 | 47.5 | 1,509.67 | 45.4 | 75.1× | 1,020 | +| 10,000 | 139.60 | 3.2 | 12.7 | 185.15 | 51.1 | 18,128.12 | 51.6 | 97.9× | 10,020 | +| 50,000 | 714.32 | 11.8 | 34.9 | 932.03 | 74.2 | 92,048.93 | 68.6 | 98.8× | 50,020 | +| 100,000 | 1,340.99 | 23.5 | 65.5 | 1,816.75 | 104.5 | 209,928.19 | 91.9 | 115.6× | 100,020 | +| 500,000 | 7,273.32 | 98.6 | 304.7 | 9,840.58 | 342.6 | 1,063,693.86 | 208.9 | 108.1× | 500,020 | + +## churn — Churn (delete 1% of docs + add 1% + commit) + +| docs | native ms | native heap MB | native rss MB | sdsearch ms | sdsearch rss MB | zend ms | zend rss MB | speedup vs zend | doc_count | +|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:| +| 1,000 | 0.25 | 0.2 | 8.6 | 1.06 | 43.5 | 6.66 | 44.5 | 6.3× | 1,020 | +| 10,000 | 1.38 | 0.5 | 12.8 | 3.19 | 43.9 | 93.93 | 44.6 | 29.4× | 10,020 | +| 50,000 | 6.54 | 1.7 | 35.1 | 11.34 | 45.3 | 461.22 | 44.8 | 40.7× | 50,020 | +| 100,000 | 13.25 | 3.2 | 65.3 | 21.48 | 51.0 | 927.81 | 45.2 | 43.2× | 100,020 | +| 500,000 | 73.97 | 9.3 | 305.7 | 111.76 | 65.5 | 5,861.84 | 46.2 | 52.5× | 500,020 | + +## search — term latency by result class (many / few / none) + +p50 latency in ms, measured at two realistic paging depths. `hits` is the TRUE match count for the class (must be identical across engines — correctness gate). Each query reopens the index, as the engines do per request; at larger N the index-open cost dominates low-hit queries. Zend's `find()` has no top-K limit, so its top-20 and top-100 latencies coincide. + +### top-20 + +| docs | class | native p50 | sdsearch p50 | zend p50 | speedup vs zend | hits | +|---:|---|---:|---:|---:|---:|---:| +| 1,000 | many | 2.4733 | 1.1220 | 45.2299 | 40.3× | 1,000 | +| 1,000 | few | 0.5714 | 0.5610 | 23.1121 | 41.2× | 5 | +| 1,000 | none | 0.6404 | 0.6249 | 23.0150 | 36.8× | 0 | +| 10,000 | many | 5.9287 | 5.5599 | 237.5951 | 42.7× | 10,000 | +| 10,000 | few | 0.6820 | 0.6981 | 28.3499 | 40.6× | 5 | +| 10,000 | none | 0.7604 | 0.7601 | 28.3911 | 37.4× | 0 | +| 50,000 | many | 28.2357 | 26.2308 | 1,085.1121 | 41.4× | 50,000 | +| 50,000 | few | 0.6883 | 0.6790 | 19.2802 | 28.4× | 5 | +| 50,000 | none | 0.7198 | 0.6940 | 19.0599 | 27.5× | 0 | +| 100,000 | many | 54.3672 | 59.6919 | 2,063.2579 | 34.6× | 100,000 | +| 100,000 | few | 0.8188 | 0.8190 | 17.0100 | 20.8× | 5 | +| 100,000 | none | 0.7656 | 0.7391 | 13.1478 | 17.8× | 0 | +| 500,000 | many | 539.0156 | 559.2749 | 10,325.0971 | 18.5× | 500,000 | +| 500,000 | few | 2.4895 | 2.3952 | 20.0241 | 8.4× | 5 | +| 500,000 | none | 2.4090 | 2.2871 | 13.8309 | 6.0× | 0 | + +### top-100 + +| docs | class | native p50 | sdsearch p50 | zend p50 | speedup vs zend | hits | +|---:|---|---:|---:|---:|---:|---:| +| 1,000 | many | 1.1545 | 1.2589 | 44.9741 | 35.7× | 1,000 | +| 1,000 | few | 0.5645 | 0.5529 | 23.1988 | 42.0× | 5 | +| 1,000 | none | 0.6391 | 0.6192 | 23.1290 | 37.4× | 0 | +| 10,000 | many | 5.9922 | 5.6951 | 235.9731 | 41.4× | 10,000 | +| 10,000 | few | 0.6676 | 0.7029 | 28.5289 | 40.6× | 5 | +| 10,000 | none | 0.7790 | 0.7539 | 28.1141 | 37.3× | 0 | +| 50,000 | many | 28.4841 | 26.7911 | 1,089.5112 | 40.7× | 50,000 | +| 50,000 | few | 0.6864 | 0.6800 | 19.3150 | 28.4× | 5 | +| 50,000 | none | 0.7140 | 0.7010 | 19.1391 | 27.3× | 0 | +| 100,000 | many | 57.8445 | 58.9969 | 2,057.3859 | 34.9× | 100,000 | +| 100,000 | few | 0.8201 | 0.7961 | 17.0660 | 21.4× | 5 | +| 100,000 | none | 0.7655 | 0.7439 | 13.1581 | 17.7× | 0 | +| 500,000 | many | 543.4759 | 561.6472 | 10,282.6638 | 18.3× | 500,000 | +| 500,000 | few | 2.5341 | 2.3811 | 19.9978 | 8.4× | 5 | +| 500,000 | none | 2.4181 | 2.3220 | 13.7968 | 5.9× | 0 | + diff --git a/benches/run.sh b/benches/run.sh new file mode 100755 index 0000000..8621ace --- /dev/null +++ b/benches/run.sh @@ -0,0 +1,206 @@ +#!/usr/bin/env bash +# +# sdsearch benchmark orchestrator: runs the full {engine × size × workload} matrix, each +# measurement in a FRESH process (so peak-RSS is not cross-contaminated), collects one JSON +# line per run into benches/results.jsonl, and renders benches/RESULTS.md. +# +# Engines: +# native — the Rust core via examples/bench_engine (reports EXACT heap_peak_kb + rss). +# sdsearch — our PHP extension (SdSearch\Writer / SdSearch\Engine); RSS is the comparable mem. +# zend — Zend_Search_Lucene (requires ZEND_LUCENE_PATH). Pure-PHP, so it is SLOW to index; +# guarded by a timeout (+ an address-space cap on very large sizes). If a guarded +# run does not finish it is recorded as "skipped", not fatal. +# +# Workloads: rebuild (build N docs from the KB base), churn (delete 1% + add 1%), search +# (many/few/none result classes). See benches/README.md for the exact corpus + query definitions. +# +# Usage: +# ./benches/run.sh [--quick] [--sizes "1000 10000 ..."] [--workloads "rebuild churn search"] +# [--engines "native sdsearch zend"] [--iters N] [--no-build] [--append] +# +# Env knobs (defaults in brackets): +# ZEND_LUCENE_PATH path to a ZSL library/ root; unset => zend engine is skipped. +# BENCH_ZEND_MAX_DOCS [100000] zend is skipped for sizes strictly above this. +# BENCH_ZEND_TIMEOUT [900] per-zend-run wall-clock budget (seconds); over => skipped. +# BENCH_ZEND_MEM_MB [0] address-space cap (MB) applied to zend runs at/above +# BENCH_ZEND_ULIMIT_FROM; 0 disables the cap. +# BENCH_ZEND_ULIMIT_FROM [500000] size at/above which BENCH_ZEND_MEM_MB is enforced. +set -uo pipefail + +REPO="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +cd "$REPO" + +SIZES="1000 10000 50000 100000 500000" +WORKLOADS="rebuild churn search" +ENGINES="native sdsearch zend" +ITERS=50 +DO_BUILD=1 +APPEND=0 + +while [[ $# -gt 0 ]]; do + case "$1" in + --quick) SIZES="1000 10000"; shift ;; + --sizes) SIZES="$2"; shift 2 ;; + --workloads) WORKLOADS="$2"; shift 2 ;; + --engines) ENGINES="$2"; shift 2 ;; + --iters) ITERS="$2"; shift 2 ;; + --no-build) DO_BUILD=0; shift ;; + --append) APPEND=1; shift ;; + *) echo "unknown flag: $1" >&2; exit 2 ;; + esac +done + +BENCH_ZEND_MAX_DOCS="${BENCH_ZEND_MAX_DOCS:-100000}" +BENCH_ZEND_TIMEOUT="${BENCH_ZEND_TIMEOUT:-900}" +BENCH_ZEND_MEM_MB="${BENCH_ZEND_MEM_MB:-0}" +BENCH_ZEND_ULIMIT_FROM="${BENCH_ZEND_ULIMIT_FROM:-500000}" + +RESULTS="$REPO/benches/results.jsonl" +NATIVE_BIN="$REPO/target/release/examples/bench_engine" +EXT_SO="$REPO/target/release/libsdsearch.so" +COMPARE="$REPO/tools/bench_compare.php" +SCRATCH="${TMPDIR:-/tmp}/sdsearch_bench_$$" +mkdir -p "$SCRATCH" +trap 'rm -rf "$SCRATCH"' EXIT + +log() { echo "[bench] $*" >&2; } + +# ---- build (release) ---- +if [[ "$DO_BUILD" == 1 ]]; then + log "building native example + PHP extension (release)…" + cargo build -p sdsearch-core --release --example bench_engine >&2 || { log "native build failed"; exit 1; } + cargo build -p sdsearch-php --release >&2 || { log "extension build failed"; exit 1; } +fi +[[ -x "$NATIVE_BIN" ]] || { log "missing $NATIVE_BIN (drop --no-build or build it)"; exit 1; } + +# ---- environment detection ---- +# Only add -d extension if the .so is not already loaded by php.ini (avoids a "already loaded" +# warning that would pollute stdout). +PHP_EXT=() +if php -m 2>/dev/null | grep -qi '^sdsearch$'; then + log "sdsearch extension already loaded by php.ini" +elif [[ -f "$EXT_SO" ]]; then + PHP_EXT=(-d "extension=$EXT_SO") + log "loading sdsearch extension from $EXT_SO" +fi + +ZEND_OK=0 +if [[ -n "${ZEND_LUCENE_PATH:-}" && -f "${ZEND_LUCENE_PATH%/}/Zend/Search/Lucene.php" ]]; then + ZEND_OK=1 + log "Zend oracle: $ZEND_LUCENE_PATH" +else + log "ZEND_LUCENE_PATH unset/invalid — zend engine will be recorded as skipped" +fi + +[[ "$APPEND" == 1 ]] || : > "$RESULTS" + +# appends a raw JSON line (already emitted by a bench) to results.jsonl. +emit_line() { printf '%s\n' "$1" >> "$RESULTS"; } +# appends a synthetic skip record. +emit_skip() { + emit_line "{\"engine\":\"$1\",\"workload\":\"$2\",\"n\":$3,\"status\":\"skipped\",\"reason\":\"$4\"}" + log "SKIP $1/$2 n=$3 ($4)" +} +# runs a command, extracts its LAST JSON object line ({...}) and records it; logs on failure. +capture() { + local engine="$1" workload="$2" n="$3"; shift 3 + local out rc line + out="$("$@" 2>/dev/null)"; rc=$? + line="$(printf '%s\n' "$out" | grep -E '^\{.*\}$' | tail -1)" + if [[ $rc -ne 0 || -z "$line" ]]; then + emit_skip "$engine" "$workload" "$n" "exit=$rc/no-json" + return + fi + emit_line "$line" + log "OK $engine/$workload n=$n" +} + +php_run() { php "${PHP_EXT[@]}" "$COMPARE" "$@"; } + +# runs a zend measurement under the timeout (+ optional address-space cap); records skip on any +# non-zero/timeout/OOM exit. Returns the captured JSON via capture(). +zend_run() { + local workload="$1" n="$2"; shift 2 # remaining args: extra bench_compare args + local cap_kb=0 + if [[ "$BENCH_ZEND_MEM_MB" -gt 0 && "$n" -ge "$BENCH_ZEND_ULIMIT_FROM" ]]; then + cap_kb=$((BENCH_ZEND_MEM_MB * 1024)) + fi + capture zend "$workload" "$n" bash -c ' + [[ "$1" -gt 0 ]] && ulimit -v "$1" + shift + exec timeout "$1"s "${@:2}" + ' _ "$cap_kb" "$BENCH_ZEND_TIMEOUT" php "$COMPARE" zend "$workload" "$n" "$@" +} + +for n in $SIZES; do + log "===== size N=$n =====" + + for engine in $ENGINES; do + case "$engine" in + native) + # rebuild/churn run TWICE: a TIME pass (tracking off → honest ms/rss, the engine floor) + # and a HEAP pass (tracking on → accurate heap, ms taxed & ignored). search runs the + # TIME pass only (its heap is not a headline). The report merges the two passes. + for w in $WORKLOADS; do + case "$w" in + search) + capture native search "$n" env BENCH_TRACK_HEAP=0 "$NATIVE_BIN" search "$n" "$ITERS" ;; + *) + capture native "$w" "$n" env BENCH_TRACK_HEAP=0 "$NATIVE_BIN" "$w" "$n" # time + capture native "$w" "$n" env BENCH_TRACK_HEAP=1 "$NATIVE_BIN" "$w" "$n" ;; # heap + esac + done + ;; + + sdsearch) + pidx="$SCRATCH/sd_$n" + need_persist=0 + for w in $WORKLOADS; do [[ "$w" == churn || "$w" == search ]] && need_persist=1; done + [[ "$need_persist" == 1 ]] && php_run sdsearch build "$n" 0 "$pidx" >/dev/null 2>&1 + for w in $WORKLOADS; do + case "$w" in + rebuild) capture sdsearch rebuild "$n" php "${PHP_EXT[@]}" "$COMPARE" sdsearch rebuild "$n" ;; + churn) capture sdsearch churn "$n" php "${PHP_EXT[@]}" "$COMPARE" sdsearch churn "$n" 0 "$pidx" ;; + search) capture sdsearch search "$n" php "${PHP_EXT[@]}" "$COMPARE" sdsearch search "$n" "$ITERS" "$pidx" ;; + esac + done + rm -rf "$pidx" + ;; + + zend) + if [[ "$ZEND_OK" != 1 ]]; then + for w in $WORKLOADS; do emit_skip zend "$w" "$n" "ZEND_LUCENE_PATH unset"; done + continue + fi + if [[ "$n" -gt "$BENCH_ZEND_MAX_DOCS" ]]; then + for w in $WORKLOADS; do emit_skip zend "$w" "$n" "n>BENCH_ZEND_MAX_DOCS=$BENCH_ZEND_MAX_DOCS"; done + continue + fi + zidx="$SCRATCH/zend_$n" + need_persist=0 + for w in $WORKLOADS; do [[ "$w" == churn || "$w" == search ]] && need_persist=1; done + if [[ "$need_persist" == 1 ]]; then + # build under the same guard; if it fails, churn/search for this size are skipped. + if ! zend_build_out="$(timeout "${BENCH_ZEND_TIMEOUT}s" php "$COMPARE" zend build "$n" 0 "$zidx" 2>/dev/null)"; then + log "zend build n=$n failed/timed out — churn/search skipped" + fi + fi + for w in $WORKLOADS; do + case "$w" in + rebuild) zend_run rebuild "$n" ;; + churn) [[ -d "$zidx" ]] && zend_run churn "$n" 0 "$zidx" || emit_skip zend churn "$n" "no prebuilt index" ;; + search) [[ -d "$zidx" ]] && zend_run search "$n" "$ITERS" "$zidx" || emit_skip zend search "$n" "no prebuilt index" ;; + esac + done + rm -rf "$zidx" + ;; + + *) log "unknown engine: $engine"; exit 2 ;; + esac + done +done + +log "rendering report → benches/RESULTS.md" +php "$REPO/tools/bench_report.php" "$RESULTS" > "$REPO/benches/RESULTS.md" \ + && log "done. results: benches/results.jsonl report: benches/RESULTS.md" \ + || log "report rendering failed (results.jsonl is intact)" diff --git a/sdsearch-core/examples/bench_engine.rs b/sdsearch-core/examples/bench_engine.rs new file mode 100644 index 0000000..53be732 --- /dev/null +++ b/sdsearch-core/examples/bench_engine.rs @@ -0,0 +1,376 @@ +//! NATIVE engine benchmark: characterizes the sdsearch Rust engine on three workloads as a +//! function of index size N, reporting wall time AND the two memory numbers the PHP harness +//! cannot obtain: +//! - `heap_peak_kb`: peak HEAP bytes attributed to the measured op alone (tracking global +//! allocator; the counter is reset to steady-state right before the op). This is the +//! number PHP's `memory_get_peak_usage()` cannot see, because the Rust heap lives outside +//! the Zend memory manager. +//! - `rss_peak_kb`: process peak RSS (VmHWM from /proc/self/status). Includes the mmap of the +//! source segments; this is the number comparable to the PHP harness's RSS reading. +//! +//! This example does NOT measure Zend — it characterizes our engine only. The cross-engine +//! comparison (sdsearch extension vs Zend) lives in `tools/bench_compare.php`. +//! +//! Every workload starts from a COPY of the committed KB fixture (same base as +//! `optimize_bench.rs` / `perf_writer.php`), so the three engines share an identical starting +//! point. The ~20 base docs are negligible noise at N >= 1000. The measured op adds/searches +//! `N` synthetic docs with planted terms (see `gen_one`) so the three search classes have an +//! exact, size-independent doc_freq. +//! +//! Usage: +//! cargo run -p sdsearch-core --release --example bench_engine -- rebuild [cap] +//! cargo run -p sdsearch-core --release --example bench_engine -- churn [cap] +//! cargo run -p sdsearch-core --release --example bench_engine -- search [iters] +//! +//! Prints ONE JSON line per run, e.g.: +//! {"engine":"native","workload":"rebuild","n":1000,"ms":12.34,"heap_peak_kb":.., +//! "rss_peak_kb":..,"doc_count":1020} +//! {"engine":"native","workload":"search","n":1000,"iters":50,"heap_peak_kb":..,"rss_peak_kb":.., +//! "many":{"hits":1000,"top20":{"p50_ms":..,"p95_ms":..},"top100":{"p50_ms":..,"p95_ms":..}}, +//! "few":{...}, "none":{...}} +//! Search latency is measured at two realistic paging depths (top-20 and top-100); `hits` is the +//! TRUE match count for the class (an unlimited call, unmeasured). + +use sdsearch_core::index::IndexReader; +use sdsearch_core::query::QueryParams; +use sdsearch_core::zsl::index::ZslIndex; +use sdsearch_core::zsl::runner::search_index; +use sdsearch_core::zsl::writer::{FieldKind, IndexWriter, WriterDoc, WriterField, WriterOpts}; +use std::alloc::{GlobalAlloc, Layout, System}; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; + +// ---- tracking allocator: measures live + peak heap bytes ---- +// +// The atomic bookkeeping on every alloc/free is not free — it taxes the hot path, so a +// heap-tracked run's WALL TIME is NOT comparable to the extension's (which uses the plain +// system allocator). We therefore run natives in two passes: a TIME pass with TRACK=false (no +// atomics → honest ms, the "engine floor" without the FFI/JSON boundary) and a HEAP pass with +// TRACK=true (accurate heap, ms ignored). `BENCH_TRACK_HEAP=0` selects the time pass. +static LIVE: AtomicUsize = AtomicUsize::new(0); +static PEAK: AtomicUsize = AtomicUsize::new(0); +static TRACK: AtomicBool = AtomicBool::new(true); + +struct TrackingAlloc; + +unsafe impl GlobalAlloc for TrackingAlloc { + unsafe fn alloc(&self, layout: Layout) -> *mut u8 { + let p = System.alloc(layout); + if !p.is_null() && TRACK.load(Ordering::Relaxed) { + let live = LIVE.fetch_add(layout.size(), Ordering::Relaxed) + layout.size(); + PEAK.fetch_max(live, Ordering::Relaxed); + } + p + } + unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) { + System.dealloc(ptr, layout); + if TRACK.load(Ordering::Relaxed) { + LIVE.fetch_sub(layout.size(), Ordering::Relaxed); + } + } + unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, new_size: usize) -> *mut u8 { + let p = System.realloc(ptr, layout, new_size); + if !p.is_null() && TRACK.load(Ordering::Relaxed) { + let old = layout.size(); + if new_size >= old { + let live = LIVE.fetch_add(new_size - old, Ordering::Relaxed) + (new_size - old); + PEAK.fetch_max(live, Ordering::Relaxed); + } else { + LIVE.fetch_sub(old - new_size, Ordering::Relaxed); + } + } + p + } +} + +/// whether this process is the HEAP pass (TRACK on) or the TIME pass (TRACK off). Read once from +/// `BENCH_TRACK_HEAP` (default: on). Startup allocations before this are minimal and precede any +/// `reset_peak`, so LIVE accounting stays consistent for the whole measured region. +fn init_track() -> bool { + let on = std::env::var("BENCH_TRACK_HEAP") + .map(|v| v != "0") + .unwrap_or(true); + TRACK.store(on, Ordering::Relaxed); + on +} + +#[global_allocator] +static GLOBAL: TrackingAlloc = TrackingAlloc; + +/// Anchors PEAK to the current live bytes, so the next PEAK read reflects only what is +/// allocated from here on (i.e. the measured op's own heap high-water mark above steady state). +fn reset_peak() { + let live = LIVE.load(Ordering::Relaxed); + PEAK.store(live, Ordering::Relaxed); +} + +fn heap_peak_kb() -> u64 { + PEAK.load(Ordering::Relaxed) as u64 / 1024 +} + +/// "heap" when the tracking allocator is on (heap number is accurate, ms is taxed), "time" +/// otherwise (ms is honest, heap is meaningless). The report keys native rows on this. +fn pass_label() -> &'static str { + if TRACK.load(Ordering::Relaxed) { + "heap" + } else { + "time" + } +} + +/// process peak RSS (VmHWM from /proc/self/status), in KB; 0 if unavailable. +fn rss_peak_kb() -> u64 { + std::fs::read_to_string("/proc/self/status") + .ok() + .and_then(|s| { + s.lines() + .find(|l| l.starts_with("VmHWM:")) + .and_then(|l| l.split_whitespace().nth(1)) + .and_then(|kb| kb.parse().ok()) + }) + .unwrap_or(0) +} + +// ---- planted-term corpus (MUST stay in sync with tools/bench_compare.php) ---- +// +// Every added doc's `body` carries COMMON in every doc (doc_freq == N → "many"); RARE is +// planted in ~RARE_K docs, spread out (doc_freq == RARE_K → "few"); MISSING is never emitted +// (doc_freq == 0 → "none"). The three query tokens are chosen so search cost is exact and +// size-independent across engines. +// Distinct 3-char prefixes (wid/spa/abs) so the PHP fuzzy path (prefix_len 3) can never +// cross-match one class token to another — the three classes stay exact on both engines. +const COMMON: &str = "widetoken"; +const RARE: &str = "sparsetoken"; +const MISSING: &str = "absenttoken"; +const RARE_K: usize = 5; + +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", +]; + +/// True when doc `i` of an N-doc batch should carry the RARE token, spread over the batch so +/// its doc_freq is `RARE_K` (approximately, when N is not a multiple of RARE_K). +fn is_rare(i: usize, n: usize) -> bool { + let step = (n / RARE_K).max(1); + i.is_multiple_of(step) && i / step < RARE_K +} + +/// ONE deterministic doc numbered `i`. `id_prefix` keeps churn-added docs in a distinct id +/// namespace from the rebuild docs. Body: 40 POOL tokens + COMMON (+ RARE for planted docs) + +/// `ref{i}`. Generated one at a time (never a whole-corpus Vec) so the measured heap reflects +/// the ENGINE, not the corpus. Matches `gen_one` in tools/bench_compare.php token-for-token. +fn gen_one(i: usize, n: usize, id_prefix: &str) -> WriterDoc { + let np = POOL.len(); + // title/body draw ONLY from the fixed POOL (bounded vocabulary, like real text) — no unique + // per-doc token, so the term dictionary stays realistic. The only unique-per-doc term is the + // `id` keyword (as a real index of N records has N ids). + let title = format!( + "ticket {} {} {}", + POOL[i % np], + POOL[(i * 3) % np], + POOL[(i * 7) % np] + ); + let mut body: String = (0..40) + .map(|j| POOL[(i * 7 + j * 5) % np]) + .collect::>() + .join(" "); + body.push(' '); + body.push_str(COMMON); + if is_rare(i, n) { + body.push(' '); + body.push_str(RARE); + } + 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!("{id_prefix}-{i}"), + kind: FieldKind::Keyword, + stored: true, + }, + ], + } +} + +/// copies the committed KB fixture to a fresh temp dir (skips locks and `.sti`), returns it. +fn copy_kb_base(tag: &str) -> PathBuf { + let src = PathBuf::from(concat!( + env!("CARGO_MANIFEST_DIR"), + "/tests/fixtures/zsl_index_kb" + )); + let dst = std::env::temp_dir().join(format!("sdsearch_bench_{}_{}", std::process::id(), tag)); + if dst.is_dir() { + std::fs::remove_dir_all(&dst).ok(); + } + std::fs::create_dir_all(&dst).expect("create temp dir"); + for entry in std::fs::read_dir(&src).expect("read KB fixture") { + let p = entry.unwrap().path(); + let name = p.file_name().unwrap().to_string_lossy().to_string(); + if name.contains("lock") || name.ends_with(".sti") { + continue; + } + std::fs::copy(&p, dst.join(&name)).expect("copy fixture file"); + } + dst +} + +/// builds an N-doc index on top of a fresh KB base copy and OPTIMIZES it to a single segment +/// (production shape — the host runs `optimize()` per batch). Unmeasured setup for churn/search. +fn build_index(n: usize, cap: usize, tag: &str) -> PathBuf { + let dir = copy_kb_base(tag); + let opts = WriterOpts { + max_buffered_docs: cap, + ..WriterOpts::default() + }; + let mut w = IndexWriter::open(&dir, opts).expect("open (build) failed"); + for i in 0..n { + w.add_document(gen_one(i, n, "REC")) + .expect("add_document failed"); + } + w.optimize().expect("optimize failed"); + dir +} + +/// `rebuild [cap]`: measures producing a PRODUCTION-shaped index from scratch — copy the KB +/// base, add N docs, then `optimize()` (merge to a single segment, as the host does per batch). +/// The copy is unmeasured (filesystem, not heap); the writer open/add/optimize is the measured op. +fn run_rebuild(n: usize, cap: usize) { + let dir = copy_kb_base("rebuild"); + reset_peak(); + let t0 = std::time::Instant::now(); + let opts = WriterOpts { + max_buffered_docs: cap, + ..WriterOpts::default() + }; + let mut w = IndexWriter::open(&dir, opts).expect("open (build) failed"); + for i in 0..n { + w.add_document(gen_one(i, n, "REC")) + .expect("add_document failed"); + } + w.optimize().expect("optimize failed"); + let ms = t0.elapsed().as_secs_f64() * 1000.0; + // total LIVE docs (base + added) via a post-commit reader — CommitReport.doc_count is the + // per-session count, not the index total (see index_writer.rs commit_inner). + let doc_count = ZslIndex::open(&dir).map(|r| r.num_docs()).unwrap_or(0); + println!( + "{{\"engine\":\"native\",\"pass\":\"{}\",\"workload\":\"rebuild\",\"n\":{},\"cap\":{},\"ms\":{:.3},\"heap_peak_kb\":{},\"rss_peak_kb\":{},\"doc_count\":{}}}", + pass_label(), n, cap, ms, heap_peak_kb(), rss_peak_kb(), doc_count + ); + std::fs::remove_dir_all(&dir).ok(); +} + +/// `churn [cap]`: on a committed N-doc index, deletes the first 1% of docs (by global id) +/// and adds 1% fresh docs, then commits — the measured op. The initial build is unmeasured. +fn run_churn(n: usize, cap: usize) { + let dir = build_index(n, cap, "churn"); + let one_pct = (n / 100).max(1); + let doc_count_before = ZslIndex::open(&dir).expect("open reader").num_docs(); + + reset_peak(); + let t0 = std::time::Instant::now(); + let opts = WriterOpts { + max_buffered_docs: cap, + ..WriterOpts::default() + }; + let mut w = IndexWriter::open(&dir, opts).expect("open (churn) failed"); + for gid in 0..one_pct { + w.delete_document(gid); + } + for i in 0..one_pct { + w.add_document(gen_one(i, n, "CHURN")) + .expect("add_document failed"); + } + w.commit().expect("commit failed"); + let ms = t0.elapsed().as_secs_f64() * 1000.0; + let doc_count = ZslIndex::open(&dir).map(|r| r.num_docs()).unwrap_or(0); + println!( + "{{\"engine\":\"native\",\"pass\":\"{}\",\"workload\":\"churn\",\"n\":{},\"cap\":{},\"pct1\":{},\"ms\":{:.3},\"heap_peak_kb\":{},\"rss_peak_kb\":{},\"doc_count_before\":{},\"doc_count\":{}}}", + pass_label(), n, cap, one_pct, ms, heap_peak_kb(), rss_peak_kb(), doc_count_before, doc_count + ); + std::fs::remove_dir_all(&dir).ok(); +} + +/// the free-text query params for `token`, IDENTICAL to what the PHP extension builds +/// (fuzzy 0.5 / prefix 3, wildcard min-prefix 0) — so the native and sdsearch search columns +/// measure the same query, minus the FFI/JSON boundary. +fn params_for(token: &str) -> QueryParams { + QueryParams { + text: token.to_string(), + where_groups: vec![], + in_groups: vec![], + fuzzy_similarity: 0.5, + fuzzy_prefix_len: 3, + wildcard_min_prefix: 0, + } +} + +/// times `iters` `search_index` runs at `limit`, discards a warm-up, returns (p50, p95) in ms. +/// Reopens the index per call, exactly as SdSearch\Engine::search does, so the two are comparable. +fn time_search(dir: &Path, params: &QueryParams, limit: usize, iters: usize) -> (f64, f64) { + let _ = search_index(dir, params, 0.0, limit); // warm-up (not sampled) + let mut samples: Vec = Vec::with_capacity(iters); + for _ in 0..iters { + let t0 = std::time::Instant::now(); + let r = search_index(dir, params, 0.0, limit).expect("search failed"); + samples.push(t0.elapsed().as_secs_f64() * 1000.0); + std::hint::black_box(r); + } + samples.sort_by(|a, b| a.partial_cmp(b).unwrap()); + let p = |q: f64| samples[((samples.len() - 1) as f64 * q).round() as usize]; + (p(0.50), p(0.95)) +} + +/// `search [iters]`: builds an N-doc index (unmeasured), then times the three query classes +/// at two realistic paging depths (top-20 and top-100). Reports the TRUE hit count per class +/// (one unlimited call, unmeasured) alongside the paged latencies. +fn run_search(n: usize, iters: usize) { + let dir = build_index(n, 1000, "search"); + + reset_peak(); + let mut parts: Vec = Vec::new(); + for (label, token) in [("many", COMMON), ("few", RARE), ("none", MISSING)] { + let params = params_for(token); + let hits = search_index(&dir, ¶ms, 0.0, 0).expect("count").len(); // true freq (unlimited) + let (t20_50, t20_95) = time_search(&dir, ¶ms, 20, iters); + let (t100_50, t100_95) = time_search(&dir, ¶ms, 100, iters); + parts.push(format!( + "\"{label}\":{{\"hits\":{hits},\"top20\":{{\"p50_ms\":{t20_50:.4},\"p95_ms\":{t20_95:.4}}},\ +\"top100\":{{\"p50_ms\":{t100_50:.4},\"p95_ms\":{t100_95:.4}}}}}" + )); + } + println!( + "{{\"engine\":\"native\",\"pass\":\"{}\",\"workload\":\"search\",\"n\":{},\"iters\":{},\"heap_peak_kb\":{},\"rss_peak_kb\":{},{}}}", + pass_label(), n, iters, heap_peak_kb(), rss_peak_kb(), parts.join(",") + ); + std::fs::remove_dir_all(&dir).ok(); +} + +fn main() { + init_track(); // TIME pass (BENCH_TRACK_HEAP=0) vs HEAP pass (default), before any measured op. + let args: Vec = std::env::args().collect(); + let workload = args.get(1).map(String::as_str).unwrap_or("rebuild"); + let n: usize = args.get(2).and_then(|s| s.parse().ok()).unwrap_or(1000); + match workload { + "rebuild" => run_rebuild(n, args.get(3).and_then(|s| s.parse().ok()).unwrap_or(1000)), + "churn" => run_churn(n, args.get(3).and_then(|s| s.parse().ok()).unwrap_or(1000)), + "search" => run_search(n, args.get(3).and_then(|s| s.parse().ok()).unwrap_or(50)), + other => { + eprintln!("unknown workload: {other} (expected rebuild|churn|search)"); + std::process::exit(2); + } + } +} diff --git a/sdsearch-core/src/zsl/postings.rs b/sdsearch-core/src/zsl/postings.rs index 8c6baf3..e14484a 100644 --- a/sdsearch-core/src/zsl/postings.rs +++ b/sdsearch-core/src/zsl/postings.rs @@ -118,7 +118,7 @@ mod tests { use super::*; use crate::zsl::cfs::CompoundFile; use crate::zsl::fields::read_field_infos; - use crate::zsl::terms::TermDict; + use crate::zsl::terms::EagerTermDict; use std::path::PathBuf; fn cfs() -> CompoundFile { @@ -147,9 +147,9 @@ mod tests { .into_iter() .map(|f| f.name) .collect(); - let dict = TermDict::read(&sub(".tis"), &names).unwrap(); + let dict = EagerTermDict::read(&sub(".tis"), &names).unwrap(); let info = dict.info("title", "new").unwrap(); - let freqs = read_freqs(&sub(".frq"), info).unwrap(); + let freqs = read_freqs(&sub(".frq"), &info).unwrap(); // "new" is in all 4 docs (all "New workflow"), freq 1 each assert_eq!(freqs, vec![(0, 1), (1, 1), (2, 1), (3, 1)]); } @@ -165,26 +165,31 @@ mod tests { } #[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() + fn for_each_posting_decodes_written_postings() { + // Independent ground truth: encode known postings with the writer, then assert the + // decoder reproduces them EXACTLY. Doc ids ascend and the per-doc position counts VARY + // (3, 1, 4, 1) so a scratch buffer that is not cleared between docs would leak stale + // positions into later docs and fail here. NOT compared against `read_all_positions` + // (which now delegates to `for_each_posting`, so that comparison is tautological). + let docs: Vec<(usize, Vec)> = vec![ + (0, vec![0, 3, 7]), + (2, vec![5]), + (5, vec![1, 2, 9, 40]), + (6, vec![0]), + ]; + + let mut frq: Vec = Vec::new(); + let mut prx: Vec = Vec::new(); + let (freq_pointer, prox_pointer) = + crate::zsl::writer::postings::write_term_postings(&mut frq, &mut prx, &docs); + let info = TermInfo { + doc_freq: docs.len() as u32, + freq_pointer, + prox_pointer, }; - 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); + for_each_posting(&frq, &prx, &info, |doc, pos| got.push((doc, pos.to_vec()))).unwrap(); + assert_eq!(got, docs); } } diff --git a/sdsearch-core/src/zsl/segment.rs b/sdsearch-core/src/zsl/segment.rs index 1741cce..da11ec0 100644 --- a/sdsearch-core/src/zsl/segment.rs +++ b/sdsearch-core/src/zsl/segment.rs @@ -6,7 +6,7 @@ use crate::zsl::fields::{read_field_infos, FieldInfo}; use crate::zsl::norms::{approx_field_len, read_norms}; 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 crate::zsl::terms::{TermCursor, TermDict, TermInfo}; use std::collections::HashMap; use std::path::Path; @@ -72,7 +72,33 @@ impl ZslSegment { return; }; // degrade: a corrupt tail stops iteration rather than panicking across FFI - let _ = for_each_posting(frq, prx, ti, |d, pos| { + let _ = for_each_posting(frq, prx, &ti, |d, pos| { + if !self.deletes.is_deleted(d) { + f(d, pos); + } + }); + } + + /// Like [`for_each_live_posting`](Self::for_each_live_posting) but takes a `TermInfo` + /// the caller already has (captured from a [`TermCursor::peek_info`] during the merge) + /// instead of resolving it via `self.dict.info(field, term)`. Semantics are otherwise + /// identical: no-op if `.prx` is absent, deletes filtered, a corrupt tail stops iteration + /// rather than panicking across FFI. + /// + /// This is the merge's fast path: the lazy `TermDict::info` forward-decodes up to + /// `INDEX_INTERVAL` (128) `.tis` entries per call, so re-looking-up every term during a + /// full-index merge is O(term_count × 128); the cursor already decoded each `TermInfo` + /// while walking `.tis` once, so threading it through here avoids all the re-seeking. + pub fn for_each_live_posting_ti(&self, term_info: &TermInfo, mut f: impl FnMut(usize, &[u32])) { + if self.prx_name.is_empty() { + 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, term_info, |d, pos| { if !self.deletes.is_deleted(d) { f(d, pos); } @@ -83,7 +109,7 @@ impl ZslSegment { /// (field names ascending, terms ascending within each field), without /// materializing a `Vec` of all terms like `all_terms` does. Used by the /// bounded-memory k-way streaming merge. - pub fn term_cursor(&self) -> TermCursor<'_> { + pub fn term_cursor(&self) -> TermCursor { self.dict.cursor() } @@ -135,10 +161,15 @@ impl ZslSegment { let name_ending = |ext: &str| cfs.names().into_iter().find(|n| n.ends_with(ext)); let fnm = name_ending(".fnm").ok_or_else(|| std::io::Error::other("no .fnm"))?; let tis = name_ending(".tis").ok_or_else(|| std::io::Error::other("no .tis"))?; + let tii_name = name_ending(".tii").ok_or_else(|| std::io::Error::other("no .tii"))?; let fields = read_field_infos(cfs.sub(&fnm).unwrap())?; let field_names: Vec = fields.iter().map(|f| f.name.clone()).collect(); - let dict = TermDict::read(cfs.sub(&tis).unwrap(), &field_names)?; + let dict = TermDict::open( + cfs.sub(&tis).unwrap(), + cfs.sub(&tii_name).unwrap(), + &field_names, + )?; let fdx_name = name_ending(".fdx").ok_or_else(|| std::io::Error::other("no .fdx"))?; let num_docs_total = cfs.sub(&fdx_name).unwrap().len() / 8; @@ -207,7 +238,7 @@ impl IndexReader for ZslSegment { fn postings_for(&self, field: &str, term: &str) -> Vec<(usize, u32)> { match self.dict.info(field, term) { // degrade: a corrupt .frq yields no postings rather than a panic across FFI - Some(ti) => read_freqs(self.cfs.sub(&self.frq_name).unwrap(), ti) + Some(ti) => read_freqs(self.cfs.sub(&self.frq_name).unwrap(), &ti) .unwrap_or_default() .into_iter() .filter(|(d, _)| !self.deletes.is_deleted(*d)) @@ -251,7 +282,7 @@ impl IndexReader for ZslSegment { Some(ti) => read_positions( self.cfs.sub(&self.frq_name).unwrap(), self.cfs.sub(&self.prx_name).unwrap(), - ti, + &ti, doc_id, ) .unwrap_or_default(), @@ -266,7 +297,7 @@ impl IndexReader for ZslSegment { Some(ti) if !self.prx_name.is_empty() => read_all_positions( self.cfs.sub(&self.frq_name).unwrap(), self.cfs.sub(&self.prx_name).unwrap(), - ti, + &ti, ) .unwrap_or_default() .into_iter() @@ -439,7 +470,7 @@ mod tests { .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 = 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), diff --git a/sdsearch-core/src/zsl/terms.rs b/sdsearch-core/src/zsl/terms.rs index acb3377..7e27e5d 100644 --- a/sdsearch-core/src/zsl/terms.rs +++ b/sdsearch-core/src/zsl/terms.rs @@ -13,8 +13,13 @@ pub struct TermInfo { /// with parallel arrays of offsets and infos. Cuts RAM sharply (less allocator /// overhead) while keeping O(log N) binary search. ZSL writes terms grouped by field /// and sorted by text, so the buffer stays sorted. +/// +/// `#[cfg(test)]`: the eager full-parse dictionary is kept ONLY as the +/// differential oracle for the lazy production `TermDict` (see module tests); +/// production never compiles it. +#[cfg(test)] #[derive(Default)] -struct FieldTerms { +struct EagerFieldTerms { /// term texts concatenated, in ascending order. text: Vec, /// n+1 offsets: term i is `text[offsets[i]..offsets[i+1]]`. @@ -23,7 +28,8 @@ struct FieldTerms { infos: Vec, } -impl FieldTerms { +#[cfg(test)] +impl EagerFieldTerms { fn len(&self) -> usize { self.infos.len() } @@ -37,15 +43,21 @@ impl FieldTerms { /// unlike `iter_terms`'s owned `String::from_utf8_lossy`. fn term_str(&self, i: usize) -> &str { std::str::from_utf8(self.term(i)) - .expect("FieldTerms invariant violated: term bytes are valid UTF-8") + .expect("EagerFieldTerms invariant violated: term bytes are valid UTF-8") } } -pub struct TermDict { - by_field: std::collections::HashMap, +/// Eager full-parse term dictionary, kept ONLY as a `#[cfg(test)]` differential +/// oracle: it walks the ENTIRE `.tis` up front into per-field compact buffers. +/// Production uses the lazy `.tii`-backed `TermDict` below; the module tests +/// assert the two agree on every query over the fixtures and a synthetic corpus. +#[cfg(test)] +pub struct EagerTermDict { + by_field: std::collections::HashMap, } -impl TermDict { +#[cfg(test)] +impl EagerTermDict { /// Decodes the `.tis` (format 2.1+, marker 0xFFFFFFFD). /// /// Each entry shares a prefix with the previous term ONLY over the @@ -55,7 +67,7 @@ impl TermDict { /// distinguishing field) reproduces exactly what `DictionaryLoader::load` does. /// The freq/prox pointers are deltas accumulated across the WHOLE file, /// not per field. - pub fn read(tis: &[u8], field_names: &[String]) -> std::io::Result { + pub fn read(tis: &[u8], field_names: &[String]) -> std::io::Result { let mut pos = 0usize; let _marker = read_i32_be(tis, &mut pos)?; let term_count = read_u64_be(tis, &mut pos)?; @@ -63,7 +75,7 @@ impl TermDict { let _skip_interval = read_i32_be(tis, &mut pos)?; let _max_skip_levels = read_i32_be(tis, &mut pos)?; - let mut by_field: std::collections::HashMap = + let mut by_field: std::collections::HashMap = std::collections::HashMap::new(); let mut prev_text = String::new(); let mut freq_ptr: u64 = 0; @@ -96,11 +108,11 @@ impl TermDict { for ft in by_field.values_mut() { ft.offsets.push(ft.text.len() as u32); } - Ok(TermDict { by_field }) + Ok(EagerTermDict { by_field }) } /// Looks up (field, term) by binary search over the field's compact buffer. - pub fn info(&self, field: &str, term: &str) -> Option<&TermInfo> { + pub fn info(&self, field: &str, term: &str) -> Option { let ft = self.by_field.get(field)?; let target = term.as_bytes(); let (mut lo, mut hi) = (0usize, ft.len()); @@ -109,7 +121,7 @@ impl TermDict { match ft.term(mid).cmp(target) { std::cmp::Ordering::Less => lo = mid + 1, std::cmp::Ordering::Greater => hi = mid, - std::cmp::Ordering::Equal => return Some(&ft.infos[mid]), + std::cmp::Ordering::Equal => return Some(ft.infos[mid].clone()), } } None @@ -151,8 +163,8 @@ impl TermDict { /// (`fieldName · \0 · text`, i.e. field names ascending, terms ascending /// within each field). Unlike `iter_terms`, never materializes a `Vec` of /// all terms up front — the k-way streaming merge walks this instead. - pub fn cursor(&self) -> TermCursor<'_> { - TermCursor::new(self) + pub fn cursor(&self) -> EagerTermCursor<'_> { + EagerTermCursor::new(self) } /// Enumerates ALL terms as `(field, text)`. Order not guaranteed (grouped by @@ -172,25 +184,371 @@ impl TermDict { } } -/// Lazy cursor over all `(field, term)` pairs of a `TermDict`, in ZSL +/// One sampled entry of the sparse `.tii` index: a full term (text + `TermInfo`) +/// plus the `.tis` byte offset from which sequential scanning must resume to +/// reach it (see `TermDict::open`'s module-level accumulation note). +pub struct IndexEntry { + pub field: String, + pub text: String, + pub info: TermInfo, + pub tis_offset: usize, +} + +/// Lazily-queryable term dictionary: holds the raw `.tis` bytes uninterpreted +/// and only the SPARSE `.tii` index parsed up front (a small fraction of the +/// terms), as a single GLOBAL list (not grouped by field). Unlike the eager +/// `EagerTermDict`, opening this does not walk every `.tis` entry — `info` seeks to +/// the nearest `IndexEntry` and scans forward from there. +/// +/// `.tis`/`.tii` are physically ordered by the composite key `(field_name +/// ascending, text ascending)` (the eager `EagerTermCursor` / writer guarantee +/// this), so `index` is already sorted by that key and can be binary-searched +/// with `partition_point`. `index[0]` is a synthetic anchor `("", "")` — see +/// `open` — so every real target has a well-defined predecessor entry. +pub struct TermDict { + tis: Vec, + index: Vec, + field_names: Vec, +} + +impl TermDict { + /// Decodes the `.tii` sparse index (mirrors `EagerTermDict::read`'s header handling, + /// plus the synthetic first entry ZSL always writes — see `zsl/writer/terms.rs`) + /// and stashes a copy of `.tis` for later on-demand decoding by `info`. + /// + /// `.tii` layout: 24-byte header (same shape as `.tis`) + one synthetic entry + /// (VInt prefix=0, empty String suffix, RAW Int32 field=0xFFFFFFFF, literal byte + /// 0x0F, VInt docFreq=0, VInt freqDelta=0, VInt proxDelta=0, VInt IndexDelta=24 — + /// the `.tis` offset of the first real term, right after its 24-byte header) + + /// N real entries, each like a `.tis` entry but with an extra trailing + /// VInt IndexDelta. + /// + /// CRITICAL: per `zsl/writer/terms.rs::dump_tis_entry`, each entry's IndexDelta + /// is captured from `index_position = self.tis_len` AFTER the sampled term's + /// `.tis` entry was written — so an accumulated `tis_offset` points to the + /// `.tis` byte position IMMEDIATELY AFTER the indexed term (the start of + /// whatever term follows it), not the indexed term's own start. `info` relies on + /// this: after an exact hit on an index entry it can only resume decoding from + /// `tis_offset`, seeded with that entry's own text/pointers as predecessor state. + /// `INDEX_INTERVAL` also counts GLOBALLY across all fields, so a field's first + /// term may have no preceding `.tii` sample of its own — the synthetic anchor + /// (`tis_offset = 24`, the first real `.tis` term's start) covers that case. + pub fn open(tis: &[u8], tii: &[u8], field_names: &[String]) -> std::io::Result { + let mut pos = 0usize; + // header (same 24 bytes as .tis) + let _marker = read_i32_be(tii, &mut pos)?; + let _index_count = read_u64_be(tii, &mut pos)?; + let _index_interval = read_i32_be(tii, &mut pos)?; + let _skip_interval = read_i32_be(tii, &mut pos)?; + let _max_skip_levels = read_i32_be(tii, &mut pos)?; + + // synthetic first entry: VInt prefix, String suffix, RAW Int32 field (0xFFFFFFFF), + // literal byte 0x0F, VInt docFreq, VInt freqDelta, VInt proxDelta, VInt IndexDelta. + let _pfx = read_vint(tii, &mut pos)?; + let _suf = read_modified_utf8(tii, &mut pos)?; + let _raw_field = read_i32_be(tii, &mut pos)?; // raw Int, not VInt + // consume the literal 0x0F marker byte: + if pos >= tii.len() { + return Err(std::io::Error::other("tii: truncated synthetic entry")); + } + pos += 1; + let _df = read_vint(tii, &mut pos)?; + let _fd = read_vint(tii, &mut pos)?; + let _pd = read_vint(tii, &mut pos)?; + let _synthetic_index_delta = read_vint(tii, &mut pos)? as usize; // = 24 + + // synthetic anchor: ("", "") sorts before every real (field, text), and its + // tis_offset (24) is the .tis byte position of the first real term, right + // after the 24-byte header. + let mut index = vec![IndexEntry { + field: String::new(), + text: String::new(), + info: TermInfo { + doc_freq: 0, + freq_pointer: 0, + prox_pointer: 0, + }, + tis_offset: 24, + }]; + let mut prev_text = String::new(); + let (mut freq_ptr, mut prox_ptr) = (0u64, 0u64); + let mut running = 24usize; // synthetic IndexDelta already consumed above == 24 + while pos < tii.len() { + let shared = read_vint(tii, &mut pos)? as usize; + let suffix = read_modified_utf8(tii, &mut pos)?; + let field_num = read_vint(tii, &mut pos)? as usize; + let doc_freq = read_vint(tii, &mut pos)? as u32; + freq_ptr = freq_ptr.wrapping_add(read_vint(tii, &mut pos)?); + prox_ptr = prox_ptr.wrapping_add(read_vint(tii, &mut pos)?); + let index_delta = read_vint(tii, &mut pos)? as usize; + let prefix: String = prev_text.chars().take(shared).collect(); + let text = format!("{prefix}{suffix}"); + let field = field_names.get(field_num).cloned().unwrap_or_default(); + running = running.wrapping_add(index_delta); // ADD FIRST … + index.push(IndexEntry { + // … THEN assign tis_offset = running + field, + text: text.clone(), + info: TermInfo { + doc_freq, + freq_pointer: freq_ptr, + prox_pointer: prox_ptr, + }, + tis_offset: running, + }); + prev_text = text; + } + Ok(TermDict { + tis: tis.to_vec(), + index, + field_names: field_names.to_vec(), + }) + } + + /// Test-only: number of parsed `.tii` index entries (including the synthetic + /// `("","")` anchor). Lets tests assert a synthesized corpus actually produced + /// more than the trivial single-anchor index, i.e. exercised the sparse seek + /// path in `info` rather than degenerating to a full `.tis` scan from offset 24. + #[cfg(test)] + pub fn index_len(&self) -> usize { + self.index.len() + } + + /// Looks up `(field, term)` by seeking to the nearest `.tii` index entry + /// at or before it (composite key `(field, text)`) and, unless that's an + /// exact hit, forward-decoding `.tis` from there until the target is found, + /// passed, or the file ends. + pub fn info(&self, field: &str, term: &str) -> Option { + let key = (field, term); + // largest index entry with (field, text) <= key. index[0] is the synthetic ("","") anchor, + // so partition_point is always >= 1. + let gt = self + .index + .partition_point(|e| (e.field.as_str(), e.text.as_str()) <= key); + let anchor = &self.index[gt - 1]; + if !anchor.text.is_empty() && anchor.field == field && anchor.text == term { + return Some(anchor.info.clone()); // exact hit on an index term (no .tis scan) + } + // scan forward from the anchor's tis_offset (already PAST the anchor's own term), seeded + // with the anchor's text/pointers as the predecessor state. + let mut pos = anchor.tis_offset; + let mut prev = anchor.text.clone(); + let (mut fp, mut pp) = (anchor.info.freq_pointer, anchor.info.prox_pointer); + while pos < self.tis.len() { + let (f, t, ti) = decode_entry( + &self.tis, + &mut pos, + &prev, + &mut fp, + &mut pp, + &self.field_names, + ) + .ok()?; + match (f.as_str(), t.as_str()).cmp(&key) { + std::cmp::Ordering::Equal => return Some(ti), + std::cmp::Ordering::Greater => return None, // passed the key in canonical order ⇒ absent + std::cmp::Ordering::Less => prev = t, + } + } + None + } + + /// Terms of `field` that start with `prefix`, matching the eager + /// `EagerTermDict::terms_with_prefix` exactly. Finds the `.tii` anchor at or before + /// `(field, prefix)` in canonical `(field_name, text)` order, seeks its + /// `tis_offset`, and forward-decodes `.tis` from there — collecting matches of + /// `field` until the field changes or the text passes the prefix range. + /// + /// The anchor may sit in an EARLIER field than `field` (e.g. when `field`'s own + /// first term precedes any `.tii` sample taken from it): entries with `f < field` + /// are skipped, `f == field` entries are collected/filtered by prefix, and the + /// scan stops as soon as `f > field` (canonical order guarantees nothing further + /// can belong to `field`). + pub fn terms_with_prefix(&self, field: &str, prefix: &str) -> Vec { + let key = (field, prefix); + // index[0] is the synthetic ("","") anchor ⇒ partition_point is always >= 1. + let gt = self + .index + .partition_point(|e| (e.field.as_str(), e.text.as_str()) <= key); + let anchor = &self.index[gt - 1]; + let mut out = Vec::new(); + // the anchor itself is a real term that may already match (e.g. prefix == an index term). + if !anchor.text.is_empty() && anchor.field == field && anchor.text.starts_with(prefix) { + out.push(anchor.text.clone()); + } + let mut pos = anchor.tis_offset; + let mut prev = anchor.text.clone(); + let (mut fp, mut pp) = (anchor.info.freq_pointer, anchor.info.prox_pointer); + while pos < self.tis.len() { + let Ok((f, t, _)) = decode_entry( + &self.tis, + &mut pos, + &prev, + &mut fp, + &mut pp, + &self.field_names, + ) else { + break; + }; + // canonical order is (field_name, text); stop once we pass this field. + if f.as_str() > field { + break; + } + if f == field { + if t.starts_with(prefix) { + out.push(t.clone()); + } else if t.as_str() > prefix { + break; // past the range in-field + } + } + prev = t; + } + out + } + + /// Sequentially decodes the WHOLE `.tis` from just past its 24-byte header. + /// `.tis` is physically written in canonical `(field_name asc, text asc)` + /// order (see `EagerTermCursor`'s doc comment and `zsl/writer/terms.rs`), so a + /// plain top-to-bottom decode already yields that order — no sorting or + /// per-field grouping needed, unlike the eager `EagerTermDict::iter_terms` + /// (which enumerates a `HashMap` and so has no ordering guarantee of its + /// own). Used by both `iter_terms` and `cursor`. + fn decode_all(&self) -> Vec<(String, String, TermInfo)> { + let mut pos = 24usize; // past the header + let mut prev = String::new(); + let (mut fp, mut pp) = (0u64, 0u64); + let mut out = Vec::new(); + while pos < self.tis.len() { + match decode_entry( + &self.tis, + &mut pos, + &prev, + &mut fp, + &mut pp, + &self.field_names, + ) { + Ok((f, t, ti)) => { + prev = t.clone(); + out.push((f, t, ti)); + } + Err(_) => break, + } + } + out + } + + /// Enumerates ALL terms as `(field, text)`, in canonical `.tis` order (see + /// `decode_all`). Used by the merge to walk each source segment's terms + /// and copy their postings via `positions_all(field, text)`. + pub fn iter_terms(&self) -> Vec<(String, String)> { + self.decode_all() + .into_iter() + .map(|(f, t, _)| (f, t)) + .collect() + } + + /// Cursor over every `(field, term)` pair in ZSL canonical order, + /// matching the eager `EagerTermCursor`'s semantics exactly (the merge relies + /// on this order to interleave sources correctly). Unlike `EagerTermCursor`, + /// this OWNS its decoded `Vec` instead of borrowing the dict — it's built + /// from raw `.tis` bytes rather than an already-grouped `HashMap`, and the + /// merge only ever needs `peek`/`advance`, so no lifetime is required. + pub fn cursor(&self) -> TermCursor { + TermCursor { + terms: self.decode_all(), + i: 0, + } + } +} + +/// Owned cursor over a `TermDict`'s terms in canonical order (see +/// `TermDict::cursor`). Pre-decoded at construction time since the merge +/// reads every entry anyway, so there's no benefit to decoding lazily on +/// `advance`. Each entry keeps the term's decoded `TermInfo` alongside its +/// `(field, text)` so the streaming merge can read that term's postings +/// WITHOUT a second `TermDict::info` seek — see `peek_info`. +pub struct TermCursor { + terms: Vec<(String, String, TermInfo)>, + i: usize, +} + +impl TermCursor { + /// current `(field, term)` pair, or `None` once every pair is exhausted. + pub fn peek(&self) -> Option<(&str, &str)> { + self.terms + .get(self.i) + .map(|(f, t, _)| (f.as_str(), t.as_str())) + } + + /// `TermInfo` of the current term (the one `peek` returns), or `None` once + /// exhausted. The streaming merge captures this at the moment it pops a + /// term from its k-way heap — BEFORE `advance` moves the cursor on — so it + /// can read the term's postings directly instead of re-seeking the dict. + pub fn peek_info(&self) -> Option<&TermInfo> { + self.terms.get(self.i).map(|(_, _, ti)| ti) + } + + /// moves to the next pair in canonical order. No-op once exhausted. + pub fn advance(&mut self) { + if self.i < self.terms.len() { + self.i += 1; + } + } +} + +/// Decodes a single `.tis` entry starting at `*pos`, advancing it past the entry. +/// Mirrors `EagerTermDict::read`'s per-entry decoding (shared prefix taken over the +/// previous TEXT only, never across field boundaries — see that function's doc +/// comment) but for exactly one entry, so `TermDict::info` can resume a scan +/// from any `.tii`-indexed offset instead of decoding the whole `.tis`. +fn decode_entry( + tis: &[u8], + pos: &mut usize, + prev_text: &str, + freq_ptr: &mut u64, + prox_ptr: &mut u64, + field_names: &[String], +) -> std::io::Result<(String, String, TermInfo)> { + let shared = read_vint(tis, pos)? as usize; + let suffix = read_modified_utf8(tis, pos)?; + let field_num = read_vint(tis, pos)? as usize; + let doc_freq = read_vint(tis, pos)? as u32; + *freq_ptr = freq_ptr.wrapping_add(read_vint(tis, pos)?); + *prox_ptr = prox_ptr.wrapping_add(read_vint(tis, pos)?); + let prefix: String = prev_text.chars().take(shared).collect(); + let text = format!("{prefix}{suffix}"); + let field = field_names.get(field_num).cloned().unwrap_or_default(); + Ok(( + field, + text, + TermInfo { + doc_freq, + freq_pointer: *freq_ptr, + prox_pointer: *prox_ptr, + }, + )) +} + +/// Cursor over all `(field, term)` pairs of an `EagerTermDict`, in ZSL /// canonical order (field names sorted ascending, terms ascending within each /// field — equivalent to sorting `(fieldName, text)` tuples since `\0` is the /// minimum byte and never appears inside a field name or term). Backed -/// directly by each field's already-sorted `FieldTerms` buffer: no `Vec` of -/// all terms is built up front, so memory stays bounded regardless of how -/// many terms the segment holds. -pub struct TermCursor<'a> { - dict: &'a TermDict, +/// directly by each field's already-sorted `EagerFieldTerms` buffer. Kept as a +/// `#[cfg(test)]` oracle for the production lazy `TermCursor`. +#[cfg(test)] +pub struct EagerTermCursor<'a> { + dict: &'a EagerTermDict, fields: Vec<&'a str>, field_idx: usize, term_idx: usize, } -impl<'a> TermCursor<'a> { - fn new(dict: &'a TermDict) -> TermCursor<'a> { +#[cfg(test)] +impl<'a> EagerTermCursor<'a> { + fn new(dict: &'a EagerTermDict) -> EagerTermCursor<'a> { let mut fields: Vec<&str> = dict.by_field.keys().map(String::as_str).collect(); fields.sort_unstable(); - TermCursor { + EagerTermCursor { dict, fields, field_idx: 0, @@ -233,7 +591,7 @@ mod tests { use crate::zsl::fields::read_field_infos; use std::path::PathBuf; - fn dict() -> TermDict { + fn dict() -> EagerTermDict { let dir = PathBuf::from(concat!( env!("CARGO_MANIFEST_DIR"), "/tests/fixtures/zsl_index" @@ -259,7 +617,7 @@ mod tests { .into_iter() .map(|f| f.name) .collect(); - TermDict::read(cf.sub(&tis).unwrap(), &names).unwrap() + EagerTermDict::read(cf.sub(&tis).unwrap(), &names).unwrap() } #[test] @@ -268,7 +626,7 @@ mod tests { let mut buf = vec![0xFF, 0xFF, 0xFF, 0xFD]; buf.extend_from_slice(&5u64.to_be_bytes()); buf.extend_from_slice(&[0, 0, 0, 1, 0, 0, 0, 1, 0, 0, 0, 0]); - assert!(TermDict::read(&buf, &[]).is_err()); + assert!(EagerTermDict::read(&buf, &[]).is_err()); } #[test] @@ -306,4 +664,251 @@ mod tests { // total count = sum of terms per field (non-empty) assert!(!got.is_empty()); } + + fn fixture_dict_bytes() -> (Vec, Vec, Vec) { + use crate::zsl::cfs::CompoundFile; + use crate::zsl::fields::read_field_infos; + let dir = std::path::PathBuf::from(concat!( + env!("CARGO_MANIFEST_DIR"), + "/tests/fixtures/zsl_index" + )); + let path = std::fs::read_dir(&dir) + .unwrap() + .filter_map(|e| e.ok().map(|e| e.path())) + .find(|p| p.extension().map(|x| x == "cfs").unwrap_or(false)) + .unwrap(); + let cf = CompoundFile::open(&path).unwrap(); + let find = |ext: &str| cf.names().into_iter().find(|n| n.ends_with(ext)).unwrap(); + let names: Vec = read_field_infos(cf.sub(&find(".fnm")).unwrap()) + .unwrap() + .into_iter() + .map(|f| f.name) + .collect(); + ( + cf.sub(&find(".tis")).unwrap().to_vec(), + cf.sub(&find(".tii")).unwrap().to_vec(), + names, + ) + } + + fn fixture_dict_bytes_multiseg() -> (Vec, Vec, Vec) { + use crate::zsl::cfs::CompoundFile; + use crate::zsl::fields::read_field_infos; + let dir = std::path::PathBuf::from(concat!( + env!("CARGO_MANIFEST_DIR"), + "/tests/fixtures/zsl_index_multiseg" + )); + let path = std::fs::read_dir(&dir) + .unwrap() + .filter_map(|e| e.ok().map(|e| e.path())) + .find(|p| p.extension().map(|x| x == "cfs").unwrap_or(false)) + .unwrap(); + let cf = CompoundFile::open(&path).unwrap(); + let find = |ext: &str| cf.names().into_iter().find(|n| n.ends_with(ext)).unwrap(); + let names: Vec = read_field_infos(cf.sub(&find(".fnm")).unwrap()) + .unwrap() + .into_iter() + .map(|f| f.name) + .collect(); + ( + cf.sub(&find(".tis")).unwrap().to_vec(), + cf.sub(&find(".tii")).unwrap().to_vec(), + names, + ) + } + + #[test] + fn lazy_info_matches_eager_for_every_term_and_absentees() { + for (tis, tii, names) in [fixture_dict_bytes(), fixture_dict_bytes_multiseg()] { + let eager = EagerTermDict::read(&tis, &names).unwrap(); + let lazy = TermDict::open(&tis, &tii, &names).unwrap(); + for (field, text) in eager.iter_terms() { + assert_eq!( + lazy.info(&field, &text), + eager.info(&field, &text), + "mismatch at {field}:{text}" + ); + } + assert_eq!(lazy.info("title", "definitelymissing"), None); + assert_eq!(lazy.info("no_such_field", "x"), None); + } + } + + /// Synthesizes a >128-term, 2-field term dictionary directly via the writer's + /// batch `write_term_dict` (mirroring `tii_samples_every_index_interval_terms` + /// / `multi_field_sample_terms` in `zsl/writer/terms.rs`, scaled up to 560 terms — + /// the same order of magnitude the review used to demonstrate ~429/560 mismatches + /// when the accumulation order is reverted). Guarantees `index_len() > 1`, i.e. + /// the `.tii` actually got multiple real samples rather than just the synthetic + /// `("","")` anchor — the committed fixtures (`zsl_index`, `zsl_index_multiseg`) + /// both have far fewer than `INDEX_INTERVAL` (128) terms and can't exercise the + /// sparse seek path on their own. + /// + /// field 0 ("body"): 400 terms, zero-padded so lexicographic order == insertion + /// order, sharing the "shared" prefix (exercises prefix-sharing across many + /// consecutive .tis/.tii entries). Every 10th term gets a second, far-away doc + /// (exercises multi-doc freq/prox deltas). + /// field 1 ("title"): 160 more terms, zero-padded ascending, disjoint doc-id + /// range from field 0 (prefix sharing must not leak across fields — see + /// `dump_entry`'s doc comment). + fn large_dict_bytes() -> (Vec, Vec, Vec) { + use crate::zsl::writer::invert::TermPostings; + use crate::zsl::writer::terms::write_term_dict; + + let mut terms: Vec = Vec::new(); + + for i in 0..400usize { + let text = format!("shared{i:04}"); + let docs = if i % 10 == 0 { + vec![(i, vec![0u32, 3]), (i + 10_000, vec![1u32])] + } else { + vec![(i, vec![0u32])] + }; + terms.push(TermPostings { + field_num: 0, + text, + docs, + }); + } + for i in 0..160usize { + let text = format!("word{i:04}"); + let docs = if i % 7 == 0 { + vec![(20_000 + i, vec![0u32, 2]), (20_000 + i + 500, vec![1u32])] + } else { + vec![(20_000 + i, vec![0u32])] + }; + terms.push(TermPostings { + field_num: 1, + text, + docs, + }); + } + assert_eq!( + terms.len(), + 560, + "corpus must comfortably exceed INDEX_INTERVAL (128)" + ); + + let field_names = vec!["body".to_string(), "title".to_string()]; + let dict_files = write_term_dict(&terms); + (dict_files.tis, dict_files.tii, field_names) + } + + #[test] + fn lazy_seek_path_exercised_by_synthetic_multi_sample_tii() { + let (tis, tii, field_names) = large_dict_bytes(); + + let eager = EagerTermDict::read(&tis, &field_names).unwrap(); + let lazy = TermDict::open(&tis, &tii, &field_names).unwrap(); + + // sanity: the synthesized corpus must actually produce multiple .tii samples, + // i.e. more than just the synthetic ("","") anchor — otherwise this test would + // silently degenerate back into the same full-scan-only coverage as the + // fixture-based differential test above. + assert!( + lazy.index_len() > 1, + "corpus too small to exercise the sparse .tii seek path: index_len={}", + lazy.index_len() + ); + + for (field, text) in eager.iter_terms() { + assert_eq!( + lazy.info(&field, &text), + eager.info(&field, &text), + "mismatch at {field}:{text}" + ); + } + + assert_eq!(lazy.info("body", "definitelymissing"), None); + assert_eq!(lazy.info("no_such_field", "x"), None); + } + + /// for each `(field, prefix)` asserts `lazy.terms_with_prefix` == `eager.terms_with_prefix` + /// (both sorted, since neither API guarantees an order). + fn assert_prefix_parity( + tis: &[u8], + tii: &[u8], + names: &[String], + fields: &[&str], + prefixes: &[&str], + ) { + let eager = EagerTermDict::read(tis, names).unwrap(); + let lazy = TermDict::open(tis, tii, names).unwrap(); + for &field in fields { + for &pfx in prefixes { + let mut a = lazy.terms_with_prefix(field, pfx); + a.sort(); + let mut b = eager.terms_with_prefix(field, pfx); + b.sort(); + assert_eq!(a, b, "prefix mismatch {field}:{pfx:?}"); + } + } + } + + #[test] + fn lazy_prefix_matches_eager() { + // small fixtures (fields title/body/id_key) — only the synthetic anchor, no + // real .tii samples. + for (tis, tii, names) in [fixture_dict_bytes(), fixture_dict_bytes_multiseg()] { + assert_prefix_parity( + &tis, + &tii, + &names, + &["title", "body", "id_key", "no_such_field"], + &["", "a", "ne", "work", "zzz"], + ); + } + + // large synthetic corpus (real, multi-sample .tii): "body" has shared0000..shared0399, + // "title" has word0000..word0159 (INDEX_INTERVAL=128 globally, body written first, so + // .tii samples land inside "body" at ~terms 128/256/384 and spill into "title" after + // term 400). Prefixes chosen to hit all three terms_with_prefix branches: + // "" / "a" precede the first real sample (anchor may be an earlier field or the + // synthetic ("","") one); "shared" matches broadly within one field; "shared01" + // (shared0100..shared0199) straddles the ~128 sample boundary; "shared05" is past + // the last real "body" term (max shared0399) so it matches nothing; "zzz" matches + // nothing in any field. + let (tis, tii, names) = large_dict_bytes(); + assert_prefix_parity( + &tis, + &tii, + &names, + &names.iter().map(String::as_str).collect::>(), + &["", "a", "shared", "shared01", "shared05", "zzz"], + ); + } + + #[test] + fn lazy_iter_and_cursor_match_eager() { + for (tis, tii, names) in [ + fixture_dict_bytes(), + fixture_dict_bytes_multiseg(), + large_dict_bytes(), + ] { + let eager = EagerTermDict::read(&tis, &names).unwrap(); + let lazy = TermDict::open(&tis, &tii, &names).unwrap(); + + // iter_terms: same multiset AND, critically for the merge, cursor gives the same ORDER. + let mut a = lazy.iter_terms(); + a.sort(); + let mut b = eager.iter_terms(); + b.sort(); + assert_eq!(a, b); + + // cursor sequence identical, element by element (canonical .tis order — the merge relies on it) + let (mut ec, mut lc) = (eager.cursor(), lazy.cursor()); + loop { + let (e, l) = ( + ec.peek().map(|(f, t)| (f.to_string(), t.to_string())), + lc.peek().map(|(f, t)| (f.to_string(), t.to_string())), + ); + assert_eq!(e, l); + if e.is_none() { + break; + } + ec.advance(); + lc.advance(); + } + } + } } diff --git a/sdsearch-core/src/zsl/writer/durability.rs b/sdsearch-core/src/zsl/writer/durability.rs index 2ae21ba..c07f0cf 100644 --- a/sdsearch-core/src/zsl/writer/durability.rs +++ b/sdsearch-core/src/zsl/writer/durability.rs @@ -22,10 +22,27 @@ pub(crate) fn write_atomic(path: &Path, bytes: &[u8]) -> std::io::Result<()> { Ok(()) } -/// Writes `bytes` to `.tmp`, does `fsync` (`File::sync_all`) and renames to `path` -/// (atomic replace). fsync + rename guarantees that, after a crash, `path` holds either the old -/// content or the new content COMPLETE and durable — never a partial write. `std`-only +/// Best-effort `fsync` of a directory, to make durable the directory ENTRIES (names) of files +/// created or renamed inside it. On POSIX, fsync of a file makes its content durable but does +/// NOT make its directory entry or a `rename` durable — that needs an fsync of the containing +/// directory. On Unix, opening a directory as a `File` and calling `sync_all` flushes the +/// dirents; on Windows, `File::open` on a directory fails, so this is a harmless no-op (NTFS +/// relies on metadata journaling for the same guarantee). Portable WITHOUT `cfg`: errors are +/// ignored, since this is a best-effort durability improvement, never load-bearing for +/// correctness on this path. +pub(crate) fn sync_dir(dir: &Path) { + let _ = std::fs::File::open(dir).and_then(|d| d.sync_all()); +} + +/// Writes `bytes` to `.tmp`, does `fsync` (`File::sync_all`), renames to `path` (atomic +/// replace), then best-effort fsyncs the parent directory via [`sync_dir`]. The file fsync makes +/// the CONTENT durable; the directory fsync makes the renamed directory ENTRY durable, so after +/// a crash `path` holds either the old content or the new content COMPLETE. `std`-only /// (Windows-safe: `File::sync_all` == `FlushFileBuffers`, `rename` == `MoveFileEx`). +/// +/// Durability caveat: the parent-directory fsync is a real fsync on Unix but a no-op on Windows +/// (there is no portable std API to fsync a directory), where the cross-crash durability of the +/// directory entry relies on NTFS metadata journaling — so the guarantee is platform-dependent. /// A single writer holds the write-lock, so the `.tmp` suffix never collides. pub(crate) fn write_durable(path: &Path, bytes: &[u8]) -> std::io::Result<()> { let file_name = path @@ -38,6 +55,9 @@ pub(crate) fn write_durable(path: &Path, bytes: &[u8]) -> std::io::Result<()> { f.sync_all()?; // fsync: the bytes are on disk before the rename } std::fs::rename(&tmp, path)?; + if let Some(parent) = path.parent() { + sync_dir(parent); // make the renamed directory entry durable (best-effort) + } Ok(()) } @@ -86,4 +106,29 @@ mod tests { assert_eq!(std::fs::read(&target).unwrap(), b"payload-2-longer"); std::fs::remove_dir_all(&dir).ok(); } + + #[test] + fn sync_dir_on_existing_directory_does_not_panic_and_keeps_contents() { + let n = COUNTER.fetch_add(1, Ordering::Relaxed); + let dir = std::env::temp_dir().join(format!("sdsearch_syncd_{}_{}", std::process::id(), n)); + std::fs::create_dir_all(&dir).unwrap(); + std::fs::write(dir.join("f"), b"contents").unwrap(); + + sync_dir(&dir); // best-effort: real dir-fsync on Unix, no-op on Windows; must not panic + + assert_eq!(std::fs::read(dir.join("f")).unwrap(), b"contents"); + std::fs::remove_dir_all(&dir).ok(); + } + + #[test] + fn sync_dir_on_nonexistent_path_is_a_noop() { + let n = COUNTER.fetch_add(1, Ordering::Relaxed); + let missing = std::env::temp_dir().join(format!( + "sdsearch_syncd_missing_{}_{}", + std::process::id(), + n + )); + // must not panic even though the path does not exist (errors are ignored). + sync_dir(&missing); + } } diff --git a/sdsearch-core/src/zsl/writer/index_writer.rs b/sdsearch-core/src/zsl/writer/index_writer.rs index 4c1b670..4843812 100644 --- a/sdsearch-core/src/zsl/writer/index_writer.rs +++ b/sdsearch-core/src/zsl/writer/index_writer.rs @@ -209,25 +209,31 @@ impl IndexWriter { let new_gen = segments::write_optimized_generation(&self.dir, &gen, &merged_name, doc_count as u32)?; - // 6) orphan cleanup POST-flip (best-effort): .cfs/.del/.sti of the old segments. - // Never before the flip (crash-safety invariant: the merged segment is referenced only - // after the atomic segments.gen flip). The merged _ is not in `infos`. + // 6) old-segment cleanup POST-flip (best-effort), DEFERRED one generation. + // Never before the flip (crash-safety: the merged segment is referenced only after the + // atomic segments.gen flip). We do NOT delete the old .cfs/.sti/.del now: the prune in + // step 5 keeps the previous generation's manifest as a grace window for lock-free + // readers, and that manifest still references these files — deleting them now would + // leave such a reader with dangling references (and on Windows the unlink of a mapped + // .cfs fails silently, leaking it). Instead we hand this round's superseded files to + // `process_pending_deletions`, which deletes the PREVIOUS round's files (whose manifest + // the step-5 prune just removed) and records these for the next flip. The merged _ + // is not in `infos`, so it is never listed. + let mut superseded: Vec = Vec::new(); for info in &infos { - let _ = std::fs::remove_file(self.dir.join(format!("{}.cfs", info.name))); - let _ = std::fs::remove_file(self.dir.join(format!("{}.sti", info.name))); - let del_name = match info.del_gen { - -1 => None, - 0 => Some(format!("{}.del", info.name)), - g => Some(format!( + superseded.push(format!("{}.cfs", info.name)); + superseded.push(format!("{}.sti", info.name)); + match info.del_gen { + -1 => {} + 0 => superseded.push(format!("{}.del", info.name)), + g => superseded.push(format!( "{}_{}.del", info.name, crate::zsl::segments::to_base36(g as u64) )), - }; - if let Some(dn) = del_name { - let _ = std::fs::remove_file(self.dir.join(dn)); } } + segments::process_pending_deletions(&self.dir, &superseded); Ok(CommitReport { generation: new_gen, @@ -309,6 +315,11 @@ impl IndexWriter { self.next_name_counter, )?; + // A flip happened (this point is unreachable on the empty-commit no-op above): reclaim any + // deferred deletions from a previous optimize whose manifest the flip's prune just removed. + // This add-only commit supersedes nothing, so it records no new files. + segments::process_pending_deletions(&self.dir, &[]); + Ok(CommitReport { generation, segments, @@ -360,6 +371,45 @@ mod tests { } } + #[test] + fn optimize_keeps_previous_generation_segment_files_as_a_grace_window() { + let dir = temp_kb_full(); + // build a multi-segment index: KB base (_2) + several flushed segments in one commit. + 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(doc_mark(i)).unwrap(); + } + w.commit().unwrap(); + + let pre = read_segment_infos(&dir).unwrap(); + assert!( + pre.len() >= 2, + "need a multi-segment index for optimize to merge" + ); + let old_names: Vec = pre.iter().map(|s| s.name.clone()).collect(); + + // optimize collapses everything into one segment and flips the generation. + let w2 = IndexWriter::open(&dir, WriterOpts::default()).unwrap(); + w2.optimize().unwrap(); + + // The prune keeps the PREVIOUS generation's manifest as a grace window for lock-free + // readers that read segments.gen an instant before the flip. Those readers open the + // previous manifest, which still references the old segments — so their .cfs must ALSO + // still exist, or the grace window points at deleted data. Deletion is deferred one + // generation for exactly this reason. + for name in &old_names { + assert!( + dir.join(format!("{name}.cfs")).exists(), + "old segment {name}.cfs must survive one generation as a reader grace window" + ); + } + std::fs::remove_dir_all(&dir).ok(); + } + #[test] fn committed_name_counter_equals_writer_high_water_mark() { let dir = temp_kb_full(); @@ -597,10 +647,29 @@ mod tests { assert_eq!(idx.num_docs(), live); assert_eq!(idx.doc_freq("title", "zqxmark"), 3); // 4 - 1 deleted - // orphan cleanup: the old .cfs were deleted; the merged one exists - assert!(!dir.join("_2.cfs").exists()); + // Deferred cleanup: the old segments' .cfs are NOT deleted immediately — they survive one + // generation as a grace window for lock-free readers (the kept previous manifest still + // references them). The merged segment exists. + assert!( + dir.join("_2.cfs").exists(), + "old .cfs kept one generation as a grace window" + ); + assert!(dir.join("_3.cfs").exists()); + assert!(dir.join("_4.cfs").exists()); + assert!(dir.join(format!("{}.cfs", infos[0].name)).exists()); + + // The NEXT flip (another commit) prunes the previous manifest and reclaims the deferred + // files, so the old .cfs are finally gone — deferral does not leak. + let mut w4 = IndexWriter::open(&dir, WriterOpts::default()).unwrap(); + w4.add_document(doc_mark(99)).unwrap(); + w4.commit().unwrap(); + assert!( + !dir.join("_2.cfs").exists(), + "old .cfs reclaimed on the next flip" + ); assert!(!dir.join("_3.cfs").exists()); assert!(!dir.join("_4.cfs").exists()); + // the merged segment (referenced by the current manifest) is untouched. assert!(dir.join(format!("{}.cfs", infos[0].name)).exists()); std::fs::remove_dir_all(&dir).ok(); diff --git a/sdsearch-core/src/zsl/writer/merge.rs b/sdsearch-core/src/zsl/writer/merge.rs index 93d499b..767b999 100644 --- a/sdsearch-core/src/zsl/writer/merge.rs +++ b/sdsearch-core/src/zsl/writer/merge.rs @@ -16,6 +16,7 @@ use super::invert::{FieldMeta, StoredField, TermPostings}; use super::{assemble_cfs, fnm, norms, stored, terms}; use crate::index::IndexReader; use crate::zsl::segment::ZslSegment; +use crate::zsl::terms::TermInfo; use std::cmp::Reverse; use std::collections::{BTreeMap, BinaryHeap, HashMap}; use std::fs::File; @@ -207,6 +208,10 @@ pub fn merge_segments_streaming( let fdt_tmp = index_dir.join(format!("{merged_name}.fdt.tmp")); let frq_tmp = index_dir.join(format!("{merged_name}.frq.tmp")); let prx_tmp = index_dir.join(format!("{merged_name}.prx.tmp")); + // The final `.cfs` is assembled into this staging file and only renamed onto the real + // `{merged_name}.cfs` once fully written + fsynced, so a mid-write I/O error can never leave + // a partial FINAL `.cfs` on disk — only this `.cfs.tmp`, which is cleaned up below. + let cfs_tmp = index_dir.join(format!("{merged_name}.cfs.tmp")); let result = merge_streaming_inner( index_dir, @@ -215,12 +220,15 @@ pub fn merge_segments_streaming( &fdt_tmp, &frq_tmp, &prx_tmp, + &cfs_tmp, ); - // Clean up temp files on BOTH the success and error paths (best-effort). + // Clean up staging files on BOTH the success and error paths (best-effort). `.cfs.tmp` is + // included so a failed assembly (or a failed rename) never leaks a partial staging file. let _ = std::fs::remove_file(&fdt_tmp); let _ = std::fs::remove_file(&frq_tmp); let _ = std::fs::remove_file(&prx_tmp); + let _ = std::fs::remove_file(&cfs_tmp); result } @@ -232,6 +240,7 @@ fn merge_streaming_inner( fdt_tmp: &Path, frq_tmp: &Path, prx_tmp: &Path, + cfs_tmp: &Path, ) -> io::Result { let segs: Vec = segments .iter() @@ -370,7 +379,18 @@ fn merge_streaming_inner( while let Some(Reverse((field, term, first_si))) = heap.pop() { // gather EVERY segment currently positioned at this exact (field, term). Advancing a // cursor and re-pushing its next (strictly greater) term keeps the heap in sync. - let mut contributing: Vec = vec![first_si]; + // + // GOTCHA: each contributing cursor's `TermInfo` is captured HERE, at pop time, BEFORE + // `advance()` moves it past this term. A segment has at most one key on the heap at a + // time, pushed by a `peek()` that did NOT advance the cursor — so when that key is + // popped, `cursors[si]` is still positioned exactly at `(field, term)` and `peek_info()` + // is this term's info. After `advance()` it would be the NEXT term's, which is why we + // must not read it at posting-copy time below. + let first_ti = cursors[first_si] + .peek_info() + .expect("a cursor whose key is on the heap is positioned at that term") + .clone(); + let mut contributing: Vec<(usize, TermInfo)> = vec![(first_si, first_ti)]; cursors[first_si].advance(); if let Some((f, t)) = cursors[first_si].peek() { heap.push(Reverse((f.to_string(), t.to_string(), first_si))); @@ -381,7 +401,11 @@ fn merge_streaming_inner( break; } let Reverse((_, _, si)) = heap.pop().unwrap(); - contributing.push(si); + let ti = cursors[si] + .peek_info() + .expect("a cursor whose key is on the heap is positioned at that term") + .clone(); + contributing.push((si, ti)); cursors[si].advance(); if let Some((f, t)) = cursors[si].peek() { heap.push(Reverse((f.to_string(), t.to_string(), si))); @@ -390,20 +414,21 @@ 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). `for_each_live_posting` + // (dense renumbering assigns lower ids to earlier segments). `for_each_live_posting_ti` // 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(); + contributing.sort_unstable_by_key(|(si, _)| *si); let mf = field_index[&field]; dict_writer.begin_term(mf, &term)?; - for &si in &contributing { + for (si, ti) in &contributing { + let si = *si; let map = &doc_maps[si]; - // `for_each_live_posting`'s callback is `FnMut` and can't propagate `?`, so capture + // `for_each_live_posting_ti`'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| { + segs[si].for_each_live_posting_ti(ti, |local, positions| { if posting_err.is_some() { return; } @@ -439,12 +464,19 @@ fn merge_streaming_inner( (".prx", CfsSource::Path(prx_tmp)), ]; - // Write the merged .cfs durably: stream it into the file, then fsync BEFORE returning so the - // generation flip in optimize() only references a durable .cfs. + // Write the merged .cfs durably and ATOMICALLY: stream it into `{merged_name}.cfs.tmp`, fsync + // the content, then rename it onto the real `{merged_name}.cfs`. The staging + rename means a + // failure partway through `write_cfs_streaming` (disk full, I/O error) can only leave the + // `.cfs.tmp` (cleaned up by the caller), never a truncated FINAL `.cfs`. The directory fsync + // makes the rename durable so the generation flip in optimize() only references a durable .cfs. let cfs_path = index_dir.join(format!("{merged_name}.cfs")); - let mut cfs_file = File::create(&cfs_path)?; - write_cfs_streaming(&mut cfs_file, merged_name, &files)?; - cfs_file.sync_all()?; + { + let mut cfs_file = File::create(cfs_tmp)?; + write_cfs_streaming(&mut cfs_file, merged_name, &files)?; + cfs_file.sync_all()?; + } + std::fs::rename(cfs_tmp, &cfs_path)?; + super::durability::sync_dir(index_dir); Ok(doc_count) } @@ -626,13 +658,15 @@ mod tests { let dict = terms::write_term_dict(&[]); // directory built by hand (NOT `assemble_cfs`): only the extensions REQUIRED by - // `ZslSegment::open_from` (.fdx .fdt .fnm .tis .frq), without `.nrm`/`.tii` (optional) and - // without `.prx` (deliberately omitted). + // `ZslSegment::open_from` (.fdx .fdt .fnm .tis .tii .frq), without `.nrm` (optional) and + // without `.prx` (deliberately omitted). `.tii` is required by `open_from` (the lazy dict + // seeks through it) and the writer always emits one, so it is included here. let files: Vec<(&str, &[u8])> = vec![ (".fdx", &fdx), (".fdt", &fdt), (".fnm", &fnm_bytes), (".tis", &dict.tis), + (".tii", &dict.tii), (".frq", &dict.frq), ]; let cfs_bytes = crate::zsl::writer::cfs::write_cfs("_x", &files); @@ -667,6 +701,8 @@ mod tests { assert!(!dir.join("_m.fdt.tmp").exists()); assert!(!dir.join("_m.frq.tmp").exists()); assert!(!dir.join("_m.prx.tmp").exists()); + // the .cfs staging file was renamed onto the final .cfs, not left behind. + assert!(!dir.join("_m.cfs.tmp").exists()); } #[test] @@ -693,6 +729,35 @@ mod tests { std::fs::remove_dir_all(&dir).ok(); } + #[test] + fn streaming_merge_cleans_staging_and_leaves_no_partial_cfs_when_final_rename_fails() { + // Regression lock for the atomic-.cfs write: a pre-existing directory at the final + // `_m.cfs` path makes the rename of the fully-written `.cfs.tmp` staging file fail. The + // merge must return Err, must NOT leak any staging file (`.cfs.tmp` / `.fdt/.frq/.prx.tmp`), + // and must never leave a truncated FINAL `.cfs` (the blocking directory stays untouched). + let dir = temp_kb_full(); + let infos = read_segment_infos(&dir).unwrap(); + let refs: Vec<(String, i64)> = infos.iter().map(|s| (s.name.clone(), s.del_gen)).collect(); + + std::fs::create_dir(dir.join("_m.cfs")).unwrap(); // blocks the final rename target + + let result = merge_segments_streaming(&dir, "_m", &refs); + assert!( + result.is_err(), + "merge must fail when the final rename is blocked" + ); + + // no staging file leaked (cleaned on the error path)... + assert!(!dir.join("_m.cfs.tmp").exists()); + assert!(!dir.join("_m.fdt.tmp").exists()); + assert!(!dir.join("_m.frq.tmp").exists()); + assert!(!dir.join("_m.prx.tmp").exists()); + // ...and the final path is still the blocking directory, never a truncated .cfs file. + assert!(dir.join("_m.cfs").is_dir()); + + std::fs::remove_dir_all(&dir).ok(); + } + #[test] fn streaming_matches_batch_deletes_across_two_base_segments() { let dir = temp_kb_full(); @@ -830,6 +895,7 @@ mod tests { (".fdt", &fdt), (".fnm", &fnm_bytes), (".tis", &dict.tis), + (".tii", &dict.tii), (".frq", &dict.frq), ]; let cfs_bytes = crate::zsl::writer::cfs::write_cfs("_x", &files); diff --git a/sdsearch-core/src/zsl/writer/segments.rs b/sdsearch-core/src/zsl/writer/segments.rs index 93e1158..0f37ac5 100644 --- a/sdsearch-core/src/zsl/writer/segments.rs +++ b/sdsearch-core/src/zsl/writer/segments.rs @@ -106,6 +106,65 @@ fn prune_old_generations(index_dir: &Path, keep_from_gen: u64) { } } +/// Name of the deferred-deletion sidecar. Not a `segments_*` manifest, `.cfs`, or `.del`, so it +/// is invisible to `prune_old_generations`, to readers, and to segment enumeration. +const PENDING_DELETIONS_FILE: &str = "pending_deletions"; + +/// Best-effort, one-generation-DEFERRED reclamation of segment data files (`_.cfs/.sti/ +/// .del`) superseded by a merge. `optimize()` cannot delete the old segments immediately: the +/// flip keeps the previous generation's manifest as a grace window for lock-free readers, and +/// that manifest still references the old files — deleting them now would leave such a reader +/// with dangling references (and on Windows the unlink of a file still mapped by a reader fails +/// silently, leaking it). So each flip instead does two things. First, it deletes the files +/// recorded by the PREVIOUS flip — whose referencing manifest this flip's `prune_old_generations` +/// has just removed, so they are now safe — retrying/carrying forward any that still fail (e.g. a +/// Windows reader still holds the mapping). Second, it records `new_superseded` (this flip's +/// just-superseded files, still referenced by the kept previous manifest) for the NEXT flip. +/// +/// Because every flip advances the generation by exactly one, a file recorded here is always +/// referenced solely by manifests the next prune removes. The ONLY failure mode of a bug here is +/// a disk leak — it NEVER deletes a file that is still referenced (it only ever lists already- +/// superseded files). Must be called AFTER the durable flip + prune. Best-effort: all errors +/// (missing sidecar, unreadable, unwritable) are ignored. +/// +/// `new_superseded` are file NAMES relative to `index_dir` (e.g. `"_2.cfs"`), not paths. +pub(crate) fn process_pending_deletions(index_dir: &Path, new_superseded: &[String]) { + let sidecar = index_dir.join(PENDING_DELETIONS_FILE); + + // 1) reclaim the previous round's files; carry forward any that could not be deleted yet. + let mut carry: Vec = Vec::new(); + if let Ok(contents) = std::fs::read_to_string(&sidecar) { + for name in contents.lines() { + let name = name.trim(); + if name.is_empty() { + continue; + } + let path = index_dir.join(name); + match std::fs::remove_file(&path) { + Ok(()) => {} + // already gone (deleted earlier / never existed): treat as reclaimed. + Err(_) if !path.exists() => {} + // still present but unlink failed (e.g. Windows: mapped by a reader) → retry next flip. + Err(_) => carry.push(name.to_string()), + } + } + } + + // 2) next round's list = carried-forward failures + this flip's newly-superseded files. + let mut next = carry; + for f in new_superseded { + if !next.contains(f) { + next.push(f.clone()); + } + } + + if next.is_empty() { + let _ = std::fs::remove_file(&sidecar); + } else { + let _ = std::fs::write(&sidecar, next.join("\n")); + } +} + fn read_gen_number(index_dir: &Path) -> std::io::Result { let data = std::fs::read(index_dir.join("segments.gen"))?; let mut pos = 0usize; @@ -320,6 +379,11 @@ pub fn write_optimized_generation( write_i64_be(&mut g, new_gen as i64); write_i64_be(&mut g, new_gen as i64); super::durability::write_atomic(&index_dir.join("segments.gen"), &g)?; + // Make the segments.gen rename durable at the directory level (best-effort; real fsync on + // Unix, no-op on Windows). write_durable already fsync'd the segments_{N+1} dirent; the + // segments.gen flip goes through write_atomic (no fsync), so sync the directory here so the + // optimize() durability promise also covers the generation pointer's rename. + super::durability::sync_dir(index_dir); // best-effort prune of stale generation manifests, only now that the flip is durable. prune_old_generations(index_dir, new_gen.saturating_sub(1)); @@ -350,6 +414,50 @@ mod tests { dir } + fn temp_empty_dir(tag: &str) -> std::path::PathBuf { + let n = COUNTER.fetch_add(1, Ordering::Relaxed); + let dir = + std::env::temp_dir().join(format!("sdsearch_pd_{tag}_{}_{}", std::process::id(), n)); + std::fs::create_dir_all(&dir).unwrap(); + dir + } + + #[test] + fn process_pending_deletions_deletes_listed_and_records_new() { + let dir = temp_empty_dir("rec"); + // previous round's file, listed in the sidecar, is now safe to delete. + std::fs::write(dir.join("_old.cfs"), b"x").unwrap(); + std::fs::write(dir.join(PENDING_DELETIONS_FILE), "_old.cfs").unwrap(); + // this round's just-superseded file, still referenced by the kept manifest. + std::fs::write(dir.join("_new.cfs"), b"y").unwrap(); + + process_pending_deletions(&dir, &["_new.cfs".to_string()]); + + // the previously-listed file is reclaimed... + assert!(!dir.join("_old.cfs").exists()); + // ...the new superseded file is recorded for the NEXT flip, NOT deleted now... + assert!(dir.join("_new.cfs").exists()); + let sidecar = std::fs::read_to_string(dir.join(PENDING_DELETIONS_FILE)).unwrap(); + assert_eq!(sidecar.trim(), "_new.cfs"); + + std::fs::remove_dir_all(&dir).ok(); + } + + #[test] + fn process_pending_deletions_removes_sidecar_when_nothing_pending() { + let dir = temp_empty_dir("empty"); + std::fs::write(dir.join("_old.cfs"), b"x").unwrap(); + std::fs::write(dir.join(PENDING_DELETIONS_FILE), "_old.cfs").unwrap(); + + process_pending_deletions(&dir, &[]); // no new superseded files + + assert!(!dir.join("_old.cfs").exists()); + // nothing left to defer → the sidecar file itself is removed (no empty-file litter). + assert!(!dir.join(PENDING_DELETIONS_FILE).exists()); + + std::fs::remove_dir_all(&dir).ok(); + } + #[test] fn appends_new_segment_over_real_kb_generation() { let dir = temp_kb_gen(); diff --git a/sdsearch-core/src/zsl/writer/terms.rs b/sdsearch-core/src/zsl/writer/terms.rs index d031576..42c44e1 100644 --- a/sdsearch-core/src/zsl/writer/terms.rs +++ b/sdsearch-core/src/zsl/writer/terms.rs @@ -410,7 +410,7 @@ mod tests { use super::*; use crate::zsl::bytes::read_u64_be; use crate::zsl::postings::{read_all_positions, read_freqs}; - use crate::zsl::terms::TermDict; + use crate::zsl::terms::EagerTermDict; use crate::zsl::writer::invert::invert; use crate::zsl::writer::{WriterDoc, WriterField, WriterOpts}; @@ -437,17 +437,17 @@ mod tests { let inv = invert(&docs, &WriterOpts::default()); let f = write_term_dict(&inv.terms); let names = field_names(&inv); - let dict = TermDict::read(&f.tis, &names).unwrap(); + let dict = EagerTermDict::read(&f.tis, &names).unwrap(); assert_eq!(dict.info("title", "workflow").unwrap().doc_freq, 2); assert_eq!(dict.info("title", "done").unwrap().doc_freq, 1); let body_new = dict.info("body", "new").unwrap(); - assert_eq!(read_freqs(&f.frq, body_new).unwrap(), vec![(0, 2)]); + assert_eq!(read_freqs(&f.frq, &body_new).unwrap(), vec![(0, 2)]); let title_wf = dict.info("title", "workflow").unwrap(); assert_eq!( - read_all_positions(&f.frq, &f.prx, title_wf).unwrap(), + read_all_positions(&f.frq, &f.prx, &title_wf).unwrap(), vec![(0, vec![2]), (1, vec![1])] ); } @@ -502,7 +502,7 @@ mod tests { assert_eq!(read_u64_be(&f.tii, &mut pos).unwrap(), 3); // and the .tis is still readable - let dict = TermDict::read(&f.tis, &field_names(&inv)).unwrap(); + let dict = EagerTermDict::read(&f.tis, &field_names(&inv)).unwrap(); assert_eq!(dict.info("t", "w0000").unwrap().doc_freq, 1); assert_eq!(dict.info("t", "w0299").unwrap().doc_freq, 1); } diff --git a/sdsearch-core/tests/zsl_merge_memory_bound.rs b/sdsearch-core/tests/zsl_merge_memory_bound.rs new file mode 100644 index 0000000..d52ff32 --- /dev/null +++ b/sdsearch-core/tests/zsl_merge_memory_bound.rs @@ -0,0 +1,219 @@ +//! CI guard for the bounded-memory streaming k-way segment merge. +//! +//! The whole point of [`merge::merge_segments_streaming`] is that its peak HEAP does NOT grow +//! with a near-stopword term's `doc_freq` (nor with total text volume): it streams postings +//! per-doc through temp files, whereas the batch oracle [`merge::merge_segments`] materializes +//! the ENTIRE merged segment in RAM (a `term_map` holding every term's postings/positions for +//! every doc, plus the full stored column, plus the full `.cfs` byte buffer). +//! +//! Until now the only thing measuring this property was the manual `examples/optimize_bench.rs` +//! binary; there was no automated test. This file is that test. +//! +//! Design: it drives BOTH public merge entry points over the SAME multi-segment input and +//! compares their peak heap using a tracking global allocator. The assertion is DIFFERENTIAL +//! (streaming peak << batch peak over identical input), which gives it teeth by construction and +//! keeps it robust across debug/release (release strips debug_asserts and shifts alloc patterns, +//! but the batch path still builds the whole segment in RAM while streaming does not). +//! +//! NOTE: a `#[global_allocator]` in an integration-test file only affects THIS test binary — each +//! `tests/*.rs` file is compiled as its own crate — so it does not pollute other tests. + +use sdsearch_core::zsl::segments::read_segment_infos; +use sdsearch_core::zsl::writer::merge; +use sdsearch_core::zsl::writer::{FieldKind, IndexWriter, WriterDoc, WriterField, WriterOpts}; +use std::alloc::{GlobalAlloc, Layout, System}; +use std::path::PathBuf; +use std::sync::atomic::{AtomicUsize, Ordering}; + +// ---- tracking allocator: measures live + peak heap bytes (same pattern as optimize_bench) ---- +static LIVE: AtomicUsize = AtomicUsize::new(0); +static PEAK: AtomicUsize = AtomicUsize::new(0); + +struct TrackingAlloc; + +unsafe impl GlobalAlloc for TrackingAlloc { + unsafe fn alloc(&self, layout: Layout) -> *mut u8 { + let p = System.alloc(layout); + if !p.is_null() { + let live = LIVE.fetch_add(layout.size(), Ordering::Relaxed) + layout.size(); + PEAK.fetch_max(live, Ordering::Relaxed); + } + p + } + unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) { + System.dealloc(ptr, layout); + LIVE.fetch_sub(layout.size(), Ordering::Relaxed); + } + unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, new_size: usize) -> *mut u8 { + let p = System.realloc(ptr, layout, new_size); + if !p.is_null() { + let old = layout.size(); + if new_size >= old { + let live = LIVE.fetch_add(new_size - old, Ordering::Relaxed) + (new_size - old); + PEAK.fetch_max(live, Ordering::Relaxed); + } else { + LIVE.fetch_sub(old - new_size, Ordering::Relaxed); + } + } + p + } +} + +#[global_allocator] +static GLOBAL: TrackingAlloc = TrackingAlloc; + +/// Anchors PEAK to the current live bytes, so the next PEAK read reflects only the heap +/// high-water mark ABOVE this steady state (i.e. the merge's own allocations). +fn reset_peak() { + let live = LIVE.load(Ordering::Relaxed); + PEAK.store(live, Ordering::Relaxed); +} + +/// Copies the committed KB fixture to a fresh temp dir (skips lock files and `.sti`), returns it. +/// `IndexWriter::open` requires an existing base index (it reads the generation), so we start +/// from the real KB fixture — exactly as `examples/optimize_bench.rs` and the merge unit tests do. +fn copy_kb_base() -> PathBuf { + let src = PathBuf::from(concat!( + env!("CARGO_MANIFEST_DIR"), + "/tests/fixtures/zsl_index_kb" + )); + let dst = std::env::temp_dir().join(format!( + "sdsearch_merge_mem_{}_{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0) + )); + if dst.is_dir() { + std::fs::remove_dir_all(&dst).ok(); + } + std::fs::create_dir_all(&dst).expect("create temp dir"); + for entry in std::fs::read_dir(&src).expect("read KB fixture") { + let p = entry.unwrap().path(); + let name = p.file_name().unwrap().to_string_lossy().to_string(); + if name.contains("lock") || name.ends_with(".sti") { + continue; + } + std::fs::copy(&p, dst.join(&name)).expect("copy fixture file"); + } + dst +} + +/// N docs that all share ONE "stopword" token (so its `doc_freq == N` in both `title` and +/// `body` — the near-stopword shape the batch path would hold fully resident) PLUS a per-doc +/// unique token, so the postings genuinely differ doc to doc. +fn gen_hot_docs(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!("stopword ticket{i} {}", POOL[i % np]); + let filler: String = (0..30) + .map(|j| POOL[(i * 7 + j * 5) % np]) + .collect::>() + .join(" "); + let body = format!("stopword {filler} 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() +} + +#[test] +fn streaming_merge_peak_heap_is_far_below_batch() { + // ---- build a multi-segment index with a large-doc_freq term ---- + // N deliberately in the hundreds+ so "materialize the whole term / whole segment" (batch) + // vs "stream one posting" (streaming) is a clear heap difference, not allocator noise. + const N: usize = 1500; + let dir = copy_kb_base(); + + { + // small buffer => ceil(N / cap) flushed segments => a genuine k-way merge. + let opts = WriterOpts { + max_buffered_docs: 50, + ..WriterOpts::default() + }; + let mut w = IndexWriter::open(&dir, opts).expect("open (build) failed"); + for d in gen_hot_docs(N) { + w.add_document(d).expect("add_document failed"); + } + w.commit().expect("commit failed"); + } + + let infos = read_segment_infos(&dir).expect("read segment infos"); + assert!( + infos.len() >= 3, + "expected a multi-segment index for a meaningful k-way merge, got {}", + infos.len() + ); + let refs: Vec<(String, i64)> = infos.iter().map(|s| (s.name.clone(), s.del_gen)).collect(); + + // ---- measure STREAMING peak heap over the input ---- + // (source segments are mmap'd, not heap-allocated, so they don't count toward PEAK.) + reset_peak(); + let streaming_docs = + merge::merge_segments_streaming(&dir, "_stream", &refs).expect("streaming merge failed"); + let streaming_peak = PEAK.load(Ordering::Relaxed); + // the streaming merge writes its output to disk; drop it so it can't skew the batch run. + std::fs::remove_file(dir.join("_stream.cfs")).ok(); + + // ---- measure BATCH peak heap over the SAME input ---- + reset_peak(); + let batch = merge::merge_segments(&dir, "_batch", &refs).expect("batch merge failed"); + let batch_peak = PEAK.load(Ordering::Relaxed); + let batch_docs = batch.doc_count; + // hold cfs_bytes only until the peak is read, then release. + drop(batch); + + // sanity: both paths merged the same corpus (KB fixture's 20 docs + N added). + assert_eq!(streaming_docs, batch_docs, "doc counts must agree"); + assert_eq!(streaming_docs, N + 20, "unexpected merged doc count"); + + // ---- differential assertion ---- + // The batch path holds the entire merged segment resident (term_map with every posting, + // the full stored column, and the full .cfs byte buffer), so its peak scales with total + // content. The streaming path keeps only small per-field/index blocks resident. Measured + // (see report) the batch peak is an order of magnitude larger; we require a conservative 3x + // so the guard is robust across debug/release yet still fails loudly if the streaming merge + // ever regresses to materializing the term (which would make the two peaks converge). + eprintln!( + "streaming_peak = {} bytes ({} KB), batch_peak = {} bytes ({} KB), ratio = {:.2}x", + streaming_peak, + streaming_peak / 1024, + batch_peak, + batch_peak / 1024, + batch_peak as f64 / streaming_peak.max(1) as f64, + ); + assert!( + streaming_peak.saturating_mul(3) < batch_peak, + "streaming merge peak heap ({streaming_peak} B) is not far below the batch peak \ + ({batch_peak} B): the bounded-memory guarantee appears to have regressed", + ); + + std::fs::remove_dir_all(&dir).ok(); +} diff --git a/tools/bench_compare.php b/tools/bench_compare.php new file mode 100644 index 0000000..1afcdce --- /dev/null +++ b/tools/bench_compare.php @@ -0,0 +1,279 @@ + [iters] [index_dir] + * engine ∈ {sdsearch, zend} + * workload ∈ {build, rebuild, churn, search} + * N index size (docs added on top of the ~20-doc KB base) + * iters search only: sampled query iterations (default 50) + * index_dir build/churn/search: the persistent index dir (build writes it; churn/search read + * it). rebuild ignores it (uses its own temp dir). + * + * `zend` requires ZEND_LUCENE_PATH (see zsl_bootstrap.php); without it the tool no-ops (exit 0), + * which the orchestrator records as "skipped". `sdsearch` requires the extension to be loaded + * (`php -d extension=target/release/libsdsearch.so ...`). + * + * The corpus generator and the three query tokens MUST stay byte-identical to + * sdsearch-core/examples/bench_engine.rs, or the two engines compare different work. + */ + +$REPO = dirname(__DIR__); +$engine = $argv[1] ?? ''; +$workload = $argv[2] ?? ''; +$N = (int)($argv[3] ?? 1000); +$iters = (int)($argv[4] ?? 50); +$indexDir = $argv[5] ?? (sys_get_temp_dir() . "/sdsearch_bench_persist_{$engine}_{$N}"); + +if (!in_array($engine, ['sdsearch', 'zend'], true)) { + fwrite(STDERR, "engine must be sdsearch|zend\n"); + exit(2); +} +if (!in_array($workload, ['build', 'rebuild', 'churn', 'search'], true)) { + fwrite(STDERR, "workload must be build|rebuild|churn|search\n"); + exit(2); +} + +if ($engine === 'zend') { + // no-ops (exit 0) if ZEND_LUCENE_PATH is unset → orchestrator records a clean skip. + require __DIR__ . '/zsl_bootstrap.php'; +} else { + if (!class_exists('SdSearch\\Writer') || !class_exists('SdSearch\\Engine')) { + fwrite(STDERR, "sdsearch extension not loaded — run with: php -d extension=" + . "$REPO/target/release/libsdsearch.so ...\n"); + exit(2); + } +} + +// ---- planted-term corpus (byte-identical to examples/bench_engine.rs) ---- +const COMMON = 'widetoken'; // in EVERY doc → "many" (doc_freq == N) +const RARE = 'sparsetoken'; // in RARE_K docs → "few" (doc_freq == RARE_K) +const MISSING = 'absenttoken'; // never emitted → "none" (doc_freq == 0) +const RARE_K = 5; +const POOL = ['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']; + +function is_rare(int $i, int $n): bool { + $step = max(1, intdiv($n, RARE_K)); + return $i % $step === 0 && intdiv($i, $step) < RARE_K; +} + +/** one deterministic doc (title/body/id), matching gen_one() in bench_engine.rs. title/body + * draw ONLY from the fixed POOL (bounded vocabulary); the only unique-per-doc term is `id`. */ +function gen_one(int $i, int $n, string $prefix): array { + $np = count(POOL); + $title = 'ticket ' . POOL[$i % $np] . ' ' . POOL[($i * 3) % $np] . ' ' . POOL[($i * 7) % $np]; + $bodyWords = []; + for ($j = 0; $j < 40; $j++) { $bodyWords[] = POOL[($i * 7 + $j * 5) % $np]; } + $body = implode(' ', $bodyWords) . ' ' . COMMON; + if (is_rare($i, $n)) { $body .= ' ' . RARE; } + return ['title' => $title, 'body' => $body, 'id' => "$prefix-$i"]; +} + +/** process peak RSS (VmHWM) in KB; 0 if unavailable. */ +function rss_peak_kb(): int { + $s = @file_get_contents('/proc/self/status'); + if ($s !== false && preg_match('/^VmHWM:\s+(\d+)/m', $s, $m)) { return (int)$m[1]; } + return 0; +} + +$KB_BASE = "$REPO/sdsearch-core/tests/fixtures/zsl_index_kb"; + +/** copies the KB fixture (or any index dir) to $dst, skipping lock/.sti files. */ +function copy_index(string $src, string $dst): void { + if (!is_dir($src)) { fwrite(STDERR, "missing base: $src\n"); exit(2); } + if (is_dir($dst)) { array_map('unlink', glob("$dst/*") ?: []); } else { mkdir($dst, 0777, true); } + foreach (glob("$src/*") as $f) { + $b = basename($f); + if (str_contains($b, 'lock') || str_ends_with($b, '.sti')) { continue; } + copy($f, "$dst/$b"); + } +} +function cleanup_dir(string $d): void { + foreach (glob("$d/*") ?: [] as $f) { @unlink($f); } + @rmdir($d); +} + +// ---- per-engine add / delete / search primitives (generate docs on the fly) ---- + +/** adds docs numbered start..start+count into $dir; returns nothing. Streaming feed (no array). */ +function add_docs(string $engine, $writer, int $start, int $count, int $n, string $prefix): void { + for ($i = $start; $i < $start + $count; $i++) { + $d = gen_one($i, $n, $prefix); + if ($engine === 'zend') { + $doc = new Zend_Search_Lucene_Document(); + $doc->addField(Zend_Search_Lucene_Field::text('title', $d['title'])); + $doc->addField(Zend_Search_Lucene_Field::text('body', $d['body'])); + $doc->addField(Zend_Search_Lucene_Field::keyword('id', $d['id'])); + $writer->addDocument($doc); + } else { + $writer->add_document(json_encode(['fields' => [ + ['name' => 'title', 'value' => $d['title'], 'kind' => 'text'], + ['name' => 'body', 'value' => $d['body'], 'kind' => 'text'], + ['name' => 'id', 'value' => $d['id'], 'kind' => 'keyword'], + ]])); + } + } +} + +/** opens a writer over $dir (Zend index handle or SdSearch\Writer). */ +function open_writer(string $engine, string $dir) { + if ($engine === 'zend') { return Zend_Search_Lucene::open($dir); } + $w = new SdSearch\Writer(); + $w->open($dir); + return $w; +} + +/** Runs the free-text query $token at paging depth $limit (0 = unlimited) and returns the hit + * count. Uses the SAME fuzzy+wildcard+parser shape on both engines (mirrors query.rs + * text_subquery / perf_zsl.php buildText) and reopens the index per call, exactly as + * SdSearch\Engine::search does — so both engines measure the same per-request cost. + * NOTE: Zend's find() has no top-K limit (it always scores + returns all), so for zend the + * $limit only affects the returned count, not the work — top-20 and top-100 latencies coincide. */ +function query_count(string $engine, string $dir, string $token, int $limit): int { + if ($engine === 'sdsearch') { + $eng = new SdSearch\Engine(); + $json = $eng->search($dir, json_encode(['text' => $token, 'limit' => $limit])); + return count(json_decode($json, true) ?: []); + } + $F = fn($t, $s, $p, $f = null) => new Zend_Search_Lucene_Search_Query_Fuzzy(new Zend_Search_Lucene_Index_Term($t, $f), $s, $p); + $W = fn($p, $f = null) => new Zend_Search_Lucene_Search_Query_Wildcard(new Zend_Search_Lucene_Index_Term($p, $f)); + $b = new Zend_Search_Lucene_Search_Query_Boolean(); + $b->addSubquery($F($token, 0.5, 3), null); + $b->addSubquery($W($token . '*'), null); + $b->addSubquery(Zend_Search_Lucene_Search_QueryParser::parse($token), null); + $top = new Zend_Search_Lucene_Search_Query_Boolean(); + $top->addSubquery($b, true); + $idx = Zend_Search_Lucene::open($dir); + $hits = $idx->find($top); + return $limit > 0 ? min($limit, count($hits)) : count($hits); +} + +/** + * Commits the open writer INSIDE the timed region and returns the resulting LIVE doc count for + * sdsearch (read from document_count() — its contract is "live base + buffered − deletes" = the + * post-commit total — while the writer is still open, so no reopen is timed), or null for zend + * (whose count needs a reopen, done by the caller AFTER stopping the clock). Zend is never + * loaded in the sdsearch path, so the branch is unambiguous. + */ +function finish_writer(string $engine, $w): ?int { + if ($engine === 'sdsearch') { + $c = $w->document_count(); + $w->commit(); + return $c; + } + $w->commit(); + return null; +} + +/** Like finish_writer(), but OPTIMIZES to a single segment (production shape). Used by build + * (churn/search setup) and rebuild. SdSearch\Writer::optimize() commits + merges and consumes + * the writer; Zend commits then optimizes its still-open index handle. */ +function finish_writer_optimized(string $engine, $w): ?int { + if ($engine === 'sdsearch') { + $c = $w->document_count(); + $w->optimize(); + return $c; + } + $w->commit(); + $w->optimize(); + return null; +} + +function emit(array $row): void { echo json_encode($row) . "\n"; } + +// ---- workloads ---- + +switch ($workload) { + case 'build': { + // UNMEASURED setup: create a persistent N-doc index for churn/search to reuse. + copy_index($KB_BASE, $indexDir); + $w = open_writer($engine, $indexDir); + add_docs($engine, $w, 0, $N, $N, 'REC'); + $dc = finish_writer_optimized($engine, $w); // production shape: single segment + if ($dc === null) { $dc = Zend_Search_Lucene::open($indexDir)->numDocs(); } + emit(['engine' => $engine, 'workload' => 'build', 'n' => $N, + 'index_dir' => $indexDir, 'doc_count' => $dc]); + break; + } + + case 'rebuild': { + $dir = sys_get_temp_dir() . "/sdsearch_bench_rebuild_{$engine}_{$N}_" . getmypid(); + copy_index($KB_BASE, $dir); + $t0 = microtime(true); + $w = open_writer($engine, $dir); + add_docs($engine, $w, 0, $N, $N, 'REC'); + $dc = finish_writer_optimized($engine, $w); // production rebuild ends optimized + $ms = (microtime(true) - $t0) * 1000.0; + if ($dc === null) { $dc = Zend_Search_Lucene::open($dir)->numDocs(); } + emit(['engine' => $engine, 'workload' => 'rebuild', 'n' => $N, 'ms' => round($ms, 3), + 'php_peak_mb' => round(memory_get_peak_usage(true) / 1048576, 2), + 'rss_peak_mb' => round(rss_peak_kb() / 1024, 2), 'doc_count' => $dc]); + cleanup_dir($dir); + break; + } + + case 'churn': { + // requires a pre-built index at $indexDir; copy to scratch so re-runs are idempotent. + $scratch = sys_get_temp_dir() . "/sdsearch_bench_churn_{$engine}_{$N}_" . getmypid(); + copy_index($indexDir, $scratch); + $onePct = max(1, intdiv($N, 100)); + // $before only when a Zend reader is available (zend path); sdsearch path has no Zend. + $before = class_exists('Zend_Search_Lucene') ? Zend_Search_Lucene::open($scratch)->numDocs() : null; + $t0 = microtime(true); + $w = open_writer($engine, $scratch); + for ($gid = 0; $gid < $onePct; $gid++) { + if ($engine === 'zend') { $w->delete($gid); } else { $w->delete_document($gid); } + } + add_docs($engine, $w, 0, $onePct, $N, 'CHURN'); + $after = finish_writer($engine, $w); + $ms = (microtime(true) - $t0) * 1000.0; + if ($after === null) { $after = Zend_Search_Lucene::open($scratch)->numDocs(); } + emit(['engine' => $engine, 'workload' => 'churn', 'n' => $N, 'pct1' => $onePct, + 'ms' => round($ms, 3), + 'php_peak_mb' => round(memory_get_peak_usage(true) / 1048576, 2), + 'rss_peak_mb' => round(rss_peak_kb() / 1024, 2), + 'doc_count_before' => $before, 'doc_count' => $after]); + cleanup_dir($scratch); + break; + } + + case 'search': { + // requires a pre-built index at $indexDir (read-only). Times two realistic paging depths + // (top-20, top-100); reports the TRUE hit count per class (one unlimited call, unmeasured). + $classes = ['many' => COMMON, 'few' => RARE, 'none' => MISSING]; + $out = ['engine' => $engine, 'workload' => 'search', 'n' => $N, 'iters' => $iters]; + foreach ($classes as $label => $token) { + $entry = ['hits' => query_count($engine, $indexDir, $token, 0)]; // true freq (unlimited) + foreach ([20, 100] as $lim) { + query_count($engine, $indexDir, $token, $lim); // warm-up (not sampled) + $samples = []; + for ($k = 0; $k < $iters; $k++) { + $t0 = microtime(true); + query_count($engine, $indexDir, $token, $lim); + $samples[] = (microtime(true) - $t0) * 1000.0; + } + sort($samples); + $p = fn($q) => $samples[(int)round((count($samples) - 1) * $q)]; + $entry["top$lim"] = ['p50_ms' => round($p(0.50), 4), 'p95_ms' => round($p(0.95), 4)]; + } + $out[$label] = $entry; + } + $out['php_peak_mb'] = round(memory_get_peak_usage(true) / 1048576, 2); + $out['rss_peak_mb'] = round(rss_peak_kb() / 1024, 2); + emit($out); + break; + } +} diff --git a/tools/bench_report.php b/tools/bench_report.php new file mode 100644 index 0000000..28aac09 --- /dev/null +++ b/tools/bench_report.php @@ -0,0 +1,141 @@ + # prints Markdown to stdout + */ + +$path = $argv[1] ?? ''; +if ($path === '' || !is_file($path)) { + fwrite(STDERR, "usage: php tools/bench_report.php \n"); + exit(2); +} + +// data[workload][n][engine] = row (assoc); skips tracked separately. +$data = []; +$skips = []; +foreach (file($path, FILE_IGNORE_NEW_LINES | FILE_SKIP_EMPTY_LINES) as $line) { + $r = json_decode($line, true); + if (!is_array($r) || !isset($r['engine'], $r['workload'], $r['n'])) { continue; } + if (($r['status'] ?? '') === 'skipped') { + $skips[] = $r; + continue; + } + // native is emitted in two passes: 'time' (honest ms/rss) and 'heap' (accurate heap). Route + // the heap pass to a separate slot so it does not overwrite the time pass. + $key = $r['engine']; + if ($key === 'native' && ($r['pass'] ?? 'heap') === 'heap') { $key = 'native_heap'; } + $data[$r['workload']][(int)$r['n']][$key] = $r; +} + +/** collects the set of sizes seen for a workload, sorted ascending. */ +function sizes_of(array $byN): array { + $s = array_keys($byN); + sort($s, SORT_NUMERIC); + return $s; +} +function fmt($v, int $dec = 2): string { + return $v === null ? '—' : number_format((float)$v, $dec); +} +function speedup($zend, $ours): string { + if ($zend === null || $ours === null || (float)$ours == 0.0) { return '—'; } + return number_format((float)$zend / (float)$ours, 1) . '×'; +} + +$commit = trim(@shell_exec('git rev-parse --short HEAD 2>/dev/null') ?: '') ?: 'unknown'; +$cpu = 'unknown'; +if (is_readable('/proc/cpuinfo')) { + foreach (file('/proc/cpuinfo') as $l) { + if (str_starts_with($l, 'model name')) { $cpu = trim(explode(':', $l, 2)[1]); break; } + } +} + +echo "# sdsearch benchmarks\n\n"; +echo "Generated by `benches/run.sh` → `tools/bench_report.php`. Reproducible: deterministic corpus, " + . "fixed iterations, warm-up discarded. See [README.md](README.md) for the corpus and query definitions.\n\n"; +echo "| | |\n|---|---|\n"; +echo "| Date | " . date('Y-m-d H:i') . " |\n"; +echo "| Commit | `$commit` |\n"; +echo "| Host | " . php_uname('s') . ' ' . php_uname('r') . " |\n"; +echo "| CPU | $cpu |\n"; +echo "| PHP | " . PHP_VERSION . " |\n\n"; + +echo "**Engines.** `native` = the Rust core (examples/bench_engine) — reports the EXACT Rust heap. " + . "`sdsearch` = our PHP extension. `zend` = Zend_Search_Lucene (pure PHP).\n\n"; +echo "**Memory columns.** `heap` (native only) = peak Rust heap attributed to the op " + . "(`memory_get_peak_usage()` cannot see this). `rss` = process peak RSS (VmHWM), the " + . "cross-engine-comparable footprint. `php_peak` (PHP engines) omitted from tables because it " + . "understates sdsearch (Rust heap is invisible to it); it is in results.jsonl for reference.\n\n"; + +// ---- rebuild / churn: time + memory, one row per size ---- +foreach (['rebuild' => 'Rebuild (build the whole index from the KB base)', + 'churn' => 'Churn (delete 1% of docs + add 1% + commit)'] as $wl => $title) { + if (empty($data[$wl])) { continue; } + echo "## $wl — $title\n\n"; + echo "| docs | native ms | native heap MB | native rss MB | sdsearch ms | sdsearch rss MB | zend ms | zend rss MB | speedup vs zend | doc_count |\n"; + echo "|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|\n"; + foreach (sizes_of($data[$wl]) as $n) { + $row = $data[$wl][$n]; + $na = $row['native'] ?? null; // time pass: honest ms + rss + $naH = $row['native_heap'] ?? null; // heap pass: accurate heap + $sd = $row['sdsearch'] ?? null; $ze = $row['zend'] ?? null; + $dc = $na['doc_count'] ?? $naH['doc_count'] ?? $sd['doc_count'] ?? $ze['doc_count'] ?? null; + printf("| %s | %s | %s | %s | %s | %s | %s | %s | %s | %s |\n", + number_format($n), + fmt($na['ms'] ?? null, 2), + fmt(isset($naH['heap_peak_kb']) ? $naH['heap_peak_kb'] / 1024 : null, 1), + fmt(isset($na['rss_peak_kb']) ? $na['rss_peak_kb'] / 1024 : null, 1), + fmt($sd['ms'] ?? null, 2), + fmt($sd['rss_peak_mb'] ?? null, 1), + fmt($ze['ms'] ?? null, 2), + fmt($ze['rss_peak_mb'] ?? null, 1), + speedup($ze['ms'] ?? null, $sd['ms'] ?? null), + $dc === null ? '—' : number_format((int)$dc)); + } + echo "\n"; +} + +// ---- search: one table per paging depth (top-20, top-100); a row per (size, class) ---- +if (!empty($data['search'])) { + echo "## search — term latency by result class (many / few / none)\n\n"; + echo "p50 latency in ms, measured at two realistic paging depths. `hits` is the TRUE match " + . "count for the class (must be identical across engines — correctness gate). Each query " + . "reopens the index, as the engines do per request; at larger N the index-open cost " + . "dominates low-hit queries. Zend's `find()` has no top-K limit, so its top-20 and " + . "top-100 latencies coincide.\n\n"; + foreach (['top20' => 'top-20', 'top100' => 'top-100'] as $depth => $title) { + echo "### $title\n\n"; + echo "| docs | class | native p50 | sdsearch p50 | zend p50 | speedup vs zend | hits |\n"; + echo "|---:|---|---:|---:|---:|---:|---:|\n"; + foreach (sizes_of($data['search']) as $n) { + $row = $data['search'][$n]; + foreach (['many', 'few', 'none'] as $cls) { + $na = $row['native'][$cls][$depth] ?? null; $sd = $row['sdsearch'][$cls][$depth] ?? null; $ze = $row['zend'][$cls][$depth] ?? null; + $hits = $row['native'][$cls]['hits'] ?? $row['sdsearch'][$cls]['hits'] ?? $row['zend'][$cls]['hits'] ?? null; + printf("| %s | %s | %s | %s | %s | %s | %s |\n", + number_format($n), $cls, + fmt($na['p50_ms'] ?? null, 4), + fmt($sd['p50_ms'] ?? null, 4), + fmt($ze['p50_ms'] ?? null, 4), + speedup($ze['p50_ms'] ?? null, $sd['p50_ms'] ?? null), + $hits === null ? '—' : number_format((int)$hits)); + } + } + echo "\n"; + } +} + +// ---- skips ---- +if ($skips) { + echo "## Skipped runs\n\n"; + echo "| engine | workload | docs | reason |\n|---|---|---:|---|\n"; + foreach ($skips as $s) { + printf("| %s | %s | %s | %s |\n", $s['engine'], $s['workload'], number_format((int)$s['n']), $s['reason'] ?? ''); + } + echo "\n"; +}