DBSP：丰富查询语言的自动增量视图维护

摘要。

增量视图维护长期以来一直是数据库理论（gupta-idb93，）的中心问题。 针对有限类别的数据库语言（例如关系代数或 Datalog），提出了许多解决方案。 这些技术并不能自然地推广到更丰富的语言。 在本文中，我们分三步给出了这个问题的通用解决方案：（1）我们描述了一种简单但富有表现力的语言，称为 DBSP，用于描述数据流上的计算； （2）我们给出了解决任意 DBSP 程序的增量视图维护问题的通用算法，并且（3）我们展示了如何对许多丰富的数据库查询语言进行建模（包括完整的关系查询、分组和聚合、单调和非单调）递归和流式聚合）使用 DBSP。 因此，我们获得了针对所有这些丰富语言的高效增量视图维护技术。

1. 介绍

在本文中，我们提出了一种用于建模流和增量计算的简单数学理论。 该模型在流数据库和增量视图维护的设计和实现中具有直接的实际应用。 我们的模型基于离散数字信号处理 (DSP) (rabiner-book75, ) 中使用的数学形式，但我们将其应用于数据库计算。 因此，我们将其称为“DBSP”。 DBSP 受到差分数据流 (mcsherry-cidr13, ) (DD) 的启发，最初是为了提供比 Abadi 等人 (abadi-fossacs15, )（如第 §9 中所讨论），但已经在该目的的背后演变。

DBSP 的核心概念是流：类型为${𝒮}_A$的流$s$将“时间”时刻$t\in \mathbb{N}$映射到值$s[t]$ 类型为 $A$；将其视为“无限向量”。 流计算是一种消耗一个或多个流并生成另一个流的函数。 我们用典型的 DSP 框箭头图（也称为“电路”）来描述流计算，其中框是计算，流是箭头，如下图所示，它显示了一个流运算符 $T$ 消耗了两个输入流 $s_0$ 和 $s_1$ 并生成一个输出流 $s$：

我们通常将流视为小值的序列，并且我们将以这种方式使用它们。 然而，我们想象力的飞跃，也将整个数据库视为流值。 什么是数据库流？ 它是数据库快照序列。 我们将数据库 $DB$ 的时间演化建模为流 $DB\in {𝒮}_{SCH}$，其中 $SCH$ 是数据库模式。 时间不是挂钟时间，而是本质上应用于数据库的事务序列的计数器。 由于事务是可线性化的，因此它们具有全序，全序定义了线性时间$t$维度：流$DB[t]$的值是$t$之后数据库内容的快照t2> 交易已被应用。 我们假设$DB[0]=0$，即数据库开始为空。

数据库事务还形成应用于数据库的流$T$、更改流或增量。 $t$时刻的数据库快照是应用序列中直到$t$的所有事务的累积结果：$DB[t]={\sum }_{i\le t}T[i]\stackrel{\text{def}}{=}ℐ(T)[t]$（我们将“加法”的概念精确化）之后。）。 将所有变化相加的操作就是流集成。 下图使用 $ℐ$ 运算符进行流集成来表达这种关系：

相反，我们可以说事务是数据库的更改，并写成$T=𝒟(DB)$或$T[t]=DB[t]-DB[t-1]$。 这就是流微分的定义，用$𝒟$表示；该操作计算流的变化，是流积分的逆操作。 §2精确定义了流、积分和微分，并分析了它们的性质。

让我们将这些概念应用到视图维护中。 考虑一个数据库 $DB$ 和一个查询 $Q$，将视图 $V$ 定义为数据库快照 $V=Q(DB)$ 的函数。 对应于数据库快照流$DB$，我们有一个视图快照流:$V[t]$是$t$之后的视图内容 -第 交易已被应用。 我们使用下图展示这种关系：

符号 $\uparrow Q$($Q$ 的“提升”）表明查询 $Q$ 独立应用于数据库快照流 $DB$ $\uparrow Q$ 是一个“流查询”，因为它对值流进行操作。 增量视图维护问题需要一种算法来计算视图$V$的更改的流$\mathrm{\Delta }V$，即$𝒟(V)$，如下所示流$T$的函数。通过将这些定义链接在一起，我们得到视图维护问题的以下基本方程:$\mathrm{\Delta }V=𝒟(\uparrow Q(DB))=𝒟(\uparrow Q(ℐ(T)))$，以图形方式显示为：

此定义可以推广到比提升的逐点查询$Q$更丰富的更通用的流式查询$S:{𝒮}_A\to {𝒮}_B$。 流式查询$S$的增量版本用$S^{\mathrm{\Delta }}$表示，根据上式定义，也可以写为：$S^{\mathrm{\Delta }}=𝒟\circ S\circ ℐ$。

通常假设数据集的变化远小于数据集本身；因此，对变化流进行计算可能会产生显着的性能优势。

应用查询增量化运算符$S\mapsto S^{\mathrm{\Delta }}$构造一个直接计算更改的查询；但是，生成的查询并不比在整个数据集上计算的查询更有效，因为它使用积分运算符来重建完整的数据集。 §3 显示了如何使用 ${\cdot }^{\mathrm{\Delta }}$ 运算符的代数属性来优化 $S^{\mathrm{\Delta }}$ 的实现：

1. 第一个属性是许多类原始操作都有非常有效的增量版本。 特别是，线性查询具有属性$Q=Q^{\mathrm{\Delta }}$。 几乎所有关系查询和数据记录查询都基于线性运算符。 因此，可以按照与变化的大小成比例的时间来计算此类查询的增量版本。 双线性运算符（例如连接）具有更复杂的实现，尽管如此，它仍然执行与更改大小成比例的工作，但需要存储与关系大小成比例的数据量。

2. 第二个关键属性是链式法则：${(S_1\circ S_2)}^{\mathrm{\Delta }}=S_1^{}{}_{}{}^{\mathrm{\Delta }}\circ S_2^{}{}_{}{}^{\mathrm{\Delta }}$。 该规则将复杂查询的增​​量版本作为其组件的增量版本的组合。 因此，我们可以将任何增量查询实现为原始增量查询的组合，所有这些查询都执行与更改大小成比例的工作。

借助增量计算的一般理论，我们在第 §4 中展示了如何在 DBSP 中对关系查询进行建模。 这立即为我们提供了计算任何关系查询的增量版本的通用算法。 这些结果众所周知，但它们是由 DBSP 清晰建模的。

将 DBSP 应用于递归查询需要扩展此计算模型。 在§5中，我们引入了两个额外的运算符：${\delta }_0$从标量值创建流，$\int$从流创建标量值。 这些运算符可用于通过 while 循环实现计算。 因此，除了对不断变化的输入和数据库进行建模之外，我们还使用流作为循环迭代变量的连续值序列的模型。 通过此添加，DBSP 变得足够丰富以实现递归查询。 §5.1 显示了如何在 DBSP 中实现带否定的分层递归 Datalog 程序。

在§6中，我们使用 DBSP 对嵌套流的计算进行建模，其中一个流的每个值都是另一个流。 这使我们能够定义递归程序的增量流计算。 因此，我们得出了一种用于增量任意流数据记录程序的通用算法。

DBSP 是一种简单语言：基本的 DBSP 流模型本质上是由两个基本数学运算符构建的：提升 $\uparrow$ 和延迟 $z^{-1}$。 嵌套流模型添加了两个额外的运算符，${\delta }_0$ 用于流构造，$\int$ 用于销毁。

DBSP 还具有表现力：例如，它比分层 Datalog 更强大。 DBSP还可以描述流式窗口查询，或者嵌套关系（例如分组）的查询，以及非单调递归查询。 我们在§7中简要讨论了 DBSP 在更丰富的语言中的应用。

本文省略了大部分证明；完整的证明可在一份内容广泛的配套技术报告(tr, )中找到。

本文做出以下贡献：

1. 它定义了 DBSP，一种用于流计算的小型语言，但它仍然可以表达嵌套的非单调递归；

2. 它提供了增量所有DBSP程序的算法；

3. 对于对应于关系代数和分层单调数据记录的 DBSP 片段，自动增量算法提供了与最先进方法相匹配的结果。 此外，我们的方法还适用于更强大的语言，例如 while 关系型和非单调 Datalog (Abiteboul-book95, )。

4. 它为流式计算和增量计算的操作奠定了正式、健全的基础，使人们能够正式地推理程序转换并设计新的高效实现。

5. DBSP 可以在单个框架中表达流式计算模型和增量计算模型。 我们认为这种统一是一项重大贡献。

2. 流计算

在本节中，我们正式介绍流作为无限值序列的概念，并定义流上的计算。 流运算符 (§2.1) 是流计算的基本构建块。 我们使用（§2.2)受限类型的流运算符：因果运算符（无法“展望未来”）和严格运算符（甚至无法“展望现在”）。 此外，我们所有的运算符都是“同步的”：它们以相同的“速率”消费和生成数据。 因果运算符可以链接成复杂的非循环计算电路；循环电路仅限于在后沿使用严格的运算符。 最后，我们定义（§2.3)两个有用的流运算符：积分和微分。

