ADR 0003: Stages and aggregations outside the graph by SimonHeybrock · Pull Request #245 · scipp/sciline

SimonHeybrock · 2026-09-11T10:30:58Z

Proposes replacing Pipeline.map/reduce with two objects built from an ordinary flat pipeline, outside the graph: a Stage, the part of a pipeline from named input keys to named output keys with the static part computed once and held, and an Aggregation, two stages with an accumulator per accumulation key between them. This PR is the additive half: the ADR (status proposed), the design document with a rollout plan, the two modules, their tests, and a user-guide page. map, reduce, and cyclebane are untouched; nothing breaks.

Why

Map/reduce inside the graph has been implemented twice and is the source of every breaking change and open question in the PEP 695 generics prototypes (#236, #237). Three consumers need an operation it cannot express: run one part of a graph repeatedly with some values supplied per call, combine the results outside, run the rest once. ess.reduce.streaming.StreamProcessor builds that partition by hand (#241 records the requirements), the essapps reduction-service architecture needs it in three places, and every ESS reduction package folds runs at one or more keys and grafts the result back onto the pipeline.

The ADR states the decision, the alternatives tried and rejected during prototyping, and the consequences. The design document has the semantics, the survey of every map/reduce site in the ESS packages and esslivedata, the evidence, and a PR-level rollout plan across sciline, ess.reduce, the reduction packages, esslivedata, and essapps.

What is in the PR

docs/developer/adr/0003-replace-map-reduce-with-stages-outside-the-graph.md: the decision. Read this first.
docs/developer/architecture-and-design/map-reduce-outside-the-graph.md: design, survey, evidence, migration, rollout plan, settled and open questions.
sciline.Stage, sciline.warm: the graph partition.
sciline.Accumulator (protocol), sciline.Aggregation, sciline.compute_members: the table-fold shape and the replacement for compute_mapped.
sciline.Buffered and sciline.Reduced: accumulator factories for an n-ary function applied to all pushed values, and for a running result of an associative binary function. Reduced is the memory-cheap choice for sums of large arrays; it never updates in place, so it cannot corrupt values the caller still holds.
docs/user-guide/stages-and-aggregations.ipynb: a guide next to the parameter-tables page. It does not claim to replace anything while the ADR is proposed, but reviewers may find it the quickest way to see the semantics.
docs/developer/architecture-and-design/loki_validation.py: the esssans multi-run reduction over stages against the map/reduce reference. Results identical, provider call counts identical.
scheduler_or_default in sciline.scheduler, shared by TaskGraph and Stage.

Tracking issues: #246 for the sciline side, scipp/ess#746 for ess.reduce, the reduction packages, and esslivedata.

Not in the PR

provide(key, callable), a reporter argument on Stage, and visualization over stages are listed as a follow-up in the rollout plan. The breaking release that removes map/reduce waits until the ESS packages and esslivedata have migrated; the plan gives the order.

Test plan

New tests in tests/stage_test.py and tests/aggregation_test.py; the full suite passes.
mypy strict is clean on the new modules.
loki_validation.py run locally against the reference: identical results and call counts.
The user-guide notebook executes cleanly under nbconvert.

🤖 Generated with Claude Code

A design proposal with a prototype. Sciline's map/reduce put a loop over members and a combine step inside the graph; that is the source of every breaking change in the PEP 695 work and cannot express what ess.reduce.streaming.StreamProcessor and the essapps architecture need. The proposal makes the partition of a flat pipeline at named keys the primitive (Stage: static part held, inputs per call) and builds the replacement for map/reduce on it (Fold: member table, cut keys with combine functions, contribute/combine/finalize), with StreamProcessor, the essapps warm workflow, and split workflows as the same objects. The prototype runs against main and reproduces the LoKI multi-run reduction identically against the existing with_sample_runs. Related: #235, #241, #233.

…embers The first draft kept as_pipeline() so that esssans's with_* helpers could keep returning a pipeline. That bridge was the only way two folds could compose, fixed the member table at construction, and hid a loop inside a provider. It is dropped; a drop-in pipeline is a non-goal. Stage is unchanged in kind: it passes through an output that is an input, exposes the keys it reads, and gets warm(), which computes the static parts of several stages of one pipeline in one run so that shared static work, such as a mask file every group reads, is done once. Between stages sit connectors with push, value, and clear: reducers where members are combined, a Forwarder where a context is held. Drivers are loops over stages and connectors: Fold for the table-fold shape, with groups of member keys, settable members and parameters, contributions held per member so that adding a run costs one contribution, and the three entry points contribute, combine, finalize; StreamProcessor for streams. A generic network object and a boundary builder were considered and deferred, with the reasons recorded. The LoKI validation uses one Fold with two groups and the pixel masks as a list parameter, since their cut sits inside the per-run work. Results are identical to the reference and provider call counts are equal. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

Fold had grown three jobs: orchestration, a cache-invalidation engine (a __setitem__ deciding per stage what to drop, set_members keeping held values by a row-equality heuristic, a held member frontier), and a member table with groups. The last two reproduced the map/reduced pipeline of esssans, and they were the part that was hard to reason about. Simon asked for parameters on the pipeline and Fold as stage orchestration only. Fold is now an immutable snapshot: contribute stage, finalize stage (optional), combine functions, the three entry points, and compute over a table that holds nothing. compute_members is a function. Groups are gone; several folds sharing a finalize are composed by the caller, and the "set runs, set parameters, compute" experience lives in a package object. The LoKI validation shows that object, SansReduction, in about thirty lines, with results identical to the reference and equal call counts. The doc records why the earlier Fold was wrong and where a network object would come from if the package objects repeat. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

Condenses the design document and its three prototype passes into a proposed decision: Stage and warm in sciline, connectors and a stateless Fold in ess.reduce, package objects owning topology and state, no drop-in replacement for the map/reduced pipeline. Lists the alternatives tried and dropped along the way. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

Where the branch, prototype, environment, and companion notes are; the design in ten lines; how three passes led here and why the first two were rejected; open questions with recommendations; a staging to evaluate that lands Stage and warm additively before removing map/reduce; per-package migration sites; things to verify in review. Working document, to be dropped before the branch merges. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

Fold and compute_members move from ess.reduce to sciline: they need nothing beyond Stage, hold no policy, and are the documented replacement for map(...).reduce(...) that users outside ESS need. Connectors stay in ess.reduce as a push/value/clear convention without a base class, since nothing in sciline consumes one. ADR corrections from the review against the design doc, prototype, and source: the PEP 695 PRs are closed prototypes, not landed work; groupby never existed; constraints= is removed by the generics, not by this change; Fold holds its stages' frontiers; only esssans, essreflectometry and bifrost fold runs in library code; BatchProcessor lives in essreflectometry; the LoKI timing carries its dask caveat. Terms (frontier, cut, member, contribution, driver, package object) are defined at first use. Added the coexistence alternative and the missing negative consequences: silently static cut keys, held frontiers and contributions, graph walks per parameter change, lost mapped-graph visualization, progress reporting, networkx as a direct dependency. Open questions are listed under Decision. Design doc: test count is 238, not 335; two prototype tests fail on the PEP 695 branch, both inside map; DREAM's mask fold is static, not nested; esssans folds runs in four map/reduce pairs; bifrost has three folds; migration recommends shipping the additive part in a minor release so packages migrate one at a time.

…tion Decisions from the review discussion, applied to the ADR, the design doc, the handoff note, and the prototype: - The combine per accumulation key is an accumulator, not an n-ary function. Sciline defines a structural Accumulator protocol (push, value) and Buffered(func), a factory wrapping an n-ary function. Aggregation takes a factory per key, makes fresh accumulators per combine, and compute(table) pushes each contribution as it is made, so peak memory is the accumulator's choice. The ess.reduce accumulators satisfy the protocol once maybe_hist leaves their base class; clear stays an ess.reduce convention. - Parallelism over members is the caller's job; contribute is a plain call to map over rows. The guide will show an example. - Names: Fold becomes Aggregation, cut key becomes accumulation key, at= becomes accumulators=, fold.cut becomes accumulation_keys. fold clashes with scipp's reshaping fold; the contribute/combine/finalize shape is what Spark, Flink, Beam, and pandas call aggregation, and there the stateful object is the accumulator. - No experimental label; the additive minor release with esssans migrated is the trial period. - Member tables are Mapping[label, Mapping[Key, value]] only. Prototype: drops the redundant single-sink trick in _compute (both schedulers keep requested keys), rejects an empty member table, and adds tests for Buffered, custom accumulator classes, streaming push order, and snapshot semantics under grafting. LoKI validation is identical to the map/reduce reference with equal call counts.

The prototype from docs/developer/architecture-and-design/stage-prototype becomes src/sciline/stage.py (Stage, warm) and src/sciline/aggregation.py (Accumulator, Buffered, Aggregation, compute_members), exported from the package and listed in the API reference. This is the additive part of ADR 0003; map/reduce are untouched. Changes from the prototype: the concrete graph comes from to_task_graph instead of TaskGraph._graph; the scheduler default is shared with TaskGraph through scheduler_or_default, so stages use dask when it is installed; outputs not in the pipeline are rejected at build time; NumPy-style docstrings; mypy strict clean. Tests move to tests/stage_test.py and tests/aggregation_test.py, with additions for scheduler handling, snapshot semantics, and warm skipping warm stages. The LoKI validation script moves next to the design doc, imports from sciline, and runs both sides on the naive scheduler so the timing compares call structure only; results remain identical. The design doc gains a rollout plan: PR-level order across sciline, ess.reduce, the reduction packages, esslivedata, and essapps, with the file-level sites from the survey and the scheduling dependencies.

keys is what a caller tests to decide whether a parameter change touches a stage. With an intermediate result as input, the input's ancestors are cut off and a change to them does not reach the stage, so they must not be in keys. keys is now the union of the held part and the per-call part.

A user-guide page next to the parameter-tables page, using one example that grows through the sections: a stage and what it holds, warming several stages, an aggregation with Buffered, a running-sum accumulator, the three steps called separately and a member dropped by recombining held contributions, members contributed in parallel from a thread pool, compute_members, an accumulation key that turns out static, and a table with two member keys. The page does not claim to replace parameter tables; ADR 0003 is not accepted yet.

The ADR, the design document, and its rollout plan carry everything the note held.

A type: ignore that only a newer mypy needs, and an Any escaping from the validation script where ess.sans is not installed.

Clearing inside the aggregation would either forbid combining in batches over several calls or silently add to stale state, and holding instances would make combine non-reentrant. With factories, whoever asks for accumulators owns their lifetime.

The dask.distributed workers cannot import the test module, so keys and providers defined there fail to unpickle and the workers die; on CI the scheduler kept retrying and the dask job never finished. Builtin keys and local providers pickle by value, as in the existing distributed test.

Buffered holds every pushed value, which is the wrong default for sums of large arrays. Running-total accumulators were hand-written in the tests and the user guide; Reduced(func) replaces them. It never updates in place, so it cannot mutate values owned by the caller, and requires an associative function so that combining in groups gives the same result. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

…ge.static Aggregation.combine takes combined values as well as contributions, and the chained test relies on it, but neither the Accumulator protocol nor Buffered said what that requires: a pushable value and a result that does not depend on grouping. The docstrings and the design document now say so. Stage computed its held part on first use without a guard, while the design document recommends mapping contribute over rows with threads; two cold calls would compute the frontier twice. A lock makes it once. Design document: the essapps paragraph described the chained series as held accumulators fed per arrival, which is the fold; the chained series is combine([previous, new]) per throwaway process. The D13 declaration of finalize parameters stays in the essapps spec, checked against the graph, since the backend validates without importing workflow code. The evidence paragraph still described a prototype reading TaskGraph._graph with a hardcoded scheduler. Rollout item F updated to match stages.md. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

The sciline proposal settled its open choices: Fold is Aggregation, the combine protocol is an accumulator factory per accumulation key with Buffered for n-ary functions, Aggregation lives in sciline, the member table is a plain mapping, and parallelism over members is the caller's. The note follows those, revises one judgment (keep the D13 declaration of finalize parameters and check it against the graph, for the reason D8 keeps cheap parameters declared), adds the closure condition that chaining puts on accumulators, and lists what should go back to the sciline proposal. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

Sciline's proposal (scipp/sciline#245) calls the key at which per-member values are combined an accumulation key; the sketch used accumulation point for the same thing. One word across sciline, ess.reduce, and this framework. The stages note now records "fold" as kept for the process shape, and the review log gets the eleventh pass. Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>

A stage holds only the values at its frontier, not the whole static part. warm uses the scheduler of the first stage that is not yet warm. Aggregation also raises if an accumulation key or an output is unknown. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Context now shows what map/reduce does and states the reasons for removal without the mechanics of the failed prototypes. Decision introduces Stage and Aggregation with examples and a diagram, defines terms before use, and separates what sciline provides from what callers own. Semantic detail is left to the docstrings and the design document. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

The document had grown through several rounds of design and prototyping. It is now organised as survey, overview with a glossary, the three building blocks, composition patterns for each current use, design choices, validation, migration, and open questions. Motivation is left to ADR 0003. Details of the generics prototype and superseded naming history are dropped; the rollout plan is kept with its file references. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Phase 3 in essapps is interactive applications, of which splitting a workflow is one of three models; say so, and explain spec, record, and binding for readers outside essapps. Replace the per-item rollout plan and its file and line references with what changes per project and the order of releases. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

jl-wynen · 2026-09-14T11:17:58Z

+- `Pipeline` (called v2 below, whether it ships as a namespace or a major release; see Migration): the flat graph from type hints, with PEP 695 generics, without `map`, `reduce`, `groupby`, `constraints`, or cyclebane.
+- `Stage`, in sciline: the part of a pipeline from a set of input keys to a set of output keys, with everything that does not depend on the inputs computed once and the inputs supplied per call. This is what `StreamProcessor` builds by hand today, what scipp/sciline#241 asks for, and what the essapps warm workflow (D8) and split workflows (phase 3) are.
+- `Accumulator`, in sciline: the structural protocol for what sits between stages, `push` a value and read the combined `value`. `Buffered(func)` is a factory for an accumulator that holds all pushed values and applies an n-ary function on `value`; `Reduced(func)` is a factory for one that holds only a running result of an associative binary function. The accumulators of `ess.reduce.streaming` satisfy the protocol. `Forwarder`, which holds the latest value pushed, is the connector for a context between stages and stays in ess.reduce.
+- `Aggregation`, in sciline: the table-fold shape as two stages of one pipeline, contribute and finalize, with an accumulator per accumulation key between them and the three entry points exposed. It holds nothing but its stages; parameters are set on the pipeline, and whoever loops over members owns the contributions.


I don't understand this paragraph. What is a 'table-fold shape'? What are the three entry points?

It referred to the table we map over and then reduce, now named Aggregation but Fold was the working class name and not cleaned up here. This term is avoided in the new version.

The three entry points meant the three functions on the Aggregation interface that allow running it "by hand", the contribute stage, the combine accumulation callable, the finalize stage. There is a drawing of this in the new version.

jl-wynen · 2026-09-14T13:43:28Z

+  `Reduced(func)` is a factory for an accumulator that holds only a running result of an associative binary function.
+  The accumulators of `ess.reduce.streaming` satisfy the protocol.
+  `Forwarder`, which holds the value of the latest push, is the connector for a context between stages and stays in ess.reduce.
+- `Aggregation`, in sciline: the table-fold shape as two stages of one pipeline, contribute and finalize, with one accumulator per accumulation key between them and with the three entry points exposed.


What is a 'table-fold shape'? And which three entry points?

jl-wynen · 2026-09-14T13:44:45Z

+- `StreamProcessor`, in ess.reduce, as a loop over stages and connectors that carries only its policy.
+
+There is no drop-in replacement for the pipeline that `with_sample_runs` and the related `with_*` helpers return today.
+Each package returns its own object, about thirty lines over `Aggregation`, which holds the pipeline, the aggregations, the contributions, and the "set runs, set parameters, compute" experience.


What does 'about thirty lines over Aggregation' mean?

jl-wynen · 2026-09-14T13:45:36Z

+
+`map` relabels every reachable node at call time, before the targets are known.
+The demand-driven generics work paid for that twice.
+First with forward chaining and its seeding, which the Q1 spike removed with the deferral of the labeling into cyclebane (scipp/cyclebane#32, unreleased, which is why the CI of the branch is red).


I still have no clue what a 'Q1 spike' is supposed to be.

It does not belong here and is removed in the latest version. Q1 referred to a question/option in a design document that led to the two generics prototypes (both PRs closed), the "spike" was the quick experimental implementation of that:

Q1: Forward chaining, or deferred mapped-labeling in cyclebane?

jl-wynen · 2026-09-14T13:51:50Z

+
+## Problem
+
+### In sciline


For future reference, I think it needs to be clear that this is not a problem with Sciline but rather a problem with the proposed changes. This might currently read as through the released version of Sciline has these problems (which are implementation details)

jl-wynen · 2026-09-14T14:39:16Z

+In both cases `combine` takes combined values as well as contributions, and thus the combine can proceed in groups or as a chain.
+That asks two things of an accumulator: `push` must accept what `value` returns, and the result must not depend on the groups of the pushes.
+`Buffered` has both if its function is associative, `Reduced` requires it, and an accumulator of ess.reduce has them if `push` accepts what `value` returns, which the histogramming ones must keep once `maybe_hist` moves out of the base class.
+For concat-like combines the two cost about the same, twice the total, and `Buffered` is the correct choice.


I don't understand. What 'two' and what 'total'? And how can total be less than its constituents?

I think this paragraph is too long and uses too many relative expressions that make it hard to understand what each sentence refers to. Do we even need this here? It does not seem important to the design itself.

jl-wynen · 2026-09-14T14:41:45Z

+Whether the values arrive from one input over time or from many members is the picture of the driver.
+Lifetime (`clear`), identity (contributions by label), and the dependence on the order belong to the driver.
+
+`Stage` itself holds nothing but its frontier, and even that frontier is a forwarder from a stage without inputs, kept inside because it is the common case.


I thought it holds the graph from its inputs to outputs?

jl-wynen · 2026-09-14T14:43:15Z

+
+`Stage` itself holds nothing but its frontier, and even that frontier is a forwarder from a stage without inputs, kept inside because it is the common case.
+Every held value is an object that the driver can inspect, clear, serialize, or place on a process boundary, which is the explicit lifetime that scipp/sciline#241 asks for.
+A value leaves one driver and enters another as a parameter of the flat pipeline: an aggregation over banks whose result feeds a `StreamProcessor` is `pipeline[EmptyDetector] = banks.compute(table)[EmptyDetector]` before the processor is built.


'driver' has not been defined. How can values move between drivers? I would understand a driver as a single piece of code that runs several stages.

jl-wynen · 2026-09-14T14:57:40Z

+
+Semantics:
+
+- An aggregation is two stages of one flat pipeline with one accumulator per accumulation key between them: contribute from the member keys to the accumulation keys, and finalize from the accumulation keys to the outputs.


How do Aggregations compose if you have more than stages?

Currently I did not foresee composing aggregations. I someone needs a multi-level aggregation sth. should likely built for that purpose, similar to how StreamProcessor serves a specific purpose.

jl-wynen · 2026-09-14T15:00:22Z

I read about half of this document and frankly, the writing is terrible. The language is obtuse and convoluted and the content is poorly structured, mixing motivations, detailed descriptions of implementation, alternatives, etc. I feel like I need to have read the entire PR and some extra documents in several repos before I can understand it.

I agree, it was still a working document, I apologize for not making that clear.

The ADR again says that parameter values are held by reference, that the StreamProcessor rewrite is not done, and why pinning is offered instead of a separate namespace. The design document again states the obligation on histogramming accumulators, the requirements Stage places on the new Pipeline, the driver's ownership of order dependence, why Accumulator and Forwarder live where they do, and what StreamProcessor.visualize needs. The survey findings that matter for the migration and the dependencies in the release order are back in short form. The Stage docstring states the by-reference behaviour too. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Stage is mostly introspection (inputs, outputs, keys, frontier, dynamic), so a call on the object hid the one action among its members, and call sites such as `agg.contribute_stage(row)` read like a getter. `compute` matches `Aggregation.compute` and `Pipeline.compute`. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Reading the static property computed the held part on first access, which runs providers and may read files; a method makes that visible. It delegates to warm, which now takes each stage's lock, so the held part is computed once also when warm and compute run in different threads. Locks are taken in a fixed order to avoid deadlocks between overlapping warm calls. Buffered's docstring now says that the function is applied on every read of value. Also wrap test lines that exceeded the line length after the rename to compute. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

SimonHeybrock and others added 11 commits September 10, 2026 19:44

Drop the session handoff note

77153ac

The ADR, the design document, and its rollout plan carry everything the note held.

This was referenced Sep 11, 2026

Tracking: replace map/reduce with stages and aggregations (ADR 0003) #246

Open

Tracking: migrate from sciline map/reduce to stages and aggregations (sciline ADR 0003) scipp/ess#746

Open

SimonHeybrock and others added 5 commits September 11, 2026 10:43

Fix mypy under the pinned version

8e0f2da

A type: ignore that only a newer mypy needs, and an Any escaping from the validation script where ess.sans is not installed.

SimonHeybrock and others added 4 commits September 14, 2026 14:32

SimonHeybrock force-pushed the map-reduce-outside-the-graph branch from a1e56ed to 26d2f33 Compare September 14, 2026 14:47

jl-wynen reviewed Sep 14, 2026

View reviewed changes

SimonHeybrock and others added 4 commits September 15, 2026 06:14

Remove confusing reference

ad3f97d


		Semantics:

		- An aggregation is two stages of one flat pipeline with one accumulator per accumulation key between them: contribute from the member keys to the accumulation keys, and finalize from the accumulation keys to the outputs.

Uh oh!

Conversation

SimonHeybrock commented Sep 11, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Why

What is in the PR

Not in the PR

Test plan

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SimonHeybrock Sep 15, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

SimonHeybrock commented Sep 11, 2026 •

edited

Loading

SimonHeybrock Sep 15, 2026 •

edited

Loading