Jammi AI

Jammi is an embeddable AI engine that brings model inference into your data pipeline. Register data sources, run SQL queries, generate embeddings, search with vector similarity, fine-tune models on your domain, and evaluate results — all without leaving your application.

What Jammi does

• Query local data with SQL — register Parquet, CSV, JSON, and JSONL files, run full SQL via DataFusion
• Federate external databases — query PostgreSQL and MySQL alongside local files
• Generate embeddings — load any BERT-family model from HuggingFace Hub (or local safetensors / ONNX), persist results to Parquet with sidecar ANN indexes
• Vector search — ANN similarity search over embedding tables with automatic fallback to brute-force; search returns a table directly, same shape embedded or remote
• Compound query — join sources, filter, and run a model over a relation (the annotate SQL table function), in-process or over the Flight SQL lane in one round-trip; a fluent QueryBuilder composes the same operations in Rust
• Evidence provenance — retrieved_by and annotated_by tracking on the fluent query builder’s results
• Fine-tuning — LoRA adapters with contrastive loss to improve embeddings for your domain
• Evaluation — retrieval metrics (recall@k, precision@k, MRR, nDCG), classification (accuracy, F1), and A/B model comparison
• Per-row error handling — null or invalid text produces error status per row, not a batch failure
• Model caching — LRU eviction, ref-counted guards, single-flight loading
• GPU scheduling — memory-budget admission control with RAII permits
• Crash recovery — on restart, recovers result tables stuck in “building” state
• Inference observability — attach observers to hook into every output batch

Three ways to use Jammi

All three interfaces share the same engine, configuration, and storage format. Embeddings generated from Python are queryable from the CLI, and vice versa.

For multi-language access or BI tool integration, the jammi-server binary starts an Arrow Flight SQL server — any Arrow client can connect and query via standard SQL. The jammi CLI is a strict gRPC client of that server.

Crates

jammi-db has no dependency on jammi-ai. You can use it standalone for SQL queries over local data without pulling in the AI layer.

Design Philosophy

Jammi is an engine of generic primitives, not a substrate-platform. This page states what that means, where the line falls, and how the engine is meant to be consumed and deployed. It is principle-level on purpose — it pins no versions, index types, or other details that move, so it stays true as the implementation evolves.

The one rule everything else follows from

Jammi names no consumer. Not in code, config, docs, tests, fixtures, or scripts. References point one way only: a consumer may depend on Jammi; Jammi depends on no consumer. A consumer’s name anywhere in the engine repo is a bug.

A real consumer need is a fine forcing function for the roadmap — but the thing that lands in the engine is the generic primitive the need pointed at, with the name filed off. Two unrelated consumers independently reaching for the same primitive is the strongest evidence it is right; being able to justify a primitive only by naming one consumer is the strongest evidence it is wrong.

The discipline test

Before any capability enters the engine:

Would a user who has never heard of any particular consumer reach for this on its own?

Justify it against unrelated, hypothetical consumers — a feature store, an ad-attribution chain, a clinical-trial data fabric, a personal-knowledge search tool. If it survives only with a real name attached, it is domain pull masquerading as a primitive, and it belongs in that consumer’s own repo, built on a published Jammi version.

Where the line falls

Left column: every cell is something a user with no knowledge of any specific consumer would still want. Right column: every cell is a composition a consumer builds in its own repo out of the left column. The substrate-platform shape — append-only typed substrate + pluggable reactors + read-back loops at decision moments — recurs across consumers, but it recurs in their domain layers, not here. If Jammi codified that shape, the next consumer, in a domain we have not met, would have to bend to fit or fork.

Leak-guards

Domain pull leaks in through “almost-generic” primitives that quietly assume a consumer’s semantics. Three guards, all the same shape — the primitive transports / persists / merges; the semantics live above it:

• The trigger stream knows nothing about the payload. A topic is a name; a message is an Arrow batch; a subscription is a SQL predicate over batch columns. No typed-event taxonomy, no required headers (actor, timestamp, signature), no ordering guarantee beyond per-topic FIFO. Adding a “typed event” with mandatory headers is where it would break.
• Mutable tables expose CRUD through DML, nothing more. No built-in transition log, no automatic versioning, no lifecycle-column convention. A consumer that wants append-only-with-history builds it from two table registrations. Adding a LifecycleTable wrapper is where it would break.
• Provenance channels merge declared columns at query time. The engine never writes to a provenance column. What a channel column means — signed-by, retrieved-by, scored-by, attributed-to — is the caller’s vocabulary. Adding channel-writing helpers (record_actor(), sign_with()) is where it would break.

How embeddings are consumed: search

There is one consumption verb. The embedding producers differ only in the encoder; the moment vectors land in a result-table plus its sidecar index, consumption is identical: search(source, query, k) returns top-k ids and scores — ANN over the sidecar index, with an exact scan as the fallback when no index is present.

search is the curated path because it is where the engine adds value: the ANN index, the exact-scan fallback, and the evidence/provenance attached to results. The raw vector itself is a column in a SQL-addressable result-table, so it is reachable through the generic SQL surface (SELECT <vector_column> FROM <result_table> …) — the same as selecting any other column. That generic read path is legitimate, and making it ergonomic and documented is ordinary engine work; it is not a violation of anything here. It is the right home for the rare genuine need — exporting embeddings, a custom-metric re-rank, debugging.

What the engine does not add is a dedicated vector-retrieval verb on the embedding/search API. A consumer should not pull vectors back to reconstruct a ranking the engine already computes, or to compare across models (re-encode for that). A bespoke get_vector verb with no caller is speculative domain-convenience: it competes with search as “how you consume embeddings” and owes a per-id contract across every storage backend. So the line is precise — the capability is the SQL surface, not a verb; the verb is what fails the discipline test.

How it deploys: one binary, pluggable backends

The same engine binary serves every topology. Differences are configuration (which backend driver) and process count (1 vs N) — never a topology-specific code path or a server-only feature the library cannot do.

Four canonical shapes — points on a configuration surface, not tiers to graduate through:

• A — Single-process embedded. Library mode, SQLite catalog, local Parquet, in-memory trigger stream, in-process model cache. Notebooks, CLI, single-machine, laptop dev.
• B — Single-tenant server. One process, Flight SQL + gRPC trigger stream exposed, optional Postgres catalog for HA, local disk + object-store backup. Physical-isolation requirements, on-prem.
• C — Multi-tenant server. One process (or stateless fleet), Postgres catalog, shared object store, shared trigger broker; tenant scope filters every catalog query. SaaS hosting many tenants — e.g. an edge-function deployment with the engine as a container sidecar reached over gRPC, or a managed multi-tenant service.
• D — Disaggregated. Catalog process ↔ stateless query workers ↔ GPU-resident inference workers ↔ trigger broker, each scaling independently; the same binary in every role. Very high scale, specialized GPU pools, split compliance posture.

The five pluggable backends — the entire deployment-knob surface:

Everything else — load balancing, ingress, TLS, secrets, IAM, observability stack, orchestration, autoscaling — is the consumer’s runtime, not the engine’s.

Three properties this preserves, and that a consumer evaluating Jammi should be able to test:

• The library is never less capable than the server. Anything the server does, the library does in-process. No feature gated to “clustered mode.”
• The default deployment fits on a laptop. SQLite + local filesystem + in-memory trigger stream + HF Hub cache is a complete deployment, no cloud service required.
• Production is a configuration change, not a fork. Moving from Shape A to Shape C swaps backend drivers; the engine code, schema, catalog discipline, and trigger-stream contract are unchanged.

Extending third-party libraries at their seams

Where a capability is better bought than built, the engine extends the library at the seam it already exposes for exactly that — its own extension points, never a fork and never a vendored copy — and keeps the retry/actuator/ownership discipline (attempts, terminal writes, who is allowed to write a row) on jammi’s own side of that seam. Two examples, one pattern:

• Fused CUDA kernels extend candle, never replace it: a jammi kernel is a candle CustomOp with a CPU reference arm and an optional fused CUDA arm, so the same tensor graph runs identically (bit-for-bit on the CPU arm, parity-tested on CUDA) whether or not the fused kernel compiles in. Candle’s own eager composition is always the fallback, never a second implementation to keep in sync.
• The Ballista compute plane extends Ballista, never forks it: a codec (PhysicalExtensionCodec), an execution engine wrapper, and a custom task distribution policy are all extension points Ballista ships for exactly this — a scheduler/executor role hosts them inside the SAME jammi-server binary, on jammi’s own shutdown (never Ballista’s own process-level start_server/start_executor_process, which install their own signal handlers). Every operator the codec does not know still crosses the wire through Ballista’s own codec unchanged, so extending the plan surface never touches Ballista’s own node set. Retries stay off on the Ballista side entirely (task_max_failures = stage_max_failures = 0) — a task fault surfaces to jammi’s own attempts/reclaim accounting, never a second, competing retry loop.

Positioning

With these primitives the engine is, precisely: an embeddable AI engine — federated SQL, durable result-tables with ANN indexes, mutable companion tables, embeddings / inference / search / fine-tune / eval, evidence provenance on every row, and a trigger-stream event surface. That is a general-utility data-and-AI engine. It is deliberately not a substrate-platform engine — because the substrate-platform shape belongs to the consumers that are that shape, and the next consumer may not be.

Installation

Rust

Add Jammi to your Cargo.toml:

[dependencies]
jammi-db = "0.25"
jammi-ai = "0.25"
tokio = { version = "1", features = ["full"] }

CLI

The jammi CLI registers sources, runs SQL, and starts the server. There are three ways to get it.

cargo install (CPU)

Builds from source on your machine. Needs the build dependencies below.

cargo install jammi-cli

The installed binary is jammi.

Prebuilt binary (CPU)

Download a stripped, ready-to-run binary from the GitHub releases. No build toolchain required. Assets are published per release:

• jammi-<version>-x86_64-unknown-linux-gnu.tar.gz — Linux x86-64 (built on a glibc 2.28 floor, so it runs on any newer Linux)
• jammi-<version>-aarch64-apple-darwin.tar.gz — macOS on Apple silicon

tar -xzf jammi-0.25.0-x86_64-unknown-linux-gnu.tar.gz
./jammi --help

GPU (CUDA 12)

GPU inference ships as a container image, not a bare binary. The jammi-ai-server-cu12 image runs jammi-server as its entrypoint and also carries the jammi admin CLI; it is turnkey:

docker run --gpus all \
  -p 127.0.0.1:8080:8080 -p 127.0.0.1:8081:8081 \
  ghcr.io/f-inverse/jammi-ai-server-cu12:latest

Both ports bind to 127.0.0.1: the server performs no authentication of its own (see The identity seam), so a loopback bind keeps the unauthenticated admin surface off the host’s public network until a terminator or reverse proxy is put in front of it.

Both :latest tags are re-pointed by every v* release tag (never by a prerelease); the CPU :latest can additionally be re-pointed to the current main by a manual build-and-push-main dispatch. Pin an exact :vX.Y.Z tag for a reproducible pull.

That runs jammi-server with zero config. See Deploy as a Server for GPU configuration and persistence.

Alternatively, install the CUDA server as a pip wheel — it ships the same jammi-server binary and pulls the CUDA runtime from nvidia-*-cu12 wheels, so no system CUDA install is required (only an NVIDIA driver on the host):

pip install jammi-server-cu12
jammi-server

The jammi-ai embed wheel is CPU-only; GPU inference runs in the server, reached from Python via jammi.connect("grpc://…").

Build dependencies (Linux)

If building from source, you need a C compiler and protoc:

# Debian/Ubuntu
apt-get install protobuf-compiler gcc g++ pkg-config

# RHEL/AlmaLinux
yum install protobuf-compiler gcc gcc-c++ pkg-config

All other native libraries (lzma, zstd, zlib, sqlite) are vendored and compiled from source automatically. These tools are pre-installed in the devcontainer and CI images.

Building jammi-db with the postgres or mysql source feature additionally requires OpenSSL’s development headers: these features link a native TLS stack (native-tls -> OpenSSL) rather than rustls, and jammi-db does not vendor OpenSSL.

# Debian/Ubuntu
apt-get install libssl-dev

# RHEL/AlmaLinux
yum install openssl-devel

# macOS (Homebrew)
brew install openssl pkg-config

See Connect to PostgreSQL / MySQL.

Python

pip install jammi-ai

Requires Python 3.8+. Pre-built wheels are available for Linux, macOS, and Windows.

From source

git clone https://github.com/f-inverse/jammi-ai.git
cd jammi-ai
cargo build --release

The CLI binary is at target/release/jammi (a strict gRPC client) and the server binary at target/release/jammi-server.

For the Python package from source:

pip install maturin
maturin develop --release

Runtime requirements

Jammi has no mandatory runtime dependencies beyond the binary itself.

Optional:

• CUDA toolkit + cuDNN for GPU inference (CPU works out of the box)
• HuggingFace Hub access for downloading models (first run downloads ~90MB for MiniLM, cached thereafter)
• PostgreSQL / MySQL client libraries if using federated database sources

Set HF_TOKEN for gated models, or HF_HOME to control the cache location — both are read as fallbacks when the config’s own [models] section (see Configuration) leaves hub_token/hub_cache_dir unset; a config value always wins over the environment variable.

Quickstart: Rust

This walkthrough registers a local data file, runs a SQL query, generates embeddings, and performs a semantic search — all in one program.

Full example

extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use std::sync::Arc;
use jammi_ai::session::InferenceSession;
use jammi_db::config::JammiConfig;
use jammi_db::source::{FileFormat, SourceConnection, SourceType};
use jammi_db::store::CachePolicy;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let config = JammiConfig::load(None)?;
    let session = Arc::new(InferenceSession::new(config).await?);

    // 1. Register a data source
    session.add_source("patents", SourceType::File, SourceConnection {
        url: Some("file:///path/to/patents.parquet".into()),
        format: Some(FileFormat::Parquet),
        ..Default::default()
    }).await?;

    // 2. Query with SQL
    let rows = session.sql(
        "SELECT id, title, year FROM patents.public.patents WHERE year > 2020 LIMIT 5"
    ).await?;
    for batch in &rows {
        println!("{batch:?}");
    }

    // 3. Generate embeddings
    let (record, _outcome) = session.generate_text_embeddings(
        "patents",
        "sentence-transformers/all-MiniLM-L6-v2",
        &["title".to_string()],
        "id",
        CachePolicy::Bypass,
        None,
    ).await?;
    println!("Embedded {} rows", record.row_count);

    // 4. Semantic search
    let query = session.encode_text_query(
        "sentence-transformers/all-MiniLM-L6-v2",
        "quantum computing applications",
    ).await?;

    let results = session.search("patents", query, 5, None, None).await?
        .sort("similarity", true)?
        .run().await?;

    for batch in &results {
        println!("{batch:?}");
    }

    Ok(())
}

The first run downloads the model from HuggingFace Hub (~90MB). Subsequent runs load from cache.

What’s happening

1. JammiConfig::load(None) loads config from jammi.toml, $JAMMI_CONFIG, or defaults
2. InferenceSession wraps the query engine with model loading, caching, and GPU scheduling
3. add_source registers a file in the catalog — it survives session restarts
4. sql runs any SQL query via DataFusion, returns Vec<RecordBatch>
5. generate_text_embeddings runs the model over every row, persists vectors to Parquet with a sidecar ANN index
6. encode_text_query encodes a text string into the same vector space
7. search finds the nearest neighbors, hydrates all source columns, and returns results with similarity scores

Next steps

• Query Your Data with SQL — SQL features, joins, aggregations
• Generate Embeddings — persistence, multiple models, crash recovery
• Semantic Search — SearchBuilder API, filtering, evidence provenance

Quickstart: Python

The full quickstart — install, connect, register, search — lives in the repo’s cookbook tree under cookbook/quickstart/ with a runnable quickstart.py that’s exercised end-to-end on every PR by tests/cookbook_smoke.py. This page mirrors the cookbook’s overview so the mdBook site renders a self-contained quickstart; the cookbook is the source of truth.

Goal: a fresh user goes from pip install jammi-ai to a successful vector query in five minutes. The end-to-end script lives next to this file in quickstart.py — copy-paste it, run it, then read the four step-by-step pages for the explanation.

Steps

1. Install — pip install jammi-ai
2. Connect — open a session against a local artifact dir
3. Register a source — attach a Parquet file
4. Generate embeddings + search — build a vector index and run a similarity query

Run it

python cookbook/quickstart/quickstart.py

Expected output: a header row and three top-3 matches with cosine similarity scores. The script exits 0 in under 30 seconds on CPU.

Production substitution

The script uses the local cookbook/fixtures/tiny_bert/ model (32-dim, 88 KB, single-layer) so the example needs no network access. In a real workload you would swap in a Hub model — for example sentence-transformers/all-MiniLM-L6-v2 (384-dim, English) — by changing the MODEL constant. Everything else stays the same.

Quickstart: CLI

The jammi CLI is a strict gRPC client: it talks to a running jammi-server over the wire and never touches the catalog or storage in-process. Start a server (see Deploy as a Server), then point the CLI at it with --target.

Register a source

# Register a remote source (the URL is resolved server-side)
jammi --target grpc://127.0.0.1:8081 \
  sources add patents --url /path/to/patents.parquet --format parquet

# List registered sources
jammi --target grpc://127.0.0.1:8081 sources list

The CLI is a control-plane tool — SQL itself runs over the server’s Flight SQL surface through the Rust or Python client, not through a CLI verb; see Query Your Data with SQL.

The default --target is grpc://127.0.0.1:8081, so a CLI talking to a local server can omit the flag.

Check the server

# Report version, compiled features, storage backends, and mounted services.
# A successful response also confirms reachability.
jammi status

Available commands

Global options

jammi --target <ENDPOINT> <command>   # Server endpoint (default grpc://127.0.0.1:8081)
jammi --tenant <UUID> <command>       # Bind a tenant scope for the session

--target accepts grpc://host:port, http://host:port, or a bare host:port — all plaintext h2. TLS termination is the consumer’s runtime, not the CLI’s: put a TLS-terminating proxy in front and point --target at it in plaintext. --tenant binds a tenant scope before any verb runs, so every read and write is scoped to that tenant.

Next steps

• Deploy as a Server — jammi-server, configuration, preloading models
• Configuration — full config reference

Runnable Recipes

Every recipe under cookbook/recipes/ ships as a runnable example.py next to a markdown README and is wired into CI via tests/cookbook_smoke.py — a broken recipe blocks the merge. These recipes are the OSS source of truth; this page mirrors each README below.

For the long-form, measured companion — The Cookbook, a Theory↔Computation book that shows one recipe = one equation in the graph-signal-processing monograph = one line of the GNN canon, executed against committed goldens — see The Cookbook. The two are complementary: these How-To Guides are the short, dual-language (Rust + Python), compile-tested “how do I call this verb” reference; The Cookbook is the long-form, Python, executed-and-measured narrative.

The recipes shipped at MVP:

Mutable tables

End-to-end create / insert / select / drop on a Jammi mutable table — the OSS primitive for state that needs to live alongside read-only result tables.

When to use this pattern. You need a writable table that sits in the same SQL catalog as your registered sources and embedding tables — for caching enriched rows, holding cursor state, recording user feedback, or any “small table I want to UPDATE / DELETE / INSERT from SQL” workload — without standing up an external Postgres.

What example.py does

1. Connects to a temporary artifact dir
2. Creates a notes mutable table with an int64 primary key + utf8 body column
3. Inserts three rows through DataFusion DML (INSERT INTO ...)
4. Verifies count and ordering via SELECT
5. Drops the table, then asserts a SELECT after the drop raises
6. Demonstrates the idempotent drop_mutable_table(..., if_exists=True)

API surface exercised

• Database.create_mutable_table(name, *, schema, primary_key, ...)
• Database.sql("INSERT INTO mutable.public.<name> ...")
• Database.sql("SELECT ... FROM mutable.public.<name>")
• Database.drop_mutable_table(name, *, if_exists=False)

The DataFusion namespace for mutable tables is always mutable.public.<name> — distinct from registered sources, which live under <source>.public.<source>.

Run it

python cookbook/recipes/mutable_tables/example.py

Exits 0 on success, prints mutable_tables: OK on the last line.

Trigger streams

End-to-end publish + subscribe on a Jammi topic, plus the registration and listing surface. Uses the embedded in-process broker — no NATS or external broker needed.

When to use this pattern. You need a low-friction event bus inside your application — for fan-out to downstream consumers, fan-in from batch jobs, or replay-from-offset semantics — without bringing up Kafka or NATS in dev/test. The same surface scales out to NATS JetStream by flipping a config flag at deploy time.

What example.py does

1. Connects to a temporary artifact dir
2. Registers a topic events.demo with a typed schema and broker metadata
3. Confirms list_topics() returns the new topic
4. Publishes a 3-row batch through publish_topic — captures the broker-assigned offset
5. Subscribes from from_offset=0 and round-trips the same rows back
6. Drops the topic, confirms it’s gone from list_topics()
7. Demonstrates idempotent drop_topic(..., if_exists=True) and strict-mode failure when dropping a missing topic

API surface exercised

• Database.register_topic(name, *, schema, broker_metadata=None)
• Database.list_topics()
• Database.publish_topic(name, *, batch) — returns the assigned offset
• Database.subscribe_collect(name, *, from_offset, max_batches)
• Database.drop_topic(name, *, if_exists=False)

The subscribe_collect path drives the replay-from-backing-table flow when from_offset=0; the live-tail flow is exercised in the broker integration suite.

Run it

python cookbook/recipes/trigger_streams/example.py

Exits 0 on success, prints trigger_streams: OK on the last line.

Evaluate retrieval quality

Measure recall@k, precision@k, MRR, and nDCG of an embedding index against a golden relevance set.

When to use this pattern. You have a corpus and a small set of (query, expected document) judgments, and you need a number that tells you “is my new encoder better than the one I shipped last month?” The same loop powers nightly regression dashboards and A/B model comparison.

What example.py does

1. Connects to a temporary artifact dir
2. Registers the tiny corpus as a Parquet source
3. Builds 32-dim embeddings over the content column with the local tiny_bert fixture
4. Reads cookbook/fixtures/tiny_golden.json, expands it into the (query_id, query_text, relevant_id) CSV shape eval_embeddings consumes, and registers it as a golden source
5. Calls db.eval_embeddings(source="corpus", golden_source="golden.public.golden", k=5)
6. Asserts each aggregate metric is in [0.0, 1.0] and the per-query records carry their golden-set query_id

API surface exercised

• Database.generate_embeddings(*, source, model, columns, key, modality="text")
• Database.eval_embeddings(*, source, golden_source, model=None, k=10)

The returned dict carries aggregate (mean across queries — recall_at_k, precision_at_k, mrr, ndcg) and per_query (one entry per query with query_id and a metrics sub-dict of the same four names, un-averaged).

Golden source shape

eval_embeddings requires a registered source with these columns:

Image queries are supported via a query_image BLOB column instead of query_text; cross-modal eval is out of scope for this recipe.

Run it

python cookbook/recipes/eval_embeddings/example.py

Exits 0 on success, prints the metrics dict + eval_embeddings: OK.

Image search

Run image-to-image semantic search over a corpus with an OpenCLIP-format vision model, measure retrieval quality, and adapt the vision tower itself on caller-supplied image triplets.

When to use this pattern. You have a corpus of images (figures, drawings, photos) and want to find the ones most similar to a query image — and a number that tells you how good the retrieval is. This is the image counterpart of the text eval_embeddings recipe.

Flow

1. Load a small image corpus (inline image bytes in a Parquet source)
2. Generate L2-normalized vision embeddings over the image column
3. Search the index with an encoded image query (cosine ANN)
4. Eval retrieval quality (Recall@K / MRR) against a held-out golden set
5. Fine-tune LoRA adapters inside the vision tower on image triplets (adapted ≠ base), then watch a wrong selector get refused

Model

A domain-specialized CLIP checkpoint is a drop-in for the reference model when your corpus is technical drawings or diagrams rather than photographs — a generic CLIP has seen few of them, and a checkpoint tuned on that kind of imagery separates them far better. patentclip/PatentCLIP_Vit_B on the Hugging Face Hub is one such checkpoint:

JAMMI_IMAGE_MODEL=patentclip/PatentCLIP_Vit_B \
    python cookbook/recipes/image_search/example.py

patentclip/PatentCLIP_Vit_B is pulled from the Hugging Face Hub on first use and produces 512-dim L2-normalized embeddings. Any OpenCLIP-format model works the same way — OpenAI CLIP, LAION CLIP-ViT-B-32-*, EVA-CLIP, etc. — the encoder is auto-detected from the model’s open_clip_config.json.

By default (no env var) the recipe runs against the hermetic cookbook/fixtures/tiny_open_clip fixture so it runs offline in CI — about two seconds on a laptop, nearly all of it the tower-LoRA and refusal legs (the search-and-eval flow alone is a fraction of a second). That fixture has random weights, so its retrieval numbers are meaningless — it exercises the full pipeline, not model quality. Point JAMMI_IMAGE_MODEL at any real checkpoint for real numbers.

What example.py does

1. Connects to a temporary artifact dir
2. Reads the 20 committed 224×224 PNGs under cookbook/fixtures/tiny_image_corpus/ into a Parquet corpus source (image_id, image bytes)
3. db.generate_embeddings(source="corpus", model=MODEL, columns=["image"], key="image_id", modality="image")
4. db.encode_query(model=MODEL, query=png_bytes, modality="image") → db.search("corpus", query=vec, k=5) (returns a pyarrow.Table)
5. Builds the image-query golden source from tiny_image_golden.json and calls db.eval_embeddings(source="corpus", golden_source="golden.public.golden", k=5)
6. Prints the aggregate Recall@K / precision@K / MRR / nDCG and the per-query records. It reports the metrics; it does not assert a quality bar.
7. Builds synthetic (anchor, positive, negative) image triplets from the corpus (positive = same shape family, negative = a different family) and calls db.fine_tune(source="triplets", base_model=MODEL, columns=["anchor","positive","negative"], method="lora", task="image_embedding", target_modules=["in_proj","c_fc"], ...). A non-empty target_modules puts LoRA inside the vision tower — in_proj is the transformer block’s fused-QKV projection, c_fc the MLP’s first linear — so the tower’s own representation moves. (The audio recipe runs both modes side by side: an empty list instead trains a projection head on a frozen tower — cheaper, less capacity.) It then re-encodes the same query image through the adapted model and asserts the embedding vector changed (max elementwise |Δ| > 1e-4 versus the base encoding).
8. Submits one more job with target_modules=["q_proj"] — a real selector on plenty of decoder checkpoints and on nothing in an OpenCLIP tower — and asserts job.wait() raises jammi.errors.TrainingError whose message echoes q_proj and names this tower’s real sites (in_proj, …). It prints the message.

What each leg proves, and the honesty rule

