Parallel training reduction

Objective and contracts

Distribute Training Run computation across the configured CPU budget while preserving the recommendation and source contracts in the service specification. One consistent, ordered source transaction remains responsible for scope, ordering, and cutoff validation. Raw rows and source Session/Order identifiers never enter work-store tables or worker batches. Serving and publication are unchanged.

Evidence and options

The previous implementation consumes rows, enumerates all product pairs, updates Python counters, and builds string-based NumPy columns on one thread. Only one DuckDB insertion can overlap it. A running container used approximately 1.14–1.18 cores out of 12 available, without a CPU quota. A synthetic 200,000-row, 1.9-million-pair reduction reproduced approximately one-core utilization.

• Python threads around the existing counters retain Python execution contention.
• Process workers bypass that contention, but require serialization, separate source-transaction handling or a dispatcher, and a separate memory budget per interpreter. Do not transfer raw rows or source credentials to child processes.
• Native reduction moves pair expansion and counting into DuckDB, using bounded Arrow inputs and thread-local cursors. It reuses the existing spillable aggregate store and exact integer/decimal semantics. This is the selected implementation to measure.

Primary references: DuckDB thread-local cursors, Arrow ingestion, Python process executors, and sparse-dot-topn parallel multiplication.

Implementation plan

1. Keep streaming validation, group boundaries, oversized exclusions, and exact popularity accumulation in the source consumer. At group completion, retain only sorted unique product and category memberships plus the temporal partition; discard the source identity. Combine identical summaries with an integer multiplicity.
2. Bound these summaries by their possible expanded contribution, including singleton support. Expand product/category pairs and count support in native SQL over Arrow arrays. Persist only the existing derived aggregate tables. Replace stringified numeric/date ingestion with typed Arrow inputs while preserving the existing decimal rounding boundary.
3. Use a bounded set of background native tasks with thread-local cursors. Apply backpressure before constructing another submitted batch, join before reading, propagate task failures, and join every task before closing the connection and deleting scratch. threads: 1 stays synchronous. No unbounded executor submission or additional source readers.
4. Pass the same configured thread count to metadata sparse multiplication and enable OpenMP in the Linux build. Keep candidate ordering and numeric behavior stable.
5. Verify serial/parallel and reference-reducer equivalence, holdout boundaries, repeated rows, exact decimals, unusual identifiers, exclusions, boundedness, overlap, failures, and cleanup. Measure elapsed time and total process CPU on deterministic synthetic workloads, then run repository gates and installed-package/container checks.

Acceptance

Observable reductions and recommendation results remain equivalent across CPU settings. A representative pair-heavy synthetic workload must demonstrate improved throughput and execution on multiple cores. Report source-consumer bottlenecks and small-input overhead honestly; CPU saturation alone is not success. No production source, running service, or published snapshot is modified during verification.

Implementation and verification record

Implemented in pipeline/workstore.py with Arrow 23.0.1 and the existing DuckDB 1.5.5 lock. pipeline.threads also reaches sp_matmul_topn(n_threads=...); the Docker source build now enables OpenMP. No API, migration, source query, durable evidence schema, or serving contract changed.

The final ingestion budget reserves one CPU for the consumer, floor(threads/2) background callers, and the remaining CPUs for a shared DuckDB pool. All native tasks share one DuckDB memory/spill budget. Python/Arrow buffers have separate bounds documented in the implementation reference. View popularity uses integer counters; purchase popularity preserves exact Decimal accumulation and the existing insertion rounding boundary.

Checks