本节中的所有结果几十年来都已为人所知，但我们重述它们以澄清我们模型的正式假设。

2.1. 流和流运算符

$\mathbb{N}$ 是自然数集合，$𝔹$ 是布尔集合，$\mathbb{Z}$ 是整数集合。

Definition 2.1 （流）:

给定一组 $A$、来自 $A$ 的值的 stream 或 $A$ -stream，是一个函数$\mathbb{N}\to A$。 我们用 ${𝒮}_A\stackrel{\text{def}}{=}\{s|s:\mathbb{N}\to A\}$ 表示所有 $A$ 流的集合。

当 $s\in {𝒮}_A$ 和 $t\in \mathbb{N}$ 时，我们为流 $s$ 的第 $t$ 个元素编写 $s[t]$，而不是常用$s(t)$来区别于其他功能应用。 我们将索引 $t\in \mathbb{N}$ 视为（离散）时间，并将 $s[t]\in A$ 视为流 $s$“在时间”$t$ 的值>。例如，$\mathrm{𝑖𝑑}[t]=t$给出的自然数流$id\in {𝒮}_{\mathbb{N}}$是值序列。

Definition 2.2 （流运算符）:

具有 $n$ 输入的（类型化）流运算符是一个函数$T:{𝒮}_{A_0}\times \mathrm{\cdots }\times {𝒮}_{A_{n-1}}\to {𝒮}_B$。

一般来说，我们将使用“运算符”来表示流上的函数，使用“函数”来计算“标量”值。

我们正在使用简单类型 lambda 演算的扩展来编写 DBSP 程序；我们将逐步介绍其要素。 然而，我们发现使用类似信号处理的电路图来描述 DBSP 程序更具可读性。 在电路图中，矩形代表运算符应用程序（标有运算符名称，例如$T$)，而箭头表示流。

流运算符组合（函数组合）显示为链式电路。 二元运算符 $T:{𝒮}_A\times {𝒮}_B\to {𝒮}_A$ 与一元运算符 $S:{𝒮}_A\to {𝒮}_B$ 组合成计算 $\lambda s.T(T(s,S(s)),S(s)):{𝒮}_A\to {𝒮}_A$ 为：

（图表模糊了运算符输入的顺序；对于非交换运算符，我们必须提供更多信息。）

Definition 2.3:

（提升）给定一个（标量）函数$f:A\to B$，我们通过在时间上逐点提升函数$f$来定义流运算符$\uparrow f:{𝒮}_A\to {𝒮}_B$ ：$(\uparrow f)(s)\stackrel{\text{def}}{=}f\circ s$。 等效地，$(\uparrow f)(s)[t]\stackrel{\text{def}}{=}f(s[t])$。 这扩展到多个参数的函数。

例如， (↑(λx。(2x)))(id)=arg02468[arg⁡[1]arg⁡[2]arg⁡[3]arg⁡[4]arg⁡[5]⋯].

Proposition 2.4 （分配性）:

提升分布在函数组合上：$\uparrow (f\circ g)=(\uparrow f)\circ (\uparrow g)$。

如果两个 DBSP 程序在流上计算相同的输入输出函数，我们就说它们等价。 我们使用符号$\cong$来表示两个电路是等效的。 例如，命题 2.4 陈述了以下电路等价：

$\cong$

2.2. 阿贝尔群上的流

对于技术开发的其余部分，我们需要流${𝒮}_A$的值集$A$来形成交换群$(A,+,0,-)$。 现在我们介绍DBSP使用的原始流运算符。

2.2.1. 延迟和时不变性

Definition 2.5 （延迟）:

这 延迟算子 通过将输入延迟一步来生成输出流：$z_A^{-1}:{𝒮}_A\to {𝒮}_A$：

我们经常省略类型参数$A$，只写$z^{-1}$。 例如，。

Definition 2.6 （时间不变性）:

流运算符 $S:{𝒮}_A\to {𝒮}_B$ 是 时间不变 iff $S(z_A^{-1}(s))=z_B^{-1}(S(s))$ 对于所有 $s\in {𝒮}_A$，或者换句话说，当且仅当以下两个电路是等价的：

$\cong$

这个定义自然地扩展到具有多个输入的运算符。

任意数量的输入的时不变运算符的组合是时不变的。 延迟运算符$z^{-1}$是时不变的。 DBSP 仅使用时不变运算符。

Definition 2.7:

如果$f(0_A)=0_B$，我们说组$f:A\to B$之间的函数具有零保留属性。 我们写$\text{zpp}(f)$。

提升运算符 $\uparrow f$ 是时不变的，当且仅当 $\text{zpp}(f)$ 时。

2.2.2. 因果运算符和严格运算符

Definition 2.8 （因果关系）:

当对于所有$s,s^{\prime }\in {𝒮}_A$和所有时间$t$我们有：$(\forall i\le ts[i]=s^{\prime }[i])\Rightarrow S(s)[t]=S(s^{\prime })[t].$时，流运算符$S:{𝒮}_A\to {𝒮}_B$是因果

换句话说，时间$t$的输出值只能取决于时间$t^{\prime }\le t$的输入值。 提升产生的算子是因果的，$z^{-1}$是因果的。 所有 DBSP 运算符都是因果的。 任意数量的输入的因果运算符的组合都是因果的。

Definition 2.9 （严格性）:

流运算符 $F:{𝒮}_A\to {𝒮}_B$ 是 严格因果关系（缩写为 strict)，如果 $\forall s,s^{\prime }\in {𝒮}_A,\forall t\in \mathbb{N}$ 我们有：

因此，$F(s)$ 的第 $t$ 输出只能依赖于输入 $s$ 的“过去”值，介于 $0$ 和 $t-1$。 特别是，$F(s)[0]=0_B$ 对于所有 $s\in {𝒮}_A$ 都是相同的。 严格算子是因果关系。 一般来说，提升运算符并不严格。 $z^{-1}$ 是严格的。

Proposition 2.10:

对于严格的$F:{𝒮}_A\to {𝒮}_A$，方程$\alpha =F(\alpha )$有唯一解$\alpha \in {𝒮}_A$，用$\text{fix}\alpha .F(\alpha )$表示。

因此，从集合到自身的每个严格运算符都有一个唯一的不动点。 这个简单的证明依赖于强归纳法，表明 $\alpha [t]$ 仅取决于 $t$ 之前的 $\alpha$ 的值。

我们证明以下电路具有严格的“反馈”边缘$F$，是流上定义良好的函数：

Lemma 2.11:

如果 $F:{𝒮}_B\to {𝒮}_B$ 是严格的并且 $T:{𝒮}_A\times {𝒮}_B\to {𝒮}_B$ 是因果的，那么对于固定的 $s$ 运算符 $\lambda \alpha .T(s,F(\alpha )):{𝒮}_A\to {𝒮}_B$ 是严格的。

Corollary 2.12:

如果 $F:{𝒮}_B\to {𝒮}_B$ 是严格的，而 $T:{𝒮}_A\times {𝒮}_B\to {𝒮}_B$ 是因果的，则运算符 $Q(s)=\text{fix}\alpha .T(s,F(\alpha ))$ 是明确定义的并且是因果的。 此外，如果$F$和$T$是时间不变的，那么$Q$也是时间不变的。

DBSP 中的所有流计算都是根据我们描述的原始运算符构建的：提升运算符和延迟（我们在§6 中添加了两个运算符）。 由此类算子组成的电路可以使用数据流机（lee-ieee95，）有效地实现。

带反馈的电路有两个用途：定义积分运算符（在下一节中）和定义递归计算（§5)。 反过来，积分运算符将有助于定义增量计算 (§3)。

2.3. 整合与差异化

请记住，我们要求流的元素来自阿贝尔群$A$。 流本身形成一个阿贝尔群：

Proposition 2.13:

通过将 $+$ 和一元 $-$ 运算从 $A$ 提升到 ${𝒮}_A$ 获得的结构 $({𝒮}_A,+,0,-)$ 是一个阿贝尔群。

流加法和求反是因果的、时不变的运算符。

Definition 2.14:

给定阿贝尔群$A$和$B$，如果它是群同态，我们称其为流运算符$S:{𝒮}_A\to {𝒮}_B$ 线性，即$S(a+b)=S(a)+S(b)$（因此$S(0)=0$和$S(-a)=-S(a)$)。

提升线性函数 $f:A\to B$ 会生成线性、时不变 (LTI) 的流运算符 $\uparrow f$。 $z^{-1}$ 是 LTI。

Definition 2.15:

（双线性）具有 $A,B,C$ 组的两个参数 $f:A\times B\to C$ 的函数，如果它在每个参数中分别是线性的（即，它在加法上分布），则为 双线性 ：$\forall a,b,c,d.f(a+b,c)=f(a,c)+f(b,c)$和$f(a,c+d)=f(a,c)+f(c,d).$

该定义扩展到流运算符。 提升双线性函数 $f:A\times B\to C$ 会生成双线性流运算符 $\uparrow f$。 ${𝒮}_{\mathbb{Z}}$ 上的双线性运算符示例是提升乘法：$f:{𝒮}_{\mathbb{N}}\times {𝒮}_{\mathbb{N}}\to {𝒮}_{\mathbb{N}},f(a,b)[t]=a[t]\cdot b[t]$。

(双)线性算子与线性算子的组合是(双)线性的(因为同态组成)。

使用线性算子产生的反馈回路是线性的：

Proposition 2.16:

令 $S$ 为一元因果 LTI 运算符。 运算符 $Q(s)=\text{fix}\alpha .S(s+z^{-1}(\alpha ))$ 定义明确且 LTI：

Definition 2.17 （微分）:

微分运算符${𝒟}_{{𝒮}_A}:{𝒮}_A\to {𝒮}_A$定义为：$𝒟(s)\stackrel{\text{def}}{=}s-z^{-1}(s)$。

我们通常省略类型，当可以从上下文推断类型时，只写 $𝒟$ 。 $t$ 时刻 $𝒟(s)$ 的值是 $s$ 的当前（时间 $t$)值与前一个（时间 $t-1$)$s$ 的值。例如，。

如果$s$是一个流，那么$𝒟(s)$就是$s$的变化流。

Proposition 2.18:

分化$𝒟$是因果和LTI。

Differentiation	Integration

集成运算符根据其变化“重建”流：

Definition 2.19 （集成）:

积分运算符 ${ℐ}_{{𝒮}_A}:{𝒮}_A\to {𝒮}_A$ 定义为 $$ℐ​(s)=def0>λ2>​3>s4>1>。5>修复7>​8>α9> 6>。0>(s+z −10>​1>(3>α4>)5>2>)6>.

我们一般也省略类型，只写$ℐ$。 这是根据命题 2.16 使用 $S$ 的恒等函数构造的。

Proposition 2.20:

$ℐ(s)$ 是应用于流 $s$ 的离散（不定）积分：$ℐ(s)[t]={\sum }_{i\le t}s[i]$。

例如，。

Proposition 2.21:

$ℐ$ 是因果关系和 LTI。

Theorem 2.22 （反演）:

积分和微分互为逆：$\forall s.ℐ(𝒟(s))=𝒟(ℐ(s))=s$。

$\cong$

$\cong$

3. 增量计算

Definition 3.1:

给定一元流运算符$Q:{𝒮}_A\to {𝒮}_B$，我们将$Q$的增量版本定义为$Q^{\mathrm{\Delta }}\stackrel{\text{def}}{=}𝒟\circ Q\circ ℐ$。 $Q^{\mathrm{\Delta }}$ 与 $Q$ 具有相同的“类型”：$Q^{\mathrm{\Delta }}:{𝒮}_A\to {𝒮}_B$。 对于具有多个输入的运算符，我们通过独立地将 $ℐ$ 应用于每个输入来定义增量版本：例如，if $T:{𝒮}_A\times {𝒮}_B\to {𝒮}_C$ then $T^{\mathrm{\Delta }}(a,b)\stackrel{\text{def}}{=}𝒟(T(ℐ(a),ℐ(b)))$。

下图说明了这个定义背后的直觉：

如果 $Q(s)=o$ 是计算，则 $Q^{\mathrm{\Delta }}$ 执行与 $Q$ 相同的计算，但在更改流 $\mathrm{\Delta }s$ 和 $\mathrm{\Delta }o$。 这是引言中的图表，用 $\mathrm{\Delta }s$ 代替事务流 $T$，用 $o$ 代替视图版本流 $V$。

请注意，我们对增量计算的定义仅对流计算有意义；这与经典定义形成对比，例如(gupta-idb95, ) 仅考虑一项更改。 将定义推广到对流进行操作为我们提供了额外的功能，特别是在使用递归查询进行操作时。

以下命题是我们的核心成果之一。

Proposition 3.2:

（增量版本的属性）：对于适当类型的计算，以下内容成立：

反转::

$Q\mapsto Q^{\mathrm{\Delta }}$ 是双射的；它的倒数是$Q\mapsto ℐ\circ Q\circ 𝒟$。

不变性::

推拉：：

$Q\circ ℐ=ℐ\circ Q^{\mathrm{\Delta }}$； $𝒟\circ Q=Q^{\mathrm{\Delta }}\circ 𝒟$

链：：

${(Q_1\circ Q_2)}^{\mathrm{\Delta }}=Q_1^{}{}_{}{}^{\mathrm{\Delta }}\circ Q_2^{}{}_{}{}^{\mathrm{\Delta }}$ （这概括为具有多个输入的运算符。）

添加：：

循环：：

这些性质的证明依赖于基本的代数运算。 尽管它们很简单，但它们非常有用。 例如，链式法则指出以下两个电路是等效的：

$\cong$

换句话说，要增量化复合查询，您可以独立增量化每个子查询。 这为我们提供了一个简单的确定性方法来计算任意复杂查询的增​​量版本。

我们通过给出链式法则的证明来进行说明，链式法则很简单，并且基于函数组合结合性：

${(Q_1\circ Q_2)}^{\mathrm{\Delta }}$ $=𝒟\circ Q_1\circ Q_2\circ ℐ$

$=𝒟\circ Q_1\circ (ℐ\circ 𝒟)\circ Q_2\circ ℐ$

$=(𝒟\circ Q_1\circ ℐ)\circ (𝒟\circ Q_2\circ ℐ)$

$={(Q_1)}^{\mathrm{\Delta }}\circ {(Q_2)}^{\mathrm{\Delta }}.$

循环规则指出以下电路是等效的：

$\cong$

围绕查询的反馈循环的增量版本就是增量查询的反馈循环。 当我们实现递归查询时，这个结果的重要性就会显而易见。

为了有效地执行增量查询，我们希望直接计算更改流而不集成它们。 上面的不变性属性表明流运算符 $+$、$-$ 和 $z^{-1}$ 与其增量版本相同，因此 $ℐ$ 和 $𝒟$ 可以省略：$Q^{\mathrm{\Delta }}=ℐ\circ Q\circ 𝒟=Q$。 以下定理将其推广到线性和双线性运算符：

Theorem 3.3 （线性）:

对于 LTI 运算符 $Q$，我们有 $Q^{\mathrm{\Delta }}=Q$。

Theorem 3.4 （双线性）:

对于双线性时不变运算符 $\times$，我们有 ${(a\times b)}^{\mathrm{\Delta }}=a\times b+z^{-1}(ℐ(a))\times b+a\times z^{-1}(ℐ(b))$。

通过使用 $\mathrm{\Delta }a$ 重写此语句作为 $a$ 的更改流，我们得到了熟悉的增量等连接公式：$\mathrm{\Delta }(a\times b)=\mathrm{\Delta }a\times \mathrm{\Delta }b+a\times (\mathrm{\Delta }b)+(\mathrm{\Delta }a)\times b$。

这并不奇怪，因为等连接是双线性的，正如我们在下一节中讨论的那样。

4. 增量视图维护

§2 和 §3 中的结果适用于任意组值的流。 在本节中，我们将注意力转向在关系视图维护的上下文中使用这些结果。 正如简介中所解释的，我们希望有效地计算更新数据库视图的任何关系查询 $Q$ 的增量版本。

然而，我们面临一个技术问题：$ℐ$和$𝒟$运算符是在阿贝尔群上定义的，而关系数据库通常不是阿贝尔群，因为它们是在集合上操作的。 幸运的是，数据库文献中有一个众所周知的工具，它通过使用 $\mathbb{Z}$-sets（也称为 z-relations (green-pods07, ))将集合操作转换为组操作) 而不是集合。

我们首先定义 $\mathbb{Z}$ 集组，然后回顾如何通过 $\mathbb{Z}$ 集将关系查询转换为 DBSP 电路。 使这种翻译高效增量化的原因是许多基本关系查询可以使用 LTI $\mathbb{Z}$ 集合运算符来表达。

4.1. $\mathbb{Z}$-设置为阿贝尔群

给定一个集合$A$，我们定义$\mathbb{Z}$-套 $A$ 上的函数具有从 $A$ 到 $\mathbb{Z}$ 的有限支持。 这些是函数$f:A\to \mathbb{Z}$，其中$f(x)\ne 0$最多代表有限数量的值$x\in A$。 我们还为带有 $A$ 元素的 $\mathbb{Z}$ 集的类型编写 $\mathbb{Z}[A]$。 $\mathbb{Z}[A]$ 中的值可以被视为键值映射，其中键位于 $A$ 中，值位于 $\mathbb{Z}$ 中，从而证明了数组索引符号的合理性。 我们写$f[a]$而不是$f(a)$。 由于 $\mathbb{Z}$ 是阿贝尔环，因此 $\mathbb{Z}[A]$ 也是阿贝尔环（因此是一个群）。 该组$(\mathbb{Z}[A],{+}_{\mathbb{Z}[A]},0_{\mathbb{Z}[A]},{-}_{\mathbb{Z}A})$具有逐点定义的加法和减法：$(f{+}_{\mathbb{Z}[A]}g)(x)=f(x)+g(x).\forall x\in A.$ $\mathbb{Z}[A]$ 的 $0$ 元素是 $0_{\mathbb{Z}[A]}(x)=0.\forall x\in A$ 定义的函数 $0_{\mathbb{Z}[A]}$。

