Coordinated State Maintenance

Recall that in Chapter 5 two representatives for each input tuple, called

build tuple and probe tuple, are used, each with distinct assigned responsi-

bilities. Build tuples will be inserted into the states of the join operators and stay there until being purged. The probe tuples instead will be prop- agated throughout the ring structure for probing corresponding states of other streams to identically matching the join predicates and then perform the actual result construction. Note that all the states are composed of the build tuples from corresponding streams. In the ring structure, the build tuples will “move” from the head node towards the tail node steadily, as

12.2. STATE-SLICE RING WITH LIFE CONTROL 141

we will further explain this in Chapter 12.3, until finally it can be surly purged after it has been probed at the tail of the join pipeline. The follow- ing example routine shows how each node processes state insertion and purging.

Example: Suppose at time t, a tuple a_tarrives from stream A at the entry state- sliced join operator J1 in the ring. Then two copies abt (build) and apt (probe) are

made. Tuple ab

tis inserted into the current state (abt, abj, ..., abi+1, abi,abi−1), ordered

in decreasing order of their timestamps. Suppose ab

i and abi−1are the only tuples

which are now fall outside of the current window range due to the arrival of ap_t at the sliced state. Then the state will be (ab

t, abj, ..., abi+1) after purging triggered

by ap_t and the output queue then is augmented by pushing (ap_t,ab_i,ab_i−1) into the queue. Later when ab

i−1 and abi are processed by the next join operator J2, they will be inserted into the state of J₂. Thereafter when ap_t is processed by J₂, it will be used to purge and probe the corresponding state in J2.

From the above example, we have following lemma.

Lemma 7 For any node holding Ji with a state sliced window S :: [Wis, Wie]

on certain input stream S, at the time that a probe tuple sp _{with timestamp T} sp

finishes the purge step, but has not yet began the probe step, we have: (1) ∀s0b _∈

S :: [W_is, W_ie] ⇒ W_is≤ Tsp− T_s0b< W_ie; and (2) ∀s0btuple in the input steam S

that Ws

i ≤ Tsp− T_s0b < W_ie⇒ s0b∈ S :: [W_is, W_ie]. Here S :: [W_is, W_ie] denotes

the sliced state of stream S at J_i.

Proof: (1). In the purge step, the processing of sp _{will purge any tuple s}0b with Tsp − T_s0b ≥ W_ie. Thus ∀s0b ∈ S :: [W_is, W_ie], T_sp − T_s0b < W_ie. For

the first sliced window join in the ring, Ws

12.2. STATE-SLICE RING WITH LIFE CONTROL 142

For other joins Ji in the ring, at any moment let tuple s00b denote the tuple in S :: [W_is, W_ie] that has the maximum timestamp. Tuple s00b must have been purged by sp (or another probe tuple with smaller timestamp) from the state of the up-stream join operator in the ring. Thus we have Ws

i ≤

T_sp− T_s0b, for ∀s0b ∈ S :: [W_is, W_ie].

(2). We use a proof by contradiction. (a) If s0b _{∈ S :: [W/} s

i, Wie], then we assume s0b _{∈ S :: [W}s

j, Wje], j < i. Given Wie ≤ Tsp − T_s0b, we know

We

j ≤ Tsp − T_s0b. Then s0b cannot be inside the state S :: [W_js, W_je] since

s0b _{would have been purged by s}p _{when it is processed by the up-stream} join operator Jj. A[Wj−1, Wj]

s

n B. (b) Let us assume s0b _{in the input} queue of Ji. Since s0b is purged by sp and is inserted into the queue before

sp_{, s}0b _{cannot be in the input queue when s}p _{is being processed by J} i. (c) Let us assume s0b _{in the output queue of J}

i or down-stream joins. Since

Tsp − T_s0b < W_ie, no probe tuple will purge s0b from J_i. In a summary, we

got contradictions in all the possible cases.

Lemma 7 indicates the implicit synchronization of the sliced states in the ring of nodes by using the probe tuples, since the probing tuple is placed behind all purged tuples in the output queue. Lemma 7 is guar- anteed due to the FIFO property of the network connections between pro- cessing nodes. The state maintenance at each node is coordinated by every probe tuple. Thus even though the state maintenance processes will not happen at the same time at all the nodes, the states are guaranteed to be consistent in terms of join probing process.

Coordinated state maintenance achieves implicit synchronization in the cluster. That is, the state synchronization is postponed as long as possible

12.2. STATE-SLICE RING WITH LIFE CONTROL 143

until right before the join probing process commences. Also this coordi- nation involves no extra network messages since the probing tuples have to be propagated for join probing purposes anyway. Along with the join progress, the probe tuples are propagated step-by-step along the ring.