• The tower leg proves the adapter is trained and applied when the model is served: an adapter that trained but was silently dropped at serve time leaves the two query vectors bit-identical, and that is what the |Δ| check catches. It asserts change, not improvement — the default fixture has random weights, so the direction of the change carries no information. The vector check is also the deterministic one: asserting a top-k metric moved is flaky, because on this tiny eval set the rankings rarely flip even when the vectors do. What the tower leg canNOT check from here is the saved adapter’s kind — that it is an encoder-adapters bundle carrying the vision tower’s id. That is engine-internal and is pinned by the engine’s own integration tests; the client surface (describe_model) reports only the model’s id, backend, task and status, so the recipe asserts the task and leans on the |Δ| check for the rest.
• The refusal leg proves a selector that matches no site fails the job rather than publishing an adapter that changes nothing — and that the message is actionable, carrying this architecture’s own site vocabulary.
• The independently-known improvement number — tuned retrieval quality beating the base by a measured margin — is not this recipe’s to claim. It belongs to a real-checkpoint chapter (built from a committed GPU-produced cache; planned under issue #421), which does not exist yet. A recipe running a random-weight fixture on a laptop can honestly prove mechanism; it cannot prove quality.

The pairing semantics (what a “positive” means) are the caller’s training data, not the trainer’s: the trainer only minimizes the contrastive triplet loss over whatever images you pair.

Stepwise scripts

example.py runs every phase in one process (this is the version wired into tests/cookbook_smoke.py). The numbered scripts decompose the search-and-eval flow and share a persistent workdir, so run them in order:

python cookbook/recipes/image_search/01-load-corpus.py
python cookbook/recipes/image_search/02-generate-embeddings.py
python cookbook/recipes/image_search/03-search.py
python cookbook/recipes/image_search/04-eval.py

API surface exercised

• Database.generate_embeddings(*, source, model, columns, key, modality="image")
• Database.encode_query(*, model, query, modality="image") → list[float]
• Database.search(source, *, query, k, filter=None, select=None) → pyarrow.Table
• Database.eval_embeddings(*, source, golden_source, model=None, k=10)
• Database.fine_tune(*, source, base_model, columns, method, task="image_embedding", target_modules=[...], ...) → TrainingJob
• Database.describe_model(model_id) → dict | None

Image triplet schema (fine-tune input)

Same column shape as text and audio triplets — task="image_embedding" is what tells the loader to read the three columns as encoded images rather than text.

Vision-tower LoRA sites

target_modules names sites on this architecture. The OpenCLIP towers offer in_proj (fused QKV), out_proj, c_fc and c_proj; all-linear selects every one. A selector matches a site name exactly or as a suffix of it. A list matching nothing fails the job with a message that echoes what you submitted and lists the tower’s real names.

Input schema

Preprocessing (pad-to-square, no center crop, normalization, L2-normalized output) is handled inside the encoder per the model’s preprocess_cfg.

Golden source shape (image mode)

eval_embeddings switches to image-query mode when the golden source carries a query_image (binary) column instead of query_text:

Fixtures

• cookbook/fixtures/tiny_image_corpus/ — 20 synthetic 224×224 PNGs in 5 shape families (circle / triangle / square / hexagon / grating), 4 per family, plus a held-out query image per family under queries/. Rendered programmatically by cookbook/fixtures/generate.py — no real-world imagery (licensing).
• cookbook/fixtures/tiny_image_golden.json — per-query → expected corpus IDs (same shape family).
• cookbook/fixtures/tiny_open_clip/ — tiny offline OpenCLIP fixture used as the default CI model.

Run it

python cookbook/recipes/image_search/example.py

Exits 0 on success, prints the top-K and the metrics dict + image_search: OK.

Evaluate inference (classification)

Run a classifier over a registered source and score its predictions against gold labels.

When to use this pattern. You have a labelled holdout set and you want a single number — accuracy, macro F1, per-class F1 — to compare two classifiers, or to track drift over time on the same classifier.

What example.py does

1. Connects to a temporary artifact dir
2. Registers the tiny corpus as corpus (parquet)
3. Registers tiny_labels.csv as golden (csv) — (id, label) rows
4. Runs db.eval_inference with the local tiny_modernbert_classifier fixture against the content column
5. Prints the returned aggregate accuracy, macro f1, per-class metrics, and the count of per-record predictions
6. Asserts every reported rate is in [0.0, 1.0]

API surface exercised

• Database.eval_inference(*, model, source, columns, task, golden_source, label_column)

The returned dict carries aggregate (tagged by "task" — currently "classification") with accuracy, f1, and per_class, plus per_record (one entry per aligned {record_id, predicted, gold}).

The task argument is the string form of the inference task — "classification" here. For NER, see ../eval_inference_ner/.

Golden source shape

eval_inference requires a registered source with these columns:

label_column is the kwarg you pass at call time — label in this recipe. Every id in the golden source must resolve to a row in the input source; rows without a gold label are silently dropped from the metric.

Run it

python cookbook/recipes/eval_inference/example.py

Exits 0 on success, prints the metrics dict + eval_inference: OK.

Evaluate inference (NER)

Run a token-classification model over a registered source and score its predicted entity spans against gold spans.

When to use this pattern. You have a labelled NER holdout set (one gold span per row) and you want strict entity-level precision, recall, and F1 — both overall and per entity type — to compare two NER models or to track regressions on the same one.

What example.py does

1. Connects to a temporary artifact dir
2. Registers tiny_ner_corpus.parquet as corpus (parquet)
3. Registers tiny_ner_gold.csv as golden (csv) — one row per gold entity span: (id, label, start, end)
4. Runs db.eval_inference with the local tiny_modernbert_ner fixture against the text column, task="ner"
5. Prints the returned aggregate precision, recall, f1, the per-type breakdown, and the count of per-record predictions
6. Asserts every reported rate is in [0.0, 1.0]

API surface exercised

• Database.eval_inference(*, model, source, columns, task, golden_source, label_column)

The returned dict carries aggregate (tagged by "task" — "ner" for this recipe) with precision, recall, f1, and per_type (one breakdown per entity type the model emitted or the gold set carried), plus per_record (one entry per aligned {record_id, predicted, gold} where predicted and gold are entity-span lists, each tagged "task": "ner").

The task argument is the string form of the inference task — "ner" here. For classification, see ../eval_inference/.

Golden source shape

eval_inference with task="ner" requires a registered source with these columns — one row per entity span (multiple spans on the same id accumulate into one per-row gold set):

label_column is the kwarg you pass at call time — label in this recipe. start is inclusive, end is exclusive, both byte offsets into the source row’s text column. The label set must match the shipped model’s id2label minus the B-/I- prefixes — tiny_modernbert_ner knows PER and ORG only.

Rows in the source without a matching gold id are silently dropped from the metric (same alignment rule the classification recipe uses).

Run it

python cookbook/recipes/eval_inference_ner/example.py

Exits 0 on success, prints the metrics dict + eval_inference (ner): OK.

Fine-tune an encoder

Run a LoRA fine-tune on top of an existing text encoder, poll the job to completion, and use the resulting checkpoint to encode a query.

When to use this pattern. Your domain (legal contracts, medical abstracts, patent claims, internal product docs) doesn’t match the distribution the base encoder was trained on, and you have a few hundred to a few thousand labelled or contrastive pairs. LoRA gets you ~80% of the lift of a full fine-tune at a fraction of the cost; the resulting adapter is small enough to ship as an attachment to the base model rather than a re-distributed full checkpoint.

What example.py does

1. Connects to a temporary artifact dir
2. Registers tiny_pairs.csv (30 contrastive pairs) as training
3. Calls db.fine_tune(...) with the local tiny_bert base, a small LoRA rank, and one epoch — kept fast for CI
4. Waits for terminal status via job.wait()
5. Asserts the resulting model_id starts with jammi:fine-tuned:
6. Encodes a query through the fine-tuned model to confirm it loads

API surface exercised

• Database.fine_tune(*, source, base_model, columns, method, task=..., ...)
• Job.wait()
• Job.job_id, Job.output_model_id
• Database.encode_query(*, model, query, modality="text")

The full keyword list on fine_tune covers LoRA rank/alpha/dropout, learning rate, epochs, batch size, max sequence length, validation fraction, early-stopping patience/metric, warmup, gradient accumulation, backbone dtype, weight decay, and gradient clipping — the recipe uses the defaults for everything except rank and epochs.

Performance note

This recipe is excluded from the per-PR smoke matrix because even at one epoch it runs ~30 seconds on CPU. The nightly cron with JAMMI_COOKBOOK_SLOW=1 includes it. Override the gate locally:

JAMMI_COOKBOOK_SLOW=1 python tests/cookbook_smoke.py

Run it

python cookbook/recipes/fine_tune/example.py

Exits 0 on success, prints job_id, model_id, and fine_tune: OK.

Connect via Flight SQL

Run a query against a remote jammi-server over Arrow Flight SQL.

When to use this pattern. You’re connecting from a non-Python client (Tableau, dbt, JDBC tools, Rust binaries), or you want to expose Jammi to multiple readers without each one holding an embedded session. The same protocol is what dbt-flightsql, the official Flight SQL JDBC driver, and BI tools speak natively.

What example.py does

1. Spawns target/release/jammi-server as a child process pointed at a temp artifact_dir
2. Polls the health endpoint (http://127.0.0.1:8080/healthz) until the server is ready (5 s budget)
3. Opens a pyarrow.flight.FlightClient against grpc://127.0.0.1:8081
4. Submits SELECT 1 AS one over Flight SQL and confirms the response
5. Tears down the server process cleanly

This recipe is gated out of the per-PR CI matrix — it depends on the jammi-server binary being built (cargo build --release -p jammi-server), and the build cost dominates the test wall-clock. The nightly cookbook job builds the binary and runs the recipe behind JAMMI_COOKBOOK_SLOW=1.

Prerequisites

• cargo build --release -p jammi-server — produces target/release/jammi-server
• pip install pyarrow (already a jammi-ai dependency)

The script auto-detects JAMMI_BIN (env var) or falls back to the workspace’s target/release/jammi-server.

API surface exercised

• pyarrow.flight.FlightClient.execute(query) over the Flight SQL command dialect
• jammi-server — the OSS deployment-shape binary entrypoint

Run it

cargo build --release -p jammi-server      # one-time build
python cookbook/recipes/flight_sql/example.py

Exits 0 on success, prints the query result + flight_sql: OK.

Audio search

Run audio-to-audio similarity search over a corpus with a CLAP-format audio model, measure retrieval quality, and domain-tune the audio embeddings on caller-supplied triplets.

When to use this pattern. You have a corpus of sounds (clips, stems, loops, recordings) and want to find the ones most similar to a query clip — and a number that tells you how good the retrieval is. This is the audio counterpart of the image eval_embeddings recipe; audio is simply the third embedding modality the engine supports alongside text and images.

Flow

1. Load a small audio corpus (inline audio bytes in a Parquet source)
2. Generate L2-normalized audio embeddings over the audio column
3. Search the index with an encoded audio query (cosine ANN)
4. Eval retrieval quality (Recall@K / MRR) against a held-out golden set
5. Fine-tune a projection head on audio triplets and re-eval (tuned ≠ base)
6. Fine-tune LoRA adapters inside the audio tower on the same triplets (adapted ≠ base), then watch a wrong selector get refused

Model

Any HuggingFace CLAP audio model works — its config.json declares model_type = "clap_audio_model" (or lists ClapModel / ClapAudioModelWithProjection in architectures), its checkpoint exposes the audio_model.audio_encoder.* + audio_projection.* HTSAT-Swin tower keys, and a preprocessor_config.json carries the feature-extractor geometry. The encoder is auto-detected from that config, exactly as the image recipe auto-detects OpenCLIP:

JAMMI_AUDIO_MODEL=<hf-repo-id-or-local-path> \
    python cookbook/recipes/audio_search/example.py

By default (no env var) the recipe runs against the hermetic cookbook/fixtures/htsat_clap_tiny fixture so it runs offline in CI — on the order of ten seconds on a laptop, dominated by the three training legs (the projection head, the tower LoRA, and the refused job). That fixture has random weights, so its retrieval numbers are meaningless — it exercises the full pipeline, not model quality. Point JAMMI_AUDIO_MODEL at a real CLAP checkpoint for real numbers.

What example.py does

1. Connects to a temporary artifact dir
2. Reads the 20 committed mono WAV clips under cookbook/fixtures/tiny_audio_corpus/ into a Parquet corpus source (clip_id, audio bytes)
3. db.generate_embeddings(source="corpus", model=MODEL, columns=["audio"], key="clip_id", modality="audio")
4. db.encode_query(model=MODEL, query=wav_bytes, modality="audio") → db.search("corpus", query=vec, k=5) (returns a pyarrow.Table)
5. Builds the audio-query golden source from tiny_audio_golden.json and calls db.eval_embeddings(source="corpus", golden_source="golden.public.golden", k=5)
6. Prints the base aggregate Recall@K / precision@K / MRR / nDCG and the per-query records. It reports the metrics; it does not assert a quality bar.
7. Builds synthetic (anchor, positive, negative) audio triplets from the corpus (positive = same timbre family, negative = a different family) and calls db.fine_tune(source="triplets", base_model=MODEL, columns=["anchor","positive","negative"], method="lora", task="audio_embedding", ...). Empty target_modules ⇒ a trainable projection head on the frozen CLAP audio tower (the cheap, low-risk lightweight mode). It then re-embeds the corpus with the tuned model, re-evals, and prints base-vs-tuned metrics for narrative. For correctness it re-encodes the same query clip through the tuned model and asserts the embedding vector changed (max elementwise |Δ| > 1e-4 versus the base encoding) — the real invariant fine-tuning guarantees, and a deterministic check. (Asserting on the coarse top-k metrics instead is flaky: on this tiny eval set the rankings rarely flip even when the vectors move.) It proves the adapter alters audio retrieval — not that it improves it; the random-weight fixture’s direction is not meaningful, real lift comes from a real checkpoint.
8. Runs the other fine-tune mode on the same triplets: target_modules=["query", "value", "linear1"] puts LoRA inside the HTSAT-Swin tower itself — query/value are the Swin blocks’ attention projections (indexed by stage), linear1 the audio projection head’s first linear (an unindexed site). It re-encodes the same query clip through the adapted model and asserts the same |Δ| > 1e-4 change.
9. Submits one more job with target_modules=["q_proj"] — a real selector on plenty of decoder checkpoints and on nothing in an HTSAT-Swin tower — and asserts job.wait() raises jammi.errors.TrainingError whose message echoes q_proj and names this tower’s real sites (query, linear1, …). It prints the message.

What each leg proves, and the honesty rule

Both fine-tune legs ship and both are real; they are different capabilities:

• The projection-head leg (empty target_modules) trains a new map on top of a tower whose weights never move — cheap, low-risk, no site names needed.
• The tower leg (non-empty target_modules) moves the tower’s own representation — more capacity for a domain the base checkpoint never saw, at more compute, and it needs the site vocabulary of this architecture.

Both prove the adapter is trained and applied when the model is served: an adapter that trained but was silently dropped at serve time leaves the two query vectors bit-identical, and that is what the |Δ| check catches. Both assert change, not improvement — the default fixture has random weights, so the direction of the change carries no information.

What neither leg can check from here is the saved adapter’s kind — that it is an encoder-adapters bundle carrying the audio tower’s id. That is engine-internal and is pinned by the engine’s own integration tests; the client surface (describe_model) reports only the model’s id, backend, task and status, so the recipe asserts the task and leans on the |Δ| check for the rest.

The refusal leg proves a selector matching no site fails the job rather than publishing an adapter that changes nothing, and that the message is actionable — it carries this architecture’s own site names.

The independently-known improvement number — tuned retrieval quality beating the base by a measured margin — is not this recipe’s to claim. It belongs to a real-checkpoint chapter (built from a committed GPU-produced cache; planned under issue #421), which does not exist yet. A recipe running a random-weight fixture on a laptop can honestly prove mechanism; it cannot prove quality.

The pairing semantics (what a “positive” means) are the caller’s training data, not the trainer’s: the trainer only minimizes the contrastive triplet loss over whatever clips you pair.

Stepwise scripts

example.py runs every phase in one process (this is the version wired into tests/cookbook_smoke.py). The numbered scripts decompose the search-and-eval flow and share a persistent workdir, so run them in order:

python cookbook/recipes/audio_search/01-load-corpus.py
python cookbook/recipes/audio_search/02-generate-embeddings.py
python cookbook/recipes/audio_search/03-search.py
python cookbook/recipes/audio_search/04-eval.py

API surface exercised

• Database.generate_embeddings(*, source, model, columns, key, modality="audio")
• Database.encode_query(*, model, query, modality="audio") → list[float]
• Database.search(source, *, query, k, filter=None, select=None) → pyarrow.Table
• Database.eval_embeddings(*, source, golden_source, model=None, k=10)
• Database.fine_tune(*, source, base_model, columns, method, task="audio_embedding", target_modules=[...], ...) → TrainingJob
• Database.describe_model(model_id) → dict | None

Audio triplet schema (fine-tune input)

Same column shape as text triplets — task="audio_embedding" is what tells the loader to read the three columns as encoded audio rather than text.

Audio-tower LoRA sites

target_modules names sites on this architecture. An empty list means “no tower sites” and selects the projection-head mode instead. The HTSAT-Swin audio tower offers query, key, value, attention_output, intermediate_dense, output_dense, reduction, linear1 and linear2; all-linear selects every one. A selector matches a site name exactly or as a suffix of it. A non-empty list matching nothing fails the job with a message that echoes what you submitted and lists the tower’s real names.

Input schema

Preprocessing (decode → resample to the model’s sample rate → CLAP fusion log-mel spectrogram → HTSAT-Swin tower → L2-normalized output) is handled inside the encoder per the model’s preprocessor_config.json feature-extractor geometry. The audio column may also hold file-path strings instead of inline bytes.

Golden source shape (audio mode)

eval_embeddings switches to audio-query mode when the golden source carries a query_audio (binary) column instead of query_text / query_image:

Fixtures

• cookbook/fixtures/tiny_audio_corpus/ — 20 synthetic mono WAV clips in 5 timbre families (sine / harmonic / square / saw / noise), 4 per family, plus a held-out query clip per family under queries/. Synthesised programmatically by cookbook/fixtures/generate.py — no recorded audio (licensing), no tenant data.
• cookbook/fixtures/tiny_audio_golden.json — per-query → expected corpus IDs (same timbre family).
• cookbook/fixtures/htsat_clap_tiny/ — tiny offline HTSAT-Swin CLAP fixture used as the default CI model, generated by tests/fixtures/generate_htsat_clap.py.

Run it

python cookbook/recipes/audio_search/example.py

Exits 0 on success, prints the top-K and the metrics dict + audio_search: OK.

Per-query search audit

Record a tamper-evident audit row for every search: what was queried, with what model, what came back, and when. The substrate signs each record, stores it tenant-scoped, and publishes it to a trigger topic — so you do not hand-roll an audit schema, a signature scheme, and a stream integration in every project.

This is the primitive every audited-ML deployment in a regulated setting (finance, healthcare, legal, and the like) needs to answer “show me exactly what this model returned for this query, and prove the record hasn’t been altered.”

What this recipe shows

• Build a PerQueryAudit record (query id, model id/version, query lineage, top-K result ids, retrieval scores).
• db.audit.log([...]) — the substrate injects tenant_id, signs the record with a per-tenant HMAC-SHA256 key, stores it, and publishes it.
• db.audit.fetch_by_query_id(...) / db.audit.fetch_recent(...) — typed reads, tenant-scoped.
• record.verify() — re-derive the key and check the signature.
• Plain SQL over mutable.public."_jammi_search_audit" — same tenant scope.
• db.subscribe_collect("jammi.audit.search.v1", ...) — every logged record is also delivered on a trigger topic for alerting / analytics / warehouse sinks.

Run it

The audit master key is required — the substrate refuses to sign without it:

export JAMMI_AUDIT_MASTER_KEY=$(python -c "import secrets; print(secrets.token_hex(32))")
python cookbook/recipes/search_audit/example.py

The key derives a distinct signing secret per tenant via HKDF-SHA256 and is deterministic across restarts, so signatures written today verify after a redeploy. Source it from your secret manager — never hard-code it.

Key points

• Lineage is capped. query_lineage JSON may not exceed 8 KiB (override with JAMMI_AUDIT_MAX_LINEAGE_BYTES). Store image hashes and row IDs, not raw payloads — compliance posture is structural, not advisory.
• top_k_result_ids and retrieval_scores must be the same length. This is checked when you construct the record.
• The table is reserved. _jammi_search_audit is created implicitly on the first log; you cannot create or directly INSERT into it (that would bypass signing). Read it freely via SQL.
• Tenant isolation is automatic. A record logged under tenant A is invisible to tenant B, through both the typed API and raw SQL.

Ephemeral session storage

A session-scoped storage context whose tables are auto-deleted when the session ends — on explicit close(), on context-manager exit, or when the 60-second timeout scanner force-closes a session past its deadline. Every transition publishes to the jammi.audit.session_lifecycle.v1 trigger topic, giving an audit-log aggregator durable proof that the data was deleted.

Run it:

python cookbook/recipes/session_lifecycle/example.py

When to use it

Use an ephemeral session for sensitive transient data that must not outlive the request that produced it: uploaded images, derived embeddings, draft model inputs. The session is always tenant-scoped — tenant A can never see tenant B’s ephemeral tables.

When NOT to use it

Do not store long-lived data in an ephemeral session. The audit record, the persistent corpus, and anything compliance needs to read later belong in ordinary mutable tables. The pattern is: keep the throwaway working set (raw bytes, embeddings) in the ephemeral session, and write only durable lineage (hashes, ids, scores) to a persistent table — before you close the session, while the working data still exists.

API

with db.ephemeral_session(timeout_seconds=3600) as ephem:
    ephem.create_ephemeral_table("imgs", schema=schema, primary_key=["image_id"])
    ephem.insert("imgs", batch=table)
    rows = ephem.sql("imgs", "SELECT image_hash FROM {table}")
# close() runs on exit: tables dropped, `closed` event published

{table} in a sql query is replaced by the tenant-scoped reference to the named ephemeral table. The context manager is the recommended path; Drop is best-effort. Lifecycle events (opened, closed, timed_out, partial_deletion_failure) carry the session id, tenant, table count, and deleted-row count.

Query Your Data with SQL

Register data files as named sources, then query them with full SQL. Sources are persisted in the catalog and survive session restarts.

Register a source

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use jammi_ai::session::InferenceSession;
async fn ex(session: &InferenceSession) -> jammi_db::error::Result<()> {
use jammi_db::source::{FileFormat, SourceConnection, SourceType};

session.add_source("patents", SourceType::File, SourceConnection {
    url: Some("file:///data/patents.parquet".into()),
    format: Some(FileFormat::Parquet),
    ..Default::default()
}).await?;
Ok(()) }
}

Python

db.add_source("patents", url="/data/patents.parquet", format="parquet")

CLI

jammi sources add patents --url /data/patents.parquet --format parquet

Supported formats

Run a SQL query

Sources are accessible via three-part SQL names: <source_id>.public.<table_name>. The table name is derived from the file name (e.g., patents.parquet becomes patents).

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use jammi_ai::session::InferenceSession;
async fn ex(session: &InferenceSession) -> jammi_db::error::Result<()> {
let results = session.sql(
    "SELECT id, title, year FROM patents.public.patents WHERE year > 2020 ORDER BY year"
).await?;

for batch in &results {
    println!("{batch:?}");
}
Ok(()) }
}

Python

table = db.sql("SELECT id, title, year FROM patents.public.patents WHERE year > 2020 ORDER BY year")
print(table.to_pandas())

CLI

jammi query "SELECT id, title, year FROM patents.public.patents WHERE year > 2020 ORDER BY year"

Aggregations

SELECT category, COUNT(*) as count, AVG(citation_count) as avg_citations
FROM patents.public.patents
WHERE year > 2020
GROUP BY category
ORDER BY count DESC

Joins across sources

Register multiple sources and join them in a single query:

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use jammi_ai::session::InferenceSession;
use jammi_db::source::{FileFormat, SourceConnection, SourceType};
async fn ex(session: &InferenceSession) -> jammi_db::error::Result<()> {
session.add_source("companies", SourceType::File, SourceConnection {
    url: Some("file:///data/companies.csv".into()),
    format: Some(FileFormat::Csv),
    ..Default::default()
}).await?;

let results = session.sql("
    SELECT p.title, c.company_name
    FROM patents.public.patents p
    JOIN companies.public.companies c ON p.assignee_id = c.id
").await?;
Ok(()) }
}

Python

db.add_source("companies", url="/data/companies.csv", format="csv")

table = db.sql("""
    SELECT p.title, c.company_name
    FROM patents.public.patents p
    JOIN companies.public.companies c ON p.assignee_id = c.id
""")

Source lifecycle

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use jammi_ai::session::InferenceSession;
async fn ex(session: &InferenceSession) -> jammi_db::error::Result<()> {
// List registered sources
let sources = session.catalog().list_sources().await?;

// Remove a source
session.remove_source("patents").await?;
Ok(()) }
}

CLI

jammi sources list

Sources persist in the SQLite catalog at <artifact_dir>/catalog.db. Registering the same source ID twice returns an error — remove it first.

Execution plans

Use EXPLAIN (or the CLI explain command) to see how DataFusion will execute your query:

jammi explain "SELECT * FROM patents.public.patents WHERE year > 2020"

Generate Embeddings

Measured companion: for the long-form, executed-and-measured Python treatment, see The Cookbook → Constructing the Graph.

Generate vector embeddings by running a model over text columns from a registered source. Results are persisted to Parquet with sidecar ANN indexes for fast similarity search.

Basic usage

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use jammi_ai::session::InferenceSession;
use jammi_db::store::CachePolicy;
async fn ex(session: &InferenceSession) -> jammi_db::error::Result<()> {
let (record, _outcome) = session.generate_text_embeddings(
    "patents",
    "sentence-transformers/all-MiniLM-L6-v2",
    &["abstract".to_string()],
    "id",
    CachePolicy::Bypass,
    None,
).await?;

println!("Embedded {} rows, {} dimensions", record.row_count, record.dimensions().unwrap());
Ok(()) }
}

Python

db.generate_embeddings(
    source="patents",
    model="sentence-transformers/all-MiniLM-L6-v2",
    columns=["abstract"],
    key="id",
    modality="text",
)

What gets created

Each call creates a timestamped Parquet file plus a sidecar ANN index bundle:

{artifact_dir}/jammi_db/
├── patents__embedding__all-MiniLM-L6-v2__20260325T120000.parquet
├── patents__embedding__all-MiniLM-L6-v2__20260325T120000.usearch
├── patents__embedding__all-MiniLM-L6-v2__20260325T120000.rowmap
└── patents__embedding__all-MiniLM-L6-v2__20260325T120000.manifest.json

• Parquet file — source of truth. Contains _row_id, _source_id, _model_id, vector, _content_hash. Readable by external tools (DuckDB, Polars, pandas).
• .usearch — USearch HNSW graph for ANN search.
• .rowmap — maps internal USearch keys to _row_id strings.
• .manifest.json — metadata (dimensions, count, metric, backend).

The sidecar files are disposable — deleting them falls back to brute-force exact search. The Parquet file is the only thing that matters.

Embedding table schema

Failed rows (null or empty text) are excluded — only successfully embedded rows appear in the output. Rows are written in key order: CAST(key AS Utf8) ascending, ties broken by _content_hash, so the table’s bytes are identical across engine.execution_threads.

A NULL in the key column is not a per-row failure: the whole call is refused with the typed InvalidKey { column, null_count } before the model runs (the null count is exact; zero rows are embedded and nothing is written). Every row needs a key.

_content_hash is what makes the table refreshable: after the source changes, refresh_embeddings re-embeds only the rows whose hash differs and publishes a new version of the same table — see Refresh an Embedding Table Incrementally.

Text column format

A text column can be:

• Utf8 / LargeUtf8 / Utf8View — read directly, no conversion. Utf8View is what a plain Parquet Utf8 column becomes through a DataFusion scan under this workspace’s pinned Arrow/DataFusion versions, so an ordinary registered source’s text column takes this arm on the real, unmodified scan output.
• Any other non-binary type (e.g. Int64, Float64) — cast to text ("42", "3.14", …). The whole call is refused if the cast would introduce a null the source column did not have.
• Binary / LargeBinary / BinaryView / FixedSizeBinary — refused outright, naming the column’s data type. These hold raw bytes, not text (e.g. image or audio bytes submitted under a text-embedding task by mistake); the call never silently embeds an empty string for every row.

A null value in an otherwise-text column reads as the empty string (""), which the row-failure rule above then excludes from the output.

Multiple text columns

Pass multiple column names to concatenate them (space-separated) before embedding:

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use jammi_ai::session::InferenceSession;
use jammi_db::store::CachePolicy;
async fn ex(session: &InferenceSession) -> jammi_db::error::Result<()> {
session.generate_text_embeddings(
    "papers",
    "sentence-transformers/all-MiniLM-L6-v2",
    &["title".to_string(), "abstract".to_string()],
    "doi",
    CachePolicy::Bypass,
    None,
).await?;
Ok(()) }
}

Python

db.generate_embeddings(
    source="papers",
    model="sentence-transformers/all-MiniLM-L6-v2",
    columns=["title", "abstract"],
    key="doi",
    modality="text",
)

Multiple embedding tables

Each call creates a new table. Multiple tables can coexist for the same source (different models, different columns):

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use jammi_ai::session::InferenceSession;
use jammi_db::store::CachePolicy;
async fn ex(session: &InferenceSession) -> jammi_db::error::Result<()> {
session.generate_text_embeddings("patents", "all-MiniLM-L6-v2", &["abstract".into()], "id", CachePolicy::Bypass, None).await?;
session.generate_text_embeddings("patents", "bge-small-en-v1.5", &["title".into()], "id", CachePolicy::Bypass, None).await?;
Ok(()) }
}

When searching, the latest ready embedding table is used by default.

Import precomputed embeddings

When the vectors already exist — computed by an offline batch, migrated from another store, or upserted from a remote encoder — register them directly as a ready embedding table instead of re-running the model. The input is a Parquet object with a _row_id (Utf8) column and a vector (FixedSizeList<Float32> of width dimensions) column, one row per key.

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use jammi_ai::session::InferenceSession;
async fn ex(session: &InferenceSession) -> jammi_db::error::Result<()> {
use jammi_db::storage::StorageUrl;

let vectors = StorageUrl::parse("file:///data/precomputed.parquet")?;
let record = session.import_embeddings(
    "patents",
    "sentence-transformers/all-MiniLM-L6-v2",
    &vectors,
    "id",
    &["abstract".to_string()],
    384,
).await?;

println!("Imported {} rows, {} dimensions", record.row_count, record.dimensions().unwrap());
Ok(()) }
}

Python

db.import_embeddings(
    source="patents",
    model="sentence-transformers/all-MiniLM-L6-v2",
    vectors_url="file:///data/precomputed.parquet",
    key="id",
    text_columns=["abstract"],
    dimensions=384,
)

The result is indistinguishable from a generated table — same (_row_id, _source_id, _model_id, vector) schema, same sidecar ANN index — so search queries it exactly like any other embedding table. Three behaviours are specific to import:

• Vectors are L2-normalized on import. Every embedding table holds unit vectors (the cosine ANN sidecar assumes it), so each incoming vector is normalized and a zero-norm vector is rejected — it cannot be cosine-searched.
• The model is validated, not loaded. model is parsed to its canonical form and recorded as the table’s derivation provenance; import never loads the encoder or downloads weights, so it needs no GPU. key and text_columns are recorded as catalog provenance (which source column the keys came from, which content columns produced the vectors); the physical key stays _row_id.
• The table is recompute-inert. The engine did not compute these vectors, so a recompute of an imported table is a typed refusal rather than a re-run guessed from its columns.

The input vectors are read fully into memory; a streaming variant is future work.

Supported models

Any encoder model on HuggingFace Hub with safetensors weights. Supported architectures:

BERT family — BERT, RoBERTa, DistilBERT, CamemBERT, XLM-RoBERTa:

• sentence-transformers/all-MiniLM-L6-v2 (384-dim, fast)
• sentence-transformers/all-mpnet-base-v2 (768-dim, higher quality)
• BAAI/bge-small-en-v1.5, BAAI/bge-base-en-v1.5

ModernBERT — modernized encoder with rotary embeddings, 8192-token context, GeGLU:

• answerdotai/ModernBERT-base (768-dim)
• answerdotai/ModernBERT-large (1024-dim)

Or any local directory with config.json + model.safetensors + tokenizer.json. The architecture is detected automatically from model_type in config.json.

Use a local model:

#![allow(unused)]
fn main() {
extern crate jammi_ai;
use jammi_ai::model::ModelSource;
let model = ModelSource::local("/path/to/my-model");
}

Pooling

Pooling — how per-token hidden states collapse into one sentence vector — is model-declared, not hardcoded. On load, the engine reads the model’s 1_Pooling/config.json (the sentence-transformers convention) and pools with the strategy it declares: pooling_mode_cls_token selects CLS pooling (first token — the mode BGE, GTE, and many E5-family models require), and pooling_mode_mean_tokens (or pooling_mode_mean_sqrt_len_tokens, which is exactly equivalent after the mandatory L2 normalization) selects mean pooling. Max and weighted-mean pooling are also supported.

A model whose repository ships no 1_Pooling/ directory — many bare BERT checkpoints — falls back to mean pooling, the historical sentence-transformers default. A model whose 1_Pooling/config.json declares a mode the engine cannot represent (e.g. last-token pooling, or more than one enabled mode at once) fails to load rather than silently pooling incorrectly.

Raw inference (no persistence)

To get embeddings as RecordBatch without writing to disk:

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use std::sync::Arc;
use jammi_ai::session::InferenceSession;
use jammi_db::store::CachePolicy;
async fn ex(session: &Arc<InferenceSession>) -> jammi_db::error::Result<()> {
use jammi_ai::model::{ModelSource, ModelTask};

let model = ModelSource::hf("sentence-transformers/all-MiniLM-L6-v2");
let (_results, _outcome) = session.infer("patents", &model, ModelTask::TextEmbedding, &["abstract".into()], "id", CachePolicy::Bypass).await?;
Ok(()) }
}

Python

results = db.infer(
    source="patents",
    model="sentence-transformers/all-MiniLM-L6-v2",
    columns=["abstract"],
    task="text_embedding",
    key="id",
)

Each RecordBatch has prefix columns (_row_id, _ordinal, _source, _model, _status, _error, _latency_ms) plus task-specific columns (e.g., vector for embeddings). _ordinal is a stream-scoped, 0-based row counter in model emission order; infer rows read back ordered by _row_id, _ordinal. The materialised embedding table generate_embeddings registers keeps only _row_id, _source_id, _model_id, vector — no _ordinal, since its _row_id is unique by construction.

Error handling

Inference never panics on bad input. _status/_error track per-row input validation, applied before the model ever runs:

The batch continues processing even when individual rows fail this validation. A null key is the one input fault that is never per-row: infer, generate_embeddings and an incremental refresh all refuse the whole call with InvalidKey { column, null_count } before any model call — _row_id is non-nullable by contract, so a silent drop would break the per-row disclosure this table promises. A model-forward failure itself — a broken kernel, a contiguity/PTX/dtype mismatch, or a model incapable of the requested task — is always systemic (every row fails identically), never a per-row event, so it fails the whole infer/embedding call with an error rather than being served as an all-"error" relation or an empty “ready” embedding table.

Dynamic batch sizing

The runner starts with the configured inference.batch_size (default: 32). If an out-of-memory error occurs:

1. Halve the batch size
2. Retry (up to 3 times)
3. If OOM persists at batch size 1, the call fails with an error

The reduced batch size is sticky for the remainder of the stream.

Crash recovery

If the process dies mid-generation, the table is left in “building” status. On the next session start, recovery runs automatically:

• Parquet missing — mark as failed
• Parquet corrupt — delete file, mark as failed
• Parquet valid but stuck in “building” — promote to “ready”, rebuild ANN index

No data is lost if the Parquet file was fully written.

DataFusion integration

Result tables are automatically registered in DataFusion and queryable via SQL:

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use jammi_ai::session::InferenceSession;
use jammi_db::catalog::result_repo::ResultTableRecord;
async fn ex(session: &InferenceSession, record: &ResultTableRecord) -> jammi_db::error::Result<()> {
let results = session.sql(&format!(
    "SELECT _row_id, _source_id FROM \"jammi.{}\" LIMIT 10",
    record.table_name
)).await?;
Ok(()) }
}

Use a Local Model Checkpoint

Every model-accepting argument in the engine — generate_embeddings’ model, fine-tune’s base_model, annotate()’s first argument, the Python Session API — takes a model reference string. A reference that names a filesystem location loads the checkpoint from local disk, with no Hub access and no network: this is the supported path for air-gapped hosts and for checkpoints you have already downloaded or trained elsewhere.

Reference forms

ModelSource::parse recognizes four spellings; the first three are local:

A local path is resolved against the filesystem of the host running the engine — for a remote client that is the server, not the client machine — so the directory must exist there.

What the directory must contain

• A config: config.json (or open_clip_config.json for OpenCLIP models).
• Weights: model.safetensors (or open_clip_model.safetensors), model.onnx, and/or model.gguf.

When both safetensors and ONNX weights are present, the ONNX file wins and the model runs on the ORT backend; safetensors alone selects the Candle backend. An explicit backend hint overrides this choice. model.gguf is considered only when neither safetensors nor ONNX weights are present — see Quantized (GGUF) checkpoints below.

Resolution is fail-loud: a nonexistent directory, a directory with no config, and a directory with no recognized weights file each produce a typed error naming what is missing — there is no silent fallback to the Hub.

Quantized (GGUF) checkpoints

A directory with no model.safetensors/model.onnx but a model.gguf loads on the Candle backend as a quantized checkpoint — matmul-site weights stored at a k-quant format (q4_0 through q6k) stay resident in that compressed form; everything else (embeddings, norms, classifier/NER heads, and any matmul-site weight that happens to be stored densely) is dequantized to the model’s compute dtype at load.

Requirements and limits:

• The weights file must be named exactly model.gguf — any other *.gguf filename in the directory produces a typed error naming the convention.
• config.json is still required, and still the source of the model’s architecture and layer count — a GGUF file’s own embedded metadata (the convention some GGUF exporters use in place of a sidecar config.json) is never read for this. A directory with a model.gguf but no config.json is not a supported checkpoint shape.
• Supported architectures: BERT and its variants (RoBERTa, CamemBERT, XLM-RoBERTa), DistilBERT, and ModernBERT. Any other architecture (OpenCLIP, CLAP) is a typed refusal, not a best-effort load.
• Every tensor in the file must be a supported k-quant format (q4_0, q4_1, q5_0, q5_1, q8_0, q2k, q3k, q4k, q5k, q6k) or stored densely as f32/f16/bf16 — any other GGML dtype is a typed refusal.

base_model in fine-tuning accepts a model.gguf checkpoint the same way: an encoder-adapters LoRA job trains its low-rank adapters over the frozen quantized backbone automatically when the resolved base is GGUF — there is no separate QLoRA flag or config field.

Examples

Embeddings over a local checkpoint (Python):

db.generate_embeddings(
    source="patents",
    model="local:/models/all-MiniLM-L6-v2",
    columns=["abstract"],
    key="id",
    modality="text",
)

Fine-tune from a local base model (see Fine-Tune for Your Domain):

db.fine_tune(
    base_model="local:/models/bioclinical-modernbert-large",
    ...
)

SQL, over Flight SQL (see Compound Retrieval and Inference):

SELECT * FROM annotate('local:/models/all-MiniLM-L6-v2', 'text_embedding',
                       ARRAY['abstract'])

Model registration — a durable catalog entry with an id, stage transitions, and evidence — is a platform concern and lives outside the OSS engine; the engine consumes local checkpoints directly through the reference forms above.

Configuring the Hub cache, endpoint, and token

Every other spelling (a bare repo id, hf://owner/repo) resolves against the Hugging Face Hub through one client built once from [models]:

[models]
hub_endpoint = "https://huggingface.co"
hub_cache_dir = "/var/cache/jammi"
hub_token = { file = "/run/secrets/hf-token" }
offline = false

Every field is also settable through the standard JAMMI_MODELS__HUB_* / JAMMI_MODELS__OFFLINE env-override layer (e.g. JAMMI_MODELS__HUB_CACHE_DIR, JAMMI_MODELS__OFFLINE=true) — that tier sits ABOVE the HF_* fallbacks below: a JAMMI_* override behaves exactly like the equivalent TOML key, one precedence step above HF_HUB_CACHE/HF_HOME/HF_ENDPOINT/HF_TOKEN/ HF_HUB_OFFLINE/TRANSFORMERS_OFFLINE, not alongside them.

Empty values are absent, and every value is trimmed: every HF_* environment variable above (HF_HUB_CACHE, HF_HOME, HF_ENDPOINT, HF_TOKEN, HUGGING_FACE_HUB_TOKEN, HF_TOKEN_PATH, HF_HUB_OFFLINE, TRANSFORMERS_OFFLINE) is treated as unset when it is present but empty (after trimming whitespace) — the shape a Compose/Kubernetes env block produces for a variable named with no value (HF_HUB_OFFLINE:), or a shell export HF_TOKEN=. Without this rule, an empty HF_HUB_OFFLINE would shadow the TRANSFORMERS_OFFLINE alias and silently resolve online. The value used downstream is also the TRIMMED string, not the raw one — HF_HOME=" /data/hf" resolves to the cache root /data/hf/hub, never a current-working-directory-relative root the padded, untrimmed value would otherwise produce. One case is a genuine divergence, not merely a stricter reading of upstream, and it fails in opposite directions: a whitespace-only HF_HUB_OFFLINE=" " falls through to TRANSFORMERS_OFFLINE here (failing toward offline, the safe direction), where huggingface_hub itself stops at the whitespace-only value and resolves online.

No home directory, no HF_HUB_CACHE, no HF_HOME, and no hub_cache_dir is a typed JammiError::Config at session construction — never a panic. A token that resolves from none of the four tiers sends no Authorization header, the same as an anonymous huggingface-cli session. The token FILE is HF_TOKEN_PATH, when set, naming the file directly; otherwise <HF_HOME>/token, even when HF_HUB_CACHE wins the cache-root precedence above — HF_HUB_CACHE names the cache directory directly, with no hub path component to derive HF_HOME back out of, so the token lookup reads HF_TOKEN_PATH/HF_HOME on their own rather than working backwards from the cache root.

offline = true refuses every Hub network fetch: a HuggingFace-sourced model loads only when it resolves against local: or an already-populated catalog row — a warm, on-disk Hub cache directory with no matching catalog row is still a miss, because the catalog (not the cache) is offline’s source of truth. It does not reach the fine-tune worker’s adapter fetch for an already-trained model, which always reads the adapter bundle from the artifact store, offline or not. An explicit offline = false wins over HF_HUB_OFFLINE/TRANSFORMERS_OFFLINE in the environment (and vice versa for offline = true) — only an OMITTED offline key falls back to either environment variable at all.

Every HF_*/HF_HUB_OFFLINE/TRANSFORMERS_OFFLINE fallback above is read from the server process’s own environment, at session construction — never from a remote client’s environment (a gRPC/Flight SQL/Python client connecting to a running jammi-server has no way to influence these; only that server process’s own [models]/env decides resolution).

A configured hub_endpoint/mirror is not part of a resolved model’s identity: two endpoints can serve different bytes for the same repo id (a stale or divergent mirror), so once a model resolves, its catalog artifact_path pins the actual bytes fetched — re-resolving under a different endpoint later never silently swaps them out from under an already-registered model.

Generate Image Embeddings

Generate vector embeddings from images using an OpenCLIP-compatible vision model. Results are persisted to Parquet with sidecar ANN indexes, identical to text embeddings — the same search(), evaluation, and SQL tools work on both.

The OpenCLIP family is cross-modal: the vision tower and the text tower in the same checkpoint produce embeddings in a shared latent space, so a text query encoded with the same model can search image embeddings directly. See Search Text Against Images (Cross-Modal) for the full text-to-image recipe.

Basic usage

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use jammi_ai::session::InferenceSession;
use jammi_db::store::CachePolicy;
async fn ex(session: &InferenceSession) -> jammi_db::error::Result<()> {
let (record, _outcome) = session.generate_image_embeddings(
    "figures",
    "laion/CLIP-ViT-B-32-laion2B-s34B-b79K",
    "image",       // column containing image data
    "figure_id",   // key column
    CachePolicy::Bypass,
    None,
).await?;

println!("Embedded {} images, {} dimensions", record.row_count, record.dimensions().unwrap());
Ok(()) }
}

Python

