PostgreSQL sink

Role

Backend postgres — registry/observation/plain tables in the dataset’s configured schema (pipeline is the conventional default). Transport: framework/db/postgres/. Adapter: backend_builtins._Postgres*.

In the current Compose reference, pipeline tables live on postgres-pipeline (DB drtoller, published on VPC :5432 for KubeRay). Airflow metadata is a separate postgres container with no host/VPC publish. Celery results use Redis, not that metadata DB. drtoller update migrate and Grafana SQL target the pipeline instance.

Naming (mandatory)

Every object in schema pipeline uses a fixed prefix. Validated by framework/db/postgres/naming.py (require_table_name / require_proc_name) at DRTML parse and DDL codegen.

Kind	Prefix	Example
Table	drt_tbl_	pipeline.drt_tbl_lemma_run_staging_v1
Stored procedure / function	drt_proc_	pipeline.drt_proc_dataset_metric_snapshot

Exception: catalog table pipeline.schema_version (no prefix). Rows in schema_version.table_name still store the physical name (drt_tbl_* or drt_proc_*).

Forbidden:

• new tables/procs in pipeline without the prefix;
• dual names (legacy + prefixed) after migrate — orphans are dropped;
• Python / DRTML / Grafana references to bare names without drt_tbl_ / drt_proc_;
• domain SQL or CREATE TABLE during a run.

Where objects are declared

Object	Source of truth	Deploy
Step / observation table	DRTML datasets.*.postgres.table (or v4 physical.table) = drt_tbl_…	sync-postgres-ddl → *_generated_schema.sql → migrate
Platform tables (metric window, staging, …)	platform/stack/postgres-init/framework-procs/**	05-framework-procs.sql
Platform procs	same SQL: CREATE OR REPLACE FUNCTION pipeline.drt_proc_…	05-…
Allowlist	postgres-init/pg_object_inventory.txt	drop-pg-orphans.sh

New object → add a line to pg_object_inventory.txt immediately, or the next drtoller update will drop it as an orphan.

Migrate sequence (drtoller update → migrate-postgres.sh)

1. Codegen step DDL
2. Bootstrap (04)
3. 07-rename-to-drt-prefixes.sql — rename legacy tables → drt_tbl_* (data kept)
4. Framework procs (05) — create/replace drt_proc_* / drt_tbl_*
5. Drop known legacy (06)
6. Apply step generated DDL
7. drop-pg-orphans.sh — drop anything in pipeline not listed in inventory
8. Print schema_version

Orphan = table/function in pipeline missing from inventory (e.g. old step02 count tables). Do not keep “just in case” tables outside the inventory.

Cleanup is set-based: it removes every overload of an unlisted function and deletes the corresponding schema_version rows. Corpus-growth deployment includes generic final, document-staging, source-relation, and key-dispatch procedures; callers reference managed pipeline.drt_tbl_* source relations, never arbitrary bare tables.

DRTML (inline only)

PostgreSQL datasets declare connection/schema/table inline under datasets.*.postgres.
Top-level postgres: / postgres.tables and dataset postgres_table pointers are rejected at parse/validate.

datasets:
  lemma_vocab:
    role: output
    backend: postgres
    kind: registry
    uniq: lemma_token_id
    columns:
      run_id: { type: string, not_null: true }
      lemma_token_id: { type: string, not_null: true }
      token_count: { type: int64, not_null: true }
    postgres:
      connection: postgres_app
      schema: pipeline
      table: drt_tbl_lemma_run_staging_v1
      schema_version: "2026.07.3"
      on_conflict:
        target: [run_id, lemma_token_id]
        action: do_update
        update_additive: [token_count]
    checkpoint: { column: doc_id, every_n: 100 }
    metric_defs:
      cardinality: { method: count_distinct, column: lemma_token_id }

For a non-registry dataset, every on_conflict.target must exactly match either:

• postgres.primary_key; or
• a declared postgres.indexes.* with unique: true.

Compilation rejects an unbacked conflict target; otherwise PostgreSQL would fail only at write time.

DDL vs runtime

When	What
Deploy	ddl_codegen → static/postgres/ → migrate-postgres.sh
Run start	assert_schema_version
Run	INSERT/UPSERT / metric snapshot procs (drt_proc_*)

Forbidden at runtime: CREATE TABLE / ALTER.
Forbidden in platform init: step-specific DDL (bootstrap + generic procs only).

Saturation

See Saturation. Homogeneity uses a single formula B.

Bulk-write contract

One buffer drain must become one bulk database operation:

• use execute_values, COPY, or one stored-procedure call with a batch payload;
• split large writes into explicit pages when needed;
• never use executemany, per-row INSERT, or per-row commit;
• keep SQL and connection handling in framework/db/postgres/; storage adapters only dispatch and coordinate buffers.

Runtime DDL remains forbidden regardless of the write path.

On Ray partition_loop, compute actors do not each open Postgres. They enqueue rows to one PG writer actor (processing/partition_loop/parallel/backends/pg_writer.py) so max_connections stays independent of workers.

Checklist (PG change)

1. Name uses drt_tbl_ / drt_proc_?
2. Line added to pg_object_inventory.txt?
3. DRTML table: / Grafana function: / Python SQL use the same names?
4. Dual-impl metrics (Python + PG) stay in sync?
5. No runtime DDL — only deploy migrate?