ToySims

Differential Dataflow: Incremental Views for Upserts

Explore signed changes, incremental joins, antichain frontiers, compaction and replay through an advanced interactive guide with a Pinot upsert design lab.

Lesson summary

Maintain exact contributions, track when their answers are complete, and forget only the history readers have released.

Row visibility does not retract the old contribution from a separately maintained result.
The previous winning full row and the source version policy; stale or duplicate arrivals must not emit another correction.
Equal records at equal times can consolidate immediately. Cross-time cancellation is valid for an eligible snapshot, or after safe logical compaction.
It must remove exactly what the old row contributed; the new row may satisfy a different predicate.
Maintain SUM and COUNT, update both with signed contributions, then divide when the count is nonzero.
If its last supporting copy is removed, retained alternatives are needed to choose the replacement.
Join changed records with retained matching records on the other side, multiplying their weights.
It cancels mixed old/new contributions and includes new/new contributions exactly once.
No. It avoids unrelated keys, but every genuine matching result may still need an update.
Ordinary key exchange sends equal keys to one owner. Splitting that work needs another algorithm.
No. It defines the snapshot; progress evidence establishes that no more relevant updates can arrive.
Their least upper bound is (2,3); neither original pair dominates the other.
When no element f of F satisfies f ≤ t. Equality is still potentially unfinished.
An in-flight message or held operator capability can still produce work at 8.
Yes, if the newly learned correction enters at a permitted later logical epoch while the business date remains payload.
Another derivation through D still supports C. The output retracts only when the new fixed point loses it.
All readers of the trace must release the historical queries that need them; input completion alone is insufficient.
It merges batches and changes access structure; logical compaction can coarsen timestamps and release old answers.
Replaying an additive correction twice corrupts a sink unless effects and progress are committed or deduplicated coherently.
Before-image corrections, complete publication, operator state, replay discipline, and explicit retention and resource policies.

Why can the latest row be correct while a stored total is wrong?

Row visibility does not retract the old contribution from a separately maintained result.

What must a normalizer know before retracting an upsert?

The previous winning full row and the source version policy; stale or duplicate arrivals must not emit another correction.

When can opposite weights cancel?

Equal records at equal times can consolidate immediately. Cross-time cancellation is valid for an eligible snapshot, or after safe logical compaction.

Why does filtering a retraction use the old fields?

It must remove exactly what the old row contributed; the new row may satisfy a different predicate.

What state makes AVG maintainable under deletes?

Maintain SUM and COUNT, update both with signed contributions, then divide when the count is nonzero.

Why is the current MIN insufficient state?

If its last supporting copy is removed, retained alternatives are needed to choose the replacement.

How is a one-sided join delta computed?

Join changed records with retained matching records on the other side, multiplying their weights.

Why is the delta-delta join needed when both inputs change?

It cancels mixed old/new contributions and includes new/new contributions exactly once.

Does an arrangement eliminate large correction fanout?

No. It avoids unrelated keys, but every genuine matching result may still need an update.

Why can a hot key remain a bottleneck after adding workers?

Ordinary key exchange sends equal keys to one owner. Splitting that work needs another algorithm.

Does a logical timestamp prove the corresponding result is complete?

No. It defines the snapshot; progress evidence establishes that no more relevant updates can arrive.

What is the join of times (2,1) and (1,3)?

Their least upper bound is (2,3); neither original pair dominates the other.

When is time t complete for frontier F?

When no element f of F satisfies f ≤ t. Equality is still potentially unfinished.

Why can a source at 9 leave time 8 unfinished?

An in-flight message or held operator capability can still produce work at 8.

Can a late business event revise an old day after logical progress?

Yes, if the newly learned correction enters at a permitted later logical epoch while the business date remains payload.

Why can C stay reachable after deleting B → C?

Another derivation through D still supports C. The output retracts only when the new fixed point loses it.

Who permits historical distinctions to be forgotten?

All readers of the trace must release the historical queries that need them; input completion alone is insufficient.

How does physical compaction differ from logical compaction?

It merges batches and changes access structure; logical compaction can coarsen timestamps and release old answers.

Why does a correct delta need a recovery protocol?

Replaying an additive correction twice corrupts a sink unless effects and progress are committed or deduplicated coherently.

What does a Pinot-oriented maintained view add beyond winner selection?

Before-image corrections, complete publication, operator state, replay discipline, and explicit retention and resource policies.

