Add a dataset backend

Use this recipe only when a storage engine has meaningfully different transport semantics. A new bucket/profile is a binding, not a dataset backend.

1. Define the DRTML contract

Add the backend name to the DRTML whitelist and DatasetBackend literal. For engine-specific settings, create an inline block such as:

datasets:
  graph:
    role: output
    backend: graphdb
    columns:
      source_id: string
      target_id: string
    graphdb:
      connection: graph_app
      graph: ontology_v1
      source_column: source_id
      target_column: target_id

Create a typed immutable spec in storage.plan, parser under drtml/storage, and cross-field validation. Reject unsupported kinds, flush options, indexes and missing columns during compile.

2. Implement external transport

Place client creation, connection resolution, schema assertions, batch write/read and engine queries under framework/db/graphdb/. Do not import processing or telemetry there.

Runtime schema creation is forbidden for deployment-managed systems. Add static codegen/migration support if the backend needs schemas/indexes.

3. Implement adapters

Implement both protocols when the engine is readable and writable:

class GraphWriteBackend:
    name = "graphdb"

    def write(self, session, dataset_key, rows, *, shard_id="0") -> None: ...
    def flush_buffers(self, session) -> None: ...
    def drain_all_buffers(self, session) -> None: ...

class GraphReadBackend:
    name = "graphdb"

    def read_manifest_total_rows(self, session, dataset_key, *, run_id): ...
    def read_run_rows(
        self, session, dataset_key, *, run_id,
        filters=None, group_by=None, sum_columns=None,
    ): ...
    def iter_run_rows(
        self, session, dataset_key, *, run_id,
        filters=None, group_by=None, sum_columns=None, fetch_rows=65_536,
    ): ...

Adapters are thin dispatch objects. Put substantial buffering/query logic in focused storage/run/<backend>_* modules and transport details in db/<engine>.

4. Register adapters

register_write_backend(GraphWriteBackend())
register_read_backend(GraphReadBackend())

Register builtins idempotently. External plugin registration must happen during framework/bootstrap initialization before session dispatch.

Never add if spec.backend == "graphdb" to StorageSession, session_write, or session_read.

5. Preserve logical rows

The public contract is logical row dictionaries. Backend wire representations, vector payloads, graph edges and database types must remain behind the adapter/transport. If a generic list encoding is needed, reuse storage wide-row/materialization helpers rather than duplicating conversion.

6. Add contract tests

Start with a fake in-memory adapter:

• registry resolves the name;
• StorageSession.write dispatches to it;
• flush and drain calls propagate;
• read/iter/row-count methods preserve logical rows;
• unknown names fail with registered-name diagnostics.

Then add engine integration tests for filters, aggregation, conflict semantics, batching, reconnect/failure, schema mismatch and idempotent close.

7. Complete the feature

• Add metric backend capabilities if dataset metrics are supported.
• Add DDL/provisioning codegen if required.
• Add connection integration for local/Airflow deployment.
• Update DRTML/storage references and FILES.md.
• Run layer-boundary tests.

A parser-only backend or a registered adapter that always silently drops unsupported operations is incomplete. Use an explicit NotImplementedError stub when reserving a backend name.