Maintained Result Sets: Exact Top-N Under Continuous Change

#Abstract

A client subscribing to a filtered, sorted and limited query needs to be told exactly how the result set changes, not merely that it changed. This paper describes a construction maintaining the in-engine window as an exact prefix of the source-ordered result, using a cushion and keyset boundary refill, so that enter, leave, move and update deltas are exact at all times. The single-node case is implemented. The sharded case is not, and the correctness argument for it is incomplete.

#The subscription problem

Let R be a relation, σ a predicate over it, ≺ a total order on a key k, and N a limit. The query Q = (σ, ≺, N) returns the first N rows of R satisfying σ under ≺. A client subscribes to Q and receives an initial snapshot. Thereafter, as committed changes arrive, it must receive a sequence of deltas such that applying them to the snapshot yields, at every point, exactly what Q would return if re-executed against R at that point.

Four delta types suffice, and the requirement is that each is emitted exactly when it occurs and not otherwise:

• enter, with the row and its position, when a row that was not in the result becomes part of it;
• leave, with the identity and prior position, when a row that was in the result ceases to be;
• move, with both positions, when a row remains in the result but its rank changes;
• update, with the changed attributes, when a row remains in the result at the same rank and its non-key attributes change.

The distinction between these matters to the client in a way that "something changed" does not. A client that receives only an invalidation must re-fetch, which costs a round trip and a query, and which reproduces the load the subscription was intended to remove. A client that receives a move can animate a row to its new position; one that receives an enter and a leave for the same row cannot tell the two situations apart.

#Three implementations that do not work

Re-execute on every change. Correct by definition, and the cost is a full query per change per subscription. Deltas must then be recovered by diffing consecutive results, which is O(N) work on top. This scales with the product of subscription count and change rate, which is the wrong product.

Maintain the top N in memory. The obvious optimisation. Hold the N result rows, apply changes to them, emit deltas from the difference. It scales, and it is wrong at exactly one point: when a row leaves the window, the row that should be promoted into position N is not held, because only N rows were ever loaded. The implementation must either issue a query at that moment, which it usually does without accounting for the concurrency hazard discussed below, or emit a result of N minus one rows, which is silently incorrect and is what several implementations do.

Filter a raw changefeed at the client. Ship every change matching σ and let the client maintain its own ordering. This removes server state and reintroduces the original problem, since the client now holds an unbounded set in order to know which N of them are the top N, and a client that holds only N has the boundary problem again in a place with less information.

The boundary is the entire problem. Everything away from position N is straightforward, and every implementation that is wrong is wrong at position N.

#The prefix invariant

The construction rests on maintaining a set W that is not the result but a strict superset of it, chosen so that a single stated property makes the four deltas derivable.

Invariant: W is an exact prefix of the sequence of rows in R satisfying σ, ordered by ≺.

Exactness in both directions is what matters. Every row of R satisfying σ that ranks before the last member of W is in W, and W contains no row that does not satisfy σ. Given the invariant, the first N members of W are exactly Q, and the deltas follow by comparing successive states of that prefix. Nothing else about R needs to be known.

W is sized at N + c, where c is a cushion. The cushion exists solely so that a departure from the first N has a successor already in hand.

#The algorithm

1func (s *Subscription) apply(ch Change) []Delta {
2	before := s.window.snapshot()          // W as an ordered slice
3
4	switch {
5	case ch.Kind == Insert && s.sigma(ch.Row):
6		s.window.insertOrdered(ch.Row)     // no-op if it ranks past the last held key
7	case ch.Kind == Delete || !s.sigma(ch.Row):
8		s.window.remove(ch.RowID)
9	default:
10		s.window.replace(ch.Row)           // may reposition if the sort key moved
11	}
12
13	if s.window.len() < s.n+s.cushion && !s.window.exhausted {
14		s.refillFromBoundary()             // keyset query, strictly after the last held key
15	}
16	return diffPrefix(before, s.window.snapshot(), s.n)
17}
Copy

Refill uses a keyset predicate rather than an offset. Offsets are not stable under concurrent insertion: a row inserted before the offset shifts every subsequent position by one, so an offset-based refill can skip a row or return one already held. Keyset pagination is stated in terms of the ordering key itself and is therefore stable under insertion anywhere [1]. This is not an optimisation. An offset-based refill breaks the invariant and therefore breaks the deltas.

Note that the ordered insert discards rows ranking past the last held key. This is sound precisely because of the invariant: a row beyond the prefix is not part of the prefix, and ignoring it leaves W a prefix. It is also the reason the construction is cheap for large relations, since the vast majority of changes to a large R fall outside W and cost a comparison.

#Correctness

Four cases must preserve the invariant. The first three are the change types; the fourth is refill itself, which is the case usually omitted and the one where implementations go wrong.

1. Insert. A row satisfying σ is positioned by ≺. If it ranks within the held prefix it enters W, and W grows by one, possibly displacing a row past position N which then leaves the result. If it ranks beyond the last held key it is discarded and W is unchanged. In both branches W remains an exact prefix: in the first because an insertion into a prefix at its correct rank is a prefix, in the second because the prefix boundary did not move.
2. Delete or predicate exit. The row leaves W. If it was within the first N, the row formerly at position N + 1 enters the result. Such a row exists in W whenever the cushion is non-empty, and when the cushion is empty the refill case applies. W remains a prefix because removing an element from a prefix and keeping the rest in order yields a prefix of the remaining sequence.
3. Update. If k is unchanged, an update delta is emitted and ordering is untouched. If k changed, the operation is treated as a delete followed by an insert, reducing to the two cases above, and the pair is reported as a move when both endpoints fall within the first N.
4. Refill. The keyset query returns rows satisfying σ that rank strictly after the last held key, in order, up to the deficit. Appending them extends the prefix. The subtlety is concurrency: a change committed between the moment the deficit was observed and the moment the refill query executes may insert a row into the range being queried. Because the refill is expressed by key rather than by position, such a row is either returned by the query, in which case it is placed correctly, or it arrives as a change and is positioned by case one. What must not happen is that it is processed as a change before the refill result is merged, since that would place it and then the refill would place it again. The implementation serialises refill against change application per subscription for this reason.