特定的$\mathbb{Z}$集$m\in \mathbb{Z}[A]$可以通过枚举映射到非零值及其对应值的输入来表示：$m=\{x_1\mapsto w_1,\mathrm{\dots },x_n\mapsto w_n\}$。 我们将$w_i\in Z$称为$x_i\in A$的重数（或权重）。 重数可以是负数。 我们将$x\in m$写为$x\in A$，当且仅当$m[x]\ne 0$。

例如，让我们考虑一个由 $R=\{\text{joe}\mapsto 1,\text{anne}\mapsto -1\}$ 定义的具体 $\mathbb{Z}$-set $R\in \mathbb{Z}[\text{string}]$。 $R$ 在其域中有两个元素，joe 的重数为 1（因此 $R[\text{joe}]=1$)，以及 anne 的重数$-1$。 我们说joe$\in R$和anne$\in R$。

$\mathbb{Z}$-sets概括了sets和bags。 通过将权重 1 与每个集合元素相关联，包含 $A$ 元素的集合可以表示为 $\mathbb{Z}$ 集合。 当将集合上的查询转换为 DBSP 程序时，我们在集合和 $\mathbb{Z}$ 集合之间来回转换数据值。

Definition 4.1:

如果每个元素的重数都是一，我们就说 $\mathbb{Z}$-set 代表 set。 我们定义一个函数来检查这个属性 $\text{isset}:\mathbb{Z}[A]\to 𝔹$ 给出的：

对于我们的示例 $\text{isset}(R)=\text{false}$，从 $R[\text{anne}]=-1$ 开始。

Definition 4.2:

如果每个元素的重数都是正的，我们就说$\mathbb{Z}$集合是正（或bag)。 我们定义一个函数来检查这个属性$\text{ispositive}:\mathbb{Z}[A]\to 𝔹$。 给出的

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

自$R[\text{anne}]=-1$以来，我们有$\text{ispositive}(R)=\text{false}$。

当$m$为正时，我们写$m\ge 0$。 对于正的$m,n$，我们为$m,n\in \mathbb{Z}[A]$写$m\ge n$，当且仅当$m-n\ge 0$。 $\ge$ 是偏序。

如果函数将正值映射到正值，我们将其称为$f:\mathbb{Z}[A]\to \mathbb{Z}[B]$ 正值:$\forall x\in \mathbb{Z}[A],x\ge 0_{\mathbb{Z}[A]}\Rightarrow f(x)\ge 0_{\mathbb{Z}[B]}$。 我们也将此表示法应用于函数：$\text{ispositive}(f)$。

Definition 4.3 （不同）:

函数 $\mathrm{𝑑𝑖𝑠𝑡𝑖𝑛𝑐𝑡}:\mathbb{Z}[A]\to \mathbb{Z}[A]$ 将 $\mathbb{Z}$ 集合投影到基础集合中（但 结果仍然是 $\mathbb{Z}$ 集合)：

$\mathrm{𝑑𝑖𝑠𝑡𝑖𝑛𝑐𝑡}$“删除”具有负重数的元素。 $\mathrm{𝑑𝑖𝑠𝑡𝑖𝑛𝑐𝑡}(R)=\{\text{joe}\mapsto 1\}$。

虽然非常简单，但 $\mathrm{𝑑𝑖𝑠𝑡𝑖𝑛𝑐𝑡}$ 的定义经过精心选择，使我们能够精确定义 $\mathbb{Z}$ 集合运算符中的所有关系（集合）运算符。

从关系查询派生的电路仅在正 $\mathbb{Z}$ 集上进行计算；负值仅用于表示对$\mathbb{Z}$集的更改。 负权重从集合中“删除”元素。

§2 中的所有结果都扩展到 $\mathbb{Z}$ 集上的流。

Definition 4.4:

（单调性）如果流的每个值都为正，则流 $s\in {𝒮}_{\mathbb{Z}[A]}$ 为 正:$s[t]\ge 0.\forall t\in \mathbb{N}$。 如果$s[t]\ge s[t-1],\forall t\in \mathbb{N}$，则流$s\in {𝒮}_{\mathbb{Z}[A]}$是单调。

如果 $s\in {𝒮}_{\mathbb{Z}[A]}$ 为正，则 $ℐ(s)$ 是单调的。 如果$s\in {𝒮}_{\mathbb{Z}[A]}$是单调的，则$𝒟(s)$是正数。

概括盒箭头图

从现在开始，我们将使用电路来计算标量和流。 我们对流或标量上的函数使用相同的图形表示：带有输入和输出箭头的框。 对于标量函数，箭头的“值”是标量而不是流；否则，框作为函数应用的解释不会改变。

4.2. 实现关系运算符

事实上，关系代数可以通过 $\mathbb{Z}$ 集的计算来实现，这一事实之前已经被证明过，例如（绿色-pods07，）。 所有核心关系运算符的翻译如表4.2所示。 翻译本质上是通过查询结构的归纳给出的。

Operation	SQL example	DBSP circuit	Details
Composition	⬇ SELECT DISTINCT ... FROM (SELECT ... FROM ...)		$C_I$ circuit for inner query, $C_O$ circuit for outer query.
Union	⬇ (SELECT * FROM I1) UNION (SELECT * FROM I2)
Projection	⬇ SELECT DISTINCT I.c FROM I		$\pi \left(i\right)\left[y\right]\stackrel{\text{def}}{=}{\sum }_{x\in i,{x|}_c=y}i\left[x\right]$ ${x|}_c$ is projection on column $c$ of the tuple $x$ $\pi$ is linear; $\text{ispositive}\left(\pi \right),\text{zpp}\left(\pi \right)$.
Filtering	⬇ SELECT * FROM I WHERE p(I.c)		$P:A\to 𝔹$ is a predicate. ${\sigma }_P$ is linear; $\text{ispositive}\left({\sigma }_P\right),\text{zpp}\left({\sigma }_P\right)$.
Selection	⬇ SELECT DISTINCT f(I.c, ...) FROM I		For a function $f$ $\text{map}\left(f\right)$ is linear, $\text{ispositive}\left(\text{map}\left(f\right)\right),\text{zpp}\left(\text{map}\left(f\right)\right)$ .
Cartesian product	⬇ SELECT I1.*, I2.* FROM I1, I2		$\left(a\times b\right)\left((x,y)\right)\stackrel{\text{def}}{=}a\left[x\right]\times b\left[y\right]$. $\times$ is bilinear, $\text{ispositive}(\times ),\text{zpp}(\times )$.
Equi-join	⬇ SELECT I1.*, I2.* FROM I1 JOIN I2 ON I1.c1 = I2.c2		$\left(a\bowtie b\right)\left((x,y)\right)\stackrel{\text{def}}{=}{a\left[x\right]\times b\left[y\right]\text{if}x|}_{c1}={y|}_{c2}$. $\bowtie$ is bilinear, $\text{ispositive}(\bowtie ),\text{zpp}(\bowtie )$.
Intersection	⬇ (SELECT * FROM I1) INTERSECT (SELECT * FROM I2)		Special case of equi-join when both relations have the same schema.
Difference	⬇ SELECT * FROM I1 EXCEPT SELECT * FROM I2

