To Share, or not to Share Online Event Trend Aggregation
Over Bursty Event Streams
Technical Report

Time is represented by a linearly ordered set of time points $(\mathbb{T},\leq)$, where $\mathbb{T}\subseteq\mathbb{Q^{+}}$ are the non-negative rational numbers. An event $e$ is a data tuple describing an incident of interest to the application. An event $e$ has a time stamp $e.time\in\mathbb{T}$ assigned by the event source. An event $e$ belongs to a particular event type $E$, denoted e.type=E and described by a schema that specifies the set of event attributes and the domains of their values. A specific attribute $\mathit{attr}$ of $E$ is referred to as $E.\mathit{attr}$. Table 2 summarizes the notation.

Events are sent by event producers (e.g., vehicles and mobile devices) to an event stream $I$. We assume that events arrive in order by their time stamps. Existing approaches to handle out-of-order events can be applied [11, 26, 27, 41].

An event consumer (e.g., Uber stream analytics) continuously monitors the stream with event queries. We adopt the commonly used query language and semantics from SASE [9, 44, 45]. The query workload in Figure 1 is expressed in this language. We assume that the workload is static. Adding or removing a query from a workload requires migration of the execution plan to a new workload which can be handled by existing approaches [24, 48].

Aggregation of Event Trends. Within each window specified by the query $q$, event trends are grouped by the values of grouping attributes $G$. Aggregates are then computed per group. Hamlet focuses on distributive (COUNT, MIN, MAX, SUM) and algebraic aggregation functions (AVG) since they can be computed incrementally [16]. Let $E$ be an event type, $\mathit{attr}$ be an attribute of $E$, and $e$ be an event of type $E$. While $\textsf{\footnotesize COUNT}(*)$ returns the number of all trends per group, $\textsf{\footnotesize COUNT}(E)$ computes the number of all events $e$ in all trends per group. $\textsf{\footnotesize SUM}(E.\mathit{attr})$ ($\textsf{\footnotesize AVG}(E.\mathit{attr})$) calculates the summation (average) of the value of $\mathit{attr}$ of all events $e$ in all trends per group. $\textsf{\footnotesize MIN}(E.\mathit{attr})$ ($\textsf{\footnotesize MAX}(E.\mathit{attr})$) computes the minimal (maximal) value of $\mathit{attr}$ for all events $e$ in all trends per group.

Given a workload of event trend aggregation queries $Q$ and a high-rate event stream $I$, the Multi-query Event Trend Aggregation Problem is to evaluate the workload $Q$ over the stream $I$ such that the average query latency of all queries in $Q$ is minimal. The latency of a query $q\in Q$ is measured as the difference between the time point of the aggregation result output by the query $q$ and the arrival time of the last event that contributed to this result.

We design the Hamlet framework in Figure 2 to tackle this problem. To reveal all sharing opportunities in the workload at compile time, the Hamlet Optimizer identifies sharable queries and translates them into a Finite State Automaton-based representation, called the merged query template (Section 3.1). Based on this template, the optimizer reveals which sub-patterns could potentially be shared by which queries. At runtime, the optimizer estimates the sharing benefit depending on the current stream characteristics to make fine-grained sharing decisions. Each sharing decision determines which queries share the processing of which Kleene sub-patterns and for how long (Section 4). These decisions along with the template are encoded into the runtime configuration to guide the executor.

Hamlet Executor partitions the stream by the values of grouping attributes. To enable shared execution despite different windows of sharable queries, the executor further partitions the stream into panes that are sharable across overlapping windows [10, 17, 24, 25]. Based on the merged query template for each set of sharable queries, the executor compactly encodes matched trends within a pane into the Hamlet graph. More precisely, matched events are modeled as nodes, while event adjacency relations in a trend are edges of the graph. Based on this graph, we incrementally compute trend aggregates by propagating intermediate aggregates along the edges from previously matched events to new events – without constructing the actual trends. This reduces the time complexity of trend aggregation from exponential to quadratic in the number of matched events compared to two-step approaches [19, 22, 32, 45].

The Hamlet graph is partitioned into sub-graphs, called graphlets, by event type and time stamps to maximally expose runtime opportunities to share these graphlets among queries. Since the value of aggregates may be differ for distinct queries, we capture these aggregate values per query as ”snapshots” and share the propagation of snapshots through shared graphlets (Section 3.3).

Lastly, the executor implements the sharing decisions imposed by the optimizer. This may involve dynamically splitting a shared graphlet into several non-shared graphlets or, vice-versa, merging several non-shared graphlets into one shared graphlet (Section 4.2).

