Compressed Corpus Search

Corpus, Separators, and Baseline

For documents $D_0,\ldots,D_{k-1}$, index $T=D_0\#_0D_1\#_1\cdots D_{k-1}\#_{k-1}\$$. Every separator is unique and outside the query alphabet; the sentinel occurs once and is globally smallest. Maintain an origin array mapping content positions to (document_id, offset) and structural positions to no document.

The baseline searches each document directly or binary-searches an uncompressed suffix array. It provides an independent oracle for FM-index counts and locations. For banana, bandana, and ananas, the indexed sequence contains 19 content symbols, three separators, and one sentinel.

Statistic or structure	Sample value	Calculation	Logical payload	Purpose
Documents	3	`banana`, `bandana`, `ananas`	Application metadata	Identity and display
Document characters	19	$6+7+6$	19 symbols	Searchable content
Separators	3	One per document	3 symbols	Prevent cross-document matches
Sentinel	1	One global minimum	1 symbol	Terminate suffix order
Indexed length	23	$19+3+1$	23 symbols	BWT and suffix count
Full suffix array	23 positions	One position per suffix	92 bytes at 32 bits	Baseline order and location
Inverse suffix array	23 ranks	One rank per position	92 bytes at 32 bits	Position-to-row mapping
BWT	23 symbols	One preceding symbol per row	23 bytes with byte alphabet	Compressed-index foundation
Naive suffix materialization	276 symbol slots	$23+22+\cdots+1$	At least 276 units	Educational only

BWT and Rank Data Structure

Given suffix array SA, BWT row $r$ stores the symbol preceding SA[r]. Let $C[c]$ count indexed symbols smaller than $c$ and $\operatorname{rank}_c(i)$ count $c$ in BWT[0:i]. Backward search updates $[l,r)$ by $l'=C[c]+\operatorname{rank}_c(l)$ and $r'=C[c]+\operatorname{rank}_c(r)$.

The implementation uses rank checkpoints every $q$ positions and suffix-array samples at text positions divisible by $s$. Rank scans at most $q-1$ BWT symbols; locating applies LF until it reaches a sample, requiring at most $s-1$ steps under text-position sampling.

from collections import Counter
from dataclasses import dataclass


@dataclass(frozen=True, order=True)
class Location:
    document: int
    offset: int


class CorpusFMIndex:
    def __init__(self, documents: list[str], checkpoint_rate: int = 4, sample_rate: int = 4):
        if not documents or checkpoint_rate <= 0 or sample_rate <= 0:
            raise ValueError("documents and positive rates are required")
        self.documents = documents.copy()
        self.q, self.s, count = checkpoint_rate, sample_rate, len(documents)
        alphabet = sorted(set().union(*(set(document) for document in documents)))
        self.code = {c: count + 1 + i for i, c in enumerate(alphabet)}
        self.text: list[int] = []
        self.origin: list[Location | None] = []
        for document_id, document in enumerate(documents):
            for offset, character in enumerate(document):
                self.text.append(self.code[character])
                self.origin.append(Location(document_id, offset))
            self.text.append(document_id + 1)
            self.origin.append(None)
        self.text.append(0)
        self.origin.append(None)
        self.n = len(self.text)

        self.sa = sorted(range(self.n), key=lambda start: self.text[start:])
        self.bwt = [self.text[(start - 1) % self.n] for start in self.sa]
        frequencies = Counter(self.bwt)
        self.cumulative: dict[int, int] = {}
        total = 0
        for symbol in sorted(frequencies):
            self.cumulative[symbol] = total
            total += frequencies[symbol]

        running = {symbol: 0 for symbol in frequencies}
        self.checkpoints: list[dict[int, int]] = []
        for position in range(self.n + 1):
            if position % self.q == 0:
                self.checkpoints.append(running.copy())
            if position < self.n:
                running[self.bwt[position]] += 1

        self.samples = {
            row: start for row, start in enumerate(self.sa) if start % self.s == 0
        }

    def rank(self, symbol: int, end: int) -> int:
        if not 0 <= end <= self.n:
            raise IndexError("rank endpoint outside BWT")
        if symbol not in self.cumulative:
            return 0
        checkpoint = end // self.q
        start = checkpoint * self.q
        value = self.checkpoints[checkpoint].get(symbol, 0)
        return value + sum(self.bwt[i] == symbol for i in range(start, end))

    def interval(self, pattern: str) -> tuple[int, int]:
        if not pattern:
            raise ValueError("empty query requires an explicit policy")
        encoded = []
        for character in pattern:
            if character not in self.code:
                return 0, 0
            encoded.append(self.code[character])
        left, right = 0, self.n
        for symbol in reversed(encoded):
            left = self.cumulative[symbol] + self.rank(symbol, left)
            right = self.cumulative[symbol] + self.rank(symbol, right)
            if left >= right:
                return 0, 0
        return left, right

    def count(self, pattern: str) -> int:
        left, right = self.interval(pattern)
        return right - left

    def lf(self, row: int) -> int:
        symbol = self.bwt[row]
        return self.cumulative[symbol] + self.rank(symbol, row)

    def locate_row(self, row: int) -> int:
        steps = 0
        while row not in self.samples:
            row = self.lf(row)
            steps += 1
            if steps > self.n:
                raise AssertionError("sample not reached")
        return (self.samples[row] + steps) % self.n

    def locate(self, pattern: str) -> list[Location]:
        left, right = self.interval(pattern)
        result = []
        for row in range(left, right):
            location = self.origin[self.locate_row(row)]
            if location is not None:
                result.append(location)
        return sorted(result)

    def baseline(self, pattern: str) -> list[Location]:
        if not pattern:
            raise ValueError("empty query requires an explicit policy")
        result = []
        for document_id, document in enumerate(self.documents):
            start = 0
            while True:
                match = document.find(pattern, start)
                if match < 0:
                    break
                result.append(Location(document_id, match))
                start = match + 1
        return sorted(result)


