StreamInsight Checkpoints: What, How and Why?

I’ve been fielding some questions this week on checkpoints in StreamInsight. I’ll share my way of thinking about checkpoints in the hopes that it will help others build applications leveraging the feature. First, I’ll define what a StreamInsight checkpoint represents. Then I’ll explain how to interpret the high-water mark information StreamInsight provides to input and output adapters when a query is resumed after StreamInsight downtime (planned or unplanned). There’s an important piece of trivia here that may surprise even veteran users!

The MSDN documentation outlines three levels of resiliency. In this post I will focus on the strictest resiliency level, but the concepts outlined are relevant for all three.

Think of a StreamInsight query as a black box. It takes a one or more sequences of events as inputs. It produces a sequence of events as output (in the upcoming 2.1 StreamInsight release, multiple output sequences are supported as well) ^*. It probably comes as no surprise that the black box contains some state. If your query is computing averages over windows, incoming events will contribute to some number of windows. Until the average over a particular window can be committed to the output, sums and counts need to be maintained internally. When you checkpoint a query, you’re really just saving the internal state of the query. It’s that simple!

Well, almost. In addition to the query state, a checkpoint captures the position of input and output sequences as of the checkpoint^**. For example, if a checkpoint could speak it might say “between enqueuing input events x₂ and x­₃, and between dequeing output events y₂ and y₃, this was the state of the query”.

After a StreamInsight server instance has been stopped, a query can be resumed using the checkpoint state. Input and output adapters need to do some work as well. After the checkpoint was taken (but before the server instance was stopped), the input sequence may have progressed (let’s say x₃ was enqueued) and the output sequence may have progressed as well (let’s say y₃ was dequeued). Ideally, the input adapter would then replay x₃ and subsequent events. And the ideal output adapter would forget that it had ever seen y₃ or anything after it.

Instead of forgetting – by, say, deleting rows from a table or removing lines from a log file – the output adapter may instead choose to suppress output events it knows have been emitted already. The latter approach is relatively difficult to get right however:

1. The output may include multiple events with the same timestamp(s) and payload value. Consider the case where the checkpoint is taken between two identical events with same timestamp. A naïve de-duplication implementation may incorrectly suppress the second event because the first (equivalent) event has been emitted already.
2. While StreamInsight deterministically produces logically consistent results, there is no guarantee about the specific sequence of output events between runs. The number of CTIs may vary. The order of events with the same timestamp may also vary between runs (or even the order of events with different timestamps when using StreamEventOrder.ChainOrdered).
3. The output adapter needs to reliably determine which events have already been emitted.

In any case, the rule of thumb when resuming a query:

Input adapters must replay events after the checkpoint; output adapters must forget events after the checkpoint.

StreamInsight provides a high-water mark (HWM, pronounced huh-wim I think) value to input adapters that can be used to determine where in the input sequence a checkpoint occurred (see IHighWaterMarkInputAdapterFactory and IHighWaterMarkTypedInputAdapterFactory). Output adapters get both an HWM value and an offset (see IHighWaterMarkOutputAdapterFactory and IHighWaterMarkTypedOutputAdapterFactory). How can these values be used to identify an element of a sequence? First, let’s define HWM. An event x_i has high-water mark value h if its timestamp is h and for all j < i, x_j has a lower timestamp. Whenever an event in a sequence has a higher timestamp than any preceding event, it has a HWM value. Assuming that your input sequence is conveniently described by an IEnumerable<PointEvent<T>> sequence xs, you can find the event corresponding to an hwm as follows:

var inputCheckpointEvent = xs.FirstOrDefault(x => x.StartTime == hwm)

Assuming that the output from the query is captured in a (resilient and persistent) List<PointEvent<T>> , and given hwm and offset values for an output sequence, you just need to pop ahead offset positions to find the checkpoint event:

var outputCheckpointEvent = ys
.Where(x => x.StartTime >= hwm)
.Skip(offset).FirstOrDefault()

I promised a piece of trivia. While I’ve shown you how to find the “checkpoint event”, I haven’t told you where the checkpoint occurred: did it occur before the checkpoint event, or after? It turns out that the answer is before for an input checkpoint event but after for an output checkpoint event. Returning to the earlier example, if the checkpoint occurred between input events x₂ and x₃, the checkpoint event will be x₃, but if the checkpoint occurred between output events y₂ and y₃, the checkpoint event will be y₂! Again using the simple IEnumerable<> contract, I can demonstrate how replay works:

// replay all events after and including the checkpoint event
var replayXs = hwm.HasValue
? xs.SkipWhile(x => x.StartTime < hwm.Value)
: xs;

When forgetting output events and again assuming (somewhat unrealistically) that a List<> is the external reliable authority on which events have been emitted:

// remove all outputs events after but not including the checkpoint event
if (hwm.HasValue)
{
int hwmEvent = ys
.Select((y, i) => new { y.StartTime, Index = i })
.FirstOrDefault(y => y.StartTime == hwm.Value);
if (hwmEvent != null)
{
int forgetIndex = hwmEvent.Index + offset + 1;
if (forgetIndex < ys.Count)
{
ys.RemoveRange(forgetIndex, ys.Count – forgetIndex);
}
}
}

Let’s tie all of this information together with a simple walkthrough. In this example, the query state is described by the input events contributing to the state.

Action	Query state	Output list	Notes
Enqueue x1	{ x1 }	{ y1 }
Begin checkpoint	{ x1 }	{ y1 }	Waiting for HWM event on input
Enqueue x2	{ x1, x2 }	{ y1, y2 }	Let’s assume x2 does not increase the HWM, so the checkpoint cannot be initiated yet
Enqueue x3	{ x1, x2 }	{ y1, y2 }	x3 is the input checkpoint event, current query state is captured before processing x3
(continued)	{ x1, x2, x3 }	{ y1, y2, y3 }	y2 is the output checkpoint event, as it is the last event before the checkpoint
End checkpoint	{ x1, x2, x3 }	{ y1, y2, y3 }	Checkpoint <{ x1, x2 }, x3, y2> committed to disk
Enqueue x4	{ x1, x2, x3, x4 }	{ y1, y2, y3, y4 }
Downtime!
Resume	{ x1, x2 }	{ y1, y2, y3, y4 }	StreamInsight recovers query state, output adapter forgets all output events after y2
Enqueue x3	{ x1, x2, x3 }	{ y1, y2, y3 }	Input replays from x3
Enqueue x4	{ x1, x2, x3, x4 }	{ y1, y2, y3, y4 }
…

Notice that by replaying input events and forgetting output events, we see correct output after resuming the query!

* Notice that I’m talking about “sequences of events” rather than “temporal streams”. While StreamInsight operators are defined in terms of temporal streams – sequences of timestamped events with ordering constraints imposed by common time increments (or CTIs) – a checkpoint can be understood relative to generic sequences.

** StreamInsight doesn’t actually freeze the query operators, inputs and outputs in order to take a snapshot of the internal state, but this is a useful and accurate way of characterizing the logical contract for checkpoints. Details of the internal mechanism aren’t covered in this blog post.