db.generate_embeddings(
    source="figures",
    model="laion/CLIP-ViT-B-32-laion2B-s34B-b79K",
    columns=["image"],
    key="figure_id",
    modality="image",
)

Image column format

The image column can be either:

• Binary / LargeBinary / BinaryView — inline image bytes (PNG, JPEG, TIFF) stored directly in Parquet
• Utf8 / LargeUtf8 / Utf8View — file paths pointing to images on disk. Utf8View is what a plain Parquet Utf8 column becomes through a DataFusion scan under this workspace’s pinned Arrow/DataFusion versions, so an ordinary registered source’s path column takes this arm on the real, unmodified scan output — not just a hand-built array (arrow_to_images, crates/jammi-ai/src/inference/mod.rs:210).

Image preprocessing

Each image is automatically preprocessed before embedding:

1. Pad to square — white canvas, image centered (preserves aspect ratio)
2. Resize — bicubic interpolation to the model’s input size (224x224 for CLIP)
3. Normalize — per-channel normalization using constants from the model’s config

Preprocessing parameters (mean, std, image size) are model-driven — parsed from the model’s config file, not hardcoded.

Encode a single image

To embed one image without persistence (e.g., for a query):

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use jammi_ai::session::InferenceSession;
async fn ex(session: &InferenceSession) -> Result<(), Box<dyn std::error::Error>> {
let image_bytes = std::fs::read("query.png")?;
let vector = session
    .encode_image_query("laion/CLIP-ViT-B-32-laion2B-s34B-b79K", &image_bytes)
    .await?;
// vector: Vec<f32>, L2-normalized, dimensionality = model's embed_dim
Ok(()) }
}

Python

with open("query.png", "rb") as f:
    image_bytes = f.read()

vector = db.encode_query(model="laion/CLIP-ViT-B-32-laion2B-s34B-b79K", query=image_bytes, modality="image")

Supported models

OpenCLIP-compatible models with safetensors weights. The repo must carry:

• open_clip_config.json with model_cfg.vision_cfg (and model_cfg.text_cfg if you want cross-modal text queries)
• open_clip_model.safetensors with OpenCLIP weight key naming (visual.* for vision, root-level for text)
• Either a tokenizer.json or the OpenCLIP-native bpe_simple_vocab_16e6.txt.gz (only required for text-side queries)

The architecture (ViT width, layers, heads, patch size, pooling strategy), the shared latent dimensionality (embed_dim), and the preprocessing config (mean, std, image size) are detected automatically from the config — no per-model code path.

Known-working models include patentclip/PatentCLIP_Vit_B (512-dim, tuned for technical drawings or diagrams rather than photographs — uses global average pooling), OpenAI CLIP, and the LAION CLIP-ViT-B-32-* variants. For an end-to-end image-to-image search + retrieval-eval walkthrough, see the runnable image_search recipe.

Output schema

Same as text embeddings:

Search

Image embeddings work with the same search() API as text embeddings:

vector = db.encode_query(model="laion/CLIP-ViT-B-32-laion2B-s34B-b79K", query=query_bytes, modality="image")
results = db.search("figures", query=vector, k=10)  # pyarrow.Table

search returns a pyarrow.Table directly; for compound retrieval (join / annotate(...)) use db.sql(...).

Error handling

Search Text Against Images (Cross-Modal)

OpenCLIP-family models carry both a vision tower and a text tower in the same checkpoint, with both towers projecting into a shared latent space. That means a text query embedded with the text tower lives in the same vector space as image embeddings produced by the vision tower — vector search against an image corpus accepts a text query directly, no separate text encoder, no projection bridge.

This recipe shows the full path: index images with the vision tower, embed a text query with the text tower, run search().

1. Index the image corpus with the vision tower

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use jammi_ai::session::InferenceSession;
use jammi_db::store::CachePolicy;
async fn ex(session: &InferenceSession) -> jammi_db::error::Result<()> {
session.generate_image_embeddings(
    "figures",
    "laion/CLIP-ViT-B-32-laion2B-s34B-b79K",
    "image",       // column containing image data
    "figure_id",   // key column
    CachePolicy::Bypass,
    None,
).await?;
Ok(()) }
}

Python

db.generate_embeddings(
    source="figures",
    model="laion/CLIP-ViT-B-32-laion2B-s34B-b79K",
    columns=["image"],
    key="figure_id",
    modality="image",
)

2. Embed a text query with the same model’s text tower

encode_query dispatches to the OpenCLIP text tower when the model ID resolves to an OpenCLIP checkpoint. The output vector dimensionality matches embed_dim — the same dim the image embeddings carry.

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use jammi_ai::session::InferenceSession;
async fn ex(session: &InferenceSession) -> jammi_db::error::Result<()> {
let query_vec = session
    .encode_text_query(
        "laion/CLIP-ViT-B-32-laion2B-s34B-b79K",
        "a red circle on a white background",
    )
    .await?;
// query_vec: Vec<f32>, L2-normalized, same length as the image embedding vector
Ok(()) }
}

Python

query_vec = db.encode_query(model="laion/CLIP-ViT-B-32-laion2B-s34B-b79K", query="a red circle on a white background",)

3. Search image embeddings with the text vector

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use std::sync::Arc;
use jammi_ai::session::InferenceSession;
async fn ex(session: Arc<InferenceSession>, query_vec: Vec<f32>) -> jammi_db::error::Result<()> {
let results = session.search("figures", query_vec, 10, None, None).await?.run().await?;
Ok(()) }
}

Python

results = db.search("figures", query=query_vec, k=10)  # pyarrow.Table

search returns a pyarrow.Table directly, carrying your source’s columns (figure_id) alongside the similarity score; pass filter= / select= to refine. For compound retrieval — joining sources or running a model over the results with annotate(...) — use db.sql(...) (see Compound Retrieval and Inference over Flight SQL); it composes identically for cross-modal search.

Why this works

Both towers in an OpenCLIP checkpoint emit vectors of size embed_dim (the shared latent dimensionality declared at the top of open_clip_config.json). The vision tower applies a visual.proj matrix after pooling its patch tokens; the text tower applies a text_projection matrix after pooling at the <|endoftext|> token. The two projections are jointly trained so the cosine similarity between a text vector and an image vector reflects semantic alignment.

If you embed text and images with separate models (e.g. a BERT encoder + a vision model that wasn’t jointly trained with it), the resulting vectors don’t share a latent space and the similarities are meaningless. Cross-modal search only works when both modalities are projected by the same CLIP-style joint training.

Model requirements

Same as Generate Image Embeddings, plus:

• open_clip_config.json must contain a populated model_cfg.text_cfg (with width, layers, and either heads or a width that is a multiple of 64).
• The safetensors checkpoint must contain the text-tower keys: token_embedding.weight, positional_embedding, transformer.resblocks.*, ln_final.*, and text_projection.
• A tokenizer must be available — either an HF-converted tokenizer.json or the OpenCLIP-native bpe_simple_vocab_16e6.txt.gz.

Classify Text

Run a classification model over text columns to assign labels and confidence scores. Any HuggingFace model with id2label in its config works out of the box.

Basic usage

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use std::sync::Arc;
use jammi_ai::session::InferenceSession;
use jammi_db::store::CachePolicy;
async fn ex(session: &Arc<InferenceSession>) -> jammi_db::error::Result<()> {
use jammi_ai::model::{ModelSource, ModelTask};

let model = ModelSource::hf("answerdotai/ModernBERT-base-classification");
let (_results, _outcome) = session.infer(
    "patents",
    &model,
    ModelTask::Classification,
    &["abstract".to_string()],
    "id",
    CachePolicy::Bypass,
).await?;
Ok(()) }
}

Python

results = db.infer(
    source="patents",
    model="answerdotai/ModernBERT-base-classification",
    columns=["abstract"],
    task="classification",
    key="id",
)

Output schema

Each RecordBatch has prefix columns plus classification-specific columns:

Supported model architectures

Classification models must have id2label in their config.json. Supported architectures:

BERT family — BERT, RoBERTa, DistilBERT, CamemBERT, XLM-RoBERTa:

• Loads classifier.weight + classifier.bias from safetensors
• CLS token pooling + linear classifier + softmax

ModernBERT — uses the built-in ModernBertForSequenceClassification:

• CLS or MEAN pooling (configured via classifier_pooling in config)
• Head (dense + GELU + LayerNorm) + classifier + softmax

Fine-tuning for classification

Train a LoRA adapter with a classification head on your labeled data:

Prepare training data

text,label
"quantum error correction","physics"
"CRISPR gene editing","biology"

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use jammi_ai::session::InferenceSession;
async fn ex(session: &InferenceSession) -> jammi_db::error::Result<()> {
use jammi_ai::fine_tune::FineTuneMethod;
use jammi_db::ModelTask;

let job = session.fine_tune(
    "training",
    "sentence-transformers/all-MiniLM-L6-v2",
    &["text".into(), "label".into()],
    FineTuneMethod::Lora,
    ModelTask::Classification,
    None,
).await?;

job.wait().await?;
Ok(()) }
}

Python

job = db.fine_tune(
    source="training",
    base_model="sentence-transformers/all-MiniLM-L6-v2",
    columns=["text", "label"],
    method="lora",
    task="classification",
)
job.wait()

The fine-tuned model trains a LoRA projection plus a linear classification head using cross-entropy loss. Both are saved to adapter.safetensors.

Error handling

Same per-row error tracking as embeddings:

Extract Entities (NER)

Run a Named Entity Recognition model over text columns to extract person names, organizations, locations, and other entities. Results are returned as JSON arrays of entity spans with character positions and confidence scores.

Basic usage

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use std::sync::Arc;
use jammi_ai::session::InferenceSession;
use jammi_db::store::CachePolicy;
async fn ex(session: &Arc<InferenceSession>) -> jammi_db::error::Result<()> {
use jammi_ai::model::{ModelSource, ModelTask};

let model = ModelSource::hf("dslim/bert-base-NER");
let (_results, _outcome) = session.infer(
    "patents",
    &model,
    ModelTask::Ner,
    &["abstract".to_string()],
    "id",
    CachePolicy::Bypass,
).await?;
Ok(()) }
}

Python

results = db.infer(
    source="patents",
    model="dslim/bert-base-NER",
    columns=["abstract"],
    task="ner",
    key="id",
)

Output schema

Entity span format

Each entity in the JSON array has:

{
  "text": "Google",
  "label": "ORG",
  "start": 15,
  "end": 21,
  "confidence": 0.97
}

Supported models

NER models must have id2label with BIO-tagged labels (e.g., B-PER, I-PER, O) in their config.json.

BERT family — loads classifier.weight + classifier.bias on top of the encoder:

• dslim/bert-base-NER (English, 4 entity types)
• dbmdz/bert-large-cased-finetuned-conll03-english

ModernBERT — same pattern, modern encoder architecture.

How it works

text → tokenize (with character offsets)
     → encoder forward → hidden states [batch, seq_len, hidden]
     → Linear(hidden, num_labels) per token → logits
     → softmax → argmax → BIO tag per token
     → merge consecutive B-/I- tags into entity spans
     → map character offsets back to original text

The BIO decoding handles:

• B-TYPE: starts a new entity of that type
• I-TYPE: continues the current entity (must match type)
• O: outside any entity
• Special tokens ([CLS], [SEP], padding) are automatically skipped

Conformal Prediction: Distribution-Free Coverage

Measured companion: for the long-form, executed-and-measured Python treatment, see The Cookbook → Conformal Prediction.

Conformal prediction wraps any existing predictor and turns a point output into a prediction set (classification) or interval (regression) carrying a finite-sample, distribution-free coverage guarantee. Given a held-out calibration set, the marginal coverage of the emitted sets is at least 1 − alpha under exchangeability — for any underlying model, any data distribution, and any sample size. No retraining: a calibration pass and an empirical quantile. Deterministic given the calibration set, which is the audit property.

The serving primitive lives in the open engine because a calibrated set is a serving output — it must work with no license. The operationalization of the guarantee — rolling coverage monitoring, coverage-SLA gating, and managed recalibration under drift — is a governed concern provided by the Jammi platform, built on this same primitive.

The one assumption

The guarantee holds if and only if the calibration and serving data are exchangeable. Under distribution drift it degrades silently. Two levers correct for known structure:

• Weighted conformal applies importance weights for a known covariate shift.
• Mondrian conformal keeps a per-cohort quantile keyed on a group column, the principled approximation to conditional coverage (full per-input coverage is provably impossible distribution-free).

The primitive applies the weights and the grouping; detecting drift and choosing the cohorts is governance, not a serving output.

The three-way split

Reusing test points to calibrate inflates coverage. The calibration set must be disjoint from both the training set and the test/serving data. The calibration source is a distinct argument throughout the API.

Classification: prediction sets

The classification scores read the per-class softmax mass the classifier already emits.

• LAC — nonconformity 1 − p_y; the smallest sets at the nominal level, but non-adaptive.
• APS (default) — the cumulative mass of classes ranked most- to least-probable up to the true class; set size adapts to input difficulty.
• RAPS — APS plus a tail-rank penalty that shrinks sets on easy inputs.

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
fn ex() -> jammi_db::error::Result<()> {
use jammi_ai::predict::{ClassScore, ConformalModel};

// Held-out calibration: per-class probabilities + the realised class index.
let calibration: Vec<Vec<f64>> = vec![
    vec![0.7, 0.2, 0.1],
    vec![0.1, 0.8, 0.1],
    // ... one row per calibration example
];
let true_labels: Vec<usize> = vec![0, 1 /* ... */];

// Calibrate at 90% nominal coverage with the adaptive APS score.
let model = ConformalModel::classification(&calibration, &true_labels, ClassScore::Aps, 0.1)?;

// Serving: emit the prediction set for a new row of class probabilities.
let probabilities = vec![0.45, 0.4, 0.15];
let prediction_set = model.predict_set(&probabilities, None)?; // e.g. [0, 1]
let _ = prediction_set;
Ok(()) }
}

Python

sets = db.conformalize(
    calibration=[[0.7, 0.2, 0.1], [0.1, 0.8, 0.1]],  # per-class probabilities
    true_labels=[0, 1],                               # realised class indices
    test=[[0.45, 0.4, 0.15]],                         # rows to predict
    alpha=0.1,
    score="aps",                                      # "lac" | "aps" | "raps"
)
# sets -> [[0, 1]]  (one list of admitted class indices per test row)

Regression: prediction intervals

• Absolute residual — nonconformity |y − ŷ|; a constant-width interval [ŷ − q̂, ŷ + q̂]. Distribution-free but uninformative under heteroscedasticity.
• CQR (Conformalized Quantile Regression) — nonconformity max(q_lo − y, y − q_hi) over a predictor’s lower/upper quantile estimates; an adaptive-width interval whose width tracks local uncertainty.

#![allow(unused)]
fn main() {
extern crate jammi_ai;
extern crate jammi_db;
fn ex() -> jammi_db::error::Result<()> {
use jammi_ai::predict::{ConformalModel, IntervalScore};

// Absolute-residual conformal over held-out (prediction, observation) pairs.
let predictions = vec![1.0, 2.0, 3.0 /* ... */];
let observed = vec![1.2, 1.7, 3.1 /* ... */];
let model = ConformalModel::regression(
    &predictions,
    &[],   // lower quantiles (CQR only)
    &[],   // upper quantiles (CQR only)
    &observed,
    IntervalScore::AbsoluteResidual,
    0.1,
)?;

// Serving: a 90% interval around a new point estimate.
let (lower, upper) = model.predict_interval(2.5, 0.0, 0.0, None)?;
let _ = (lower, upper);
Ok(()) }
}

Python

# Constant-width absolute-residual intervals.
intervals = db.conformalize_interval(
    predictions=[1.0, 2.0, 3.0],   # calibration point estimates
    observed=[1.2, 1.7, 3.1],      # calibration targets
    test_predictions=[2.5],        # point estimates to bound
    alpha=0.1,
)
# intervals -> [(lower, upper)]

# Adaptive-width CQR intervals from quantile estimates.
intervals = db.conformalize_cqr(
    lower=[0.5, 1.5, 2.5],         # calibration lower-quantile estimates
    upper=[1.5, 2.5, 3.5],         # calibration upper-quantile estimates
    observed=[1.2, 1.7, 3.1],
    test_lower=[2.0],
    test_upper=[3.0],
    alpha=0.1,
)

The finite-sample quantile

The conformal threshold is the ⌈(n+1)(1 − alpha)⌉-th smallest calibration score, not the naive ⌈n(1 − alpha)⌉ order statistic. The (n + 1) correction is what makes the guarantee exact rather than merely asymptotic; the naive quantile leaves a ~1/n coverage gap and under-covers. When the calibration set is too small for the requested level — fewer than ⌈1/alpha⌉ − 1 points — the threshold is +∞: the honest, conservative answer is “every label”. A full set is a real signal that the input is hard or the base model is miscalibrated, not a bug.

The conformal evidence channel

Conformal outputs ride the evidence substrate exactly as vector and inference do — one channel, four declared columns, no new provenance machinery:

Register the channel once, then attach a contribution per result batch:

#![allow(unused)]
fn main() {
extern crate jammi_ai;
extern crate jammi_db;
async fn ex(catalog: &jammi_db::catalog::Catalog) -> jammi_db::error::Result<()> {
use jammi_ai::evidence::conformal::{channel_spec, contribution, ConformalOutput};

catalog.channels().register(&channel_spec()?).await?;

let _contrib = contribution(&[
    ConformalOutput::Set { classes: vec![0, 2], alpha: 0.1 },
    ConformalOutput::Interval { lower: -1.0, upper: 1.0, alpha: 0.1 },
])?;
// `_contrib` merges into result batches via `merge_channels`.
Ok(()) }
}

Verifying coverage

The realised coverage and mean set size of a labelled batch are pure functions in jammi-numerics — the same functions the platform’s coverage monitor calls on a rolling window:

#![allow(unused)]
fn main() {
extern crate jammi_numerics;
fn ex() -> Result<(), jammi_numerics::error::NumericsError> {
use jammi_numerics::calibration::{coverage, mean_set_size};

let hits = [true, true, false, true];     // did each set contain the true label?
let sizes = [2usize, 1, 3, 2];            // cardinality of each set
let realised = coverage(&hits)?;          // ~ 1 - alpha when calibrated
let efficiency = mean_set_size(&sizes)?;  // smaller is sharper
let _ = (realised, efficiency);
Ok(()) }
}

What lives in the platform, not here

This is the serving primitive only. The governed layer — a rolling realised-vs-nominal coverage monitor with drift detection and online adaptation, a coverage-SLA gate, and managed recalibration under shift — is provided by the Jammi platform. It consumes this primitive and the OSS coverage function; it is not part of the open engine.

Distributional Inference: Predict a Distribution, Not a Point

A ModelTask::Regression head returns a predictive distribution per row — a Gaussian (predicted_mean, predicted_std) or a set of quantiles — instead of a single number. Where conformal prediction wraps any predictor with a distribution-free interval, a regression head is trained to emit calibrated uncertainty directly, with proper-scoring objectives that make that uncertainty honest.

Two output forms, both standard:

• Parametric Gaussian — the head predicts μ and a raw scale; serving maps the scale to a positive σ = floor + softplus(raw). Smooth, cheap, a closed- form density. The default.
• Quantile — the head predicts a set of levels (e.g. 0.05, 0.5, 0.95) directly. Distribution-free in shape, robust to non-Gaussian outcomes, and the input to conformal CQR. The serving adapter sorts each row’s quantiles so they never cross.

Choose the objective by your label

The four regression objectives are all proper scores — minimising them rewards a calibrated distribution, not merely an accurate mean. (MSE on the predicted mean is not proper for a distribution and is only a secondary point- accuracy diagnostic.)

• β-NLL (default) — Seitzer’s variance-weighted Gaussian NLL. The plain joint μ,σ² NLL has a well-documented pathology: it down-weights high-error points by inflating their variance, starving the mean’s gradient and collapsing to overconfidence elsewhere. β-NLL re-weights each row’s NLL by a detached σ^{2β}, restoring the mean’s gradient and removing the collapse. β = 0.5 is the recommended default.
• CRPS — the closed-form Gaussian continuous ranked probability score, the other collapse-resistant choice: strictly proper, in the outcome’s units, and far more stable under joint μ,σ² training than NLL.
• Gaussian NLL — the classic mean-variance objective, provided for completeness and as the pathology baseline. Prefer β-NLL or CRPS.
• Pinball — the quantile objective; trains each quantile to its level, with a non-crossing penalty that discourages crossing during training.

The same CRPS / NLL math headlines the calibration eval — one source of truth for the score, used as both the training loss and the eval metric.

Calibrated, not merely accurate

A regression head is not done until its coverage is verified. Two models can share a mean (identical MSE) yet one be badly miscalibrated. The calibration eval is the gate: the central interval should cover at ≈ its nominal level, and the head’s proper score (CRPS/NLL) should beat a constant-variance baseline. Verify coverage; never ship on NLL alone.

Aleatoric, not epistemic

A parametric Gaussian head models aleatoric (irreducible data) noise only. It does not know what it has not seen: off-distribution it can be confidently wrong. For uncertainty about the unseen, reach for the rest of the spectrum:

• distribution-free coverage with no model assumption → conformal prediction;
• amortized epistemic posteriors → a future neural-process head.

Do not read this head’s σ as epistemic.

The uncertainty evidence channel

A served distribution rides the uncertainty evidence channel (predicted_mean, predicted_std, quantiles, context_ref) — the same additive substrate as vector, inference, and conformal. When the prediction was conditioned on an assembled context set, the channel records which rows informed it in context_ref — data-driven provenance applied to prediction. Register it like any custom channel (see Declare a Custom Provenance Channel); the distribution columns then merge into the result and are SQL-reachable.

Train an In-Context Predictor (Amortized, Adapts Without Retraining)

Measured companion: for the long-form, executed-and-measured Python treatment, see The Cookbook → Context-Conditioned Prediction.

An in-context predictor meta-learns to turn a context set — a target’s retrieved neighbours and their outcomes — into a predictive distribution, in one forward pass with no gradient update at inference. Trained once over many tasks, it adapts to a new target’s neighbourhood the way a prior-fitted network does: condition on the context, emit the posterior, move on. It is the learned-aggregation point of the uncertainty spectrum, above the cheaper distribution-free and parametric options.

Three curated architectures, selected by config (never authored as tensor ops):

• CNP — a DeepSets encoder that mean-pools the context, then a decoder MLP. The baseline; the learned twin of fixed pooling.
• Attentive CNP (attncnp) — attention pooling over the context, so the target query weights the neighbours that matter. The payoff over fixed pooling, and the member that widens its uncertainty when its context is thin or unfamiliar (epistemic uncertainty).
• TNP — a transformer over the (context ∪ target) token set; the strongest member, the prior-fitted-network point.

When to reach for it — the spectrum

The substrate offers three honest-uncertainty tools. Pick the cheapest one that covers your need:

The in-context predictor is the heaviest and most expressive. It does not replace the other two — an amortized posterior is sharp but not automatically calibrated, so it is wrapped by conformal for a coverage guarantee (below). Conformal remains the deterministic, audit-reproducible option; this predictor is what a continual, adapt-per-target setting reaches for.

Train

Training is episodic: each task (the distinct values of a task column — a cohort, a time window, a source partition) is split into a context set and held-out targets, and the target’s outcome is scored under a proper objective. Tasks — not points — are partitioned into train/test, so generalisation is measured on held-out tasks. The target is never in its own context (self-exclusion plus a same-task split), and a meta-dataset with too few tasks is rejected rather than meta-trained into memorisation.

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
extern crate jammi_encoders;
use std::sync::Arc;
use jammi_ai::session::InferenceSession;
use jammi_ai::pipeline::context_predictor::{
    ContextArchitecture, ContextPredictorTrainConfig, GaussianObjective, PredictiveHead,
};
async fn ex(session: &Arc<InferenceSession>) -> jammi_db::error::Result<()> {
let spec = ContextPredictorTrainConfig {
    model_id: "patents-context-predictor".into(),
    architecture: ContextArchitecture::AttnCnp, // CNP | AttnCnp | TNP, by config
    key_column: "_row_id".into(),               // the per-row identity
    task_column: "cohort".into(),               // distinct values = the tasks
    value_column: "outcome".into(),             // the scalar y to regress
    context_k: 32,                              // retrieval / context size
    hidden_dim: 64,
    num_heads: 4,
    num_layers: 2,
    head: PredictiveHead::Gaussian {            // an S18 head + proper score
        objective: GaussianObjective::Crps,
    },
    epochs: 100,
    learning_rate: 0.005,
    grad_clip: 1.0,
    test_task_fraction: 0.2,                    // tasks held out for eval
    min_task_count: 4,                          // the meta-overfitting guard
    seed: 0,
};
// Training is a durable, lease-claimed job: `train_context_predictor` submits a
// queued job and returns a handle immediately; a worker claims it, re-samples
// the episodic meta-dataset from the spec, trains it, and registers the model.
let job = session.train_context_predictor("patents", &spec).await?;
job.wait().await?; // block until a worker drives the job to completion
let model_id = job.model_id(); // the spec's `model_id`, now registered
let _ = model_id;
Ok(()) }
}

The objective is one of the proper scores the distributional head uses — no new loss code. A PredictiveHead::Gaussian serves (mean, std); a PredictiveHead::Quantile serves a non-crossing set of quantile levels.

Predict — adapt to a new target, no retraining

Predicting assembles the target’s live context (the serving corpus, with the target excluded) and runs one in-context forward. There is no optimizer and no weight update — the adaptation lives entirely in the forward.

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use std::sync::Arc;
use jammi_ai::session::InferenceSession;
use jammi_ai::pipeline::context_predictor::{ContextServeOptions, PredictedDistribution};
async fn ex(session: &Arc<InferenceSession>) -> jammi_db::error::Result<()> {
// Reload the trained predictor for inference, serving over a corpus.
// `ContextServeOptions::default()` is embedding-similarity (ANN) context with no
// serving split — pass an edge-bearing source to condition on declared edges.
let served = session
    .load_context_predictor(
        "patents-context-predictor",
        "patents",
        ContextServeOptions::default(),
    )
    .await?;

// One forward over the target's live context — no gradient update.
let dist = session
    .predict_with_context_predictor(&served, "US-7654321")
    .await?;
match dist {
    PredictedDistribution::Gaussian { mean, std } => {
        let _ = (mean, std);
    }
    PredictedDistribution::Quantile { levels } => {
        let _ = levels; // ascending (level, value) pairs
    }
}
Ok(()) }
}

The serving source need not be the training source — a predictor meta-trained on one corpus serves a target’s neighbourhood in another of the same shape (the inductive prior-fitted-network property). An optional split predicate scopes the serving context.

Wrap with conformal for a coverage guarantee

An amortized posterior is sharp but can be overconfident off its training tasks — its raw interval under-covers. Calibrate a conformal wrap on a held-out calibration set (tasks disjoint from training) and the served interval recovers its nominal coverage. A Gaussian head wraps with absolute-residual conformal over its mean; a quantile head wraps with CQR over its lower/upper quantiles.

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use std::sync::Arc;
use jammi_ai::session::InferenceSession;
use jammi_ai::pipeline::context_predictor::{ConformalLevers, ServedContextPredictor};
async fn ex(
    session: &Arc<InferenceSession>,
    served: &ServedContextPredictor,
    held_out: &[(String, f64)], // (target_key, observed y), tasks disjoint from training
) -> jammi_db::error::Result<()> {
// Calibrate at 90% nominal coverage on the held-out set. `Marginal` is plain
// split-conformal; governance may instead supply a Mondrian cohort or weights.
let wrap = session
    .calibrate_context_predictor_conformal(served, held_out, 0.1, ConformalLevers::Marginal)
    .await?;

// Serving: turn a prediction into a coverage-guaranteed interval. The optional
// group is the test point's Mondrian cohort (`None` for a marginal wrap).
let dist = session
    .predict_with_context_predictor(served, "US-7654321")
    .await?;
let (lower, upper) = wrap.interval(&dist, None)?;
let _ = (lower, upper);
Ok(()) }
}

Epistemic uncertainty — and its honest caveat

The attentive members (attncnp, tnp) widen their predicted uncertainty when a target’s context is sparse or unfamiliar — the property a fixed distributional head lacks. This is primarily an attention property: a plain CNP’s mean-pool barely widens its σ as the context thins (it conditions on the context size rather than reweighting members), so reach for attncnp or tnp when epistemic widening matters. If you only need aleatoric noise on a continuous outcome, the cheaper distributional head is the right tool.

From Python

# Submits a durable training job and returns its handle; an embedded worker
# runs it. Block on `.wait()`, then read `.model_id` for the registered model.
job = db.train_context_predictor(
    "patents",
    key_column="_row_id",
    task_column="cohort",
    value_column="outcome",
    architecture="attncnp",   # "cnp" | "attncnp" | "tnp"
    output="gaussian",        # or "quantile" with levels=[0.1, 0.5, 0.9]
    objective="crps",         # "crps" | "nll" | "betanll"
    context_k=32,
)
job.wait()
model_id = job.model_id

dist = db.predict_with_context_predictor(
    model_id, source="patents", target_key="US-7654321"
)
# {"kind": "gaussian", "mean": ..., "std": ...}

Semantic Search

Measured companion: for the long-form, executed-and-measured Python treatment, see The Cookbook → Retrieval.

Perform ANN vector similarity search over embedding tables. Results include all original source columns, similarity scores, and evidence provenance.

Basic search

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use jammi_ai::session::InferenceSession;
use jammi_db::config::JammiConfig;
async fn ex(config: JammiConfig) -> jammi_db::error::Result<()> {
use std::sync::Arc;

let session = Arc::new(InferenceSession::new(config).await?);

// Encode a query
let query = session.encode_text_query(
    "sentence-transformers/all-MiniLM-L6-v2",
    "quantum computing applications",
).await?;

// Search — returns top 10 results
let results = session.search("patents", query, 10, None, None).await?
    .run().await?;
Ok(()) }
}

Python

query_vec = db.encode_query(model="sentence-transformers/all-MiniLM-L6-v2", query="quantum computing applications")

results = db.search("patents", query=query_vec, k=10)  # pyarrow.Table
print(results.to_pandas())

What search returns

Results are RecordBatch / pyarrow.Table with:

• All original source columns (e.g., id, title, abstract, year)
• _row_id — the source key
• _source_id — which source the row came from
• similarity — cosine similarity score (1.0 = identical, 0.0 = orthogonal)
• retrieved_by — List<Utf8> provenance: which channels found this row
• annotated_by — List<Utf8> provenance: which channels added evidence post-retrieval

Refining a search

search carries the two knobs the bounded primitive owns directly: a SQL filter predicate over the hydrated results and a select column projection. In Python they are keyword arguments and search returns the table; in Rust they are methods on the fluent QueryBuilder (session.search(...) returns the builder, which also carries sort / limit / join / annotate and a .run()).

Filter and select

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use jammi_ai::session::InferenceSession;
async fn ex(session: &std::sync::Arc<InferenceSession>, query: Vec<f32>) -> jammi_db::error::Result<()> {
session.search("patents", query, 20, None, None).await?
    .filter("year > 2020")?
    .sort("similarity", true)?  // descending
    .limit(5)
    .select(&["_row_id".into(), "title".into(), "similarity".into()])?
    .run().await?;
Ok(()) }
}

Python

results = db.search(
    "patents", query=query_vec, k=20,
    filter="year > 2020",
    select=["_row_id", "title", "similarity"],
)  # pyarrow.Table

Compound query (join, annotate)

Joining other sources and running a model over the results is open composition, so in Python and over the wire it is SQL — db.sql(...), with the annotate(...) table function for inference. In Rust the same operations compose on the fluent builder:

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use jammi_ai::session::InferenceSession;
async fn ex(session: &std::sync::Arc<InferenceSession>, query: Vec<f32>) -> jammi_db::error::Result<()> {
let results = session.search("patents", query, 100, None, None).await?
    .filter("year > 2020")?
    .sort("similarity", true)?
    .limit(10)
    .select(&["title".into(), "similarity".into()])?
    .run().await?;
Ok(()) }
}

Python

results = db.sql("""
    SELECT title, vector
    FROM annotate('local:/models/all-MiniLM-L6-v2', 'text_embedding',
                  'patents.public.patents', 'id', 'abstract') AS a
    JOIN patents.public.patents AS p ON a._row_id = arrow_cast(p.id, 'Utf8')
    WHERE p.year > 2020
    LIMIT 10
""")

See Compound Retrieval and Inference over Flight SQL for the full compound surface — it runs the same SQL in-process or against a remote engine over Flight SQL.

ANN vs exact search

Search automatically selects the best path:

• ANN (fast) — when sidecar index files (.usearch + .rowmap + .manifest.json) exist and load successfully
• Exact (brute-force) — fallback when sidecar files are missing or corrupt

The caller never knows the difference. Deleting sidecar files degrades performance but not correctness.

Embedding table resolution

When multiple embedding tables exist for a source, search uses the most recently created “ready” table. The resolution order:

1. Explicit table name (if provided)
2. Latest ready embedding table for the source (by created_at)
3. Error if no embedding table exists

Search over gRPC (edge runtimes)

EmbeddingService exposes Search on the typed gRPC surface, so a process that reaches the engine over gRPC-web — an edge function that cannot speak Flight SQL’s bidirectional HTTP/2 — can run the same similarity search it already uses for AddSource, GenerateAudioEmbeddings, and EncodeAudioQuery. It is the same engine capability on an additional transport, not a second search path.

A SearchRequest carries the source, a k, an optional SQL filter (predicate pushdown), and an optional select column list. The query is a oneof:

• query_vector — a precomputed vector. The usual flow is encode-then-search: call EncodeAudioQuery (or any client-side encoder) to get the vector, then feed it back as the query.
• row_key — query-by-example. The engine resolves that row’s stored vector internally and ranks by it (“rows like this row”). The vector never crosses the wire.

// encode-then-search
embedding = EncodeAudioQuery{ model_id, audio_bytes }.embedding
hits      = Search{ source_id, query_vector: { values: embedding }, k: 10 }.hits

// query-by-example (no re-encode round-trip; vector stays in the engine)
hits      = Search{ source_id, row_key: "clip_1", k: 10 }.hits

Each SearchHit carries the key (the matched row’s key-column value), the score (similarity), and a columns map. columns is empty unless select is non-empty, in which case it holds the requested columns stringified — the engine always projects the key and score alongside them so a hit is fully formed. Heavy clients that want Arrow batches keep using Flight SQL; Search returns lightweight structured rows so an edge bundle needs no Arrow reader.

Building a similarity graph

build_neighbor_graph materializes the k-nearest-neighbour graph of an existing embedding table: for every row it finds the k most similar rows within the same table and writes one directed edge per pair as a queryable edge relation.

When to reach for it — and when not to

Use search for the neighbours of specific rows. Use build_neighbor_graph only when you need the whole edge set at once.

If you want “rows like this row”, call search (or search_by_id). It loads the index once per query and hydrates results on demand — that is the per-query path, and build_neighbor_graph is not for it.

build_neighbor_graph exists for global-structure work, where you consume all edges as a durable artifact:

• near-duplicate detection / semantic dedup,
• clustering and connected components,
• entity resolution,
• generating training pairs for graph-aware fine-tuning.

For those, looping search over every row would reopen the index per row, pay an n× hydration round-trip, and leave you with n detached result sets instead of one catalogued, tenant-scoped, queryable table. The edge table this verb writes closes that gap — and only for the global case.

The edge relation

The result is an ordinary result_table you can query, join, and federate like any other. One row per directed edge:

src and dst are the embedding table’s keys, so the edge table joins directly to your source data — no detour through the embedding table.

Approximate by default; exact on demand

The default driver is index-assisted (HNSW). It is fast (n · log n) but its output is:

