Proving dataflow operators coordination-free in Lean 4

Ryle · Jul 16, 2026

Distributed query engines spend most of their coordination on barriers, the points where every partition has to finish before anything downstream can run. Most engines decide where those go with a hand-maintained table. Spark calls map and filter narrow and groupByKey, join and distinct wide, and engine authors extend that list one operator at a time. In a dataflow engine I've been writing as a side project, I wanted the classification to come from a property of each operator instead, and I wanted that property machine-checked for the leaf operators. This is how far I got with that, including the parts that are only tested or still just a design.

CALM in one paragraph

The CALM theorem (Hellerstein and Alvaro, "Keeping CALM", CACM 2020) says a computation has a consistent, coordination-free distributed implementation exactly when it's monotone. Monotone here means the output only grows as the input grows. A new row can add to the answer but can never take back something that was already emitted. With a non-monotone operator one more row can change the answer, so it has to wait until the input is complete.

For a planner this turns into a rule: an operator needs a barrier over a partitioning when it's non-monotone over that partitioning, and everything upstream of the first such operator can stream. Watermarks follow the same rule, since a watermark is basically the signal "input is complete up to time T". The non-monotone list is what you'd expect: sort, top-K, negation, distinct-count finalize, and any final aggregate that depends on complete input. The more interesting cases are the ones that look wide but are monotone in the right lattice, like a running max, a grow-only sum, or transitive closure.

Two corollaries of one lattice

I organized the Lean development around two theorems. They share a setting, but they aren't the same proof.

The first one is about sharded reduction. If a reducer is a fold with an associative, commutative combine, then reducing each shard and merging the partials with that same combine gives the same result as reducing the whole batch. That's what lets an executor split a group across partitions. In Lean it's a short lemma over lists, which I instantiate per aggregate:

theorem reduce_append (combine : β → β → β) (e : β) (g : γ → β)
    (assoc : ∀ a b c, combine (combine a b) c = combine a (combine b c))
    (comm : ∀ a b, combine a b = combine b a)
    (left_id : ∀ a, combine e a = a) (xs ys : List γ) :
    reduce combine e g (xs ++ ys)
      = combine (reduce combine e g xs) (reduce combine e g ys)

The second is about iteration. A monotone, deflationary map on a complete lattice of finite height, iterated from a start state, stabilizes at the greatest fixed point below the start. I bundled the map, its monotonicity proof and the deflation invariant into one structure and proved once that iterating it equals Mathlib's OrderHom.prevFixed. After that, an iterative algorithm gets convergence by supplying the operator and a finite-height instance. The link between the two theorems is that the combine in the first one is the same lattice meet that drives the second.

The aggregation kernels

The leaf operators are integer SUM, MIN, MAX, COUNT and AVG. I pulled the per-group fold out of the engine into a tiny no_std Rust crate with no dependencies, because Charon translates every reachable item and each dependency makes the generated Lean bigger. The functions just take a slice and return a scalar:

pub fn fold_sum(values: &[i64]) -> i64 {
    let mut acc: i64 = 0;
    let mut i: usize = 0;
    while i < values.len() {
        acc += values[i];
        i += 1;
    }
    acc
}

Charon compiles that to an intermediate representation, and Aeneas (Ho and Protzenko, ICFP 2022) translates it into a pure Lean model. I check the generated Lean in, and a script re-runs extraction and diffs the output, so if a kernel body gets edited the proofs can't keep passing against stale Lean.