Stage 1 of 22

One corrected order, many derived consequences

An order moved from Mumbai to Pune, and its amount changed from 500 to 700. The latest row is fixed. Somewhere downstream, an old result still contains it. How do we find and undo that contribution without rerunning everything?

We will use a delta, meaning a signed change to a collection; a trace, meaning indexed change history; and logical time, meaning a version of the computation. You can assume ordinary SQL and distributed ingestion. We will build the less familiar algebra and progress rules explicitly.

First turn replacements into corrections. Then propagate corrections through operators and joins. Finally make time, compaction, and recovery precise enough to design an upsert-aware maintained view for Pinot. All order values and timings in the diagrams are an executable teaching fixture, not measurements.

Stage 2 of 22

The row is correct. Why is the total wrong?

Order o7 contributes 500 rupees to Mumbai. Another order, o8, contributes 300. Our stored city total is therefore 800. Assume you already know SQL, joins, and partitioned ingestion; the new problem is maintaining their results as inputs change.

A correction replaces o7 with 700 rupees in Pune. The primary-key lookup now returns the right row. But our separately maintained city totals still contain its old contribution.

A fresh GROUP BY over the latest rows gives Mumbai 300 and Pune 700. The stored result still says Mumbai 800. Updating row visibility and maintaining a derived result are different operations.

We could rerun the query after every correction. Instead, we want the change to tell us which existing contributions must disappear and which new ones must appear. Start at the boundary that turns a replacement into those two facts.

Stage 3 of 22

What information does a replacement need?

The input says what o7 should become. To retract what it used to contribute, the normalizer must recover its previous full row. A primary key alone does not tell us its old city or amount.

Compare source versions before emitting anything. Our fixture uses strictly increasing versions per key: version 8 supersedes version 7. A duplicate or older version produces no change. Equal-version conflicts need an explicit production policy; this toy treats equality as a duplicate.

For an accepted replacement, emit the old row with weight minus one and the new row with weight plus one, both at logical epoch 8. A delete emits only the old negative row and keeps enough version information to reject stale resurrection.

Use Incoming version to compare a real correction, a duplicate, and a stale arrival. Inspect both the winner and the emitted records.

The version check decides which row wins. The negative and positive weights describe what changed after that decision. Those are separate semantics; differential dataflow does not infer your upsert winner rule.

Stage 4 of 22

A negative row is an instruction, not bad data

A collection is a multiset: a record has an integer multiplicity. Weight plus two means two copies of that exact record. Weight minus one removes one copy when the changes are accumulated.

Here the old o7 row arrives with plus one at epoch 7. Its retraction carries minus one at epoch 8. They have identical row contents, but their times differ. Keep that distinction if readers still need the epoch-7 result.

At epoch 8, accumulate the weights at times up to 8. The old row sums to zero and disappears; the new row sums to one. We can reconstruct a snapshot from a history of differences.

Consolidation adds weights only for equal records and equal logical times. Snapshot reconstruction also accumulates across eligible times. Do not erase the time dimension just because two rows would cancel in the latest result.

Stage 5 of 22

Can the query run on changes alone?

Take the signed correction from o7. A projection that keeps city and amount maps both records and preserves their weights. If several inputs project to the same record, their weights add.

Now add a filter: amount at least 600. Evaluate the predicate on each full record, including the negative old one. The old 500 row never contributed to this filtered view, so it needs no retraction there.

The filtered delta contains only plus one of Pune 700. Map and filter distribute over collection addition, so they can transform the changes without scanning the unchanged input.

Switch Predicate between All rows and At least 600. Compare the surviving signed rows. The filter applies to the removed record’s old fields, not the replacement’s fields.

This convenient rule depends on a fixed, deterministic transformation. A lookup into changing external state is a join in disguise; treating it as a timeless map loses the changes caused by that lookup.

Stage 6 of 22

What state makes an average retractable?

Return to all orders. Mumbai starts with sum 800 and count 2. Store those sufficient statistics instead of storing only the average, which is 400.

Retract o7: subtract 500 from the sum and one from the count. Mumbai becomes sum 300, count 1. Insert its replacement into Pune: add 700 and one.

Recompute each average from its maintained sum and count: Mumbai 300 and Pune 700. This is exact for our integer fixture. Real implementations must define null, overflow, decimal, and floating-point behavior.