• approximate — HNSW recall is below 100%, so some true neighbours are missed, and
• non-deterministic — two builds can differ in the long tail of weak edges.

For dedup and clustering this is exactly what you want. When you need reproducible, auditable edges, pass exact = true: a brute-force, deterministic, complete n² pass (gated by a row-count ceiling, so it refuses to run on very large tables).

Example: near-duplicate detection

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use std::sync::Arc;
use jammi_ai::session::InferenceSession;
use jammi_db::config::JammiConfig;
use jammi_ai::pipeline::neighbor_graph::BuildNeighborGraph;
use jammi_db::store::CachePolicy;
async fn ex(config: JammiConfig, model_id: &str) -> jammi_db::error::Result<()> {
let session = Arc::new(InferenceSession::new(config).await?);

// Embed the corpus first (any embedding model).
session
    .generate_text_embeddings("patents", model_id, &["abstract".into()], "id", CachePolicy::Bypass, None)
    .await?;

// Materialize the kNN graph, keeping only strong, reciprocal edges.
let (edges, _outcome) = session
    .build_neighbor_graph(
        "patents",
        None, // resolve the latest embedding table
        &BuildNeighborGraph {
            k: 5,
            min_similarity: Some(0.9), // near-duplicate threshold
            mutual: true,              // both rows agree they are neighbours
            ..Default::default()
        },
        CachePolicy::Bypass,
    )
    .await?;

// The edge table is now an ordinary relation. Group by src to list each
// row's near-duplicates, joined directly to the source on the key.
let dupes = session
    .sql(&format!(
        "SELECT e.src, e.dst, e.similarity, p.title \
         FROM \"jammi.{}\" e \
         JOIN patents.public.patents p ON p.id = e.src \
         ORDER BY e.similarity DESC",
        edges.table_name
    ))
    .await?;
Ok(()) }
}

Example: graph traversal stays in SQL

build_neighbor_graph transports adjacency and weight — it never walks the graph. Two-hop expansion, paths, and reachability are plain SQL over the edge relation, on every transport:

-- Two-hop neighbours via a self-join on the edge table.
SELECT a.src AS origin, b.dst AS two_hops_away
FROM "jammi.<edge_table>" a
JOIN "jammi.<edge_table>" b ON b.src = a.dst
WHERE a.src = '<some_key>';

For deeper walks, use a WITH RECURSIVE CTE. There is no traversal operator and no graph DSL — the edge table is just a relation.

Example: training-data prep for graph-aware fine-tuning

The edge table is the raw material for neighbour-contrastive embedding fine-tuning. Turn edges into (anchor, positive, negative) triplets — a neighbour is a positive, a non-neighbour a negative — and feed them to the existing triplet/contrastive fine-tune path. The walk policy, negative sampling, and objective are yours; Jammi supplies the edges and the loss.

Propagate Embeddings over a Graph (Decoupled GNN)

Measured companion: for the long-form, executed-and-measured Python treatment, see The Cookbook → Graph Signal Processing.

propagate_embeddings is the forward pass of a graph convolution, run as a data-plane operation. For every row of an embedding table it replaces the row’s vector with an aggregate of its k-hop neighbourhood — ÂᵏX — and writes the result as a new, ordinary embedding table (searchable, joinable, re-graphable).

This is the propagate half of a decoupled GNN. SGC showed the nonlinearities between graph-conv layers are removable: precompute the propagated features, then learn a simple head on top. APPNP added the teleport restart that keeps deep propagation from collapsing. Neither needs autograd, an architecture, or message-passing code — ÂᵏX is a graph join plus a grouped vector average, and that is all this verb is.

It composes with anything that consumes an embedding table: search the propagated vectors, evaluate them, build a neighbour graph over them, or fine-tune a head on them (the SGC/APPNP order — propagate first, then fine-tune).

When propagation helps — measure homophily first

Smoothing helps only when neighbours share signal.

Averaging a node with its neighbours denoises it when the graph is homophilous — neighbours tend to be the same kind of thing (papers cite papers on the same topic; co-purchased items share a category; KG entities of one type link to one type). Then the propagated vectors cluster tighter and downstream search / classification improves.

On a heterophilous graph — neighbours tend to differ — propagation mixes in opposing signal and is beaten by ignoring the graph entirely. This is not a silent failure mode to discover in production: measure it first. The per-edge-type homophily diagnostic reports, for each edge type, how often its endpoints share a label. Propagate over the homophilous types; for genuinely heterophilous structure the answer is learned attention (a later spec), not fixed averaging.

The default is over-smoothing-safe

Iterated averaging is exactly the operation that collapses every node into one indistinguishable point as the hop count grows (rank collapse). Three defaults keep that in check:

• PageRank-decay weighting (DegreeNormalized + an α-teleport restart). Each hop re-mixes a share α of every node’s original embedding back in, so a node stays anchored to itself however deep you go (the APPNP fix). α defaults to 0.1.
• Two hops by default, capped at three. Beyond that, more hops add collapse, not signal.
• Self-loops (à = A + I). Every node aggregates over itself, so an isolated node propagates to its own embedding rather than vanishing, and the symmetric normalisation has no oscillating eigenmode.

Weightings

Output: final block, or Jumping Knowledge

By default the output is the final propagated block X⁽ᴷ⁾, a d-dimensional embedding table in the input’s vector space.

PropagationOutput::JumpingKnowledge instead concatenates the per-hop blocks [X⁽⁰⁾ ‖ … ‖ X⁽ᴷ⁾], each L2-normalised before concat so the raw block does not dominate cosine search. This lets a downstream head pick the right receptive depth per node, but the output is (K+1)·d-dimensional and indexes in its own space — do not search it against the original d-dimensional vectors.

Example: propagate over a citation graph

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use std::sync::Arc;
use jammi_ai::session::InferenceSession;
use jammi_db::config::JammiConfig;
use jammi_ai::pipeline::graph_neighbourhood::{EdgeDirection, EdgeSourceRef};
use jammi_ai::pipeline::graph_propagation::{PropagateRequest, PropagationWeighting};
use jammi_db::store::CachePolicy;
async fn ex(config: JammiConfig, model_id: &str) -> jammi_db::error::Result<()> {
let session = Arc::new(InferenceSession::new(config).await?);

// Embed the documents first (any embedding model).
session
    .generate_text_embeddings("papers", model_id, &["abstract".into()], "id", CachePolicy::Bypass, None)
    .await?;

// Propagate over a declared citation edge source (src/dst are the paper ids,
// which are the embedding keys). Citations are undirected for smoothing.
let (propagated, _outcome) = session
    .propagate_embeddings(
        &PropagateRequest::new(
            "papers",
            EdgeSourceRef::Registered {
                source_id: "citations".into(),
                src_column: "citing".into(),
                dst_column: "cited".into(),
                type_column: None,
                weight_column: None,
                as_of_column: None,
            },
        )
        .with_direction(EdgeDirection::Undirected)
        .with_weighting(PropagationWeighting::DegreeNormalized)
        .with_hops(2),
        CachePolicy::Bypass,
    )
    .await?;

// The result is an ordinary embedding table: search it, evaluate it, or graph
// it like any other.
let neighbours = session
    .sql(&format!(
        "SELECT _row_id FROM \"jammi.{}\" LIMIT 5",
        propagated.table_name
    ))
    .await?;
let _ = neighbours;
Ok(())
}
}

Propagating over an S9 similarity graph

You can also propagate over the similarity graph Jammi itself builds — pass its table name as the edge source:

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use std::sync::Arc;
use jammi_ai::session::InferenceSession;
use jammi_db::config::JammiConfig;
use jammi_ai::pipeline::graph_neighbourhood::EdgeSourceRef;
use jammi_ai::pipeline::graph_propagation::PropagateRequest;
use jammi_db::store::CachePolicy;
async fn ex(config: JammiConfig, graph_table: &str) -> jammi_db::error::Result<()> {
let session = Arc::new(InferenceSession::new(config).await?);
let (_propagated, _outcome) = session
    .propagate_embeddings(
        &PropagateRequest::new(
            "items",
            EdgeSourceRef::NeighborGraph {
                table_name: graph_table.into(),
            },
        ),
        CachePolicy::Bypass,
    )
    .await?;
Ok(())
}
}

But note the caveat from graph-supervised fine-tuning: a similarity graph is k-NN under the base metric, so propagating over it mostly re-averages things the model already thinks are close. Declared edges (a citation network, a co-purchase log, a knowledge graph’s typed relations) carry structure the base metric does not already encode — that is where propagation adds signal.

Determinism

Propagation is deterministic: every fold, teleport, and weighted sum runs in f64 over a fixed (node, neighbour) order, so the output is byte-identical regardless of how many threads the engine runs, on a machine. It is the reproducible point on the structure-aware spectrum — fixed averaging, no learned parameters.

Bounds

The edge set is loaded under a row ceiling (PropagateRequest::max_rows); a graph larger than that is refused loudly rather than risking an out-of-memory pass. Whole-graph propagation beyond memory (chunking by the join) is future work.

Hybrid Retrieval: Lexical (BM25) + Reciprocal-Rank Fusion

Measured companion: for the long-form, executed-and-measured Python treatment, see The Cookbook → Retrieval.

Dense vector search finds rows that mean the same thing as your query; lexical (BM25) search finds rows that contain the same words. Each misses what the other catches — dense search fumbles rare identifiers and exact phrases, lexical search misses paraphrase. Hybrid retrieval runs both and fuses their rankings, and it is the standard production recipe because it reliably beats either alone.

Jammi ships two pieces for this:

• A lexical sidecar (LexicalIndex) — a tantivy BM25 inverted index that rides beside a result table’s Parquet object, the lexical peer of the USearch ANN sidecar.
• Reciprocal-rank fusion (rrf_fuse) — merges any number of ranked lists by rank, not score.

Fusing by rank is the whole point: BM25 scores and cosine similarities live on incompatible scales, so averaging them is meaningless. RRF never looks at a raw score — it sums 1 / (k_rrf + rank) across the lists a row appears in, so the fused order depends only on where a row landed in each list. The default k_rrf is 60 (Cormack et al., SIGIR 2009; robust across 40–80).

Build a lexical index

A LexicalIndex is built over (row_id, text) pairs — the text is whatever text columns of the row you want searchable, joined by the caller. The analyzer is configurable; English (lowercase + Porter stemming) is the default, and Raw (lowercase, no stemming) is the escape hatch for text the English stemmer would mangle (codes, identifiers, non-English).

#![allow(unused)]
fn main() {
extern crate jammi_ai;
extern crate jammi_db;
fn ex() -> jammi_db::error::Result<()> {
use jammi_ai::index::{Analyzer, LexicalIndex};

let rows = vec![
    ("doc-1", "a method for reducing turbine blade vibration"),
    ("doc-2", "an apparatus for cooling turbine engine blades"),
    ("doc-3", "a recipe for baking sourdough bread"),
];

let lexical = LexicalIndex::build(rows, Analyzer::English)?;
let hits = lexical.search("turbine engine", 10)?;
for hit in &hits {
    println!("{} bm25={:.3} rank={}", hit.row_id, hit.bm25_score, hit.rank);
}
Ok(()) }
}

Each LexicalHit carries the row_id, its raw bm25_score, and its 0-based rank — the rank is what fusion consumes.

Lifecycle and scope

The lexical sidecar’s lifecycle equals the ANN sidecar’s: it is built (and rebuilt) with the table. An immutable result table that is rebuilt produces a fresh sidecar; for a mutable-table source, re-ingesting the changed rows into a new index is the caller’s mode. Search applies no row-level filter — isolation is table-level, exactly as the ANN search path: resolve the table through the tenant-scoped catalog and hand the index only that table’s rows.

Fuse dense and lexical rankings

rrf_fuse takes a slice of ranked lists — each a best-first list of _row_ids — and returns one fused ranking. The dense list is the ANN search result; the lexical list is the LexicalIndex result. A third list (e.g. a graph-retrieval channel) fuses identically, with no special-casing.

#![allow(unused)]
fn main() {
extern crate jammi_ai;
use jammi_ai::query::{rrf_fuse, DEFAULT_K_RRF};

// Best-first row-id lists from each retriever.
let dense = vec!["doc-2", "doc-1", "doc-5"];   // ANN cosine order
let lexical = vec!["doc-1", "doc-2", "doc-9"]; // BM25 order

let fused = rrf_fuse(&[dense, lexical], DEFAULT_K_RRF);
for hit in &fused {
    println!("{} rrf={:.4}", hit.row_id, hit.rrf_score);
}
}

Rows that both retrievers surface rise to the top — cross-list agreement is exactly what RRF rewards. The output is fully deterministic: it is sorted by fused score descending, ties broken ascending by row_id, and it does not depend on the order you pass the lists in. A row repeated within a single list counts only once, at its best rank.

k_rrf is exposed, not forced. Larger values flatten the gap between adjacent ranks (a deep-but-agreed-upon row matters more); smaller values sharpen the reward for top-of-list placement. DEFAULT_K_RRF (60) is the recommended start.

Record the evidence

BM25 contributions ride the built-in bm25 evidence channel, the lexical peer of vector’s similarity. It declares two columns — bm25_score (Float32) and bm25_rank (Int64) — and a contribution is supplied to merge_channels exactly as the vector channel’s is, so a fused result carries both its dense and its lexical provenance side by side. See Declare a Custom Provenance Channel for the contribution mechanics; bm25 needs no registration — it is seeded with the catalog.

Assemble a Context Set for Conditioned Prediction

Measured companion: for the long-form, executed-and-measured Python treatment, see The Cookbook → Retrieval.

A prediction is often best made conditioned on a neighbourhood: not “what is the label of this row” in the abstract, but “given the k most similar labelled rows, what is the label of this row.” That neighbourhood is a context set — C = {(xᵢ, yᵢ)} — and in a database it is not an abstraction to invent. It is a search joined to its labels.

assemble_context makes that first-class. It retrieves a target’s k nearest neighbours, pairs them with their outcome columns, and pools the neighbour vectors — permutation-invariantly — into one fixed-width context vector a predictor can condition on. It is the encode-and-aggregate half of a Neural Process; the decode half (a learned predictor over the representation) composes on top.

When this is the right tool

Reach for assemble_context when you want a reusable set representation of a target’s retrieved neighbourhood — a conditioning vector, a prototype/centroid, a bag-of-evidence summary. If you only need to aggregate one specific, already known set of rows once, that is a SQL GROUP BY you already have; this is the operator that turns any target’s retrieval into a representation, reproducibly, with the leakage guards a prediction context needs.

Assemble and encode

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use jammi_ai::session::InferenceSession;
use jammi_ai::pipeline::context_set::{ContextRequest, SetAggregator};
async fn ex(session: &std::sync::Arc<InferenceSession>, query: Vec<f32>) -> jammi_db::error::Result<()> {
let mut request = ContextRequest::new("patents", query, 10);
request.value_columns = vec!["category".into()];     // the labels carried per context row
request.aggregator = SetAggregator::Mean;             // mean | sum | max pooling

let context = session.assemble_context(&request).await?;

if let Some(vector) = &context.context_vector {
    // condition a predictor on `vector` + `context.context_size`
    let _ = (vector, context.context_size);
}
Ok(()) }
}

The result carries:

• context_vector — the pooled ρ(Σ φ(xᵢ)) representation, or None for an empty context (no neighbour survived the guards below — treat as low-confidence / fall back to the prior, never as a one-element average).
• context_size — the number of neighbours that entered the pool, carried separately so a decoder can use the count signal without it corrupting the pooled vector.
• context_keys — the context members’ keys, in retrieval (descending similarity) order.
• value_rows — the requested value_columns of each member, in the same order.

The leakage guards (on by default)

A target that retrieves itself as its own context trivially leaks the answer when a value column is the prediction target. So:

• exclude_self defaults on. Set exclude_key to the target’s own row key (when the query vector belongs to a stored row) and that same-key neighbour is dropped before pooling; the retrieval over-fetches by one so a self-hit never shrinks the context below k.
• split scopes the context to a train split. When the context feeds a training or evaluation target, pass e.g. split = Some("split = 'train'".into()) so the target’s own outcome stays held out — the same train/target line the evaluation harness enforces.

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
use jammi_ai::pipeline::context_set::ContextRequest;
fn ex(query: Vec<f32>) {
let mut request = ContextRequest::new("patents", query, 10);
request.exclude_key = Some("US-1234567".into());   // drop the target's own row
request.split = Some("split = 'train'".into());     // context from the train split only
let _ = request;
}
}

Pooling: fixed, permutation-invariant, deterministic

The encoder pools through the engine’s vector-aggregation functions (vector_mean / vector_sum / vector_max) — the same element-wise aggregation the engine ships for grouped vector reduction. The pool is:

• permutation-invariant in value — shuffling the context rows yields the same vector value, but not necessarily byte-identical bits: the pool runs through the engine’s vector-aggregation SQL UDAF (vector_mean/vector_sum/ vector_max), and f64 + is non-associative, so a different query partitioning can move the last bits even though the aggregate is commutative and associative at the value level;
• stable run-to-run under a fixed execution plan on one host — the pooled vector is reproducible across runs, not asserted byte-identical across an arbitrary re-partitioning or a different host.

mean discards set size; sum encodes it; max is robust but lossy. None is universally right, which is why the aggregator is a knob and context_size is always carried alongside.

This is fixed pooling — the DeepSets / Conditional-Neural-Process expressiveness ceiling. It cannot model which context element matters; learned attention pooling (the AttnCNP point on the spectrum) is a separate, downstream capability, not a silent extension of this one.

Materialise for batch workflows

For batch pipelines, pool every target once and land the results as a normal embedding-shaped result table — searchable and joinable like any other embedding table, with its own sidecar ANN index:

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use jammi_ai::session::InferenceSession;
use jammi_ai::pipeline::context_set::{ContextRequest, MaterializedContext};
use jammi_db::store::CachePolicy;
async fn ex(session: &std::sync::Arc<InferenceSession>, rows: Vec<(String, Vec<f32>)>) -> jammi_db::error::Result<()> {
// The recipe carries *how* every row was pooled — the source, candidate set,
// value columns, aggregator, self-exclusion, and split — and names the source.
let recipe = ContextRequest::new("patents", vec![0.0_f32; 32], 5);
let (table, _outcome) = session
    .materialize_context(
        MaterializedContext {
            rows: &rows,
            dimensions: 32,
            recipe: &recipe,
            // These targets are `patents` rows, so their keys are `patents.id`.
            key_column: Some("id"),
        },
        CachePolicy::Bypass,
    )
    .await?;
// `table` is a normal embedding result table: search it, join it, index it.
println!("materialised context table: {}", table.table_name);
Ok(()) }
}

Each target’s key becomes the table’s _row_id; the pooled context vector becomes its vector. A materialised context set is a first-class member of the same table family every embedding table belongs to. The recipe materialize_context takes is the batch’s shared assembly definition (the source comes from it); the per-target query vectors are the inputs the recipe ran over, which is why they are the rows, not part of the recipe.

key_column names which column of the source those target keys came from, and is recorded as the table’s provenance: it is what lets a reader join patents.id = <context table>._row_id to get back to the rows the targets are. You declare it because the targets are yours — materialize_context receives only (key, vector) pairs and cannot see where the keys came from — but it does check the half it can: a column the source does not have is rejected rather than recorded. Pass None when the targets are free vectors that correspond to no stored row; the table is still searchable, but nothing joins it back to a source.

Condition a Prediction on Declared-Edge Context (Bring Your Own Graph)

assemble_context builds a target’s context set from its embedding-similar neighbours — search(target, k). That is the right neighbourhood when similarity is the relationship you want to condition on. But often the relationship that matters is one only your domain declares: the papers a paper cites, the products co-purchased with this one, the concepts an entity is an is-a of. Those edges carry structure an embedding metric does not reconstruct — and they are exactly the context a prediction is most defensible conditioned on.

A context set is a search. It is also a walk — and the walk you care about is the one only you can declare.

S16-G makes that first-class: a second context source for the same assemble_context. You register an edge relation, and a target’s context becomes its bounded, target-anchored declared-edge neighbourhood — pooled through the same permutation-invariant set encoder, under the same leakage and tenancy contracts, decoded by the same calibrated predictor. The engine transports adjacency; it never learns what an edge means.

When this is the right tool

Reach for declared-edge context when a target’s most informative neighbours are the ones your graph names, not the ones the base metric happens to place nearby: a citation classifier, a co-purchase recommender, a knowledge-graph-backed labeller, a transaction-graph scorer. Use ANN context when similarity is the signal; use Hybrid (below) when declared edges are the signal but the graph is sparse and you want similarity to densify it.

It is not a general graph-traversal verb. The gather is target-anchored and depth-/fan-out-bounded — never a free walk — which is what keeps it inside the tenant-scope guarantee.

The edge source

Any relation with two key columns is an edge source. Register it like any other source, then point the gather at it:

• a similarity graph you already built (neighbor_graph), or
• an external edge table you register (two key columns, optionally a type and a weight column).

The edge endpoints are row keys: a neighbour joins to its stored vector and its outcome columns by key, exactly as an ANN neighbour does.

Assemble declared-edge context

Rust

#![allow(unused)]
fn main() {
extern crate jammi_ai;
extern crate jammi_db;
use std::sync::Arc;
use jammi_ai::session::InferenceSession;
use jammi_ai::pipeline::context_set::{ContextRequest, ContextSource};
use jammi_ai::pipeline::graph_neighbourhood::{EdgeGather, EdgeSourceRef};

async fn demo(session: Arc<InferenceSession>, target_vector: Vec<f32>) -> jammi_db::error::Result<()> {
// A registered edge relation: a node's declared neighbours.
let gather = EdgeGather::new(EdgeSourceRef::Registered {
    source_id: "citations".into(),
    src_column: "from_id".into(),
    dst_column: "to_id".into(),
    type_column: None,
    weight_column: None,
    as_of_column: None,
});

let mut request = ContextRequest::new("papers", target_vector, 0);
request.source = ContextSource::Edges(gather);
// The target's own row key is the gather anchor (and is excluded from its
// own context — the leakage guard).
request.exclude_key = Some("paper-42".into());
request.value_columns = vec!["topic".into()];

let context = session.assemble_context(&request).await?;
println!("conditioned on {} declared-edge neighbours", context.context_size);
Ok(())
}
}

Python

out = db.predict_with_context_predictor(
    model_id,
    source="papers",
    target_key="paper-42",
    edge_source="citations",     # a registered edge relation
    edge_src_column="from_id",
    edge_dst_column="to_id",
    edge_hops=1,                 # bounded depth (default 1)
    edge_fanout=25,              # sample at most 25 neighbours per node per hop
)
# The prediction carries how its context was assembled and which rows it used:
assert out["source"] == "edges"
print(out["context_ref"])        # the declared-edge member keys

The bounds — and why they are caps, not knobs to crank

• hops (default 1, hard-capped). Beyond ~2–3 hops, neighbour pooling is Laplacian over-smoothing — the pooled vector washes out and loses signal. Depth is a precision/recall trade-off, not “more context is better.”
• fanout (sample, don’t enumerate). A high-degree node’s neighbourhood is intractable to enumerate; fanout bounds the neighbours sampled per node per hop. The sample is seeded-deterministically from the target, so a gather reproduces byte-identically. None is exact (enumerate all) and uses no randomness; a truncated neighbourhood is reported, never silently dropped.
• edge_types / min_weight. Filter which edges the walk follows. Types are a filter, never learned aggregation — a consumer wanting learned multi-relational message passing runs it in a graph library and registers the resulting node embeddings back as a source.

Hybrid: declared edges ∪ similarity

When the graph is sparsely connected, union the declared-edge neighbours with the ANN neighbours and pool once — declared edges as the signal, similarity as the densifier:

#![allow(unused)]
fn main() {
extern crate jammi_ai;
use jammi_ai::pipeline::context_set::{ContextSource, HybridMerge};
use jammi_ai::pipeline::graph_neighbourhood::{EdgeGather, EdgeSourceRef};
fn demo(gather: EdgeGather) -> ContextSource {
ContextSource::Hybrid {
    ann_k: 10,
    edges: gather,
    merge: HybridMerge::Union,
}
}
}

Check homophily before you trust it

A declared edge can be heterophilous — cites may connect dissimilar items — and pooling over a heterophilous edge type degrades a prediction rather than helping it. Declared-edge context is an option on the spectrum, never unconditionally better than similarity. Before you rely on a type, read its homophily:

#![allow(unused)]
fn main() {
extern crate jammi_ai;
extern crate jammi_db;
use std::sync::Arc;
use jammi_ai::session::InferenceSession;
use jammi_ai::pipeline::graph_neighbourhood::EdgeGather;
async fn demo(session: Arc<InferenceSession>, gather: &EdgeGather) -> jammi_db::error::Result<()> {
// Per-edge-type label-agreement over a labelled set: a type near the label's
// chance rate is heterophilous — pooling over it is unlikely to help.
let homophily = session
    .homophily_by_edge_type(gather, "papers", "id", "topic")
    .await?;
for (edge_type, agreement) in &homophily {
    println!("{edge_type}: {agreement:.2} label agreement");
}
Ok(())
}
}

The decoder also receives the target’s own (ego) features alongside the pooled neighbour vector, so it can down-weight an unhelpful neighbourhood rather than being forced to trust it.

Coverage over graph context

A graph-conditioned prediction is decoded and conformally wrapped exactly like an ANN-conditioned one, and it always serves. But graph correlation can break the exchangeability that marginal split-conformal coverage assumes — so the served prediction carries the assembly source fact and its member keys, and the coverage claim is attributed, never silently presented as a guarantee. Choosing whether to apply a group-conditional (Mondrian) or importance-weighted lever, and which cohort or weights to use, is a governance decision the serving layer applies but never makes. The engine surfaces the fact; governance chooses the lever.

Tenancy

An edge source is tenant-scoped like every other source. The gather runs inside the session’s tenant scope, so an edge whose endpoint belongs to another tenant is filtered before it is ever materialised — a declared edge cannot leak one tenant’s rows into another’s context.

Point-in-time joins: matching facts to the instant they were known

Measured companion: for the long-form, executed-and-measured Python treatment, see The Cookbook → Point-in-Time Correctness.

An as-of join matches each row of a spine relation to the at-most-one row of a facts relation that was valid as of the spine row’s instant, within the same group. It is the relational primitive for point-in-time correctness.

The problem it solves is leakage. If you assemble a table by joining each spine row to any fact in its group, you import facts stamped after the spine instant — information that was not yet known. A forward join imports the future. The as-of join takes only the fact valid at or before each instant, so every attached value reflects what was knowable then, and nothing later.

The engine exposes this as one verb, asof_join, over two registered relations. It carries only what every time-aware caller needs — an equality grouping, a temporal ordering key, a match direction, boundary inclusivity, an optional look-back tolerance, and a deterministic tie-break — and writes a result table that carries the same materialization manifest every other producer does.

The call

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use std::sync::Arc;
use jammi_ai::session::InferenceSession;
use jammi_db::config::JammiConfig;
use jammi_ai::pipeline::asof::{
    AsofJoinSpecBuilder, AsofKey, Boundary, MatchDirection, TieBreak, Tolerance,
};
async fn ex(config: JammiConfig) -> jammi_db::error::Result<()> {
let session = Arc::new(InferenceSession::new(config).await?);

// `events` is the spine (its `t` column is the as-of instant); `facts` carries
// the values that were valid over time (its `vt` column is their validity time).
// `key` groups the match so a fact only matches an event in the same group.
let spec = AsofJoinSpecBuilder::new(
        AsofKey { by: vec!["key".into()], time: "t".into() },  // spine
        AsofKey { by: vec!["key".into()], time: "vt".into() }, // facts
    )
    .direction(MatchDirection::Backward)  // most recent at or before
    .boundary(Boundary::Inclusive)        // a fact stamped exactly at t matches
    .tolerance(Some(Tolerance::Duration(5_000_000))) // ignore facts >5s stale
    .tie_break(TieBreak::ByColumnDesc("seq".into())) // newest seq wins a tie
    .project(vec!["value".into()])        // attach this fact column
    .build();

let table = session.asof_join("events", "facts", &spec).await?;

// The result is an ordinary relation: every spine row, with the matched fact's
// `value` attached (null where nothing matched within the rules). Read it via SQL.
let _rows = session
    .sql(&format!(
        "SELECT t, value FROM \"{name}\".public.\"{name}\" ORDER BY t",
        name = table.table_name,
    ))
    .await?;
Ok(())
}
}

The spine is always fully preserved — an unmatched spine row keeps its columns and carries nulls for the fact columns. Dropping unmatched rows would silently shrink the result; a caller who wants inner semantics filters on a non-null fact column themselves.

The four pinned knobs

Each knob is the choice every engine gets subtly different. The engine pins them once, on the spec, never inferred — and each one changes the result.

Direction

MatchDirection::Backward (the default) takes the most recent fact at or before the instant — the only leakage-safe choice when the spine instant is a point you must not see past. Forward takes the first fact at or after (e.g. “the next scheduled event after each reading”). Nearest takes the smallest absolute distance, resolving equidistant candidates toward the past, and requires a numeric temporal key.

Boundary

Boundary::Inclusive (the default, <=) lets a fact stamped exactly at the instant match; Boundary::Exclusive (<) excludes it. Over identical inputs, the two differ exactly on the rows that have a fact coincident with the instant.

Tolerance

None (the default) looks back arbitrarily far. Some(Tolerance::Duration(µs)) for a temporal key, or Some(Tolerance::Steps(n)) for an integer key, discards a candidate farther than the limit — the spine row goes unmatched rather than matching a stale fact. The limit is measured relative to each spine instant.

Tie-break

When two facts share the matched instant within a group, the match is ambiguous. TieBreak::ByColumnDesc("seq") disambiguates by a secondary column, the maximal value winning (the transaction-time column). TieBreak::Error makes a true duplicate a loud AsofError::AmbiguousMatch rather than a silent, non-deterministic pick. With a tie-break in force the output is bit-reproducible.

The temporal key must be totally ordered

The temporal key on each side must be a totally-ordered Arrow type — any Timestamp(..), Date32/Date64, or a signed/unsigned integer. A float key is rejected: NaN has no total order, so “most recent at or before” would be undefined. The two sides’ temporal keys must share a type, and a null temporal value is never ordered — a null-time spine row is preserved with null facts, and a null-time fact is never a candidate.

One verb, many shapes

The same asof_join assembles a leakage-free labelled set keyed on past instants, matches each transaction to the value in effect when it occurred, and pairs a measurement with the reading valid at the time it was taken. The engine provides the as-of relational primitive and the determinism contract; what a caller assembles on top of it is theirs.

Enrich Results with Joins and Annotations

Search results can be enriched by joining with other data sources and annotating with additional model inference.

There are two surfaces for this, and they are deliberately different:

• search is the bounded, jammi-defined primitive: nearest-neighbor top-k with optional filter/select, returning a table directly. Same call, same shape, embedded or remote.
• Compound query — open, caller-shaped composition (join / filter / select and model inference over the results) — rides SQL. In Rust the fluent QueryBuilder (returned by session.search(...)) builds the same plan in-process; in Python and over the wire the surface is db.sql(...), where the annotate(...) table function runs a model over a relation. Both descend through the one inference operator, so an in-process query and a Flight-SQL query run the same plan node.

The fluent Rust builder tracks every enrichment step in the evidence provenance columns (retrieved_by / annotated_by).

Join with another source

Join search results with a registered source to add context columns (e.g., company name, category labels):

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use std::sync::Arc;
use jammi_ai::session::InferenceSession;
use jammi_db::source::{FileFormat, SourceConnection, SourceType};
async fn ex(session: &Arc<InferenceSession>, query: Vec<f32>) -> jammi_db::error::Result<()> {
session.add_source("assignees", SourceType::File, SourceConnection {
    url: Some("file:///data/assignees.csv".into()),
    format: Some(FileFormat::Csv),
    ..Default::default()
}).await?;

let results = session.search("patents", query, 10, None, None).await?
    .join("assignees", "assignee_id=id", None).await?  // left join by default
    .run().await?;
// Results now include company_name, country from assignees
Ok(()) }
}

Python

In Python the compound query is SQL. search returns a pyarrow.Table directly; to join, run SQL that the engine plans (in-process for the embed wheel, over Flight SQL for a remote engine — same SQL either way):

db.add_source("assignees", path="/data/assignees.csv", format="csv")

results = db.sql("""
    SELECT p.title, a.company_name, a.country
    FROM patents.public.patents AS p
    JOIN assignees.public.assignees AS a ON p.assignee_id = a.id
""")
# Results now include company_name, country from assignees

In Rust, the fluent builder’s on parameter is "left_col=right_col" and the optional join type is "inner" or "left" (default).

Annotate with model inference

Run a model over search results to add new columns:

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use std::sync::Arc;
use jammi_ai::session::InferenceSession;
use jammi_db::ModelTask;
async fn ex(session: &Arc<InferenceSession>, query: Vec<f32>) -> jammi_db::error::Result<()> {
let results = session.search("patents", query, 10, None, None).await?
    .annotate(
        "sentence-transformers/all-MiniLM-L6-v2",
        ModelTask::TextEmbedding,
        &["abstract".to_string()],
    ).await?
    .run().await?;
Ok(()) }
}

Python

annotate(model, task, relation, key_column, content_column, …) is a SQL table function: it runs the model over the named relation’s columns and returns the inference output (_row_id keyed from key_column, plus the task’s columns — e.g. vector). Join it back to the source on _row_id to enrich:

results = db.sql("""
    SELECT p.title, a.vector
    FROM annotate('sentence-transformers/all-MiniLM-L6-v2', 'text_embedding',
                  'patents.public.patents', 'id', 'abstract') AS a
    JOIN patents.public.patents AS p ON a._row_id = arrow_cast(p.id, 'Utf8')
""")

The same SQL — the same annotate function — runs in-process (embed wheel) or over the Flight SQL lane (remote engine, jammi-client), so compound retrieval + inference is one round-trip.

Evidence provenance

Every search result carries provenance tracking that records how each row was found and enriched:

These are List<Utf8> columns — each row has its own list of contributing channels.

Composing everything

All operations compose freely:

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use std::sync::Arc;
use jammi_ai::session::InferenceSession;
use jammi_db::ModelTask;
async fn ex(session: &Arc<InferenceSession>, query: Vec<f32>) -> jammi_db::error::Result<()> {
let results = session.search("patents", query, 100, None, None).await?
    .join("assignees", "assignee_id=id", None).await?
    .annotate("all-MiniLM-L6-v2", ModelTask::TextEmbedding, &["abstract".into()]).await?
    .filter("country = 'US'")?
    .sort("similarity", true)?
    .limit(10)
    .select(&["title".into(), "company_name".into(), "similarity".into()])?
    .run().await?;
Ok(()) }
}

Python

The compound query is SQL, so everything composes as JOIN / WHERE / ORDER BY / LIMIT / projection over annotate(...) and registered sources:

results = db.sql("""
    SELECT p.title, a.company_name, ann.vector
    FROM annotate('all-MiniLM-L6-v2', 'text_embedding',
                  'patents.public.patents', 'id', 'abstract') AS ann
    JOIN patents.public.patents  AS p ON ann._row_id = arrow_cast(p.id, 'Utf8')
    JOIN assignees.public.assignees AS a ON p.assignee_id = a.id
    WHERE a.country = 'US'
    LIMIT 10
""")

Each surface plans a DataFusion execution plan under the hood. No data is processed until the result is read.

Compound Retrieval and Inference over Flight SQL

search is the bounded primitive — nearest-neighbor top-k, returning a table directly. Compound query — joining sources, filtering, and running a model over a relation — is open, caller-shaped composition, so it rides SQL. The same SQL runs in-process on the embedded engine and over the Flight SQL lane against a remote engine; the annotate(...) table function makes model inference available inside that SQL on both.