if __name__ == "__main__":
    index = CorpusFMIndex(["banana", "bandana", "ananas"], 3, 4)
    assert index.locate("ana") == [
        Location(0, 1), Location(0, 3), Location(1, 4),
        Location(2, 0), Location(2, 2),
    ]
    assert index.locate("nab") == []
    queries = {"a", "ana", "banana", "band", "dana", "nanas", "absent"}
    for document in index.documents:
        for left in range(len(document)):
            for right in range(left + 1, len(document) + 1):
                queries.add(document[left:right])
    for query in queries:
        assert index.locate(query) == index.baseline(query)
        assert index.count(query) == len(index.baseline(query))

Counting and Locating Matches

Backward search yields one suffix-array interval. Counting returns its width. Locating enumerates its rows, follows LF to sampled suffix positions, reconstructs global positions, and maps them to (document, offset). Count-only indexes need no SA samples; locate and display features add samples or access to original text. Results may remain in suffix order or be sorted by document and offset.

Count and locate

flowchart TD
    A["Pattern"] --> B["Backward search with C and rank"]
    B --> C["Suffix-array interval [l, r)"]
    C --> D["Count = r − l"]
    C --> E["Enumerate rows"]
    E --> F{"SA value sampled?"}
    F -- "No" --> G["Apply LF; increment steps"]
    G --> F
    F -- "Yes" --> H["Recover global position modulo N"]
    H --> I["Origin map"]
    I --> J{"Content position?"}
    J -- "No" --> K["Discard structural position"]
    J -- "Yes" --> L["Emit document and offset"]

Compression and Parameter Study

Checkpoint rate $q$ trades rank space for at most $q-1$ scanned symbols. Sampling rate $s$ trades SA-sample space for at most $s-1$ LF steps per occurrence. Count latency is independent of $s$, while locate-heavy workloads may be dominated by LF walks and output size. Compare serialized bytes, not host-object overhead, and report median plus tail latency.

Benchmark Lab

Tune the Index

Build every pair $q\in\{16,32,64,128,256\}$ and $s\in\{4,8,16,32,64\}$. Run absent, count-heavy, and locate-heavy query sets. Record BWT, rank, and sample bytes; median and p99 count/locate latency; LF steps per occurrence; and output count.

Given a 64 MiB limit, p99 count below 500 microseconds, and p99 locate below 5 milliseconds, choose the smallest qualifying index. Verify that changing $s$ leaves counts unchanged, maximum LF steps do not exceed $s-1$, increasing $q$ reduces checkpoint bytes while increasing scan work, and all configurations reproduce baseline locations.

Correctness and Robustness

Compare every query with direct per-document search. Include every substring of small documents, absent symbols, repeated patterns, matches at boundaries, empty documents, and strings that would match only across a separator. Assert count == len(locate) and that every located slice equals the query.

Reject missing or duplicate sentinels, reused separators, invalid samples, malformed checkpoints, non-monotone $C$ tables, and incompatible versions. Persist encoding and normalization policies with the index.

The sentinel must occur exactly once and compare smaller than every other symbol; every document separator must lie outside the query alphabet and differ from the sentinel. These invariants make suffix order unambiguous and prevent ordinary-symbol patterns from crossing document boundaries.

Cookie Consent