Datasets and stores

Datasets live under inputs and outputs. Role is implied by the section. Physical backends are declared once in storage.stores.

Stores

storage:
  default: artifacts
  resume: overwrite
  stores:
    artifacts: {backend: local, root: /tmp/drtoller-data}
    warehouse:
      backend: postgres
      connection: postgres_app
      schema: pipeline
    vectors:
      backend: qdrant
      connection: qdrant_app

• Object stores (local / s3) become storage bindings.
• Database/vector stores keep connection defaults; datasets add physical: overrides.
• storage.default is used when a dataset omits store.

Input

inputs:
  documents:
    store: artifacts
    schema: {document_id: string, text: string}
    run: mapping.upstream_runs
    feed:
      batch: 256
      group_by: [document_id]
      group_scope: shard

Useful fields:

• schema: column types (replaces columns).
• run: run-id param (replaces run_id_param).
• feed / attach: primary or attach feed.
• aggregate.by / aggregate.sum: reduce-style input aggregation.
• identity: mapping merge keys (by, source, hash, count).
• as: attach alias for processing.feed_attach lowering.

Output

outputs:
  uppercase_documents:
    schema: {run_id: string, document_id: string, text: string}
    from: documents
  document_registry:
    kind: registry
    key: document_id
    schema: {run_id: string, document_id: string}
    measurements:
      cardinality:
        method: count_distinct
        column: document_id
        publish:
          name: unique_documents
          kind: gauge

Useful fields:

• from: logical processor output key → physical dataset.
• key: registry / vector uniq column.
• group: split dimension.
• kind: plain | registry | observation | vector_vocab.
• flush / checkpoint: write cadence.
• shuffle: early bucketed layout for parquet (see below).
• physical: postgres/qdrant table/collection overrides.
• measurements: dataset metrics (+ optional Prometheus publish/panel).

Early shuffle (parquet)

Declare on a parquet output so writers spill by stable key buckets (reduce-friendly):

outputs:
  lemma_train_pairs_run:
    schema: { … }
    shuffle:
      by: [kind, center_lemma_token_id, context_lemma_token_id]
      buckets: 64

Rules:

• by columns must exist in the dataset schema;
• buckets ≥ 1;
• parquet only (not postgres / qdrant);
• prefer moderate buckets (one filling writer can stay open per touched bucket until the part-size limit).

Runtime override: storage.shuffle_buckets rewrites buckets on every shuffled output at session open (does not invent shuffle: on plain outputs).

Runtime (storage/dataset/shuffle.py + part rotation):

• bucket id = blake2 over key parts (not process-salted hash());
• parts under bucket-NNN/;
• parts grow to max_mb_per_part — the bucket writer stays open across flushes until that limit;
• manifest shard ids bNNN-… (for example b017-w0-0);
• no post-hoc compact: driver merge persists the manifest only.

Downstream reduce_merge treats each shuffle bucket as a parallel work unit when the upstream layout is fully shuffled.

Backends

Backend	Store profile	Dataset extras
parquet	local / s3	binding via store
postgres	backend: postgres	physical: {schema, table, on_conflict, …}
qdrant	backend: qdrant	physical: {collection, vector_size, …}
neo4j	stub	requires explicit binding until db/neo4j/ lands

One dataset → one backend. Dual-write is forbidden.

For a PostgreSQL store:

• physical.table in schema pipeline must use drt_tbl_*;
• a non-registry physical.on_conflict.target must match physical.primary_key or an exact unique index;
• table creation/changes happen only during deploy codegen/migrate.

See PostgreSQL naming and lifecycle.