Given that the workload may contain queries with different Kleene patterns, aggregation functions, windows, and groupby clauses, Hamlet takes the following pre-processing steps: (1) it breaks the workload into sets of sharable queries at compile time; (2) it then constructs the Hamlet query template for each sharable query set; and (3) it partitions the stream by window and groupby clauses for each query template at runtime.

However, sharable Kleene sub-patterns cannot always be shared due to other query clauses. For example, queries having $\textsf{\footnotesize COUNT}(*)$, $\textsf{\footnotesize MIN}(E.$ $\mathit{attr})$ or $\textsf{\footnotesize MAX}(E.\mathit{attr})$ can only be shared with queries that compute these same aggregates. In contrast, since $\textsf{\footnotesize AVG}(E.\mathit{attr})$ is computed as $\textsf{\footnotesize SUM}(E.\mathit{attr})$ divided by $\textsf{\footnotesize COUNT}(E)$, queries computing $\textsf{\footnotesize AVG}(E.\mathit{attr})$ can be shared with queries that calculate $\textsf{\footnotesize SUM}(E.\mathit{attr})$ or $\textsf{\footnotesize COUNT}(E)$. We therefore define sharable queries below.

To facilitate the shared runtime execution of each set of sharable queries, each pattern is converted into its Finite State Automaton-based representation [9, 14, 44, 45], called query template. We adopt the state-of-the-art algorithm [33] to convert each pattern in in the workload $Q$ into its template.

Figure 3(a) depicts the template of query $q_{1}$ with pattern $\textsf{\footnotesize SEQ}(A,$ $B+)$. States, shown as rectangles, represent event types in the pattern. If a transition connects a type $E_{1}$ with a type $E_{2}$ in a template of a query $q$, then events of type $E_{1}$ precede events of type $E_{2}$ in a trend matched by $q$. $E_{1}$ is called a predecessor type of $E_{2}$, denoted $E_{1}\in\textit{pt}(E_{2},q)$. A state without ingoing edges is a start type, and a state shown as a double rectangle is an end type in a pattern.

Our Hamlet system processes the entire workload $Q$ instead of each query in isolation. To expose all sharing opportunities in $Q$, we convert the entire workload $Q$ into one Hamlet query template. It is constructed analogously to a query template with two additional rules. First, each event type is represented in the merged template only once. Second, each transition is labeled by the set of queries for which this transition holds.

The event stream is first partitioned by the grouping attributes. To enable shared execution despite different windows of sharable queries, Hamlet further partitions the stream into panes that are sharable across overlapping windows [10, 17, 24, 25]. The size of a pane is the greatest common divisor (gcd) of all window sizes and window slides. For example, for two windows $(\textsf{\footnotesize WITHIN}\ 10\ min\ \textsf{\footnotesize SLIDE}$ $5\ min)$ and $(\textsf{\footnotesize WITHIN}\ 15\ min$ $\textsf{\footnotesize SLIDE}\ 5$ $min)$, the gcd is 5 minutes. In this example, a pane contains all events per 5 minutes interval. For each set of sharable queries, we apply the Hamlet optimizer and executor within each pane.

For the non-shared execution, we describe below how the Hamlet executor leverages state-of-the-art online trend aggregation approach  [33] to compute trend aggregates for each query independently from all other queries. Given a query $q$, it encodes all trends matched by $q$ in a query graph. The nodes in the graph are events matched by $q$. Two events $e^{\prime}$ and $e$ are connected by an edge if $e^{\prime}$ and $e$ are adjacent in a trend matched by $q$. The event $e^{\prime}$ is called a predecessor event of $e$. At runtime, trend aggregates are propagated along the edges. In this way, we aggregate trends online, i.e., without actually constructing them.

Assume a query $q$ computes the number of trends $\textsf{\footnotesize COUNT}(*)$. When an event $e$ is matched by $q$, $e$ is inserted in the graph for $q$ and the intermediate trend count of $e$ (denoted $\mathit{count}(e,q)$) is computed. $\mathit{count}(e,q)$ corresponds to the number of trends that are matched by $q$ and end at $e$. If $e$ is of start type of $q$, $e$ starts a new trend. Thus, $\mathit{count}(e,q)$ is incremented by one (Equation 1). In addition, $e$ extends all trends that were previously matched by $q$. Thus, $\mathit{count}(e,q)$ is incremented by the sum of the intermediate trend counts of the predecessor events of $e$ that were matched by $q$ (denoted $\textit{pe}(e,q)$) (Equation 2). The final trend count of $q$ is the sum of intermediate trend counts of all matched events of end type of $q$ (Equation 3).