This is what lets a remote caller do search → join → annotate in one round-trip, with the model running inside the engine — no per-row RPC, no bespoke compound-search verb.

The annotate table function

annotate(model, task, relation, key_column, content_column [, content_column…])

It runs model (a local:<path>, an HF repo id, or a fine-tuned id) for task over the named relation’s content_column(s), and returns the inference output: the prefix _row_id / _ordinal / _source / _model / _status / _error / _latency_ms — with _row_id carried from key_column and _ordinal a 0-based row counter in model emission order (so ORDER BY _row_id, _ordinal is deterministic even when a key repeats) — followed by the task’s columns (e.g. a vector FixedSizeList for an embedding task). Join it back to the source on _row_id to place inference columns alongside source columns.

Remote: jammi over Flight SQL

import jammi

db = jammi.connect("grpc://engine.internal:8081")  # behind your TLS terminator or on a trusted network

# Compound retrieval + inference in one Flight SQL round-trip:
table = db.sql("""
    SELECT p.title, a.vector
    FROM annotate('sentence-transformers/all-MiniLM-L6-v2', 'text_embedding',
                  'patents.public.patents', 'id', 'abstract') AS a
    JOIN patents.public.patents AS p ON a._row_id = arrow_cast(p.id, 'Utf8')
    WHERE p.year >= 2020
""")
# table is a pyarrow.Table

db.sql carries the connection’s tenant scope (the same jammi-session-id the typed gRPC verbs use), so SQL reads observe the same tenant as db.search.

Embedded: the same SQL, in-process

The embed wheel runs the identical SQL against its in-process DataFusion engine — the annotate function is registered on the same context:

import jammi

db = jammi.connect("file:///var/lib/jammi")
table = db.sql("""
    SELECT a._row_id, a.vector
    FROM annotate('local:/models/all-MiniLM-L6-v2', 'text_embedding',
                  'patents.public.patents', 'id', 'abstract') AS a
""")

Productionising from the embed wheel to the remote client changes only the target (file:// → grpc://) — the import jammi and the sql call are unchanged.

In-process Rust: the fluent builder

In Rust, session.search(source_id, vec, k, embedding_table, oversample) returns a QueryBuilder that composes the same operations as a fluent chain (the annotate node it builds is the very plan node the SQL table function builds):

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use std::sync::Arc;
use jammi_ai::session::InferenceSession;
use jammi_db::ModelTask;
async fn ex(session: &Arc<InferenceSession>, query: Vec<f32>) -> jammi_db::error::Result<()> {
let results = session.search("patents", query, 10, None, None).await?
    .annotate("local:/models/all-MiniLM-L6-v2", ModelTask::TextEmbedding, &["abstract".into()]).await?
    .run().await?;
Ok(()) }
}

Notes

• A WHERE over the annotated output runs above the inference node — the table function declares inference non-pushdown, since a model runs row-wise and a predicate can’t push below it. Filter the source (inside the relation a join scans) when you want to shrink the input the model sees.
• The output schema is fixed at planning time; the embedding dimension is read by loading the model, which is then warm for execution.
• Classification and NER ride the same prefix + task-column shape; pass their task string ('classification', 'ner') and the content column.

Declare a Custom Provenance Channel

Measured companion: for the long-form, executed-and-measured Python treatment, see The Cookbook → Provenance Channels.

Every row that flows through Jammi carries provenance — retrieved_by and annotated_by lists that record how the row was found and what was added after retrieval. Jammi ships two built-in channels — vector (declares similarity) and inference (declares inference_model, inference_task, inference_confidence) — but the catalog accepts any channel a consumer wants to register. Each channel declares the columns it contributes; the engine merges those columns into every result RecordBatch at query time.

This recipe walks through registering a third channel, scored_by, for a multi-stage retrieval pipeline where a federated reranker rescores the vector hits. The same shape applies to any non-built-in provenance signal: a citation graph, an attribution chain, a quality-grading pass.

Setup

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate arrow;
extern crate tokio;
use jammi_db::config::JammiConfig;
async fn ex(config: JammiConfig) -> jammi_db::error::Result<()> {
use std::sync::Arc;
use arrow::array::{ArrayRef, Float32Array, StringArray};
use jammi_ai::evidence::{merge_channels, ChannelContribution};
use jammi_ai::session::InferenceSession;
use jammi_db::catalog::channel_repo::{ChannelColumn, ChannelColumnType, ChannelSpec};
use jammi_db::ChannelId;
use jammi_db::source::{FileFormat, SourceConnection, SourceType};

let session = Arc::new(InferenceSession::new(config).await?);
session.add_source("patents", SourceType::File, SourceConnection {
    url: Some("file:///data/patents.parquet".into()),
    format: Some(FileFormat::Parquet),
    ..Default::default()
}).await?;
Ok(()) }
}

Python

import jammi

db = jammi.connect("file:///var/lib/jammi")
db.add_source("patents", path="/data/patents.parquet", format="parquet")

Declare the channel

Channel declarations are catalog rows. Each declared column has a name, an Arrow type, and an ordinal. The set is append-only — once ranker: Utf8 is declared, the engine refuses to redeclare it as Int32 or drop it.

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use std::sync::Arc;
use jammi_ai::session::InferenceSession;
use jammi_db::catalog::channel_repo::{ChannelColumn, ChannelColumnType, ChannelSpec};
use jammi_db::ChannelId;
async fn ex(session: &Arc<InferenceSession>) -> jammi_db::error::Result<()> {
session.catalog().channels().register(&ChannelSpec {
    id: ChannelId::new("scored_by")?,
    priority: 3,
    columns: vec![
        ChannelColumn {
            name: "ranker".into(),
            data_type: ChannelColumnType::Utf8,
        },
        ChannelColumn {
            name: "rank_score".into(),
            data_type: ChannelColumnType::Float32,
        },
    ],
}).await?;
Ok(()) }
}

Python

db.register_channel(
    "scored_by",
    priority=3,
    columns=[("ranker", "Utf8"), ("rank_score", "Float32")],
)

priority controls the order columns appear in the merged output — vector (1) and inference (2) come first, then scored_by (3).

To add more columns to an already-registered channel:

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use std::sync::Arc;
use jammi_ai::session::InferenceSession;
use jammi_db::catalog::channel_repo::{ChannelColumn, ChannelColumnType};
use jammi_db::ChannelId;
async fn ex(session: &Arc<InferenceSession>) -> jammi_db::error::Result<()> {
session.catalog().channels().add_columns(
    &ChannelId::new("scored_by")?,
    &[ChannelColumn { name: "scored_at".into(), data_type: ChannelColumnType::Utf8 }],
).await?;
Ok(()) }
}

db.add_channel_columns("scored_by", columns=[("scored_at", "Utf8")])

add_columns is append-only by construction. Trying to redeclare ranker with the same or a different type returns JammiError::ChannelCatalog(_).

Use the channel

Build a ChannelContribution for each batch your reranker produces. The arrays must align 1:1 with the channel’s declared columns (ranker first, rank_score second) and have the same length as the batch’s row count.

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate arrow;
extern crate tokio;
use std::sync::Arc;
use arrow::array::{ArrayRef, Float32Array, RecordBatch, StringArray};
use jammi_ai::evidence::{merge_channels, ChannelContribution};
use jammi_ai::session::InferenceSession;
use jammi_db::ChannelId;
fn rerank_scores(_batch: &RecordBatch) -> Vec<f32> { vec![] }
async fn ex(session: &Arc<InferenceSession>, batches: Vec<RecordBatch>) -> jammi_db::error::Result<()> {
let scored_by = ChannelId::new("scored_by")?;
let vector = ChannelId::new("vector")?;

let mut contributions = Vec::with_capacity(batches.len());
for batch in &batches {
    let n = batch.num_rows();
    let ranker: ArrayRef = Arc::new(StringArray::from(vec!["bm25"; n]));
    let rank_score: ArrayRef = Arc::new(Float32Array::from(rerank_scores(batch)));
    contributions.push(vec![ChannelContribution {
        channel: scored_by.clone(),
        columns: vec![ranker, rank_score],
    }]);
}

let merged = merge_channels(
    session.catalog(),
    &batches,
    &[vector.clone(), scored_by.clone()],
    &[vector, scored_by],   // retrieved_by
    &[],                     // annotated_by
    &contributions,
).await?;
Ok(()) }
}

Verify

The merged output schema includes the declared columns. Rows where the channel did not supply a value carry NULL.

Rust

#![allow(unused)]
fn main() {
extern crate arrow;
use arrow::array::RecordBatch;
fn ex(merged: Vec<RecordBatch>) {
let schema = merged[0].schema();
assert!(schema.field_with_name("ranker").is_ok());
assert!(schema.field_with_name("rank_score").is_ok());

for batch in &merged {
    let ranker = batch.column_by_name("ranker").unwrap();
    println!("first ranker: {:?}", ranker);
}
}
}

Python

From the SQL surface, the declared columns show up in any query that touches the result table — Python sees them as plain Arrow columns:

table = db.sql(
    "SELECT _row_id, similarity, ranker, rank_score FROM results LIMIT 3"
)
for row in table.to_pylist():
    print(row["ranker"], row["rank_score"])

What you cannot do

The channel declaration is append-only. Once scored_by ships with ranker: Utf8, you cannot:

• Redeclare ranker as Int32 — add_columns rejects with JammiError::ChannelCatalog(ChannelCatalogError::ColumnConflict { … }), whose message is "channel 'scored_by': column 'ranker' was declared Utf8, cannot redeclare as Int32". From Python, the same call raises RuntimeError carrying the identical message:

  db.add_channel_columns("scored_by", columns=[("ranker", "Int32")])
  # RuntimeError: channel 'scored_by': column 'ranker' was declared Utf8, cannot redeclare as Int32
  

• Add a second column with the same name — add_columns rejects with JammiError::ChannelCatalog(ChannelCatalogError::ColumnAlreadyDeclared { … }), whose message is "channel 'scored_by': column 'ranker' already declared".

• Drop ranker from the channel — there is no drop_column method by design.

If a column needs to change shape, declare a new column under a new name and migrate consumers. This preserves byte-for-byte readability of any backing table or downstream artifact that already references the original column.

Fine-Tune for Your Domain

Measured companion: for the long-form, executed-and-measured Python treatment, see The Cookbook → Fine-Tuning Methods.

Train LoRA adapters on your data to improve embedding quality for your domain. The base model stays frozen — only a small projection layer is trained and saved.

Prepare training data

Create contrastive pairs with a similarity score:

text_a,text_b,score
"quantum error correction","superconducting qubit stabilization",0.88
"quantum error correction","medieval poetry analysis",0.08

High scores mean similar; low scores mean dissimilar.

Register the training data as a source:

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use jammi_ai::session::InferenceSession;
use jammi_db::source::{FileFormat, SourceConnection, SourceType};
async fn ex(session: &InferenceSession) -> jammi_db::error::Result<()> {
session.add_source("training", SourceType::File, SourceConnection {
    url: Some("file:///data/training_pairs.csv".into()),
    format: Some(FileFormat::Csv),
    ..Default::default()
}).await?;
Ok(()) }
}

Python

db.add_source("training", path="/data/training_pairs.csv", format="csv")

Start a fine-tuning job

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use jammi_ai::session::InferenceSession;
async fn ex(session: &InferenceSession) -> jammi_db::error::Result<()> {
use jammi_ai::fine_tune::FineTuneMethod;
use jammi_db::ModelTask;

let job = session.fine_tune(
    "training",
    "sentence-transformers/all-MiniLM-L6-v2",
    &["text_a".into(), "text_b".into(), "score".into()],
    FineTuneMethod::Lora,
    ModelTask::TextEmbedding,
    None,  // default config
).await?;

println!("Job: {}", job.job_id);
job.wait().await?;
println!("Model: {}", job.model_id());
Ok(()) }
}

Python

job = db.fine_tune(
    source="training",
    base_model="sentence-transformers/all-MiniLM-L6-v2",
    columns=["text_a", "text_b", "score"],
    method="lora",
    task="embedding",
)

job.wait()
print(f"Model: {job.model_id}")

base_model accepts any model reference form, including a local checkpoint (local:/path, file:///path, or a bare filesystem path) — see Use a Local Model Checkpoint.

Custom configuration

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use jammi_ai::session::InferenceSession;
use jammi_ai::fine_tune::{FineTuneMethod, LrSchedule};
async fn ex(session: &InferenceSession, model: &str, columns: Vec<String>) -> jammi_db::error::Result<()> {
use jammi_ai::fine_tune::FineTuneConfig;
use jammi_db::ModelTask;

let config = FineTuneConfig {
    lora_rank: 4,
    learning_rate: 5e-4,
    epochs: 5,
    batch_size: 4,
    warmup_steps: 10,
    lr_schedule: LrSchedule::CosineDecay,
    early_stopping_patience: 2,
    validation_fraction: 0.2,
    gradient_accumulation_steps: 4,  // effective batch = 4 x 4 = 16
    ..Default::default()
};

let job = session.fine_tune(
    "training", model, &columns, FineTuneMethod::Lora, ModelTask::TextEmbedding, Some(config),
).await?;
Ok(()) }
}

Configuration reference

Use the fine-tuned model

The fine-tuned model is automatically registered and can be used anywhere a model ID is accepted:

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use jammi_ai::session::InferenceSession;
use jammi_ai::fine_tune::training_job::TrainingJob;
use jammi_db::store::CachePolicy;
async fn ex(session: &InferenceSession, job: &TrainingJob) -> jammi_db::error::Result<()> {
let model_id = job.model_id();

let embedding = session.encode_text_query(model_id, "quantum computing").await?;
println!("query embedding has {} dims", embedding.len());
session.generate_text_embeddings("patents", model_id, &["abstract".into()], "id", CachePolicy::Bypass, None).await?;
Ok(()) }
}

Python

model_id = job.model_id

query_vec = db.encode_query(model=model_id, query="quantum computing")
db.generate_embeddings(source="patents", model=model_id, columns=["abstract"], key="id", modality="text")

Run metrics

job.metrics() (Python) returns the run summary recorded on the job — a dict with final_loss (the best value seen, on whichever metric early_stopping_metric monitored), early_stopping_metric ("train_loss" or "val_loss"), total_steps, and started_at/completed_at timestamps. It returns {} for a job that has not recorded anything yet (still queued, or running before its first stamp), and carries error_message instead when the job failed.

job.wait()
metrics = job.metrics()
print(f"final loss: {metrics['final_loss']} ({metrics['early_stopping_metric']})")

This is a run summary, not a per-epoch curve: the trainer computes and logs avg_train_loss/avg_val_loss at every epoch boundary but does not retain them past that boundary, so no per-epoch trajectory is available through this surface today.

How it works

text -> encoder (frozen) -> base embedding -> LoRA projection (trained) -> output

1. The base encoder model (BERT, ModernBERT, etc.) is loaded and frozen
2. A LoRA projection layer (identity + low-rank A/B matrices) is added after pooling
3. For each batch: text is encoded, projected through LoRA, and loss is computed
4. Only the A/B matrices receive gradients
5. The adapter is saved as adapter.safetensors in the artifact directory

The training set a run reads

A fine-tune trains from a committed snapshot, not from your live source relation. It materialises its training set once as an immutable result table of kind TrainingSet — a Parquet artifact carrying a definition hash and a materialization manifest attestation — and reads that table back in one canonical full-tuple order (every projected column, in declared order, is part of the sort key, so identical tuples are identical rows and the read order is the committed order).

The definition hash folds the source query, the projected columns, the model task, the training format and the order rule — and nothing about how a run consumes the rows, so a run’s world size, batch size or validation split never enters the table’s identity.

Sharing one table across jobs takes more than a matching definition. A ready training set is reused only when its definition hash and every recorded input anchor match the request exactly, and an input anchored unpinned-at-an-instant never matches: an instant is not a reproducible id, so equal anchors would not prove equal rows. That is the engine’s standing reuse rule — the same (definition, input anchors) key the embedding and as-of producers are probed on — with no training-set exception. Which path ran is reported on the returned table rather than left to be inferred.

A registered source relation exposes no version or digest surface to pin, so a fine-tune anchors it unpinned-at-an-instant and materialises its own training set on every run: two runs over the same query and columns leave two tables. Changing the source’s rows therefore never serves a stale training set — the earlier table cannot be served at all. Sharing rests on the rule’s other arm, an input pinned by content digest, as a result table is; a fine-tune source does not reach that arm, because a source resolves as a registered relation and a result table does not resolve as one. The manifest keeps the anchors either way, so a staleness check over the table answers the same honest Undecidable it gives for every unpinned input.

A projection that yields no rows is refused with a typed EmptyTrainingSet error before any catalog row or byte exists, so a run never trains on an empty set in silence.

A graph fine-tune does not go through this table: its sampled pairs are sampled in memory and trained on directly. Giving the graph arm a TrainingSet table of its own is tracked at https://github.com/f-inverse/jammi-ai/issues/538.

Model-level cache reuse (cache = Use) is not yet supported

fine_tune (the column-source kind) and fine_tune_graph both accept the same opt-in cache dial the compute verbs (generate_embeddings, infer, …) carry, but neither honors cache="use". It is refused, typed (jammi.errors.InvalidArgument on both transports): model-level cache reuse — binding a fine-tune job to an earlier run’s already-published model instead of training — is not yet supported (https://github.com/f-inverse/jammi-ai/issues/562). cache="bypass" (the default, or omitting cache entirely) is unaffected on either kind: a fine-tune job always trains.

Python

job = db.fine_tune(
    source="training",
    base_model="sentence-transformers/all-MiniLM-L6-v2",
    columns=["text_a", "text_b", "score"],
    method="lora",
    task="embedding",
)
result = job.wait()
print(result["cache_outcome"])  # always "computed"
print(f"Model: {result['model_id']}")

Rust

The embedded surface’s cache dial lives on jammi_wire::request::FineTuneRequest (submitted through InferenceSession::submit_fine_tune), not on the loose InferenceSession::fine_tune/fine_tune_graph methods shown above, which always train (cache: CachePolicy::Bypass, unconditionally). The remote jammi_client::DataClient::submit_fine_tune carries the same field, and Use is refused there too. Neither Rust surface returns cache_outcome from TrainingJob::wait() — only model_id() is exposed there.

A stray cache key found under graph_fine_tune in a persisted jobs.spec row (the row is engine-written from an already-decoded spec, so this only arises from a hand-edited row) is silently dropped at deserialize rather than refused, since the type has nowhere to put it; making an unexpected key a hard error across the persisted-row format is a separate reshape (https://github.com/f-inverse/jammi-ai/issues/548).

Encoder-adapters fine-tuning (PEFT-style adapter injection)

The default flow above trains a single low-rank projection head sitting outside the frozen encoder. For higher capacity at the same parameter budget, Jammi also supports encoder adapters — LoRA injected into named linear layers inside the encoder stack, matching the PEFT convention.

Switch to encoder adapters by populating target_modules on FineTuneConfig:

#![allow(unused)]
fn main() {
extern crate jammi_ai;
use jammi_ai::fine_tune::FineTuneConfig;
fn make() -> FineTuneConfig {
let config = FineTuneConfig {
    lora_rank: 8,
    lora_alpha: 16.0,
    // Inject LoRA into BERT's attention query and value projections.
    target_modules: vec!["query".to_string(), "value".to_string()],
    ..Default::default()
};
config }
}

job = db.fine_tune(
    source="training",
    base_model="sentence-transformers/all-MiniLM-L6-v2",
    columns=["text_a", "text_b", "score"],
    method="lora",
    task="text_embedding",
    target_modules=["query", "value"],
)

Target-module conventions

Pick target_modules per the architecture you’re fine-tuning:

Names match the trailing module-name segment in the HuggingFace weight layout. Suffix matching is the rule, so "query" matches "attention.self.query".

in_proj on the two OpenCLIP towers is the fused QKV projection — one site covering query, key and value, the way Wqkv does on ModernBERT.

A target_modules list that matches nothing on the selected tower fails the job with an error naming that tower’s real site names, rather than training an adapter with zero parameters.

Fine-tuning an image or audio tower

The tower is selected by the job’s task, not by a separate flag: an image_embedding job on an OpenCLIP checkpoint fine-tunes its vision tower, a text_embedding job on the same checkpoint fine-tunes its text tower, and an audio_embedding job on an HF-CLAP checkpoint fine-tunes its HTSAT audio tower. A task the base checkpoint has no tower for is refused before training starts, with a message naming the towers it does have.

Media jobs read triplets of encoded bytes — three binary columns anchor, positive, negative holding whole files (PNG/JPEG/… for images, WAV/FLAC/ MP3/Ogg for audio). The modality comes from the declared task and is never sniffed from the bytes, so passing an image corpus to an audio_embedding job is a decoding error, not a silently mis-encoded run. The three groups are encoded as one joined forward pass and then split.

job = db.fine_tune(
    source="image_triplets",
    base_model="local:/models/open_clip_vit_b32",
    columns=["anchor", "positive", "negative"],
    method="lora",
    task="image_embedding",
    target_modules=["in_proj", "out_proj"],
)

What makes a blob a “positive” — an augmentation of the anchor, a co-occurring item — is your data’s concern; the triplet loss only separates whatever pairs you supply.

Layer ranges and per-module ranks

Two optional refinements:

• layers_to_transform — restrict injection to specific 0-based layer indices. None (default) applies to every layer.
• rank_pattern — override lora_rank for individual modules. Keys are substring matches against the module name; values are the override rank.

layers_to_transform indexes the first numbered segment of the weight name, matching PEFT’s own rule. On the BERT family and the two OpenCLIP towers that is the transformer layer. On the HTSAT audio tower, whose blocks are named layers.{stage}.blocks.{block}, it is the stage index. A site that sits in no numbered unit at all — the CLAP audio projection head’s linear1/linear2 — is excluded whenever layers_to_transform is set, again matching PEFT: a restriction to specific layers cannot be satisfied by a module that belongs to no layer.

#![allow(unused)]
fn main() {
extern crate jammi_ai;
use jammi_ai::fine_tune::FineTuneConfig;
fn make() -> FineTuneConfig {
let mut rank_pattern = std::collections::HashMap::new();
rank_pattern.insert("query".to_string(), 16);  // higher capacity on Q
rank_pattern.insert("value".to_string(), 4);   // lower on V

let config = FineTuneConfig {
    lora_rank: 8,                                     // default rank
    target_modules: vec!["query".into(), "value".into()],
    layers_to_transform: Some(vec![6, 7, 8, 9, 10, 11]), // top half only
    rank_pattern,
    ..Default::default()
};
config }
}

On-disk artifact

Every fine-tuned model writes adapter.safetensors plus an adapter_config.json whose adapter_type tag discriminates between the two adapter shapes Jammi produces.

Encoder-adapters example:

{
  "adapter_type": "encoder_adapters",
  "model_type": "bert",
  "lora_rank": 8,
  "lora_alpha": 16.0,
  "use_rslora": false,
  "target_modules": ["query", "value"],
  "layers_to_transform": [6, 7, 8, 9, 10, 11],
  "rank_pattern": {"query": 16, "value": 4},
  "backbone_dtype": "f32"
}

Projection-head example:

{
  "adapter_type": "projection_head",
  "lora_rank": 8,
  "lora_alpha": 16.0,
  "head_layers": ["projection"]
}

model_type records the base architecture the adapter was trained on — one of bert, distilbert, modernbert, open_clip (an OpenCLIP checkpoint, which ships no model_type field of its own) or clap_audio_model.

A checkpoint that holds more than one tower carries one extra key, tower, naming which one the adapter installs on:

{
  "adapter_type": "encoder_adapters",
  "model_type": "open_clip",
  "lora_rank": 8,
  "lora_alpha": 16.0,
  "use_rslora": false,
  "target_modules": ["in_proj", "out_proj"],
  "layers_to_transform": null,
  "rank_pattern": {},
  "backbone_dtype": "f32",
  "tower": "vision"
}

The key is written only when it applies: a single-tower adapter’s adapter_config.json carries no tower key at all.

The Candle inference backend reads adapter_config.json on model load and dispatches on adapter_type: encoder_adapters rebuilds the encoder with frozen backbone weights plus the LoRA A/B from adapter.safetensors; projection_head loads the saved projection weights as a LoraLinear applied after pooling.

Before any of that, the backend checks the adapter and the base agree on architecture family: an open_clip adapter on a CLAP base, a clap_audio_model adapter on a BERT base, or a tower the base checkpoint does not have, is refused with a typed error rather than loaded onto whatever the base happens to be. On an OpenCLIP base the adapted tower is rebuilt with the adapter’s weights and the sibling tower is rebuilt frozen at the same backbone precision — a fine-tuned model has one identity and one precision.

When to use each

• Projection head — fastest training, smallest artifact, lowest memory. The default when target_modules is empty. Best for adapting embedding direction without changing per-token attention behaviour.
• Encoder adapters — higher representational ceiling per adapter parameter; required if the task needs to reshape attention behaviour (e.g. a domain where the base attention pattern mismatches the query distribution). Costs a slightly slower forward pass since the LoRA path runs per layer.

QLoRA (encoder adapters over a quantized base)

base_model accepts a GGUF checkpoint (see Quantized (GGUF) checkpoints) the same way it accepts a safetensors checkpoint. When the resolved base is model.gguf, an encoder-adapters job trains its LoRA A/B matrices over the frozen quantized backbone automatically — the base artifact selects this, not a separate flag or config field. The quantized weights themselves are never trained (LoRA never updates a frozen base, quantized or dense); only the low-rank adapters receive gradients, exactly as with a dense base.

Training safety

• Divergence detection: if loss is NaN or >100 for 3 consecutive batches, the job fails with a clear error
• Early stopping: training stops when validation loss doesn’t improve for patience epochs, best checkpoint weights are restored
• Checkpoints: saved at ~10% intervals for crash recovery
• Multi-host runs (world_size > 1): a rank that aborts, drops its stream, or stays silent past [worker] rank_timeout_secs retires the whole attempt — no partial model is ever published and nothing terminal is written; the job is requeued from its last epoch checkpoint and the retry costs one attempt, unless the rank’s host was draining (a rolling restart), which costs none. A job whose ranks keep failing fails once its attempts are exhausted, never retries forever.

Memory

Training memory is dominated by the frozen backbone’s weights and activations, scaled by batch_size and max_seq_length. backbone_dtype defaults to f32 for numerical conservatism — every job runs the backbone at full precision unless you opt into a lower-precision dtype; the trained LoRA A/B matrices always stay f32 regardless of backbone_dtype, for numerical stability.

backbone_dtype only takes effect on the encoder-adapters arm (target_modules non-empty) — the projection-head arm (the default, empty target_modules) never re-dtypes the frozen backbone, so bf16 is not an available remedy there.

This guidance applies to the fine-tune training kinds (embedding and classification training), and only when the failure’s own error text carries a recognized out-of-memory spelling. Two cases it does NOT cover: a host OOM-kill that terminates the worker process outright leaves no error message at all to classify — the job is picked up by lease reclaim instead, not this guidance; and a ContextPredictor training run carries no OOM guidance at all (it doesn’t route through this classifier).

When it applies, the job’s terminal error is rewritten to name the exact batch_size, max_seq_length, and (on the encoder-adapters arm) backbone_dtype it ran with, and suggests remedies in the order they’re cheapest to try:

• Encoder adapters: (1) backbone_dtype: bf16 — substantially reduces memory on this arm. Requires a CUDA device; a bf16 backbone on a non-CUDA device is refused before training starts, rather than silently falling back to f32. (2) A smaller batch_size, or trade batch size for gradient_accumulation_steps to hold the same effective batch size while shrinking the per-step activation memory. (3) A smaller max_seq_length.
• Projection head (default): backbone_dtype does not apply and is omitted from the message — the message says so outright. (1) A smaller batch_size, or trade batch size for gradient_accumulation_steps. (2) A smaller max_seq_length.

For a fine-tune job whose failure was classified this way, jammi jobs status (and the Python job.status()) surfaces the rewritten message directly, so you don’t need to read raw driver output to find the fix.

Fine-Tune from a Graph (Graph-Supervised)

Measured companion: for the long-form, executed-and-measured Python treatment, see The Cookbook → Representation Learning on Graphs.

Fine-tune embeddings so that graph-neighbours are close in embedding space. This is node2vec / DeepWalk realised as Jammi config: it samples a graph into contrastive (anchor, positive, [hard_negative]) pairs and feeds them through the existing fine-tune trainer. It authors no GNN — no message passing, no new loss — it is a new training-data shape (TrainingFormat::Graph) that drives the same in-batch-negative (MNRL) / triplet objective as Fine-Tune for Your Domain.

Use it when your supervision is a graph rather than hand-built pairs: a hierarchy, a crosswalk, a citation network, a set of coder-confirmed matches, or the neighbour graph Jammi itself builds.

The load-bearing caveat: where the signal comes from

Declared edges teach; similarity edges echo.

If you train on S9-similarity edges (the neighbour graph, which is k-NN under the base embedding metric), the walk-positives are mostly “things the model already thinks are close” — so fine-tuning largely re-learns the base metric. That is a degenerate feedback loop with little new signal.

Genuine gain comes from declared / external edges — structure the base metric does not already encode:

• a hierarchy (parent/child categories),
• a crosswalk (version-A code ↔ version-B code),
• a citation / reference network,
• coder-confirmed pairs.

Tag your edges with their provenance. Similarity edges are an acceptable weak bootstrap (e.g. to expand a sparse declared graph), but never the sole supervision. The sampler tracks provenance and can report whether any declared edge is present.

Prepare the graph

Two sources: node text (what the encoder embeds) and edges.

nodes.csv — every node must be text-bearing (the encoder needs text; pure-vector nodes are out of scope here):

id,text
c01,"acute myocardial infarction, initial"
c02,"acute myocardial infarction, subsequent"
c03,"benign essential hypertension"

edges.csv — directed edges; endpoints join to id:

src,dst
c01,c02
c02,c01

Register both as sources:

Python

db.add_source("nodes", path="/data/nodes.csv", format="csv")
db.add_source("edges", path="/data/edges.csv", format="csv")

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
use jammi_ai::session::InferenceSession;
use jammi_db::source::{FileFormat, SourceConnection, SourceType};
async fn ex(session: &InferenceSession) -> jammi_db::error::Result<()> {
for name in ["nodes", "edges"] {
    session.add_source(name, SourceType::File, SourceConnection {
        url: Some(format!("file:///data/{name}.csv")),
        format: Some(FileFormat::Csv),
        ..Default::default()
    }).await?;
}
Ok(()) }
}

Run the graph fine-tune

The sampler runs biased random walks (node2vec) over the edges: from each node it walks walk_length (L) steps, biased by the return parameter p and the in-out parameter q, and treats co-walked nodes as positives. L > 1 captures higher-order / community structure — L = 1 is the degenerate 1-hop case. Negatives are in-batch (every other pair’s positive) plus structure-mined hard negatives drawn from outside the anchor’s exclude_hops-hop neighbourhood (the false-negative guard — a node inside that radius is likely a missing edge, i.e. a true positive).

Python

job = db.fine_tune_graph(
    node_source="nodes", id_column="id", text_column="text",
    edge_source="edges", src_column="src", dst_column="dst",
    base_model="local:/models/tiny_bert",
    edge_provenance="declared",   # "declared" teaches; "similarity" echoes
    walk_length=4, walks_per_node=2, return_p=1.0, in_out_q=1.0,
    graph_hard_negatives=1, exclude_hops=1, min_negatives=1,
    embedding_loss="mnrl",        # in-batch negatives (default); or "triplet"
    epochs=3, batch_size=8,
)
job.wait()

Rust

#![allow(unused)]
fn main() {
extern crate jammi_ai;
extern crate jammi_db;
use jammi_ai::session::InferenceSession;
use jammi_ai::fine_tune::FineTuneConfig;
use jammi_ai::fine_tune::graph_sampler::{EdgeProvenance, GraphFineTuneSources, GraphSampleConfig};
async fn ex(session: &InferenceSession) -> jammi_db::error::Result<()> {
let sources = GraphFineTuneSources {
    node_source: "nodes".into(), id_column: "id".into(), text_column: "text".into(),
    edge_source: "edges".into(), src_column: "src".into(), dst_column: "dst".into(),
    // Declared edges carry signal the base metric does not already encode.
    provenance: EdgeProvenance::Declared,
};
let sample = GraphSampleConfig {
    walk_length: 4, walks_per_node: 2, return_p: 1.0, in_out_q: 1.0,
    hard_negatives: 1, exclude_hops: 1, min_negatives: 1, seed: 0,
};
let job = session
    .fine_tune_graph(&sources, "local:/models/tiny_bert", sample, Some(FineTuneConfig::default()))
    .await?;
job.wait().await?;
Ok(()) }
}

The output is a fine-tuned model; regenerate embeddings with it and they encode the graph’s structure (build_neighbor_graph, search, and propagation all benefit).

Tuning knobs

Compose with propagation

Both graph fine-tune and embedding propagation encode homophily; stacking them naively double-counts the same smoothing. The recommended order is propagate first, then fine-tune the head (the SGC/APPNP decoupling) — not two independent smoothing passes.

Did it work? The circularity check

To confirm declared edges actually helped (and that you did not just re-learn the base metric), evaluate on a held-out golden set — see Did Structure Help? A Graph-ML Evaluation Recipe:

1. Build two supervision graphs over the same nodes — one from declared edges, one from S9-similarity edges.
2. fine_tune_graph each; hold out a golden relevance set.
3. eval_embeddings the base model vs each fine-tune, with a paired significance test.
4. Expect the declared-edge model to beat the base significantly, and the similarity-edge model’s gain to be near-zero — the degenerate feedback loop, measured rather than assumed.

Evaluate and Compare Models

Measured companion: for the long-form, executed-and-measured Python treatment, see The Cookbook → Evaluation & Provenance Channels.

Measure embedding quality and classification accuracy against golden datasets. Results are recorded in the catalog for tracking over time.

Prepare a golden dataset

A golden dataset is any registered source with the right columns. No special format required.

Retrieval golden set

query_id,query_text,relevant_id
q1,quantum computing applications,1
q1,quantum computing applications,4
q2,machine learning for science,2

Register it as a source:

db.add_source("golden", path="/data/golden_relevance.csv", format="csv")

Evaluate embedding quality

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use jammi_ai::session::InferenceSession;
async fn ex(session: &InferenceSession) -> jammi_db::error::Result<()> {
let report = session.eval_embeddings(
    "patents",
    None,                                // use latest embedding table
    "golden.public.golden_relevance",    // golden dataset
    10,                                  // k for recall@k, precision@k
    &std::collections::HashMap::new(),   // no cohort tags
).await?;

println!("recall@10:    {}", report.aggregate.recall_at_k);
println!("precision@10: {}", report.aggregate.precision_at_k);
println!("MRR:          {}", report.aggregate.mrr);
println!("nDCG:         {}", report.aggregate.ndcg);
Ok(()) }
}

Python

metrics = db.eval_embeddings(
    source="patents",
    golden_source="golden.public.golden_relevance",
    k=10,
)

agg = metrics["aggregate"]
print(f"recall@10:    {agg['recall_at_k']:.3f}")
print(f"precision@10: {agg['precision_at_k']:.3f}")
print(f"MRR:          {agg['mrr']:.3f}")
print(f"nDCG:         {agg['ndcg']:.3f}")

Per-query drill-down

The report also carries a per_query array — one record per golden-set query, in golden order. This is what sample-based statistical rules (Welch’s t, Mann-Whitney U) consume at gate time.

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use jammi_ai::session::InferenceSession;
async fn ex(session: &InferenceSession) -> jammi_db::error::Result<()> {
let report = session.eval_embeddings("patents", None, "golden.public.golden_relevance", 10, &std::collections::HashMap::new()).await?;
for record in &report.per_query {
    println!("{}: recall={:.3} ndcg={:.3}",
        record.query_id, record.metrics.recall, record.metrics.ndcg);
}
Ok(()) }
}