An average can be maintained because we retained more information than the average itself. That raises the harder question: what must we retain for an operation such as minimum, where subtraction has no inverse?

Stage 7 of 22

What replaces the minimum when it disappears?

Use a smaller fixture for this boundary: one group contains 300, 300, and 500. Its minimum is 300. Its distinct values are 300 and 500. Neither output tells us that 300 has two supporting copies.

Remove one 300. Its support count falls from two to one. The minimum and the distinct set stay unchanged, even though the input has changed.

Remove the last 300 and the support count reaches zero. Retract 300 from DISTINCT; for MIN, retract output 300 and insert output 500. We needed retained alternatives to discover the replacement.

Use Remove copies to compare one deletion with both. Follow the support count, remaining multiset, and minimum together.

A nonlinear operator can still emit signed output changes. It usually needs indexed state and more work per affected key. Signed weights are a representation of change, not a promise that every operator reduces to arithmetic on one counter.

Stage 8 of 22

A change must meet the other side’s state

Introduce a customer relation: c1 currently has tier Silver. Join orders to customers on customer ID. Before the correction, o7 contributes the joined row Mumbai, 500, Silver.

Probe the customer relation with the negative old order. Its match emits a negative Mumbai, 500, Silver row. Probe with the positive new order and emit positive Pune, 700, Silver.

With customers unchanged, the join delta is delta-orders joined with customers. Each output weight is the product of the two input weights. A matching row of multiplicity two would double both contributions.

Retaining the customer state lets the order change find its matches. But customer records also change. If both relations change in one epoch, two independent one-sided calculations are not enough.

Stage 9 of 22

What if both sides of the join change?

At epoch 8, o7 changes from 500 to 700 and customer c1 changes from Silver to Gold. The column headings abbreviate the joined records; 500 means the old Mumbai order and 700 means its Pune replacement.

First join the order delta with the old customer relation. This retracts 500 Silver and inserts 700 Silver. The first row records those signed contributions, not the final answer.

Next join the old order relation with the customer delta. This also retracts 500 Silver, then inserts 500 Gold. Pause here: both rows were computed against old state.

Which additional term fixes the combined delta?

Join the two deltas. Minus times minus restores one 500 Silver; the two mixed signs retract 500 Gold and 700 Silver; plus times plus inserts 700 Gold. All four cells matter.

Add the three contribution rows. The mixed records cancel, leaving exactly minus one 500 Silver and plus one 700 Gold. This is the old joined row removed and the new joined row inserted.

The cross-term cancels mixed old/new join results and produces exactly the correct replacement.

Use Cross-term to omit it, then include it. Inspect the residual weights below the matrix. Omitting it leaves an invalid negative old record and two mixed records.

This matrix is the algebra, not a requirement to materialize three intermediates. An implementation may join the order delta against new customer state, then join old orders against the customer delta. That assigns the cross-term once; doing both against new state counts it twice.

Stage 10 of 22

How do changes find only their matching records?

The algebra says which records must meet. An arrangement makes that meeting efficient: partition by a key, then retain an indexed trace of values, logical times, and weights. Here the key is customer ID.

A correction to c1’s tier seeks c1 in the order arrangement. It finds o7 and o9. It does not need to inspect the c2 and c3 groups. Both matching orders must receive a correction.

The index narrows the search, but cannot remove genuine fanout. If c1 had a million matching orders, changing its tier could require a million joined retractions and a million insertions.

Use Changed customer to probe different key groups. The highlighted rows are the actual work set, not a sampled animation.

Multiple operators can reuse a compatible arrangement. That saves duplicated indexes, while giving each reader a say in which historical distinctions must remain available. We will return to that constraint when compacting the trace.

Stage 11 of 22

What does distributing the index actually solve?

An exchange sends equal join keys to the same worker. For this toy, worker equals the numeric part of the customer ID modulo three. The eight order tiles expose every assignment.

With distinct customer keys, the work spreads across workers. A production hash differs from our tiny mapping, but the required property is the same: matching keys must meet.

Six orders with key c1 all land on worker 1. Adding workers does not split that key under ordinary hash partitioning. A hot key can dominate state and correction work.

Switch Key distribution between Distinct and Hot c1. Compare actual tile ownership and worker loads; the total stays eight.

Sharding a hot key requires a second design, such as replication or two-stage aggregation where the algebra permits it. An index avoids unrelated scans; it cannot make an intrinsically large join delta small.