Complexity Analysis. Figure 4(a) illustrates that each event of type $B$ is stored and processed once for each query in the workload $Q$, introducing significant re-computation and replication overhead. Let $k$ denote the number of queries in the workload $Q$ and $n$ the number of events. Each query $q$ stores each matched event $e$ and computes the intermediate count of $e$ per Equation 2. All predecessor events of $e$ must be accessed, with $e$ having at most $n$ predecessor events. Thus, the time complexity of non-shared online trend aggregation is computed as follows:

Events that are matched by $k$ queries are replicated $k$ times (Figure 4(a)). Each event stores its intermediate trend count. In addition, one final result is stored per query. Thus, the space complexity is $O(k\times n+k)=O(k\times n)$.

In Equation 4, the overhead of processing each event once per query in the workload $Q$ is represented by the multiplicative factor $k$. Since the number of queries in a production workload may reach hundreds to thousands [37, 43], this re-computation overhead can be significant. Thus, we design an efficient shared online trend aggregation strategy that encapsulates bursts of events of the same type in a graphlet such that the propagation of trend aggregates within these graphlets can be shared among several queries.

Example 5 illustrates the following two challenges of online shared event trend aggregation.

Challenge 1. Given that event $b_{3}$ has different predecessors for queries $q_{1}$ and $q_{2}$, the computation of the intermediate trend count of $b_{3}$ (and all other events in graphlets $B_{3}$ and $B_{6}$) cannot be directly shared by queries $q_{1}$ and $q_{2}$.

Challenge 2. If queries $q_{1}$ or $q_{2}$ have predicates, then not all previously matched events are qualified to contribute to the trend count of a new event. Assume that the edge between events $b_{4}$ and $b_{5}$ holds for $q_{1}$ but not for $q_{2}$ due to predicates, and all other edges hold for both queries. Then $count(b_{4},q_{1})$ contributes to $count(b_{5},q_{1})$, but $count(b_{4},q_{2})$ does not contribute to $count(b_{5},q_{2})$.

We tackle these challenges by introducing snapshots. Intuitively, a snapshot is a variable that its value corresponds to an intermediate trend aggregate per query. In Figure 4(b), the propagation of a snapshot $x$ within graphlet $B_{3}$ is shared by queries $q_{1}$ and $q_{2}$. We store the values of $x$ per query (e.g., $x=2$ for $q_{1}$ and $x=1$ for $q_{2}$).

To enable shared trend aggregation despite expressive predicates, we now introduce snapshots at the event level.

Shared Online Trend Aggregation Algorithm computes the number of trends per query $q\in Q$ in the stream $I$. For simplicity, we assume that the stream $I$ contains events within one pane. For each event $e\in I$ of type $E$, Algorithm 1 constructs the Hamlet graph and computes the trend count as follows.

Hamlet graph construction (Lines 4–14). When an event $e$ of type $E$ is matched by a query $q\in Q$, $e$ is inserted into a graphlet $G_{E}$ that stores events of type $E$ (Line 14). if there is no active graphlet $G_{E}$ of events of type $E$, we create a new graphlet $G_{E}$, mark it as active and store it in the Hamlet graph $G$ (Lines 7–8). If the graphlet $G_{E}$ is shared by queries $Q_{E}\subseteq Q$, then we create a snapshot $x$ at graphlet level (Line 9). $x$ captures the values of intermediate trend counts per query per Equation 5 at the end of graphlet $G_{E^{\prime}}$ that stores events of type $E^{\prime},\ E^{\prime}\in pt(E,q)$. We save the value of $x$ per query in the table of snapshots $S$ (Lines 10–13). Also, for each query $q\in Q$ with event types $T$, we mark all graphlets $G_{E^{\prime}}$ of events of type $E^{\prime}\in T,\ E^{\prime}\neq E,$ as inactive (Lines 4–6).

Trend count computation (Lines 16–24). If $G_{E}$ is shared by queries $Q_{E}\subseteq Q$ and the set of predecessor events of $e$ is identical for all queries $q\in Q_{E}$, then we compute $count(e,q)$ per Equation 2 (Lines 16–18). If $G_{E}$ is shared but the sets of predecessor events of $e$ differ among the different queries in $Q_{E}$ due to predicates, then we create a snapshot $y$ as the intermediate trend count of $e$ (Line 19). We compute the value of $y$ for each query $q\in Q_{E}$ per Equation 2 and save it in the table of snapshots $S$ (Line 20). If $G_{E}$ is not shared, the algorithm defaults to the non-shared trend count propagation per Equation 2 (Line 21). If $E$ is an end type for a query $q\in Q$, we increment the final trend count of $q$ in the table of results $R$ by the intermediate trend count of $e$ for $q$ per Equation 3 (Lines 22–23). Lastly, we return the table of results $R$ (Line 24).