Each kernel gets a loop spec (the extracted loop computes a left fold, proved with Aeneas's weakest-precondition loop rule) and a merge theorem (the fold over xs ++ ys equals the merge of the two folds). SUM needed one more. Aeneas models += as two's-complement wrapping addition, so I could only prove the fold equals the mathematical integer sum under a bound that every prefix sum stays inside i64. The merge law holds unconditionally though, even under overflow, because wrapping addition is still associative and commutative. The production reducer uses checked addition and errors on overflow, which is the side of the bound it needs.

AVG turned out to be the most useful case, because it shows where the barrier actually sits. Averaging the shard averages gives the wrong answer. The (sum, count) pair is a monoid merged componentwise, though, and the division is a separate step. Lean proves the pair merges and gives a counterexample for the division: shards (10, 1), (0, 1) and (10, 1), (0, 9) have the same per-shard averages of 10 and 0 but combine to 5 and 1. So the accumulation is monotone and the division is where the boundary goes.

DISTINCT works the same way with sets. A grow-only set merged by union is a join-semilattice, and Lean proves the merged set equals the whole-batch set and absorbs duplicates. Its cardinality doesn't add across overlapping shards (two shards of [1] have a combined distinct count of 1), and Lean proves counts do add when shards are disjoint. Put together, that's exactly why distinct finalization needs a barrier and the set accumulation doesn't.

The graph kernel

The end-to-end case I did is connected components by label propagation. Every vertex starts labeled with its own id, and each sweep sets both endpoints of every edge to the smaller label, in place over a u64 slice. I proved one sweep is monotone and deflationary on Fin n → ℕ∞, which gets convergence from the shared structure, and then proved the converged labeling correct against Mathlib's definition of reachability:

theorem cc_eq_iff_reachable (edges : List (Fin n × Fin n)) (u v : Fin n) :
    cc edges u = cc edges v ↔ (graphOf edges).Reachable u v

The generic convergence theorem only says that some number of sweeps is enough, which doesn't help a caller pick a loop bound. So a follow-up proves n sweeps are enough for n vertices, since a label moves at least one hop per sweep and any path has fewer than n edges. Refinement proofs connect the extracted Rust initializer, sweep and driver to the pure model, so the composed theorem says the Rust code with a budget of at least n returns the connected components. The caller passes the slice length, which is n.

Using it in the planner

A planner pass walks the plan's stages and tags each operator. SUM, MIN, MAX and COUNT get tagged monotone because they're the verified leaves. AVG and COUNT DISTINCT get tagged non-monotone because they emit the finalized value, and anything the pass can't model is also non-monotone. I think that fail-closed default is the most important property of the pass, because a wrong classification can only add a barrier.

Planning records each stage's tag and the index of the first non-monotone stage, which is where a distributed executor would insert an exchange. Right now the engine runs in-memory batches on one node, so the boundary gets recorded and doesn't do anything yet. Streaming the monotone prefix across machines has to wait until the engine has distributed execution.

Tying production code to the proofs

The proofs cover the extracted kernel crate. On the graph side, production code calls that kernel directly. The engine's aggregate reducers are separate code, and the link to them is tests. One runs the engine's SUM, MIN and MAX over a fixed seven-value input and compares against the kernel, and another splits input across two record batches and compares against the kernel on the concatenation. Planner binding tests check that each verified leaf is tagged monotone and replay the AVG and DISTINCT counterexamples.

Those are spot checks on concrete inputs. They catch semantic drift in a production reducer and they document the link from a planner tag back to a theorem, but equivalence of the engine reducer and the kernel is unproven, and so is the classifier. There's also a test that doesn't need the toolchain and checks every public kernel function has a counterpart in the generated Lean, which catches a missed extraction by name. The extraction diff covers the bodies, but it needs Charon and Aeneas installed.

Column provenance for tenant isolation

I also implemented a second planner pass that propagates a provenance tag per column: PUBLIC, PRIVATE(tenant), or REDUCED(op, side conditions). A value derived from a private column can only leave the plan as REDUCED with its side conditions met. The read path runs the structural check on multi-owner reads and applies the group-size check at emit time.

The Lean backing only covers one condition, stride. For tumbling windows, Lean proves any two raw series whose windows agree in sum produce the same release, and for a two-value window it proves infinitely many raw windows map to each released sum. With overlapping windows, consecutive sums can be differenced back to the raw values, so the guard rejects them. The general argument for wider windows is still open, and so is anything that depends on realized group sizes. The guard fails closed on those. A private-derived aggregate starts with an unestablished group-size floor and gets refused until the emit path records a realized size of at least two. Private MIN and MAX are refused outright, since they reveal an exact value.

What is proven and what is open

I scanned both Lean projects for sorry, admit and axiom and found none, so everything I list as proven below is a closed theorem with no escape hatch. A CI job is set up to build both projects against pinned Lean and Mathlib versions and re-run the extraction diff.

The main thing I learned was where to stop. Proving the leaves was doable because I cut the kernels down to slices and scalars before extracting them, which kept the generated Lean small enough to reason about. The price is that the link back to the engine's reducers is only tests. I also found the counterexample theorems carried as much weight as the positive ones. "AVG's division does not distribute over shards" is a one-line norm_num proof, and it's the formal reason for a barrier the planner would otherwise place on intuition. Next I want to do the accumulate/finalize split for AVG, so the accumulate stage can be tagged monotone, and work out the reconstruction argument for wider windows.