July 2026

July 26 — streaming cluster and richer offline evaluation

Fit and assign without a full matrix

Corpus-scale clustering is now a first-class processing path: stream bounded vector batches, fit one shared centroid model, then assign and write in chunks. Parallel workers (threads / processes / Ray under runtime.parallelism) own vector shards; the driver merges centroid stats on fit and assignment aggregates on assign. Peak RAM tracks centroids + one batch + write buffer — not N×D.

Storage contributes iter_vector_batches: a numpy window over logical rows, separate from feed batches and embed-pair batches.

Docs: Streaming cluster, StorageSession, Parallelism.

Evaluation: cluster and graph artifacts

pattern: evaluation no longer stops at vector geometry. Jobs may declare:

artifact_kind: cluster_rows — sample assignments, join vectors by id, emit cluster quality metrics;
artifact_kind: graph_edges — sample edges, emit topology / association metrics.

Compile-time validation checks required columns and method/artifact_kind pairing. Formulas live in algorithms/evaluation/; the engine owns sample, join, and write.

Docs: evaluation, algorithm catalog.

July 25 — probe loop and row assembly

Probe: try candidates, domain picks the winner

Framework runs prepare → nested evaluate → domain decide → publish. Bindings and metrics are opaque DTOs; the step decides which candidate wins. Nested evaluate shares the parent’s parallelism budget (no multiplicative Ray explosion). Probe subsets sample prepare/evaluate only; final publish uses the full upstream stream.

Row assembly

row_assembly loads declared inputs, applies probe subset, calls domain build, and writes only when persist is enabled — shared glue for nested evaluate without I/O in the step.

Docs: Probe loop.

July 24 — shuffle FD limits, bucket reduce workers, batch mean-pool

Close shuffle writers after flush

Early shuffle used to keep one ParquetWriter open per touched bucket for the whole run. With large shuffle.buckets that exhausted process FD limits (Errno 24 Too many open files).

part_rotation.write_table now rotates/closes writers for every bucket touched in the flush. Prefer moderate bucket counts (and/or storage.shuffle_buckets) rather than holding writers open.

`storage.shuffle_buckets`

Optional flattened param. On session open, apply_shuffle_buckets_override rewrites shuffle.buckets on every output that already declares shuffle:. Does not add shuffle to datasets that omit it.

reduce_merge: work unit = shuffle bucket

Parallel reduce no longer means “one Ray task per input dataset”. bucket_plan.build_reduce_work_units emits one unit per discovered shuffle bucket (flat layout → one unit per input group). Units are round-robin assigned; each worker owns a write lane; the driver merges manifests.

Canonical pipeline shape: producer writes shuffled parts → reduce_merge SUMs per bucket.

Docs: reduce_merge, Datasets shuffle, StorageSession.

Batch mean-pool (algorithms)

mean_pool_rows_batch mean-pools many occurrence index lists against one vocab matrix in a single NumPy pass. Used by vocab-index helpers for embed_infer-scale batching. Clustering / K selection remains outside infer (one output row = one occurrence).

Docs: embed_infer, algorithm catalog.

July 23 — honest Ray train, early shuffle, streaming reduce

Embed train: parameter-server W shards

Ray embed_train no longer fakes multi-worker training with a single shared W actor when workers > 1.

Setting	Behavior
`runtime.parallelism.backend=ray`, `workers=1`	one `W` actor (previous single-table path)
`runtime.parallelism.backend=ray`, `workers>=2`	`W` sharded across `workers` actors (`row_id % n_shards`)

Per batch the trainer:

gathers needed rows from owning shards;
runs SGNS update locally (algorithms/embed/sgns_ps.py);
pushes row deltas back to shard actors (ray_w_shards.py).

Step03 embed_train defaults moved to backend: ray / workers: 8. UI help documents the PS semantics.

Early shuffle on parquet outputs

Datasets may declare:

shuffle:
  by: [lemma_token_id]   # columns must exist
  buckets: 64            # >= 1; parquet only

Writes land under stable bucket-NNN/ object prefixes with manifest shard ids like b017-w0-0. Buckets use blake2 over key parts (not process-salted hash()). Part rotation and reduce build consume this layout for spill-friendly merges.

Streaming parquet reads

session_read.iter_parquet_run_rows streams logical rows with optional filters; optional group_by + sum_columns aggregates incrementally over the filtered stream. Peak RAM without group_by tracks one read batch.