Stage 12 of 22

Which version does this result describe?

So far we have used epoch numbers without asking how they define an answer. Treat a logical timestamp as a coordinate in the computation. It need not be the event’s wall-clock time or a Kafka offset.

At query time 7, include only updates whose logical time is at most 7. That includes plus one of Mumbai 500, but neither part of the correction at 8.

At query time 8, include all three updates. The old row cancels and Pune 700 remains. The negative and positive changes share a time, so the mathematical snapshot has no half-applied version.

Use Read epoch to compare 7 and 8. The inclusion marks show exactly which updates contribute to each snapshot.

A timestamp specifies which answer we mean. It does not by itself prove that every update for that answer has arrived, nor make an external sink publish atomically. To reason about completion, we first need the shape of time.

Stage 13 of 22

Time can have more than one direction

An iterative computation has two reasons to change: a new input epoch and another loop iteration. Represent its time as a pair, epoch and iteration. The grid uses epoch horizontally and iteration vertically.

Compare (2,1) with (1,3). One has the later epoch; the other has the later iteration. Under the product order, neither is at most the other. A lexicographic sort would invent an order the computation does not require.

A joined output must include both inputs. Its time is their least upper bound, here (2,3), computed by taking the maximum in each coordinate. The shaded region shows the history included at (2,3).

A collection at a pair is the sum of differences at all coordinate-wise earlier pairs. Sharing a two-dimensional history lets a new epoch reuse work at corresponding iterations. The same partial order also changes how we represent unfinished work.

Stage 14 of 22

How can one frontier contain two times?

Suppose work may still arrive at or beyond (2,4), or at or beyond (4,2). Neither point dominates the other, so the frontier retains both. Such a set of incomparable minimal times is an antichain.

Shade every time greater than or equal to either frontier element. These are times that may still change. The unshaded points cannot receive further contributions under this progress contract.

Time (3,3) is complete even though neither frontier point is globally past it in a total order. Neither (2,4) nor (4,2) is at most (3,3). The exact test is: no frontier element is less than or equal to the queried time.

Use Query time to test the middle gap, a frontier point, and a later point. Compare the selected cell with the shaded future region.

Complete means no more updates for that logical answer, assuming all participants honor their capabilities. It does not mean durable storage or transactional publication. A frontier is evidence about possible computation, not a clock reading.

Stage 15 of 22

An empty source is not a finished computation

Return to scalar epoch 8. A capability is permission to produce records at that time or a later one. Sources and operators must release or advance that permission before progress can rule out earlier work.

The source advances to 9, but an epoch-8 message is still in transit. Our tracked obligations retain 8. A source high-water mark alone would declare completion too early.

Receiving the message transfers the remaining work to the operator. That operator may produce more epoch-8 output, so the frontier still cannot pass 8.

After the operator finishes and drops its epoch-8 capability, no obligation at 8 remains. The frontier advances to 9, which proves completion through 8 in this acyclic, scalar-time fixture.

Real Timely progress accounting combines capability changes and message counts across workers, following timestamp summaries along dataflow edges. A held capability can delay completion without any buffered records. CPU idleness and empty queues are not sufficient proofs.

Stage 16 of 22

Can a completed time still accept a correction?

A business event dated 3 arrives after logical time 8 is complete. The event’s date is payload data in our design. Assign its newly learned correction logical ingestion epoch 9.

The correction at 9 may revise a result grouped by business date 3. That does not reopen logical snapshot 3: the version of our knowledge has changed at 9.

If instead you try to submit at logical time 3 after relinquishing the right to send there, the contract rejects that action. Event-time windows and allowed lateness are source and application policies, not something the dataflow can guess.

Use Timestamp policy to contrast a new logical epoch with reusing the old business time. The destination slot shows whether the update is legal.

A late arrival is not automatically a newer upsert version. First apply the source’s winner rule, then timestamp the resulting accepted change. This separates business ordering, delivery order, and computational progress.

Stage 17 of 22

A retraction can travel more than one hop

Use a small acyclic reachability graph to expose iteration. From seed A, one route reaches C through B and another through D. Each round extends the prior reached set by one hop, then applies DISTINCT.

Remove edge B to C in a new input epoch. Compare the old and new reached sets at the same iteration, rather than comparing the new input with the old computation’s final round.

After one round, both epochs still reach A, B, and D. After two rounds, the surviving route through D still reaches C. Removing one derivation must not retract a result with another supporting path.

