Datasets and storage reference

datasets is the canonical DRTML I/O contract. A dataset has a role, logical columns, one backend, and optional feed, flush, checkpoint, index, registry, and metric configuration.

Complete output example

storage:
  resume: overwrite               # skip | fail | overwrite
  bindings:
    artifacts:
      backend: s3                 # local | s3
      root: pipeline-data
      profile: minio_app

datasets:
  token_counts:
    role: output
    binding: artifacts
    backend: parquet              # parquet | postgres | qdrant | neo4j
    kind: registry                # plain | registry | observation | vector_vocab
    uniq: token_id
    group_by: [language]

    columns:
      run_id: {type: string, not_null: true}
      language: string
      document_id: string
      token_id: {type: string, not_null: true}
      count: {type: int64, not_null: true, default: "0"}
      context_ids[16:2]: int64

    flush:
      mode: rows                  # rows | mb
      value: 50000
      max_rows_per_part: 250000
      max_mb_per_part: 128
      boundary_columns: [language, document_id]

    checkpoint:
      column: document_id         # column in the feed input
      every_n: 100

    indexes:
      token_id:
        kind: routing
        columns: [token_id]

    metric_defs:
      cardinality:
        method: count_distinct
        column: token_id
        unit_column: document_id
        cadence: checkpoint
        accuracy: exact

Roles and compatibility

The canonical datasets block is split into input/output plans by role.

• input: read-only for this run; may declare run_id_param, aggregate read, feed, and routing indexes.
• output: write target; may declare flush, checkpoint, kind, uniq, group_by, metric profile and metric definitions.

One dataset has exactly one backend. Dual-write is intentionally unsupported: declare two logical outputs if a workflow truly needs two sinks.

Columns

Short form:

columns:
  id: string
  count: int64
  score: float64
  active: bool
  tags: list<string>

Rich form:

columns:
  id: {type: string, not_null: true}
  count: {type: int64, not_null: true, default: "0"}

not_null and default are used by backend schema/DDL generation. Keep domain rows logical; storage performs wire conversion.

Fixed-width logical lists

columns:
  token_ids[15]: int64       # token_ids_0 … token_ids_14
  mask[8:2]: bool            # mask_00 … mask_07

The processor reads/writes one logical list (token_ids, mask). Storage expands it into fixed wire columns and reconstructs it on read. [width:pad] controls numeric suffix padding. The legacy wide_lists block still parses but must not be combined with inline fixed-width columns.

Parquet flush and shard boundaries

flush.mode controls when the in-memory buffer drains:

• rows: value is buffered row count.
• mb: value is estimated buffered MiB.

max_rows_per_part and max_mb_per_part cap physical parquet parts independently of buffer flush cadence.

boundary_columns provides semantic segmentation:

flush:
  mode: rows
  value: 50000
  boundary_columns: [tenant_id, document_id]

Rows are segmented by the tuple (tenant_id, document_id) before part rotation. The framework does not invent a default such as doc_id; every boundary column must exist in columns, must be unique in the list, and is valid only for parquet. PostgreSQL and Qdrant reject flush.boundary_columns.

A boundary is a physical write/rotation concern. It is not the same as feed partitioning or checkpoint cadence.

Checkpoints

checkpoint:
  column: document_id
  every_n: 100

checkpoint.column belongs to the feed input, not necessarily the output dataset. The partition loop counts distinct transitions in that input column and persists registry/output state every every_n units. Multi-worker checkpoint/resume is currently unsupported.

Routing indexes

indexes:
  token_id:
    kind: routing
    columns: [token_id]

For parquet, a routing sidecar maps key values to (shard, row offset). Attach feeds require such an index; full attach-dataset scans are rejected at compile time. route_by defaults to the first join_on column, and an index with that name must exist.

Dataset-level indexes currently parse routing indexes. PostgreSQL physical indexes are declared inside the inline postgres.indexes block (see below).

Input aggregation

An input may ask a backend to aggregate partial counts:

partial_counts:
  role: input
  backend: postgres
  columns:
    language: string
    token_id: string
    count: int64
  group_by: [language, token_id]
  sum_columns: [count]

group_by and sum_columns must be declared together and all columns must exist.

Dataset kinds

• plain: regular rows.
• registry: requires uniq; Qdrant is unsupported. PostgreSQL registry requires inline postgres.
• observation: time/history rows, commonly PostgreSQL.
• vector_vocab: requires uniq; PostgreSQL is unsupported in v1; parquet or Qdrant are valid.

PostgreSQL inline block

backend: postgres
columns:
  run_id: {type: string, not_null: true}
  token_id: {type: string, not_null: true}
  count: {type: int64, not_null: true}
postgres:
  connection: postgres_app
  schema: pipeline
  table: token_counts_v1
  schema_version: "2026.07.3"
  lifecycle:
    on_run_start: assert_schema
  ddl:
    create_table: if_not_exists
    create_index: if_not_exists
    emit: true
  unlogged: false
  primary_key: [run_id, token_id]
  indexes:
    run_token_idx:
      columns: [run_id, token_id]
      unique: true
  on_conflict:
    target: [run_id, token_id]
    action: do_update              # do_update | do_nothing
    returning: [token_id]
    update_columns: []
    update_additive: [count]

The inline block drives static DDL codegen and runtime transport. Runtime never creates or alters step tables. Deploy applies generated DDL; run start can assert schema_version.

PostgreSQL does not require a storage binding (__postgres__ is implicit). postgres_table is a legacy pointer; new datasets should use the inline block.

Qdrant inline block

backend: qdrant
kind: vector_vocab
uniq: token_id
columns:
  token_id: string
  vector: list<float32>
qdrant:
  connection: qdrant_app
  collection: token_vectors_v1
  vector_size: 128
  distance: Cosine              # Cosine | Euclid | Dot
  vector_column: vector
  point_id_column: token_id

The vector and point-id columns must exist. If point_id_column is omitted, uniq is used. Qdrant does not require a storage binding (__qdrant__ is implicit).

Neo4j

backend: neo4j is accepted by the current whitelist so contracts can be authored ahead of transport support. Builtin read/write adapters deliberately raise NotImplementedError until db/neo4j/ exists.

Validation checklist

• Input datasets need columns unless they are mapping-merge inputs.
• uniq and group_by columns must exist.
• Fixed-width lists and legacy wide_lists cannot coexist.
• Routing index columns must exist.
• PostgreSQL/Qdrant blocks are required for their backends.
• Backend/kind compatibility is validated during compile.