read_parquet_run_rows still materializes the full result and is reserved for small fixtures / tests. Hot paths must use the iterator.

reduce_merge scale-out (prelude)

Independent input-key job groups and early-shuffle spill / one-pass multi-output land in reduce/build.py. Per-bucket Ray/process work units and write-lane merge landed on July 24 (see above).

Operator / migration notes

Deploy Ray-capable stack (drtoller update / rebuild ray-head if needed).
Steps that want reduce-friendly parquet layout must add shuffle: to DRTML outputs and re-run producers.
Prefer fixing OOM with streaming/spill/SQL — not by forcing inprocess / workers=1.

Canonical docs: embed_train, reduce_merge, StorageSession, Datasets shuffle, Ray, Parallelism.

July 22 — embed metrics, evaluation points, train result contract

Shared embed metric catalog

Live and terminal embed gauges live in contracts/embed_train_metric_catalog.py and are consumed by Prometheus emit and observation writers (train and infer metrics steps). Geometry metrics (L2, hubness) and vector evaluation registry methods were expanded for offline evaluation jobs.

Evaluation → observation table

Evaluation point rows can be mapped into the corpus-growth observation shape (evaluation/point_writer.py) for PostgreSQL dashboard continuity (corpus_fraction=1.0, order_policy=evaluation).

Public train_stats

run_embed_train returns EmbedTrainStats.as_public_dict() (and nested public stats) so Airflow XCom / UI consumers never see non-JSON dataclass payloads.

July 21 — dataset metrics and corpus-growth curves

Final and milestone jobs

Offline dataset_metrics jobs now have three explicit modes:

buckets — cumulative stable-hash buckets;
final — one exact merged point;
milestones — deterministic document-progress points.

Runtime selection comes from metrics.mode; only jobs matching the selected mode execute. A metrics step can therefore declare final and milestone jobs separately while sharing outputs and observability.

Per-document evidence

Milestone jobs can declare:

mode: milestones
evidence: document_counts
document_column: doc_id
key_columns: [lemma_token_id]
mass_column: token_count
dataset_label: step01_lemma_vocab_run

document_counts requires:

PostgreSQL executor and PostgreSQL input;
mode: milestones;
a declared document_column;
upstream rows materialized per document and key.

Documents are ordered deterministically using the configured order seed. metrics.milestone_step controls document cadence; compute.corpus_growth.max_milestones limits generated windows.

Direct PostgreSQL final path

Methods marked postgres_direct_final can compute exact final metrics directly against a PostgreSQL source table. Other jobs continue through run-scoped staging. Both paths write the same metric-point contract.

New metric methods must still provide matching Python and PostgreSQL implementations.

Dataset metric capabilities

MetricExecutionMode now includes milestone_exact. Distribution, threshold, graph, and grouped-comparison methods advertise milestone support explicitly; unsupported method/mode combinations fail validation.

The builtin catalog is organized by semantic family:

core distribution;
assignment and growth;
graph and group comparison;
vector and cross-dataset methods.

Use the registry API; do not import family internals as a public extension point.

Grafana

PostgreSQL panels support xychart. Corpus-growth XY queries return:

corpus_fraction as X;
selected metric values as Y.

Grafana variables passed to PostgreSQL use the sqlstring formatter. The run_id variable is scoped by step_id, preventing runs from unrelated steps appearing in the selector.

Step01

Step01 now writes drt_tbl_lemma_document_count_v1 with one row per (run_id, doc_id, lemma_token_id). This is the evidence source for exact milestone curves.

The spaCy processor was split into focused modules for model loading, parsed types, document extraction, token extraction, noun chunks, profiling, and result materialization. Public step behavior remains process_batch / process_one through spacy_batch.

PostgreSQL lifecycle

Deployment enforces:

tables: drt_tbl_*;
functions/procedures: drt_proc_*;
every managed pipeline object listed in pg_object_inventory.txt;
orphan removal during migrate-postgres.sh.

The migration also renames legacy objects before applying framework procedures and generated step DDL.

Operator action

After deploying these changes:

drtoller update
drtoller verify

drtoller update applies PostgreSQL migrations, regenerates dashboards, and recreates affected services. Rerun Step01 to create per-document evidence before running step01_metrics in milestone mode.