for record in metrics["per_query"]:
    m = record["metrics"]
    print(f"{record['query_id']}: recall={m['recall']:.3f} ndcg={m['ndcg']:.3f}")

Retrieval metrics

All metrics are in [0, 1]. Higher is better.

Compare models (A/B)

Compare a base model against a fine-tuned model:

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use jammi_ai::session::InferenceSession;
async fn ex(session: &InferenceSession, base_table: String, finetuned_table: String) -> jammi_db::error::Result<()> {
let comparison = session.eval_compare(
    &[base_table.clone(), finetuned_table.clone()],
    "patents",
    "golden.public.golden_relevance",
    10,
).await?;

// The first entry is the baseline (`delta: None`); every subsequent entry
// carries a delta against it.
for entry in comparison.per_table.iter().skip(1) {
    let delta = entry.delta.as_ref().expect("non-baseline entries carry a delta");
    println!(
        "{}: recall@10 delta {:+.3} ({:+.1}%)",
        entry.table_name,
        delta.recall_at_k.absolute,
        delta.recall_at_k.relative * 100.0,
    );
}
Ok(()) }
}

Python

comparison = db.eval_compare(
    embedding_tables=[base_table, finetuned_table],
    source="patents",
    golden_source="golden.public.golden_relevance",
    k=10,
)
# `per_table[0]` is the baseline (`delta` is None); subsequent entries
# carry a `delta` dict keyed by metric name (recall_at_k, precision_at_k,
# mrr, ndcg) with `absolute` and `relative` sub-keys.
for entry in comparison["per_table"][1:]:
    d = entry["delta"]["recall_at_k"]
    print(f"{entry['table_name']}: recall@10 delta {d['absolute']:+.3f} ({d['relative']*100:+.1f}%)")

The first table is the baseline. Deltas (absolute and relative) are computed for all subsequent tables.

Evaluate classification

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use jammi_ai::session::InferenceSession;
async fn ex(session: &InferenceSession) -> jammi_db::error::Result<()> {
use jammi_ai::eval::{EvalTask, InferenceAggregate};

let report = session.eval_inference(
    "facebook/bart-large-mnli",
    "test_data",
    &["text".into()],
    EvalTask::Classification,
    "golden.public.labels",
    "category",
).await?;

match &report.aggregate {
    InferenceAggregate::Classification(c) => {
        println!("Accuracy: {}", c.accuracy);
        println!("Macro F1: {}", c.f1);
    }
    InferenceAggregate::Ner(n) => {
        println!("NER F1: {}", n.f1);
    }
}
println!("per_record predictions: {}", report.per_record.len());
Ok(()) }
}

Python

metrics = db.eval_inference(
    model="facebook/bart-large-mnli",
    source="test_data",
    columns=["text"],
    task="classification",
    golden_source="golden.public.labels",
    label_column="category",
)

# `aggregate` is tagged by `task`; for classification it carries
# `accuracy`, `f1`, and `per_class`.
agg = metrics["aggregate"]
print(f"Accuracy: {agg['accuracy']:.3f}")
print(f"Macro F1: {agg['f1']:.3f}")
# `per_record` is one entry per aligned predicted/gold pair.
print(f"per_record predictions: {len(metrics['per_record'])}")

Eval runs in the catalog

Every evaluation is recorded automatically:

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use jammi_ai::session::InferenceSession;
async fn ex(session: &InferenceSession) -> jammi_db::error::Result<()> {
let runs = session.catalog().list_eval_runs().await?;
for run in &runs {
    println!("{}: {} on {} (k={:?})", run.eval_run_id, run.eval_type, run.golden_source, run.k);
}
Ok(()) }
}

Schema validation

Golden datasets are validated before evaluation starts. Missing or wrong-type columns produce clear errors:

Eval error: Golden dataset missing required column 'query_text'
Eval error: Golden dataset column 'query_id' has type Boolean, expected Utf8

Integer ID columns (Int32, Int64) are accepted where Utf8 is expected.

Did Structure Help? A Graph-ML Evaluation Recipe

When you produce a structure-aware embedding table — a fine-tuned model, a propagated table, or any treatment that folds graph context into the representation — the only question that matters is whether it beats plain text embeddings on a held-out retrieval task. This recipe is the discipline around eval_compare that turns a delta into a defensible conclusion.

Nothing here is new engine surface. eval_compare already computes recall@k / precision@k / MRR / nDCG per table, the per-metric delta against a baseline, and — paired by query_id over the per-query records — a distribution-free significance result for each metric delta. The recipe is the protocol: a clean split, the judgment-matched metric, multiple seeds for trained treatments, and cohort slicing.

The four steps

1. Baseline. Produce a plain text-embedding table with generate_text_embeddings.
2. Treatment. Produce each structure-aware table (a fine-tuned model’s embeddings, a propagated table, etc.) over the same source rows.
3. Compare. Run eval_compare with the baseline table first. Read the per-metric delta and its paired significance.
4. Conclude and slice. Declare a win only when the judgment-matched metric improves with a significant paired test; then slice by cohort to see where structure helped.

1 & 2 — produce baseline and treatment tables

The baseline is a plain text-embedding table over your corpus. Each treatment table must be built over the same rows so the comparison is apples-to-apples.

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use jammi_ai::session::InferenceSession;
use jammi_db::store::CachePolicy;
async fn ex(session: &InferenceSession, baseline_model: &str, treatment_model: &str) -> jammi_db::error::Result<()> {
// Baseline: plain text embeddings.
let (baseline, _) = session
    .generate_text_embeddings("patents", baseline_model, &["abstract".into()], "id", CachePolicy::Bypass, None)
    .await?;

// Treatment: embeddings from a structure-aware model (e.g. a fine-tuned
// checkpoint), over the same source and key column.
let (treatment, _) = session
    .generate_text_embeddings("patents", treatment_model, &["abstract".into()], "id", CachePolicy::Bypass, None)
    .await?;

let baseline_table = baseline.table_name;
let treatment_table = treatment.table_name;
let _ = (baseline_table, treatment_table);
Ok(()) }
}

baseline = db.generate_embeddings(
    source="patents", model=baseline_model,
    columns=["abstract"], key="id", modality="text",
)
treatment = db.generate_embeddings(
    source="patents", model=treatment_model,
    columns=["abstract"], key="id", modality="text",
)

Leakage contract (read before you build anything). The graph and any graph-supervised training must use the train split only. The golden set — the (query, judgments) pairs you evaluate against — is held out and must never feed graph construction or training-pair selection. A structure-aware representation that has seen the eval rows will “win” by memorizing them, and the verdict is worthless. Split first, then build.

3 — compare on a held-out golden set

Run eval_compare with the baseline table first; every subsequent table carries its delta against that baseline. The golden set is a registered source of (query_id, query_text, relevant_id[, relevance_grade]) rows — see Evaluate and Compare Models for its schema.

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use jammi_ai::session::InferenceSession;
async fn ex(session: &InferenceSession, baseline_table: String, treatment_table: String) -> jammi_db::error::Result<()> {
let comparison = session
    .eval_compare(
        &[baseline_table, treatment_table], // baseline FIRST
        "patents",
        "golden.public.golden_relevance",   // held-out golden set
        10,                                 // k for recall@k / precision@k
    )
    .await?;

for entry in comparison.per_table.iter().skip(1) {
    let delta = entry.delta.as_ref().expect("non-baseline entries carry a delta");
    println!(
        "{}: nDCG {:+.3} ({:+.1}%)",
        entry.table_name,
        delta.ndcg.absolute,
        delta.ndcg.relative * 100.0,
    );

    // The paired significance of each metric delta. `None` only when the two
    // runs share no `query_id` (nothing to pair).
    if let Some(sig) = delta.significance.as_ref() {
        let s = &sig.ndcg;
        println!(
            "  nDCG p={:.4}  95% CI [{:+.3}, {:+.3}]",
            s.p_value, s.ci_lower, s.ci_upper,
        );
    }
    let _ = entry;
}
Ok(()) }
}

comparison = db.eval_compare(
    embedding_tables=[baseline_table, treatment_table],  # baseline FIRST
    source="patents",
    golden_source="golden.public.golden_relevance",
    k=10,
)
for entry in comparison["per_table"][1:]:
    delta = entry["delta"]
    d = delta["ndcg"]
    print(f"{entry['table_name']}: nDCG {d['absolute']:+.3f} ({d['relative']*100:+.1f}%)")
    sig = delta.get("significance")
    if sig is not None:
        s = sig["ndcg"]
        print(f"  nDCG p={s['p_value']:.4f}  95% CI [{s['ci_lower']:+.3f}, {s['ci_upper']:+.3f}]")

Reading the significance

For each metric, eval_compare attaches a MetricSignificance carrying a p_value and a [ci_lower, ci_upper] interval:

• ci_lower / ci_upper are a percentile bootstrap confidence interval on the mean paired difference (treatment − baseline), at the 95% level. A CI that lies entirely above zero is the resampling analogue of “the delta is real, not noise.” A CI that brackets zero means you cannot distinguish the treatment from the baseline on this metric.
• p_value is the two-tailed Mann–Whitney U p-value comparing the baseline and treatment per-query distributions — distribution-free, and robust to the bounded, tie-heavy shape retrieval metrics have. Smaller is stronger evidence.

Both are deterministic: the bootstrap runs under a pinned seed and a fixed iteration count, so the same inputs always yield the same interval. Two identical runs collapse to a [0, 0] CI with p ≈ 1.

A delta of +0.02 is a headline, not a conclusion. Report it only as +0.02, p=0.003, CI [+0.008, +0.031] — the delta with its significance.

Discipline contracts

These are the contracts a “did structure help?” claim must satisfy. State each one explicitly when you report a result.

Strict held-out split (the leakage contract)

The graph and any graph-supervised training use the train split only; the golden set is held out and never feeds construction. This is the single most important contract — restated here because it is the one that silently invalidates a result. If you cannot point to the split boundary, you do not have a clean number.

Judgment-matched metric

Pick the metric from the judgment type, not by habit:

eval_compare always computes all four metrics, but read the one that matches your golden set. Using recall on graded judgments throws away signal you paid a human to produce.

≥3 seeds for trained treatments

A trained treatment (a fine-tuned checkpoint, or any treatment whose construction samples) varies by seed. One lucky seed can fake a win. Run the treatment under ≥3 seeds, compare each against the same baseline, and report the mean ± variance of the delta plus the significance across seeds — not a single run.

A deterministic treatment (e.g. a pure propagation with no sampling) does not vary by seed, so it needs only the leakage and significance discipline — a quiet advantage worth stating when it applies.

Cohort slicing

The most useful output is where structure helped — by source family, period, or any segment you care about. Tag each query with cohort labels at eval time, then group the persisted per-query records.

Cohort tags are supplied per query to eval_embeddings (the per-table entry point); eval_compare itself does not surface cohort tagging, so to slice a comparison you run each table through eval_embeddings with the same cohort map. Every per-query record — its metrics and its cohort tags — is persisted to _jammi_eval_per_query, keyed by the run’s eval_run_id, and read back with eval_per_query.

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use std::collections::{BTreeMap, HashMap};
use jammi_ai::session::InferenceSession;
async fn ex(session: &InferenceSession, treatment_table: &str) -> jammi_db::error::Result<()> {
// Tag each query with its cohort(s) at eval time.
let mut cohorts: HashMap<String, BTreeMap<String, String>> = HashMap::new();
cohorts.insert(
    "q1".into(),
    BTreeMap::from([("family".into(), "A".into())]),
);
// ... one entry per query_id ...

let report = session
    .eval_embeddings(
        "patents",
        Some(treatment_table),
        "golden.public.golden_relevance",
        10,
        &cohorts,
    )
    .await?;

// Read the persisted per-query rows back by run id; each row carries its
// metrics and cohort tags as JSON, ready to group by cohort.
let rows = session.eval_per_query(&report.eval_run_id).await?;
for row in &rows {
    println!("{}: cohorts={} metrics={}", row.query_id, row.cohorts_json, row.metrics_json);
}
Ok(()) }
}

cohorts = {"q1": {"family": "A"}, "q2": {"family": "B"}}  # one entry per query_id
report = db.eval_embeddings(
    source="patents",
    embedding_table=treatment_table,
    golden_source="golden.public.golden_relevance",
    k=10,
    cohorts=cohorts,
)
rows = db.eval_per_query(report["eval_run_id"])
# Group `rows` by their cohort tags and aggregate per-cohort metrics.

Then group by cohort and report per-cohort n and a confidence interval — a 12-query cohort has a wide CI, and a swing inside it is not a finding. Small cohorts go noisy; report n so a reader does not over-read them.

What this recipe does not cover

• New metrics. Recall / precision / MRR / nDCG suffice; this recipe does not add others.
• Online drift monitoring. This is an offline held-out harness, not a production-drift monitor.
• The golden set itself. Constructing relevance judgments is the expensive human step — budget for it. The eval is cheap; the labels are not.

See also

• Evaluate and Compare Models — the golden-set schema, the metric definitions, and the full eval_compare / eval_embeddings API.
• Fine-Tune for Your Domain — producing a trained treatment table to evaluate with this recipe.

Evaluate Uncertainty and Calibration

Measured companion: for the long-form, executed-and-measured Python treatment, see The Cookbook → Calibration & Uncertainty.

eval_embeddings and eval_inference answer “is the prediction accurate?”. eval_calibration answers the orthogonal question — “does the prediction know what it doesn’t know?”. The two are independent: a model can be accurate and badly calibrated, or perfectly calibrated and useless. When a predictor emits a distribution or an interval, a point-accuracy metric cannot tell you whether that uncertainty is honest. This harness can.

What it reports

Every calibration eval reports three things together — reporting any one alone is a trap:

• A proper score (the headline). CRPS (continuous ranked probability score) and NLL (negative log-likelihood). Strictly proper scores are uniquely minimised by the true distribution, so they reward calibration and sharpness jointly — the only safe headline metric.
• A calibration diagnostic. The adaptive, debiased PIT-calibration error: under calibration the probability-integral-transform of the outcomes is uniform, and this scores its departure from uniform. It is a diagnostic, never the verdict — reporting it alone admits the marginal-predictor degenerate (a model that predicts the global average is perfectly calibrated and worthless).
• Sharpness and coverage. The mean width of the nominal 90% interval and how often it actually contains the outcome. Sharper is better only at fixed coverage.

The held-out, three-way-split contract

Calibration is measured on a held-out test set that is disjoint from both the training data and any calibration set used to fit the predictor. Re-using calibration points to also test inflates coverage — it is the single most common conformal/calibration bug. The harness measures exactly the predictions you give it; the split discipline is yours.

Prepare a calibration golden set

A calibration golden set pairs a held-out predictive distribution with its realised outcome. Two predictor shapes are supported, each reading different columns.

Parametric (Gaussian) predictor

For a predictor that emits a predictive Normal(mean, sd) per record:

record_id,mean,sd,outcome
r1,4.2,0.5,4.0
r2,1.1,0.9,2.3
r3,7.8,0.3,7.7

Ensemble (Sample) predictor

For a predictor that emits an ensemble of predictive draws per record, store the draws as a JSON array in a draws column:

record_id,draws,outcome
r1,"[3.9, 4.1, 4.3, 4.0]",4.0
r2,"[0.8, 1.4, 1.0, 2.1]",2.3

Register either as a source:

db.add_source("calib", path="/data/calibration_holdout.csv", format="csv")

Run the eval

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
use jammi_ai::session::InferenceSession;
use jammi_ai::eval::EvalCalibrationShape;
async fn ex(session: &InferenceSession) -> jammi_db::error::Result<()> {
let report = session.eval_calibration(
    "patents",                          // source under test
    "calib.public.calibration_holdout", // held-out predictions + outcomes
    EvalCalibrationShape::Gaussian,     // or ::Sample for an ensemble
    &std::collections::HashMap::new(),  // no cohort tags
).await?;

// The proper score is the headline; the diagnostics explain it.
println!("CRPS (headline):  {}", report.aggregate.crps);
println!("NLL:              {}", report.aggregate.nll);
println!("PIT-calibration:  {}", report.aggregate.adaptive_ece);
println!("sharpness (90%):  {}", report.aggregate.sharpness);
println!("coverage (90%):   {}", report.aggregate.coverage);
Ok(())
}
}

Python

report = db.eval_calibration(
    "patents",
    "calib.public.calibration_holdout",
    shape="gaussian",   # or "sample"
)
print("CRPS:", report["aggregate"]["crps"])
print("coverage:", report["aggregate"]["coverage"])

Slice by cohort

Marginal coverage hides conditional miscoverage: a predictor can hit 90% coverage globally while systematically under-covering a subgroup. Tag records with opaque cohort segments — keyed by record_id — and the report slices coverage and CRPS per cohort, each with its sample size n and a bootstrap confidence interval on the proper score, so a small cohort is not over-read.

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
use std::collections::{BTreeMap, HashMap};
use jammi_ai::session::InferenceSession;
use jammi_ai::eval::EvalCalibrationShape;
async fn ex(session: &InferenceSession) -> jammi_db::error::Result<()> {
let mut cohorts: HashMap<String, BTreeMap<String, String>> = HashMap::new();
cohorts.insert(
    "r1".to_string(),
    BTreeMap::from([("region".to_string(), "emea".to_string())]),
);

let report = session
    .eval_calibration(
        "patents",
        "calib.public.calibration_holdout",
        EvalCalibrationShape::Gaussian,
        &cohorts,
    )
    .await?;

for cohort in &report.per_cohort {
    println!(
        "{}={}: n={} coverage={} crps={}",
        cohort.key, cohort.value, cohort.n, cohort.coverage, cohort.crps
    );
}
Ok(())
}
}

Compare two predictors with a p-value

The per-record scores are persisted to _jammi_eval_per_query keyed by the run id, exactly like the embedding eval. Pairing the per-record CRPS of two runs by record_id and running the same distribution-free paired significance test the retrieval comparison uses turns “B is better-calibrated than A” into a CRPS delta with a confidence interval and a p-value — not a vibe.

Determinism

Given the same inputs on the same host the report is bit-for-bit reproducible: every scoring function is deterministic and the only randomness — the cohort confidence-interval bootstrap — runs under a pinned seed. The scoring folds are f32/f64 reductions, so they are a same-host guarantee, not a cross-host one — a different CPU host can produce the same score at the value level while differing in the last float bits.

Register a Mutable Companion Table

Measured companion: for the long-form, executed-and-measured Python treatment, see The Cookbook → Feature Store.

A mutable companion table lives in the same backend database as the Jammi catalog (SQLite by default, Postgres in shared deployments), supports transactional INSERT / UPDATE / DELETE through DataFusion DML, and federates with Parquet result tables and external sources in one SQL surface. Reach for it when a tenant needs a relation it can edit row by row — a feature-store slowly-changing dimension table, a per-user state table, a config-driven lookup — that still has to participate in the same JOINs as your immutable result tables.

The primitive carries only what every consumer needs: a schema, a primary key, optional tenant scope, optional secondary indexes, optional ordering column. No history semantics, no lifecycle vocabulary, no audit columns.

Goal

This recipe walks through registering one mutable companion table for a neutral third-party use case (a feature-store team called Polaris Features maintaining slowly-changing dimensions for their recommender) and shows the equivalent Rust / Python / CLI surface.

Setup

Assumes a working JammiSession. The session opens the catalog at the configured artifact directory; nothing else is needed.

Define the schema

Polaris keeps one row per (item_id, valid_from, valid_to) interval:

#![allow(unused)]
fn main() {
extern crate arrow_schema;
fn make() {
use std::sync::Arc;
use arrow_schema::{DataType, Field, Schema};

let schema = Arc::new(Schema::new(vec![
    Field::new("item_id",      DataType::Utf8,    false),
    Field::new("price_tier",   DataType::Utf8,    false),
    Field::new("availability", DataType::Utf8,    false),
    Field::new("valid_from",   DataType::Int64,   false),  // epoch milliseconds
    Field::new("valid_to",     DataType::Int64,   true),   // epoch milliseconds; NULL = open
]));
}
}

The catalog encoder accepts the closed primitive subset enforced by every MutableBackend impl — Boolean, the integer family, Float32 / Float64, Utf8, Binary. Wider types (e.g. Timestamp, Decimal) round-trip via their natural numeric encoding (Int64 epoch milliseconds, scaled Int64) so the schema stays narrow and the rule stays one-line at the boundary.

The engine reserves tenant_id and any column whose name starts with _ — the schema builder rejects them at build time, per the tenant-identifier discipline (Design Philosophy). (The tenant_id column is always present on the storage table; the engine appends it implicitly.)

Build the definition

MutableTableDefinitionBuilder chains the field validations:

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate arrow_schema;
use std::sync::Arc;
use arrow_schema::Schema;
use jammi_db::store::mutable::definition::{
    MutableIndexDef, MutableTableDefinitionBuilder, MutableTableId,
};

fn make(schema: Arc<Schema>) -> jammi_db::store::mutable::definition::MutableTableDefinition {
let def = MutableTableDefinitionBuilder::new(
        MutableTableId::new("item_dimensions").unwrap(),
        schema,
    )
    .primary_key(vec!["item_id".into(), "valid_from".into()])
    .index(MutableIndexDef {
        name: "idx_item_dim_active".into(),
        columns: vec!["item_id".into(), "valid_to".into()],
        unique: false,
    })
    .build()
    .unwrap();
def
}
}

The primary key must be a non-empty subset of the schema; secondary indexes are optional but persisted on the storage table so the backend can use them for WHERE clauses.

Register

The registration is atomic: catalog row + storage CREATE TABLE + every secondary CREATE INDEX commit together. If any step fails, nothing lands.

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate tokio;
use jammi_db::store::mutable::definition::MutableTableDefinition;
use jammi_db::session::JammiSession;
async fn ex(session: &JammiSession, def: MutableTableDefinition) -> jammi_db::error::Result<()> {
let id = session.create_mutable_table(def).await?;
// The table is now queryable as `mutable.public.item_dimensions` in the
// same SQL surface that federates result tables and external sources.
Ok(())
}
}

Python

import pyarrow as pa
import jammi

db = jammi.connect("file:///var/lib/jammi")
# The Python wrapper exposes mutable-table registration through the
# `create_mutable_table` accessor (see `jammi.mutable`). The recipe below
# is illustrative; consult the API reference for the binding shape your
# version ships.

CLI

The jammi CLI exposes mutable-table registration through the lower-level sources surface for now; programmatic clients should use the Rust or Python APIs.

Verify

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate tokio;
async fn ex(session: &jammi_db::session::JammiSession) -> jammi_db::error::Result<()> {
let zero_rows = session
    .sql("SELECT * FROM mutable.public.item_dimensions LIMIT 0")
    .await?;
assert_eq!(zero_rows[0].schema().fields().len(), 5);
Ok(())
}
}

The query returns a zero-row batch with the declared schema — confirmation that the table is registered and DataFusion can route mutable.public.<id> correctly.

Federation tease

The mutable table now JOINs with your existing result tables and sources:

SELECT  d.item_id, d.price_tier, e.embedding
FROM    mutable.public.item_dimensions d
JOIN    itemembs.public.item_embeddings e ON e.item_id = d.item_id
WHERE   d.valid_to IS NULL
  AND   d.price_tier = 'premium'
LIMIT 10;

See the Run Transactional Updates on a Mutable Table recipe for INSERT / UPDATE / DELETE round-trips and the SCD Type 2 close-and-open pattern.

Run Transactional Updates on a Mutable Table

Once a mutable companion table is registered (see Register a Mutable Companion Table), you update its rows with the same SQL surface that runs your read queries. Every INSERT / UPDATE / DELETE lands in one backend transaction — either every row commits or none does — and federates with your immutable result tables in subsequent SELECTs.

Goal

Walk through the three DML verbs against the item_dimensions table from the previous recipe, then demonstrate the Slowly-Changing Dimension Type 2 close-and-open pattern Polaris uses to record a price-tier change.

Insert

INSERT INTO mutable.public.item_dimensions
    (item_id, price_tier, availability, valid_from)
VALUES
    ('sku-1842', 'standard', 'in_stock', '2026-04-01T00:00:00Z'),
    ('sku-2901', 'premium',  'in_stock', '2026-04-01T00:00:00Z'),
    ('sku-3457', 'standard', 'out_of_stock', '2026-04-01T00:00:00Z');

The RecordBatch returned by session.sql(...) carries a single-row UInt64 column called count per DataFusion’s TableProvider::insert_into contract. Three rows landed; the JOIN-against-result-tables query from the previous recipe now returns three rows.

Update

UPDATE mutable.public.item_dimensions
   SET availability = 'low_stock'
 WHERE item_id = 'sku-2901';

Predicate columns that participate in an index pushdown become a backend WHERE clause; the rest filter above the scan node. The update commits in one transaction; if the predicate matches zero rows, the call succeeds with rows_affected = 0.

Delete

DELETE FROM mutable.public.item_dimensions
 WHERE item_id = 'sku-3457';

DELETE follows the same shape. Row-level cascades are SQLite’s job (the foreign-key declarations on the storage table); the engine does not model cascades above the backend.

SCD Type 2 — close-and-open

Polaris records a price-tier change by closing the active row’s valid_to and inserting a new row with the new tier. Both statements must land atomically; today the supported pattern is to issue them as a single multi-statement SQL string through session.sql, which DataFusion plans as one DML batch under one transaction:

-- Single sql() call so both statements land in one transaction.
UPDATE mutable.public.item_dimensions
   SET valid_to = '2026-05-15T12:00:00Z'
 WHERE item_id = 'sku-1842' AND valid_to IS NULL;

INSERT INTO mutable.public.item_dimensions
    (item_id, price_tier, availability, valid_from)
VALUES
    ('sku-1842', 'premium', 'in_stock', '2026-05-15T12:00:00Z');

A future JammiSession::transaction(|tx| async { … }) API will make multi-statement DML atomicity explicit; today the multi-statement SQL string is the supported surface.

Federation join

The mutable table now joins with the embedding table to surface recommender candidates filtered by current tier:

SELECT  d.item_id, d.price_tier, e.embedding
  FROM  mutable.public.item_dimensions d
  JOIN  itemembs.public.item_embeddings e ON e.item_id = d.item_id
 WHERE  d.valid_to IS NULL
   AND  d.price_tier = 'premium'
 LIMIT 10;

The federation is the engine’s existing FederationOptimizerRule work — no special integration needed; mutable tables register under the same SessionContext as your Parquet result tables and external sources.

Crash recovery

If the process dies mid-write, no partial commit is visible on restart. SQLite’s WAL mode (documentation) and Postgres’s MVCC each guarantee that an open transaction either commits as a whole or is rolled back on connection loss. The engine inherits that guarantee through the CatalogBackend::transaction closure shape: when the closure returns Err(_), the backend rolls back; when the process is killed mid-execution, the backend rolls back the in-flight transaction.

Direct-access append + replay (Phase 4 trigger streams)

Two lower-level methods bypass DataFusion’s planner for high-throughput event paths:

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate arrow;
extern crate tokio;
async fn ex(
    session: &jammi_db::session::JammiSession,
    batch: arrow::array::RecordBatch,
) -> jammi_db::error::Result<()> {
use jammi_db::store::mutable::definition::MutableTableId;
use jammi_db::catalog::backend::TxOptions;

let id = MutableTableId::new("events").unwrap();
let registry = session.mutable_tables_arc();
let backend = session.catalog().backend_arc();

// Direct INSERT via insert_batch — caller owns the transaction.
backend
    .transaction(TxOptions::default(), move |tx| {
        let registry = registry.clone();
        let id = id.clone();
        let batch = batch.clone();
        Box::pin(async move {
            registry
                .insert_batch(tx, &id, &batch)
                .await
                .map_err(|e| jammi_db::BackendError::Execution(e.to_string()))?;
            Ok::<(), jammi_db::BackendError>(())
        })
    })
    .await?;
Ok(())
}
}

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate futures;
extern crate tokio;
use futures::StreamExt;
async fn ex(
    session: &jammi_db::session::JammiSession,
) -> jammi_db::error::Result<()> {
use jammi_db::store::mutable::definition::MutableTableId;

let id = MutableTableId::new("events").unwrap();
// Stream rows where the registered `order_column` value > 100.
let mut stream = session
    .mutable_tables()
    .scan_after(&id, 100)
    .await
    .map_err(|e| jammi_db::error::JammiError::Catalog(e.to_string()))?;
while let Some(batch) = stream.next().await {
    let _batch = batch
        .map_err(|e| jammi_db::error::JammiError::Catalog(e.to_string()))?;
    // …
}
Ok(())
}
}

These are the surface Phase 4’s trigger broker uses to publish events into a backing table and replay subscribers; general consumers should prefer the SQL surface.

Publish Events to a Topic

Measured companion: for the long-form, executed-and-measured Python treatment, see The Cookbook → Change Data Capture.

A trigger-stream topic is a catalog-registered Arrow schema plus a backing mutable table. Publishers append RecordBatches; subscribers filter and receive them. The engine owns the offset counter and the durable event log; the broker (in-memory by default, NATS JetStream in clustered deployments) fans live deliveries out to attached subscribers.

Reach for the trigger stream when a tenant needs event semantics — a CDC pipeline, a feature-store update bus, a job-completion notification fan-out — that has to coexist with the SQL surface the rest of the platform already uses. Every published event lands as a row in the topic’s backing mutable table; that table is queryable with the same Flight SQL surface as any other mutable companion table, so ad-hoc analytics on the event log come for free.

Goal

Walk through registering one topic for a neutral third-tenant use case (a small CDC pipeline pulling Postgres change events into a downstream search index) and publish a batch of events from Rust.

Setup

Assumes a JammiSession whose JammiConfig.trigger_broker is left at its default — the embedded InMemoryBroker. Production deployments swap in JetStreamBroker via configuration; the publisher API does not change.

Define the topic schema

#![allow(unused)]
fn main() {
extern crate arrow_schema;
fn make() {
use std::sync::Arc;
use arrow_schema::{DataType, Field, Schema};

let schema = Arc::new(Schema::new(vec![
    Field::new("op",         DataType::Utf8,  false),
    Field::new("ts_ms",      DataType::Int64, false),
    Field::new("key",        DataType::Utf8,  false),
    Field::new("after",      DataType::Utf8,  true),
]));
}
}

The schema is the contract every published batch must satisfy. The engine reserves the _offset, _row_idx, and _produced_at column names (all leading-underscore names are reserved); user schemas must not include them.

Register the topic

Topic registration is a typed lifecycle verb, not a SQL statement: build a TopicDefinition and register it. The Session::register_topic surface (and the gRPC CatalogService.RegisterTopic verb it rides) does this in one call.

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate arrow_schema;
use std::collections::BTreeMap;
use std::sync::Arc;
use arrow_schema::SchemaRef;
use jammi_db::trigger::{TopicDefinition, TopicId};

fn make(schema: SchemaRef) -> TopicDefinition {
let topic = TopicDefinition {
    id: TopicId::new(),
    name: "cdc.orders".into(),
    schema,
    tenant: None,                          // None = global; Some(t) scopes to t
    broker_metadata: BTreeMap::new(),      // driver-specific opts (e.g. retention)
};
topic
}
}

The CLI exposes the same shape via jammi trigger register --name … --schema ….

import jammi
import pyarrow as pa

db = jammi.connect("file:///var/lib/jammi")
db.register_topic(
    "cdc.orders",
    schema=pa.schema([
        ("op", pa.string()),
        ("ts_ms", pa.int64()),
        ("key", pa.string()),
        ("after", pa.string()),
    ]),
    broker_metadata={"retention_seconds": "604800"},
)

The id is a UUIDv7 minted at construction — time-ordered so the catalog index keeps insert locality. The name is opaque to the engine beyond catalog lookup; pick a hierarchical namespace that suits your platform (e.g. cdc.orders, feature_store.user_features).

Registration is atomic: the topics row, the backing mutable table, and any broker-side state commit together; nothing lands on failure.

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate tokio;
use std::sync::Arc;
use jammi_db::trigger::{TopicDefinition, TriggerBroker};
async fn ex(
    topic_repo: &jammi_db::catalog::topic_repo::TopicRepo,
    broker: Arc<dyn TriggerBroker>,
    topic: &TopicDefinition,
) -> Result<(), jammi_db::trigger::TriggerError> {
broker.register_topic(topic).await?;
topic_repo.register_topic(topic).await?;
Ok(())
}
}

Publish a batch

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate arrow;
extern crate arrow_schema;
extern crate tokio;
use std::sync::Arc;
use arrow::array::{Int64Array, RecordBatch, StringArray};
use arrow_schema::SchemaRef;
use jammi_db::trigger::{Publisher, TopicDefinition};
use jammi_db::TenantId;
async fn ex(
    publisher: &Publisher,
    topic: &TopicDefinition,
    schema: SchemaRef,
    tenant: Option<TenantId>,
) -> Result<(), jammi_db::trigger::TriggerError> {
let batch = RecordBatch::try_new(
    schema,
    vec![
        Arc::new(StringArray::from(vec!["c", "u", "d"])),
        Arc::new(Int64Array::from(vec![1700_000_000_000, 1700_000_000_100, 1700_000_000_200])),
        Arc::new(StringArray::from(vec!["order-1", "order-2", "order-3"])),
        Arc::new(StringArray::from(vec![Some("{...}"), Some("{...}"), None])),
    ],
)
.unwrap();
let offset = publisher.publish_scoped(topic, tenant, batch).await?;
println!("published offset = {}", offset.value());
Ok(())
}
}

publish_scoped tags every row’s tenant_id column from the explicit tenant: Option<TenantId> argument — no silent dependency on session state at publish time. Pass None for global topics; pass the session’s current tenant (session.tenant()) for tenant-scoped publishes.

Python equivalent — publish_topic accepts a pyarrow.Table via the Arrow C Stream Interface so the conversion is zero-copy:

import pyarrow as pa

table = pa.table({
    "op":    ["c", "u", "d"],
    "ts_ms": [1700_000_000_000, 1700_000_000_100, 1700_000_000_200],
    "key":   ["order-1", "order-2", "order-3"],
    "after": ["{...}", "{...}", None],
})
offset = db.publish_topic("cdc.orders", batch=table)
print(f"published offset = {offset}")

publish_scoped validates the batch schema against the topic schema before opening a transaction. A mismatch returns BatchSchemaMismatch and nothing lands in the backing table. If the topic is tenant-pinned (TopicDefinition::tenant = Some(t)) and the tenant argument doesn’t match, the publish is rejected up front with PublishTenantMismatch.

What just happened

1. The Publisher minted the next monotonic offset for the topic (seeded lazily from MAX(_offset) on the backing table the first time the topic is touched).
2. The augmented batch — user columns plus _offset, _row_idx, and _produced_at — was inserted into the topic’s backing mutable table inside one CatalogBackend::transaction. On commit the offset advances; on rollback it is reused for the next attempt so no gaps appear in the log.
3. The broker received the batch for best-effort fan-out to any live subscribers. A broker fan-out failure after commit is logged but does not fail the publish — subscribers replay missed offsets from the backing table on next reconnect.

See also

• Subscribe with a SQL Predicate Filter — open a subscription with a WHERE-style predicate and consume the matching batches.
• Replay Events from the Backing Table — run ad-hoc Flight SQL over the durable event log.

Subscribe to a Topic with a SQL Predicate Filter

Measured companion: for the long-form, executed-and-measured Python treatment, see The Cookbook → Change Data Capture.

A subscription tails a topic and yields only the batches whose rows satisfy a SQL WHERE-clause predicate. The predicate is parsed once through DataFusion at subscribe time; the broker delivers each batch, the engine evaluates the predicate against it, and rows that match flow to the consumer.

