Add fast parallel decoding, indexed access, and native distribution

.github/workflows/ci.yml

@@ -15,11 +15,13 @@ jobs:
     steps:
       - uses: actions/checkout@v7
       - uses: dtolnay/rust-toolchain@stable
-      - run: cargo test
+      - run: cargo test --release
       - run: cargo check --all-features
       - uses: actions/setup-python@v7
         with:
           python-version: '3.12'
+      - run: cargo build --release --bins
+      - run: python tools/stage_binaries.py
       - run: pip install -e '.[dev]'
       - run: pytest -q
 
@@ -31,10 +33,17 @@ jobs:
     runs-on: ${{ matrix.os }}
     steps:
       - uses: actions/checkout@v7
+      - uses: dtolnay/rust-toolchain@stable
+      - name: Stage native binary
+        if: runner.os != 'Linux'
+        run: |
+          cargo build --release --bins
+          python tools/stage_binaries.py
       - uses: PyO3/maturin-action@v1
         with:
           args: --release --out dist -i python3.10 -i python3.11 -i python3.12 -i python3.13
           manylinux: auto
+          before-script-linux: cargo build --release --bins && python3.10 tools/stage_binaries.py
       - uses: actions/upload-artifact@v7
         with:
           name: wheels-${{ matrix.os }}

Cargo.toml

@@ -13,12 +13,26 @@ documentation = "https://github.com/AnswerDotAI/fastbz2"
 name = "fastbz2"
 crate-type = ["cdylib", "rlib"]
 
+[[bin]]
+name = "fastbz2"
+path = "src/bin/fastbz2.rs"
+test = false
+
 [profile.release]
 lto = true
 codegen-units = 1
 
 [dependencies]
-pyo3 = { version = ">=0.28", optional = true }
+blake3 = "1.8.7"
+clap = { version = "4.6.6", features = ["derive"] }
+memmap2 = "0.9.11"
+pyo3 = { version = ">=0.29.2", optional = true }
+rayon = "1.12.0"
+tempfile = "3.27.0"
+
+[dev-dependencies]
+crabz2 = "0.4.0"
+libbz2-rs-sys = { version = "0.2.5", default-features = false, features = ["std"] }
 
 [features]
 python = ["dep:pyo3"]

DEV.md

@@ -6,34 +6,67 @@
 
 ```text
 src/bitreader.rs      bounded MSB-first in-memory bit reads
+src/block.rs          independently decodable block construction and validation
 src/crc.rs            bzip2 block and combined-stream CRC primitives
+src/decode.rs         serial/parallel decode scheduling and index construction
+src/decoder.rs        bzip2 block machinery and 12-bit Huffman fast tables
 src/format.rs         cheap structural scan for header and marker candidates
+src/index.rs           stable persistent index format
+src/indexed.rs         seekable decoded view and block cache
 src/lib.rs            public Rust API and private PyO3 binding
+src/source.rs         owned and memory-mapped compressed sources
 python/fastbz2/       thin Python I/O wrapper over fastbz2._core
-tests/                Python API and integration tests
+tests/corpus/         selected upstream conformance and corruption fixtures
+tests/                Rust CLI/corpus and Python API integration tests
+tools/stage_binaries.py copy the release executable into Maturin wheel data
 ```
 
 The current scanner deliberately does not treat 48-bit marker matches or later `BZh` headers as validated structure. Full decoding must establish the exact block chain and validate every block CRC plus the combined stream CRC before marker candidates can become trusted index entries. Python integration tests use standard-library `bz2`/libbz2 as an independent fixture generator.
 
 The decoder remains independent of files, threads, Python, and the CLI. Parallel scanning/decoding and indexed seeking are layered over it. Native workers never call Python. Large offsets use explicit 64-bit bit/byte types, and speculative block-marker hits are accepted only when they form an exact stream chain with valid block and combined stream CRCs.
 