Remove both routes into C and the new fixed point excludes C. The output change is a retraction of C. The paired iteration histories show where the difference first appears.

Use Removed paths to compare one route with both. Watch the deleted edges and the same-round reached sets together.

This fixture computes a least fixed point on a DAG; it is not a general deletion algorithm for cyclic graphs. Cycles require correct recursive semantics, not naïve path-count subtraction. Differential dataflow’s timed differences support such computations, but arbitrary programs are not guaranteed to converge.

Stage 18 of 22

Which historical distinctions can readers release?

An indexed trace cannot keep every update time forever. Here o7 at amount 500 is inserted at time 2, retracted at 4, and replaced by amount 700 at 4. Reading time 3 still needs the old 500 row.

If every trace reader releases queries before time 5, scalar logical compaction may advance earlier update times to 5. Their contributions to queries at 5 or later are unchanged.

At the common time 5, plus and minus 500 can consolidate to zero. Only plus 700 remains. We removed historical distinctions, not the current row. Reading the old answer at 3 is no longer supported.

Use Reader frontier to keep time 3 readable or release history through 5. Compare the retained timestamped updates.

The input progress frontier and a reader’s compaction frontier answer different questions. One bounds future arrivals; the other releases historical queries. Readers of a shared trace must all permit the loss. For product time, advancement uses the lattice and frontier, not an arbitrary coordinate-wise maximum with one chosen point.

Stage 19 of 22

Fewer batches does not mean less history

Logical compaction changes which historical answers remain distinguishable. Physical compaction addresses a different cost: a growing list of immutable update batches that each lookup would otherwise need to visit.

Merge the ordered entries from the two batches into one ordered run. The same keys, values, times, and weights remain available. Equal entries may consolidate, but a batch merge need not erase timestamp distinctions.

The trace now has one batch instead of two. Its time-3 snapshot still contains 500 because we have not advanced logical compaction beyond that query. The storage organization changed; the answer did not.

Trace readers can constrain physical batch boundaries as well as logical history. A join may need a cursor through a particular boundary to assign new work exactly once. Compaction is coordinated with readers, not merely a background cleanup timer.

Stage 20 of 22

Correct deltas can still corrupt an external sink

The algebra computes the right correction once: Mumbai minus 500, Pune plus 700. Suppose a sink applies it, then the process crashes before its input checkpoint is durably recorded.

On restart the same input is replayed. If the sink blindly applies the additive correction again, Mumbai loses another 500 and Pune gains another 700. Negative weights do not deduplicate delivery.

Our transactional teaching sink records the correction’s identity with its effects. Replay recognizes the already-applied identity and leaves the totals unchanged. This models an atomic commit or equivalent idempotency protocol, not a built-in guarantee of a dataflow frontier.

Use Sink protocol to compare Blind increments with Deduplicated commit. Read the actual city totals after replay.

A production recovery design must align source positions, winner state, operator state, and sink publication. Completion tells us computation at a time is done; durability and exactly-once external effects need their own protocol.

Stage 21 of 22

Can an upsert also maintain a correct city view?

Return to o7 and o8. Pinot’s upsert machinery selects current records using primary keys, comparison values, and validity metadata. That addresses which rows queries should see. The extra system drawn here is a proposed maintained city view, not a claim that Pinot implements differential dataflow.

The proposed normalizer needs the previous winning full row before its contribution can be retracted. A patch must first become a new full winner. Retaining only the new value leaves no evidence that Mumbai should lose 500.

Even with both deltas, publishing after only the retraction exposes a partial result. Our proposed sink stages changes and publishes the complete epoch. Across shards, defining that publication boundary is additional engineering work.

Repair both city totals. Use Before image and Publication; compare each result with its oracle.

Retaining the before image and publishing the complete epoch makes the view match the oracle. If either condition is missing, inspect the affected city cell to locate the error. This is a minimal transfer exercise; the previous stages identify the join, retention, skew, and recovery requirements a real implementation must also satisfy.

A practical first experiment would maintain SUM and COUNT for a bounded set of groups outside the serving path, replay real upsert histories, and compare every completed epoch with a full recomputation. Measure correction fanout, retained state, and freshness before extending to joins or recursion. Differential dataflow supplies useful semantics; a Pinot integration still has to earn its operational guarantees.

Stage 22 of 22

A result is a maintained history of contributions