Reach for predicate-filtered subscriptions when different downstream consumers need different selectivity on the same event stream — a search-index that only cares about op = 'c', an audit log that wants every event, a cache invalidator that wants op IN ('u', 'd').

Goal

Open a subscription on the cdc.orders topic with a predicate that matches only deletes, and consume the stream from Rust.

Setup

Assumes the topic was registered (see Publish Events to a Topic) and you have a Subscriber constructed against the same broker the publisher uses.

Open the subscription

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate arrow;
extern crate arrow_schema;
extern crate datafusion;
extern crate futures;
extern crate tokio;
use std::sync::Arc;
use arrow_schema::SchemaRef;
use datafusion::execution::context::SessionContext;
use futures::StreamExt;
use jammi_db::trigger::{Predicate, Subscriber, TopicDefinition};
async fn ex(
    subscriber: &Subscriber,
    session: &SessionContext,
    topic: &TopicDefinition,
) -> Result<(), jammi_db::trigger::TriggerError> {
let predicate = Predicate::from_sql(session, Arc::clone(&topic.schema), "op = 'd'")?;

let mut stream = subscriber
    .subscribe(topic, predicate, None /* from_offset: None = live tail */)
    .await?;

while let Some(delivered) = stream.next().await {
    let batch = delivered?;
    handle_deletes(batch.batch);
}
Ok(())
}
fn handle_deletes(_: arrow::array::RecordBatch) {}
}

from_offset = None starts the subscription at the broker’s live tail (no replay). Some(0) starts from the earliest retained event; the engine joins backing-table replay with the live broker stream so the client sees one continuous sequence of DeliveredBatch.

Predicate dialect

Predicates are a subset of DataFusion SQL. The whitelist:

The string-function whitelist is lower, upper, length, starts_with, ends_with. Anything outside this list returns PredicateUnsupported at subscribe time — the stream never opens. An unparseable predicate returns PredicateParse for the same reason.

Reconnection and replay

If your consumer disconnects and reconnects, pass the last-seen offset as from_offset to resume without missing events:

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate chrono;
extern crate tokio;
use chrono::Utc;
use std::sync::Arc;
use jammi_db::trigger::{Offset, Predicate, Subscriber, TopicDefinition};
async fn ex(
    subscriber: &Subscriber,
    topic: &TopicDefinition,
    last_seen: u64,
) -> Result<(), jammi_db::trigger::TriggerError> {
let resume_from = Offset::new(last_seen + 1, Utc::now());
let _stream = subscriber
    .subscribe(topic, Predicate::match_all(), Some(resume_from))
    .await?;
Ok(())
}
}

The engine reads the backing table for offsets >= resume_from, then attaches the broker live stream starting strictly above the last replayed offset — the two halves never deliver the same offset twice.

Backpressure

A slow consumer slows the producer; events are not dropped. The chain is: the broker tail backs up onto a bounded mpsc::channel, the channel’s send() future awaits, the broker poll loop pauses, publishers awaiting the broker fan-out experience matching back- pressure. The backing table — the authoritative log — is still written without delay, so a consumer that disconnects under load can always catch up via replay.

See also

• Publish Events to a Topic — the publisher side.
• Replay Events from the Backing Table — bypass the broker entirely and read the event log via Flight SQL.

Replay Events from the Backing Table

Every topic’s event log is a Phase-2 mutable companion table named __topic_<topic_id>. The double-underscore prefix is reserved for engine-controlled tables; consumers do not register tables under that namespace. Flight SQL queries against the backing table compose with the same federation surface the rest of Jammi exposes — joins with result tables, predicate pushdown, aggregates over event history.

Reach for direct replay when a tenant needs ad-hoc analytics on the event log that would be awkward through the subscribe surface — counting events per key, computing per-day rollups, joining the event stream against a Parquet result table.

Goal

Run an ad-hoc query that returns the count of op = 'd' events per hour over the durable log for cdc.orders.

Backing table naming

Every registered topic has a backing table whose name is __topic_<topic_id> where <topic_id> is the hyphenated lowercase TopicId::Display. To find the name, query topics:

SELECT topic_id, name, backing_table FROM topics WHERE name = 'cdc.orders';

Schema

The backing table’s columns are the topic’s user schema with three engine-controlled columns prepended:

The primary key is (_offset, _row_idx); _offset is the order column so scan_after and ORDER BY _offset agree.

Query

SELECT
    DATE_TRUNC('hour', TIMESTAMP_MICROS(_produced_at))      AS hour,
    COUNT(*)                                                 AS deletes
FROM    mutable.public.__topic_019088da_1234_7890_abcd_ef1234567890
WHERE   op = 'd'
GROUP BY hour
ORDER BY hour;

Substitute your topic’s backing_table (looked up from the topics catalog row) for the literal name in the example. The query runs through Flight SQL like any other federated query — predicate pushdown applies, joins compose, aggregates run.

Tenant scoping

The backing table carries the tenant_id column added by the Phase-2 mutable backend. Sessions bound to a tenant see only rows whose tenant_id matches or is NULL, per Phase 3’s predicate-injection analyzer rule — the same guard that scopes the rest of the catalog.

See also

• Publish Events to a Topic — the publisher side.
• Subscribe with a SQL Predicate Filter — the live-tail surface for these same events.
• Register a Mutable Companion Table — the substrate the backing table reuses; same schema validation, same federation properties.

Scope a Session to a Tenant

Measured companion: for the long-form, executed-and-measured Python treatment, see The Cookbook → Tenancy.

When more than one logical tenant shares a Jammi engine — a SaaS feature store serving two ML teams, a research workbench shared across three labs, a notebook product hosting one project per student — every catalog read and write needs to belong to the right tenant. Jammi’s session-scoped tenant binding does this without the caller having to spell a WHERE tenant_id = … clause on every query.

Goal

After this recipe you can:

1. Bind a tenant to a session in Rust, Python, and on the CLI.
2. Verify that two sessions on the same process see disjoint rows.
3. Bind a tenant on a remote client via the gRPC CatalogService so subsequent Flight SQL queries from the same connection observe the tenant.

Setup

Every example below assumes a configured JammiConfig (defaults are fine for the recipe). The tenant identifier is a UUID v4 or v7 string — the engine refuses the nil UUID (00000000-…) at the TenantId newtype boundary.

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate tokio;
use std::str::FromStr;
use jammi_db::TenantId;
use jammi_db::session::JammiSession;
use jammi_db::config::JammiConfig;

async fn ex() -> jammi_db::error::Result<()> {
let config = JammiConfig::default();
let alice = TenantId::from_str("018f5a0e-c4c8-7e10-9c4f-3b6f7c5a8e9a")?;

let session = JammiSession::new(config).await?.with_tenant(alice);
// Every catalog read and write on `session` now scopes to Alice.
Ok(())
}
}

with_tenant is a builder that consumes self and returns Self, so it chains naturally. If you hold a session behind Arc, use bind_tenant(&t) to update the binding in place — the session shares one TenantBinding across all references.

Python

import jammi

db = jammi.connect("file:///tmp/jammi")
db.set_tenant("018f5a0e-c4c8-7e10-9c4f-3b6f7c5a8e9a")

# Subsequent calls observe Alice's tenant scope.
db.add_source("inbox", path="/data/alice/inbox.parquet", format="parquet")
db.sql("SELECT * FROM inbox.public.inbox")

set_tenant is a sticky setter — it mutates the connection in place and stays in effect until the next set_tenant. Pass an empty string to clear: db.set_tenant("").

For a binding scoped to a single block — the prior tenant restored on exit, and nesting handled — use tenant_scope as a context manager:

with db.tenant_scope("018f5a0e-c4c8-7e10-9c4f-3b6f7c5a8e9a"):
    # Reads here observe Alice's tenant scope.
    db.sql("SELECT * FROM inbox.public.inbox")
# Prior scope restored here.

The same surface is available on a remote connection (jammi.RemoteDatabase), where the prior tenant is captured client-side and rebound on exit.

CLI

The --tenant flag is global; it applies to every subcommand.

jammi --tenant 018f5a0e-c4c8-7e10-9c4f-3b6f7c5a8e9a sources list
jammi --tenant 018f5a0e-c4c8-7e10-9c4f-3b6f7c5a8e9b models list

Remote clients (gRPC + Flight SQL)

A programmatic client (Python, Go, Java) binds the tenant once per connection via the jammi.v1.catalog.CatalogService.SetTenant RPC. The server records the tenant against the jammi-session-id request metadata header; every Flight SQL query the same connection issues afterwards inherits the binding through the same resolver — the engine-default SessionIdTenantResolver — applied by the single async tenant-binding layer (TenantResolverLayer) that fronts both the CatalogService and the Flight SQL provider. Browser clients reach the same CatalogService over HTTP/1.1 via the gRPC-Web shim (application/grpc-web+proto) — no separate REST surface, same jammi-session-id header semantics.

import grpc
from jammi.v1 import catalog_pb2, catalog_pb2_grpc

channel = grpc.insecure_channel("jammi.example.com:50051")
metadata = [("jammi-session-id", "my-client-uuid")]

client = catalog_pb2_grpc.CatalogServiceStub(channel)
client.SetTenant(
    catalog_pb2.SetTenantRequest(
        tenant=catalog_pb2.Tenant(id="018f5a0e-c4c8-7e10-9c4f-3b6f7c5a8e9a")
    ),
    metadata=metadata,
)
# Subsequent Flight SQL queries on the same channel + jammi-session-id
# observe Alice's tenant scope.

This flow assumes a trusted network. The jammi-session-id header is a client-minted, opaque transport correlation id — it identifies a connection, not a principal. The server does not authenticate it: anyone who presents another session’s id assumes that session’s tenant. SetTenant writes a tenant the caller asserts; nothing verifies the caller is entitled to it. That is the right trade-off when every client is inside your trust boundary (a private VPC, a sidecar mesh, a single-process notebook), and the wrong one the moment an untrusted caller can reach the port. Do not treat jammi-session-id as an authentication or authorization boundary.

Bring your own auth

Jammi authenticates nothing on its own — it is a substrate, and identity is a consumer’s vocabulary. To put a tenant boundary in front of untrusted callers, you supply the authentication and authorization yourself by implementing a TenantResolver and passing it to GrpcChain.tenant_resolver when you assemble the chain via assemble_grpc_chain. One resolver, plugged in once, binds every engine gRPC verb AND the Flight SQL db.sql lane — the same async tenant-binding layer (TenantResolverLayer) applies the resolved scope to both transports, so there is nothing separate to wire up for Flight.

1. Authenticate the principal. In resolve, read the caller’s credential — a bearer token, a session cookie your gateway exchanges, a service-to-service token — and verify it. A missing or invalid credential returns Err(Status::unauthenticated(..)) here, before any handler runs.
2. Authorize the tenant from the verified claim. Derive the tenant from the verified claim — never from a header the caller controls. This is where your policy lives: which tenant this principal may act as. Return Ok(TenantScope::Tenant(t)).
3. The engine binds it. The async TenantResolverLayer maps the resolved scope onto the SessionTenant request extension every verb handler reads, and the Flight SQL provider (TenantBoundProvider) binds the same scope for db.sql — you write only resolve.

Because resolve runs in front of every handler, the tenant the engine acts on is the one the credential proves, not one the caller asserts. The jammi-session-id header plays no part in this path. Reject, don’t default: an authenticating resolver returns Tenant/Err and NEVER TenantScope::Global — returning Global on a failed check runs the request unscoped, which for a tenant_id IS NULL-bearing catalog is a global read, so a rejected caller must fail the request. TenantScope::Global is the explicit unscoped choice the engine-default resolver (SessionIdTenantResolver) returns when no tenant is bound — never a value a rejection falls through to.

use tonic::{Status, metadata::MetadataMap};
use jammi_db::TenantId;
use jammi_server::grpc::session::{TenantResolver, TenantScope};

/// A consumer's authenticating resolver. `verify_credential` is the
/// consumer's own identity logic — it authenticates the caller and returns the
/// tenant the verified claim authorizes, or `None` to reject the request.
struct AuthResolver;

#[tonic::async_trait]
impl TenantResolver for AuthResolver {
    async fn resolve(&self, metadata: &MetadataMap) -> Result<TenantScope, Status> {
        // 1. Authenticate: pull the credential the caller presented.
        let credential = metadata
            .get("authorization")
            .and_then(|v| v.to_str().ok())
            .ok_or_else(|| Status::unauthenticated("missing credential"))?;

        // 2. Authorize: derive the tenant from the *verified* claim. A failed
        //    check rejects the request — it never falls through to an unscoped
        //    read that could surface another tenant's rows.
        let tenant: TenantId = verify_credential(credential)
            .ok_or_else(|| Status::unauthenticated("invalid credential"))?;

        // 3. Bind: return the resolved scope. The engine's tenant-binding layer
        //    applies it to every gRPC verb and to Flight SQL.
        Ok(TenantScope::Tenant(tenant))
    }
}
fn verify_credential(_c: &str) -> Option<TenantId> { None }

Plug it in at assembly time, in place of the engine default:

use std::sync::Arc;
use jammi_server::runtime::{assemble_grpc_chain, GrpcChain};

let chain = GrpcChain {
    // .. addr, flight_ctx, flight_binding, store, trigger, engine, tiers, metrics ..
    tenant_resolver: Arc::new(AuthResolver),
    ..chain_defaults
};
let assembled = assemble_grpc_chain(chain)?;

The seam types are TenantResolver (the trait you implement), TenantScope (Tenant/Global), and SessionTenant (the per-request binding every verb reads, which the engine sets for you). This one resolver replaces the engine-default SessionIdTenantResolver for the whole chain — the gRPC verbs and the Flight db.sql lane both read the scope it resolves, closing the cross-transport gap where a boundary authenticated the gRPC plane but Flight still bound from the unauthenticated jammi-session-id header.

Gating the cross-tenant admin pass: AdminAuthorizer

TenantResolver above binds which tenant an ordinary request acts as. A separate, narrower seam gates a different question: whether a caller may run CatalogService.Reconcile’s cross-tenant all = true admin pass at all — the one verb that deliberately crosses every tenant’s data in a single call. This is AdminAuthorizer, a synchronous capability check (&MetadataMap -> Result<(), Status>) supplied at GrpcChain.admin_authorizer:

use jammi_server::grpc::catalog::AdminAuthorizer;
use tonic::{Status, metadata::MetadataMap};

/// A consumer's admin-capability check — a signed capability header, an
/// mTLS peer identity the transport already verified, or any other
/// consumer-owned policy. `Ok(())` permits the `all = true` pass; `Err`
/// is returned to the caller verbatim, so this decides its own status code.
struct AdminHeaderCheck;

impl AdminAuthorizer for AdminHeaderCheck {
    fn authorize(&self, metadata: &MetadataMap) -> Result<(), Status> {
        match metadata.get("x-admin-capability") {
            Some(v) if v == "granted" => Ok(()),
            _ => Err(Status::permission_denied("admin capability required")),
        }
    }
}

use std::sync::Arc;
use jammi_server::runtime::{assemble_grpc_chain, GrpcChain};

let chain = GrpcChain {
    admin_authorizer: Some(Arc::new(AdminHeaderCheck)),
    ..chain_defaults
};

The shipped default is admin_authorizer: None, which refuses EVERY all = true request with PERMISSION_DENIED — a deployment that never wires one cannot accidentally expose the cross-tenant sweep. A tenant-scoped Reconcile (all = false) never consults this seam at all; it runs under the caller’s own TenantResolver-resolved scope like every other verb. The seam is gRPC-only by construction (Reconcile has no Flight SQL analogue), unlike TenantResolver, which binds both transports from one grant — see Security Posture for the full comparison.

Disjoint views — what to expect

Two sessions on the same process, bound to different tenants, will:

1. Read each other as invisible: list_sources() returns the calling tenant’s sources plus any globally-scoped (tenant_id IS NULL) sources.
2. Write into different lanes: a register_source from Alice produces a row tagged tenant_id = alice; Bob’s list_sources does not see it.
3. Share globally-scoped rows: an unscoped (tenant_id IS NULL) registration — typically a public reference dataset — is visible to every tenant.

The engine enforces the binding at four layers (the SPEC-03 defence-in-depth discipline):

• Read-side predicate injection — TenantScopeAnalyzerRule injects tenant_id = $current OR tenant_id IS NULL on every TableScan whose schema declares the column.
• Result-table resolution gate — a Jammi-owned result table is wholly owned by one tenant (or GLOBAL), so its Parquet carries no tenant_id column for the analyzer to filter on. The tenant-gating result-table schema provider instead gates resolution on the catalog owner: over every lane that names jammi.{table} (Flight db.sql, gRPC sql, search), a correctly-bound tenant resolves only its own and GLOBAL result tables, and a peer’s private table resolves not-found (and is absent from the schema’s table enumeration) — the same (tenant_id = $current OR tenant_id IS NULL) visibility the catalog read API applies.
• Write-side guard — every catalog register_* and the mutable-table sink calls Transaction::assert_tenant_matches before INSERT.
• Storage-side filter — catalog repo reads also pass the predicate to the backend SQL layer, so the wrong tenant’s rows never leave the database.

A buggy caller that constructs a row with the wrong tenant_id gets BackendError::TenantMismatch from the guard layer.

When the binding doesn’t apply

• External federated sources without a tenant_id column — Jammi’s analyzer rule has no column to inject against, so those sources show every row to every tenant unless the source declaration registers a tenant_column override. Catalog tables and mutable companion tables always carry the column.
• Cross-tenant WHERE clauses the caller writes by hand — a query that contains WHERE tenant_id = 'other-tenant' runs against the injected predicate plus the user’s clause; the analyzer rule does not remove user-written predicates.
• Single-tenant deployments — bind nothing and every row is global; no predicate is injected beyond tenant_id IS NULL.

See also

• The discipline test in SPEC-03
• Register a Mutable Companion Table for how a mutable companion table also honours the tenant binding on write

Scope a Federated Source by Tenant

Measured companion: for the long-form, executed-and-measured Python treatment, see The Cookbook → Tenancy.

The session-scoped tenant binding (multi-tenant.md) relies on every table the engine reads carrying a tenant_id column. That works for mutable companion tables and Parquet result tables Jammi produced itself — both emit the column per the tenant-identifier discipline (Design Philosophy). But a federated source — a remote Postgres warehouse, a S3 Parquet lake, a CSV from someone else’s pipeline — usually doesn’t. It may carry a customer_id, an organization, a workspace column, or no tenant discriminator at all.

This recipe shows how to tell Jammi which column on a federated source plays the role of the tenant discriminator, so the predicate-injection analyzer rule scopes scans against that column instead of looking for the engine’s built-in tenant_id name.

Goal

After this recipe you can:

1. Register a federated source whose tenant discriminator is named differently from tenant_id.
2. Tell the analyzer rule which column to use.
3. Verify two tenants get disjoint rows from the same physical source.
4. Recognise what set_source_tenant_column does not do.

Setup

The recipe assumes you have a Parquet file (or any other federated source) whose schema includes a column that already carries the tenant identifier — for example a customer_id column populated with the UUID of the customer who owns each row. The column’s value must be the same canonical hyphenated lowercase form TenantId::Display emits; the analyzer rule does a string comparison after coercing the column to Utf8.

Register a federated source

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate tokio;
use jammi_db::session::JammiSession;
use jammi_db::source::{FileFormat, SourceConnection, SourceType};

async fn ex(session: &JammiSession) -> jammi_db::error::Result<()> {
session
    .add_source(
        "notes",
        SourceType::File,
        SourceConnection {
            url: Some("file:///data/notes.parquet".into()),
            format: Some(FileFormat::Parquet),
            ..Default::default()
        },
    )
    .await?;
Ok(())
}
}

Python

db.add_source("notes", path="/data/notes.parquet", format="parquet")

Declare its tenant column

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
use jammi_db::session::JammiSession;
fn ex(session: &JammiSession) {
session.set_source_tenant_column("notes", Some("customer_id".into()));
}
}

set_source_tenant_column registers the override on the session’s SourceTenantColumns map. The next time the analyzer rule sees a scan against notes.public.notes, it discovers the override and injects WHERE CAST(customer_id AS Utf8) = $current_tenant OR CAST(customer_id AS Utf8) IS NULL (or IS NULL only when the session is unscoped).

The Python and CLI surfaces do not expose this method today — it lives on Rust JammiSession only. If you embed Jammi as a library this is the right hook; if you reach Jammi over Flight SQL / gRPC, the source registration and tenant-column declaration happen on the server side before the server starts accepting client connections.

Schema column tenant_id always wins. Only call set_source_tenant_column when your federated source carries the discriminator under a different name, or when it carries the discriminator at all — sources without any tenant column remain globally visible.

Verify the predicate

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate tokio;
use std::str::FromStr;
use jammi_db::TenantId;
use jammi_db::session::JammiSession;
use jammi_db::config::JammiConfig;
async fn ex() -> jammi_db::error::Result<()> {
let alice = TenantId::from_str("018f5a0e-c4c8-7e10-9c4f-3b6f7c5a8e9a")?;
let bob = TenantId::from_str("018f5a0e-c4c8-7e10-9c4f-3b6f7c5a8e9b")?;

let session_a = JammiSession::new(JammiConfig::default()).await?.with_tenant(alice);
session_a.set_source_tenant_column("notes", Some("customer_id".into()));

let session_b = JammiSession::new(JammiConfig::default()).await?.with_tenant(bob);
session_b.set_source_tenant_column("notes", Some("customer_id".into()));

let count_a = session_a.sql("SELECT COUNT(*) FROM notes.public.notes").await?;
let count_b = session_b.sql("SELECT COUNT(*) FROM notes.public.notes").await?;

// Each session sees only its own rows — `count_a` and `count_b` are
// disjoint subsets of the on-disk file.
Ok(())
}
}

For a file with 10 rows split 6 (customer_id = alice) + 4 (customer_id = bob), the two sessions get 6 and 4 respectively.

What you cannot do

• You cannot point set_source_tenant_column at a column that doesn’t exist on the source. The analyzer rule emits a column reference that DataFusion later fails to resolve at execution time, surfacing as a DataFusionError::SchemaError. The override is a trust contract — the engine does not validate the column’s presence at registration time.
• You cannot mix tenant_id and a non-tenant_id column on the same source. When the source’s schema already declares tenant_id, the built-in column wins and the override is ignored.
• You cannot remove the discriminator at runtime once tenants are actively querying. Call set_source_tenant_column("notes", None) to drop the override; subsequent queries on notes.public.notes will not be tenant-scoped at all.

If the federated source you are wrapping carries no tenant discriminator, two options are open: (1) re-shape upstream so each tenant lands in its own table, registered as a separate source, or (2) accept that the source is globally visible to every session and gate access at a higher layer (Flight SQL session interceptor, gRPC auth middleware). The engine itself does not authenticate; Design Philosophy § the engine does not invent tenants applies.

See also

• Scope a Session to a Tenant — the broader session-binding recipe this one extends.
• Register a Mutable Companion Table — for sources Jammi owns, the tenant_id column is emitted by default.

The Materialization Contract: Verifiable Result-Table Identity

Measured companion: for the long-form, executed-and-measured Python treatment, see The Cookbook → Incremental Recompute.

Every result table Jammi publishes carries a verifiable identity: a sidecar attestation that lets a later reader assert “this artifact is the output of definition D over input-state S” — without trusting a name, a path, or an out-of-band convention. This recipe is the operator’s view of that contract: what the attestation contains, how it is written, how to check a table against it, and how recovery reconciles it after a crash. Everything below describes the system as it ships today.

What a materialization manifest is

A result table is published as an immutable Parquet object (plus, for embedding tables, an ANN-index sidecar bundle). Alongside it the engine writes a separate .materialization.json sidecar — for every result table, not only embedding tables — carrying an in-toto-shaped attestation that binds three things to the artifact’s content digest:

• a definition hash of how the table was produced;
• the as-of anchors of every input the producer read; and
• the producing-run identity and instant.

The on-disk shape is MaterializationManifest:

The artifact digest deliberately covers the Parquet data, never the ANN index sidecar: the index is a derived accelerator reconstructible from the data, so a verdict attests the data-of-record, not the search structure. The two sidecars never collide — .manifest.json describes the ANN accelerator; .materialization.json attests the Parquet data.

The two halves of identity: descriptor and environment

The definition_hash is not a hash of a logical plan. Result-table producers in this engine are hand-built physical pipelines, so there is no single plan to canonicalise. Instead the hash folds two typed, deterministically-serialisable values:

• ProducingDescriptor — how the table was computed: the verb plus its typed parameters. Each producer fills in exactly one variant from its own parameters — Inference, Embedding, NeighborGraph, GraphPropagation, or ContextSet. A stable, sorted-key JSON encoding yields canonical bytes.
• MaterializationEnv — the output-affecting environment that is not part of the description itself: the engine semantic version, the compute device (Cpu / Cuda { ordinal } / Metal { ordinal }), and the identity + backend kind of every model the producer invoked.

The device is part of the environment for a concrete reason: a model produces different float outputs on CPU versus an accelerator while carrying the same model identity, so a hash that omitted the device would yield a false “match” when only the device changed. The two halves are length-prefixed and domain-separated before hashing, so a descriptor field can never alias an environment field. Two runs of the same producer, with the same parameters, over the same inputs, in the same environment, hash identically; any output-affecting change to any of the three changes the hash.

The hash folds every declared parameter of the descriptor and the environment — it does not fold the producing host’s CPU microarchitecture. Two CPU hosts of a different ISA (or even two same-ISA hosts with different runtime vector-dispatch decisions) can produce the same definition_hash for tables whose last float bits differ, because a CPU float reduction is a same-box guarantee only (measured directly: bits_snapshot.rs pins CPU-float bit output per box, not per (target_arch, target_os)). A definition_hash match therefore asserts which definition produced the table, not that its bytes are reproducible byte-for-byte on a different CPU host — identity across hosts is the catalog row plus the definition_hash, never the raw bytes. This CPU-variant gap in the hash is a known, open limitation, not a defect this contract claims to close.

The input anchors are recorded but are deliberately not part of the definition hash: the definition is how a table is produced, the anchors are over what. A consumer that wants a combined “code + data” identity composes the two itself.

BuildingTable::finish is the sole building→ready transition

There is no manifest-free finalize. ResultStore::create_table returns a lease-owned BuildingTable handle, and its finish is the single building → ready path every producer goes through, so no table reaches ready without an attestation. It performs the steps in a crash-safe order:

1. renew the writer’s lease by compare-and-set — a writer whose lease was claimed by recovery learns it here, before it attests bytes it no longer owns;
2. ResultStore::write_attestation: read the Parquet bytes and compute the artifact digest, compute the manifest from the descriptor, environment, and resolved inputs, and write the .materialization.json sidecar;
3. flip the catalog row building → ready by a compare-and-set naming the writer (recording the definition_hash and the input anchors as summary columns, and clearing the lease);
4. register the table in DataFusion under the row’s own catalog owner.

Because the bytes and the sidecar are durable before the status flip, a crash in the window leaves a building row — never a queryable ready table missing its manifest — and the row’s lease then expires, so the next recovery sweep promotes it from that very sidecar or reaps it. Every producer that materialises an embedding table — graph propagation and context-set pooling — routes through this funnel too: ResultStore::materialize_embedding_table writes the table and then calls finish with the producer’s Materialization (descriptor, environment, and resolved input anchors).

How to verify a table

verify_materialization is the read-only verb that recomputes a ready table’s artifact digest and checks it — and, optionally, an expected definition hash — against the table’s manifest. It returns a MatchVerdict; it never acts on one. What a reader does with a mismatch (refuse, alarm, fall back) is the reader’s policy, not the engine’s.

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use std::sync::Arc;
use jammi_ai::session::InferenceSession;
use jammi_db::store::manifest::{DefinitionHash, MatchVerdict};
async fn ex(session: &Arc<InferenceSession>, table: &str) -> jammi_db::error::Result<()> {
let record = session
    .catalog()
    .get_result_table(table)
    .await?
    .expect("result table exists");

// No expectation: just assert the bytes still match the attestation.
let verdict = session
    .result_store()
    .verify_materialization(&record, None)
    .await?;

match verdict {
    MatchVerdict::Match => { /* artifact is the attested output */ }
    MatchVerdict::MatchWithUnpinnedInputs { unpinned } => {
        // Verified, but at least one input was anchored only to a read
        // instant, so reproducibility cannot be fully asserted. Honest,
        // not silent — downgrade confidence accordingly.
        let _ = unpinned;
    }
    MatchVerdict::Mismatch { expected, found } => {
        // The served artifact is not the output of the expected definition.
        let _ = (expected, found);
    }
    MatchVerdict::MissingManifest => {
        // No sidecar — a pre-contract table. A truthful unknown, never a
        // fabricated match.
    }
}

// Pin an expected definition hash to assert *which* definition produced it.
let expected = DefinitionHash("…".into());
let _ = session
    .result_store()
    .verify_materialization(&record, Some(&expected))
    .await?;
Ok(()) }
}

Python

verify_materialization takes the table name and an optional expected definition hash, and returns the verdict as a dict tagged by verdict:

verdict = db.verify_materialization("results__text_embedding__…")

if verdict["verdict"] == "match":
    pass  # artifact is the attested output
elif verdict["verdict"] == "match_with_unpinned_inputs":
    unpinned = verdict["unpinned"]      # sources anchored only to an instant
elif verdict["verdict"] == "mismatch":
    expected, found = verdict["expected"], verdict["found"]
elif verdict["verdict"] == "missing_manifest":
    pass  # pre-contract table — a truthful unknown

# Pin the definition you expect produced the table:
db.verify_materialization("results__…", expected_definition="<hex definition hash>")

Reading the verdict

An anchor is “pinned” when it points at an immutable id: a result table’s content digest, a mutable companion table’s monotonic version, or an external source’s as-of/version value (an Iceberg snapshot id, a Delta version, an LSN, a watermark). It is “unpinned” only when the source exposes no version surface, in which case the anchor is the read instant and the verdict says so.

How recovery reconciles manifest sidecars

ResultStore::recover() runs at startup and restores the crash-consistency invariant of the catalog↔storage boundary across every tenant (it runs under an admin scope — see Catalog Backend and Trigger Broker → Multi-writer safety for the lease primitive this section assumes). Its domain is deliberately narrow: it enumerates only building rows whose lease is absent or expired (Catalog::list_expired_building_tables). A building row under a live lease belongs to a writer that is still producing it — recovery does not even look at its Parquet or manifest state, on any tenant, from any session. For each expired-lease row it finds, recovery first claims it by a one-row compare-and-set naming the recoverer the row’s new owner (the recoverer becomes a writer, heartbeating a fresh lease of its own) — only after that claim succeeds does it inspect bytes or delete anything, so a writer that renews mid-sweep, or a peer recoverer that claims first, is never raced. For the materialization contract specifically, the claimed row’s Parquet and manifest state decide the terminal CAS:

• A claimed row with valid Parquet but no manifest is reaped. The write was torn in the window between the Parquet landing and the manifest being written — before the building → ready flip. The contract forbids promoting a table without an attestation, and the producing descriptor cannot be reconstructed after the fact, so the row is driven to failed by CAS and its bytes are deleted only after that CAS affects exactly one row — never promoted manifest-less.
• A claimed row that does have its manifest is promoted, backfilling the summary columns (definition_hash, input anchors, the true Parquet footer row count — never the count the writer intended before it crashed) the live path records, and rebuilding the ANN sidecar from the Parquet for an embedding table so it self-heals even if its segment set never landed.
• A ready table whose manifest has since vanished is reaped. This is reconcile_ready_manifests, which recover() also runs (across every post-contract ready row — one whose catalog definition_hash is set, so it was promoted under the contract): its .materialization.json being absent is a corruption, since the attestation a verifier would read is gone. Such a row is driven to failed by a status-guarded CAS and its bytes reaped, rather than left queryable with a silently missing manifest.

Every deletion in both reaping arms above follows its own one-row CAS — the row’s new owner (the recoverer, having just claimed it) deletes what it now owns; a failed delete is logged and left for a later jammi reconcile pass, never silently swallowed. This is two of the engine’s three deletion arms for a building row’s bytes; the third is a writer’s own abort() after its own CAS. See Operability for the full crash-mid-publish failure-mode entry, and Backup and Restore for what a catalog/storage restore that lands outside this lease-driven window (an object whose row was never written at all, or vice versa) requires — jammi reconcile, not recover().

Pre-contract tables report honestly rather than being penalised. A row created before migration 021 carries definition_hash IS NULL in the catalog and legitimately has no sidecar; recovery leaves it untouched, and verify_materialization returns MissingManifest for it — a truthful unknown. This is the distinction the contract draws: a bug (post-contract, no sidecar) is reaped; a legitimate historical table is preserved.

Versioned tables: the identity chain

An embedding table that has been refreshed (see Refresh an Embedding Table Incrementally) carries a chain of versions beside its base manifest. The base version’s identity is the .materialization.json artifact digest — publishing the base changes no anchor. Every later version N records, in {table}__v{N}.version.json, its parent, the definition hash, the delta descriptor (EmbeddingDelta or EmbeddingCompaction, which name the parent version and its identity), the digest of every fragment it serves and the digest of its deletion mask; its identity is a domain-separated SHA-256 over exactly those inputs. Counts, the ANN segments and produced_by/produced_at are outputs, not inputs.

verify_materialization on a versioned table runs the base check unchanged, then recomputes every fragment digest and the mask digest from the bytes, recomputes the chain identity from the parent’s recorded identity, and compares it with both the version manifest and the catalog row. A Mismatch names the artifact that diverged. The version’s input anchors are the source at the instant of the refresh (unpinned), so the verdict is MatchWithUnpinnedInputs naming the source, exactly as for the base embed.

staleness and every derives_from anchor use the current version’s identity, so a refresh that changed content advances dependents to Stale { InputAdvanced } and a no_change refresh leaves them Fresh. recompute of a versioned table starts a new chain in a new table.

Why this identity matters

The materialization manifest gives every result table a content-addressed, verifiable identity that is independent of its name or path. That identity is the nucleus a future freshness-and-caching layer builds on: once a table’s output is bound to the definition that produced it and the as-of state of its inputs, an incremental-recompute layer can decide whether a cached artifact is still valid by comparing definition hashes and input anchors — rather than re-running the producer blind. The contract ships that identity and the verify primitive today; it ships no policy. What a reader does with a verdict, and when a downstream layer chooses to recompute, are decisions left to the consumer.

Refresh an Embedding Table Incrementally

An embedding table produced by generate_embeddings is a function of its source: D(S). When the source changes, refresh_embeddings re-embeds only the rows whose content changed and publishes the result as a new version of the same table — the table keeps its name, its ANN index and every downstream anchor, and a reader sees exactly one version at a time. “Publish” here is the storage-commit sense — the catalog’s publish_version CAS that makes a building version the table’s current one — distinct from the domain-lifecycle “publish/install/bind” a consumer builds on top of Jammi’s primitives (see Philosophy).

db.generate_embeddings("docs", model="sentence-transformers/all-MiniLM-L6-v2",
                       columns=["title", "body"], key_column="id")
# ... rows are edited, added, or deleted in the source ...
report = db.refresh_embeddings("docs__embedding__all-MiniLM-L6-v2__...")
# {'table': ..., 'version': 1, 'parent_version': 0, 'inferred_rows': 3,
#  'added': 1, 'changed': 2, 'deleted': 1, 'unchanged': 9996, 'dropped_rows': 0,
#  'live_rows': 9999, 'masked_rows': 3, 'outcome': 'published'}

#![allow(unused)]
fn main() {
extern crate jammi_ai;
extern crate jammi_db;
extern crate tokio;
use jammi_ai::Session;
use jammi_ai::pipeline::embedding_refresh::RefreshOptions;
async fn ex(session: &Session, table: &str) -> jammi_db::error::Result<()> {
let report = session.refresh_embeddings(table, RefreshOptions::default()).await?;
let _ = report;
Ok(()) }
}

