# DBSP: Automatic Incremental View Maintenance for Rich Query Languages

Mihai Budiu  VMware Research<EMAIL_ADDRESS>,  Frank McSherry  Materialize
Inc<EMAIL_ADDRESS>,  Leonid Ryzhyk  VMware Research
<EMAIL_ADDRESS>and  Val Tannen  University of Pennsylvania
<EMAIL_ADDRESS>
###### Abstract.

Incremental view maintenance has been for a long time a central problem in
database theory (gupta-idb93, ). Many solutions have been proposed for
restricted classes of database languages, such as the relational algebra, or
Datalog. These techniques do not naturally generalize to richer languages. In
this paper we give a general solution to this problem in 3 steps: (1) we
describe a simple but expressive language called DBSP for describing
computations over data streams; (2) we give a general algorithm for solving
the incremental view maintenance problem for arbitrary DBSP programs, and (3)
we show how to model many rich database query languages (including the full
relational queries, grouping and aggregation, monotonic and non-monotonic
recursion, and streaming aggregation) using DBSP. As a consequence, we obtain
efficient incremental view maintenance techniques for all these rich
languages.

##  1\. Introduction

In this paper we present a simple mathematical theory for modeling streaming
and incremental computations. This model has immediate practical applications
in the design and implementation of streaming databases and incremental view
maintenance. Our model is based on mathematical formalisms used in discrete
digital signal processing (DSP) (rabiner-book75, ), but we apply it to
database computations. Thus, we have called it “DBSP”. DBSP is inspired from
Differential Dataflow (mcsherry-cidr13, ) (DD), and started as an attempt to
provide a simpler formalization of DD than the one of Abadi et al. (abadi-
fossacs15, ) (as discussed in §9), but has evolved behind that purpose.

The core concept of DBSP is the _stream_ : a stream $s$ with type
$\mathcal{S}_{A}$ maps “time” moments $t\in\mathbb{N}$ to values $s[t]$ of
type $A$; think of it as an ”infinite vector”. A streaming computation is a
function that consumes one or more streams and produces another stream. We
depict streaming computations with typical DSP box-and-arrow diagrams (also
called “circuits”), where boxes are computations and streams are arrows, as in
the following diagram, which shows a stream operator $T$ consuming two input
streams $s_{0}$ and $s_{1}$ and producing one output stream $s$:

$s_{0}$$s_{1}$$T$$s$

We generally think of streams as sequences of _small_ values, and we will use
them in this way. However, we make a leap of imagination and also treat a
_whole database_ as a stream value. What is a stream of databases? It is a
_sequence of database snapshots_. We model the time-evolution of a database
$DB$ as a stream $DB\in\mathcal{S}_{SCH}$, where $SCH$ is the database schema.
Time is not the wall-clock time, but essentially a counter of the _sequence of
transactions_ applied to the database. Since transactions are linearizable,
they have a total order, which defines a linear time $t$ dimension: the value
of the stream $DB[t]$ is the snapshot of the database contents after $t$
transactions have been applied. We assume that $DB[0]=0$, i.e., the database
starts empty.

Database transactions also form a stream $T$, a stream of _changes_ , or
_deltas_ that are applied to our database. The database snapshot at time $t$
is the cumulative result of applying all transactions in the sequence up to
$t$: $DB[t]=\sum_{i\leq t}T[i]\stackrel{{\scriptstyle\textrm{\scriptsize
def}}}{{=}}\mathcal{I}(T)[t]$ (we make the notion of “addition” precise
later.). The operation of adding up all changes is _stream integration_. The
following diagram expresses this relationship using the $\mathcal{I}$ operator
for stream integration:

$T$$\mathcal{I}$$DB$

Conversely, we can say that transactions are the _changes_ of a database, and
write $T=\mathcal{D}(DB)$, or $T[t]=DB[t]-DB[t-1]$. This is the definition of
stream differentiation, denoted by $\mathcal{D}$; this operation computes the
changes of a stream, and is the inverse of stream integration. §2 precisely
defines streams, integration and differentiation, and analyzes their
properties.

Let us apply these concepts to view maintenance. Consider a database $DB$ and
a query $Q$ defining a view $V$ as a function of a database snapshot
$V=Q(DB)$. Corresponding to the stream of database snapshots $DB$ we have a
_stream of view snapshots_ : $V[t]$ is the view’s contents after the $t$-th
transaction has been applied. We show this relationship using the following
diagram:

$DB$${\uparrow}Q$$V$

The symbol ${\uparrow}Q$ (the “lifting” of $Q$) shows that the query $Q$ is
applied independently to every element of the stream of database snapshots
$DB$. ${\uparrow}Q$ is a “streaming query” since it operates on a stream of
values. The incremental view maintenance problem requires an algorithm to
compute the stream $\Delta V$ of _changes_ of the view $V$, i.e.,
$\mathcal{D}(V)$, as a function of the stream $T$. By chaining these
definitions together we get the following fundamental equation of the view
maintenance problem: $\Delta
V=\mathcal{D}({\uparrow}Q(DB))=\mathcal{D}({\uparrow}Q(\mathcal{I}(T)))$,
graphically shown as:

$T$$\mathcal{I}$${\uparrow}Q$$\mathcal{D}$$\Delta V$$DB$

This definition can be generalized to more general streaming queries
$S:\mathcal{S}_{A}\to\mathcal{S}_{B}$ that are richer than lifted pointwise
queries $Q$. The incremental version of streaming query $S$ is denoted by
${S}^{\Delta}$ and is defined according to the above equation, which can also
be written as: ${S}^{\Delta}=\mathcal{D}\circ S\circ\mathcal{I}$.

It is generally assumed that the changes to a dataset are much smaller than
the dataset itself; thus, computing on streams of changes may produce
significant performance benefits.

Applying the query _incrementalization_ operator $S\mapsto{S}^{\Delta}$
constructs a query that computes directly on changes; however, the resulting
query is no more efficient than a query that computes on the entire dataset,
because it uses an integration operator to reconstitute the full dataset. §3
shows how algebraic properties of the ${\cdot}^{\Delta}$ operator are used to
optimize the implementation of ${S}^{\Delta}$:

  1. (1)

The first property is that many classes of primitive operations have very
efficient incremental versions. In particular, linear queries have the
property $Q={Q}^{\Delta}$. Almost all relational and Datalog queries are based
on linear operators. Thus, the incremental version of such queries can be
computed in time proportional to the size of the changes. Bilinear operators
(such as joins) have a more complex implementation, which nevertheless still
performs work proportional to the size of the changes, but require storing an
amount of data proportional to the size of the relations.

  2. (2)

The second key property is the chain rule: ${(S_{1}\circ
S_{2})}^{\Delta}={S_{1}}^{\Delta}\circ{S_{2}}^{\Delta}$. This rule gives the
incremental version of a complex query as a composition of incremental
versions of its components. It follows that we can implement any incremental
query as a composition of primitive incremental queries, _all of which perform
work proportional to the size of the changes_.

Armed with this general theory of incremental computation, in §4 we show how
to model relational queries in DBSP. This immediately gives us a general
algorithm to compute the incremental version of any relational query. These
results are well-known, but they are cleanly modeled by DBSP.

Applying DBSP to recursive queries requires extending this computational
model. In §5 we introduce two additional operators: $\delta_{0}$ creates a
stream from a scalar value, and $\int$ creates a scalar value from a stream.
These operators can be used to implement computations with while loops. So, in
addition to modelling changing inputs and database, we also use streams as a
model for sequences of _consecutive values of loop iteration variables_. With
this addition DBSP becomes rich enough to implement recursive queries. §5.1
shows how stratified recursive Datalog programs with negation can be
implemented in DBSP.

In §6 we use DBSP to model computations on nested streams, where each value of
a stream is another stream. This allows us to define _incremental streaming
computations for recursive programs_. As a consequence we derive a universal
algorithm for incrementalizing arbitrary streaming Datalog programs.

DBSP is a simple language: the basic DBSP streaming model is built essentially
from two elementary mathematical operators: lifting ${\uparrow}$ and delay
$z^{-1}$. The nested streams model adds two additional operators, $\delta_{0}$
for stream construction and $\int$ for destruction.

DBSP is also expressive: for example, it is more powerful than stratified
Datalog. DBSP can also describe streaming window queries, or queries on nested
relations (such as grouping), and non-monotone recursive queries. We discuss
briefly the application of DBSP to richer languages in §7.

This paper omits most proofs; the full proofs are available in an expansive
companion technical report (tr, ).

This paper makes the following contributions:

  1. (1)

It defines DBSP, a small language for streaming computation, which nonetheless
can express nested non-monotonic recursion;

  2. (2)

It provides an algorithm incrementalizing all DBSP programs;

  3. (3)

For fragments of DBSP corresponding to the relational algebra and stratified-
monotonic Datalog, the automatic incrementalization algorithm provides results
matching state-of-the-art approaches. Moreover, our approach also applies to
more powerful languages, such as while-relational and non-monotonic Datalog
(Abiteboul-book95, ).

  4. (4)

It develops a formal, sound foundation for the manipulation of streaming and
incremental computations, which allows one to reason formally about program
transformations and design new efficient implementations.

  5. (5)

DBSP can express both streaming and incremental computation models in a single
framework. We regard this unification as a significant contribution.

##  2\. Stream computations

In this section we introduce formally the notion of a stream as an infinite
sequence of values, and we define computations on streams. Stream operators
(§2.1) are the basic building block of stream computations. We employ (§2.2)
restricted types of stream operators: causal operators (which cannot “look
into the future”), and strict operators (which cannot even “look into the
present”). Moreover, all our operators are “synchronous”: they consume and
produce data at the same “rate”. Causal operators can be chained into complex
acyclic computational circuits; cyclic circuits are restricted to using strict
operators on back-edges. Finally, we define (§2.3) two useful stream
operators: integration and differentiation.

All the results in this section have been known for decades, but we
recapitulate them to clarify our model’s formal assumptions.

###  2.1. Streams and stream operators

$\mathbb{N}$ is the set of natural numbers, $\mathbb{B}$ is the set of
Booleans, and $\mathbb{Z}$ is the set of integers.

######  Definition 2.1 (stream):

Given a set $A$, a stream _of values from $A$_, or an _$A$ -stream_, is a
function $\mathbb{N}\rightarrow A$. We denote by
$\mathcal{S}_{A}\stackrel{{\scriptstyle\textrm{\scriptsize
def}}}{{=}}\\{s\,|\,s:\mathbb{N}\to A\\}$ the set of all $A$-streams.

When $s\in\mathcal{S}_{A}$ and $t\in\mathbb{N}$ we write $s[t]$ for the $t$-th
element of the stream $s$ instead of the usual $s(t)$ to distinguish it from
other function applications. We think of the index $t\in\mathbb{N}$ as
(discrete) time and of $s[t]\in A$ as the value of the the stream $s$ “at
time” $t$. For example, the stream of natural numbers
$id\in\mathcal{S}_{\mathbb{N}}$ given by $\mathit{id}[t]=t$ is the sequence of
values
$\setsepchar{}\readlist\arg{01234}{[}\begin{array}[]{cccccc}\arg[1]&\arg[2]&\arg[3]&\arg[4]&\arg[5]&\cdots\end{array}{]}$.

######  Definition 2.2 (stream operator):

A (typed) stream operator with $n$ inputs is a function
$T:\mathcal{S}_{A_{0}}\times\cdots\times\mathcal{S}_{A_{n-1}}\to\mathcal{S}_{B}$.

In general we will use “operator” for functions on streams, and “function” for
computations on “scalar” values.

We are using an extension of the simply-typed lambda calculus to write DBSP
programs; we will introduce its elements gradually. However, we find it more
readable to also use signal-processing-like circuit diagrams to depict DBSP
programs. In a circuit diagram a rectangle represents an operator application
(labeled with the operator name, e.g., $T$), while an arrow is a stream.

Stream operator _composition_ (function composition) is shown as chained
circuits. The composition of a binary operator
$T:\mathcal{S}_{A}\times\mathcal{S}_{B}\to\mathcal{S}_{A}$ with the unary
operator $S:\mathcal{S}_{A}\to\mathcal{S}_{B}$ into the computation $\lambda
s.T(T(s,S(s)),S(s)):\mathcal{S}_{A}\to\mathcal{S}_{A}$ is:

$s$$S$$T$$T$$S$$o$

(Diagrams obscure the _order_ of the inputs of an operator; for non-
commutative operators we have to provide more information.)

######  Definition 2.3:

(lifting) Given a (scalar) function $f:A\to B$, we define a stream operator
${\uparrow}f:\mathcal{S}_{A}\to\mathcal{S}_{B}$ by _lifting_ the function $f$
pointwise in time: $({\uparrow}f)(s)\stackrel{{\scriptstyle\textrm{\scriptsize
def}}}{{=}}f\circ s$. Equivalently,
$({\uparrow}f)(s)[t]\stackrel{{\scriptstyle\textrm{\scriptsize
def}}}{{=}}f(s[t])$. This extends to functions of multiple arguments.

For example, $({\uparrow}(\lambda
x.(2x)))(id)=\setsepchar{}\readlist\arg{02468}{[}\begin{array}[]{cccccc}\arg[1]&\arg[2]&\arg[3]&\arg[4]&\arg[5]&\cdots\end{array}{]}$.

######  Proposition 2.4 (distributivity):

Lifting distributes over function composition: ${\uparrow}(f\circ
g)=({\uparrow}f)\circ({\uparrow}g)$.

We say that two DBSP programs are equivalent if they compute the same input-
output function on streams. We use the symbol $\cong$ to indicate that two
circuits are equivalent. For example, Proposition 2.4 states the following
circuit equivalence:

$s$${\uparrow}g$${\uparrow}f$$o$ |  $\cong$ |  $s$${\uparrow}(f\circ g)$$o$
---|---|---

###  2.2. Streams over abelian groups

For the rest of the technical development we require the set of values $A$ of
a stream $\mathcal{S}_{A}$ to form a commutative group $(A,+,0,-)$. Now we
introduce the primitive stream operators that DBSP uses.

####  2.2.1. Delays and time-invariance

######  Definition 2.5 (Delay):

The delay operator111The name $z^{-1}$ comes from the DSP literature, and is
related to the z-transform (rabiner-book75, ). produces an output stream by
delaying its input by one step:
$z^{-1}_{A}:\mathcal{S}_{A}\to\mathcal{S}_{A}$:

$z^{-1}_{A}(s)[t]\stackrel{{\scriptstyle\textrm{\scriptsize def}}}{{=}}\begin{cases}0&\text{when}~{}t=0_{A}\\\ s[t-1]&\text{when}~{}t\geq 1\end{cases}$ |  $s$$z^{-1}$$o$
---|---

We often omit the type parameter $A$, and write just $z^{-1}$. For example,
$z^{-1}(\mathit{id})=\setsepchar{}\readlist\arg{00123}{[}\begin{array}[]{cccccc}\arg[1]&\arg[2]&\arg[3]&\arg[4]&\arg[5]&\cdots\end{array}{]}$.

######  Definition 2.6 (Time invariance):

A stream operator $S:\mathcal{S}_{A}\to\mathcal{S}_{B}$ is time-invariant iff
$S(z^{-1}_{A}(s))=z^{-1}_{B}(S(s))$ for all $s\in\mathcal{S}_{A}$, or, in
other words, iff the two following circuits are equivalent:

$s$$S$$z^{-1}$$o$ |  $\cong$ |  $s$$z^{-1}$$S$$o$
---|---|---

This definition extends naturally to operators with multiple inputs.

The composition of time-invariant operators of any number of inputs is time
invariant. The delay operator $z^{-1}$ is time-invariant. DBSP only uses time-
invariant operators.

######  Definition 2.7:

We say that a function between groups $f:A\to B$ has the _zero-preservation
property_ if $f(0_{A})=0_{B}$. We write $\mbox{zpp}(f)$.

A lifted operator ${\uparrow}f$ is time-invariant iff $\mbox{zpp}(f)$.

####  2.2.2. Causal and strict operators

######  Definition 2.8 (Causality):

A stream operator $S:\mathcal{S}_{A}\to\mathcal{S}_{B}$ is causal when for all
$s,s^{\prime}\in\mathcal{S}_{A}$, and all times $t$ we have: $(\forall i\leq
t~{}s[i]=s^{\prime}[i])~{}~{}\Rightarrow~{}~{}S(s)[t]=S(s^{\prime})[t].$

In other words, the output value at time $t$ can only depend on input values
from times $t^{\prime}\leq t$. Operators produced by lifting are causal, and
$z^{-1}$ is causal. All DBSP operators are causal. The composition of causal
operators of any number of inputs is causal.

######  Definition 2.9 (Strictness):

A stream operator, $F:\mathcal{S}_{A}\to\mathcal{S}_{B}$ is strictly causal
(abbreviated strict) if $\forall s,s^{\prime}\in\mathcal{S}_{A},\forall
t\in\mathbb{N}$ we have: $(\forall
i<t.~{}s[i]=s^{\prime}[i])~{}~{}\Rightarrow~{}~{}F(s)[t]=F(s^{\prime})[t].$

So the $t$-th output of $F(s)$ can depend only on “past” values of the input
$s$, between $0$ and $t-1$. In particular, $F(s)[0]=0_{B}$ is the same for all
$s\in\mathcal{S}_{A}$. Strict operators are causal. Lifted operators in
general are _not_ strict. $z^{-1}$ is strict.

######  Proposition 2.10:

For a strict $F:\mathcal{S}_{A}\to\mathcal{S}_{A}$ the equation
$\alpha=F(\alpha)$ has a unique solution $\alpha\in\mathcal{S}_{A}$, denoted
by $\mbox{fix}\,\alpha.F(\alpha)$.

Thus every strict operator from a set to itself has a unique fixed point. The
simple proof relies on strong induction, showing that $\alpha[t]$ depends only
on the values of $\alpha$ prior to $t$.

We show that the following circuit, having a strict “feedback” edge $F$, is a
well-defined function on streams:

$s$$T$$\alpha$$F$

######  Lemma 2.11:

If $F:\mathcal{S}_{B}\to\mathcal{S}_{B}$ is strict and
$T:\mathcal{S}_{A}\times\mathcal{S}_{B}\to\mathcal{S}_{B}$ is causal, then for
a fixed $s$ the operator
$\lambda\alpha.T(s,F(\alpha)):\mathcal{S}_{A}\to\mathcal{S}_{B}$ is strict.

######  Corollary 2.12:

If $F:\mathcal{S}_{B}\to\mathcal{S}_{B}$ is strict and
$T:\mathcal{S}_{A}\times\mathcal{S}_{B}\to\mathcal{S}_{B}$ is causal, the
operator $Q(s)=\mbox{fix}\,\alpha.T(s,F(\alpha))$ is well-defined and causal.
If, moreover, $F$ and $T$ are time-invariant then so is $Q$.

All stream computations in DBSP are built from the primitive operators we have
described: lifted operators and delays (we add two more operators in §6).
Circuits composed of such operators can be efficiently implemented using
Dataflow machines (lee-ieee95, ).

Circuits with feedback are used for two purposes: defining an integration
operator (in the next section), and defining recursive computations (§5). In
turn, the integration operator will be instrumental in defining incremental
computations (§3).

###  2.3. Integration and differentiation

Remember that we require the elements of a stream to come from an abelian
group $A$. Streams themselves form an abelian group:

######  Proposition 2.13:

The structure $(\mathcal{S}_{A},+,0,-)$, obtained by lifting the $+$ and unary
$-$ operations from $A$ to $\mathcal{S}_{A}$, is an abelian group.

Stream addition and negation are causal, time-invariant operators.

######  Definition 2.14:

Given abelian groups $A$ and $B$ we call a stream operator
$S:\mathcal{S}_{A}\rightarrow\mathcal{S}_{B}$ linear if it is a group
homomorphism, that is, $S(a+b)=S(a)+S(b)$ (and therefore $S(0)=0$ and
$S(-a)=-S(a)$).

Lifting a linear function $f:A\to B$ produces a stream operator ${\uparrow}f$
that is linear, time-invariant (LTI). $z^{-1}$ is LTI.

######  Definition 2.15:

(bilinear) A function of two arguments $f:A\times B\to C$ with $A,B,C$ groups,
is _bilinear_ if it is linear separately in each argument (i.e., it
distributes over addition): $\forall a,b,c,d.f(a+b,c)=f(a,c)+f(b,c)$, and
$f(a,c+d)=f(a,c)+f(c,d).$

This definition extends to stream operators. Lifting a bilinear function
$f:A\times B\to C$ produces a bilinear stream operator ${\uparrow}f$. An
example bilinear operator over $\mathcal{S}_{\mathbb{Z}}$ is lifted
multiplication:
$f:\mathcal{S}_{\mathbb{N}}\times\mathcal{S}_{\mathbb{N}}\to\mathcal{S}_{\mathbb{N}},f(a,b)[t]=a[t]\cdot
b[t]$.

The composition of (bi)linear operators with linear operators is (bi)linear
(since homomorphisms compose).

The feedback loop produced with a linear operator is linear:

######  Proposition 2.16:

Let $S$ be a unary causal LTI operator. The operator
$Q(s)=\mbox{fix}\,\alpha.S(s+z^{-1}(\alpha))$ is well-defined and LTI:

$s$$+$$S$$\alpha$$z^{-1}$

######  Definition 2.17 (Differentiation):

The differentiation operator
$\mathcal{D}_{\mathcal{S}_{A}}:\mathcal{S}_{A}\to\mathcal{S}_{A}$ is defined
by: $\mathcal{D}(s)\stackrel{{\scriptstyle\textrm{\scriptsize
def}}}{{=}}s-z^{-1}(s)$.

We generally omit the type, and write just $\mathcal{D}$ when the type can be
inferred from the context. The value of $\mathcal{D}(s)$ at time $t$ is the
difference between the current (time $t$) value of $s$ and the previous (time
$t-1$) value of $s$. As an example,
$\mathcal{D}(\mathit{id})=\setsepchar{}\readlist\arg{01111}{[}\begin{array}[]{cccccc}\arg[1]&\arg[2]&\arg[3]&\arg[4]&\arg[5]&\cdots\end{array}{]}$.

If $s$ is a stream, then $\mathcal{D}(s)$ is the _stream of changes_ of $s$.

######  Proposition 2.18:

Differentiation $\mathcal{D}$ is causal and LTI.

$s$$+$$\mathcal{D}(s)$$z^{-1}$$-$ | $s$$+$$\mathcal{I}(s)$$z^{-1}$
---|---
Differentiation | Integration

The integration operator “reconstitutes” a stream from its changes:

######  Definition 2.19 (Integration):

The integration operator
$\mathcal{I}_{\mathcal{S}_{A}}:\mathcal{S}_{A}\to\mathcal{S}_{A}$ is defined
by $\mathcal{I}(s)\stackrel{{\scriptstyle\textrm{\scriptsize
def}}}{{=}}\lambda s.\mbox{fix}\,\alpha.(s+z^{-1}(\alpha))$.

We also generally omit the type, and write just $\mathcal{I}$. This is the
construction from Proposition 2.16 using the identity function for $S$.

######  Proposition 2.20:

$\mathcal{I}(s)$ is the discrete (indefinite) integral applied to the stream
$s$: $\mathcal{I}(s)[t]=\sum_{i\leq t}s[i]$.

As an example,
$\mathcal{I}(\mathit{id})=\setsepchar{}\readlist\arg{013610}{[}\begin{array}[]{cccccc}\arg[1]&\arg[2]&\arg[3]&\arg[4]&\arg[5]&\cdots\end{array}{]}$.

######  Proposition 2.21:

$\mathcal{I}$ is causal and LTI.

######  Theorem 2.22 (Inversion):

Integration and differentiation are inverses of each other: $\forall
s.\mathcal{I}(\mathcal{D}(s))=\mathcal{D}(\mathcal{I}(s))=s$.

$s$$\mathcal{I}$$\mathcal{D}$$o$ |  $\cong$ |  $s$$o$ |  $\cong$ |  $s$$\mathcal{D}$$\mathcal{I}$$o$
---|---|---|---|---

##  3\. Incremental computation

######  Definition 3.1:

Given a unary stream operator $Q:\mathcal{S}_{A}\to\mathcal{S}_{B}$ we define
the incremental version of $Q$ as
${Q}^{\Delta}\stackrel{{\scriptstyle\textrm{\scriptsize
def}}}{{=}}\mathcal{D}\circ Q\circ\mathcal{I}$. ${Q}^{\Delta}$ has the same
“type” as $Q$: ${Q}^{\Delta}:\mathcal{S}_{A}\to\mathcal{S}_{B}$. For an
operator with multiple inputs we define the incremental version by applying
$\mathcal{I}$ to each input independently: e.g., if
$T:\mathcal{S}_{A}\times\mathcal{S}_{B}\rightarrow\mathcal{S}_{C}$ then
${T}^{\Delta}(a,b)\stackrel{{\scriptstyle\textrm{\scriptsize
def}}}{{=}}\mathcal{D}(T(\mathcal{I}(a),\mathcal{I}(b)))$.

The following diagram illustrates the intuition behind this definition:
$\Delta s$$\mathcal{I}$$Q$$\mathcal{D}$$\Delta o$$s$$o$

If $Q(s)=o$ is a computation, then ${Q}^{\Delta}$ performs the “same”
computation as $Q$, but between streams of changes $\Delta s$ and $\Delta o$.
This is the diagram from the introduction, substituting $\Delta s$ for the
transaction stream $T$, and $o$ for the stream of view versions $V$.

Notice that our definition of incremental computation is meaningful only for
_streaming_ computations; this is in contrast to classic definitions, e.g.
(gupta-idb95, ) which consider only one change. Generalizing the definition to
operate on streams gives us additional power, especially when operating with
recursive queries.

The following proposition is one of our central results.

######  Proposition 3.2:

(Properties of the incremental version): For computations of appropriate
types, the following hold:

inversion::

$Q\mapsto{Q}^{\Delta}$ is bijective; its inverse is $Q\mapsto\mathcal{I}\circ
Q\circ\mathcal{D}$.

invariance::

${+}^{\Delta}=+,{(z^{-1})}^{\Delta}=z^{-1},{-}^{\Delta}=-,{\mathcal{I}}^{\Delta}=\mathcal{I},{\mathcal{D}}^{\Delta}=\mathcal{D}$

push/pull::

$Q\circ\mathcal{I}=\mathcal{I}\circ{Q}^{\Delta}$; $\mathcal{D}\circ
Q={Q}^{\Delta}\circ\mathcal{D}$

chain::

${(Q_{1}\circ Q_{2})}^{\Delta}={Q_{1}}^{\Delta}\circ{Q_{2}}^{\Delta}$ (This
generalizes to operators with multiple inputs.)

add::

${(Q_{1}+Q_{2})}^{\Delta}={Q_{1}}^{\Delta}+{Q_{2}}^{\Delta}$

cycle::

${(\lambda s.\mbox{fix}\,\alpha.T(s,z^{-1}(\alpha)))}^{\Delta}=\lambda
s.\mbox{fix}\,\alpha.{T}^{\Delta}(s,z^{-1}(\alpha))$

The proof of these properties relies on elementary algebraic manipulations.
Despite their simplicity, they are very useful. For example, the chain rule
states that the following two circuits are equivalent:

$i$$\mathcal{I}$$Q_{1}$$Q_{2}$$\mathcal{D}$$o$ |  $\cong$ |  $i$${Q_{1}}^{\Delta}$${Q_{2}}^{\Delta}$$o$
---|---|---

In other words, to incrementalize a composite query you can incrementalize
each sub-query independently. This gives us a simple deterministic recipe for
computing the incremental version of an arbitrarily complex query.

We illustrate by giving the proof of the chain rule, which is trivial, and is
based on function composition associativity:

$\displaystyle{(Q_{1}\circ Q_{2})}^{\Delta}$ $\displaystyle=\mathcal{D}\circ
Q_{1}\circ Q_{2}\circ\mathcal{I}$ $\displaystyle=\mathcal{D}\circ
Q_{1}\circ(\mathcal{I}\circ\mathcal{D})\circ Q_{2}\circ\mathcal{I}$
$\displaystyle=(\mathcal{D}\circ Q_{1}\circ\mathcal{I})\circ(\mathcal{D}\circ
Q_{2}\circ\mathcal{I})$
$\displaystyle={(Q_{1})}^{\Delta}\circ{(Q_{2})}^{\Delta}.$

The cycle rule states that the following circuits are equivalent:

$s$$\mathcal{I}$$T$$\mathcal{D}$$o$$z^{-1}$ |  $\cong$ |  $s$${T}^{\Delta}$$o$$z^{-1}$
---|---|---

The incremental version of a feedback loop around a query is just the feedback
loop with the incremental query. The significance of this result will be
apparent when we implement recursive queries.

To execute incremental queries efficiently, we want to compute directly on
streams of changes without integrating them. The invariance property above
shows that stream operators $+$, $-$, and $z^{-1}$ are identical to their
incremental versions, thus $\mathcal{I}$ and $\mathcal{D}$ can be omitted for
them: ${Q}^{\Delta}=\mathcal{I}\circ Q\circ\mathcal{D}=Q$. The following
theorems generalize this to linear and bi-linear operators:

######  Theorem 3.3 (Linear):

For an LTI operator $Q$ we have ${Q}^{\Delta}=Q$.

######  Theorem 3.4 (Bilinear):

For a bilinear time-invariant operator $\times$ we have ${(a\times
b)}^{\Delta}~{}=~{}a\times b~{}+~{}z^{-1}(\mathcal{I}(a))\times
b~{}+~{}a\times z^{-1}(\mathcal{I}(b))$.

By rewriting this statement using $\Delta a$ for the stream of changes to $a$
we get the familiar formula for incremental equi-joins: $\Delta(a\times
b)=\Delta a\times\Delta b+a\times(\Delta b)+(\Delta a)\times b$.

This should not be surprising because equi-joins are bilinear, as we discuss
in the next section.

##  4\. Incremental View Maintenance

Results in §2 and §3 apply to streams of arbitrary group values. In this
section we turn our attention to using these results in the context of
relational view maintenance. As explained in the introduction, we want to
efficiently compute the incremental version of any relational query $Q$ that
updates a database view.

However, we face a technical problem: the $\mathcal{I}$ and $\mathcal{D}$
operators were defined on abelian groups, and relational databases in general
are not abelian groups, since they operate on sets. Fortunately, there is a
well-known tool in the database literature which converts set operations into
group operations by using $\mathbb{Z}$-sets (also called z-relations (green-
pods07, )) instead of sets.

We start by defining the $\mathbb{Z}$-sets group, and then we review how
relational queries are converted into DBSP circuits over $\mathbb{Z}$-sets.
What makes this translation efficiently incrementalizable is the fact that
many basic relational queries can be expressed using LTI $\mathbb{Z}$-set
operators.

###  4.1. $\mathbb{Z}$-sets as an abelian group

Given a set $A$, we define $\mathbb{Z}$-sets222Also called
$\mathbb{Z}$-relations elsewhere (green-tcs11, ), because often A is a
Cartesian product in practice; however, we only need the set structure for
most of our results. over $A$ as functions with _finite support_ from $A$ to
$\mathbb{Z}$. These are functions $f:A\rightarrow\mathbb{Z}$ where
$f(x)\not=0$ for at most a finite number of values $x\in A$. We also write
$\mathbb{Z}[A]$ for the type of $\mathbb{Z}$-sets with elements from $A$.
Values in $\mathbb{Z}[A]$ can be thought of as key-value maps with keys in $A$
and values in $\mathbb{Z}$, justifying the array indexing notation. We write
$f[a]$ instead of $f(a)$. Since $\mathbb{Z}$ is an abelian ring,
$\mathbb{Z}[A]$ is also an abelian ring (and thus a group). This group
$(\mathbb{Z}[A],+_{\mathbb{Z}[A]},0_{\mathbb{Z}[A]},-_{\mathbb{Z}{A}})$ has
addition and subtraction defined pointwise:
$(f+_{\mathbb{Z}[A]}g)(x)=f(x)+g(x).\forall x\in A.$ The $0$ element of
$\mathbb{Z}[A]$ is the function $0_{\mathbb{Z}[A]}$ defined by
$0_{\mathbb{Z}[A]}(x)=0.\forall x\in A$.

A particular $\mathbb{Z}$-set $m\in\mathbb{Z}[A]$ can be denoted by
enumerating the inputs that map to non-zero values and their corresponding
values: $m=\\{x_{1}\mapsto w_{1},\dots,x_{n}\mapsto w_{n}\\}$. We call
$w_{i}\in Z$ the multiplicity (or weight) of $x_{i}\in A$. Multiplicities can
be negative. We write that $x\in m$ for $x\in A$, iff $m[x]\not=0$.

For example, let’s consider a concrete $\mathbb{Z}$-set
$R\in\mathbb{Z}[\texttt{string}]$, defined by $R=\\{\texttt{joe}\mapsto
1,\texttt{anne}\mapsto-1\\}$. $R$ has two elements in its domain, joe with a
multiplicity of 1 (so $R[\texttt{joe}]=1$), and anne with a multiplicity of
$-1$. We say joe $\in R$ and anne $\in R$.

$\mathbb{Z}$-sets generalize sets and bags. A set with elements from $A$ can
be represented as a $\mathbb{Z}$-set by associating a weight of 1 with each
set element. When translating queries on sets to DBSP programs we convert the
data values back and forth between sets and $\mathbb{Z}$-sets.

######  Definition 4.1:

We say that a $\mathbb{Z}$-set represents a set if the multiplicity of every
element is one. We define a function to check this property
$\mbox{isset}:\mathbb{Z}[A]\rightarrow\mathbb{B}$ given by:

