Skip to content
DrToller.Framework

Dataset backend adapters

Problem

Section titled “Problem”

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

Solution (P1)

Section titled “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 calls write_backend_for(spec.backend); session_read.read_run_rows and row iteration call read_backend_for(spec.backend).

Rules

Section titled “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.
  5. Multi-worker remote Ray: parquet stores with backend: local are rejected when ray_address is set and workers > 1 (or DRTOLLER_REQUIRE_OBJECT_STORE=1). Use backend: s3 (MinIO / AWS / GCS interop) so every actor sees the same keys — Ray object-store guard.

Transport vs adapter

Section titled “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

Section titled “Test”

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

Cookbook

Section titled “Cookbook”