-Start with safe scalar Rust designed for LLVM auto-vectorisation. Add narrowly scoped unsafe or architecture-specific SIMD only after profiling, with the safe implementation retained as a differential oracle.
+The production decoder is safe scalar Rust designed for LLVM auto-vectorisation. Huffman decoding uses a 4096-entry direct table for codes up to 12 bits and canonical fallback for longer codes. Add narrowly scoped unsafe or architecture-specific SIMD only after profiling; `libbz2-rs-sys` remains the dev-only differential oracle.
 
 ## Commands
 
 ```bash
 cargo test
+cargo test --release
 cargo check --all-features
-maturin develop
+cargo build --release --bins
+python tools/stage_binaries.py
+maturin develop --release
 pytest -q
 ship-rs-build
 ```
 
 Run `cargo fmt --check` after Rust edits and `chkstyle` after Python edits once tests pass.
 
-## Performance acceptance
+## Correctness and performance acceptance
 
-Build the checked-out `librapidarchive` implementation and compare on the same host, input, output sink, cache condition, and thread counts. Compare several-run medians for single-thread and parallel throughput, scaling, peak RSS, and time to first output. Within 20% end-to-end throughput is good enough when correctness and memory requirements pass.
+The normal release test path decodes selected valid and corrupt cases from the maintained upstream `bzip2-testfiles` collection. Generated byte distributions add differential coverage. Valid outputs are compared byte-for-byte with `libbz2-rs-sys`, which is a dev-only oracle and never part of production decoding.
+
+The same test binary contains a warmed end-to-end performance gate requiring `fastbz2` to complete a representative workload within 1.2 times the oracle. Keep the whole release test suite below five seconds on the primary development laptop; individual timed workloads should normally be about 0.1 seconds or less. Use the Simple English Wikipedia dump for heavier local throughput, scaling, memory, and time-to-first-output checks. `librapidarchive` was a one-time design comparison, not a retained baseline.
+
+Legacy randomized blocks produced by bzip2 versions before 0.9.5 are intentionally unsupported. Supporting that obsolete format would add complexity to the production decoder for data that is not realistically encountered today.
+
+### Local SimpleWiki benchmark
+
+`tests/wiki_perf.rs` contains a release-only local benchmark that is skipped by default. Its git-ignored fixture is a single bzip2 stream containing the first 84,423,012 decoded bytes (about 5%) of SimpleWiki, stored at `meta/simplewiki-first-5pct.xml.bz2`. Run it with:
+
+```bash
+cargo test --release --test wiki_perf simplewiki_first_five_percent -- --ignored --nocapture
+```
+
+It uses all available CPUs by default. Set `FASTBZ2_THREADS` to compare an explicit thread count. The timed section includes allocation and decompression but excludes reading the compressed file and calculating its BLAKE3; decoded length, block/stream CRCs, and BLAKE3 are all checked.
+
+For an occasional full-dump confirmation, create `meta/simplewiki-full.xml.bz2` as a symlink to the local dump and run:
+
+```bash
+cargo test --release --test wiki_perf simplewiki_full -- --ignored --nocapture
+```
+
+The full test streams decoded bytes into a counting sink and validates every block and stream CRC. Stream headers found during the structural scan remain speculative: only the exact byte-aligned header following a validated end-of-stream marker starts another stream.
 
 ## Platforms
 
@@ -45,8 +78,9 @@ The canonical version lives in `Cargo.toml`. `pyproject.toml` gets the Python pa
 
 ## Release
 
-1. Run `maturin develop && pytest -q`.
-2. Confirm the release version in `Cargo.toml` (`[package].version`).
-3. Run `ship-release`.
+1. Run `cargo build --release --bins && python tools/stage_binaries.py`.
+2. Run `maturin develop --release && pytest -q`.
+3. Confirm the release version in `Cargo.toml` (`[package].version`).
+4. Run `ship-release`.
 
 Fastship pushes the version tag for GitHub Actions, then bumps and pushes `Cargo.toml`.

README.md

@@ -4,9 +4,9 @@ Fast parallel and indexed bzip2 decompression for Rust and Python.
 
 `fastbz2` is initially focused solely on bzip2: a portable Rust core, a native CLI, and a thin PyO3 seekable-file API. The primary targets are Linux on x86-64 and ARM64, and macOS on ARM64; macOS Intel is best-effort. Correct output, block and stream CRC validation, bounded memory, and deterministic behaviour across thread counts are hard requirements.
 