表格1。 DBSP 中 SQL 关系集运算符的实现。 每个查询都假设输入 I、I1、I2 是集合，并生成输出集合。

The translation is fairly straightforward, but many operators require the application of a $\mathrm{𝑑𝑖𝑠𝑡𝑖𝑛𝑐𝑡}$ to produce sets. The correctness of this implementation is predicated on the global circuit inputs being sets as well. For example, $a\cup b=\mathrm{𝑑𝑖𝑠𝑡𝑖𝑛𝑐𝑡}(a+b)$, $a\setminus b=\mathrm{𝑑𝑖𝑠𝑡𝑖𝑛𝑐𝑡}(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:

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

$\text{toset}:\mathbb{Z}\left[A\right]\to 2^A$ is defined by $\text{toset}\left(m\right)\stackrel{\text{def}}{=}{\cup }_{x\in \mathrm{𝑑𝑖𝑠𝑡𝑖𝑛𝑐𝑡}\left(m\right)}\left\{x\right\}$.

$\text{tozset}:2^A\to \mathbb{Z}\left[A\right]$ is defined by $\text{tozset}\left(s\right)\stackrel{\text{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 $\mathrm{𝑑𝑖𝑠𝑡𝑖𝑛𝑐𝑡}$ 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 $\mathrm{𝑑𝑖𝑠𝑡𝑖𝑛𝑐𝑡}$. While we show below that $\mathrm{𝑑𝑖𝑠𝑡𝑖𝑛𝑐𝑡}$ 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}\left[I\right],\text{ispositive}\left(i\right)\Rightarrow Q\left(\mathrm{𝑑𝑖𝑠𝑡𝑖𝑛𝑐𝑡}\left(i\right)\right)=\mathrm{𝑑𝑖𝑠𝑡𝑖𝑛𝑐𝑡}\left(Q\left(i\right)\right)$.

This rule allows us to delay the application of $\mathrm{𝑑𝑖𝑠𝑡𝑖𝑛𝑐𝑡}$.

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}\left[I\right],\text{ispositive}\left(i\right)\Rightarrow \mathrm{𝑑𝑖𝑠𝑡𝑖𝑛𝑐𝑡}\left(Q\left(\mathrm{𝑑𝑖𝑠𝑡𝑖𝑛𝑐𝑡}\left(i\right)\right)\right)=\mathrm{𝑑𝑖𝑠𝑡𝑖𝑛𝑐𝑡}\left(Q\left(i\right)\right)$.

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

These properties allow us to “consolidate” distinct operators by performing one $\mathrm{𝑑𝑖𝑠𝑡𝑖𝑛𝑐𝑡}$ at the end of a chain of computations.

Finally, the next proposition shows that the incremental of $\mathrm{𝑑𝑖𝑠𝑡𝑖𝑛𝑐𝑡}$ can be computed with work proportional to the size of the input change.

Proposition 4.7:

The following circuit implements ${(\uparrow \mathrm{𝑑𝑖𝑠𝑡𝑖𝑛𝑐𝑡})}^{\mathrm{\Delta }}$: $\cong$

where $H:\mathbb{Z}\left[A\right]\times \mathbb{Z}\left[A\right]\to \mathbb{Z}\left[A\right]$ 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. Translate $Q$ into a circuit using the rules in Table 4.2.

2. Apply $\mathrm{𝑑𝑖𝑠𝑡𝑖𝑛𝑐𝑡}$ consolidation until convergence.

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

4. Incrementalize the whole circuit “surrounding” it with $ℐ$ and $𝒟$.

5. Apply the chain rule and other properties of the ${\cdot }^{\mathrm{\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:

Step 2: we apply the $\mathrm{𝑑𝑖𝑠𝑡𝑖𝑛𝑐𝑡}$ optimization rules; first the rule from 4.6 gives us the following equivalent circuit:

Applying the rule from 4.5 we get:

And applying again 4.6 we get:

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:

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:

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

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

Finally, replace the incremental join using the formula for bilinear operators (Theorem 3.4), and the incremental $\mathrm{𝑑𝑖𝑠𝑡𝑖𝑛𝑐𝑡}$ (Proposition 4.7), obtaining the circuit from Figure 1.

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 $\mathrm{𝑑𝑖𝑠𝑡𝑖𝑛𝑐𝑡}$. 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\left[t\right]\stackrel{\text{def}}{=}\Vert s\left[t\right]\Vert$ the size of the value of stream $s$ of changes at time $t$, and $R\left[t\right]\stackrel{\text{def}}{=}\Vert ℐ\left(s\right)\left[t\right]\Vert$ the size of the relation produced by integrating all changes in $s$. An unoptimized incremental operator $Q^{\mathrm{\Delta }}=𝒟\circ Q\circ ℐ$ 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\left[t\right]$. In addition, because of the $ℐ$ and $𝒟$ operators, it uses $O\left(R\left[t\right]\right)$ memory.

The optimizations described in §3 reduce the reduce the time complexity of an operator to be a function of $C\left[t\right]$. Assuming $C\left[t\right]\ll R\left[t\right]$, this translates to major performance improvements in practice. For example, Theorem 3.3, allows evaluating $T^{\mathrm{\Delta }}$, where $T$ is a linear operator, in time $O\left(C\left[t\right]\right)$. Interestingly, while the $ℐ$ operator uses $O\left(R\left[t\right]\right)$ memory, it can be evaluated in $O\left(C\left[t\right]\right)$ 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 $\mathrm{𝑑𝑖𝑠𝑡𝑖𝑛𝑐𝑡}$ operator is not linear, ${(\uparrow \mathrm{𝑑𝑖𝑠𝑡𝑖𝑛𝑐𝑡})}^{\mathrm{\Delta }}$ can also be evaluated in $O\left(C\left[t\right]\right)$ 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\left(C{\left[t\right]}^2\right)$ (Theorem 3.4).

The space complexity of linear operators is 0 (zero), since they store no data persistently. The space complexity of ${(\uparrow \mathrm{𝑑𝑖𝑠𝑡𝑖𝑛𝑐𝑡})}^{\mathrm{\Delta }}$ and join is $O\left(R\left[t\right]\right)$.

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 {𝒮}_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\ge t_0.s\left[t\right]=0$. Denote the set of streams that are zero almost everywhere by $\overline{{𝒮}_A}$.

Stream introduction

The delta function (named from the Dirac delta function) ${\delta }_0:A\to {𝒮}_A$ produces a stream from a scalar value: δ_0(v)[t] =def{ vif t = 00_A otherwise For example, ${\delta }_0\left(5\right)$ is the stream .

Stream elimination

We define the function $\int :\overline{{𝒮}_A}\to A$, over streams that are zero almost everywhere, as $\int \left(s\right)\stackrel{\text{def}}{=}{\sum }_{t\ge 0}s\left[t\right]$. $\int$ is closely related to $ℐ$; if $ℐ$ is the indefinite integral, $\int$ is the definite integral on the interval $0-\mathrm{\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={\mathrm{𝑖𝑑}}_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:

Proposition 5.3:

If $Q$ is time-invariant, the circuit above has the zero-preservation property: $\text{zpp}\left(\int \circ Q\circ {\delta }_o\right)$.

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. 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):

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

   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 $𝒟$ 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 $\text{isset}\left(\text{I}\right)$, 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 $\mathrm{𝑑𝑖𝑠𝑡𝑖𝑛𝑐𝑡}$ 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\left(x\right)=\lambda x.R(\text{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 $\text{ispositive}\left(S\right)$, so $\forall x.S\left(x\right)\ge x$. Also $\uparrow S$ is time-invariant, so $S\left(0\right)=0$. From §4, the definition of set union we know that $x\cup y=\mathrm{𝑑𝑖𝑠𝑡𝑖𝑛𝑐𝑡}\left(x+y\right)$. 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\left[t\right]=S^t\left(0\right)$, where by $S^t$ we mean $\underset{t}{\underbrace{S\circ S\circ \mathrm{\dots }\circ S}}$. $S$ is monotone; thus, if there is a time $k$ such that $S^k\left(0\right)=S^{k+1}\left(0\right)$, we have $\forall j\in \mathbb{N}.S^{k+j}\left(0\right)=S^k\left(0\right)$. 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: $\text{O}=\text{fix}x.R(\text{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 $ℐ$ and $𝒟$ operators. Using the cycle rule of Proposition 3.2 we can rewrite this circuit as:

(5.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 }^{\mathrm{\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, ${𝒮}_{{𝒮}_A}=\mathbb{N}\to \left(\mathbb{N}\to A\right)$, 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:{𝒮}_A\to {𝒮}_B$ yields an operator over nested streams $\uparrow S:{𝒮}_{{𝒮}_A}\to {𝒮}_{{𝒮}_B}$, such that $(\uparrow S)\left(s\right)=S\circ s$, or, pointwise: $(\uparrow S\left(s\right))\left[t_0\right]\left[t_1\right]=S\left(s\left[t_0\right]\right)\left[t_1\right],\forall t_0,t_1\in \mathbb{N}$. In particular, a scalar function $f:A\to B$ can be lifted twice to produce an operator between streams of streams: $\uparrow \uparrow f:{𝒮}_{{𝒮}_A}\to {𝒮}_{{𝒮}_B}$.

We define a partial order over timestamps: $(i_0,i_1)\le (t_0,t_1)$ iff $i_0\le t_0$ and $i_1\le t_1$. We extend the definition of strictness for operators over nested streams: a stream operator $F:{𝒮}_{{𝒮}_A}\to {𝒮}_{{𝒮}_B}$ is strict if for any $s,s^{\prime }\in {𝒮}_{{𝒮}_A}$ and all times $t,i\in \mathbb{N}\times \mathbb{N}$ we have implies $F\left(s\right)\left[t\right]=F\left(s^{\prime }\right)\left[t\right]$. Proposition 2.10 holds for this notion of strictness, i.e., the fixed point operator $\text{fix}\alpha .F\left(\alpha \right)$ is well defined for a strict operator $F$.

Proposition 6.1:

The operator $\uparrow z^{-1}:{𝒮}_{{𝒮}_A}\to {𝒮}_{{𝒮}_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 $ℐ$ operator on ${𝒮}_{{𝒮}_A}$ operates on rows of the matrix, treating each row as a single value. Lifting a stream operator computing on ${𝒮}_A$, such as $ℐ:{𝒮}_A\to {𝒮}_A$, also produces an operator on nested streams, but this time computing on the columns of the matrix $\uparrow ℐ:{𝒮}_{{𝒮}_A}\to {𝒮}_{{𝒮}_A}.$

Proposition 6.2 (Lifting cycles):

For a binary, causal $T$ we have: $\uparrow (\lambda s.\text{fix}\alpha .T(s,z^{-1}\left(\alpha \right)))=\lambda s.\text{fix}\alpha .(\uparrow T)(s,(\uparrow z^{-1})\left(\alpha \right))$ 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:

$\cong$

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 $\mathrm{𝑑𝑖𝑠𝑡𝑖𝑛𝑐𝑡}\circ R$ with $T$:

Lift the entire circuit using Proposition 6.2 and incrementalize it:

Now apply the chain rule to this circuit:

(6.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:

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$:

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$.)

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

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:

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:

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

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 }^{\mathrm{\Delta }}$ invocations:

There are two applications of ${\cdot }^{\mathrm{\Delta }}$ left in this circuit: ${(\uparrow {(\uparrow \bowtie )}^{\mathrm{\Delta }})}^{\mathrm{\Delta }}$ and ${(\uparrow {(\uparrow \mathrm{𝑑𝑖𝑠𝑡𝑖𝑛𝑐𝑡})}^{\mathrm{\Delta }})}^{\mathrm{\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 )}^{\mathrm{\Delta }}$, given by Theorem 3.4:

$\cong$

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

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

We now have three applications of ${(\uparrow \uparrow \bowtie )}^{\mathrm{\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:

Returning to ${(\uparrow {(\uparrow \mathrm{𝑑𝑖𝑠𝑡𝑖𝑛𝑐𝑡})}^{\mathrm{\Delta }})}^{\mathrm{\Delta }}$, we can compute its circuit by expanding once using Proposition 4.7:

$\cong$

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

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 {𝒮}_{{𝒮}_A},s\left[t_1\right]\left[t_2\right]$, where $t_1$ is the input timestamp and $t_2$ is the fixed point iteration number. The unoptimized loop body ${(\uparrow {(\uparrow T)}^{\mathrm{\Delta }})}^{\mathrm{\Delta }}=𝒟\circ \uparrow 𝒟\circ \uparrow \uparrow T\circ \uparrow ℐ\circ ℐ$ has the same time complexity as $T$ applied to the aggregated input of size $R\left(s\right)\left[t_1\right]\left[t_2\right]\stackrel{\text{def}}{=}\parallel (\uparrow ℐ\circ ℐ)\left(s\right)\left[t_1\right]\left[t_2\right]\parallel =\parallel {\sum }_{(i_1,i_2)\le (t_1,t_2)}s\left[i_1\right]\left[i_2\right]\parallel$. 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 )}^{\mathrm{\Delta }})}^{\mathrm{\Delta }}{s}_2$ that runs in $O(\parallel \uparrow ℐ\left(s1\right)\left[t1\right]\left[t2\right]\parallel \times \parallel ℐ\left(s2\right)\left[t1\right]\left[t2\right]\parallel )\ll O(R\left(s_1\right)\times R\left(s_2\right))$ (because each term is correspondingly smaller).

Space complexity

Integration ($ℐ$) and differentiation ($𝒟$) of a stream $s\in {𝒮}_{{𝒮}_A}$ uses memory proportional to ${\sum }_{t_2}\Vert {\sum }_{t_1}s\left[t_1\right]\left[t_2\right]\Vert$, 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 $ℐ$ and $𝒟$ 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 $\mathrm{𝑑𝑖𝑠𝑡𝑖𝑛𝑐𝑡}$, 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}\left[A\right]\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_{\text{COUNT}}:\mathbb{Z}\left[A\right]\to \mathbb{Z}$, which computes a sum of all the element weights: $a_{\text{COUNT}}\left(s\right)={\sum }_{x\in s}s\left[x\right]$. The SQL SUM aggregation function is implemented on $\mathbb{Z}$-sets by $a_{\text{SUM}}:\mathbb{Z}\left[ℝ\right]\to ℝ$ which performs a weighted sum of all (real) values: $a_{\text{SUM}}\left(s\right)={\sum }_{x\in s}x\times s\left[x\right]$.

With this definition the aggregation functions $a_{\text{COUNT}}$ and $a_{\text{SUM}}$ are in fact linear transformations between the group $\mathbb{Z}\left[A\right]$ and the result group ($\mathbb{Z}$, and $ℝ$ 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 $\text{makeset}:A\to \mathbb{Z}\left[A\right]$ function, which converts a scalar value of type $A$ to a singleton $\mathbb{Z}$-set, defined as follows: $\text{makeset}\left(x\right)\stackrel{\text{def}}{=}1\cdot x$.

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

The lifted incremental version of this circuit is interesting: since $\pi$ and $a_{\text{SUM}}$ are linear, they are equivalent to their own incremental versions. Although ${(\uparrow \text{makeset})}^{\mathrm{\Delta }}=𝒟\circ \uparrow \text{makeset}\circ ℐ$ 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

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_{\text{MIN}})}^{\mathrm{\Delta }}=𝒟\circ \uparrow a_{\text{MIN}}\circ ℐ$. Such functions perform work proportional to $R\left(s\right)$ 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}\left[A\right]=\mathbb{Z}\left[A\right]\left[K\right]$. We call values $i$ of this structure indexed $\mathbb{Z}$-sets: for each key $k\in K$, $i\left[k\right]$ is a $\mathbb{Z}$-set. Because the codomain $\mathbb{Z}\left[A\right]$ 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}\left[A\right]\to \left(K\to \mathbb{Z}\left[A\right]\right)$ as $G_p\left(a\right)\left[k\right]\stackrel{\text{def}}{=}{\sum }_{x\in a.p\left(x\right)=k}a\left[x\right]\cdot x$. When applied to a $\mathbb{Z}$-set $a$ this function returns a indexed $\mathbb{Z}$-set, where each element is called a grouping: 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\left(s\right)$ that returns the first letter of the string $s$. Then we have that $G_p\left(R\right)=\left\{\text{j}\mapsto \left\{\text{joe}\mapsto 1\right\},\text{a}\mapsto \left\{\text{anne}\mapsto -1\right\}\right\}$, 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:\left(K\times \mathbb{Z}\left[A\right]\right)\to B$ that produces values in some group $B$, and an indexed relation of type $\mathbb{Z}\left[A\right]\left[K\right]$, as defined above in §7.3. The nested relation aggregation operator $Ag{g}_a:\mathbb{Z}\left[A\right]\left[K\right]\to B$ applies $a$ to the contents of each grouping independently and adds the results: $Ag{g}_a\left(g\right)\stackrel{\text{def}}{=}{\sum }_{k\in K}a(k,g\left[k\right])$. 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}\left[A\right]$, $count(k,s)=\text{makeset}\left((k,a_{\text{COUNT}}\left(s\right))\right)$, using the $\mathbb{Z}$-set counting function $a_{\text{COUNT}}$ from §7.2; the notation $(a,b)$ is a pair of values $a$ and $b$. Then we have $Ag{g}_{count}\left(G_p\left(R\right)\right)=\left\{(\text{j},1)\mapsto 1,(\text{a},-1)\mapsto 1\right\}$.

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: $\text{flatmap}:\mathbb{Z}\left[A\right]\left[K\right]\to \mathbb{Z}\left[A\times K\right]$. flatmap is in fact a particular instance of aggregation, using the aggregation function $a:K\times \mathbb{Z}\left[A\right]\to \mathbb{Z}\left[A\times K\right]$ defined by $a(k,s)={\sum }_{x\in s\left[k\right]}s\left[k\right]\left[x\right]\cdot (k,x).$ For our previous example, $\text{flatmap}\left(G_p\left(R\right)\right)=\left\{(\text{j},\text{joe})\mapsto 1,(\text{a},\text{anne})\mapsto -1\right\}$.

If we use an aggregation function $a:K\times Z\left[A\right]$ that is linear in its second argument, then the aggregation operator $Ag{g}_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:

7.6. Streaming joins

Consider a binary query $T(s,t)=ℐ\left(s\right)\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.

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 {𝒮}_{\mathbb{Z}\left[A\right]}$ and an additional input $\theta \in {𝒮}_{ℝ}$ that carries the value of the current time.

$\cong$

where the window operator $W$ prunes input $\mathbb{Z}$-sets, only keeping values with timestamps less than an hour behind $\theta \left[t\right]$. Assuming $ts:A\to ℝ$ returns the physical timestamp of a value, $W$ is defined as $W(v,\theta )\left[t\right]\stackrel{\text{def}}{=}\{x\in v\left[t\right].ts\left(x\right)\ge \theta \left[t\right]-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:

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$:

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\left(R+\mathrm{\Delta }R\right)=Q\left(R\right)+\mathrm{\Delta }Q(R,\mathrm{\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 }^{\mathrm{\Delta }}$, which transforms a stream computation $Q$ into its incremental version $Q^{\mathrm{\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.

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 {𝒮}_{{𝒮}_{\mathbb{N}}}$ defined by $i\left[t_0\right]\left[t_1\right]=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 $ℐ$ operator on ${𝒮}_{{𝒮}_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 ${𝒮}_A$, such as $ℐ:{𝒮}_A\to {𝒮}_A$, also produces an operator on nested streams, but this time computing on the columns of the matrix: $\uparrow ℐ:{𝒮}_{{𝒮}_A}\to {𝒮}_{{𝒮}_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 $𝒟$ to nested streams $𝒟:{𝒮}_{{𝒮}_A}\to {𝒮}_{{𝒮}_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 𝒟:{𝒮}_{{𝒮}_A}\to {𝒮}_{{𝒮}_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: $ℐ\circ (\uparrow ℐ)=(\uparrow ℐ)\circ ℐ$ and $𝒟\circ (\uparrow 𝒟)=(\uparrow 𝒟)\circ 𝒟$.

(↑$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_${𝒮}_{{𝒮}_{\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_${𝒮}_{{𝒮}_{\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]⋯]⋮]