Retractions and Updating Results

A stream of window results is not a table of answers. When a window fires early and again when the watermark passes, and a third time because a late record arrived, the consumer receives three records for the same key and window. Whether that means three answers, one answer and two corrections, or one answer and two increments depends entirely on a convention the producer and consumer must agree on. A pipeline that leaves the convention implicit produces consumers that double-count, and the error grows with every late record.

The three conventions

Append mode treats every emission as a new, independent row. A consumer that sums the stream gets the sum of all emissions, which for accumulating window output is wrong: the second emission of a window already contains the first. Append mode is correct only when each emission covers disjoint events, which is the discarding-pane convention from Window Assignment and Trigger Semantics.

Replace mode, also called upsert, treats each emission as the current value for its key. A consumer that maintains a table keyed by the window’s identity overwrites the previous value, and the table’s contents after any point in the stream are the answers as known at that point. This is the natural convention for accumulating output, and it requires the record to carry a key that is stable across emissions of the same window.

Retract mode treats each emission as a pair of operations: a deletion of the previous value and an insertion of the new one. A consumer that maintains a multiset of rows applies both, and the net effect equals the replace. Retract mode is what a relational engine produces when a query’s result changes, because the change may affect a row the consumer never saw and may need to remove a row entirely rather than update it.

The distinction between replace and retract matters when the key is not available or not stable. An upsert needs the consumer to be able to identify the row; a retraction carries the entire previous row, so the consumer can remove it without knowing a key. A stream whose rows have no natural key — an aggregation grouped by nothing, a distinct over a set — can only be expressed as retractions.

What a retraction record contains

The minimum content is an operation tag and the row. The tag distinguishes an insertion from a deletion, and the row is the full value, not a delta. The row’s full value is required because a deletion must be applied to a multiset that may contain the row more than once, and matching on a key alone is ambiguous when keys are not unique.

+I  window=12:00..12:05  count=17
-I  window=12:00..12:05  count=17      # the previous answer, removed
+I  window=12:00..12:05  count=19      # the revised answer

The pair appears together, in one atomic unit if the transport allows it. A consumer that sees the deletion without the insertion — because the transport delivered them in separate transactions, or because the consumer filtered one out — observes a window with no answer. A consumer that sees the insertion without the deletion double-counts.

Systems that support retractions usually express the change as a signed update instead: a delta of +2 for the count, applied to whatever the consumer currently holds. That is compact and it works when the operation is a sum over an additive aggregate, and it fails for anything non-additive. A revised distinct count, a revised median, a revised minimum cannot be expressed as a delta; they need the old value removed and the new one inserted.

The choice between delta and full-row retraction is therefore determined by the aggregate, not by preference. Additive aggregates can ship deltas; everything else ships rows.

Why late data forces the issue

Without late data, a window fires once and every convention coincides. The conventions diverge exactly when the result is revised, and revisions are caused by records arriving after the watermark passed. The rate of revisions is therefore the rate of lateness, and the lateness budget from Measuring Disorder to Size a Lateness Budget sets how often the downstream must cope with them.

This connects the whole pipeline. A job configured with a watermark delay at the 99th percentile and an allowed lateness covering the rest produces revisions for roughly one percent of windows, and the downstream must handle one percent of its input as corrections. A job with no allowed lateness produces no revisions and silently drops the late records, so its output is consistent and incomplete. The consumer’s complexity and the pipeline’s completeness are traded against each other by a number chosen in the window configuration.

The awkward case is a window whose revision arrives after the consumer has already materialized and published it. If the consumer is a table with a primary key, the revision is an update and the publication is a second version. If the consumer is an append-only file, the revision is a second row and the file is now a log of revisions rather than a table of answers — which is fine if the reader knows to take the latest per key, and wrong if it does not. Two-Phase Commit and Exactly-Once Sinks covers the atomicity of the write; the question here is the semantics of the row once written.

The dynamic table view

The cleanest way to think about a changelog stream is as the materialization of a table that changes over time. At each point in the stream, the accumulated set of insertions and deletions defines a table’s contents. Reading the stream up to a position gives the table as of that position; reading further gives a later version. The stream is the table’s changelog, and a query over the stream is a query over a table that never stops changing.

This view explains why a stream-to-stream transformation behaves like a SQL query. A filter over a changelog emits the insertions and deletions that pass the predicate and drops the rest, which is a valid changelog for the filtered table. A join over two changelogs must emit the pairs that become valid and retract the pairs that cease to be valid when either side changes, which is why joins with outer semantics produce retractions and why their state is the buffered input.

The view also explains the boundary. A changelog is only a faithful representation of the table if every change is emitted. An operator that drops records — because they were late, because a filter excluded them, because an error handler swallowed them — produces a changelog that no longer reconstructs the table, and a consumer replaying it from the beginning arrives at the wrong contents. The completeness of the changelog is the property that matters, and it is destroyed by any silent discard.

-- The accumulated changelog materialized as a table.
SELECT window_start, window_end, SUM(delta) AS count
FROM window_changes
GROUP BY window_start, window_end
HAVING SUM(delta) <> 0;      -- rows fully retracted drop out

The HAVING clause is the detail that makes the materialization faithful: a row whose insertions and deletions cancel to zero must disappear from the table, and a query that only sums insertions keeps it forever.

Retractions across an aggregation boundary

Composing two aggregations is where retraction handling becomes hard. The upstream produces a count per window; the downstream sums those counts per hour. When a window’s count is revised, the downstream’s hourly sum must change by the difference, and the downstream must know the previous value to compute the difference.

Two implementations exist. The first has the downstream retain the previous value per input row, which is state proportional to the number of input rows it has seen. The second has the upstream emit both the old and new value, so the downstream can subtract the old and add the new without retaining anything. The second is cheaper in downstream state and more expensive in stream volume, and it is the reason retraction records carry the full previous row rather than only a delta.

The trade is the same one that appears everywhere in this cluster: state versus bandwidth. A pipeline whose aggregations are chained five deep can choose to pay in retained copies at each level or in duplicated rows on the wire. For a stream with few revisions the bandwidth is negligible and the full-row form is clearly better; for a stream where most windows are revised many times, retaining state is cheaper than shipping the history.

The consumer’s obligation

Whatever convention the producer uses, the consumer must be told which one. A consumer that assumes append mode over a retract stream adds every insertion and ignores every deletion, producing a total that grows without bound. A consumer that assumes replace mode over a retract stream that carries no key cannot overwrite, so it must maintain a multiset and apply both operations.

The practical mechanism is to make the convention part of the stream’s schema rather than part of its documentation. A record type that carries an operation tag forces every consumer to handle the tag, and a table sink that declares its primary key forces the producer to emit keys. Where the schema cannot express it, the convention has to be enforced at the boundary — a single materialization step that converts the changelog into a keyed upsert stream before anything else reads it, so that exactly one component in the pipeline understands retractions.

The limitation worth stating is that retractions make a stream’s output depend on its entire history for correctness. A consumer that starts reading from the middle, or from a compacted topic that retained only the latest value per key, sees a valid table but not the operations that produced it. That is fine for a table and wrong for anything that needs to know how a value was reached, and the two uses of the same stream require different retention policies on the transport. A compacted topic preserves the latest answer; an uncompacted one preserves the revisions, and a pipeline that needs both is really two pipelines.