-The first performance target is end-to-end throughput within 20% of `librapidarchive`'s `indexed_bzip2` on the same host and input. Portable, SIMD-friendly Rust comes first; architecture-specific SIMD is added only when profiles justify it.
+The performance floor is end-to-end decompression within 20% of the maintained pure-Rust `libbz2-rs-sys` decoder on a representative corpus. Portable, SIMD-friendly Rust comes first; architecture-specific SIMD is added only when profiles justify it. The much larger Simple English Wikipedia dump is used for local throughput measurements.
 
-The first implemented layer is a safe, portable MSB-first bit reader, bzip2 CRC primitives, and a structural scanner for stream headers and non-byte-aligned block/end markers. The scanner reports candidates rather than claiming validation: the decoder, exact stream-chain validation, indexing, parallel scheduler, CLI, and seekable Python file API are not implemented yet.
+The implementation includes a safe structural scanner, an in-repo decoder with a tuned 12-bit Huffman lookup table, CRC-validated block decoding, bounded parallel scheduling, persistent indexes, a native CLI, and a seekable Python file API. Marker scans remain speculative until decoding establishes an exact stream chain and validates block and combined-stream CRCs.
 
 ```python
 import bz2
@@ -19,15 +19,19 @@ result.blocks[0].bit_offset
 
 ## Inspiration and credit
 
-The architecture is inspired by Maximilian Knespel's [`librapidarchive`](https://github.com/mxmlnkn/librapidarchive) and [`indexed_bzip2`](https://github.com/mxmlnkn/indexed_bzip2): in particular, scanning for non-byte-aligned bzip2 block markers, independently decoding blocks, ordered prefetch, and indexed seeking. That project's specialised decoder is itself derived from Rob Landley's 0BSD [`bzcat` implementation in Toybox](https://github.com/landley/toybox). `fastbz2` is intended as a new Rust implementation, with the original project retained as the correctness and performance reference.
+The architecture was inspired by Maximilian Knespel's [`librapidarchive`](https://github.com/mxmlnkn/librapidarchive) and [`indexed_bzip2`](https://github.com/mxmlnkn/indexed_bzip2): in particular, scanning for non-byte-aligned bzip2 block markers, independently decoding blocks, ordered prefetch, and indexed seeking. That project's specialised decoder is itself derived from Rob Landley's 0BSD [`bzcat` implementation in Toybox](https://github.com/landley/toybox).
 
 ## Development
 
 ```bash
 pip install -e .[dev]
-maturin develop && pytest -q
+cargo build --release --bins && python tools/stage_binaries.py
+cargo test --release
+maturin develop --release && pytest -q
 ```
 
+Python wheels also install the native `fastbz2` executable directly into the environment's scripts directory; it is not a Python entry point or wrapper.
+
 ## Build
 
 ```bash
@@ -37,7 +41,8 @@ ship-rs-build
 ## Release
 
 ```bash
-maturin develop && pytest -q
+cargo build --release --bins && python tools/stage_binaries.py
+maturin develop --release && pytest -q
 ship-release
 ```
 

pyproject.toml

@@ -27,6 +27,7 @@ Issues = "https://github.com/AnswerDotAI/fastbz2/issues"
 features = ["extension-module"]
 python-source = "python"
 module-name = "fastbz2._core"
+data = "target/wheel-data"
 
 [tool.uv]
 cache-keys = [{ file = "pyproject.toml" }, { file = "src/**/*.rs" }, { file = "Cargo.toml" }, { file = "Cargo.lock" }]

python/fastbz2/__init__.py

@@ -1,6 +1,12 @@
 from collections import namedtuple
+import io
+import os
+from pathlib import Path
 
-from ._core import __version__, _scan, bz2_crc32
+from ._core import BadBzip2File, _IndexedReader, __version__, _build_index, _decompress, _scan, _test, bz2_crc32
+
+DEFAULT_MEMORY_LIMIT = 512 * 1024 * 1024
+DEFAULT_CACHE_LIMIT = 64 * 1024 * 1024
 
 StreamHeaderCandidate = namedtuple("StreamHeaderCandidate", "byte_offset block_size_100k")
 BlockCandidate = namedtuple("BlockCandidate", "bit_offset expected_crc randomized orig_ptr")