$\mbox{isset}(m)\stackrel{{\scriptstyle\textrm{\scriptsize
def}}}{{=}}\left\\{\begin{array}[]{ll}\mbox{true}&\mbox{ if }m[x]=1,\forall
x\in m\\\ \mbox{false}&\mbox{ otherwise}\end{array}\right.$

For our example $\mbox{isset}(R)=\mbox{false}$, since $R[\texttt{anne}]=-1$.

######  Definition 4.2:

We say that a $\mathbb{Z}$-set is positive (or a bag) if the multiplicity of
every element is positive. We define a function to check this property
$\mbox{ispositive}:\mathbb{Z}[A]\rightarrow\mathbb{B}$. given by

$\mbox{ispositive}(m)\stackrel{{\scriptstyle\textrm{\scriptsize
def}}}{{=}}\left\\{\begin{array}[]{ll}\mbox{true}&\mbox{ if }m[x]\geq
0,\forall x\in A\\\ \mbox{false}&\mbox{ otherwise}\end{array}\right.$

$\forall m\in\mathbb{Z}[A].\mbox{isset}(m)\Rightarrow\mbox{ispositive}(m)$.

We have $\mbox{ispositive}(R)=\mbox{false}$, since $R[\texttt{anne}]=-1$.

We write $m\geq 0$ when $m$ is positive. For positive $m,n$ we write $m\geq n$
for $m,n\in\mathbb{Z}[A]$ iff $m-n\geq 0$. $\geq$ is a partial order.

We call a function $f:\mathbb{Z}[A]\rightarrow\mathbb{Z}[B]$ positive if it
maps positive values to positive values: $\forall x\in\mathbb{Z}[A],x\geq
0_{\mathbb{Z}[A]}\Rightarrow f(x)\geq 0_{\mathbb{Z}[B]}$. We apply this
notation to functions as well: $\mbox{ispositive}(f)$.

######  Definition 4.3 (distinct):

The function $\mathit{distinct}:\mathbb{Z}[A]\rightarrow\mathbb{Z}[A]$
projects a $\mathbb{Z}$-set into an underlying set (but _the result is still a
$\mathbb{Z}$-set_):

$\mathit{distinct}(m)[x]\stackrel{{\scriptstyle\textrm{\scriptsize
def}}}{{=}}\left\\{\begin{array}[]{ll}1&\mbox{ if }m[x]>0\\\ 0&\mbox{
otherwise}\end{array}\right.$

$\mathit{distinct}$ “removes” elements with negative multiplicities.
$\mathit{distinct}(R)=\\{\texttt{joe}\mapsto 1\\}$.

While very simple, this definition of $\mathit{distinct}$ has been carefully
chosen to enable us to define precisely all relational (set) operators from
$\mathbb{Z}$-sets operators.

Circuits derived from relational queries only compute on positive
$\mathbb{Z}$-sets; negative values will only be used to represent _changes_ to
$\mathbb{Z}$-sets. Negative weights “remove” elements from a set.

All the results from §2 extend to streams over $\mathbb{Z}$-sets.

######  Definition 4.4:

(mononotonicity) A stream $s\in\mathcal{S}_{\mathbb{Z}[A]}$ is positive if
every value of the stream is positive: $s[t]\geq 0.\forall t\in\mathbb{N}$. A
stream $s\in\mathcal{S}_{\mathbb{Z}[A]}$ is monotone if $s[t]\geq
s[t-1],\forall t\in\mathbb{N}$.

If $s\in\mathcal{S}_{\mathbb{Z}[A]}$ is positive, then $\mathcal{I}(s)$ is
monotone. If $s\in\mathcal{S}_{\mathbb{Z}[A]}$ is monotone, $\mathcal{D}(s)$
is positive.

##### Generalizing box-and-arrow diagrams

From now on we will use circuits to compute both on scalars and streams. We
use the same graphical representation for functions on streams or scalars:
boxes with input and output arrows. For scalar functions the “values” of the
arrows are scalars instead of streams; otherwise the interpretation of boxes
as function application is unchanged.

###  4.2. Implementing relational operators

The fact that relational algebra can be implemented by computations on
$\mathbb{Z}$-sets has been shown before, e.g. (green-pods07, ). The
translation of all the core relational operators is shown in Table 4.2. The
translation is essentially given by induction on the query structure.

Operation |  SQL example |  DBSP circuit |  Details
---|---|---|---
Composition |  ⬇ SELECT DISTINCT ... FROM (SELECT ... FROM ...) |  I$C_{I}$$C_{O}$O |  $C_{I}$ circuit for inner query, $C_{O}$ circuit for outer query.
Union |  ⬇ (SELECT * FROM I1) UNION (SELECT * FROM I2) |  I1I2$+$$\mathit{distinct}$O |
Projection |  ⬇ SELECT DISTINCT I.c FROM I |  I$\pi$$\mathit{distinct}$O |  $\pi(i)[y]\stackrel{{\scriptstyle\textrm{\scriptsize def}}}{{=}}\sum_{x\in i,x|_{c}=y}i[x]$ $x|_{c}$ is projection on column $c$ of the tuple $x$ $\pi$ is linear; $\mbox{ispositive}(\pi),\mbox{zpp}(\pi)$.
Filtering |  ⬇ SELECT * FROM I WHERE p(I.c) |  I$\sigma_{P}$$\mathit{distinct}$O |  $\sigma_{P}(m)[x]\stackrel{{\scriptstyle\textrm{\scriptsize def}}}{{=}}\left\\{\begin{array}[]{ll}m[x]\cdot x&\mbox{ if }P(x)\\\ 0&\mbox{ otherwise }\\\ \end{array}\right.$ $P:A\rightarrow\mathbb{B}$ is a predicate. $\sigma_{P}$ is linear; $\mbox{ispositive}(\sigma_{P}),\mbox{zpp}(\sigma_{P})$.
Selection |  ⬇ SELECT DISTINCT f(I.c, ...) FROM I |  I$\mbox{map}(f)$$\mathit{distinct}$O |  For a function $f$ $\mbox{map}(f)$ is linear, $\mbox{ispositive}(\mbox{map}(f)),\mbox{zpp}(\mbox{map}(f))$ .
Cartesian product |  ⬇ SELECT I1.*, I2.* FROM I1, I2 |  I1I2$\times$O |  $(a\times b)((x,y))\stackrel{{\scriptstyle\textrm{\scriptsize def}}}{{=}}a[x]\times b[y]$. $\times$ is bilinear, $\mbox{ispositive}(\times),\mbox{zpp}(\times)$.
Equi-join |  ⬇ SELECT I1.*, I2.* FROM I1 JOIN I2 ON I1.c1 = I2.c2 |  I1I2$\bowtie$O |  $(a\bowtie b)((x,y))\stackrel{{\scriptstyle\textrm{\scriptsize def}}}{{=}}a[x]\times b[y]\\\ \mbox{ if }x|_{c1}=y|_{c2}$. $\bowtie$ is bilinear, $\mbox{ispositive}(\bowtie),\mbox{zpp}(\bowtie)$.
Intersection |  ⬇ (SELECT * FROM I1) INTERSECT (SELECT * FROM I2) |  I1I2$\bowtie$O |  Special case of equi-join when both relations have the same schema.
Difference |  ⬇ SELECT * FROM I1 EXCEPT SELECT * FROM I2 |  I1I2$-$$+$$\mathit{distinct}$O |

Table 1. Implementation of SQL relational set operators in DBSP. Each query
assumes that inputs I, I1, I2, are sets and it produces output sets.

The translation is fairly straightforward, but many operators require the
application of a $\mathit{distinct}$ to produce sets. The correctness of this
implementation is predicated on the global circuit inputs being sets as well.
For example, $a\cup b=\mathit{distinct}(a+b)$, $a\setminus
b=\mathit{distinct}(a-b)$, $(a\times b)((x,y))=a[x]\times b[y]$.

####  4.2.1. Correctness of the DBSP implementations

A relational query $Q$ that transforms a set $V$ into a set $U$ will be
implemented by a DBSP computation $Q^{\prime}$ on $\mathbb{Z}$-sets. The
correctness of the implementation requires that the following diagram
commutes:

$V$$VZ$$U$$UZ$$Q$tozset$Q^{\prime}$toset

The toset and tozset functions convert sets to $\mathbb{Z}$-sets and vice-
versa:

$\mbox{toset}:\mathbb{Z}[A]\to 2^{A}$ is defined by
$\mbox{toset}(m)\stackrel{{\scriptstyle\textrm{\scriptsize
def}}}{{=}}\cup_{x\in\mathit{distinct}(m)}\\{x\\}$.

$\mbox{tozset}:2^{A}\to\mathbb{Z}[A]$ is defined by
$\mbox{tozset}(s)\stackrel{{\scriptstyle\textrm{\scriptsize
def}}}{{=}}\sum_{x\in s}1\cdot x$.

All standard algebraic properties of the relational operators can be used to
optimize circuits (they can even be applied to queries before building the
circuits).

Notice that the use of the $\mathit{distinct}$ operator allows DBSP to model
the _full relational algebra_ , including difference (and not just the
positive fragment). Most of the operators that appear in the circuits in Table
4.2 are linear, and thus have very efficient incremental versions. A notable
exception is $\mathit{distinct}$. While we show below that $\mathit{distinct}$
can be computed efficiently incrementally, it does have an important cost in
terms of memory, so we try to minimize its use. For this we can use a pair of
optimizations:

######  Proposition 4.5:

Let Q be one of the following $\mathbb{Z}$-sets operators: filtering $\sigma$,
join $\bowtie$, or Cartesian product $\times$. Then we have $\forall
i\in\mathbb{Z}[I],\mbox{ispositive}(i)\Rightarrow
Q(\mathit{distinct}(i))=\mathit{distinct}(Q(i))$.

This rule allows us to delay the application of $\mathit{distinct}$.

######  Proposition 4.6:

Let Q be one of the following $\mathbb{Z}$-sets operators: filtering $\sigma$,
projection $\pi$, selection (map($f$)), addition $+$, join $\bowtie$, or
Cartesian product $\times$. Then we have $\forall
i\in\mathbb{Z}[I],\mbox{ispositive}(i)\Rightarrow\mathit{distinct}(Q(\mathit{distinct}(i)))=\mathit{distinct}(Q(i))$.

This is Proposition 6.13 in (green-tcs11, ).

These properties allow us to “consolidate” distinct operators by performing
one $\mathit{distinct}$ at the end of a chain of computations.

Finally, the next proposition shows that the incremental of
$\mathit{distinct}$ can be computed with work proportional to the size of the
input change.

######  Proposition 4.7:

The following circuit implements ${({\uparrow}\mathit{distinct})}^{\Delta}$:
$d$${({\uparrow}\mathit{distinct})}^{\Delta}$$o$ $\cong$
$d$$\mathcal{I}$$z^{-1}$${\uparrow}H$$o$$i$

where $H:\mathbb{Z}[A]\times\mathbb{Z}[A]\to\mathbb{Z}[A]$ is defined as:
H(i, d)[x] =def{-1 if  i[x] ¿ 0  and  (i + d)[x] ≤01 if  i[x] ≤0  and  (i +
d)[x] ¿ 00 otherwise

The function $H$ detects whether the multiplicity of an element in the input
set $i$ when adding change $d$ is changing from negative to positive or vice-
versa. Notice that only multiplicities of the elements that appear in the
change $d$ can change from input to output, so the work needed to compute both
$H$ is bounded by the size of $d$ and not $i$.

###  4.3. Incremental view maintenance

Let us consider a relational query $Q$ defining a view. To create a circuit
that maintains incrementally the view defined by $Q$ we apply the following
mechanical steps; this algorithm is deterministic and its running time is
proportional to the complexity of the query (number of operators in the
query):

######  Algorithm 4.8 (incremental view maintenance):

  1. (1)

Translate $Q$ into a circuit using the rules in Table 4.2.

  2. (2)

Apply $\mathit{distinct}$ consolidation until convergence.

  3. (3)

Lift the whole circuit, by applying Proposition 2.4, converting it to a
circuit operating on streams.

  4. (4)

Incrementalize the whole circuit “surrounding” it with $\mathcal{I}$ and
$\mathcal{D}$.

  5. (5)

Apply the chain rule and other properties of the ${\cdot}^{\Delta}$ operator
from Proposition 3.2 recursively on the query structure to optimize the
incremental implementation.

Step (3) yields a circuit that consumes a stream of complete database
snapshots and outputs a stream of complete view snapshots. Step (4) yields a
circuit that consumes a stream of changes to the database and outputs a stream
of view changes; however, the internal operation of the circuit is non-
incremental, as it computes on the complete state of the database
reconstructed by the integration operator. Step (5) incrementalizes the
internals of the circuit by rewriting it to compute on changes, avoiding
integration when possible (see §3).

###  4.4. Example

In this section we apply the incremental view maintenance algorithm to a
concrete query. Let us consider the following query:

⬇

CREATE VIEW v AS

SELECT DISTINCT t1.x, t2.y FROM (

SELECT t1.x, t1.id

FROM t

WHERE t.a > 2

) t1

JOIN (

SELECT t2.id, t2.y

FROM r

WHERE r.s > 5

) t2 ON t1.id = t2.id

Step 1: First we create a DBSP circuit to represent this query using the
translation rules from Table 4.2:

t1$\sigma_{a>2}$$\mathit{distinct}$$\pi_{x,d}$$\mathit{distinct}$t2$\sigma_{s>5}$$\mathit{distinct}$$\pi_{y,id}$$\mathit{distinct}$$\bowtie_{id=id}$$\pi_{x,y}$$\mathit{distinct}$V

Step 2: we apply the $\mathit{distinct}$ optimization rules; first the rule
from 4.6 gives us the following equivalent circuit:

t1$\sigma_{a>2}$$\pi_{x,d}$$\mathit{distinct}$t2$\sigma_{s>5}$$\pi_{y,id}$$\mathit{distinct}$$\bowtie_{id=id}$$\pi_{x,y}$$\mathit{distinct}$V

Applying the rule from 4.5 we get:

t1$\sigma_{a>2}$$\pi_{x,d}$t2$\sigma_{s>5}$$\pi_{y,id}$$\bowtie_{id=id}$$\mathit{distinct}$$\pi_{x,y}$$\mathit{distinct}$V

And applying again 4.6 we get:

t1$\sigma_{a>2}$$\pi_{x,d}$t2$\sigma_{s>5}$$\pi_{y,id}$$\bowtie_{id=id}$$\pi_{x,y}$$\mathit{distinct}$V

Step 3: we lift the circuit using distributivity of composition over lifting;
we obtain a circuit that computes over streams, i.e., for each new input pair
of relations t1 and t2 it will produce an output view V:

t1${\uparrow}\sigma_{a>2}$${\uparrow}\pi_{x,d}$t2${\uparrow}\sigma_{s>5}$${\uparrow}\pi_{y,id}$${\uparrow}\bowtie_{id=id}$${\uparrow}\pi_{x,y}$${\uparrow}\mathit{distinct}$V

Step 4: incrementalize circuit, obtaining a circuit that computes over
changes; this circuit receives changes to relations t1 and t2 and for each
such change it produces the corresponding change in the output view V:

$\Delta$t1$\mathcal{I}$${\uparrow}\sigma_{a>2}$${\uparrow}\pi_{x,d}$$\Delta$t2$\mathcal{I}$${\uparrow}\sigma_{s>5}$${\uparrow}\pi_{y,id}$${\uparrow}\bowtie_{id=id}$${\uparrow}\pi_{x,y}$${\uparrow}\mathit{distinct}$$\mathcal{D}$$\Delta$V

Step 5: apply the chain rule to rewrite the circuit as a composition of
incremental operators;

$\Delta$t1${({\uparrow}\sigma_{a>2})}^{\Delta}$${({\uparrow}\pi_{x,d})}^{\Delta}$$\Delta$t2${({\uparrow}\sigma_{s>5})}^{\Delta}$${({\uparrow}\pi_{y,id})}^{\Delta}$${({\uparrow}\bowtie_{id=id})}^{\Delta}$${({\uparrow}\pi_{x,y})}^{\Delta}$${({\uparrow}\mathit{distinct})}^{\Delta}$$\Delta$V

Use the linearity of $\sigma$ and $\pi$ to simplify this circuit:

$\Delta$t1${\uparrow}\sigma_{a>2}$${\uparrow}\pi_{x,d}$$\Delta$t2${\uparrow}\sigma_{s>5}$${\uparrow}\pi_{y,id}$${({\uparrow}\bowtie_{id=id})}^{\Delta}$${\uparrow}\pi_{x,y}$${({\uparrow}\mathit{distinct})}^{\Delta}$$\Delta$V

Finally, replace the incremental join using the formula for bilinear operators
(Theorem 3.4), and the incremental $\mathit{distinct}$ (Proposition 4.7),
obtaining the circuit from Figure 1.

$\Delta$t1${\uparrow}\sigma_{a>2}$${\uparrow}\pi_{x,d}$$\Delta$t2${\uparrow}\sigma_{s>5}$${\uparrow}\pi_{y,id}$$\mathcal{I}$${\uparrow}\bowtie_{id=id}$$\mathcal{I}$$z^{-1}$$z^{-1}$${\uparrow}\bowtie_{id=id}$${\uparrow}\bowtie_{id=id}$$+$${\uparrow}\pi_{x,y}$$\mathcal{I}$$z^{-1}$${\uparrow}H$$\Delta$V
Figure 1. Final version of the incremental query circuit from §4.4.

Notice that the resulting circuit contains three integration operations: two
from the join, and one from the $\mathit{distinct}$. It also contains three
join operators. However, the work performed by each operator for each new
input is proportional to the size of change, as we argue in the following
section.

###  4.5. Complexity

Incremental circuits are efficient. The work performed (and the memory used)
by a circuit is the sum of the work performed (and memory used) by its
operators. We argue that each operator in the incremental version of a circuit
is efficient.

For incrementalized circuits the input stream of each operator contains
_changes_ in its input relations. Denote
$C[t]\stackrel{{\scriptstyle\textrm{\scriptsize def}}}{{=}}\|s[t]\|$ the size
of the value of stream $s$ of changes at time $t$, and
$R[t]\stackrel{{\scriptstyle\textrm{\scriptsize
def}}}{{=}}\|\mathcal{I}(s)[t]\|$ the size of the relation produced by
integrating all changes in $s$. An unoptimized incremental operator
${Q}^{\Delta}=\mathcal{D}\circ Q\circ\mathcal{I}$ evaluates query $Q$ on the
integration of its input streams; hence its time complexity is the same as
that of the non-incremental operator, a function of $R[t]$. In addition,
because of the $\mathcal{I}$ and $\mathcal{D}$ operators, it uses $O(R[t])$
memory.

The optimizations described in §3 reduce the reduce the time complexity of an
operator to be a function of $C[t]$. Assuming $C[t]\ll R[t]$, this translates
to major performance improvements in practice. For example, Theorem 3.3,
allows evaluating ${T}^{\Delta}$, where $T$ is a linear operator, in time
$O(C[t])$. Interestingly, while the $\mathcal{I}$ operator uses $O(R[t])$
memory, it can be evaluated in $O(C[t])$ time, because all values that appear
in the output at time $t$ must be present in current input change for time
$t$. Similarly, while the $\mathit{distinct}$ operator is not linear,
${({\uparrow}\mathit{distinct})}^{\Delta}$ can also be evaluated in $O(C[t])$
according to Proposition 4.7. Bilinear operators, including join, can be
evaluated in time proportional to the product of the sizes of their input
changes $O(C[t]^{2})$ (Theorem 3.4).

The space complexity of linear operators is 0 (zero), since they store no data
persistently. The space complexity of
${({\uparrow}\mathit{distinct})}^{\Delta}$ and join is $O(R[t])$.

##  5\. Recursive queries

Recursive queries are very useful in a many applications. For example, many
graph algorithms (such as graph reachability or transitive closure) are
naturally expressed using recursive queries.

We introduce two new stream operators that are instrumental in expressing
recursive query evaluation. These operators allow us to build circuits
implementing looping constructs, which are used to iterate computations until
a fixed-point is reached.

######  Definition 5.1:

We say that a stream $s\in\mathcal{S}_{A}$ is zero almost-everywhere if it has
a finite number of non-zero values, i.e., there exists a time
$t_{0}\in\mathbb{N}$ s.t. $\forall t\geq t_{0}.s[t]=0$. Denote the set of
streams that are zero almost everywhere by $\overline{\mathcal{S}_{A}}$.

##### Stream introduction

The delta function (named from the Dirac delta function)
$\delta_{0}:A\rightarrow\mathcal{S}_{A}$ produces a stream from a scalar
value: δ_0(v)[t] =def{ vif  t = 00_A otherwise For example, $\delta_{0}(5)$ is
the stream
$\setsepchar{}\readlist\arg{50000}{[}\begin{array}[]{cccccc}\arg[1]&\arg[2]&\arg[3]&\arg[4]&\arg[5]&\cdots\end{array}{]}$.

##### Stream elimination

We define the function $\int:\overline{\mathcal{S}_{A}}\rightarrow A$, over
streams that are zero almost everywhere, as
$\int(s)\stackrel{{\scriptstyle\textrm{\scriptsize def}}}{{=}}\sum_{t\geq
0}s[t]$. $\int$ is closely related to $\mathcal{I}$; if $\mathcal{I}$ is the
indefinite integral, $\int$ is the definite integral on the interval
$0-\infty$.

For many implementation strategies (including relational and Datalog queries
given below) the $\int$ operator can be approximated finitely and precisely by
integrating until the first 0 value encountered, since it can be proven that
its input is always the derivative of a monotone stream.

$\delta_{0}$ is the left inverse of $\int$, i.e.:
$\int\circ\;\delta_{0}=\mathit{id}_{A}$.

######  Proposition 5.2:

$\delta_{0}$ and $\int$ are LTI.

##### Nested time domains

So far we used a tacit assumption that “time” is common for all streams in a
program. For example, when we add two streams, we assume that they use the
same “clock” for the time dimension. However, the $\delta_{0}$ operator
creates a stream with a “new”, independent time dimension. We require _well-
formed circuits_ to “insulate” such nested time domains by nesting them
between a $\delta_{0}$ and an $\int$ operator:

$i$$\delta_{0}$$Q$$\int$$o$

######  Proposition 5.3:

If $Q$ is time-invariant, the circuit above has the zero-preservation
property: $\mbox{zpp}(\int\circ\;Q\circ\delta_{o})$.

###  5.1. Implementing Recursive Datalog

We illustrate the implementation of recursive queries in DBSP for stratified
Datalog. Datalog is strictly more expressive than relational algebra since it
can express recursive programs, e.g.:

⬇

O(v) :- I(v). // base case

O(v) :- I(z), O(x), v = ... . // rec case

In general, a recursive Datalog program defines a set of mutually recursive
relations $O_{1},..,O_{n}$ as an equation
$(O_{1},..,O_{n})=R(I_{1},..,I_{m},O_{1},..,O_{n})$, where $I_{1},..,I_{m}$
are input relations and $R$ is a relational (non-recursive) query.

The following algorithm generates a circuit that computes a solution to this
equation. We describe the algorithm informally and for the special case of a
single input $I$ and single output $O$; the general case can be found in the
companion technical report (tr, ), and is only slightly more involved.

  1. (1)

Implement the non-recursive relational query $R$ as described in §4 and Table
4.2; this produces an acyclic circuit whose inputs and outputs are a
$\mathbb{Z}$-set (i.e., not a stream):

IO$R$O

  2. (2)

Lift this circuit to operate on streams and connect the output to the input in
a feedback cycle as follows:

I$\delta_{0}$$\mathcal{I}$${\uparrow}R$${\uparrow}\mathit{distinct}$$\mathcal{D}$$\int$O$o$$z^{-1}$

We construct ${\uparrow}R$ by lifting each operator of the circuit
individually according to Proposition 2.4.

The inner loop of the circuit computes the fixed point of $R$. The
differentiation operator $\mathcal{D}$ yields the set of new Datalog facts
(changes) computed by each iteration of the loop. When the set of new facts
becomes empty the iterations have completed. $\int$ computes the value of the
fixed point by aggregating these changes.

######  Theorem 5.4 (Recursion correctness):

If $\mbox{isset}(\mbox{{I}})$, the output of the circuit above is the relation
O as defined by the Datalog semantics as a function of the input relation I.

###### Proof.

Let us compute the contents of the $o$ stream, produced at the output of the
$\mathit{distinct}$ operator. We will show that this stream is composed of
increasing approximations of the value of O.

We define the following one-argument function: $S(x)=\lambda
x.R(\mbox{{I}},x)$. Notice that the left input of the ${\uparrow}R$ block is a
constant stream with the value I. Due to the stratified nature of the
language, we must have $\mbox{ispositive}(S)$, so $\forall x.S(x)\geq x$. Also
${\uparrow}S$ is time-invariant, so $S(0)=0$. From §4, the definition of set
union we know that $x\cup y=\mathit{distinct}(x+y)$. We get the following
system of equations:  o[0] =S(0) o[t] =S(o[t-1])  So, by induction on $t$ we
have $o[t]=S^{t}(0)$, where by $S^{t}$ we mean $\underbrace{S\circ
S\circ\ldots\circ S}_{t}$. $S$ is monotone; thus, if there is a time $k$ such
that $S^{k}(0)=S^{k+1}(0)$, we have $\forall
j\in\mathbb{N}.S^{k+j}(0)=S^{k}(0)$. Applying a derivative to this stream will
then produce a stream that is zero almost everywhere, and integrating this
derivative will return the last distinct value in the stream $o$.

This is essentially the definition of the semantics of a recursive Datalog
relation: $\mbox{{O}}=\mbox{fix}\,x.R(\mbox{{I}},x)$. ∎

Note that the use of unbounded data domains (like integers with arithmetic)
does not guarantee convergence for all programs.

In fact, this circuit implements the standard naïve evaluation algorithm
(e.g., see Algorithm 1 in (greco-sldm15, )). Notice that the inner part of the
circuit is the incremental form of another circuit, since it is sandwiched
between $\mathcal{I}$ and $\mathcal{D}$ operators. Using the cycle rule of
Proposition 3.2 we can rewrite this circuit as:

(5.1)
I$\delta_{0}$${({\uparrow}R)}^{\Delta}$${{\uparrow}(\mathit{distinct})}^{\Delta}$$\int$O$z^{-1}$

This last circuit effectively implements the semi-naïve evaluation algorithm
(Algorithm 2 from (greco-sldm15, )). The correctness of semi-naïve evaluation
is an immediate consequence of the cycle rule.

##  6\. Incremental recursive programs

In §2–4 we showed how to incrementalize a relational query by compiling it
into a circuit, lifting the circuit to compute on streams, and applying the
${\cdot}^{\Delta}$ operator to the lifted circuit. In §5.1 we showed how to
compile a recursive query into a circuit that employs incremental computation
internally to compute the fixed point. Here we combine these results to
construct a circuit that evaluates a _recursive query incrementally_. The
circuit receives a stream of updates to input relations, and for every update
recomputes the fixed point. To do this incrementally, it preserves the stream
of changes to recursive relations produced by the iterative fixed point
computation, and adjusts this stream to account for the modified inputs. Thus,
every element of the input stream yields a stream of adjustments to the fixed
point computation, using _nested streams_.

Nested streams, or streams of streams,
$\mathcal{S}_{\mathcal{S}_{A}}=\mathbb{N}\rightarrow(\mathbb{N}\rightarrow
A)$, are well defined, since streams form an abelian group. Equivalently, a
nested stream is a value in $\mathbb{N}\times\mathbb{N}\to A$, i.e., a matrix
with an infinite number of rows, indexed by two-dimensional time
$(t_{0},t_{1})$. where each row is a stream. In §A.1 we show a few example
nested stream computations.

Lifting a stream operator $S:\mathcal{S}_{A}\to\mathcal{S}_{B}$ yields an
operator over nested streams
${\uparrow}S:\mathcal{S}_{\mathcal{S}_{A}}\to\mathcal{S}_{\mathcal{S}_{B}}$,
such that $({\uparrow}S)(s)=S\circ s$, or, pointwise:
$({\uparrow}S(s))[t_{0}][t_{1}]=S(s[t_{0}])[t_{1}],\forall
t_{0},t_{1}\in\mathbb{N}$. In particular, a scalar function $f:A\rightarrow B$
can be lifted twice to produce an operator between streams of streams:
${\uparrow}{\uparrow}f:\mathcal{S}_{\mathcal{S}_{A}}\rightarrow\mathcal{S}_{\mathcal{S}_{B}}$.

We define a partial order over timestamps: $(i_{0},i_{1})\leq(t_{0},t_{1})$
iff $i_{0}\leq t_{0}$ and $i_{1}\leq t_{1}$. We extend the definition of
strictness for operators over nested streams: a stream operator
$F:\mathcal{S}_{\mathcal{S}_{A}}\to\mathcal{S}_{\mathcal{S}_{B}}$ is strict if
for any $s,s^{\prime}\in\mathcal{S}_{\mathcal{S}_{A}}$ and all times
$t,i\in\mathbb{N}\times\mathbb{N}$ we have $\forall i<t,s[i]=s^{\prime}[i]$
implies $F(s)[t]=F(s^{\prime})[t]$. Proposition 2.10 holds for this notion of
strictness, i.e., the fixed point operator $\mbox{fix}\,\alpha.F(\alpha)$ is
well defined for a strict operator $F$.

######  Proposition 6.1:

The operator
${\uparrow}z^{-1}:\mathcal{S}_{\mathcal{S}_{A}}\to\mathcal{S}_{\mathcal{S}_{A}}$
is strict.

The operator $z^{-1}$ on nested streams delays “rows” of the matrix, while
${\uparrow}z^{-1}$ delays “columns”. (See examples in §A.1).

The $\mathcal{I}$ operator on $\mathcal{S}_{\mathcal{S}_{A}}$ operates on rows
of the matrix, treating each row as a single value. Lifting a stream operator
computing on $\mathcal{S}_{A}$, such as
$\mathcal{I}:\mathcal{S}_{A}\to\mathcal{S}_{A}$, also produces an operator on
nested streams, but this time computing on the columns of the matrix
${\uparrow}\mathcal{I}:\mathcal{S}_{\mathcal{S}_{A}}\to\mathcal{S}_{\mathcal{S}_{A}}.$

######  Proposition 6.2 (Lifting cycles):

For a binary, causal $T$ we have: ${\uparrow}(\lambda
s.\mbox{fix}\,\alpha.T(s,z^{-1}(\alpha)))=\lambda
s.\mbox{fix}\,\alpha.({\uparrow}T)(s,({\uparrow}z^{-1})(\alpha))$ i.e.,
lifting a circuit containing a “cycle” can be accomplished by lifting all
operators independently, including the $z^{-1}$ back-edge.

This means that lifting a DBSP stream function can be expressed within DBSP
itself. For example, we have:

$i$${\uparrow}\mathcal{I}$$o$ |  $\cong$ |  $i$$+$$o$${\uparrow}z^{-1}$
---|---|---

This proposition gives the ability to lift entire circuits, including circuits
computing on streams and having feedback edges, which are well-defined, due to
Proposition 6.1. With this machinery we can now apply Algorithm 4.8 to
arbitrary circuits, even circuits built for recursively-defined relations.
Consider the “semi-naive” circuit (5.1), and denote $\mathit{distinct}\circ R$
with $T$:

I$\delta_{0}$${({\uparrow}T)}^{\Delta}$$\int$O$z^{-1}$

Lift the entire circuit using Proposition 6.2 and incrementalize it:

I$\mathcal{I}$${\uparrow}\delta_{0}$${\uparrow}{({\uparrow}T)}^{\Delta}$${\uparrow}\int$$\mathcal{D}$O${\uparrow}z^{-1}$

Now apply the chain rule to this circuit:

(6.1)
I${\uparrow}\delta_{0}$${({\uparrow}{({\uparrow}T)}^{\Delta})}^{\Delta}$${\uparrow}\int$O${\uparrow}z^{-1}$

This is the incremental version of an arbitrary recursive query.

###  6.1. Example

In this section we derive the incremental version of a circuit containing
recursion, by applying Algorithm 4.8. We start with a very simple program,
expressed in Datalog, which computes the transitive closure of a directed
graph:

⬇

// Edge relation with head and tail

input relation E(h: Node, t: Node)

// Reach relation with source s and sink t

output relation R(s: Node, t: Node)

R(x, x) :- E(x, _).

R(x, x) :- E(_, x).

R(x, y) :- E(x, y).

R(x, y) :- E(x, z), R(z, y).

We haven’t explained how Datalog is translated to circuits, but most Datalog
operators are relational in nature. Assuming one could write recursive queries
in SQL where a view is defined in terms of itself, the above program would be
implemented by the following (illegal) SQL query:

⬇

CREATE VIEW R AS

(SELECT E.h, E.h FROM E)

UNION

(SELECT E.t, E.t FROM E)

UNION

(SELECT * FROM E)

UNION

(SELECT E.h, R.t

FROM E JOIN R

ON E.t = R.s)

We apply the algorithm from §5.1 to create first the non-recursive circuit, by
assuming that R is already computed as a view R1, and using R1 in the
definition of R instead of itself:

⬇

CREATE VIEW Reach AS

(SELECT E.h, E.h FROM E)

UNION

(SELECT E.t, E.t FROM E)

UNION

(SELECT * FROM E)

UNION

(SELECT E.h, R1.t

FROM E JOIN R ON E.t = R1.s)

Now we implement this query as a DBSP circuit with two inputs E and R1:

ER1$\bowtie_{t=s}$$\pi_{h,t}$$\pi_{h}$$\sigma_{h,h}$$\pi_{t}$$\sigma_{t,t}$$+$$\mathit{distinct}$R

Now lift the circuit by lifting each operator pointwise, and connect it in a
feedback loop by connecting input R1 from the output R through a
$z^{-1}$operator and bracket everything with $\delta_{0}--\int$:

E$\delta_{0}$$\mathcal{I}$${\uparrow}\bowtie_{t=s}$${\uparrow}\pi_{h,t}$${\uparrow}\pi_{h}$${\uparrow}\sigma_{h,h}$${\uparrow}\pi_{t}$${\uparrow}\sigma_{t,t}$$+$${\uparrow}\mathit{distinct}$$\mathcal{D}$$\int$R$z^{-1}$

The above circuit is a complete implementation of the non-streaming recursive
query; given an input relation E it will produce its transitive closure R at
the output.

Now we use the seminaïve property 5.1 to rewrite the circuit:

(To save space in the figures we will omit the indices from $\pi$ and $\sigma$
in the subsequent figures, for example by writing just $\pi$ instead of
$\pi_{h}$.)

E$\delta_{0}$${({\uparrow}\bowtie)}^{\Delta}$${({\uparrow}\pi)}^{\Delta}$${({\uparrow}\pi)}^{\Delta}$${({\uparrow}\sigma)}^{\Delta}$${({\uparrow}\pi)}^{\Delta}$${({\uparrow}\sigma)}^{\Delta}$$+$${({\uparrow}\mathit{distinct})}^{\Delta}$$\int$R$z^{-1}$

Using the linearity of ${\uparrow}\pi$ and ${\uparrow}\sigma$, this can be
rewritten as an equivalent circuit:

E$\delta_{0}$${({\uparrow}\bowtie)}^{\Delta}$${\uparrow}\pi$${\uparrow}\pi$${\uparrow}\sigma$${\uparrow}\pi$${\uparrow}\sigma$$+$${({\uparrow}\mathit{distinct})}^{\Delta}$$\int$R$z^{-1}$

To make this circuit into a streaming computation that evaluates a new
transitive closure for a stream of inputs E, we lift it entirely, using
Proposition 6.2:

E${\uparrow}\delta_{0}$${\uparrow}{({\uparrow}\bowtie)}^{\Delta}$${\uparrow}{\uparrow}\pi$${\uparrow}{\uparrow}\pi$${\uparrow}{\uparrow}\sigma$${\uparrow}{\uparrow}\pi$${\uparrow}{\uparrow}\sigma$$+$${\uparrow}{({\uparrow}\mathit{distinct})}^{\Delta}$${\uparrow}\int$R${\uparrow}z^{-1}$

We convert this circuit into an incremental circuit, which receives in each
transaction the changes to relation E and produces the corresponding changes
to relation R:

$\Delta$E$\mathcal{I}$${\uparrow}\delta_{0}$${\uparrow}{({\uparrow}\bowtie)}^{\Delta}$${\uparrow}{\uparrow}\pi$${\uparrow}{\uparrow}\pi$${\uparrow}{\uparrow}\sigma$${\uparrow}{\uparrow}\pi$${\uparrow}{\uparrow}\sigma$$+$${\uparrow}{({\uparrow}\mathit{distinct})}^{\Delta}$${\uparrow}\int$$\mathcal{D}$$\Delta$R${\uparrow}z^{-1}$

We can now apply again the chain rule to this circuit:

$\Delta$E${({\uparrow}\delta_{0})}^{\Delta}$${({\uparrow}{({\uparrow}\bowtie)}^{\Delta})}^{\Delta}$${({\uparrow}{\uparrow}\pi)}^{\Delta}$${({\uparrow}{\uparrow}\pi)}^{\Delta}$${({\uparrow}{\uparrow}\sigma)}^{\Delta}$${({\uparrow}{\uparrow}\pi)}^{\Delta}$${({\uparrow}{\uparrow}\sigma)}^{\Delta}$$+$${({\uparrow}{({\uparrow}\mathit{distinct})}^{\Delta})}^{\Delta}$${({\uparrow}\int)}^{\Delta}$$\Delta$R${({\uparrow}z^{-1})}^{\Delta}$

We now take advantage of the linearity of ${\uparrow}\delta_{0}$,
${\uparrow}\int$, ${\uparrow}z^{-1}$, ${\uparrow}{\uparrow}\pi$, and
${\uparrow}{\uparrow}\sigma$ to simplify the circuit by removing some
${\cdot}^{\Delta}$ invocations:

$\Delta$E${\uparrow}\delta_{0}$${({\uparrow}{({\uparrow}\bowtie)}^{\Delta})}^{\Delta}$${\uparrow}{\uparrow}\pi$${\uparrow}{\uparrow}\pi$${\uparrow}{\uparrow}\sigma$${\uparrow}{\uparrow}\pi$${\uparrow}{\uparrow}\sigma$$+$${({\uparrow}{({\uparrow}\mathit{distinct})}^{\Delta})}^{\Delta}$${\uparrow}\int$$\Delta$R${\uparrow}z^{-1}$

There are two applications of ${\cdot}^{\Delta}$ left in this circuit:
${({\uparrow}{({\uparrow}\bowtie)}^{\Delta})}^{\Delta}$ and
${({\uparrow}{({\uparrow}\mathit{distinct})}^{\Delta})}^{\Delta}$. We expand
their implementations separately, and we stitch them into the global circuit
at the end. This ability to reason about sub-circuits independently highlights
the modularity of DBSP.

The join is expanded twice, using the bilinearity of ${\uparrow}\bowtie$ and
${\uparrow}{\uparrow}\bowtie$. Let’s start with the inner circuit,
implementing ${({\uparrow}\bowtie)}^{\Delta}$, given by Theorem 3.4:

$a$$b$${({\uparrow}\bowtie)}^{\Delta}$$o$ |  $\cong$ |  $a$${\uparrow}\bowtie$$b$$\mathcal{I}$$\mathcal{I}$$z^{-1}$$z^{-1}$${\uparrow}\bowtie$${\uparrow}\bowtie$$+$$o$
---|---|---

Now we lift and incrementalize to get the circuit for
${({\uparrow}{({\uparrow}\bowtie)}^{\Delta})}^{\Delta}$:

$a$$b$$\mathcal{I}$$\mathcal{I}$${\uparrow}{\uparrow}\bowtie$${\uparrow}\mathcal{I}$${\uparrow}\mathcal{I}$${\uparrow}z^{-1}$${\uparrow}z^{-1}$${\uparrow}{\uparrow}\bowtie$${\uparrow}{\uparrow}\bowtie$$+$$\mathcal{D}$$o$

Applying the chain rule and the linearity of ${\uparrow}\mathcal{I}$ and
${\uparrow}z^{-1}$ this becomes:

$a$${({\uparrow}{\uparrow}\bowtie)}^{\Delta}$$b$${\uparrow}\mathcal{I}$${\uparrow}\mathcal{I}$${\uparrow}z^{-1}$${\uparrow}z^{-1}$${({\uparrow}{\uparrow}\bowtie)}^{\Delta}$${({\uparrow}{\uparrow}\bowtie)}^{\Delta}$$+$$o$

We now have three applications of ${({\uparrow}{\uparrow}\bowtie)}^{\Delta}$.
Each of these is the incremental form of a bilinear operator, so it looks like
in the end we will have $3\times 3$ applications of
${\uparrow}{\uparrow}\bowtie$. In fact, the overall expression can be
simplified (see (tr, ) for a precise derivation), and the end result only has
4 terms in ${\uparrow}{\uparrow}\bowtie$.

Here is the final form of the expanded join circuit:

$a$$b$${\uparrow}\mathcal{I}$$\mathcal{I}$$\mathcal{I}$$z^{-1}$$\mathcal{I}$${\uparrow}\mathcal{I}$$z^{-1}$$\mathcal{I}$${\uparrow}z^{-1}$${\uparrow}z^{-1}$${\uparrow}{\uparrow}\bowtie$${\uparrow}{\uparrow}\bowtie$${\uparrow}{\uparrow}\bowtie$${\uparrow}{\uparrow}\bowtie$$+$$o$

Returning to
${({\uparrow}{({\uparrow}\mathit{distinct})}^{\Delta})}^{\Delta}$, we can
compute its circuit by expanding once using Proposition 4.7:

$i$${({\uparrow}{({\uparrow}\mathit{distinct})}^{\Delta})}^{\Delta}$$o$ |  $\cong$ |  $i$$\mathcal{I}$${\uparrow}\mathcal{I}$${\uparrow}z^{-1}$${\uparrow}{\uparrow}H$$\mathcal{D}$$o$
---|---|---

Finally, stitching all these pieces together we get the final circuit shown in
Figure 2.

$\Delta$E${\uparrow}\delta_{0}$${\uparrow}\mathcal{I}$$\mathcal{I}$$\mathcal{I}$$z^{-1}$$\mathcal{I}$${\uparrow}\mathcal{I}$$z^{-1}$$\mathcal{I}$${\uparrow}z^{-1}$${\uparrow}z^{-1}$${\uparrow}{\uparrow}\bowtie$${\uparrow}{\uparrow}\bowtie$${\uparrow}{\uparrow}\bowtie$${\uparrow}{\uparrow}\bowtie$$+$${\uparrow}{\uparrow}\pi$${\uparrow}{\uparrow}\pi$${\uparrow}{\uparrow}\sigma$${\uparrow}{\uparrow}\pi$${\uparrow}{\uparrow}\sigma$$+$$\mathcal{I}$${\uparrow}\mathcal{I}$${\uparrow}z^{-1}$${\uparrow}{\uparrow}H$$\mathcal{D}$${\uparrow}\int$$\Delta$R${\uparrow}z^{-1}$
Figure 2. Final form of circuit from §6.1.

###  6.2. Complexity of incremental recursive queries

##### Time complexity

The time complexity of an incremental recursive query can be estimated as a
product of the number of fixed point iterations and the complexity of each
iteration. The incrementalized circuit (6.1) performs the same number of
iterations as the non-incremental circuit (5.1) in the worst case: once the
non-incremental circuit reaches the fixed point, its output is constant and so
is its derivative computed by the incrementalized circuit.

Consider a nested stream of changes
$s\in\mathcal{S}_{\mathcal{S}_{A}},s[t_{1}][t_{2}]$, where $t_{1}$ is the
input timestamp and $t_{2}$ is the fixed point iteration number. The
unoptimized loop body
${({\uparrow}{({\uparrow}T)}^{\Delta})}^{\Delta}=\mathcal{D}\circ{\uparrow}\mathcal{D}\circ{\uparrow}{\uparrow}T\circ{\uparrow}\mathcal{I}\circ\mathcal{I}$
has the same time complexity as $T$ applied to the aggregated input of size
$R(s)[t_{1}][t_{2}]\stackrel{{\scriptstyle\textrm{\scriptsize
def}}}{{=}}\|({\uparrow}\mathcal{I}\circ\mathcal{I})(s)[t_{1}][t_{2}]\|=\|\sum_{(i_{1},i_{2})\leq(t_{1},t_{2})}s[i_{1}][i_{2}]\|$.
As before, an optimized circuit can be significantly more efficient. For
instance, by applying Theorem 3.4 twice, to $\bowtie$ and ${\uparrow}\bowtie$,
we obtain a circuit for nested incremental join
$s_{1}{({\uparrow}{({\uparrow}\bowtie)}^{\Delta})}^{\Delta}s_{2}$ that runs in
$O(\|{\uparrow}\mathcal{I}(s1)[t1][t2]\|\times\|\mathcal{I}(s2)[t1][t2]\|)\ll
O(R(s_{1})\times R(s_{2}))$ (because each term is correspondingly smaller).

##### Space complexity

Integration ($\mathcal{I}$) and differentiation ($\mathcal{D}$) of a stream
$s\in\mathcal{S}_{\mathcal{S}_{A}}$ uses memory proportional to
$\sum_{t_{2}}\|\sum_{t_{1}}s[t_{1}][t_{2}]\|$, i.e., the total size of changes
aggregated over columns of the matrix. The unoptimized circuit integrates and
differentiates respectively inputs and outputs of the recursive program
fragment. As we move $\mathcal{I}$ and $\mathcal{D}$ inside the circuit using
the chain rule, we additionally store changes to intermediate streams.
Effectively we cache results of fixed point iterations from earlier timestamps
to update them efficiently as new input changes arrive. Notice that space is
proportional to the number of iterations of the inner while loop.

##  7\. Extensions

The DBSP language can express a richer class of streaming computations (both
incremental and non-incremental) than those covered so far. In this section we
give several examples.

###  7.1. Multisets and bags

In §4 we have shown how to implement the relational algebra on sets. Some SQL
queries however produce _multisets_ , e.g., UNION ALL. Since $\mathbb{Z}$-sets
generalize multisets and bags, it is easy to implement query operators that
compute on such structures. For example, while SQL UNION is $\mathbb{Z}$-set
addition followed by $\mathit{distinct}$, UNION ALL is just $\mathbb{Z}$-set
addition.

###  7.2. Aggregation

Aggregation in SQL applies a function $a$ to a whole set producing a “scalar”
result with some type $R$: $a:2^{A}\to R$. We convert such aggregation
functions to operate on $\mathbb{Z}$-sets, so in DBSP an aggregation function
has a signature $a:\mathbb{Z}[A]\to R$. Correctness of the implementation is
defined as in §4.2.1.

The SQL COUNT aggregation function is implemented on $\mathbb{Z}$-sets by
$a_{\texttt{COUNT}}:\mathbb{Z}[A]\to\mathbb{Z}$, which computes a _sum_ of all
the element weights: $a_{\texttt{COUNT}}(s)=\sum_{x\in s}s[x]$. The SQL SUM
aggregation function is implemented on $\mathbb{Z}$-sets by
$a_{\texttt{SUM}}:\mathbb{Z}[\mathbb{R}]\to\mathbb{R}$ which performs a
_weighted sum_ of all (real) values: $a_{\texttt{SUM}}(s)=\sum_{x\in s}x\times
s[x]$.

With this definition the aggregation functions $a_{\texttt{COUNT}}$ and
$a_{\texttt{SUM}}$ are in fact linear transformations between the group
$\mathbb{Z}[A]$ and the result group ($\mathbb{Z}$, and $\mathbb{R}$
respectively).

If the output of the DBSP circuit can be such a “scalar” value, then
aggregation with a linear function is simply function application, and thus it
is automatically incremental. However, in general, for composing multiple
queries we require the result of an aggregation to be a singleton
$\mathbb{Z}$-set (containing a single value), and not a scalar value. In this
case the aggregation function is implemented in DBSP as the composition of the
actual aggregation and the $\mbox{makeset}:A\to\mathbb{Z}[A]$ function, which
converts a scalar value of type $A$ to a singleton $\mathbb{Z}$-set, defined
as follows: $\mbox{makeset}(x)\stackrel{{\scriptstyle\textrm{\scriptsize
def}}}{{=}}1\cdot x$.

In conclusion, the following SQL query: SELECT SUM(c) FROM I is implemented as
the following circuit:

I$\pi_{\texttt{C}}$$a_{\texttt{SUM}}$makesetO

The lifted incremental version of this circuit is interesting: since $\pi$ and
$a_{\texttt{SUM}}$ are linear, they are equivalent to their own incremental
versions. Although
${({\uparrow}\mbox{makeset})}^{\Delta}=\mathcal{D}\circ{\uparrow}\mbox{makeset}\circ\mathcal{I}$
cannot be simplified, it is nevertheless efficient, doing only O(1) work per
invocation, since its input and output are singleton values.

An aggregation function such as AVG can be written as the composition of a
more complex linear function that computes a pair of values using SUM and
COUNT, followed by a makeset and a selection operation that divides the two
columns.

⬇

SELECT AVG(c) FROM I

I$\pi_{\texttt{C}}$$(a_{\texttt{SUM}},a_{\texttt{COUNT}})$makeset$\sigma_{/}$O

Finally, some aggregate functions, such as MIN, are _not_ incremental in
general, since for handling deletions they may need to know the full set, and
not just its changes. The lifted incremental version of such aggregate
functions is implemented essentially by “brute force”, using the formula
${({\uparrow}a_{\texttt{MIN}})}^{\Delta}=\mathcal{D}\circ{\uparrow}a_{\texttt{MIN}}\circ\mathcal{I}$.
Such functions perform work proportional to $R(s)$ at each invocation.

Note that the SQL ORDER BY directive can be modeled as a non-linear aggregate
function that emits a list. However, such an implementation it is not
efficiently incrementalizable in DBSP. We leave the efficient handling of
ORDER BY to future work.

Even when aggregation results do not form a group, they usually form a
structure with a zero element. We expect that a well-defined aggregation
function maps empty $\mathbb{Z}$-sets to zeros in the target domain.

###  7.3. Grouping; indexed relations

Pick an arbitrary set $K$ of “key values.” Consider the mathematical structure
of finite maps from $K$ to $\mathbb{Z}$-sets over some other domain $A$:
$K\to\mathbb{Z}[A]=\mathbb{Z}[A][K]$. We call values $i$ of this structure
indexed $\mathbb{Z}$-sets: for each key $k\in K$, $i[k]$ is a
$\mathbb{Z}$-set. Because the codomain $\mathbb{Z}[A]$ is an abelian group,
this structure is itself an abelian group.

We use this structure to model the SQL GROUP BY operator in DBSP. Consider a
partitioning function $p:A\to K$ that assigns a key to any value in $A$. We
define the grouping function $G_{p}:\mathbb{Z}[A]\to(K\to\mathbb{Z}[A])$ as
$G_{p}(a)[k]\stackrel{{\scriptstyle\textrm{\scriptsize def}}}{{=}}\sum_{x\in
a.p(x)=k}a[x]\cdot x$. When applied to a $\mathbb{Z}$-set $a$ this function
returns a indexed $\mathbb{Z}$-set, where each element is called a
grouping333We use “group” for the algebraic structure and “grouping” for the
result of GROUP BY.: for each key $k$ a grouping is a $\mathbb{Z}$-set
containing all elements of $a$ that map to $k$ (as in SQL, groupings are
multisets, represented by $\mathbb{Z}$-sets). Consider our example
$\mathbb{Z}$-set $R$ from §4, and a key function $p(s)$ that returns the first
letter of the string $s$. Then we have that
$G_{p}(R)=\\{\mbox{{j}}\mapsto\\{\mbox{{joe}}\mapsto
1\\},\mbox{{a}}\mapsto\\{\mbox{{anne}}\mapsto-1\\}\\}$, i.e., grouping with
this key function produces an indexed $\mathbb{Z}$-set with two groupings,
each of which contains a $\mathbb{Z}$-set with one element.

The grouping function $G_{p}$ is linear for any $p$. It follows that the
group-by implementation in DBSP is automatically incremental: given some
changes to the input relation we can apply the partitioning function to each
change separately to compute how each grouping changes.

###  7.4. GROUP BY-AGGREGATE

Grouping in SQL is almost always followed by aggregation. Let us consider an
aggregation function $a:(K\times\mathbb{Z}[A])\to B$ that produces values in
some group $B$, and an indexed relation of type $\mathbb{Z}[A][K]$, as defined
above in §7.3. The nested relation aggregation operator
$Agg_{a}:\mathbb{Z}[A][K]\to B$ applies $a$ to the contents of each grouping
independently and adds the results:
$Agg_{a}(g)\stackrel{{\scriptstyle\textrm{\scriptsize def}}}{{=}}\sum_{k\in
K}a(k,g[k])$. To apply this to our example, let us compute the equivalent of
GROUP-BY count; we use the following aggregation function
$count:K\times\mathbb{Z}[A]$,
$count(k,s)=\mbox{makeset}((k,a_{\texttt{COUNT}}(s)))$, using the
$\mathbb{Z}$-set counting function $a_{\texttt{COUNT}}$ from §7.2; the
notation $(a,b)$ is a pair of values $a$ and $b$. Then we have
$Agg_{count}(G_{p}(R))=\\{(\mbox{{j}},1)\mapsto 1,(\mbox{{a}},-1)\mapsto
1\\}$.

Notice that, unlike SQL, DBSP can express naturally computations on indexed
$\mathbb{Z}$-sets, they are just an instance of a group structure. One can
even implement queries that operate on each grouping in an indexed
$\mathbb{Z}$-set. However, our definition of incremental computation is only
concerned with incrementality in the _outermost_ structures. We leave it to
future work to explore an appropriate definition of incremental computation
that operates on the _inner_ relations.

A very useful operation on nested relations is flatmap, which is essentially
the inverse of partitioning, converting an indexed $\mathbb{Z}$-set into a
$\mathbb{Z}$-set: $\mbox{flatmap}:\mathbb{Z}[A][K]\to\mathbb{Z}[A\times K]$.
flatmap is in fact a particular instance of aggregation, using the aggregation
function $a:K\times\mathbb{Z}[A]\to\mathbb{Z}[A\times K]$ defined by
$a(k,s)=\sum_{x\in s[k]}s[k][x]\cdot(k,x).$ For our previous example,
$\mbox{flatmap}(G_{p}(R))=\\{(\mbox{{j}},\mbox{{joe}})\mapsto
1,(\mbox{{a}},\mbox{{anne}})\mapsto-1\\}$.

If we use an aggregation function $a:K\times Z[A]$ that is linear in its
second argument, then the aggregation operator $Agg_{a}$ is linear, and thus
fully incremental. As a consequence, flatmap is linear. However, many
practical aggregation functions for nested relations are in fact not linear;
an example is the $count$ function above, which is not linear since it uses
the makeset non-linear function. Nevertheless, while the incremental
evaluation of such functions is not fully incremental, it is at least partly
incremental: when applying a change to groupings, the aggregation function
only needs to be re-evaluated _for groupings that have changed_.

###  7.5. Antijoin

Antijoins arise in the implementation of Datalog programs with stratified
negation. Consider the following program:

⬇

O(v, z) :- I1(v, z), not I2(v).

The semantics of such a rule is defined in terms of joins and set difference.
This rule is equivalent with the following pair of rules:

⬇

C(v, z) :- I1(v, z), I2(v).

O(v, z) :- I1(v, z), not C(v, z).

This transformation reduces an antijoin to a join followed by a set
difference. This produces the following DBSP circuit:

I1I2$\bowtie$—$+$$\mathit{distinct}$O

###  7.6. Streaming joins

Consider a binary query $T(s,t)=\mathcal{I}(s)~{}~{}{\uparrow}\bowtie~{}~{}t$.
This is the _relation-to-stream join_ operator supported by streaming
databases like ksqlDB (jafarpour-edbt19, ). Stream $s$ carries changes to a
relation, while $t$ carries arbitrary data, e.g., logs or telemetry data
points. $T$ discards values from $t$ after matching them against the
accumulated contents of the relation.

####  7.6.1. Streaming Window queries

Streaming databases often organize the contents of streams into windows, which
store a subset of data points with a predefined range of timestamps. The
circuit below (a convolution filter in DSP) computes a _fixed-size sliding-
window aggregate_ over the last four timestamps defined by the $T_{i}$
functions.

$s$$T_{0}$$z^{-1}$$T_{1}$$z^{-1}$$T_{2}$$z^{-1}$$o$

In practice, windowing is usually based on physical timestamps attached to
stream values rather than logical time. For instance, the CQL (arasu-tr02, )
query “SELECT * FROM events [RANGE 1 hour]” returns all events received within
the last hour. The corresponding circuit (on the left) takes input stream
$s\in\mathcal{S}_{\mathbb{Z}[A]}$ and an additional input
$\theta\in\mathcal{S}_{\mathbb{R}}$ that carries the value of the current
time.

$s$$\theta$$I$$W$$o$ |  $\cong$ |  $s$$\theta$$+$$W$$o$$z^{-1}$
---|---|---

where the _window operator_ $W$ prunes input $\mathbb{Z}$-sets, only keeping
values with timestamps less than an hour behind $\theta[t]$. Assuming
$ts:A\to\mathbb{R}$ returns the physical timestamp of a value, $W$ is defined
as $W(v,\theta)[t]\stackrel{{\scriptstyle\textrm{\scriptsize
def}}}{{=}}\\{x\in v[t].ts(x)\geq\theta[t]-1hr\\}$. Assuming $\theta$
increases monotonically, $W$ can be moved inside integration, resulting in the
circuit on the right, which uses bounded memory to compute a window of an
unbounded stream. This circuit is a building block of a large family of window
queries, including window joins and aggregation. We conjecture that DBSP can
express any CQL query.

###  7.7. Relational while queries

(See also non-monotonic semantics for Datalog¬ and Datalog¬¬(Abiteboul-book95,
).) To illustrate the power of DBSP we implement the following “while”
program, where $Q$ is an arbitrary relational algebra query:

⬇

x := i;

while (x changes)

x := Q(x);

The DBSP implementation of this program is:

$i$$\delta_{0}$$+$${\uparrow}Q$$\mathcal{D}$$\int$$x$$z^{-1}$

This circuit can be converted to a streaming circuit that computes a stream of
values $i$ by lifting it; it can be incrementalized using Algorithm 4.8 to
compute on changes of $i$:

$\Delta
i$${\uparrow}\delta_{0}$$+$${({\uparrow}{\uparrow}Q)}^{\Delta}$${\uparrow}\mathcal{D}$${\uparrow}\int$$\Delta
x$${\uparrow}z^{-1}$

Note that at runtime the execution of this circuit is not guaranteed to
terminate; however, if the circuit does terminate, it will produce the correct
output, i.e., the least fixpoint of $Q$ that includes $i$.

##  8\. Implementation

We are prototyping an implementation of DBSP as part of an open-source project
with an MIT license: https://github.com/vmware/database-stream-processor. The
implementation is written in Rust. The implementation consists of a library
and a runtime. The library provides APIs for basic algebraic data types: such
as groups, finite maps, $\mathbb{Z}$-set, indexed $\mathbb{Z}$-set. A separate
circuit construction API allows users to create DBSP circuits by placing
operator nodes (corresponding to boxes in our diagrams) and connecting them
with streams, which correspond to the arrows in our diagrams. The library
provides pre-built generic operators for integration, differentiation, delay,
nested integration and differentiation. The $\mathbb{Z}$-set library provides
functions for computing the basic $\mathbb{Z}$-set operations corresponding to
plus, negation, grouping, joining, aggregation.

For iterative computations we provide the $\delta_{0}$ operator and an
operator that approximates $\int$ by terminating iteration of a loop at a
user-specified condition (usually the condition is the requirement for a zero
to appear in a specified stream). The low level library allows users to
construct incremental circuits manually. Our plan is to add a higher-level
library (a compiler) which will automatically incrementalize a circuit.

##  9\. Related work

DBSP using non-nested streams is a simplified instance of a Kahn network
(kahn-ifip74, ). Johnson (johnson-phd83, ) studies a very similar
computational model without nested streams and its expressiveness. The
implementation of such streaming models of computation and their relationship
to dataflow machines has been studied by Lee (lee-ieee95, ). Lee (lee-ifip93,
) also introduced streams of streams and the ${\uparrow}z^{-1}$ operator.

In §7 we have discussed the connections with window and stream database
queries (arasu-tr02, ; aurora, ).

Incremental view maintenance (e.g. (gupta-idb93, )) is surveyed in (gupta-
idb95, ); a large bibliography is present in (motik-ai19, ). Its most formal
aspect is propagating “deltas” through algebraic expressions: $Q(R+\Delta
R)=Q(R)+\Delta Q(R,\Delta R)$. This work eventually crystallized in (koch-
pods16, ). DBSP incrementalization is both more modular and more fine-grain
since it deals with streams of updates. Both (koch-pods10, ) and (green-tcs11,
) use $\mathbb{Z}$-sets to uniformly model insertions/deletions.

Picallo et al. (picallo-scop19, ) provide a general solution to IVM for rich
languages. DBSP requires a group structure on the values operated on; this
assumption has two major practical benefits: it simplifies the mathematics
considerably (e.g., Picallo uses monoid actions to model changes), and it
provides a general, simple algorithm (4.8) for incrementalizing arbitrary
programs. The downside of DBSP is that one has to find a suitable group
structure (e.g., $\mathbb{Z}$-sets for sets) to “embed” the computation.
Picallo’s notion of “derivative” is not unique: they need creativity to choose
the right derivative definition, we need creativity to find the right group
structure.

Many heuristic algorithms were published for Datalog-like languages, e.g.,
counting based approaches (Dewan-iis92, ; motik-aaai15, ) that maintain the
number of derivations, DRed (gupta-sigmod93, ) and its variants (Ceri-VLDB91,
; Wolfson-sigmod91, ; Staudt-vldb96, ; Kotowski-rr11, ; Lu-sigmod95, ; Apt-
sigmod87, ), the backward-forward algorithm and variants (motik-aaai15, ;
Harrison-wdd92, ; motik-ai19, ). DBSP is more general than these approaches.
Interestingly, the $\mathbb{Z}$-sets multiplicities in our relational
implementation are related to the counting-number-of-derivations approaches.

DBSP is tightly related to Differential Dataflow (DD) (mcsherry-cidr13, ;
murray-sosp13, ) and its theoretical foundations (abadi-fossacs15, ) (and
recently (mchserry-vldb20, ; chothia-vldb16, )). All DBSP operators are based
on DD operators. DD’s computational model is more powerful than DBSP, since it
allows past values in a stream to be ”updated”. In contrast, our model assumes
that the inputs of a computation arrive in the time order while allowing for
nested time domains via the modular lifting transformer. However, DBSP can
express both incremental and non-incremental computations; in essence DBSP is
“deconstructing” DD into simple component building blocks; the core
Proposition 3.2 and the Algorithm based on it 4.8 are new contributions.

##  10\. Conclusions

In this paper we have introduced DBSP, a model of computation based on
infinite streams over abelian groups. In this model streams are used to model
consecutive snapshots of a database, consecutive changes (or transactions)
applied to a database, and consecutive values of loop-carried variables.

We have defined an abstract notion of incremental computation over streams,
and defined the incrementalization operator ${\cdot}^{\Delta}$, which
transforms a stream computation $Q$ into its incremental version
${Q}^{\Delta}$. The incrementalization operator has some very nice algebraic
properties, which can generate very efficient incremental implementations for
linear and bilinear computations.

We have then applied these tools to two domains: relational queries and
recursive stratified queries. This gave us a general algorithm for
incrementalizing an arbitrary query, including recursive queries. However, we
believe that both the incrementalization algorithm and DBSP are even more
powerful and can apply to even richer classes of query languages, including
languages operating on nested relations and streaming query languages.

## References

  * (1) The Aurora project.  http://cs.brown.edu/research/aurora/, December 2004.
  * (2) Martín Abadi, Frank McSherry, and Gordon Plotkin.  Foundations of differential dataflow.  In Foundations of Software Science and Computation Structures (FoSSaCS), London, UK, April 11-18 2015.  URL: http://homepages.inf.ed.ac.uk/gdp/publications/differentialweb.pdf.
  * (3) Serge Abiteboul, Richard Hull, and Victor Vianu.  Foundations of Databases.  Addison-Wesley, 1995.  URL: http://webdam.inria.fr/Alice/.
  * (4) Mario Alvarez-Picallo, Alex Eyers-Taylor, Michael Peyton Jones, and C.-H. Luke Ong.  Fixing incremental computation.  In European Symposium on Programming Languages and Systems (ESOP), pages 525–552, Prague, Czech Republic, April 6–11 2019.  URL: https://link.springer.com/chapter/10.1007/978-3-030-17184-1_19.
  * (5) Krzysztof R. Apt and Jean-Marc Pugin.  Maintenance of stratified databases viewed as a belief revision system.  In Moshe Y. Vardi, editor, ACM SIGMOD International conference on Management of data (SIGMOD), pages 136–145, San Diego, California, March 23-25 1987.  doi:10.1145/28659.28674.
  * (6) Arvind Arasu, Shivnath Babu, and Jennifer Widom.  An abstract semantics and concrete language for continuous queries over streams and relations.  Technical Report 2002-57, Stanford InfoLab, 2002.  URL: http://ilpubs.stanford.edu:8090/563/.
  * (7) Mihai Budiu, Frank McSherry, Leonid Ryzhyk, and Val Tannen.  DBSP: A language for expressing incremental view maintenance for rich query languages.  https://github.com/vmware/database-stream-processor/blob/main/doc/spec.pdf.  December 2021.
  * (8) Stefano Ceri and Jennifer Widom.  Deriving production rules for incremental view maintenance.  In Guy M. Lohman, Amílcar Sernadas, and Rafael Camps, editors, International Conference of Very Large Data Bases (VLDB), pages 577–589, Barcelona, Spain, 1991.  URL: http://www.vldb.org/conf/1991/P577.PDF.
  * (9) Zaheer Chothia, John Liagouris, Frank McSherry, and Timothy Roscoe.  Explaining outputs in modern data analytics.  Proc. VLDB Endow., 9(12):1137–1148, August 2016.  URL: https://doi.org/10.14778/2994509.2994530.
  * (10) Hasanat M. Dewan, David Ohsie, Salvatore J. Stolfo, Ouri Wolfson, and Sushil Da Silva.  Incremental database rule processing in PARADISER.  J. Intell. Inf. Syst., 1(2):177–209, 1992.  doi:10.1007/BF00962282.
  * (11) Sergio Greco and Cristian Molinaro.  Datalog and logic databases.  Synthesis Lectures on Data Management, 7(2):1–169, 2015.  URL: https://doi.org/10.2200/S00648ED1V01Y201505DTM041.
  * (12) Todd J Green, Zachary G Ives, and Val Tannen.  Reconcilable differences.  Theory of Computing Systems, 49(2):460–488, 2011.  URL: https://web.cs.ucdavis.edu/~green/papers/tocs11_differences.pdf.
  * (13) Todd J. Green, Grigoris Karvounarakis, and Val Tannen.  Provenance semirings.  In Symposium on Principles of Database Systems (PODS), page 31–40, Beijing, China, June 11-14 2007.  URL: https://doi.org/10.1145/1265530.1265535.
  * (14) Ashish Gupta, Inderpal Singh Mumick, et al.  Maintenance of materialized views: Problems, techniques, and applications.  IEEE Data Eng. Bull., 18(2):3–18, 1995.
  * (15) Ashish Gupta, Inderpal Singh Mumick, and V. S. Subrahmanian.  Maintaining views incrementally.  In Proceedings of the 1993 ACM SIGMOD International Conference on Management of Data, SIGMOD ’93, page 157–166, Washington, D.C., USA, 1993\.  URL: https://doi.org/10.1145/170035.170066.
  * (16) Ashish Gupta, Inderpal Singh Mumick, and V. S. Subrahmanian.  Maintaining views incrementally.  In ACM SIGMOD International conference on Management of data (SIGMOD), pages 157–166, Washington, DC, May 26-28 1993. ACM Press.  doi:10.1145/170035.170066.
  * (17) John V. Harrison and Suzanne W. Dietrich.  Maintenance of materialized views in a deductive database: An update propagation approach.  In Kotagiri Ramamohanarao, James Harland, and Guozhu Dong, editors, Workshop on Deductive Databases, volume CITRI/TR-92-65 of Technical Report, pages 56–65, Washington, D.C., November 14 1992. Department of Computer Science, University of Melbourne.
  * (18) Hojjat Jafarpour, Rohan Desai, and Damian Guy.  KSQL: Streaming SQL engine for Apache Kafka.  In International Conference on Extending Database Technology (EDBT), pages 524–533, Lisbon, Portugal, March 26-29 2019.  URL: http://openproceedings.org/2019/conf/edbt/EDBT19_paper_329.pdf.
  * (19) Steven Dexter Johnson.  Synthesis of Digital Designs from Recursion Equations.  PhD thesis, Indiana University, May 1983.  https://help.luddy.indiana.edu/techreports/TRNNN.cgi?trnum=TR141.
  * (20) Gilles Kahn.  The semantics of a simple language for parallel programming.  In IFIP Congress on Information Processing, 1974.  URL: http://www1.cs.columbia.edu/~sedwards/papers/kahn1974semantics.pdf.
  * (21) Christoph Koch.  Incremental query evaluation in a ring of databases.  In Symposium on Principles of Database Systems (PODS), page 87–98, Indianapolis, Indiana, USA, 2010.  URL: https://doi.org/10.1145/1807085.1807100.
  * (22) Christoph Koch, Daniel Lupei, and Val Tannen.  Incremental view maintenance for collection programming.  In Symposium on Principles of Database Systems (PODS), page 75–90, San Francisco, California, USA, 2016.  URL: https://doi.org/10.1145/2902251.2902286.
  * (23) Jakub Kotowski, François Bry, and Simon Brodt.  Reasoning as axioms change - incremental view maintenance reconsidered.  In Web Reasoning and Rule Systems RR, volume 6902 of Lecture Notes in Computer Science, pages 139–154, Galway, Ireland, August 29-30 2011. Springer.  doi:10.1007/978-3-642-23580-1\\_11.
  * (24) Edward A. Lee.  Multidimensional streams rooted in dataflow.  In IFIP Working Conference on Architectures and Compilation Techniques for Fine and Medium Grain Parallelism, Orlando, FL, January 20-22 1993\.  URL: https://ptolemy.berkeley.edu/publications/papers/93/mdsdf/.
  * (25) Edward A. Lee and Thomas M. Parks.  Dataflow process networks.  Proceedings of the IEEE, pages 773–801, May 1995.  URL: https://ptolemy.berkeley.edu/publications/papers/95/processNets/.
  * (26) James J. Lu, Guido Moerkotte, Joachim Schü, and V. S. Subrahmanian.  Efficient maintenance of materialized mediated views.  In ACM SIGMOD International conference on Management of data (SIGMOD), pages 340–351, San Jose, California, May 22-25 1995.  doi:10.1145/223784.223850.
  * (27) Frank McSherry, Andrea Lattuada, Malte Schwarzkopf, and Timothy Roscoe.  Shared arrangements: Practical inter-query sharing for streaming dataflows.  Proc. VLDB Endow., 13(10):1793–1806, June 2020.  URL: https://doi.org/10.14778/3401960.3401974.
  * (28) Frank McSherry, Derek Gordon Murray, Rebecca Isaacs, and Michael Isard.  Differential dataflow.  In Conference on Innovative Data Systems Research (CIDR), Asilomar, CA, January 6–9 2013.  URL: http://cidrdb.org/cidr2013/Papers/CIDR13_Paper111.pdf.
  * (29) Boris Motik, Yavor Nenov, Robert Piro, and Ian Horrocks.  Maintenance of Datalog materialisations revisited.  Artif. Intell., 269:76–136, 2019.  URL: https://doi.org/10.1016/j.artint.2018.12.004.
  * (30) Boris Motik, Yavor Nenov, Robert Edgar Felix Piro, and Ian Horrocks.  Incremental update of Datalog¡ materialisation: the backward/forward algorithm.  In Conference on Artificial Intelligence (AAAI), pages 1560–1568, Austin, Texas, January 25-30 2015. AAAI Press.  URL: http://www.aaai.org/ocs/index.php/AAAI/AAAI15/paper/view/9660.
  * (31) Derek G. Murray, Frank McSherry, Rebecca Isaacs, Michael Isard, Paul Barham, and Martín Abadi.  Naiad: A timely dataflow system.  In ACM Symposium on Operating Systems Principles (SOSP), page 439–455, Farminton, Pennsylvania, 2013.  URL: https://doi.org/10.1145/2517349.2522738.
  * (32) L. R. Rabiner and B. Gold, editors.  Theory and Application of Digital Signal Processing.  Prentice-Hall, 1975.
  * (33) Martin Staudt and Matthias Jarke.  Incremental maintenance of externally materialized views.  In International Conference of Very Large Data Bases (VLDB), pages 75–86, Mumbai (Bombay), India, September 3-6 1996.  URL: http://www.vldb.org/conf/1996/P075.PDF.
  * (34) Ouri Wolfson, Hasanat M. Dewan, Salvatore J. Stolfo, and Yechiam Yemini.  Incremental evaluation of rules and its relationship to parallelism.  In ACM SIGMOD International conference on Management of data (SIGMOD), pages 78–87, Denver, Colorado, May 29-31 1991. ACM Press.  doi:10.1145/115790.115799.

##  Appendix A Supporting material

###  A.1. Operations on nested streams

If a stream can be thought of as an infinite vector, a stream of streams can
be thought of as an “matrix” with an infinite number of rows, where each row
is a stream. For example, we can depict the nested stream
$i\in\mathcal{S}_{\mathcal{S}_{\mathbb{N}}}$ defined by
$i[t_{0}][t_{1}]=t_{0}+2t_{1}$ as:  i =  arg0 1 2 3 2 3 4 5 4 5 6 7 6 7 8 9
[[arg[1]arg[2]arg[3]arg[4]⋯][arg[5]arg[6]arg[7]arg[8]⋯][arg[9]arg[10]arg[11]arg[12]⋯][arg[13]arg[14]arg[15]arg[16]⋯]⋮]

($t_{0}$ is the column index, and $t_{1}$ is the row index). Let us perform
some computations on nested streams to get used to them. Lifting twice a
scalar function computes on elements of the matrix pointwise:

(↑↑(x ↦x mod2))(i) =  arg0 1 0 1 0 1 0 1 0 1 0 1 0 1 0 1
[[arg[1]arg[2]arg[3]arg[4]⋯][arg[5]arg[6]arg[7]arg[8]⋯][arg[9]arg[10]arg[11]arg[12]⋯][arg[13]arg[14]arg[15]arg[16]⋯]⋮]

The $\mathcal{I}$ operator on $\mathcal{S}_{\mathcal{S}_{A}}$ is well-defined:
it operates on rows of the matrix, treating each row as a single value:

I(i) =  arg0 1 2 3 2 4 6 8 6 9 12 15 12 16 20 24
[[arg[1]arg[2]arg[3]arg[4]⋯][arg[5]arg[6]arg[7]arg[8]⋯][arg[9]arg[10]arg[11]arg[12]⋯][arg[13]arg[14]arg[15]arg[16]⋯]⋮]

Lifting a stream operator computing on $\mathcal{S}_{A}$, such as
$\mathcal{I}:\mathcal{S}_{A}\to\mathcal{S}_{A}$, also produces an operator on
nested streams, but this time computing on the columns of the matrix:
${\uparrow}\mathcal{I}:\mathcal{S}_{\mathcal{S}_{A}}\to\mathcal{S}_{\mathcal{S}_{A}}.$

(↑I)(i) =  arg0 1 3 6 2 5 9 14 4 9 15 22 6 13 21 30
[[arg[1]arg[2]arg[3]arg[4]⋯][arg[5]arg[6]arg[7]arg[8]⋯][arg[9]arg[10]arg[11]arg[12]⋯][arg[13]arg[14]arg[15]arg[16]⋯]⋮]

Similarly, we can apply $\mathcal{D}$ to nested streams
$\mathcal{D}:\mathcal{S}_{\mathcal{S}_{A}}\to\mathcal{S}_{\mathcal{S}_{A}}$,
computing on rows of the matrix:

D(i) =  arg0 1 2 3 2 2 2 2 2 2 2 2 2 2 2 2
[[arg[1]arg[2]arg[3]arg[4]⋯][arg[5]arg[6]arg[7]arg[8]⋯][arg[9]arg[10]arg[11]arg[12]⋯][arg[13]arg[14]arg[15]arg[16]⋯]⋮]

while
${\uparrow}\mathcal{D}:\mathcal{S}_{\mathcal{S}_{A}}\to\mathcal{S}_{\mathcal{S}_{A}}$
computes on the columns:

(↑D)(i) =  arg0 1 1 1 2 1 1 1 4 1 1 1 6 1 1 1
[[arg[1]arg[2]arg[3]arg[4]⋯][arg[5]arg[6]arg[7]arg[8]⋯][arg[9]arg[10]arg[11]arg[12]⋯][arg[13]arg[14]arg[15]arg[16]⋯]⋮]

Similarly, $z^{-1}$ and its lifted variant have different outcomes:

$z^{-1}$(i) =  arg0 0 0 0 0 1 2 3 2 3 4 5 4 5 6 7
[[arg[1]arg[2]arg[3]arg[4]⋯][arg[5]arg[6]arg[7]arg[8]⋯][arg[9]arg[10]arg[11]arg[12]⋯][arg[13]arg[14]arg[15]arg[16]⋯]⋮]

Notice the following commutativity properties for integration and
differentiation on nested streams:
$\mathcal{I}\circ({\uparrow}\mathcal{I})=({\uparrow}\mathcal{I})\circ\mathcal{I}$
and
$\mathcal{D}\circ({\uparrow}\mathcal{D})=({\uparrow}\mathcal{D})\circ\mathcal{D}$.

(↑$z^{-1}$)(i) =  arg0 0 1 2 0 2 3 4 0 4 5 6 0 6 7 8
[[arg[1]arg[2]arg[3]arg[4]⋯][arg[5]arg[6]arg[7]arg[8]⋯][arg[9]arg[10]arg[11]arg[12]⋯][arg[13]arg[14]arg[15]arg[16]⋯]⋮]

$z^{-1}$ commutes with ${\uparrow}z^{-1}$:

(↑$z^{-1}$)($z^{-1}$(i)) = $z^{-1}$((↑$z^{-1}$)(i)) =  arg0 0 0 0 0 0 1 2 0 2
3 4 0 4 5 6
[[arg[1]arg[2]arg[3]arg[4]⋯][arg[5]arg[6]arg[7]arg[8]⋯][arg[9]arg[10]arg[11]arg[12]⋯][arg[13]arg[14]arg[15]arg[16]⋯]⋮]

D_$\mathcal{S}_{\mathcal{S}_{\mathbb{N}}}$(i) = (D(↑D))(i) =  arg0 1 1 1 2 0 0
0 2 0 0 0 2 0 0 0
[[arg[1]arg[2]arg[3]arg[4]⋯][arg[5]arg[6]arg[7]arg[8]⋯][arg[9]arg[10]arg[11]arg[12]⋯][arg[13]arg[14]arg[15]arg[16]⋯]⋮]

I_$\mathcal{S}_{\mathcal{S}_{\mathbb{N}}}$(i) = ((↑I)(I))(i)=  arg0 1 3 6 2 6
12 20 6 15 27 42 12 28 48 72
[[arg[1]arg[2]arg[3]arg[4]⋯][arg[5]arg[6]arg[7]arg[8]⋯][arg[9]arg[10]arg[11]arg[12]⋯][arg[13]arg[14]arg[15]arg[16]⋯]⋮]