Data Structures. Algorithm 1 utilizes the following physical data structures.

(1) Hamlet graph $G$ is a set of all graphlets. Each graphlet has two metadata flags active and shared (Definitions 6 and 7).

(2) A hash table of snapshot coefficients per event $e$. The intermediate trend count of $e$ may be an expression composed of several snapshots. In Figure 5(c), $count(b_{6},Q)=4x+z$. Such composed expressions are stored in a hash table per event that maps a snapshot to its coefficient. In this example, $x\mapsto 4$ and $z\mapsto 1$ for $b_{6}$.

(3) A hash table of snapshots $S$ is a mapping from a snapshot $x$ and a query $q$ to the value of $x$ for $q$ (Tables 5 and 5).

(4) A hash table of trend count results $R$ is a mapping from a query $q$ to its corresponding trend count.

Complexity Analysis. We use the notations in Table 2 and Algorithm 1. For each event $e$ that is matched by a query $q\in Q$, Algorithm 1 computes the intermediate trend count of $e$ in an online fashion. This requires access to all predecessor events of $e$. In the worst case, $n$ previously matched events are the predecessor events of $e$. Since the intermediate trend count of $e$ can be an expression that is composed of $s$ snapshots, the intermediate trend count of $e$ is stored in the hash table that maps snapshots to their coefficients. Thus, the time complexity of intermediate trend count computation is $O(n\times s)$. In addition, the final trend count is updated per query $q$ if $E$ is an end type of $q$ in $O(k\times s)$ time. In summary, the time complexity of trend count computation is $O(n\times(n\times s+k\times s))=O(n^{2}\times s)$ since $n\geq k$.

In addition, Algorithm 1 maintains snapshots to enable shared trend count computation. To compute the values of $s$ snapshots for each query $q$ in the workload of $k$ queries, the algorithm accesses $g$ events in $t$ graphlets $G_{E^{\prime}}$ of events of type $E^{\prime}\in T,\ E^{\prime}\neq E$. Thus, the time complexity of snapshot maintenance is $O(s\times k\times g\times t)$. In summary, time complexity of Algorithm 1 is computed as follows:

Algorithm 1 stores each matched event in the Hamlet graph once for the entire workload. Each shared event stores a hash table of snapshot coefficients. Each non-shared event stores its intermediate trend count. In addition, the algorithm stores snapshot values per query. Lastly, the algorithm stores one final result per query. Thus, the space complexity is $O(n+n\times s+s\times k+k)=O(n\times s+s\times k)$.

On the up side, shared trend aggregation avoids the re-computation overhead for each query in the workload. On the down side, it introduces overhead to maintain snapshots. Next, we quantify the trade-off between shared versus non-shared execution.

Equations 4 and 6 determine the cost of non-shared and shared strategies of all events within the window for the entire workload $Q$ based on stream statistics. In contrast to these coarse-grained static decisions, the Hamlet optimizer makes fine-grained runtime decisions for each burst of events for a sub-set of queries $Q_{E}\subseteq Q$. Intuitively, a burst is a set of consecutive events of type $E$, the processing of which can be shared by queries $Q_{E}$ that contain a $E+$ Kleene sub-pattern. The Hamlet optimizer decides at runtime if sharing a burst is beneficial. In this way, beneficial sharing opportunities are harvested for each burst at runtime.

Within each pane, events that belong to the same burst are buffered until a burst is complete. The arrival of an event of type $E^{\prime}$ or the end of the pane indicates that the burst is complete. In the following, we refer to complete bursts as bursts for compactness.

Hamlet restricts event types in a burst for the following reason. Assuming that a burst contained an event $e$ of type $E^{\prime}$, the event $e$ could be matched by one query $q_{1}$ but not by another query $q_{2}$ in $Q_{E}$. Snapshots would have to be introduced to differentiate between the aggregates of $q_{1}$ and $q_{2}$ (Section 3.3). Maintenance of these snapshots may reduce the benefit of sharing. Thus, the previous sharing decision may have to be reconsidered as soon as the first event arrives that is matched by some queries in $Q_{E}$.

Based on Definition 12, we conclude that the more queries $k$ share trend aggregation, the more events $g$ are in shared graphlets, and the fewer snapshots $s_{c}$ and $s_{p}$ are maintained at a time, the higher the benefit of sharing will be. Based on this conclusion, our dynamic Hamlet optimizer decides to share or not to share online trend aggregation (Section 4.2).