Every transport exposes the same three verbs with the same report: Session::{refresh_embeddings, compact_embeddings, expire_versions} in Rust, Database.refresh_embeddings / compact_embeddings / expire_versions in Python (embedded and remote alike), and EmbeddingService.RefreshEmbeddings / CompactEmbeddings / ExpireVersions on gRPC. A refresh run through a local session and the same refresh run through the data-plane client produce the same version identity, fragment digests and counts (K4).

What a refresh does

1. Gates. The table must be ready, produced by an embedding pipeline, and carry _content_hash (a table imported or produced before the column existed is NotRefreshable { reason: missing_content_hash } — run recompute once to get a refreshable table). The model is loaded and the table’s recorded definition is compared with the definition the current environment would produce; a different model, backend, precision or quantization is DefinitionDrift { table, recorded, current } — nothing is allocated, and the consumer’s remedy is recompute.
2. Base version. A never-refreshed table is first published as its base version, whose identity is the table’s existing .materialization.json artifact digest, so publishing the base changes no downstream anchor.
3. Diff. The current version’s (_row_id, _content_hash) pairs are read through the masked view and the source is scanned once, projecting the same content hash. Each source key is classified added, changed or unchanged; a current key the source no longer has is deleted (see the delete policy below).
4. Nothing to do → outcome: no_change, no version allocated, nothing written, every downstream anchor untouched (a dependent stays Fresh).
5. Infer the delta. A new version number is allocated (monotonic, never reused) and only the added and changed rows are run through the model — the plan is the source scan joined to the delta keys, sorted in the same key order the base uses, so the fragment’s bytes are deterministic. The rows land in one Parquet fragment and one ANN segment, both stamped with the new version.
6. Deletion mask. Every superseded or deleted key is added to the version’s cumulative mask with the horizon N − 1: the key is dead in every fragment and segment stamped at or below that horizon and live in the new fragment.
7. Publish. The version’s manifest is written and one catalog transaction flips the version building → ready and moves the table’s current_version forward. Two refreshes of one parent both allocate a number; the second to publish a BASE version finds its parent already moved (ParentMoved) and absorbs it, proceeding with the concurrent winner’s version rather than failing. The previous version stays live until the swap, so a refused refresh never changes what a reader sees.

RefreshReport

live_rows and masked_rows are the inputs to a compaction decision; the engine ships no threshold.

Delete policy

deletes is tombstone (default) or retain:

• tombstone — a key the source no longer has is masked out of every prior fragment and segment; the table stays D(S).
• retain — the key’s current row is kept (a rolling-window source that drops old rows the table should still serve).

The policy is recorded in the version’s descriptor, so two versions produced under different policies have different identities.

Uniqueness

The initial generate_embeddings tolerates duplicate source keys (every row is written; the ANN index keeps the last one). A refresh does not: a duplicated key on the source scan is NonUniqueKey { table, scan: source, keys, total } after the complete scan (up to ten keys with their exact counts, plus the total), and a parent version that already holds two physical rows under one _row_id is NonUniqueKey { scan: parent }. Neither allocates a new version; the remedy is to de-duplicate the source and, for a non-unique parent, to recompute once.

Null keys

A NULL in the key column is never a per-row event on any path — generate_embeddings, a refresh, and infer all refuse the whole call with InvalidKey { column, null_count } before the model runs. The count is exact, zero rows are embedded, and nothing is written (a base version published by a first refresh is retained; it is downstream-invisible).

Empty texts

Per-row validity is the backend’s: the candle backend marks an empty or null text a per-row error (dropped_rows); the HTTP backend forwards texts verbatim, so validity is the remote’s. A delta whose every row is dropped is still a legitimate publish — a mask-only version with no fragment and no segment, so an edit-to-empty removes the row from the table.

Reading a versioned table

A reader — search, search_by_id, every SQL SELECT, read_vectors, verify_materialization, staleness — resolves the table’s current_version and sees only that version’s live rows:

• ANN. The version’s segments are merged under its mask; a masked candidate is dropped and the search widens until k live hits are found, and the exact rescore reads the segment that owns the hit.
• SQL. The version’s fragments are unioned under the mask with _row_id always projected, so COUNT(*), SELECT vector and LIMIT n all see live rows only.
• Snapshots. A publish between a reader’s table-row read and its scan serves the older version consistently; a read in flight when its version expires completes from memory. A table whose current version’s manifest cannot be resolved (the object is gone) is bound as a placeholder: planning succeeds, every scan is the typed VersionUnavailable { table, version }, and recompute is the remedy.
• A never-refreshed table is byte-identical to today’s: no mask, no union, read_vectors reads the raw file.

Restarting the engine re-binds the current version; search results and SELECT output are identical before and after.

Compaction and expiry

compact_embeddings(table) rewrites the current version’s live rows as one fragment and one segment — no inference, vectors carried byte-for-byte — and publishes it as a new version with the same ranking and an identity that folds the parent’s. expire_versions(table, before) deletes every non-current version numbered below before and reaps its manifest, mask and every fragment and segment the current version does not still reference; {table}.parquet, its .materialization.json and the version allocator are never touched.

db.compact_embeddings(table)
db.expire_versions(table, before=db.refresh_embeddings(table)["version"])

Identity and verification

The base version’s identity is the table’s artifact digest. Every later version’s identity is a hash chain over its parent’s identity, the definition hash, the delta descriptor, the fragment digests and the mask digest — counts and the ANN index are outputs, never inputs. verify_materialization on a versioned table recomputes every fragment and mask digest and the chain and names the artifact that diverged; staleness on a dependent uses the current version’s identity, so a refresh that changed content advances the dependent to Stale { InputAdvanced } and a no_change refresh leaves it Fresh. recompute of a versioned table produces a new table with a fresh chain root.

Pinned reads for a persisting producer

Any producer that persists a durable artifact whose provenance NAMES this table as an input — an embedding delta/compaction, a context set, a propagated table, a neighbor graph — must resolve the table’s anchor and every row it reads from the SAME catalog resolution, never two independent reads of current_version: a version publish landing between them would persist an artifact whose provenance names one version while its content came from another. ResultStore::pin_current_version is that one resolution; PinnedSource::input_anchor (infallible, no second catalog read) and ResultStore::pinned_provider both derive from it.

Disclosed residual — candidate SELECTION is not pinned. Pinning closes “the artifact’s anchor and the rows it reads agree on one version” for a producer that already holds its candidate key set. It does not make “every row read anywhere in the producer’s pipeline came from this one version” true end-to-end for every producer:

• The three context producers (context_set, context_predictor, recompute) pin the POOLED VECTOR read; the candidate SET is chosen upstream by ResultStore::search_vectors, which serves ANN from the catalog’s live segment set (or, on the exact-search fallback, this session’s own registration) — neither leg is pinned.
• The neighbor-graph producer persists an artifact whose provenance carries the pinned anchor, while its edge candidates come from an unpinned segment set (resolve_search_mode_local).

Closing this means threading a pin into the search/candidate-selection path — out of scope for the seam above; tracked as a residual, not silently absorbed into “reads through pin_current_version” claims elsewhere.

Errors

A storage object that vanishes under a running scan is the typed Storage(StorageError::Io { source: object_store::Error::NotFound }), never a partial answer; every other DataFusion error keeps its source() chain under JammiError::DataFusion.

Scope

A refresh diffs the whole source by content hash (Tier 1), which covers file, federated and mutable sources alike. A table-level monotonic source version is a freshness surface the manifest already anticipates and may be built later (Tier 2); row-level change tracking on mutable tables is a transition log and is never built. Refresh applies to tables produced by an embedding pipeline — not to imported, propagated, context-set or inference tables.

Connect to PostgreSQL / MySQL

Jammi federates external databases alongside local files. Register a database as a source and query it with the same SQL interface — joins across local files and databases work seamlessly.

PostgreSQL

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use jammi_ai::session::InferenceSession;
async fn ex(session: &InferenceSession) -> jammi_db::error::Result<()> {
use jammi_db::source::{SourceConnection, SourceType};

session.add_source("pg_data", SourceType::Postgres, SourceConnection {
    url: Some("postgresql://user:pass@localhost:5432/mydb".into()),
    ..Default::default()
}).await?;

let results = session.sql(
    "SELECT id, title FROM pg_data.public.articles WHERE published = true LIMIT 10"
).await?;
Ok(()) }
}

MySQL

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use jammi_ai::session::InferenceSession;
use jammi_db::source::{SourceConnection, SourceType};
async fn ex(session: &InferenceSession) -> jammi_db::error::Result<()> {
session.add_source("mysql_data", SourceType::Mysql, SourceConnection {
    url: Some("mysql://user:pass@localhost:3306/mydb".into()),
    ..Default::default()
}).await?;
Ok(()) }
}

Cross-source joins

Once registered, external databases are queryable with the same three-part naming convention and can be joined with local files:

SELECT p.title, a.author_name
FROM local_data.public.papers p
JOIN pg_data.public.authors a ON p.author_id = a.id
WHERE a.institution = 'MIT'

Generate embeddings from external sources

External databases work as sources for embedding generation:

# Note: external databases must be registered through the Rust API,
# which exposes the typed SourceType::Postgres / SourceType::Mysql variants.
# The Python `add_source(url=…, format=…)` surface is for file-shaped sources.

db.generate_embeddings(
    source="pg_articles",
    model="sentence-transformers/all-MiniLM-L6-v2",
    columns=["title", "abstract"],
    key="id",
    modality="text",
)

Feature flags

External source support requires feature flags when building from source:

These are enabled by default in published crates and pre-built binaries. Building either feature from source links a native TLS stack (native-tls -> OpenSSL) rather than rustls, so it additionally requires OpenSSL’s development headers on the build host — libssl-dev on Debian/Ubuntu, openssl-devel on RHEL/AlmaLinux, or openssl + pkg-config on macOS via Homebrew. See Build dependencies (Linux).

Supported source types

Store Sources and Results in Cloud Object Storage

Jammi treats local disk, S3, GCS, Azure Blob, and Cloudflare R2 as interchangeable backends. Any place the engine accepts a local file path it also accepts a storage URL — file://, s3://, gs://, azure://, or r2:// — including registered file-shaped sources and the result-table Parquet that embedding and inference jobs write.

Build with the cloud features you need

The default build ships only file:// and the in-memory test driver. Cloud schemes are opt-in per provider so a deployment that only uses S3 does not pull in the GCS and Azure SDK chains:

[dependencies]
jammi-db = { version = "0.5", features = ["storage-s3", "storage-gcs"] }

Live integration tests live behind matching live-s3-tests, live-gcs-tests, live-azure-tests features so the hermetic cargo test lane never reaches the network.

Register an S3-backed source

Rust

#![allow(unused)]
fn main() {
extern crate jammi_db;
extern crate jammi_ai;
extern crate tokio;
use jammi_ai::session::InferenceSession;
async fn ex(session: &InferenceSession) -> Result<(), Box<dyn std::error::Error>> {
use jammi_db::source::{FileFormat, SourceConnection, SourceType};
use jammi_db::storage::{CloudConfig, S3Config, StorageUrl};

let url = StorageUrl::parse("s3://benchmarks/snapshots/2026/papers.parquet")?;

let conn = SourceConnection {
    url: Some(url.to_string()),
    format: Some(FileFormat::Parquet),
    cloud: Some(CloudConfig::S3(S3Config {
        region: Some("us-east-1".into()),
        ..Default::default()
    })),
    ..Default::default()
};

session.add_source("papers", SourceType::File, conn).await?;

let rows = session
    .sql("SELECT id, title FROM papers.public.papers LIMIT 10")
    .await?;
Ok(()) }
}

If the cloud field is None and the URL is a cloud scheme, the driver falls back to the SDK’s ambient credential chain — env vars, instance profile, IRSA, ADC, Managed Identity.

Python

import jammi

db = jammi.connect("file:///var/lib/jammi")
db.add_source("papers", url="s3://benchmarks/snapshots/2026/papers.parquet", format="parquet")
db.sql("SELECT id, title FROM papers.public.papers LIMIT 10")

The Python binding accepts the same URL forms as the Rust API; per-source cloud credentials are read from process environment.

CLI

jammi sources add papers \
    --url s3://benchmarks/snapshots/2026/papers.parquet \
    --format parquet

GCS and Azure

The pattern is identical — only the URL prefix and the CloudConfig variant change:

#![allow(unused)]
fn main() {
extern crate jammi_db;
use jammi_db::source::{FileFormat, SourceConnection};
fn make() -> SourceConnection {
use jammi_db::storage::{CloudConfig, GcsConfig};

let conn = SourceConnection {
    url: Some("gs://archives/2026/jan.parquet".into()),
    format: Some(FileFormat::Parquet),
    cloud: Some(CloudConfig::Gcs(GcsConfig {
        service_account_path: Some("/etc/jammi/sa.json".into()),
        ..Default::default()
    })),
    ..Default::default()
};
conn }
}

#![allow(unused)]
fn main() {
extern crate jammi_db;
use jammi_db::source::{FileFormat, SourceConnection};
fn make() -> Result<SourceConnection, Box<dyn std::error::Error>> {
use jammi_db::storage::{AzureConfig, CloudConfig};

let conn = SourceConnection {
    url: Some("azure://snapshots/model_outputs.parquet".into()),
    format: Some(FileFormat::Parquet),
    cloud: Some(CloudConfig::Azure(AzureConfig {
        account_name: Some("mystorage".into()),
        sas_token: Some(std::env::var("AZURE_SAS_TOKEN")?.into()),
        ..Default::default()
    })),
    ..Default::default()
};
Ok(conn) }
}

Cloudflare R2

R2 speaks the S3 API, so it rides the same driver — but r2:// is a first-class scheme so you supply only the R2-shaped inputs and the engine derives the two quirks R2 imposes: the account-scoped endpoint https://<account_id>.r2.cloudflarestorage.com and region = "auto". Mint an S3-style access key pair in the R2 dashboard (or via the API) and give Jammi the account id:

#![allow(unused)]
fn main() {
extern crate jammi_db;
use jammi_db::source::{FileFormat, SourceConnection};
fn make() -> Result<SourceConnection, Box<dyn std::error::Error>> {
use jammi_db::storage::{CloudConfig, R2Config};

let conn = SourceConnection {
    url: Some("r2://archives/snapshots/2026.parquet".into()),
    format: Some(FileFormat::Parquet),
    cloud: Some(CloudConfig::R2(R2Config {
        account_id: Some(std::env::var("R2_ACCOUNT_ID")?),
        access_key_id: Some(std::env::var("R2_ACCESS_KEY_ID")?),
        secret_access_key: Some(std::env::var("R2_SECRET_ACCESS_KEY")?.into()),
        ..Default::default()
    })),
    ..Default::default()
};
Ok(conn) }
}

Set endpoint instead of account_id to point at an R2 custom domain. Result tables and their sidecar ANN indexes persist to r2:// exactly as to any other cloud backend.

Persist result tables to the cloud

ResultStore accepts a [StorageUrl] root, so embedding and inference outputs land in the same bucket as the source data:

#![allow(unused)]
fn main() {
extern crate jammi_db;
use std::sync::Arc;
use jammi_db::catalog::Catalog;
fn ex(catalog: Arc<Catalog>) -> jammi_db::error::Result<()> {
use jammi_db::config::AnnIndexConfig;
use jammi_db::storage::{StorageRegistry, StorageUrl};
use jammi_db::store::ResultStore;
use std::sync::Arc;

let root = StorageUrl::parse("s3://benchmarks/jammi_db")?;
let registry = StorageRegistry::new();
// `AnnIndexConfig` tunes the HNSW sidecar index every embedding table carries;
// the default reproduces the index backend's built-in defaults. The last
// argument, `local_cache_dir`, is the PARENT of the two LOCAL caches the store
// derives: `{local_cache_dir}/index` (each remote ANN segment materialised
// before USearch opens it — a `file://` root loads its segments in place, so
// it is unused there) and `{local_cache_dir}/artifact` (the model-artifact
// fetch cache the store's own internal `ArtifactStore`, rooted at
// `{root}/models`, uses). Both are always local paths, even when `root` is a
// cloud scheme, since USearch and the model loader both read the local
// filesystem.
let local_cache_dir = std::path::PathBuf::from("/var/lib/jammi/cache");
let result_store = Arc::new(ResultStore::with_root(
    root,
    registry,
    catalog,
    AnnIndexConfig::default(),
    local_cache_dir,
)?);
Ok(()) }
}

Every result table the session creates writes its Parquet and sidecar ANN index under a tenant-attributed prefix of that root ({root}/_global/{table}.parquet for a GLOBAL table, {root}/{tenant-uuid}/{table}.parquet for a tenant’s); delete_table_files and the crash-recovery pass operate against the same backend. Model artifacts (ResultStore::artifact_store) share the root’s models/ sub-prefix, attributed the same way (models/{seg}/{job_id}/…) — one storage knob serves both result tables and trained models.

Config-driven result storage

A deployment usually does not hand-build the ResultStore — it sets a [storage] section in the config file and lets the session do it. result_root is the storage URL result tables are rooted at; cloud carries the driver credentials, and is the default cloud config the session threads to every driver it builds — both for the result root and for cloud data sources whose add_source call carries no inline credentials.

[storage]
result_root = "r2://jammi-results/prod"

[storage.cloud.r2]
account_id = "abc123def456"
# access_key_id / secret_access_key are read from the environment — see below.

Both fields are optional. With result_root unset, result tables stay on local disk under {artifact_dir}/jammi_db/. The catalog backend is independent of this setting (configure it under [catalog]); [storage] governs only result-table and source object storage.

[storage.cloud] is an externally tagged section — the provider name (s3, r2, gcs, azure) is its own table, and the remaining keys mirror the matching config-side section:

# AWS S3 (region in TOML, secrets from env)
[storage.cloud.s3]
region = "us-east-1"

# Google Cloud Storage — `service_account` accepts inline JSON or `{ file = "…" }`
[storage.cloud.gcs]
service_account = { file = "/etc/jammi/sa.json" }

# Azure Blob
[storage.cloud.azure]
account_name = "mystorage"

Credentials come from the environment

Secrets are deploy secrets, not config-file values. The S3 and R2 drivers build on object_store’s AmazonS3Builder::from_env(), which reads:

GCS reads GOOGLE_APPLICATION_CREDENTIALS (or Workload Identity); Azure reads the standard AZURE_* chain. So the R2 example above needs only account_id in the TOML — AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY in the container’s environment supply the rest. Any field you do set in [storage.cloud] overrides the value the env chain produced. A half-set credential pair (an access_key_id with no secret_access_key, or vice-versa) is rejected at config-load time rather than on the first request.

How the layout maps onto buckets

For a result table named papers__text_embedding__bge-m3__20260520T120000Z_abc12345, the engine writes the Parquet plus one ANN segment bundle per index segment. A table built in a single embedding pass has one segment, seg0:

s3://benchmarks/jammi_db/papers__text_embedding__bge-m3__….parquet
s3://benchmarks/jammi_db/papers__text_embedding__bge-m3__…__seg0.usearch
s3://benchmarks/jammi_db/papers__text_embedding__bge-m3__…__seg0.rowmap
s3://benchmarks/jammi_db/papers__text_embedding__bge-m3__…__seg0.manifest.json

A table’s ANN index is a set of segments: appending a batch of new rows writes a new bundle (…__seg1.*, …__seg2.*, …) beside the existing ones and leaves them untouched, and a search merges every segment. A quantized-precision segment adds a …__seg{N}.rawf32 rescore companion; a Binary one adds a …__seg{N}.threshold companion too.

The segment layout is the same on every backend; the only difference is the driver under the hood. USearch’s path-based FFI is bridged through a tempfile for cloud schemes so its save / load calls work unchanged, and each remote segment is materialised once into a content-addressed local cache before it is opened.

Deploy as a Server

Jammi can run as an Arrow Flight SQL server, making all registered sources and embedding tables queryable from any Arrow-compatible client. Use this when multiple services, BI tools, or non-Rust/Python consumers need to query Jammi’s data.

The workflow

The server is a read path. Deploy the server, then set up your data through it — with the jammi CLI (a strict gRPC client) or the library — so other systems can query it:

# 1. Start the server
jammi-server

# 2. Register sources against the running server with the CLI
jammi --target grpc://127.0.0.1:8081 \
  sources add patents --url /data/patents.parquet --format parquet

# 3. Generate embeddings (library or Python — not available over Flight SQL)
python3 -c '
import jammi
db = jammi.connect("file:///var/lib/jammi")
db.generate_embeddings(source="patents", model="sentence-transformers/all-MiniLM-L6-v2", columns=["abstract"], key="id", modality="text")
'

Connecting with Arrow Flight SQL

Python (pyarrow)

from pyarrow.flight import FlightClient, FlightDescriptor

client = FlightClient("grpc://localhost:8081")

# Run a SQL query
info = client.get_flight_info(
    FlightDescriptor.for_command(b"SELECT id, title, year FROM patents.public.patents WHERE year > 2020")
)
reader = client.do_get(info.endpoints[0].ticket)
table = reader.read_all()
print(table.to_pandas())

Query embedding tables

Embedding tables are registered in DataFusion and queryable via SQL:

# List all embedding tables
info = client.get_flight_info(
    FlightDescriptor.for_command(b"SELECT table_name FROM information_schema.tables WHERE table_schema = 'jammi'")
)

# Query vectors directly
info = client.get_flight_info(
    FlightDescriptor.for_command(b"SELECT _row_id, _model_id FROM \"jammi.patents__embedding__all-MiniLM-L6-v2__20260325\" LIMIT 10")
)

JDBC

Flight SQL is compatible with JDBC drivers that support the Arrow Flight SQL protocol, enabling access from Java applications, BI tools (Superset, DBeaver, Tableau), and SQL editors.

Server configuration

[server]
flight_listen = "0.0.0.0:8081"
preload_models = ["sentence-transformers/all-MiniLM-L6-v2"]

[logging]
level = "info"
format = "json"    # structured logging for production

Preloading models

Models listed in preload_models are loaded into the cache at startup, BEFORE /readyz reports ready — it answers 503 {"status":"not_ready", "detail":"preloading i/n"} meanwhile, /healthz stays 200 (no startupProbe needed), and this process’s claim loop waits at its gate with its workers.state row reading warming, so no job is claimed by a cold process. An entry is a bare id, whose task comes from the catalog’s models row, or { id, task } naming the task explicitly — required for a local: path, which has no row. A listed model that cannot load, a bare id with no models row, or an unknown task token is a startup error: the server exits non-zero instead of serving. A shutdown signal during the preload exits 0 without ever serving.

[server]
preload_models = [
    "sentence-transformers/all-MiniLM-L6-v2",
    { id = "local:/models/bge-small", task = "text_embedding" },
]

Service tiers

One server binary scales to many deployment shapes by mounting only the gRPC service tiers a deployment needs — no per-shape rebuild. The core tier is always mounted: CatalogService (the control plane — tenant binding, the GetServerInfo handshake, and source / model / channel / mutable-table / topic administration), EmbeddingService, InferenceService, PipelineService, AuditService, JobService (durable job submission — every deployment accepts a job and reports its status), and the Flight SQL surface. Two optional tiers are runtime-selectable via [server] services:

[server]
services = "all"             # all-in-one: every tier (the default)
# services = ["event"]       # serve + event box
# services = []              # serve-only: core tier only

Running jobs is not a tier. Whether a process claims and executes the jobs it accepted is [worker] enabled (see Configuration): a request node runs [worker] enabled = false and still accepts every submission; a compute node runs services = [] with [worker] enabled = true, kinds = [...] and works what the request nodes queued.

A deployment advertises exactly the tiers it mounted over the wire, so a client can negotiate capability before calling a verb:

info = db.get_server_info()
# {"version": "...", "features": [...], "storage_backends": [...],
#  "services": ["core", "eval", "event"]}
if "eval" in info["services"]:
    db.eval_per_query(...)

Reaching a verb whose tier was not mounted returns a truthful Unimplemented (“not enabled on this deployment”) rather than a misleading success — the service-mount analog of the client’s build-by-capability connect(target).

Runtime config, no compile features. Every tier compiles into every build — no cargo feature gates a tier, so there is no compile ceiling for the selection to hit; a token naming no tier is a startup error, not a silent drop. Override the selection with JAMMI_SERVER__SERVICES (all, or a comma-separated token list — empty for serve-only).

GPU configuration

For GPU-accelerated inference in production:

[gpu]
device = 0            # CUDA device index
memory_limit = "auto"
memory_fraction = 0.9
require_gpu = false   # fail fast if the GPU is unavailable instead of CPU fallback

[inference]
batch_size = 64
max_loaded_models = 3

Set gpu.device = -1 for CPU-only deployment. On a GPU build, an unavailable device degrades to CPU with a warning by default; set gpu.require_gpu = true to fail fast instead.

Environment variable overrides

Every config field can be overridden with environment variables, useful for containerized deployments: JAMMI_<PATH>, path segments joined by __, works for every field in the tree — not a hand-picked subset — and a bare JAMMI_<FIELD> (no __) works for a top-level one. See Configuration for the full rule.

JAMMI_SERVER__FLIGHT_LISTEN=0.0.0.0:9081 \
JAMMI_GPU__DEVICE=-1 \
JAMMI_LOGGING__FORMAT=json \
jammi-server

A production shape, entirely from the environment

A Postgres catalog, a JetStream broker, an S3 result root, and a file-backed audit signing key — Shape C’s stack — need no TOML file at all: every field resolves from JAMMI_* variables through the same layered loader.

export JAMMI_CATALOG__POSTGRES__URL="postgres://jammi:${POSTGRES_PASSWORD}@postgres.internal:5432/jammi?sslmode=verify-full&sslrootcert=/etc/ssl/certs/ca-certificates.crt"
export JAMMI_CATALOG__POSTGRES__POOL_SIZE=16
export JAMMI_BROKER__JET_STREAM__URL="nats://nats.internal:4222"
export JAMMI_BROKER__JET_STREAM__CREDENTIALS__FILE=/run/secrets/nats.creds
export JAMMI_STORAGE__RESULT_ROOT="s3://jammi-results/prod"
export JAMMI_STORAGE__CLOUD__S3__REGION=us-east-1
export JAMMI_SIGNING_KEY__FILE__PATH=/run/secrets/jammi-audit-master-key
export JAMMI_MODELS__HUB_TOKEN__FILE=/run/secrets/hf-token
export JAMMI_SERVER__SERVICES=all
jammi-server

The equivalent TOML file — the two are interchangeable, and either one overrides the other’s fields when both are present:

[catalog.postgres]
url = "postgres://jammi:${POSTGRES_PASSWORD}@postgres.internal:5432/jammi?sslmode=verify-full&sslrootcert=/etc/ssl/certs/ca-certificates.crt"
pool_size = 16

[broker.jet_stream]
url = "nats://nats.internal:4222"
credentials = { file = "/run/secrets/nats.creds" }

[storage]
result_root = "s3://jammi-results/prod"

[storage.cloud.s3]
region = "us-east-1"

[signing_key.file]
path = "/run/secrets/jammi-audit-master-key"

[models]
hub_token = { file = "/run/secrets/hf-token" }

[server]
services = "all"

Health, readiness, and metrics

The server exposes three HTTP side-channel endpoints on port 8080:

curl http://localhost:8080/healthz
# {"status":"ok","version":"0.8.0"}

curl http://localhost:8080/readyz
# {"status":"ready"}

curl http://localhost:8080/metrics
# jammi_grpc_requests_total 0
# jammi_flight_queries_total 0
# jammi_eval_invocations_total 0
# jammi_search_latency_seconds_bucket{...} 0

/healthz is a liveness probe — 200 while the process can keep its leases and its claim loop alive; 503 {"status":"unhealthy","lease_keeper": false|true,"claim_loop":"…"} when the lease keeper thread is dead (every lease this process holds is lost) or the claim loop task panicked (claim_loop: "failed"). A stopped or aborted loop, a process with no loop, and a DRAIN in progress are all 200 — liveness decides restarts, never routing, and there is no slow-step detection (the runtime owns “how long is too long”). /readyz is a readiness probe — 200 means the catalog backend responded; 503 means it didn’t, or the server is draining ("detail": "draining"), and traffic should be drained from this instance. Point your load balancer at /readyz.

/metrics exposes a small, substrate-level set of Prometheus counters (gRPC requests, Flight SQL queries, eval invocations, refusals at the [server.limits] edge) plus a search-latency histogram, and five gauges:

A process without a claim loop omits the worker families (absent, never 0). No gauge carries a tenant label. jammi_jobs_queued{kind} is the autoscaling input for a compute tier; a scrape issues no catalog statement.

What the server can and cannot do

The Flight SQL surface is a query interface (read path); the ML operations are not SQL, so they ride the typed gRPC surface instead. Set up your data and run training/inference through the Rust library, the jammi-ai / jammi-client Python package, or — for a remote engine — those same verbs over gRPC, then query the results over Flight SQL. The CLI is a strict gRPC client that registers sources and drives the admin surfaces against a running server; it carries no ML verbs and does not run the engine in-process.

The typed gRPC surface is what an edge runtime speaks (it has no HTTP/2 client for Flight SQL’s bidirectional streaming). EmbeddingService serves AddSource, GenerateEmbeddings, EncodeQuery, and Search over plain gRPC — and, since tonic-web is mounted, over gRPC-web — so an edge function running the engine as a sidecar can ingest, encode, and search without the library. Search accepts a precomputed vector or an existing row_key (query-by-example, with the vector resolved inside the engine); see Semantic Search. JobService (core, always mounted) serves all three training kinds over gRPC and InferenceService.Predict serves a trained context predictor — so a client can offload training and prediction to a GPU server with the same verb surface the embedded engine exposes.

Shutdown: DRAIN and RELEASE

The server has two shutdown modes, PostgreSQL’s mapping: SIGTERM = DRAIN (smart) and SIGINT = RELEASE (fast); any signal received while draining is a RELEASE. There is no engine-side drain timeout — the runtime’s grace period (terminationGracePeriodSeconds, stop_grace_period) bounds a DRAIN, then SIGKILL.

DRAIN (kill -TERM, docker stop, a Kubernetes pod deletion): /readyz flips to 503 {"status":"not_ready","detail":"draining"}; the listener closes and in-flight requests finish; every idle WaitJob / Subscribe stream is ended with UNAVAILABLE “server draining” (counted under jammi_grpc_refused_total{reason="draining"}); the embedded worker finishes the job it is running — its lease keeps renewing, every epoch bundle lands, the job reaches completed under the same attempt — and claims no more. Then the catalog is released and the process exits 0. An in-flight unary (an inline run_now such as GenerateEmbeddings) is bounded only by the grace period. A DRAIN that outlives the grace is SIGKILLed: the running job’s lease then expires after one [lease] duration_secs, a successor requeues it (resuming from its last epoch bundle) and it consumes one attempt. A gang RANK this process holds for another host’s training job ([server] peer_advertise) ends at once on DRAIN with the Drain reason: the coordinator hands that job’s lease back (releases + 1), so a rolling restart of the peer tier costs the job no attempt — the next attempt lands within one idle poll over the members still listed. Every other way a rank is lost mid-run — the process SIGKILLed, its stream dropped, or silent past [worker] rank_timeout_secs — retires the coordinator’s attempt with no terminal write: its lease is left to expire, a successor requeues the job and the retry consumes one attempt.

RELEASE (kill -INT, Ctrl+C, jammi-server release, or a second SIGTERM): connections are severed at once; every job lease this process holds is handed back — the row stays running under this instance with lease_expires_at = NULL and releases + 1, and a compute job’s linked building-table lease is NULLed with it — the loop is stopped (cooperatively while no job is under a hold, by abort while one is), the workers row is deleted, the catalog released, and the process exits 0 at once, or exit code 3 when its own evidence does not confirm every lease was handed back (see below). A released row is claimable by any other worker within one [worker] idle_poll_secs, never one lease window; the reclaim cap counts attempts - releases, so a rollout storm of releases never burns the three attempts a genuine crash does. The abandoned training thread never finalizes: its lease is gone and its next epoch boundary bails without a bundle, so the _resume manifest epoch never advances past the last landed one PROVIDED the RELEASE confirmed (exit 0) — under a degraded RELEASE (exit 3) one further epoch bundle may still land before the affected hold’s lost flag flips at its next renewal. Three named exceptions (§3.5 of the design): a claim caught between its COMMIT and its hold registration past one heartbeat keeps its live lease and is recovered by the expiry path (one lease window, attempts + 1, never failed); a compute job whose linked building sweep errored while the jobs sweep succeeded makes the successor back off once (one lease window) before it re-materializes; and a RELEASE whose own sweep statement for a given lease’s table itself errored (that statement’s own Err, read back as None) — that lease was never written and falls to the expiry path, so a successor reclaims it within one [lease] duration_secs rather than one idle poll.

Exit code 3 (DEGRADED) does not by itself say which lease, if any, is still live — it means at least one determinant of a confirmed release is missing, and WHICH determinant is missing decides what is and is not established; there is no single consequence for “degraded”:

• The RELEASE call itself returned an error: nothing further is established — whether any lease this instance held was handed back is unknown.
• A sweep statement for a given lease’s own table itself failed (the exception above): that is the one case with a definite, table-specific cost. The jobs table costs one attempt (attempts + 1, releases untouched) within one [lease] duration_secs; the linked building row lives in result_tables, which carries no attempts/releases columns at all, so its cost is the “backs off once” recovery documented above, never an attempt.
• The keeper’s per-hold pass could not be confirmed to run (Unobserved), or reported a per-hold failure, while both sweep statements still confirm: every row the sweep itself matched — queued, not under an active hold — is confirmed released, since that UPDATE’s own commit is what the sweep’s Some count reports. Nothing is established about a job under an active hold at that moment: the sweep predicate never matches a held row, and the pass whose job it was to release that hold is exactly the one whose evidence is missing.
• The loop’s terminal state was not genuinely witnessed (stop_witnessed == false) while both sweep statements still confirm: every row the sweep matched by the time it ran is confirmed released. Nothing is established about a claim that commits AFTER the sweep runs — stop_witnessed == false means precisely that such a claim is not ruled out, and a claim like that is outside the sweep’s predicate, keeping a live lease with releases untouched and falling to the expiry path, the same shape as the exception above.

The process still exits at once either way (exit code 3, never a hang).

jammi-server release [--pid N] sends SIGINT to N (default 1, the container entrypoint) and exits 0 when the signal was sent — that is ALL its own exit code means; it signals a process that is not its child, so it cannot wait on the RELEASE it triggered and never reports that outcome. It is the uniform RELEASE actuator for a preStop hook, since the distroless images carry no shell for kill, and it knows nothing about jobs. The RELEASE outcome itself (confirmed or degraded) is read from the serving process’s own exit code (0 or 3) via the supervisor (lastState.terminated.exitCode, docker inspect --format='{{.State.ExitCode}}') — under restartPolicy: Always (or equivalent), exit code 3 restarts identically to 0. The library reaches the same mechanism through EmbeddedWorker::release_and_stop and Python’s close(release=True); there the process survives, so a thread that reaches finalize before a successor claims may still land completed — the one documented divergence from the server, which exits.

ListWorkers / jammi workers show each claim loop’s state: warming (the process is preloading; nothing claimed yet), claiming, or draining.

The identity seam

The server performs no authentication on its own — treat it as trusted-network and put access control in front of it, or supply your own TenantResolver at the seam the engine ships for exactly this. The contract of record for that seam is on the security page; the sketch below is only the minimal shape a caller wires in. See:

• Security Posture for the full threat model — what the engine defends, what it explicitly does not, and the trusted-network assumption every deployment inherits; and
• Bring your own auth for the TenantResolver seam itself: one resolver, plugged into assemble_grpc_chain once, authenticates both the gRPC control plane and the Flight db.sql lane.