StorageSession

Create

session = StorageSession.from_plan(
    plan,  # StoragePlan from compile / parse_storage_plan
    run_id=...,
    params={"user_id": ..., "project_id": ...},
)

Tenant paths: user_id / project_id → object keys under the binding root.

Write path

session.write(dataset, rows, shard_id=…)
  → output_spec.backend
  → write_backend_for(backend).write(…)
  → parquet buffer / PG upsert / Qdrant upsert

Flush / drain on close iterates all registered write backends (builtins are lazy-registered).

Read path

• read_batches — parquet-only (RecordBatch).
• read_run_rows / iter_run_rows — via read backend (parquet/pg/qdrant).
• Logical rows: iter_logical_rows → row_materialize.

Checkpoint

Cadence from datasets.*.checkpoint (column from feed input, not vocab) → partition_loop glue.
flush.boundary_columns — parquet only; postgres/qdrant → validate error.

Part rotation

run/part_rotation.py — part-*.parquet by MB/rows/boundary.
Shard names may include a worker prefix (w0-0) under parallel write.