@@ -19,6 +25,78 @@ def scan(data: bytes) -> ScanResult:
     stream_ends = [EndCandidate(*item) for item in stream_ends]
     return ScanResult(streams, blocks, stream_ends)
 
+def decompress(data: bytes, *, threads=0, memory_limit=DEFAULT_MEMORY_LIMIT) -> bytes:
+    """Decompress and fully CRC-validate one or more concatenated bzip2 streams."""
+    return _decompress(data, threads, memory_limit)
+
+class IndexedBzip2File(io.RawIOBase):
+    """Seekable binary reader backed by a validated bzip2 block index."""
+
+    def __init__(self, source, *, threads=0, index=None, memory_limit=DEFAULT_MEMORY_LIMIT, cache_limit=DEFAULT_CACHE_LIMIT):
+        super().__init__()
+        if isinstance(source, (bytes, bytearray, memoryview)):
+            if index is not None and not isinstance(index, (bytes, bytearray, memoryview)):
+                index = Path(index).read_bytes()
+            self._reader = _IndexedReader.from_bytes(bytes(source), threads, memory_limit, index, cache_limit)
+        else:
+            if index is not None and isinstance(index, (bytes, bytearray, memoryview)):
+                raise TypeError("an in-memory index requires an in-memory bzip2 source")
+            index_path = None if index is None else os.fspath(index)
+            self._reader = _IndexedReader.from_path(os.fspath(source), threads, memory_limit, index_path, cache_limit)
+
+    def readable(self): return True
+    def seekable(self): return True
+
+    def read(self, size=-1):
+        self._checkClosed()
+        return self._reader.read(size)
+
+    def readinto(self, buffer):
+        data = self.read(len(buffer))
+        buffer[:len(data)] = data
+        return len(data)
+
+    def seek(self, offset, whence=io.SEEK_SET):
+        self._checkClosed()
+        return self._reader.seek(offset, whence)
+
+    def tell(self):
+        self._checkClosed()
+        return self._reader.tell()
+
+    @property
+    def size(self):
+        self._checkClosed()
+        return self._reader.size
+
+    def index_bytes(self):
+        self._checkClosed()
+        return self._reader.index_bytes()
+
+    def close(self):
+        self._reader = None
+        super().close()
+
+def open(source, *, threads=0, index=None, memory_limit=DEFAULT_MEMORY_LIMIT, cache_limit=DEFAULT_CACHE_LIMIT):
+    """Open a path or bytes object as a seekable bzip2 binary file."""
+    return IndexedBzip2File(source, threads=threads, index=index, memory_limit=memory_limit, cache_limit=cache_limit)
+
+def build_index(source, path=None, *, threads=0, memory_limit=DEFAULT_MEMORY_LIMIT) -> bytes:
+    """Fully validate *source* and return its source-bound binary block index."""
+    if isinstance(source, (bytes, bytearray, memoryview)):
+        reader = _IndexedReader.from_bytes(bytes(source), threads, memory_limit, None, DEFAULT_CACHE_LIMIT)
+        encoded = reader.index_bytes()
+    else: encoded = _build_index(os.fspath(source), threads, memory_limit)
+    if path is not None: Path(path).write_bytes(encoded)
+    return encoded
+
+def test(source, *, threads=0, memory_limit=DEFAULT_MEMORY_LIMIT):
+    """Fully decode and CRC-validate *source*, returning ``None`` on success."""
+    if isinstance(source, (bytes, bytearray, memoryview)):
+        _IndexedReader.from_bytes(bytes(source), threads, memory_limit, None, DEFAULT_CACHE_LIMIT)
+    else: _test(os.fspath(source), threads, memory_limit)
+
 __all__ = [
-    "__version__", "BlockCandidate", "EndCandidate", "ScanResult", "StreamHeaderCandidate", "bz2_crc32", "scan"
+    "__version__", "BadBzip2File", "BlockCandidate", "DEFAULT_CACHE_LIMIT", "DEFAULT_MEMORY_LIMIT", "EndCandidate",
+    "IndexedBzip2File", "ScanResult", "StreamHeaderCandidate", "build_index", "bz2_crc32", "decompress", "open", "scan", "test"
 ]