Command / check	Result
make lock and make setup	Passed; added locked Arrow dependency
make test-focused TEST=tests/unit/test_workstore.py	23 passed; reference equivalence, bounded concurrency, partitions, exact quantities, failures, cleanup
make test-focused TEST=tests/unit/test_content.py	4 passed; serial/parallel scores and ordering, including truncated top-N
make test-focused TEST=tests/integration/test_end_to_end.py	12 passed; complete results equivalent and previous serving head preserved after native failure
make -k test	331 unit, 38 contract/delivery, 29 integration tests passed; overall command fails on pre-existing lint/types below
make docs-check architecture-check (also run by make -k test)	Passed
Ruff on changed pipeline, benchmark, and test files	Passed
make smoke	Passed; wheel installed and verified outside the source tree
make compose-check	Passed; both Compose definitions validate
docker build --build-context telemetry=../telemetry -t commerce-recommendations:parallel-training-check .	Passed on Linux ARM64
Isolated image check of OpenMP capability and four-thread sparse multiplication	Passed
git diff --check	Passed

The unchanged src/recommendations/synthetic/generator.py has five existing Ruff failures (docstring returns/raises, nested context managers, spacing, long line) and three existing mypy errors assigning None to inferred strings at lines 691, 692, and 699. Its contents match the base commit. These unrelated failures prevent a green aggregate make test/make verify; no gate was disabled or weakened. make test-postgres was not run: no disposable PostgreSQL test URL was provided, and this change does not alter PostgreSQL storage. No standalone migration check was needed; the normal integration gate ran both migration tests.

The initial work-store run had 11 passing tests and one failing test that encoded the old one-in-flight-batch rule. The replacement tests verify the new configured bound and prove three native group tasks can overlap. Preliminary profiling, synthetic baseline comparisons, read-only docker stats/cgroup/thread inspection, and image/source checksums established the original one-core behavior. An initial container probe used the wrong private extension import path; the corrected probe checked the package's exported OpenMP capability and passed. A temporary local baseline rerun was invalidated when make smoke cleared its generated module; image comparisons use the same committed benchmark against the old and new installed code instead.

Measured result

Sequential runs on the same Linux ARM64 Docker host, 12 CPUs available, no CPU quota. Each run used scripts/benchmark_training_reduction.py, 50,000 groups of 20 distinct products, one million views, 9.5 million pair contributions, and batch_key_limit=100000. Aggregate pair, support, popularity, and group totals were verified after the timed interval.

Installed implementation	Threads	Wall seconds	CPU seconds	Average occupied cores	Rows/second
Base commit 3767655, existing local image	4	16.305	19.068	1.169	61,332
Native reduction, final verification image	4	4.153	7.465	1.797	240,792
Native reduction, final verification image	8	2.911	7.677	2.637	343,501

The default four-thread setting was 3.9 times faster on this workload; eight threads was 5.6 times faster than the original. Native reduction both removes work and executes on multiple cores. These are development microbenchmarks on a shared machine, not a qualification or production speedup claim. No other agent-started CPU benchmark or test ran concurrently with these image comparisons.

The container invocation was the following, run sequentially for commerce-recommendations:local at four threads and commerce-recommendations:parallel-training-check at four and eight threads:

docker run --rm --network none --read-only --tmpfs /tmp \
  --mount type=bind,src=/Users/pierre-luc-delisle/PycharmProjects/recommendations/scripts/benchmark_training_reduction.py,dst=/benchmark.py,readonly \
  commerce-recommendations:parallel-training-check \
  /app/.venv/bin/python /benchmark.py --threads 8

Local macOS checks with make benchmark-training also covered one/four/eight threads and group sizes 3, 20, and 100. Small groups stayed near one core because serial input/popularity work dominates and membership summaries compress well; the native version still improved elapsed time. The 100-product workload exercised 49.5 million pair contributions and used multiple cores.

Operational handoff: deploy/rebuild the worker through the normal authorized workflow to activate the source change. The existing worker was not restarted. The default remains four threads; larger thread budgets trade more bounded Arrow buffers and native work for greater concurrency. Source I/O/validation, popularity accumulation, candidate materialization, and portions of evaluation remain serial. No claim is made that all configured cores stay saturated throughout an entire Training Run.