Based on Definition 12, we conclude that the more queries $k$ share trend aggregation, the more events $g$ are in shared graphlets, and the fewer snapshots $s_{c}$ and $s_{p}$ are maintained at a time, the higher the benefit of sharing will be. Based on this conclusion, our dynamic Hamlet optimizer decides to share or not to share online trend aggregation (Section 4.2).

Our dynamic Hamlet optimizer monitors the sharing benefit depending on changing stream conditions at runtime. Let $B+$ be a sub-pattern sharable by queries $Q_{B}=\{q_{1},q_{2}\}$. In Figure 6, pane boundaries are depicted as dashed vertical lines and bursts of newly arrived events of type $B$ are shown as bold empty circles. For each burst, the optimizer has a choice of sharing (Figure 6(a)) versus not sharing (Figure 6(b)). It concludes that it is beneficial to share based on calculations in Equation 9.

Decision to Split. However, when the next burst of events of type $B$ arrives, a new snapshot $y$ has to be created due to predicates during the shared execution in Figure 6(c). In contrast, the non-shared strategy processes queries $q_{1}$ and $q_{2}$ independently from each other (Figure 6(d)). Now the overhead of snapshot maintenance is no longer justified by the benefit of sharing (Equation 10).

Thus, the optimizer decides to split the shared graphlet $B_{3}$ into two non-shared graphlets $B_{4}$ and $B_{5}$ for the queries $q_{1}$ and $q_{2}$ respectively in Figure 6(d). Newly arriving events of type $B$ then must be inserted into both graphlets $B_{4}$ and $B_{5}$. Their intermediate trend counts are computed separately for the queries $q_{1}$ and $q_{2}$. The snapshot $x$ is replaced by its value for the query $q_{1}$ ($q_{2}$) within the graphlet $B_{4}$ ($B_{5}$). The graphlets $A_{1}$ and $C_{2}$ are collapsed.

Decision to Merge. When the next burst of events of type $B$ arrives, we could either continue the non-shared trend count propagation within $B_{4}$ and $B_{5}$ (Figure 6(e)) or merge $B_{4}$ and $B_{5}$ into a new shared graphlet $B_{6}$ (Figure 6(f)). The Hamlet optimizer concludes that the latter option is more beneficial in Equation 11. As a consequence, a new snapshot $z$ is created as input to $B_{6}$. $z$ consolidates the intermediate trend counts of the snapshot $x$ and the graphlets $B_{3}$–$B_{5}$ per query $q_{1}$ and $q_{2}$.

Complexity Analysis. The runtime sharing decision per burst has constant time complexity because it simply plugs in locally available stream statistics into Equation 8. A graphlet split comes for free since we simply continue graph construction per query (Figure 6(d)). Merging graphlets requires creation of one snapshot and calculation of its values per query (Figure 6(f)). Thus, the time complexity of merging is $O(k\times g\times t)$ (Equation 6). Since our workload is fixed (Section 2), the number of queries $k$ and the number of types $t$ per query are constants. Thus, the time complexity of merge is linear in the number of events per graphlet $g$. Merging graphlets requires storing the value of one snapshot per query. Thus, its space complexity is $O(k)$.

To relax the assumption from Section 4.2 that a set of queries $Q_{E}$ that share a Kleene sub-pattern $E+$ is given, we now select a sub-set of queries $Q_{E}$ from the workload $Q$ for which sharing $E+$ is the most beneficial among all other sub-sets of $Q$. In general, the search space of all sub-sets of $Q$ is exponential in the number of queries in $Q$ since all combinations of shared and non-shared queries in $Q$ are considered. For example, if $Q$ contains four queries, Figure 7 illustrates the search space of 12 possible execution plans of $Q$. Groups of queries in braces are shared. For example, the plan (134)(2) denotes that queries 1, 3, 4 share their execution, while query 2 is processed separately. The search space ranges from maximally shared (top node) to non-shared (bottom node) plans. Each plan has its execution cost associated with it. For example, the cost of the plan (134)(2) is computed as the sum of $Shared(G_{E},\{1,3,4\})$ and $NonShared(G_{E}^{i},2)$ (Equation 8). The goal of the dynamic Hamlet optimizer is to find a plan with minimal execution cost.

Traversing the exponential search space for each Kleene sub-pattern and each burst of events would jeopardize real-time responsiveness of Hamlet. Fortunately, most plans in this search space can be pruned without loosing optimality (Theorems 4.1 and 4.2). Intuitively, Theorem 4.1 states that it is always beneficial to share the execution of a query that introduces no new snapshots.