The argument assumes changes are observed in the order the source committed them. A versioned log with per-entity ordering provides this [2]. Under reordering, case one may position a row against a state that no longer holds, and the invariant does not survive.

#What isolation level makes the target well defined

The specification says the deltas must yield "what Q would return if re-executed". Under concurrent transactions that phrase is not self-evidently meaningful, since re-execution at what point and observing which uncommitted work are both open questions.

The implementation assumes read-committed semantics at the source and takes the target to be the result Q would return in a fresh read-committed transaction started immediately after the change in question committed. This is stated rather than assumed because under weaker isolation the target itself becomes ambiguous, and a subscription cannot be more consistent than the source it derives from [3] [4] [5].

One consequence is worth spelling out. Under read committed, a subscriber may observe a sequence of states that no single snapshot would ever have shown, because each delta reflects a different committed point. For a live list of assets this is not merely acceptable but desired: the client wants the current state, not a consistent historical one. For a subscription feeding a computation whose correctness depends on a consistent cut, it is wrong, and such a consumer needs a snapshot read rather than a subscription.

#Cost

Memory is O(N + c) per subscription, independent of the size of R. Per-change cost is one comparison against the last held key for the common case where the change falls outside W, and O(log(N + c)) for positioning when it falls inside.

The interesting quantity is refill frequency, since refill is the only operation that touches the source. A refill occurs when the held set falls below N + c, which requires c departures from W since the last refill. Departures are driven by churn near the boundary rather than by total change rate, and those two quantities are close to independent. This is what makes the construction practical: a relation with very high change rate and a stable top N is inexpensive to maintain, because almost every change is a comparison that fails.

The pathological workload is therefore one where the ordering key is itself the frequently updated attribute, for example a list sorted by last-modified time. There, nearly every change displaces a row and the refill rate approaches the change rate divided by c, which is the point at which re-execution becomes competitive.

#Choosing the cushion

c trades memory against refill frequency, linearly in the first and inversely in the second. A cushion of roughly twenty per cent of N was adequate across the workloads examined, meaning refills were rare enough not to appear in source query profiles.

No principled method for choosing c is offered, and this is a genuine gap rather than a detail. The quantity that determines the right value is the displacement probability in the expression above, which is a property of the workload and is measurable at runtime. An implementation could adapt c from observed refill frequency, targeting a refill rate rather than fixing a size. That was not built.

#Sharding, and where the argument stops

Where R is partitioned, each shard can maintain a local prefix, and the global result is a merge of local prefixes. The merge is straightforward. The difficulty is the cushion.

A shard holding no rows near the global boundary cannot determine locally whether its next row would enter the global result, because it does not know the global boundary key. Two answers present themselves. Every shard maintains a cushion sized for the worst case, meaning the case where the entire global result comes from that shard, which costs S times the memory for S shards and is wasteful in the common case where results are spread evenly. Or the merge coordinator holds the global boundary key and pushes it to shards, which is correct but introduces a coordination round on every boundary movement and turns a local operation into a distributed one.

Neither has been implemented and the correctness argument above does not extend without a stated ordering across shards. Where k is not globally comparable, for example where it is a per-shard sequence, the construction does not apply at all, and a formulation based on conflict-free replicated types may be a better fit for the merge [6].

#Relation to incremental view maintenance

The problem is a special case of incremental view maintenance, which asks how to update a materialised view from changes to its inputs rather than recomputing it [7] [8]. That literature is far more general than what is described here, covering joins, aggregation and recursion, and the differential formulation handles arbitrary dataflow with iteration [9].

Three things distinguish this construction from applying that work directly. First, the view is bounded by N, which the general theory does not exploit and which is the source of the memory bound. Second, the output is a positional delta rather than a bag difference, since clients care about rank; the general formulation produces multiset deltas from which position must be recovered separately. Third, the interesting cost is the boundary refill against the source, which does not arise in a setting where the entire view is maintained in the engine.

The top-N aspect connects to threshold algorithms for ranked retrieval, which bound how much of a sorted source must be read to be certain of the top N [10]. The connection is real and was not pursued: a refill strategy informed by threshold reasoning might refill less often than the fixed cushion does, by bounding when a boundary crossing is possible rather than assuming it always is.

#What has not been measured

The single-node construction is implemented and in use. Beyond that, the evidence is thin, and the following are the specific things a reader should not assume have been shown.

No measurement of refill frequency against churn distribution is reported. The claim that boundary churn and total change rate are close to independent is an argument from the structure of the algorithm, not an observation, and the workload where it fails is identified above rather than characterised.

The cushion figure of twenty per cent is empirical in the weak sense: it was adequate and was not tuned against alternatives. No comparison against re-execution at varying N, relation size and churn is offered, so the crossover point where this construction stops being worth its complexity is unknown.

The construction assumes a total order on k and degrades to re-execution when the ordering involves a computed expression the source cannot index, since the refill query then cannot be expressed as a keyset predicate. How common that case is in practice was not surveyed.