Dataset backend adapters

Problem

StorageSession used to branch on if backend == …. Adding a sink meant editing session code.

Solution (P1)

Protocols + registry:

storage/dataset/backend_registry.py

class DatasetWriteBackend(Protocol):
    name: str
    def write(self, session, dataset_key, rows, *, shard_id="0"): ...
    def flush_buffers(self, session): ...
    def drain_all_buffers(self, session): ...

class DatasetReadBackend(Protocol):
    name: str
    def read_manifest_total_rows(...): ...
    def read_run_rows(...): ...
    def iter_run_rows(...): ...

Builtins (backend_builtins.py): parquet, postgres, qdrant, neo4j stub.

session_write.write / session_read.read_run_rows only call write_backend_for(spec.backend).

Rules

1. One dataset — one backend (dual-write forbidden).
2. Steps do not branch on backend.
3. New sink: adapter + register_*_backend (+ db/<engine>/ transport) — no if in session.
4. DRTML whitelist: drtml/storage/backend.py.

Transport vs adapter

Backend	Adapter	Transport
parquet	builtin → session_write.write_parquet	local/S3 backends/
postgres	builtin → postgres_write	db/postgres/
qdrant	builtin → qdrant_write	db/qdrant/
neo4j	stub NotImplementedError	future db/neo4j/

Test

dev/tests/test_dataset_backend_adapter.py — fake memory_test backend via the registry.

Cookbook

New dataset backend