The increasing number of applications that require processing and analyzing high-throughput information in real time has fostered a new field of study known as Complex Event Processing (CEP). The claimed objective of CEP is to develop techniques that are able to cope with the high-throughput requirements of modern Big Data applications. Also, it is commonly argued that CEP systems are different from relational reactive systems such as Active Database Management Systems (ADBMSs) or Data Stream Management Systems (DSMSs) because the latter see new data elements as database transactions (generally insertions or deletions) whose order is not relevant. On the contrary, CEP systems see new data elements as events (e.g. sensor measurements) whose arrival time and order have a semantic meaning. Unfortunately, these differences come from a high-level description of the underlying applications, but do not reveal fundamental computational differences between the requirements of CEP systems and relational systems. Moreover, recent developments in Dynamic Query Evaluation (DQE) show that general techniques in the relational setting can be more efficient than current CEP algorithms. In this paper, we study whether there is a fundamental difference between the computational requirements of CEP and DQE. To answer this question, we identify two concrete assumptions of CEP and investigate their effects in terms of evaluation complexity. Concretely, we show a realistic CEP query that, if the Online Matrix-vector multiplication (OMv) conjecture holds, cannot be evaluated with sub-linear time per tuple followed by sub-linear-delay enumeration, but under the CEP assumptions can be evaluated with constant time per tuple followed by constant-delay enumeration. Sub-linear here means O(n 1−ε ), where n is the size of the active domain.
In recent years, analyzing high-throughput streams in real time has become a crucial task for many communities. Examples range from Stream Processing [24] and Financial Systems [9] to Industrial Control Systems [1] and On-line Machine Learning [23]. The diversity of these communities and the nature of their applications has resulted in a variety of frameworks and systems to analyze data streams. These systems fall in different categories (see [11] for an extensive survey), out of which Complex Event Processing and Dynamic Query Evaluation are two prominent examples.
The term Dynamic Query Evaluation (DQE) refers to techniques whose aim is to efficiently maintain an up-to-date version of a query result under updates to an underlying database. DQE is usually based on a form of Incremental View Maintenance (IVM) [3,6,7,18], whose objective is to maintain a materialized version of query results under updates by applying relational operations over cached results and sub-results. In particular, in DQE the input is a stream of database transactions that can insert as well as delete tuples. Systems based on DQE find their origins in Active Database Management Systems (ADBMS) and Data Stream Management Systems (DSMS), that were designed to execute relational operations every time a transaction was executed in a database [2].
Complex Event Processing (CEP) refers to techniques where the objective is to detect certain temporal patterns in a high-throughput data stream. Therefore, elements in the input stream are not seen as database transactions but as events, that represent real-world observations and that thus cannot be retracted. Moreover, the order at which these events arrive to the system has a semantic meaning; CEP patterns commonly express sequences based on this temporal order. CEP and DQE are usually considered very different as is perhaps best illustrated by Cugola and Margara who, in their well-known survey [11], argue that "The concepts of timeliness and flow processing are crucial for justifying the need for a new class of systems". In line with this difference, specialized CEP systems were developed to deal with the throughput requirements of applications scenarios involving CEP patterns. Prominent examples of CEP systems are SASE [26], Cayuga [12], EsperTech [13], TESLA/T-Rex [10], RTEC [4], among others. While CEP and DQE have hence historically been considered different in the literature, this perceived difference is not based on fundamental computational properties, but rather on a high-level description of the targeted applications of CEP and DQE. Several recent trends make the perceived difference between CEP and DQE rather shallow. First of all, CEP and DQE application scenarios seem to intermingle. For this reason, modern distributed stream processing systems (such as Apache Flink1 , Spark Streaming2 or Apache Storm3 ) mingle both incremental query maintenance capabilities and CEP features like time windows or time-based semantics. Second, DQE has received new attention in the last couple of years [15,21] based on a simple yet important idea: instead of maintaining a materialized version of the query results, maintain a data structure that can be updated efficiently under changes to the underlying database, and from which the query results can be efficiently generated. Recent work shows that relational techniques based on this idea can be more efficient than the CEP systems cited above [16].
This hence raises the question: is there a fundamental computational difference between CEP and DQE ? In order to answer this question, we take a fundamental viewpoint, and define CEP systems as systems in which the input is a stream of tuples. In line with the fact that events cannot be retracted, the stream in insertion-only. Moreover, in line with the fact that arrival order is important, we require that every tuple contains a special attribute that represents the arrival time to the system. The value of this attribute is globally increasing across the stream. In contrast, DQE systems take as input a stream of both insertions and deletions with no further assumption on the attributes.
Concretely, we answer the question by showing a query that can be answered efficiently under the CEP assumptions, but not under the DQE assumptions. The notion of efficiency here corresponds to process each new event in constant time (under data complexity) and be able to output the results at any moment with constant-delay enumeration.
Tuples, databases, predicates and streams. We assume the existence of three disjoint infinite sets: A domain D, a set of variables {x, y, z, . . .} (which play the role of attribute names), and a set of relation names {r, s, t, . . .}.
An atom is an expression of the form r(x), where r is a relation name and x is a finite set of variables. A tuple over r(x) (or simply over x) is a function t : x → D. A relation R over r(x) is a finite set of tuples over r(x). A schema S is a finite set of atoms with different relation names, and a database D over schema S is a set of relations, one over each atom of S. A stream over schema S is a possibly infinite sequence S = (S [1], S [2], . . .), where each S[i] is a pair (t i , • i ) consisting of a tuple t i over any of the atoms in S and a symbol • i ∈ {+, −} that represents whether S[i] is an insertion or a deletion. A stream is called insertion-only if all of its elements are insertions. We might abuse notation and assume the elements of an insertion-only stream are tuples. Given a stream S, the sub-stream (S[i], . . . , S[j]) is denoted by S i,j , and the database generated by S up to position i, denoted by S[1..i], is the database consisting of all tuples t that occur in S up to position i (inclusive), and whose last occurrence in S 1,i is an insertion (i.e. (t, +)).
Finally, a predicate over x is a (not necessarily finite) decidable set P of tuples over x. For example, if x = {x, y} then P (x) = {(x, y) | x < y} is the predicate of all tuples (x 0 , y 0 ) ∈ D 2 satisfying x 0 < y 0 . As usual, we use shorthand notation for predicates (e.g. P (x) = x < y) and write P (t) to indicate that t ∈ P .
Query Language. We follow the query language of Generalized Conjunctive Queries (GCQs) introduced in [16].
where r 1 (x 1 ), . . . , r n (x n ) are atoms, P 1 , . . . , P m are predicates over z 1 , . . . , z m , respectively, and both y and
Given a database D and a GCQ Q of the form (1), the evaluation of Q over D is denoted Q(D) and is performed in the expected way: first compute the natural join r 1 (x 1 ) • • • r n (x n ), from the result keep only those tuples that satisfy all predicates P 1 , . . . , P m , and finally project them over y. We illustrate the semantics of GCQs with an example, for a full definition see [16].
Example 1. Consider the following GCQ.
Intuitively, the query specifies that we should take the natural join between r(x, ts 1 ), s(x, y, ts 2 ) and t(y, ts 3 ), from this result keep only those tuples that satisfy ts 1 < ts 2 < ts 3 and (ts 3 − ts 1 ) < 5, and finally project over {x, y, ts 3 }.
From the definitions of databases, streams and the semantics of GCQs it can be seen that we use set semantics. We stress, however, that this is only for simplicity, and the results in this paper can be extended easily to bag semantics.
GCQs for Complex Event Processing. It has been claimed in the past that CEP languages are generally informal and non-standard [10,11,26], presenting a variety of primitives and operators with sometimes problematic semantics. Nevertheless, GCQs capture some fundamental requirements of CEP. For example, in CEP frameworks users can specify that events must occur in a particular order and inside a bounded time window. Although this is not directly provided by built-in features of GCQs, it can be simulated by adding a globally-increasing attribute to tuples and using inequality predicates. For example, in query (2) we could interpret ts 1 , ts 2 and ts 3 as the time of arrival of each event. Then, this query would select three events of type r, s and t that arrived in that same order and inside a time window of 5 units of time (we assume seconds).
In contrast to GCQs, CEP languages usually have sequencing and time windows as primitives. For example, under the assumption that ts 1 , ts 2 and ts 3 represent arrival order, query (2) can be written equivalently in SASE [26] as SEQ(r,s,t) WHERE r.1 = s.1 AND s.2 = t.1 WITHIN 5 seconds.
The SEQ operator indicates that the events must arrive in the expected order and the WITHIN clause that they have to occur within a time window of 5 seconds, avoiding the explicit inequalities over the timestamps. The WHERE clause indicates that the first attribute of r must be equal to the first attribute of s and the second attribute of s must be equal to the first attribute of t, which is equivalent to using variable names in GCQs.
For the purposes of this paper, we only require the CEP features of sequencing, time windows and filtering; it is not relevant that other features of CEP languages like iteration (Kleene closure) are not expressible by means of GCQs.
Since we are interested in evaluating GCQs over streams, we need to formally define what evaluation means. Traditionally, the objective of DQE algorithms like IVM has been to maintain the current version of the result materialized at all times. Therefore, the processing of an update includes the corresponding modification of the materialized result. In contrast, recent DQE algorithms do not maintain the full materialized output, but a data structure from which the result can be generated [5]. The idea behind this is to separate the enumeration of results from the process of updating the data structure when the underlying database changes. Note that this implies that the time required to react to an update can be much shorter than the size of the change to the output, because the data structure can be a compressed representation of the output. This is actually the main reason behind the efficiency gain in recent DQE algorithms [15,16,19].
When evaluating a query Q over a stream S, the elements of S are consumed one by one in order. Once the i th element of S is processed, we expect to be able to generate Q(S[1..i]) (recall that S[1..i] is the database produced by S up to position i). Following this, a DQE algorithm A for Q consists of a data structure D A and two routines PROCESS A and ENUM A such that for every stream S:
when PROCESS A receives an element of the stream, it can modify D A and; after PROCESS A has been called with the first i elements of the stream,
Complexity. We consider query evaluation in main memory and measure time under data complexity [25]. That is, the query is considered to be fixed and not part of the input. This makes sense under DQE, where the query is known in advance and the data is constantly changing. In particular, the number of atoms and predicates in the query, their arity, and the length of the query are all constant. Also, note that we need to measure result enumeration and update processing separately. Update processing is simply measured as the worst-case time required to process an element from the stream, i.e. the worst-case time required by PROCESS A . Formally, given a function f from streams to natural numbers, we say that a DQE algorithm A provides f update-time if, for every stream S and i ∈ N,
The efficiency of enumeration is slightly more involved, since it is measured as the delay that the enumeration procedure takes between generating two consecutive results [5,22]. This is defined as follows.
Definition 2. A routine ENUM with access to a data structure D supports enumeration of a set E if ENUM(D) outputs each element of E exactly once. Moreover, ENUM(D) enumerates E with delay d ∈ N, if when ENUM(D) is invoked, the time until the output of the first element; the time between any two consecutive elements; and the time between the output of the last element and the termination of ENUM(D), are all bounded by d.
Given a function f from streams to natural numbers, we say that a DQE algorithm A for query Q provides f -delay enumeration if for every stream S and every i ∈ N, after PROCESS A has processed the first i elements of S, ENUM A enumerates Q(S[1..i]) with delay O(f (S 1,i )). If f is a constant function, A is said to provide constant-delay enumeration (CDE) of Q(S[1..i]).
Computational Model. Another important consideration when studying the efficiency of DQE algorithms is the computational model. Indeed, when dealing with fine-grained notions like constant-time per update or CDE, the computational model might affect the complexity.
We follow the extended RAM model from [15]. Concretely, we assume a model of computation where domain values take O(1) space and memory lookups are O(1). We further assume that every relation R can be represented by a data structure that allows (1) enumeration of all tuples in R with constant delay (i.e. the existence of a routine as defined above); (2) lookups in O(1), meaning that given t we can decide in constant time whether t ∈ R; (3) single-tuple insertions and deletions in O(1) time; while (4) having size that is proportional to the number of tuples in R.
We further assume the existence of linear-size hash-maps. Concretely a hash map H over a set of variables x can store an arbitrary data structure H(t) for each tuple t over x, using size proportional to the sum of the sizes of the data structures stored. Moreover, given a tuple t we can access H(t) in constant time. These assumptions amount to perfect hashing of linear size [8]. Although this is not realistic for practical computers [20], it is well known that complexity results for this model can be translated, through amortized analysis, to average complexity in real-life implementations [8].
As mentioned in the introduction, the differences between DQE and CEP generally come from high-level descriptions of the targeted applications, and not from fundamental differences in their processing techniques. It makes sense then to discuss whether CEP really requires new computational insights for achieving better performance, or DQE algorithms are already optimal for the requirements of CEP. To understand this, we first need to identify the concrete requirements and restrictions that distinguish CEP from DQE. As discussed in the introduction, CEP adds to the dynamic evaluation problem (1) the requirement of selecting tuples based on their arrival times and (2) the restriction that witnessed tuples correspond to real-world events and thus cannot be retracted. This can be concretely reflected in GCQ evaluation by 1. interpreting some attributes as timestamps and assuming that they occur in a globally increasing manner, and 2. assuming that streams are insertion-only.
We show that taking these assumptions into account can make a difference in terms of complexity of DQE algorithms. Concretely, we prove that the GCQ Q = π y,ts3 (r(x, ts 1 ) s(x, y, ts 2 ) t(y, ts 3 ) | ts 1 < ts 2 < ts 3 ∧ (ts 3 − ts 1 ) < 5) can be evaluated with constant update time followed by CDE when restricted to streams that satisfy the two assumptions above, but not in general.
We start by presenting the hardness result. We show that there is no DQE algorithm that evaluates Q over streams with sub-linear update time followed by sub-linear-delay enumeration. Here, sub-linear means O(n 1−ε ), where n is the number of elements in the active domain of the database produced by the processed portion of the stream. We use the dichotomy proved recently by Berkholz et. al. in [5], which is based on the Online Matrix-vector Multiplication (OMv) conjecture (see [14]). The full description of the dichotomy is rather involved, but it implies that a if a CQ is not q-hierarchical then it cannot be evaluated over dynamic datasets with sub-linear time per update followed by sub-linear-delay enumeration. For the sake of space, we do not give here the formal definition of q-hierarchical queries, but only state that the query Q = π y (u(x) v(x, y) w(y)) is not q-hierarchical.
Lemma 1. If the OMv conjecture holds, there is no DQE algorithm for
with sub-linear time per update followed by sub-linear-delay enumeration.
Proof (Sketch). We prove this by contradiction. Assume that there is a DQE algorithm A that evaluates Q with sub-linear update time and sub-linear-delay enumeration. We use this algorithm to define a second algorithm A that evaluates query Q = π y (u(x) v(x, y) w(y)) with the same bounds. First, the data structure D A is exactly the same as the data structure D A . Upon the arrival of an element (t, •), the routine PROCESS A will simply call PROCESS A (u, •), where u is defined as follows: if t = u(x 0 ), then u = r(x 0 , 1); if t = v(x 0 , y 0 ), then u = s(x 0 , y 0 , 2); if t = w(y 0 ), then u = t(y 0 , 3). Since PROCESS A (u, •) takes sub-linear time, it is clear that PROCESS A (t, •) also takes sub-linear time. Finally, the routine ENUM A is equivalent to ENUM A except for the fact that it needs to project out ts 3 . It is clear that this process will generate the expected output with sub-linear-delay enumeration: the filters will be satisfied because ts 1 is always 1, ts 2 is always 2 and ts 3 is always 3. Moreover, enumeration is correct and without duplicates (even though we project away ts 3 ) exactly because ts 3 is always 3. We hence have our desired contradiction as we have constructed a DQE algorithm with sub-linear update time followed by sub-linear-delay enumeration that evaluates Q , which is not q-hierarchical.
Next, we show a DQE algorithm A that evaluates Q with constant time per update followed by CDE under the assumptions that streams are insertion-only and ts 1 , ts 2 and ts 3 arrive in a globally increasing fashion. We start by presenting the data structure D A , to then describe how PROCESS A maintains it. After the first i elements of S are processed, D A contains three elements.
Algorithm 1 UPDATE A : Updates D A upon receiving a tuple t. Having access to D A the enumeration procedure is trivial, since the output is already materialized in relation T of D A . We proceed then to describe the update process.
Update processing. The objective of PROCESS A , shown in Algorithm 1, is to maintain all elements of D A consistent with the definitions given above. Whenever a new tuple t of type r(x, ts 1 ) arrives, it suffices to insert ts 1 as the last timestamp of x (Line 3). Note that we can only do this because we assume that timestamps arrive in a globally increasing fashion. As such, the value of ts 1 is the maximum across all timestamps seen so far. When a tuple t = s(x, y, ts 2 ) arrives, we need to check if the x value has already occurred in an r tuple (Line 5), and if so, update the value of H S (y) to H R (x). Again, this maintains the consistency because we know that ts 2 is larger than H R (x). Finally, when a new tuple t of type t(y, ts 3 ) arrives, we check if we need to insert it into T . To this end, we only need to check the value H S (y), as this value tells us if there are matching tuples of type r and s in the required time window (Line 8).
It is clear that PROCESS A runs in constant time under our computational model. Also, because we assume that ts 1 , ts 2 and ts 3 are globally increasing, it is easy to see that PROCESS A correctly maintains the elements of D A .
Theorem 1. There exists a DQE algorithm that evaluates Q with constant time per tuple and CDE under the assumption that streams are insertion-only and that attributes ts 1 , ts 2 and ts 3 are globally increasing.
In this section we have shown a computational difference between CEP and DQE, giving a concrete motivation for studying techniques that differ from those used in the relational setting. We conclude by further discussing this difference and providing insight into the particular aspects that might make a query easy to compute in a CEP setting.
We have presented the difference between CEP and DQE by means of a very particular query. We decided to present this query because (1) it features both sequencing and time windows, making it realistic for CEP (2) the algorithm to evaluate it with constant time per update followed by CDE allows for a clear presentation. Nevertheless, one possible criticism is that this query is too simple, because every new event can generate at most one output tuple, and therefore constant-delay enumeration is easy to achieve by materializing the output (which is actually what we do). This is a very valid argument; queries for which a single tuple can generate multiple outputs are indeed harder to evaluate. One possibility to present such a query would have been to add variable x to the output of Q (as presented in Query (2) of Example 1). In this case, we can still improve the evaluation performance by using dynamic fractional cascading techniques [17], which provide logarithmic update-time followed by logarithmicdelay enumeration. This is still sub-linear in the active domain and therefore an improvement over the general case, but the presentation of this result would have been far more involved. We are not aware of constant-time per update followed by CDE in this case. It is also important to notice that we could have presented a simpler query by removing one of the CEP features. For example, if we remove from Q the time window (ts 3 − ts 1 < 5), we obtain a simpler DQE algorithm with the same performance guarantees.
Apart from structural properties of the query, we can also look at the CEP assumptions. At the moment, we do not know whether both assumptions (insertiononly streams and globally-sorted timestamps) are required, nor how is complexity affected if we remove each of them separately. We envision this and a general understanding of the query properties that affect complexity as future work.
Finally, in CEP it is common to distinguish between push-based and pullbased systems (see for example [11,16]). Intuitively, in a pull-based setting the system must be ready to generate the complete answer Q(S[1..i]) whenever a user asks for it. This corresponds to the semantics presented in this paper. On the contrary, when element S[i] arrives to a push-based system, the system must inform the user of the output's delta, i.e. the difference between Q(S[1..i]) and Q(s[1..i − 1]). Interestingly, for the case of push-based systems we can evaluate Query (2) of Example 1 with logarithmic update time followed by constant-delay enumeration (of the output's delta). Moreover, if we take out the time window filter but keep x in the output, we can provide constant-time per update followed by CDE in the push-based case, but we are not aware of an algorithm achieving these times in a pull-based setting. All of these insights suggest that push-based evaluation is simpler than pull-based evaluation; we also envision an in-depth study of this difference as future work.