#  JanusAQP: Efficient Partition Tree Maintenance for Dynamic Approximate
Query Processing

Xi Liang  University of Chicago<EMAIL_ADDRESS>,  Stavros Sintos
University of Chicago<EMAIL_ADDRESS>and  Sanjay Krishnan  University of
Chicago<EMAIL_ADDRESS>
###### Abstract.

Approximate query processing over dynamic databases, i.e., under
insertions/deletions, has applications ranging from high-frequency trading to
internet-of-things analytics. We present JanusAQP, a new dynamic AQP system,
which supports SUM, COUNT, AVG, MIN, and MAX queries under insertions and
deletions to the dataset. JanusAQP extends static partition tree synopses,
which are hierarchical aggregations of datasets, into the dynamic setting.
This paper contributes new methods for: (1) efficient initialization of the
data synopsis in the presence of incoming data, (2) maintenance of the data
synopsis under insertions/deletions, and (3) re-optimization of the
partitioning to reduce the approximation error. JanusAQP reduces the error of
a state-of-the-art baseline by more than 60% using only 10% storage cost.
JanusAQP can process more than 100K updates per second in a single node
setting and keep the query latency at a millisecond level.

##  1\. Introduction

Approximate query processing (AQP) studies principled ways to sacrifice query
result accuracy for faster or more resource-efficient execution
(chaudhuri2017approximate, ; garofalakis2001approximate, ). AQP systems
generally employ reduced-size summaries, or “synopses”, of large datasets that
are faster to process. The simplest of such synopsis structures are histograms
and samples (cormode2011synopses, ; liang2021combining, ; agarwal2013blinkdb,
; lazaridis2001progressive, ), but many others have been proposed in the
literature. More complex synopses are more accurate for specific types of
queries (walenz2019learning, ), specific data settings (poepselland, ), or
even are learned with machine learning models (yang2019deep, ;
hilprecht2019deepdb, ; ma2021learned, ). AQP is particularly interesting and
challenging in a dynamic data setting, where a dataset is continuously
modified with insertions and deletions (garofalakis2001approximate, ;
olma2019taster, ; acharya1999aqua, ). In this setting, hereafter denoted as
DAQP, any synopsis data structures have to be continuously maintained online.

As an example use-case, consider a database aggregating per-stock order data
for the NASDAQ exchange (nasdaqbv, ). Suppose, that we would like to build a
low-latency SQL interface for approximate aggregate queries over the past
seven days of order data. On a typical day, there are 25M new orders that
correspond to trades that are placed by brokers (up to 70,000 orders in any
given second). A decent fraction of these orders are eventually canceled or
prematurely terminated, for a variety of financial reasons. Thus, this
database is highly dynamic with a large volume of new insertions (new orders)
and a small but significant number of deletions (canceled orders). This paper
explores such scenarios with similar motivating applications in internet-of-
things monitoring and enterprise stream processing.

Simple synopses like 1D histograms and uniform samples are easy to maintain
dynamically. However, such structures are often inaccurate in high-dimensional
data and selective query workloads. More complex synopses structures, e.g,
(yang2019deep, ; liang2021combining, ) can be optimized for a particular
instance (dataset and query workload), but are generally harder to maintain
online. For example, recently proposed learned synopses require expensive
retraining procedures which limit insertion/deletion throughput (yang2019deep,
; hilprecht2019deepdb, ; ma2021learned, ). Even classical stratified samples
may have to be periodically re-optimized and re-balanced based on query and
workload shifts (agarwal2013blinkdb, ). These, expensive (re-)initialization
procedures can significantly hurt insertion throughput, and accordingly,
almost all existing AQP systems focus on the static data warehousing
setting111A notable exception being the AQUA project (acharya1999aqua, ) from
20 years ago.. Unfortunately, the existing techniques that _are_ designed for
dynamic data, such as sketches and mergeable summaries (gan2020coopstore_, ;
agarwal2012mergeable, ; poepselland, ), often cannot handle arbitrary
deletions or aggregation queries with arbitrary predicates easily. Thus, it is
understood that most synopsis data structures have at least one of the
following pitfalls in our desired dynamic setting: throughput, drift, or
generality (chaudhuri2017approximate, ).

Figure 1. JanusAQP manages a collection of DPT synopses by maintaining them
online while periodically re-optimizing partitioning and sample allocation.

This paper explores the DAQP problem and studies ways that we can mitigate the
pitfalls of prior approaches with a flexible synopsis data structure that can
continuously re-optimize itself. We present JanusAQP, a new DAQP system, which
supports SUM, COUNT, AVG, MIN, and MAX queries with predicates under arbitrary
insertions and deletions to the dataset. The main data structure in JanusAQP
is a dynamic extension of our recently published work (liang2020fast, ;
liang2021combining, ), which we call a Dynamic Partition Tree (DPT). DPT is a
two-layer synopsis structure that consists of a: (1) hierarchical partitioning
of a dataset into a tree, and (2) a uniform sample of data for each of the
leaf partitions (effectively a stratified sample over the leaves). An
optimizer determines the best partitioning conditions and sample allocations
to meet a user’s performance goals. For each partition (nodes in the tree), we
calculate the SUM, COUNT, MIN, and MAX values of the partition. Any desired
SUM, COUNT, AVG, MIN, and MAX query can be efficiently decomposed into two
parts with the structure: a combination of the partial aggregates where the
predicate fully covers a partition in the tree, and an approximate part where
the predicate partially covers a leaf node (and can be estimated with a
sample). More importantly, this structure is essentially a collection of
materialized views and samples, which can be maintained incrementally.

A core contribution of JanusAQP is online synopsis optimization. JanusAQP
continuously monitors the accuracy of all of its DPT synopses to account for
data and workload drift. When a synopsis is no longer accurate, it triggers a
re-optimization procedure that resamples and repartitions the data. This re-
optimization problem is both a significant algorithmic and systems challenge.
From an algorithmic perspective, JanusAQP needs an efficient way to determine
the optimal partitioning conditions in dynamic data. We propose an efficient
algorithm based on a dynamic range tree index that finds a partitioning that
controls the minimax query error (up to an approximation factor). From a
systems perspective, re-optimization poses a bit of a logistical challenge.
New data will arrive as the new synopsis data structure is being constructed.
We design an efficient multi-threaded catch-up processing algorithm that
synchronizes new data and historical data without sacrificing the statistical
rigor of the estimates.

Our prototype version of JanusAQP is integrated with the message-broker
framework, Apache Kafka. Insertions, deletions, and user queries are processed
as Kafka topics allowing for a multi-threaded DAQP server. In our experiments,
we show that DPT is significantly more accurate than other baselines and
state-of-the-art systems such as, reservoir sampling, stratified reservoir
sampling, and the DeepDB system (hilprecht2019deepdb, ). It also achieves a
throughput of processing nearly 200k records per second, while serving sub-
millisecond query latencies.

Overall, the new system JanusAQP we proposed has the following benefits over
previously known indexes for approximate query processing. It handles dynamic
updates efficiently (comparing to the static indexes PASS (liang2021combining,
), and VecrdictDB (park2018verdictdb, )), it provides theoretical guarantees
on the confidence intervals (comparing to the machine learning based indexes
such that DeepDB (hilprecht2019deepdb, )), the query procedure accesses only a
small synopsis of data and does not touch the original data set so the
communication throughput is low (comparing to other tree-based indexes such as
(joshi2008materialized, ; jurgens1998r, )), and the estimation error is always
low without making any assumption about the spatial/value-domain distribution
of the data (comparing to (lazaridis2001progressive, )). We highlight all
benefits/differences in the related work.

##  2\. Background

We first introduce the core concepts behind the synopses used in this work.

###  2.1. Dynamic Approximate Query Processing

We assume an initial database table $\mathcal{D}^{(0)}$. This table
$\mathcal{D}^{(0)}$ is continuously modified through a stream of insertions
and deletions of tuples. As a design principle, we assume that insertions are
common but deletions are rare. With each insertion or deletion operation, the
table evolves over time with a new state at each time step $i$:
$\mathcal{D}^{(0)},\mathcal{D}^{(1)},\ldots,\mathcal{D}^{(i)},\mathcal{D}^{(i+1)},\ldots$
A synopsis is a data structure that summarizes the evolving table. For each
$\mathcal{D}^{(i)}$, there is a corresponding synopsis $\Sigma^{(i)}$:
$\Sigma^{(0)},\Sigma^{(1)},\ldots,\Sigma^{(i)},\Sigma^{(i+1)},\ldots$

In DAQP, the problem is to answer queries as best as possible from only the
$\Sigma^{(i)}$. For a query $q$, the estimation error is defined as the
difference between the estimated result (using the synopsis) and the true
result (using the current database state):

$\textsf{Error}(q,\Sigma^{(i)})=|q(\mathcal{D}^{(i)})~{}~{}-~{}~{}q(\Sigma^{(i)})|$

We further assume that there is sufficient cold/archival storage to store the
current state of the table $\mathcal{D}^{(i)}$. This data can be accessed in
an offline way for initialization, re-optimization, and logging purposes but
not for query processing.

Figure 2. The core data structure in JanusAQP is based on the PASS data
structure (liang2021combining, ) that summarizes a dataset with a tree of
aggregates at different levels of resolution (granularity of partitioning).
Associated with the leaf nodes are stratified samples. The two stage synopsis
structure can be optimally partitioned to minimize error.

There are a few notable differences from the “streaming” setting. First, most
data streaming models do not support arbitrary record deletion, i.e., as
studied in (poepselland, ). We find that in many use-cases limited support for
deletion is needed due to records that are invalidated through an out-of-band,
asynchronous data process like fraud detection or financial auditing. Next,
most streaming settings enforce a single pass over the data with limited
overall memory. We do not make this assumption and allow for archival storage
and slow access to old data. This is a more realistic AQP setting where all
data are stored, however, there is limited working memory for a fast,
approximate query answering service.

###  2.2. Related work

There is significant research in histograms and their variants that is highly
relevant to this project (jagadish1998optimal, ; koudas2000optimal, ;
jagadish2001global, ). V-Optimal histograms construct buckets to minimize the
cumulative variance (jagadish1998optimal, ). There are works on multi-
dimensional histograms (lazaridis2001progressive, ), and histograms on the
streaming/dynamic setting (guha2006approximation, ; gilbert2002fast, ). Like
histograms, JanusAQP constructs partitions over attribute domains and
aggregates within the partition. However, we contribute different partition
optimization criteria than typically used in histograms and novel techniques
based on geometric data structures to scale partitioning into higher
dimensions. Furthermore, our system works in the general dynamic setting,
unlike (gilbert2002fast, ) where the number of total items must remain the
same. Another related area of research is into mergeable summaries that
compute a partition of the data and optimize sampling at a data partition
level (rong2020approximate, ; liang2020fast, ; agarwal2012mergeable, ;
gan2020coopstore, ; poepselland, ). The DPT used in JanusAQP very much behaves
like a mergeable summary but a far greater breadth of downstream queries.
Furthermore, some prior work mostly focuses on a streaming setting without
support for deletion (poepselland, ). Similarly, sketches (cormode2011sketch,
; cormode2011synopses, ) have been used to find a summary of data to answer
approximately a variety of queries efficiently. However, they also do not
handle arbitrary range queries using space independent of the size of the full
database. Mergeable summaries and sketches usually focus on optimizing
different types of problems such that frequency queries, percentile queries,
etc. This paper shows how to operationalize a general DAQP system for
aggregation queries with both systems and algorithmic contributions relating
to the design of dynamic synopses and their continuous optimization. Our
system can handle arbitrary updates and can estimate any arbitrary predicate
query with provable confidence intervals.

In databases, a number of tree-based indexes, such as the improved $R^{*}$
tree (jurgens1998r, ), have been used to support range aggregation queries
efficiently. The space of such indexes is (super-)linear with respect to the
input items. The query procedures need to have access to the entire tree-index
that contains the entire dataset. That leads to high communication throughput
or high I/O operations comparing to JanusAQP where the queries are executed in
a small synopsis of data stored in a local machine or RAM with zero
communication throughput. In another line of work, tree-based data structured
are used to return a set of $k$ uniform samples in a query range. More
specifically, in (joshi2008materialized, ; wang2015spatial, ) the authors
construct indexes such that given a query range $Q$ and a parameter $k$, they
return $k$ uniform samples from the input items that lie inside $Q$. These
samples can be used to estimate any aggregation query in the range query $Q$.
There are several issues with these indexes in our setting. First, the design
of the index in (joshi2008materialized, ) makes their structure inherently
static and it cannot be maintained efficiently. Furthermore, the estimation
error in both indexes is the same as the error in the simple uniform random
sampling schema. In Section 6, we show that the error of our new index in real
data sets is always less than half of the error in uniform random sampling, so
our new index always outperforms these range sampling indexes. In addition,
the communication throughput or the I/O operations during a query procedure of
these indexes depend on $N$, i.e. the size of the input set, so they cannot be
used on big data. Finally, the dynamic tree structure in
(lazaridis2001progressive, ) can store a synopsis of data in a tree-based
index and use only this synopsis/index to return estimations of a range
aggregation queries, which is also the case in our system JanusAQP. However,
there are two main differences. The index in (lazaridis2001progressive, )
returns a good estimation only if an assumption about the spatial/value-domain
distribution of the data is made, while JanusAQP uses stratified sampling and
it always returns unbiased estimators with small error without assuming any
distribution over the data. Furthermore, while their partition tree in
(lazaridis2001progressive, ) can handle dynamic updates, its
structure/partitioning remains unchanged. In our index we maintain a near-
optimal partitioning over the updates. As we show in Section 6.7, running
experiments on real data, re-partitioning is essential in order to maintain a
small error.

Dynamic AQP problems have been discussed in prior work
(garofalakis2001approximate, ), however, most existing systems have focused on
a static data warehousing setting (agarwal2013blinkdb, ). The Aqua system
(acharya1999aqua, ) did consider the maintenance of its synopsis data
structures under updates. However, these synopses were relatively simple and
only samples and histograms. Furthermore, we discuss systems issues such as
catch-up processing that was not discussed in (acharya1999aqua, ) or any
subsequent work (gibbons2002fast, ).

Many new AQP techniques use machine learning. The basic ideas exist for a
while, e.g., (jermaine2003robust, ; jin2006new, ). Recently, there are more
comprehensive solutions that train from a past query workload
(park2017database, ) or directly build a probabilistic model of the entire
database (hilprecht2019deepdb, ; yang2019deep, ). We show that these systems
are not optimized for a dynamic setting. Even when they can be updated
efficiently with warm-start training, their throughput is much lower than
JanusAQP.

###  2.3. Partition Trees for AQP

We propose a new dynamic data synopsis and optimization strategy that is an
extension of our previous work (liang2021combining, ). In particular, we
proposed a system called PASS (which we call SPT for “static partition tree”).
SPT synopses are related to works such as (lazaridis2001progressive, ) in the
data cube literature and hybrid AQP techniques (peng2018aqp++, ). We showed
that with appropriate optimization of the partitioning conditions, an SPT
could achieve state-of-the-art accuracy in AQP problems.

####  2.3.1. Construction

An SPT is a synopsis data structure used for answering aggregate queries over
relational data. To use SPT, the user defines an _aggregation column_
(numerical attribute to aggregate) and a _set of predicate columns_ (columns
over which filters will be applied). An SPT consists of two pieces: (1) a
hierarchical aggregation of a dataset, and (2) a uniform sample of data for
each of the leaf partitions (effectively a stratified sample over the leaves).
The system returns a synopsis that can answer SUM, COUNT, AVG, MIN, and MAX
aggregates over the aggregation column filtered by the predicate columns.
Figure 2 illustrates a partition tree synopsis over toy stock-order data.

To understand how this structure is useful, let us overview some of its formal
properties. A _partition_ of a dataset $\mathcal{D}$ is a decomposition of
$\mathcal{D}$ into disjoint parts $\mathcal{D}_{1},...,\mathcal{D}_{B}$. Each
$\mathcal{D}_{i}$ has an associated partitioning condition, a predicate that
when applied to the full dataset as a filter retrieves the full partition.
Partitions naturally form a hierarchy and can be further subdivided into even
more partitions, which can then be subdivided further. A _static partition
tree_ $\mathcal{T}$ is a tree with $B$ nodes (where each node corresponds to a
partition) with the following invariants: (1) every child is a subset of its
parent, (2) all siblings are disjoint, and (3) the union of all siblings
equals the parent.

In an SPT, each node of the tree is associated with SUM, COUNT, MIN, and MAX
statistics over the items in $\mathcal{D}$ that lie inside the node. SPT
synopses have a flexible height to tradeoff accuracy v.s. storage. In shorter
trees, the leaf nodes of an SPT can cover large subsets of data and vice versa
in deeper trees. Note how each layer of the tree in Figure 2 aggregates the
lower layer over coarser-and-coarser aggregation conditions (first by “sector”
and then by “order type”).

This structure works well when the queries align with partition boundaries.
For example, a user aggregating total orders by “order type” in Figure 2 would
get an exact answer with no approximation. The challenge is to answer queries
with predicates that partially intersect partitions. Due to the tree
invariants, the set of partial intersections can be fully determined at the
leaf nodes. To estimate the contributions of these partial intersections, an
SPT associates a uniform sample of tuples _within that partition_ for each
leaf node.

####  2.3.2. Query Processing

Using an SPT, a user can estimate the result of a query as follows.
Essentially, the query processing algorithm identifies “fully covered” nodes
that are contained in the query predicate and “partially covered” ones that
overlap in some way. Exact statistics from the “fully covered” nodes can be
used, while estimates can be used to determine the contribution of “partially
covered” ones. We present SUM, COUNT, AVG for brevity, but it is also possible
to get estimations for MIN and MAX.

Step 1: Frontier Lookup.  Given a query predicate $q$, traverse the tree top-
down to retrieve two sets of nodes partitions: $R_{cover}$ (nodes that fully
cover the predicate) and $R_{partial}$ (nodes that partially intersect the
predicate). Nodes that do not intersect the predicate can be ignored.

Step 2: Partial Aggregation  For each partition in $R_{cover}$, we can compute
an exact “partial aggregate” for the tuples in those partitions. For a
SUM/COUNT query $q$: $agg=\sum_{R_{i}\in R_{cover}}SUM(R_{i})$, for an AVG
query, we weight the average by the relative size of the partition:
$agg=\sum_{R_{i}\in R_{cover}}SUM(R_{i})\frac{N_{i}}{N_{q}}$, where $N_{i}$ is
the size of the partition $R_{i}$, $N_{q}$ is the total size in all relevant
partitions of query $q$, and $SUM(R_{i})=\sum_{t\in R_{i}\cap\mathcal{D}}t.a$
is the sum of the aggregation values of all tuples in the partition $R_{i}$.

Step 3: Sample Estimation.  Each partition in $R_{partial}$ is a leaf node
with an associated stratified sample. Within each stratified sample, we use
standard AQP techniques to estimate that partition’s contribution to the final
query result (agarwal2013blinkdb, ). For completeness, we include those
calculations here. Suppose a partition $R_{i}$ has a set $S_{i}$ of $m_{i}$
samples and there are $N_{i}$ total tuples in $R_{i}$. We can formulate COUNT,
SUM, AVG as calculating an average over transformed attributes:
$f(S_{i})=\frac{1}{m_{i}}\sum_{t\in S_{i}}\phi_{q}(t)$, where
$\phi_{q}(\cdot)$ expresses all the necessary scaling to translate the samples
in query $q$ into an average query population. In particular, if we define
$Predicate(t,q)=1$ if tuple $t$ satisfies the predicate of query $q$, and $0$
otherwise, we have

  * •

COUNT: $\phi_{q}(t)=Predicate(t,q)\cdot N_{i}$

  * •

SUM: $\phi_{q}(t)=Predicate(t,q)\cdot N_{i}\cdot t.a$

  * •

AVG: $\phi_{q}(t)=Predicate(t,q)\cdot\frac{m_{i}}{\sum_{t\in
S_{i}}Predicate(t,q)}\cdot t.a$

We run such a calculation for each partition that is partially covered. These
results are combined with a weighted combination like before. For SUM/COUNT
queries it is: $samp=\sum_{R_{i}\in R_{partial}}f(S_{i})$. And for AVG
queries, it is: $samp=\sum_{R_{i}\in
R_{partial}}f(S_{i})\cdot\frac{N_{i}}{N_{q}}$. $N_{i}$ and $N_{q}$ can be
exactly retrieved from the statistics computed for each partition.

Step 4: Final Estimate.  The results can be found by taking a sum of the two
parts: $result=samp+agg.$ For this result estimate, confidence intervals can
be calculated using standard stratified sampling formulas.

PASS and JanusAQP comparison.  As we noted, JanusAQP is an extension of PASS
in the dynamic setting. However, there are key differences between the two
systems, since PASS is an inherently static index that cannot handle dynamic
updates. The main differences and novelties of our new system JanusAQP
comparing to PASS are the following: i) PASS finds a static partitioning that
is not changing after insertions and deletions of items. In JanusAQP we
propose algorithms (Subsection 5.4) that automatically check if a re-
partitioning is needed after the dynamic updates. As we show empirically in
Section 6 the re-partitioning method is very important and leads to much lower
errors comparing to PASS with a fixed partition tree. ii) Even if re-
partitioning is allowed in PASS, the algorithms we proposed in
(liang2021combining, ) do not run efficiently in the dynamic setting. Here we
propose dynamic indexes and algorithms with theoretical guarantees that
perform much faster than the algorithms in PASS. In Section 6 we show that our
new dynamic indexes can construct a new partitioning extremely faster than the
partitioning algorithm of PASS. iii) Even if we use our new dynamic algorithms
in PASS, there is no mechanism to compute the exact statistics of the nodes
after a re-partitioning happening and there is no mechanism handling the
updates as the re-partitioning is executed. In this paper we introduce a novel
multi-thread approach called the _catch-up_ phase. JanusAQP can improve the
estimators in the nodes of DPT after a re-partitioning, while handling new
dynamic updates and new queries. iv) Last but not least, we implement JanusAQP
on Apache Kafka, so it can be used by real database systems. On the other
hand, there is no system implementation of PASS.

##  3\. System Architecture

In this section, we describe the JanusAQP architecture.

###  3.1. Construction and Optimization API

First, we overview how users construct synopsis data structures in JanusAQP.
Unlike systems like BlinkDB (agarwal2013blinkdb, ), JanusAQP does not use a
single synopsis to answer all queries. Much like index construction in a
database, users choose which attributes to include in the synopsis structure.
Each synopsis can answer query templates of the following form:

⬇

SELECT SUM/COUNT/AVG/MIN/MAX(A) FROM D

WHERE Rectangle(D.c1,...,D.cd)

where $A$ is an aggregation attribute and $c_{1},...,c_{d}$ are predicate
attributes used in some rectangular predicate region (a conjunction of $>,<,=$
clauses). The dimensionality of a synopsis is the number of predicate
attributes $d$. To construct a synopsis, the user must define the following
basic inputs:

  * •

Aggregation Attribute.  An attribute $A$ that is the primary metric for
aggregation.

  * •

Predicate Attributes.  A collection of $d$ columns $c_{1},...,c_{d}$ that are
used to filter the data prior to aggregation.

  * •

Memory Constraint.  The maximum amount of space that the synopsis can take.

  * •

Query Processing Constraint.  The maximum bytes of data that the system should
process in answering a query.

  * •

Historical Data Limit.  How much historical data to include in the synopsis,
i.e., the earliest time-step of data included in the system.

JanusAQP contains an optimizer that integrates these constraints into a solver
that produces an optimized synopsis (one with low error). Beyond these basic
knobs that are relevant to most AQP systems, there are two other
considerations discussed in this paper: Catch-Up Processing.  Constructing a
synopsis will require some amount of computational time. While incremental
maintenance might be efficient, constructing the initial synopsis $S^{(0)}$
from the initial database state $\mathcal{D}^{(0)}$ might be very expensive if
there is a significant amount of initial data. However, as the initial
$S^{(0)}$ is being constructed new data will arrive, and the system will
require additional processing to catch up. JanusAQP optimizes the catch-up
process using a multi-threaded system and approximate internal statistics for
the partition tree. This process minimizes the amount of time where the system
is unable to process new data or queries. The user decides how much processing
to expend during catch up, the quicker the system is ready, the higher the
error will be.

Throughput.  The maximum data throughput is the maximum rate of insertions and
deletions that the system can support. Throughput depends on the complexity of
the synopsis used.

###  3.2. Data and Query API

For processing queries and data, we adopt the PSoup architecture where both
queries and data are streams (chandrasekaran2003psoup, ). JanusAQP supports
three types of requests: insertion of a new tuple, deletion of an existing
tuple and querying of the existing tuples. Thus, there are three Kafka topics
insert(tuple), delete(tuple), and execute(query).

The use of Kafka, with its timing and delivery guarantees, simplifies the
query processing semantics. The system will process the incoming stream of
queries in order. Each query will have an arrival time $i$, which is the
current database state at the time at which the query is issued. Therefore, we
define $q^{(i)}_{j}$ as the $j$th query in the sequence that arrives at
database state $i$. Query results should reflect all of the data that has
arrived until the time point $i$.

###  3.3. Summary and Algorithmic Contributions

To summarize, the usage of JanusAQP can be thought of as a life cycle. (1.
Initialization) The user triggers synopsis construction through an API call.
(2. Catch-Up) The system will online construct the synopsis while managing new
data arriving into the system. (3. Query/Data Processing) Then, JanusAQP is
ready to process requests of insertions, deletions, and queries. (4. Re-
Initialization) As JanusAQP processes more updates, the data or query workload
could drift requiring re-partitioning. This procedure re-enters the
initialization phase.

Throughout the rest of the paper, we present technical contributions
throughout this synopsis life cycle:

  * •

Dynamic Partition Trees (Section 4) The core data structure in JanusAQP is
called a dynamic partition tree (DPT). This data structure aggregates data at
multiple levels of resolutions and associates some of the aggregates with
stratified samples. The size of the tree (i.e., depth and width) can be
changed to tradeoff accuracy at the cost of increased storage and throughput.
The sampling rate at the leaf nodes can be adjusted to tradeoff query latency
with accuracy.

  * •

Warm-Start Deployment (Section 4.3) Next, we present a multi-threaded
technique that allows DPT synopses to be deployed in dynamic environments
accounting for new data that might arrive during their construction. This
component crucially allows for either partial or full online re-partitioning
and re-balancing of the data structure.

  * •

Minimum Variance Partitioning (Section 5) We describe a new, efficient
algorithm for selecting the DPT tree structure that minimizes the worst-case
estimation error.

##  4\. Dynamic Partition Trees

We discuss how Dynamic Partition Trees (DPT) are constructed, how they answer
queries, and how they are maintained under updates. Structurally, a DPT is
essentially the same data structure as an SPT; however, the way that the
partition statistics and samples are represented differ to allow for
incremental maintenance. Figure 3 summarizes the basic update process.

Table 1. Table of basic notation $\mathcal{D}$ | Full database | $H_{i}$ | $H\cap R_{i}$
---|---|---|---
$N$ | $|\mathcal{D}|$ | $m_{i}$ | $|S_{i}|$
$S$ | Set of reservoir samples | $h_{i}$ | $|H_{i}|$
$H$ | Set of catch-up samples | $m$ | $|S|$
$R_{i}$ | Partition/bucket/rectangle | $t$ |  Tuple in $\mathcal{D}$
$|R_{i}|$ | $|R_{i}\cap S|$ | $t.a$ |  Aggregation value of tuple $t$
$N_{i}$ | $\mathcal{D}\cap R_{i}$ | $\mathcal{T}$ |  Partition tree in DPT
$S_{i}$ | $S\cap R_{i}$ |  |
Figure 3. The DPT update process for an insertion or deletion. (1) A set of
samples is maintained using a reservoir sampling algorithm. (2) The leaf node
statistics are incrementally updated. (3) The updated statistics from the leaf
node propagate to the parents. (4) Updated statistics from the parents
propagate all the way to the root.

###  4.1. Incrementally Maintaining Nodes

Each node defines a partition and contains statistics (the SUM, COUNT, MIN,
and MAX aggregates) of the data contained in that partition. The key challenge
is to keep these statistics up-to-date in the presence of insertions and
deletions. When an insertion or deletion arrives, an entire path of nodes from
the leaf to the root will have to be updated.

DPT Nodes:  First, we discuss how we represent the statistics in a DPT node.
Since the SUM and COUNT are easy to incrementally maintain under both
insertions and deletions, we simply store a single SUM and COUNT value for
each aggregation attribute. The MIN and MAX values are harder to incrementally
maintain. To store the MIN and MAX values, we store the top-k and the bottom-k
values in a MIN/MAX heap respectively. The top value of these heaps is equal
to the MIN and MAX of all the data in the node.

Insert New Record:  When a new record is inserted, we first test each leaf
node to see if the record is contained in the node. Once find the appropriate
leaf node, we then increment the SUM and COUNT statistics accordingly.
Finally, we push the new aggregation values onto the heap. If the heap exceeds
the size limit $k$, then the bottom value on the heap is removed.

Delete Existing Record:  When an existing record is deleted, we first test
each leaf node to see if the record is contained in the node. Once find the
appropriate leaf node, we then decrement the SUM and COUNT statistics
accordingly. Finally, if that aggregation value is contained in the heap it is
removed from the heap.

###  4.2. Maintaining Stratified Samples

Next, we describe how to maintain the samples associated with leaf nodes. We
use a modified version of the well-known technique of _reservoir-sampling_
(vitter1985random, ) under updates (gibbons2002fast, ). The details of how we
implement this are interesting. Conceptually, each leaf node is associated
with a physically disjoint sample of just that partition, i.e., a stratified
sample. Instead of physical strata, we implement virtual partitions of a
single global sample. This global sample can be maintained using a reservoir
sampling algorithm and makes it easier to control the overall size of the
synopsis under insertions/deletions as well as simplifies concurrency control.

Sample Representation:  The DPT maintains a “pooled” sample (all the relevant
samples in a single data structure). This set of samples has a target size of
$2m$ tuples. At the construction time, we choose a set $S$ of $2m$ uniform
random samples from $\mathcal{D}$. The update procedure ensures that there are
always between $m\leq|S|\leq 2m$ samples. The leaf nodes index into this
“pooled” sample selecting only the relevant data to their corresponding
partitions.

Insert New Record:  Suppose we insert a new tuple $t$. If $|S|<2m$ we add $t$
in $S$. If $|S|=2m$, we choose $t$ with probability
$\frac{|S|}{|\mathcal{D}|}$. If it is selected then we replace $t$ with a
point from $S$ uniformly at random.

Delete Existing Record:  Next, suppose that we delete a tuple $t$ from
$\mathcal{D}$. If $t\notin S$ we do not do anything. If $t\in S$ then we check
the cardinality of $S$. If $|S|>m$ then we only remove $t$ from $S$. If
$|S|=m$ then we skip the set $S$ and we re-sample $2m$ items from
$\mathcal{D}$. As shown in (gibbons2002fast, ) this procedure always maintain
a set of uniform random samples. Using a simple dynamic search binary tree of
space $O(m)$ we can update the samples $S$ stored in $\mathcal{T}$ in
$O(\text{height}(\mathcal{T}))$ time.

Figure 4. JanusAQP synopses can be re-initialized online using a multi-
threaded implementation to minimize unavailability

###  4.3. Re-initialization and Catch-Up

As we noted before, repeated deletes on the same leaf partition can degrade
the accuracy of the synopsis. As we will see in the next section, it is also
possible for repeated insertions to degrade the accuracy as well. In such
cases, re-initialization of the DPT may be needed where the data structure is
re-built and re-optimized over existing data.

Enabling periodic re-initialization is crucial for reliable long-term
deployment but is challenging because new data will not simply stop arriving
during the re-initialization period. As the dataset size grows, the amount of
time needed for re-initialization will grow as well. We employ a multi-
threaded approach to minimize any period unavailability for processing new
data arrival as well as new queries (Figure 4). When re-initialization is
triggered, the main processing thread initiates the construction of a new DPT
synopsis and the following steps are performed:

  1. (1)

Optimization Phase (In Parallel)

     * •

The partition optimization algorithm analyzes the data in the pooled reservoir
sample to determine the optimal new partitioning criteria. It returns a new
empty DPT with no node statistics.

     * •

In parallel with (Step 1), the old synopsis is maintained under all insertions
and deletions that happen during the optimization algorithm. Queries can still
be answered with the old synopsis.

  2. (2)

(Blocking) Approximate node statistics are populated into the new synopsis
using the pooled reservoir sample $S$ (note, that this will reflect any data
that arrived during the optimization phase). This is the only blocking step in
the re-initialization routine and new data and queries will have to wait until
completion.

  3. (3)

The old synopsis is discarded.

  4. (4)

The system resamples a uniform sample of data from archival storage to be the
new pooled reservoir sample. Queries and results can still be processed on the
new synopsis even without a sample.

  5. (5)

Random samples of historical data are used to improve the node statistics in
the background until a user-specified “catch-up” time.

This process is the key difference between an SPT and an DPT, where after
catch-up the node statistics may be inexact. However, this old data is
propagated in a random order, which means that the SUM,COUNT,AVG values in
each node will be unbiased estimates of their full data statistics. The
duration of the catch-up phase can be chosen by the user. For example, in our
experiments, the catch-up phase does not stop until we get
$0.1\cdot|\mathcal{D}|$ samples. It is worth noting that queries close to the
beginning of the catch-up phase will have a higher error, however queries
towards the middle or the end of the catch-up phase will have a smaller error.
In Section 5.4, we describe how to trigger re-initialization. Furthermore,
there is only one step (2) where the synopsis is unavailable to process
queries and data (has the duration of 100s of milliseconds in our
experiments).

###  4.4. Answering Queries With a DPT

The basic structure of the result estimator is the same as before, especially
for the $R_{partial}$ partitions. However, there are a few key changes due to
the nature of the catch-up phase. In SPT, for each partition in $R_{cover}$,
we can compute an exact “partial aggregate” for the tuples in those partitions
and combine the partial aggregates. In a DPT, this process changes considering
the estimations we get from the catch-up samples. Overall, the estimation of a
partition $R_{i}\in R_{cover}$ consists of i) estimation using the catch-up
samples $H$ and the formulas of Section 2.3, ii) the exact statistics of the
new inserted tuples in $R_{i}$, and iii) the exact statistics of the deleted
tuples in $R_{i}$ (recall that the quantities in ii), iii) are stored and
maintained as described in the Incrementally Maintaining Statistics in Section
4.1). By taking the sum of i), ii) and subtracting iii) we get the unbiased
estimation in partition $R_{i}$.

Let $H$ be the set of catch-up samples and $H_{i}\subseteq H$ be the subset of
$H$ that lie in a partition $R_{i}$, and $h_{i}=|H_{i}|$. All basic notations
are defined in Table 1. The formulas for estimating COUNT and SUM queries in
both $R_{cover}$, $R_{partial}$ from Section 2.3 contain the factor
$\frac{N_{i}}{m_{i}}$ or $\frac{N_{i}}{h_{i}}$, while the formulas for
estimating the AVG contain the factor $\frac{N_{i}}{N_{q}}$. In DPT we do not
have the exact values for $N_{i}$. Instead, we use an estimate of the size of
the partition $R_{i}$ denoted by $\hat{N_{i}}$. In particular we use the
catch-up samples $H$ to estimate $\hat{N_{i}}=\frac{h_{i}}{h}N$.

####  4.4.1. Confidence Intervals

While the estimators do not significantly change from an SPT to a DPT, the
confidence intervals are calculated very differently. This is because there
are now two sources of errors: estimation errors due to the stratified samples
and estimation errors in the node statistics. Both these sources of errors
have to be integrated into a single measure of uncertainty. Assuming that all
partitions are large enough, the central limit theorem can be used to
asymptotically bound the estimation error for SUM/COUNT/AVG queries.
Informally, the central limit theorem states that this asymptomatic error is
proportional to the square-root of the ratio of estimate variance and the
amount of samples used $\propto\sqrt{\frac{var(est_{i})}{m_{i}}}$. We simply
have to match terms to this formula for all sample estimates and all node
estimates because both are derived from samples.

Error in Node Estimates. First, let’s account for all the uncertainty due to
catch-up. Recall that $H$ is the set of catch-up samples we have considered so
far and $H_{i}\subseteq H$ is the samples in partition $R_{i}$ with
$h_{i}=|H_{i}|$. We note that we do not store the set $H$ or the subsets
$H_{i}$, instead we only use the new catch-up samples to continuously improve
the statistics we store in the nodes. Using the notation in the previous
section, we can calculate the catch-up variance $\nu_{c}$:

$\nu_{c}(q)=\sum_{R_{i}\in
R_{cover}}w_{i}^{2}\frac{var(\phi_{q}(H_{i}))}{h_{i}}$

where $w_{i}=\frac{\hat{N_{i}}}{\hat{N_{q}}}$ for AVG queries and $w_{i}=1$
for SUM/COUNT queries. Calculating $\phi_{q}(H_{i})$ is straight-forward. We
simply store additional information that allows us to efficiently calculate
the variance. For any node $i$ of $\mathcal{T}$ we store $h_{i}$, $\sum_{t\in
H_{i}}t.a^{2}$, $\sum_{t\in H_{i}}t.a$.

Error in Sample Estimates. For a partition $R_{i}\in R_{partial}$, let
$S_{i}\subseteq\mathcal{D}$ be the set of samples in $S$ that lie in partition
$R_{i}$ and let $m_{i}=|S_{i}|$. Like the catch-up variance, we can calculate
the sample estimate variance $\nu_{s}$:

$\nu_{s}(q)=\sum_{R_{i}\in
R_{partial}}w_{i}^{2}\frac{var(\phi_{q}(S_{i}))}{m_{i}}$

We can calculate an overall confidence interval as:

$\pm z\cdot\sqrt{\nu_{c}(q)+\nu_{s}(q)}$

where $z$ is a normal scaling factor corresponding to the desired confidence
level, e.g., $z=1.96$ for 95%. As before,
$w_{i}=\frac{\hat{N_{i}}}{\hat{N_{q}}}$ for AVG queries and $w_{i}=1$ for
SUM/COUNT queries. In Appendix B, we show analytically all formulas for
computing the variance under different types of queries.

Communication throughput and I/O operations. The query procedure of our system
only accesses the partition tree stored in a local machine. Hence the
communication throughput (or the I/O operations to access the full data) in
the query procedure is zero.

##  5\. Optimal DPT Partitioning

We next describe a new dynamic partitioning algorithm designed for the dynamic
setting.

###  5.1. Preliminaries and Problem Setup

The partitioning algorithm analyzes the pooled reservoir sample of data to
determine how best to partition the dataset. The goal of the partitioning
algorithm is to find a partitioning such that the subsequent queries issued to
the DPT have low-error. Surprisingly enough, the partitioning algorithm does
not need an exact query workload to perform this optimization. It simply needs
a focus aggregation function (e.g., SUM, COUNT, AVG) and finds a partitioning
that minimizes the worst-case query error for sufficiently large predicates.

Given a set of $O(m)$ samples $S$, the goal is to construct a data structure
that supports the following operations. (i) Insert or delete a sample from $S$
efficiently, and (ii) when a partitioning request comes, it creates a near-
optimum partition tree $\mathcal{T}$ in $o(m)$ time.

In order to find a near-optimum partition tree we define the following
optimization problem. Let $Q$ be a set of possible aggregate queries with a
predicate. And, let $\Theta$ be the set of all DPT synopses defined by
rectangular partitioning conditions over $k$ partitions. The main optimization
objective is to minimize the maximum error over the query workload:

(1) $\min_{\mathcal{T}\in\Theta}\max_{q\in Q}\textsf{Error}(q,\mathcal{T})$

The error is defined as the length of the confidence interval, as defined in
the previous section. Since the catch-up variance is usually extremely smaller
than the sample estimate variance, we focus on minimizing the maximum length
of the confidence interval with respect to the sample estimate variance
$\nu_{s}(\cdot)$. For simplicity, when we say variance we always mean the
sample estimate variance.

The above problem still seems challenging because queries can intersect
partitions and cut a tree in arbitrary ways, but our recent work shows an
important simplification (liang2021combining, ) (under mild technical
conditions about the size of partitions). Instead of looking over all possible
queries to minimize the maximum error, one only needs to focus on single
partitions to ensure they do not have “high-variance” sub-partitions. Indeed
by considering only these sub-partitions we can still get a
$\sqrt{k}$-approximation for COUNT and SUM queries over the optimum partition
considering all queries (for $k$ leaves). The approximation factor improves to
$\sqrt{2}$ for $d=1$. For AVG queries the error of the optimum partition of
this simplification is the same with the maximum error considering every
possible query.

The error of a query $q$ inside a leaf node (partition) $R_{i}$ is defined
(expanding the equations from the definition of $\nu_{s}(\cdot)$) as

$\frac{N_{i}^{2}}{m_{i}^{3}}\\!\left[m_{i}\\!\sum_{t\in
q}\\!t.a^{2}\\!-\\!\left(\\!\sum_{t\in
q}\\!t.a\\!\right)^{2}\right],\quad\frac{1}{m_{i}|q\cap
S|^{2}}\\!\left[m_{i}\\!\sum_{t\in q}\\!t.a^{2}\\!-\\!\left(\\!\sum_{t\in
q}\\!t.a\\!\right)^{2}\right].$

for SUM/COUNT and AVG queries, respectively.

Thus, the optimization problem reduces to finding partitions that do not
contain a high-variance “rectangle” of data. Unfortunately, all algorithms in
(liang2021combining, ) are designed only for the static case and their running
is always super-linear $\Omega(m)$ with respect to the samples. Hence, their
algorithms cannot be used to satisfy the requirements in the dynamic setting.

One of the core subroutine of all our new partitioning algorithms is the
following. Given a rectangle $R$, the goal is to find a rectangular query
within $R$ with maximum variance among all possible queries in $R\cap S$. For
now, we assume that we have a dynamic index $\mathcal{M}$ with near-linear
space such that given a query rectangle $R$, it returns a query $q$ within $R$
with $\nu_{s}(q)\geq\frac{1}{\gamma}\mathcal{V}(R)$, for a parameter
$\gamma>1$, in $O(M)$ time, where $\mathcal{V}(R)$ is the variance of the
maximum variance rectangular query in $R$. Let $\mathcal{M}(R)$ be the
variance of the query returned by the index $\mathcal{M}$. We describe this
index with more details in Subsection 5.3.

###  5.2. Partitioning for $d=1$

Now, we discuss how to solve the partitioning optimization problem in one
dimension. We present results for SUM and AVG queries. COUNT can be thought of
as a special case of SUM with binary data. The basic trick is to search over a
discretized set of possible variance values. For each value $e$, we try to
construct a partitioning of $k$ partitions such that in each bucket the length
of the longest confidence interval of a query is at most $e$. By
systematically reducing $e$ in each iteration, we control for the worst-case
error.

Bounding the Error. The first step is to calculate the bounds for the maximum
length of the largest possible confidence interval among queries that
intersect one partition. We assume that the aggregation value of any item in
$\mathcal{D}$ is bounded by a maximum value $\mathcal{U}$ and a minimum non-
zero value $\mathcal{L}$. We allow items to take zero values since this is
often the case in real datasets but no item with positive value less than
$\mathcal{L}$ or larger than $\mathcal{U}$ exists. We assume that
$\mathcal{U}=O(\textrm{poly}(N))$ and
$\mathcal{L}=\Omega(1/\textrm{poly}(N))$. In Appendix C.2 we show that the
length of the longest confidence interval is also bounded by
$O(\textrm{poly}(N))$ and $\Omega(1/\textrm{poly}(N))$.

Description of Algorithm. We describe the partitioning algorithm for SUM
queries. The procedure is identical for AVG queries. For a parameter
$\rho\in\mathbb{R}$ with $\rho>1$, let $E=\\{\rho^{t}\mid
t\in\mathbb{Z},\frac{\mathcal{L}}{\sqrt{2}}\leq\rho^{t}\leq
N\mathcal{U}\\}\cup\\{0\\}$, be the discretization of the range defined by the
lower and upper bound of the longest confidence interval (as defined in the
previous paragraph). We run a binary search on the values of $E$. For each
value $e\in E$ we consider, we try to construct a partitioning of $k$
partitions such that in each partition the length of the longest confidence
interval of a query is at most $e$. If there exists such a partitioning we
continue the binary search with values $e^{\prime}<e$. If there is no such a
partitioning we continue the binary search with values $e^{\prime}>e$. In the
end of the binary search we return the last partitioning that we were able to
compute.

It remains to describe how to check if a partitioning with $k$ buckets
(intervals) with maximum length confidence interval at most $e$ exists. A high
level description of the algorithm is the following.

  1. (1)

For $i=1$ to $k$

    1. (a)

Let $b_{i}$ be the $i$-th bucket with left endpoint $t_{a}$

    2. (b)

Binary search on samples $t_{j}$ to find the maximum bucket $b_{i}$ with error
at most $e$

    3. (c)

If $\sqrt{\mathcal{M}([t_{a},t_{j}])}\leq e$

      1. (i)

Continue search for values $>j$

      2. (ii)

Else Continue search for values $<j$

  2. (2)

If the partitioning contains all samples construct $\mathcal{T}$ using $b_{i}$
as its leaf nodes. Otherwise $\mathcal{T}=\emptyset$.

We start with the leftmost sample, say $t_{1}$, which is the left boundary of
the first bucket. In order to find its right boundary we run a binary search
on the samples $S$. Let $t_{j}$ be one of the right boundaries we check in the
binary search, and let $b_{1}=[t_{1},t_{j}]$. If
$\sqrt{\mathcal{M}(b_{1})}\leq e$ then we continue the binary search with a
sample at the right side of $t_{j}$ (larger bucket). Otherwise, we continue
the binary search with a sample at the left side of $t_{j}$ (smaller bucket).
When we find the maximal bucket with longest confidence interval at most $e$
we continue with the second bucket repeating the same process for at most $k$
buckets. In the end, if all samples in $S$ are contained in $k$ buckets then
we return that there exists a partitioning (with $k$ buckets) with maximum
variance at most $e$. If we cannot cover all samples in $k$ buckets then we
return that there is no partitioning with $k$ buckets and maximum variance at
most $e$.

Correctness.  In Appendix C.2 we use the monotonic property of the longest
confidence interval (the bigger the bucket the larger the error) and we show
$\sqrt{\mathcal{V}(b^{\prime})}\leq\sqrt{\gamma\mathcal{M}(b^{\prime})}\leq\sqrt{\gamma}e^{\prime}\leq\rho\sqrt{\gamma}\sqrt{\mathcal{V}(b^{*})}$,
where $b^{\prime}$ is the bucket with the longest confidence interval in the
returned partitioning, $e^{\prime}$ is the smallest value in $E$ such that
$\sqrt{\mathcal{V}(b^{*})}\leq e^{\prime}$, and $b^{*}$ is the bucket of
optimum partitioning with the largest confidence interval. For $d=1$ we have
that $\gamma=4$ for SUM and AVG queries queries, so we get a partitioning
where the maximum error is within $2\rho\sqrt{2}$ of the optimum error for SUM
queries and within $2\rho$ of the optimum error for AVG queries.

Running time.  Since, $\mathcal{L},\mathcal{U}$ are polynomially bounded on
$N$ we have that $|E|=O(\log_{\rho}N)$ and it can be constructed in
$O(\log_{\rho}N)$ time. The binary search over $E$ takes at most
$O(\log\log_{\rho}N)$ steps. We can decide if there exists a partitioning with
error $e$ in $O(kM\log m)$ time. Overall, the running time of our algorithm is
$O(kM\log m\log\log_{\rho}N)$. If $\rho$ is a constant, for example $\rho=2$,
then the running time is $O(kM\log m\log\log N)$. In Appendix C.2 we have that
in $1$-dimension $M=O(\log m)$ for SUM and AVG queries. Notice that if we skip
the $\log$ factors the running time depends only linearly on the number of
buckets $k$ and the approximation factor is constant.

###  5.3. Partitioning in Higher Dimensions

####  5.3.1. Indexing To Find Maximum Variance

Now, we describe the core index $\mathcal{M}$ that we use in all our
partitioning algorithms for any dimension $d\geq 1$. The exact description of
the index depends on the type of aggregation queries we focus on.

For SUM and COUNT queries, we propose a simple index to find the query with
the largest variance in a query rectangle. In particular, we build a dynamic
range tree on $S$. Given a query rectangle $R$, we split it into two smaller
rectangles $R_{1},R_{2}$ such that $|R_{1}\cap S|=|R_{2}\cap S|=|R\cap S|/2$.
Using a dynamic range tree (de1997computational, ) we return the rectangle
$R_{i}$ (either $R_{1}$ or $R_{2}$) with the largest variance. We can show
that $\nu_{s}(R_{i})\geq\frac{1}{4}\mathcal{V}(R)$. The running time and the
update time is $O(\log^{d}m)$.

For AVG queries, the algorithm proposed in (liang2021combining, ) cannot be
extended to the dynamic case. Hence we propose a new dynamic index with a
better approximation factor. Similarly to (liang2021combining, ), we assume
that every valid query that is contained in a bucket of the partitioning must
contain at least $2\delta m$ samples (for a small parameter $\delta<1$),
otherwise the estimation is not accurate. For simplicity, we use the notation
$\tilde{O}(\cdot)$ to hide $\log(m)$ factors. In Appendix C.1 we show the
following crucial observation: for any rectangle $q$ inside a query rectangle
$R$ with $|q\cap S|=\delta m$ that maximizes $\sum_{t\in q\cap S}t.a^{2}$, it
holds that $\nu_{s}(q)\geq\frac{1}{4}\mathcal{V}(R)$. Hence, we build a
dynamic index so that given a query rectangle $R$ it returns a rectangle that
contains $\delta m$ samples and the sum of squares of their aggregate values
is close to the maximum sum.

We build a dynamic range tree $T^{\prime}$ over the samples $S$, storing the
number of samples in each node of the tree. Furthermore, we build another
empty dynamic range tree $T$. We will use $T$ to store weighted rectangles (as
points in $2d$) that contain at most $\delta m$ samples. More specifically, we
store in $T$ the _canonical_ rectangles of $T^{\prime}$ that contain at most
$\delta m$ samples. Notice that there are $\tilde{O}(m)$ nodes in $T^{\prime}$
hence $T$ uses $\tilde{O}(m)$ space. When we have an insertion or deletion in
$T^{\prime}$ there are only $\tilde{O}(1)$ nodes/rectangles that are updated,
hence we can update both $T^{\prime}$ and $T$ in $\tilde{O}(1)$ time. Given a
query rectangle $R$ we use $T$ to find a rectangular query $q^{*}$ with the
largest sum inside $R$ in $\tilde{O}(1)$ time. From the definition of a range
tree, for any rectangle there is a partitioning of $\log^{d+1}m$ canonical
rectangles from $T^{\prime}$. Hence we can show that
$\nu_{s}(q^{*})\geq\frac{1}{4\log^{d+1}m}\mathcal{V}(R)$. The exact
complexities depend on the dynamic range tree structure we use; our data
structure has roughly $O(m\log^{3d}m)$ space, $O(\log^{3d}m)$ update time, and
$O(\log^{2d}m)$ query time.

####  5.3.2. Partitioning

We construct a partitioning by building a k-d tree using the dynamic procedure
$\mathcal{M}$ as we described above. The construction is similar to the
construction in (liang2021combining, ). However, they construct a near optimum
k-d tree in time $O(km)$ skipping the $\log$ factors. Here, we use our
improved index $\mathcal{M}$ to construct a k-d tree faster (in roughly $O(k)$
time) with better approximation guarantees.

The high level description of the algorithm is the following.

  1. (1)

Max Heap $C$ containing partition $R_{1}$ covering all items in $\mathcal{D}$

  2. (2)

For $j=2$ to $k$

    1. (a)

Extract the partition $R_{i}$ with maximum $\mathcal{M}(R_{i})$ from $C$

    2. (b)

Create a partitioning of $R_{i}$ of two partitions $R_{i_{1}}$, $R_{i_{2}}$ by
splitting on the median of $R_{i}$

    3. (c)

Insert $\mathcal{M}(R_{i_{1}})$, $\mathcal{M}(R_{i_{2}})$ in $C$

    4. (d)

Set $R_{i_{1}}$, $R_{i_{2}}$ as children of $R_{i}$ in $\mathcal{T}$

We can show that such a tree construction returns a partitioning which is near
optimal with respect to the optimum partition tree construction following the
same splitting criterion: split on the median of the leaf node with the
largest maximum variance query. Overall we construct a data structure that can
be updated in $O(\textrm{polylog}m)$ time. For a (re-)partition activation
over a set $S$ of $m$ samples, we can construct a new $\mathcal{T}$ with the
following guarantees: For COUNT/SUM queries, $\mathcal{T}$ can be constructed
in $O(k\log^{d}m)$ time with approximation factor $2\sqrt{k}$. For AVG
queries, $\mathcal{T}$ can be constructed in $O(k\log^{2d}m)$ time with
approximation factor $2\log^{(d+1)/2}m$. In all cases we can construct near-
optimum partitions in $\tilde{O}(k)$ time.

###  5.4. Re-Partitioning Triggers

Assume that the current partitioning is $\mathcal{R}$ and let
$\mathcal{M}(\mathcal{R})$ be the (approximate) maximum variance query with
respect to the current set of samples $S$. JanusAQP first checks the number of
samples in each bucket (leaf node) of the current $\mathcal{T}$. If there is a
leaf node $i$ associated with partition $R_{i}$ such that
$|S_{i}|<<\frac{1}{\alpha}\log m$ (where $\alpha$ is the sampling rate) then
there are not enough samples in $u$ to make robust estimators. Hence, we need
to find a new re-partitioning. Even if the number of samples in each bucket is
large our system might enable a re-partitioning: For a partition $R_{i}$ in
the leaf node layer of $\mathcal{T}$ let $\mathcal{M}_{i}=\mathcal{M}(R_{i})$
be the (approximate) maximum variance at the moment we constructed
$\mathcal{T}$. Let $\beta>1$ be a parameter that controls the maximum
allowable change on the variance. It can either be decided by the user or we
can set it to $\beta=10$. Assume that an update occurred in the leaf node
associated with the partition $R_{i}$. After the update we get
$\mathcal{M}_{i}^{\prime}=\mathcal{M}(R_{i})$. If
$\frac{1}{\beta}\mathcal{M}_{i}\leq\mathcal{M}_{i}^{\prime}\leq\beta\mathcal{M}_{i}$
then the new maximum variance in partition $b_{i}$ is not very different than
before so we do not trigger a re-partition. Otherwise, the maximum variance in
bucket $b_{i}$ changed by a factor larger than $\beta$ from the initial
variance $\mathcal{M}_{i}$. In this case a re-partitioning might find a new
tree with smaller maximum error. We compute a new partitioning
$\mathcal{R}^{\prime}$ and hence a new tree $\mathcal{T}$. If
$\mathcal{M}(\mathcal{R}^{\prime})<\frac{1}{\beta}\mathcal{M}(\mathcal{R})$
then we activate a re-partition restarting the catch-up phase over the new
tree $\mathcal{T}$. On the other hand, if
$\mathcal{M}(\mathcal{R}^{\prime})\geq\frac{1}{\beta}\mathcal{M}(\mathcal{R})$
then our current partitioning $\mathcal{R}$ is good enough so we can still use
it. Of course, the user can also manually trigger re-partitioning. For
example, the user can choose to re-partition once every hour, day, or after
$\tau$ insertions and deletions have occurred. In Appendix C.4, we also
describe how JanusAQP can execute either partial or full re-partitioning.

##  6\. Experiments

We run our experiments on a Linux machine with an Intel Core i7-8700 3.2GHz
CPU and 16GB RAM.

###  6.1. Setup

To set up each experiment, we select a single aggregate attribute and one or
more predicate attributes. We generate query workloads of 2000 queries by
uniformly sampling from rectangular range queries over the predicates. We then
initialize a JanusAQP instance with a user-specified sample rate, a catch-up
ratio and a number of leaf nodes of the partition tree to compare with other
baselines (these parameters directly control the Throughput, Query Latency,
and Storage Size).

####  6.1.1. Datasets

Intel Wireless dataset. The Intel Wireless dataset (intelwireless_, ) contains
3 million rows of sensor data collected in the Berkeley Research lab in 2004.
Each row contains measurements like humidity, temperature, light, voltage as
well as the date and time each record was collected.

New York Taxi Records dataset. The New York City Taxi Trip Records dataset
(nyctaxi, ) contains 7.7 million rows of yellow and green taxi trip records
collected in January 2019. Each record contains information about the trip
including pickUpDateTime, dropOffDateTime, tripDistance, dropOffLocation,
passengerCount, etc.

NASDAQ ETF Prices dataset. The NASDAQ Exchange Traded Fund (ETF) Prices
dataset(nasdaq, ) contains 2166 ETFs traded in the NASDAQ exchange from April
1986 to April 2020. There are 4 million entries in the dataset and each entry
contains the date, the volume of transactions of an ETF on the date, and 4
prices: the price of an ETF when the market opens and closes; the highest and
the lowest of its daily price range.

####  6.1.2. Metrics and ground truth

In terms of performance, we report the wall-clock latency and the throughput,
i.e. number of requests (query/data) processed per second. To measure the
accuracy of the system, unless otherwise specified, we report the 95
percentile of the relative error which is the difference between ground truth
and estimated query result divided by the ground truth. We define the ground
truth to be w.r.t all the tuples available when the query arrives, i.e. the
true results reflect all insertions and deletions up to its arrival point.
With this setup, the results indeed depend on the sequence of requests that
are processed by the system. To make sure our experiments are deterministic,
we fix this sequence up-front and ensure they are the same for each baseline.

####  6.1.3. Baselines

We evaluate the following baselines. All of these baselines are tuned to
roughly control for query latency.

Reservoir Sampling (RS) and Stratified Reservoir Sampling (SRS). We construct
a uniform sample of the entire data set which is maintained using the
reservoir sampling algorithm (vitter1985random, ). We use a variant of RS
first designed for the AQUA system that handles both insertions and deletions
(gibbons2002fast, )222Due to its age, a direct comparison with AQUA was not
feasible. Unless otherwise noted, we use a 1% sample of data. For stratified
seservoir sampling, the strata is constructed using a equal-depth partitioning
algorithm.

DeepDB.  We also compare with a machine learning-based baseline called
DeepDB(hilprecht2019deepdb, ). DeepDB achieves state-of-the-art AQP results in
the static setting, and we chose it as a baseline since it has limited support
for dynamic data. In our baseline, DeepDB trains on 10% of the data. We set
this to be equivalent to the “catch-up” sampling in DPT.

Dynamic Partition Tree-Only (DPT).  We compare with a baseline of only using a
single DPT synopsis without online optimization. This synopsis is constructed
once and then used for the duration of the experiment. Unless otherwise noted
there are 128 leaf nodes in a balanced binary tree, the leaf nodes are
associated with 1% samples of their respective strata, and the catch-up
sampling rate 10% of the data.

JanusAQP.  Finally, we evaluate the full-featured JanusAQP system. This
includes a DPT and also performs re-partitioning if needed. Unless otherwise
noted there are 128 leaf nodes in a balanced binary tree, the leaf nodes are
associated with 1% samples of their respective strata, and the catch-up
sampling rate 10% of the data.

The storage costs of the baselines on the NYC Taxi dataset given the typical
setting (128 leaf-nodes, 10% catch-up rate, and 1% sample rate) are the
following, reservoir sampling baseline takes about 5MB, JanusAQP and DPT takes
about 6MB, a DeepDB baseline trained with 10% of the data is about 60MB.

###  6.2. Accuracy

—
---
Approach
JanusAQP
DeepDB
RS
SRS

Intel (%)
---
0.2 | 0.5 | 0.9
0.67 | 0.62 | 0.33
1.5 | 1.7 | 0.8
2.1 | 1.6 | 1.3
1.3 | 1.3 | 1.2

NYC (%)
---
0.2 | 0.5 | 0.9
0.48 | 0.22 | 0.2
4.7 | 4.7 | 4.7
3.4 | 2.1 | 0.94
2.4 | 1.2 | 0.95

ETF (%)
---
0.2 | 0.5 | 0.9
5 | 4.3 | 2.3
- | - | -
16 | 9.8 | 8.6
10 | 8.2 | 8

Intel (ms/query)
---
0.2 | 0.5 | 0.9
0.19 | 0.31 | 0.63
0.6 | 0.6 | 0.6
2.5 | 6.3 | 13.2
3.1 | 6 | 10.7

NYC (ms/query)
---
0.2 | 0.5 | 0.9
0.27 | 0.57 | 0.97
0.6 | 0.6 | 0.6
4.7 | 14.2 | 30.6
4.6 | 14.7 | 25.3

ETF (ms/query)
---
0.2 | 0.5 | 0.9
0.14 | 0.28 | 0.46
0.6 | 0.6 | 0.6
2.58 | 6.8 | 13
2.66 | 5.2 | 12.7

Table 2. Median relative error (%) of 2000 SUM random queries and average
query latency (ms/query) over three datasets.

We first evaluate the end-to-end performance of JanusAQP and the baselines on
a 1d problem (1 predicate attribute). For the NYC Taxi dataset, we use the
pickUpTime attribute as the predicate attribute and the tripDistance attribute
as the aggregate attribute; for the ETF dataset, we use the volume attribute
as the predicate attribute and the close attribute as the aggregate attribute;
for the Intel Wireless dataset, we use the time and light attributes as
predicate and aggregate attribute respectively.

We start with 10% of the data in Kafka which is used by the baselines for
initialization (simulating historical data). We incrementally add 10% more
data in increments (simulating new data arrival). After every 10% increment,
we re-train the model for DeepDB and re-initialize the DPT used by JanusAQP.
Due to space limitation, we report results when 20%, 50%, and 90% of the rows
from each dataset are inserted into the system. The median relative error and
the corresponding average query latency can be found in Table 2.

We can see that JanusAQP has the overall best accuracy while controlling for
query latency. We note that the accuracy of DeepDB is stable as a function of
progress. This is because as a learned model DeepDB has a roughly fixed
resolution of the data (it does not increase the number of parameters as more
data is inserted).333We omit the results of DeepDB on the ETF dataset in Table
2 due to very large error ($>$ 1000%) for SUM queries while the error of COUNT
queries is reasonable. These findings are consistent with results from
(liang2021combining, ). The accuracy of RS and SRS improves at a cost of a
higher query latency.

###  6.3. Performance

Next, we evaluate the throughput and re-optimization cost of JanusAQP. We
populate Kafka with the first $p$ percent of the NYC Taxi dataset ($p$ varies
from 10 to 90). Like before, we initialize JanusAQP on the first 10% of data
and then incrementally add increments of 10% more. In this experiment, we
construct a mixed update workload of both insertions and deletions.

On the left plot of Figure 5, we show the throughput of handling insertions
and deletions using a pool of 12 threads. We can see the performance of
JanusAQP is quite stable and does not change with the size of existing data or
the amount of data that have been processed. For each insertion and deletion,
we simply find the target node in $O(\log(k))$ and modify the summary. Even
though a larger reservoir size increases the overhead of manipulating the
samples for reservoir sampling, the increased overhead is unnoticeable. This
is because the stratum stored in each node is $\frac{1}{k}$ of the reservoir,
and each stratum is independent with others and race condition only happens if
two workers are working on the same node.

On the right plot of Figure 5, we show the re-optimization time cost in
seconds by JanusAQP and DeepDB. The cost to initialize JanusAQP increases with
the number of tuples stored in Kafka but it is still much cheaper than DeepDB.
It is worth noting that the re-optimization cost of DeepDB is the cost of re-
training instead of incremental training. This is mostly due to the constraint
of the API exposed by DeepDB, and we observe that re-train a model with 2$n$
samples is faster than train a model with $n$ samples then incrementally train
another $n$ samples. The results suggest that complex, learned synopses are
not ideal in the dynamic setting.

Figure 5. We evaluate the throughput of JanusAQP when handling insertions and
deletions in multi-threaded mode. We also compare the re-optimization cost
with DeepDB.

###  6.4. Handle Deletion

To demonstrate how JanusAQP handles deletions of data, we construct a JanusAQP
instance with the 50% percent of each dataset, then we delete the last $p$% of
data of the first 50% ($p$ varies from 1% to 9%). After JanusAQP process all
the deletions, a query workload of 2000 random queries is evaluated and we
record the median relative error of the 2000 queries. We use the data that
remains in the system to compute the ground truth, e.g., for $p=1\%$, the
ground truth is computed with the first 49% of each dataset.

Results can be found in Figure 6, we notice that the relative error is
relatively stable when we vary the deletion percentage. This is because the
tuples that are being deleted are uniformly distributed over the predicate
attributes of the query workloads, i.e. the deletion would occur in each leaf
node of the DPT with roughly the same probability, therefore, the DPT without
re-optimization works reasonably well. In another experiment we artificially
generate deletions that are skewed to demonstrate scenario where re-
optimization is needed, details can be found in Sec. 6.7.

Figure 6. Median relative error of JanusAQP varying the amount of deletions
from 1% to 9% over three datasets.

###  6.5. The Catch-up Phase

In this experiment, we want to understand how the catch-up phase can impact
the accuracy and performance of the entire system.

####  6.5.1. Accuracy

We use the entire Intel wireless dataset as the existing data. We compare a
set of JanusAQP (128, $c$, 1%) instances where the catch-up goal $c$ varies
from 1% to 10% with a step of 1%. When each JanusAQP instance reaches the
catch-up goal, we use it to evaluate the same set of 2000 random queries
generated using the light attribute as the aggregate attribute and the time
attribute as the predicate attribute.

The results can be found in the left plot of Figure 7. As a reference, we also
show the accuracy of an RS baseline with 1% sample rate. We notice that
JanusAQP (128,1%,1%) has no advantage against the RS baseline because neither
the samples nor the summaries built during catch-up could provide better
accuracy. As we increase the catch-up ratio, we can see an improvement in
accuracy because the quality of the summaries built by the catch-up phase
improved. Comparing with the expensive offline pre-processing used in
(liang2021combining, ), we believe the catch-up phase is a better alternative
that provides another knob to tune the tradeoff between accuracy and cost.

Figure 7. Varying the catch-up goal from 1% to 10% of the data, we evaluate
the accuracy of JanusAQP (left plot) and the time cost of the catch-up phase
(right plot).

####  6.5.2. Overhead

The overhead of the catch-up phase comes from two sources: the loading and
processing of the samples. We distinguish and measure the two types of
overhead in terms of their time cost. Data loading time measures the time
spent on calling the Kafka poll() API, transferring the data, and ETL
operations that are necessary to prepare the data for JanusAQP to process. It
is worth noting that the data loading cost is part of the essential cost that
occurs in all systems and is usually less relevant to the core design of the
system but more relevant to the design of interfaces. For example, with a
different interface, instead of dealing with the strings from Kafka that can
be expensive to parse, the system could use Protocol Buffers(protobuff, ) for
more efficient data exchanging or even offload some of the ETL duties to the
client-side as described in (ding2021ciao, ). On the other hand, the data
processing time stands for the time taken by JanusAQP to analyze the data then
accordingly modify internal data structures that will be used for query
processing.

Results can be found in the right plot of Figure 7. We can see that the data
processing with a single thread takes less than 1.5 seconds for a catch-up
ratio of 10%, which is equivalent to a throughput of processing 160,000 tuples
per second. Furthermore, the data loading cost is much higher than the data
processing cost and we believe the data loading cost can be further improved
by more engineering efforts and techniques such as client-assisted data
loading(ding2021ciao, ).

###  6.6. Multi-dimensional Query Templates

Next, we investigate the performance of JanusAQP with multi-dimensional
queries on the NASDAQ ETF Prices dataset. We randomly generate 2000 queries
from a 5-D query template that uses the volume attribute as the target
attribute, the date attribute and the 4 price attributes as predicate
attributes. We perform the same workflow as we did in Section 6.2. We first
compare the median relative error of JanusAQP (256,10%,1%) with DeepDB and the
results can be found in the left plot of Figure 8. We notice that the accuracy
of JanusAQP is better than DeepDB but the relative error increases for both.
This is because multi-dimensional queries are usually more selective. Also,
because the queries are generated using the entire dataset, we notice that
many of the ground truths generated using the first 20% of the data are 0s.
Therefore, in the experiment, we start with 30% of the data. On the right plot
of Figure 8, we can find the re-optimization cost of JanusAQP is lower than
DeepDB but is more expensive than in the 1D setting. While the increase of
dimensions can indeed make it more expensive to process the samples we fetched
during catch-up, we believe the re-optimization cost can be further improved
with more engineering efforts.

Figure 8. We compare the median relative error and the re-optimization cost of
JanusAQP with DeepDB on multi-dimensional queries.

###  6.7. Re-partitioning

Next, we consider microbenchmarks that evaluate the re-partitioning
optimizations in JanusAQP. In the first experiment, using the NYC Taxi
dataset, JanusAQP performs a periodic re-partitioning after every 10%
insertions. For comparison, the DPT baseline does not perform any re-
partitioning and we evaluate the accuracy. We deliberately skew the insertions
by sorting on pickUpDateTime so that new insertions would hit a small number
of partitions. The results are illustrated in Figure 9(left), we can see the
relative error of DPT increases drastically due to a partition tree that
becomes more and more imbalanced with new insertions. With periodic re-
partition, JanusAQP keeps the accuracy at a controlled level.

Figure 9. We compare the accuracy of JanusAQP and DPT in two scenarios that
cause imbalanced partition trees.

In the second experiment, we use the pickupTimeOfDay as the predicate
attribute. Because the dataset is randomly distributed over the
pickupTimeOfDay attribute, the insertions are not skewed as in the previous
setting. To demonstrate a situation where a re-partition is triggered by
deletions, we randomly choose 10% of the nodes and we randomly delete half of
the samples that belong to these nodes then we insert the next 10% data. After
the insertion, the re-partition will be triggered for JanusAQP. For
comparison, we use a DPT baseline that does not perform any re-partition. We
perform the same operations to the leaf nodes of the DPT baseline then we
evaluate the same set of queries. The results can be found in the right plot
of Figure 9, we can see the relative error of DPT increases due to the
imbalanced partition tree while the error of JanusAQP drops because of re-
partition.

###  6.8. A More Efficient Partitioning Algorithm

In Section 5, we propose a binary search-based (BS-based) partitioning
algorithm for 1 dimension that is much more efficient. In this experiment, we
compare the accuracy and time cost of the BS-based algorithm with the dynamic
programming-based partitioning algorithm used by PASS on the Intel Wireless
dataset. We implement the BS-based algorithm in Python in our code base of
PASS for a fair comparison. We measure the time cost in seconds of each
partitioning algorithm given different number of partitions, we also compare
the median relative error of the PASS variation over 2000 randomly generate
queries.

|  | 16 | 32 | 64 | 128
---|---|---|---|---|---
Partition Time (s) | DP | 16 | 22 | 382 | 6349
BS | 0.3 | 0.3 | 0.4 | 1.6
Median RE (CNT) | DP | 0.2% | 0.1% | 0.05% | 0.04%
BS | 0.6% | 0.4% | 0.1% | 0.1%
Median RE (SUM) | DP | 0.2% | 0.1% | 0.07% | 0.05%
BS | 1% | 0.9% | 0.2% | 0.2%
Median RE (AVG) | DP | 0.2% | 0.1% | 0.08% | 0.05%
BS | 1% | 0.7% | 0.2% | 0.15%
Table 3. We compare our new binary search-based (BS) partitioning algorithm
with the dynamic programming-based (DP) algorithm proposed in
(liang2021combining, ) on the Intel dataset.

The result can be found in Table 3, we vary the number of partitions from 16
to 128, as we increase the number of partitions, the sample size used by the
algorithms also increase. We notice that the time cost of the DP-based
algorithm increase drastically444Because we use a larger sample size than what
we used in (liang2021combining, ), the time cost of the DP algorithm increases
and the accuracy improves from what we reported in (liang2021combining, ).
with the number of partitions while the time cost of the BS-based algorithm
increase slightly. On the accuracy side, the DP-based algorithm does lead to a
lower error but the BS-based algorithm also introduce good accuracy. Overall,
we believe the BS-based algorithm is more scalable than the DP-based algorithm
and it provides favorable trade-off between cost and accuracy.

## References

  * [1] Nasdaq bookviewer, 2021.
  * [2] S. Acharya, P. B. Gibbons, and V. Poosala.  Aqua: A fast decision support system using approximate query answers.  In In Proc. of 25th Intl. Conf. on Very Large Data Bases. Citeseer, 1999.
  * [3] P. K. Agarwal, G. Cormode, Z. Huang, J. Phillips, Z. Wei, and K. Yi.  Mergeable summaries.  In Proceedings of the 31st ACM SIGMOD-SIGACT-SIGAI symposium on Principles of Database Systems, pages 23–34, 2012.
  * [4] S. Agarwal, B. Mozafari, A. Panda, H. Milner, S. Madden, and I. Stoica.  Blinkdb: queries with bounded errors and bounded response times on very large data.  In Proceedings of the 8th ACM European Conference on Computer Systems, pages 29–42, 2013.
  * [5] J. L. Bentley and J. B. Saxe.  Decomposable searching problems i. static-to-dynamic transformation.  Journal of Algorithms, 1(4):301–358, 1980.
  * [6] S. Chandrasekaran and M. J. Franklin.  Psoup: a system for streaming queries over streaming data.  The VLDB Journal, 12(2):140–156, 2003.
  * [7] S. Chaudhuri, B. Ding, and S. Kandula.  Approximate query processing: No silver bullet.  In Proceedings of the 2017 ACM International Conference on Management of Data, pages 511–519, 2017.
  * [8] G. Cormode.  Sketch techniques for approximate query processing.  Foundations and Trends in Databases. NOW publishers, 2011.
  * [9] G. Cormode, M. Garofalakis, P. J. Haas, C. Jermaine, et al.  Synopses for massive data: Samples, histograms, wavelets, sketches.  Foundations and Trends® in Databases, 4(1–3):1–294, 2011.
  * [10] M. De Berg, M. Van Kreveld, M. Overmars, and O. C. Schwarzkopf.  Computational Geometry: Algorithms and Applications.  Springer, 3rd edition, 2008.
  * [11] C. Ding, D. Tang, X. Liang, A. J. Elmore, and S. Krishnan.  Ciao: An optimization framework for client-assisted data loading.  In 2021 IEEE 37th International Conference on Data Engineering (ICDE), pages 1979–1984. IEEE, 2021.
  * [12] J. Erickson.  Static-to-dynamic transformations.  http://jeffe.cs.illinois.edu/teaching/datastructures/notes/01-statictodynamic.pdf.
  * [13] E. Gan, P. Bailis, and M. Charikar.  Coopstore: Optimizing precomputed summaries for aggregation.  Proceedings of the VLDB Endowment, 13(11).
  * [14] E. Gan, P. Bailis, and M. Charikar.  Coopstore: Optimizing precomputed summaries for aggregation.  Proceedings of the VLDB Endowment, 13(12):2174–2187, 2020.
  * [15] M. N. Garofalakis and P. B. Gibbons.  Approximate query processing: Taming the terabytes.  In VLDB, volume 10, pages 645927–672356, 2001.
  * [16] P. B. Gibbons, Y. Matias, and V. Poosala.  Fast incremental maintenance of approximate histograms.  ACM Transactions on Database Systems (TODS), 27(3):261–298, 2002\.
  * [17] A. C. Gilbert, S. Guha, P. Indyk, Y. Kotidis, S. Muthukrishnan, and M. J. Strauss.  Fast, small-space algorithms for approximate histogram maintenance.  In Proceedings of the thiry-fourth annual ACM symposium on Theory of computing, pages 389–398, 2002.
  * [18] Google.  Google protocol buffer, 2021.
  * [19] S. Guha, N. Koudas, and K. Shim.  Approximation and streaming algorithms for histogram construction problems.  Trans. on Datab. Syst., 31(1):396–438, 2006.
  * [20] B. Hilprecht, A. Schmidt, M. Kulessa, A. Molina, K. Kersting, and C. Binnig.  Deepdb: Learn from data, not from queries!  VLDB Endowment, 2019.
  * [21] H. Jagadish, H. Jin, B. C. Ooi, and K.-L. Tan.  Global optimization of histograms.  ACM SIGMOD Record, 30(2):223–234, 2001.
  * [22] H. V. Jagadish, N. Koudas, S. Muthukrishnan, V. Poosala, K. C. Sevcik, and T. Suel.  Optimal histograms with quality guarantees.  In VLDB, volume 98, pages 24–27, 1998.
  * [23] C. Jermaine.  Robust estimation with sampling and approximate pre-aggregation.  In Proceedings 2003 VLDB Conference, pages 886–897. Elsevier, 2003\.
  * [24] R. Jin, L. Glimcher, C. Jermaine, and G. Agrawal.  New sampling-based estimators for OLAP queries.  In 22nd International Conference on Data Engineering (ICDE’06), pages 18–18. IEEE, 2006.
  * [25] S. Joshi and C. Jermaine.  Materialized sample views for database approximation.  IEEE Transactions on Knowledge and Data Engineering, 20(3):337–351, 2008.
  * [26] M. Jurgens and H.-J. Lenz.  The r/sub a/*-tree: an improved r*-tree with materialized data for supporting range queries on olap-data.  In Proceedings Ninth International Workshop on Database and Expert Systems Applications (Cat. No. 98EX130), pages 186–191. IEEE, 1998.
  * [27] N. Koudas, S. Muthukrishnan, and D. Srivastava.  Optimal histograms for hierarchical range queries.  In Proceedings of the nineteenth ACM SIGMOD-SIGACT-SIGART symposium on Principles of database systems, pages 196–204, 2000.
  * [28] I. Lazaridis and S. Mehrotra.  Progressive approximate aggregate queries with a multi-resolution tree structure.  Acm sigmod record, 30(2):401–412, 2001.
  * [29] X. Liang, Z. Shang, S. Krishnan, A. J. Elmore, and M. J. Franklin.  Fast and reliable missing data contingency analysis with predicate-constraints.  In Proceedings of the 2020 ACM SIGMOD International Conference on Management of Data, pages 285–295, 2020.
  * [30] X. Liang, S. Sintos, Z. Shang, and S. Krishnan.  Combining aggregation and sampling (nearly) optimally for approximate query processing.  In Proceedings of the 2021 International Conference on Management of Data, pages 1129–1141, 2021.
  * [31] Q. Ma, A. M. Shanghooshabad, M. Almasi, M. Kurmanji, and P. Triantafillou.  Learned approximate query processing: Make it light, accurate and fast.  In CIDR, 2021.
  * [32] M. Olma, O. Papapetrou, R. Appuswamy, and A. Ailamaki.  Taster: self-tuning, elastic and online approximate query processing.  In 2019 IEEE 35th International Conference on Data Engineering (ICDE), pages 482–493. IEEE, 2019.
  * [33] O. Onyshchak.  Stock market dataset.  =https://www.kaggle.com/dsv/1054465, 2020.
  * [34] M. H. Overmars and J. van Leeuwen.  Worst-case optimal insertion and deletion methods for decomposable searching problems.  Information Processing Letters, 12(4):168–173, 1981.
  * [35] Y. Park, B. Mozafari, J. Sorenson, and J. Wang.  Verdictdb: Universalizing approximate query processing.  In Proceedings of the 2018 International Conference on Management of Data, pages 1461–1476, 2018.
  * [36] Y. Park, A. S. Tajik, M. Cafarella, and B. Mozafari.  Database learning: Toward a database that becomes smarter every time.  In Proceedings of the 2017 ACM International Conference on Management of Data, pages 587–602, 2017.
  * [37] J. Peng, D. Zhang, J. Wang, and J. Pei.  Aqp++ connecting approximate query processing with aggregate precomputation for interactive analytics.  In Proceedings of the 2018 International Conference on Management of Data, pages 1477–1492, 2018.
  * [38] W. H. Peter Bodik et al.  Intel wireless dataset.  http://db.csail.mit.edu/labdata/labdata.html, 2004.
  * [39] R. Poepsel-Lemaitre, M. Kiefer, J. von Hein, J.-A. Quiané-Ruiz, and V. Markl.  In the land of data streams where synopses are missing, one framework to bring them all.  2021\.
  * [40] K. Rong, Y. Lu, P. Bailis, S. Kandula, and P. Levis.  Approximate partition selection for big-data workloads using summary statistics.  arXiv preprint arXiv:2008.10569, 2020.
  * [41] N. Taxi and L. Commission.  New york city taxi trip records dataset.  https://www1.nyc.gov/site/tlc/about/tlc-trip-record-data.page, 2019\.
  * [42] J. S. Vitter.  Random sampling with a reservoir.  ACM Transactions on Mathematical Software (TOMS), 11(1):37–57, 1985\.
  * [43] B. Walenz, S. Sintos, S. Roy, and J. Yang.  Learning to sample: Counting with complex queries.  Proceedings of the VLDB Endowment, 13(3):390–402, 2019.
  * [44] L. Wang, R. Christensen, F. Li, and K. Yi.  Spatial online sampling and aggregation.  Proceedings of the VLDB Endowment, 9(3):84–95, 2015.
  * [45] Z. Yang, E. Liang, A. Kamsetty, C. Wu, Y. Duan, X. Chen, P. Abbeel, J. M. Hellerstein, S. Krishnan, and I. Stoica.  Deep unsupervised cardinality estimation.  arXiv preprint arXiv:1905.04278, 2019.

##  Appendix A Sampling from Kafka-like Systems

Random sampling is a key building block of JanusAQP because it is used in many
components: we use random samples to build the leaf layer of the partition
tree and initialize the reservoir that are later used to solve queries. During
the catch-up phase, we also use random samples to construct the summaries that
are stored in the leaf nodes. Random sampling affects both the accuracy and
performance of the entire JanusAQP system: a biased sample hurts the accuracy
of the system and expensive sampling operations leads to a higher latency and
lower throughput.

Designing a random sampler for message brokers like Kafka can be a non-trivial
task because the API of such systems usually does not provide random access to
the data. To retrieve data from a Kafka topic, a Kafka consumer has to send an
offset to the server indicating the location it wants to access data from.
Therefore, a naive random sampler can be implemented by using the poll() API
with a random offset.

However, such a naive implementation can be expensive because each poll could
retrieve a batch of thousands of tuples that are contiguous (therefore
biased). To guarantee unbiased sampling, we will have to keep only a small
portion from each batch and discard most of the tuples, set another random
offset, repeat the process until we collected enough samples.

To build a scalable, efficient and unbiased random sampler for Kafka, we need
more control over the polling process besides the offset, to be more specific,
we want to control the size of each poll. We propose two sampling methods:

Sequential Sampler A sequential sampler retrieves the data in a sequential
manner. In each poll, a random sample is drawn from the batch and the rest
tuples are discard. Sequential samplers work the best when the size of the
dataset is medium to large because the entire dataset is transferred from
Kafka to the sampler. Therefore, one overhead of this approach is the network
traffic of transferring the entire dataset, another drawback of a sequential
sampler is that the random samples are only available until the sampler have
retrieved the entire dataset, the high latency might not be acceptable in some
situations.

Singleton Sampler In each poll, a singleton sampler request one tuple from a
random offset, it repeats until enough sample has been collected. Singleton
samplers minimize the network traffic with the cost of server-side overhead
due to a more frequent use of the API. The main advantage of a singleton
sampler is that it offers lower latency because the random sample is built
incrementally, a small random sample can be available with a lower latency.

In general, we observe that, for light-weight sampling tasks or tasks that
requires low latency, a singleton sampler works the best. In scenarios where
the dataset size is medium to large and the expected latency is acceptable,
the sequential sampler might be preferred for a more consistent performance.
There might be an interesting mid-ground for sampling from Kafka-like systems
that can offer better efficiency and latency and we plan to investigate in a
future work.

####  A.0.1. Experiments

In this experiment, we study the performance of the two sampling strategies
discussed in Section A with the Intel wireless dataset. We implement samplers
with different poll size and measure the time cost of collecting 1 million
tuples from Kafka. Essentially, this experiment measures the overhead
introduced by transferring the data and calling Kafka API.

Results can be found in Table 4 where the Singleton sampler is represented by
the row with pollSize equals to 1. Rows with pollSize larger than 1 are
sequential samplers which retrieve the entire dataset and sample from each
poll.

pollSize | nPolls | total(ms) | ms/poll | EquivSingletonSR
---|---|---|---|---
1 | 1000000 | 19000 | 0.019 | —
10 | 10000 | 4000 | 0.04 | 0.21
100 | 10000 | 2000 | 0.02 | 0.11
1000 | 1000 | 1500 | 1.5 | 0.08
10000 | 100 | 1400 | 14 | 0.075
100000 | 10 | 1700 | 17 | 0.09
Table 4. We use a singleton sampler (pollSize=1) and sequential samplers
(pollSize¿1) to sample 1 million tuples, given the latency of each sequential
samplers, we derive a equivalent sample rate of the singleton sampler indicate
a sample rate above which sequential samplers take less time to collect all
requested random samples. If the expected total latency is acceptable,
sequential samplers might be preferred in these scenarios.

Because we have to wait a total(ms) time for sequential samplers to collect
the samples. We calculate the applicable sampleRate for each sequential
sampler above which they can achieve a lower total(ms). For example, if our
sample rate is 10% which is larger than 7.5% achieved by the best Sequential
sampler with pollSize of 10000, the singleton sampler will take more time than
the sequential sampler to complete the sampling process. And the sequential
samplers might be preferred if the estimated latency of total(ms) is
acceptable.

In JanusAQP, because the sample rate we use during initialization is no larger
than 1%, we always use a singleton sampler during initialization, i.e. to
collect sample to build the partition tree and to initialized the strata. For
the catch-up phase, if our catch-up rate is larger than 10%, when dealing with
a dataset of medium to large size with an acceptable latency, we will prefer
to use a sequential sampler because we can have a complete catch-up with a
better accuracy, otherwise, a singleton sampler is preferred for the low
latency it offers and we will keep it running in background.

##  Appendix B Variance Estimator Details

Using algebra we have the following formulas: For a COUNT/SUM query $q$ (for
COUNT we assume that $t.a=1$ for any tuple $t$),

$w_{i}\cdot mean(\phi_{q}(S_{i}))=\frac{1}{m_{i}}\sum_{t\in
S_{i}}\phi(t)=\frac{N_{i}}{m_{i}}\sum_{t\in S_{i}\cap q}t.a$ $w_{i}\cdot
mean(\phi_{q}(H_{i}))=\frac{1}{h_{i}}\sum_{t\in
H_{i}}\phi(t)=\frac{N_{i}}{h_{i}}\sum_{t\in H_{i}}t.a$
$w_{i}^{2}\frac{var(\phi_{q}(S_{i}))}{m_{i}}=\frac{N_{i}^{2}}{m_{i}^{3}}\left[m_{i}\sum_{t\in
S_{i}\cap q}t.a^{2}-\left(\sum_{t\in S_{i}\cap q}t.a\right)^{2}\right]$
$w_{i}^{2}\frac{var(\phi_{q}(H_{i}))}{h_{i}}=\frac{N_{i}^{2}}{h_{i}^{3}}\left[h_{i}\sum_{t\in
H_{i}}t.a^{2}-\left(\sum_{t\in H_{i}}t.a\right)^{2}\right].$

For an AVG query $q$ we have

$w_{i}\cdot mean(\phi_{q}(S_{i}))=\frac{N_{i}}{|S_{i}\cap
q|\cdot\sum_{i\in\mathcal{I}_{q}}N_{i}}\sum_{t\in S_{i}\cap q}t.a$ $w_{i}\cdot
mean(\phi_{q}(H_{i}))=\frac{N_{i}}{h_{i}\cdot\sum_{i\in\mathcal{I}_{q}}N_{i}}\sum_{t\in
H_{i}}t.a,$

where $\mathcal{I}_{q}$ is the set of all leaf nodes intersected (either
partially or fully) by the query $q$.

$w_{i}^{2}\frac{var(\phi_{q}(S_{i}))}{m_{i}}=\frac{w_{i}^{2}}{m_{i}\cdot|S_{i}\cap
q|^{2}}\left[m_{i}\sum_{t\in S_{i}\cap q}t.a^{2}-\left(\sum_{t\in S_{i}\cap
q}t.a\right)^{2}\right]$
$w_{i}^{2}\frac{var(\phi_{q}(H_{i}))}{h_{i}}=\frac{w_{i}^{2}}{h_{i}^{3}}\left[h_{i}\sum_{t\in
H_{i}}t.a^{2}-\left(\sum_{t\in H_{i}}t.a\right)^{2}\right]$

##  Appendix C Partition algorithms

###  C.1. Maximum variance under updates

An important procedure of all our algorithms is to find what is the maximum
error of a query that lies completely in a leaf node or more generally in a
rectangle. We explain how to do it for the queries COUNT, SUM, AVG. In
particular, we describe a stronger result: We construct a dynamic data
structure over a set of samples $S$ with efficient update time such that given
a query rectangle, it returns an approximation of the variance of the query
with the maximum variance in the query rectangle. This result leads to very
efficient dynamic algorithms for checking the maximum variance and re-
constructing a new partition, as we will see in the next subsections. Let
$\mathcal{M}(R)$ be the value of the variance returned by our approximation
algorithm in a rectangle $R$.

COUNT queries. For COUNT queries it is known [30] that the query with the
maximum variance in a rectangle $R$ contains exactly $|R\cap S|/2$ samples.
Hence, we construct a dynamic range tree $T$ over $S$ with space
$O(m\log^{d}m)$. $T$ can be constructed in $O(m\log^{d}m)$ time and can be
updated in $O(\log^{d}m)$ time. Given a query rectangle $R$, we run a binary
search using $T$ to find two rectangles that contain $|R_{i}|/2$ items. The
query runs in $O(\log^{d}m)$ time.

SUM queries. For SUM queries it is known [30] that we can get a
$\frac{1}{4}$-approximation of the maximum variance query inside $R$ with the
following simple approach: Find two non-intersecting rectangles $R_{1},R_{2}$
such that $R_{1}\cup R_{2}=R$, and $|R_{1}\cap S|=|R_{2}\cap S|=|R\cap S|/2$.
Then, they compare $\sum_{t\in R_{1}\cap S}t.a^{2}$ with $\sum_{t\in R_{2}\cap
S}t.a^{2}$ and return the variance of the rectangle with the largest sum of
squares. The variance of this rectangle is a $\frac{1}{4}$-approximation of
the maximum variance in $R$. Range trees work for any aggregation function so
we can also use them to compute the sum of the values squared in a rectangle
or the variance of a rectangle. Hence, we can use a dynamic range tree as we
had in the COUNT case returning a $\frac{1}{4}$-approximation of the maximum
variance. This data structure has exactly the same complexities as the data
structure for COUNT queries.

AVG queries. For AVG queries the offline algorithms of [30] cannot be
efficiently extended to the dynamic case we are interested in (is not known
how to achieve $\textrm{polylog}m$ update time). Here we propose a new dynamic
data structure for finding an approximation of the maximum variance AVG query
in a query rectangle efficiently. The new data structure we propose does not
only handle updates efficiently, unlike the data structures in [30], but it
also improves the approximation factor for any dimension $d$.

For a set of samples $S$ we construct a dynamic range tree $T^{\prime}$. We
also initialize an empty dynamic data structure $T$ that stores weighted
rectangles. Given a query rectangle $R$, it returns the rectangle with the
highest weight that lies completely inside $R$. For example $T$ can be a
dynamic range tree in $2d$ dimensions storing each rectangle as a $2d$ point
by creating a point from its two main opposite corners. Notice that every
$d$-th level node $u$ of $T^{\prime}$ corresponds to a rectangle $R_{u}$. If
$|R_{u}\cap S|\leq\delta m$ then we add $R_{u}$ in $T$ with weight
$S(R_{u})=\sum_{t\in R_{u}\cap S}t.a^{2}$. If $\delta m\leq|R_{u}\cap S|\leq
2\delta m$ we split $R_{u}$ into two rectangles $R_{u_{1}},R_{u_{2}}$ such
that $R_{u_{1}}\cup R_{u_{2}}=R_{u}$ and $|R_{u_{1}}\cap S|=|R_{u_{2}}\cap
S|=|R_{u}\cap S|/2$. We add $R_{u_{1}},R_{u_{2}}$ in $T$ with weights
$S(R_{u_{1}})=\sum_{t\in R_{u_{1}}\cap S}t.a^{2}$, $S(R_{u_{2}})=\sum_{t\in
R_{u_{2}}\cap S}t.a^{2}$. The tree $T^{\prime}$ can be constructed in
$\tilde{O}(m)$ time and it has $\tilde{O}(m)$ space. In $T$ we might insert
$\tilde{O}(m)$ rectangles so it has $\tilde{O}(m)$ space and can be
constructed in $\tilde{O}(m)$ time. For any insertion or deletion of a sample
in $S$, $T^{\prime}$ can be updated in $\tilde{O}(1)$ time by modifying at
most $\tilde{O}(1)$ nodes. For each modified node we update accordingly the
corresponding rectangle in $T$ in $\tilde{O}(1)$ time. Furthermore, after an
update we traverse the $d$-th level of $T^{\prime}$ from the updated leaf
nodes to their roots inserting or removing rectangles from $T$ accordingly
based on the number of points they contain. Using [5, 34, 12] we can propose a
simple dynamic data structure with amortized update time guarantee that can be
extended to worst case guarantee by standard techniques [12]. Overall our data
structure can be constructed in $O(m\log^{3d}m)$ time, has $O(\log^{3d}m)$
space and can be updated in $O(\log^{3d+1}m)$ time.

Given a query rectangle $R$ such that $R\cap S>2\delta m$ (as we had in [30])
we show how to return an approximation of the maximum variance query
efficiently. We search $T$ using the query rectangle $R$ and we get a set of
$\tilde{O}(1)$ canonical nodes that contain all rectangles completely inside
$R$. From the canonical subsets we get the rectangle $q^{\prime}$ inside
rectangle $R$ with the largest weight. If $|q^{\prime}\cap S|<\delta m$ then
using $T^{\prime}$ we run binary search over all dimensions until we find an
expansion of $q^{\prime}$ that contains exactly $\delta m$ samples. This can
be done in $\tilde{O}(1)$ time. Without loss of generality assume that
$q^{\prime}$ contains exactly $\delta m$ samples. Using $T^{\prime}$ we
measure the variance of $q^{\prime}$ in $R$ in $\tilde{O}(1)$ time. In the end
we return the variance of $q^{\prime}$ as the approximation of the maximum
variance AVG query in the query rectangle $R$. The query procedure takes
$\tilde{O}(1)$ time.

###### Lemma C.1.

It holds that $\nu_{s}(q^{\prime})\geq\frac{1}{4\log^{d+1}m}\mathcal{V}(R)$.

###### Proof.

Let $q$ be the AVG query (a rectangle) with the maximum variance in $R$. It is
known from [30] that $q$ contains at most $2\delta m$ and at least $\delta m$
samples. Since $|R\cap S|\geq 2\delta m$ we have that

$|R\cap S|\sum_{t\in q^{\prime}\cap S}t.a^{2}-\left(\sum_{t\in q^{\prime}\cap
S}t.a\right)^{2}\geq\frac{|R\cap S|}{2}\sum_{t\in q^{\prime}\cap S}t.a^{2},$

following from Lemma A.2 in [30]. Next, notice that a query procedure on
$T^{\prime}$ with the query range $q$ would give a set of $\log^{d}m$
canonical rectangles that cover $q$ where each of them contains at most
$2\delta m$ samples. We note that for each of these canonical rectangles in
$T^{\prime}$ there are at most two rectangles in $T$ containing the same
items. Let $X$ be the set of rectangles in $T$ corresponding to the canonical
rectangles in $T^{\prime}$. From its definition it holds that $|X|\leq
2\log^{d+1}m$. All of these rectangles in $X$ lie completely inside $R$ so the
query procedure will consider them to find the rectangle with the largest
weight. Hence it holds that

$\sum_{t\in q^{\prime}\cap S}t.a^{2}\geq\max_{x\in X}\sum_{t\in x\cap
S}t.a^{2}\geq\frac{1}{2\log^{d+1}m}\sum_{t\in q\cap S}t.a^{2}.$

Overall we have, (for simplicity $|R|=|R\cap S|$)

$\displaystyle\nu_{s}(q^{\prime})$
$\displaystyle=\frac{1}{|R|\cdot|q^{\prime}\cap S|^{2}}\left[|R\cap
S|\sum_{t\in q^{\prime}\cap S}t.a^{2}-\left(\sum_{t\in q^{\prime}\cap
S}t.a\right)^{2}\right]$ $\displaystyle\geq\frac{1}{|R|\cdot|q^{\prime}\cap
S|^{2}}\frac{|R\cap S|}{2}\sum_{t\in q^{\prime}\cap S}t.a^{2}$
$\displaystyle\geq\frac{1}{|R|\cdot|q^{\prime}\cap
S|^{2}}\frac{1}{4\log^{d+1}m}|R\cap S|\sum_{t\in q\cap S}t.a^{2}$
$\displaystyle\\!\geq\\!\frac{|q\cap S|^{2}}{|q^{\prime}\cap
S|^{2}}\frac{1}{4\log^{d+1}m}\frac{1}{|R|\\!\cdot\\!|q\cap S|^{2}}\left[|R\cap
S|\\!\\!\\!\sum_{t\in q\cap S}\\!\\!t.a^{2}\\!-\\!\left(\\!\sum_{t\in q\cap
S}\\!\\!t.a\\!\right)^{2}\\!\right]$
$\displaystyle\geq\frac{1}{4\log^{d}m}\mathcal{V}(R)$

∎

If $d=1$ we can modify the data structure so that it gives an approximation
factor $4$.

Overall, given a set $S$ of $m$ points in $d$ dimensions we can construct a
data structure of space $O(m\cdot\textrm{polylog}(m))$ in
$O(m\cdot\textrm{polylog}(m))$ time with update time $O(\textrm{polylog}(m))$
such that given a query rectangle it finds an approximation of the maximum
variance COUNT, SUM or AVG query inside $R$ in $O(\textrm{polylog}(m))$ time.

###  C.2. Partition for $d=1$

For COUNT queries the optimum partition in $1$D consists of equal size buckets
(intervals) so we can find the new partition in $O(k\log m)$ time by
maintaining the order of the samples $S$ under insertion or deletion using a
balanced search binary tree where the samples are stored in the leaf nodes.
Such a tree can be updated in $O(\log m)$ time while the order of the samples
on the real line is the same as the order of the leaf nodes from left to
right. When we have to (re-)construct the partition we find the right endpoint
of each bucket using the search binary tree. Overall, we need $O(\log m)$ time
to update the tree and $O(k\log m)$ time to construct a new partition.

Next, we focus on SUM and AVG queries.

Bounding the error. We first show a lemma that bounds the maximum length of
the largest possible confidence interval among queries that intersect one
bucket of the partition. We assume that the value of any item in $\mathcal{D}$
is bounded by a maximum value $\mathcal{U}$ and a minimum non-zero value
$\mathcal{L}$. We allow items to take zero values since this is often the case
in real datasets but no item with positive value less than $\mathcal{L}$ or
larger than $\mathcal{U}$ exists. We assume that
$\mathcal{U}=O(\textrm{poly}(N))$ and
$\mathcal{L}=\Omega(1/\textrm{poly}(N))$.

###### Lemma C.2.

Let $R$ be any rectangle and let $\mathcal{V}_{S}(R),\mathcal{V}_{A}(R)>0$ be
the variance of the SUM and AVG query respectively with the maximum variance
in $R$. Then it holds that
$\frac{\mathcal{L}}{\sqrt{2}}\leq\sqrt{\mathcal{V}_{S}(R)}\leq N\mathcal{U}$
and
$\frac{\mathcal{L}}{\sqrt{2}N}\leq\sqrt{\mathcal{V}_{A}(R)}\leq\sqrt{N}\mathcal{U}$.

###### Proof.

Without loss of generality let $q$ be the SUM or AVG query with the maximum
variance in $R$.

First we focus on SUM queries. Let $N_{R}=|R\cap\mathcal{D}|$ be the number of
total tuples in $R$. Unless $\nu_{s}(q)=0$, from [30], we know that there
exists a query $q^{\prime}$ with $|q^{\prime}|=|R|/2$ such that
$\nu_{s}(q)\geq\frac{N_{R}^{2}}{|R|^{3}}\frac{|R|}{2}\sum_{t\in
q^{\prime}}t.a^{2}$, where $\sum_{t\in q^{\prime}}t.a^{2}>0$. We also have
$\sum_{t\in q^{\prime}}t.a^{2}\geq\mathcal{L}^{2}$ (and $w_{u}=1$) leading to
$\sqrt{\mathcal{V}_{S}(R)}\geq\frac{N_{R}}{\sqrt{2}|R|}\mathcal{L}\geq\frac{\mathcal{L}}{\sqrt{2}}$.
Furthermore, we have
$\nu_{s}(q)\leq\frac{N_{R}^{2}}{|R|^{2}}|R|^{2}\mathcal{U}^{2}\leq
N^{2}\mathcal{U}^{2}$ leading to $\sqrt{\mathcal{V}_{S}(R)}\leq N\mathcal{U}$.

Next, we consider AVG queries. Unless $\nu_{s}(q)=0$, from [30], we know that
there exists a query $q^{\prime}$ with $|q^{\prime}|=\delta m\leq|R|/2$ such
that
$\nu_{s}(q^{\prime})\geq\frac{1}{|R|\delta^{2}m^{2}}\frac{|R|}{2}\sum_{t\in
q^{\prime}}t.a^{2}$, where $\sum_{t\in q^{\prime}}t.a^{2}>0$. We also have
$\sum_{t\in q^{\prime}}t.a^{2}\geq\mathcal{L}^{2}$ (and $w_{u}=1$) leading to
$\sqrt{\mathcal{V}_{A}(R)}\geq\frac{1}{\sqrt{2}\delta
m}\mathcal{L}\geq\frac{\mathcal{L}}{\sqrt{2}N}$. Furthermore, we have
$\nu_{s}(q)\leq\frac{|R|}{|q|^{2}}\mathcal{U}^{2}\leq|R|\mathcal{U}^{2}\leq
N\mathcal{H}^{2}$ leading to
$\sqrt{\mathcal{V}_{A}(R)}\leq\sqrt{N}\mathcal{U}$. ∎

Since, $\mathcal{U},\mathcal{L}$ are bounded by a polynomial with respect to
$N$, we have that the length of the longest confidence interval is bounded by
$O(\textrm{poly}(N))$ and $\Omega(1/\textrm{poly}(N))$, i.e.

$\Omega(1/\textrm{poly}(N))\leq\sqrt{\mathcal{V}_{S}(R)},\sqrt{\mathcal{V}_{A}(R)}\leq
O(\textrm{poly}(N)).$

Description of algorithm. We describe the partition algorithm for SUM queries.
The procedure is identical for AVG queries and we highlight the differences in
the end of this section. For a parameter $\rho\in\mathbb{R}$ with $\rho>1$,
let $E=\\{\rho^{t}\mid
t\in\mathbb{Z},\frac{\mathcal{L}}{\sqrt{2}}\leq\rho^{t}\leq
N\mathcal{U}\\}\cup\\{0\\}$, be the discretization of the range
$[\frac{\mathcal{L}}{\sqrt{2}},N\mathcal{U}]$, i.e., the lower and upper bound
of the longest confidence interval (assuming queries completely inside one
bucket), by the multiplicative parameter $\rho$. For an interval $b$, let
$\mathcal{M}(b)$ be the approximation of the query with the maximum variance
in bucket $b$ (supporting updates) as described in Section C.1. We run a
binary search on the values of $E$. For each value $e\in E$ we consider, we
try to construct a partition of $k$ buckets such that in each bucket the
length of the longest confidence interval is at most $e$. If there exists such
a partition we continue the binary search with values $e^{\prime}<e$. If there
is no such a partition we continue the binary search with values
$e^{\prime}>e$. In the end of the binary search we return the last partition
that we were able to compute.

It remains to describe how to check if a partition with $k$ buckets
(intervals) with maximum length confidence interval at most $e$ exists. We
start with the leftmost sample, say $t_{1}$, which is the left boundary of the
first bucket. In order to find its right boundary we run a binary search on
the samples $S$. Let $t_{j}$ be one of the right boundaries we check in the
binary search, and let $b_{1}=[t_{1},t_{j}]$. If $\sqrt{\mathcal{M}(b_{1})}<e$
then we continue the binary search with a sample at the right side of $t_{j}$
(larger bucket). Otherwise, we continue the binary search with a sample at the
left side of $t_{j}$ (smaller bucket). When we find the maximal bucket with
longest confidence interval at most $e$ we continue with the second bucket
repeating the same process for at most $k$ buckets. In the end, if all samples
in $S$ are contained in $k$ buckets then we return that there exists a
partition (with $k$ buckets) with maximum variance at most $e$. If we cannot
cover all samples in $k$ buckets then we return that there is no partition
(with $k$ buckets) with maximum variance at most $e$.

The same algorithm also works for AVG queries. The only difference is that $E$
should be defined with respect to the upper and lower bound of the longest
confidence interval, as shown in Lemma C.2.

Correctness. Before we start with the correctness proof of our algorithm we
recall that in [30] we showed that under a mild assumption, for two buckets
$b_{i},b_{j}$ if $b_{i}\subseteq b_{j}$ then
$\sqrt{\mathcal{V}(b_{i})}\leq\sqrt{\mathcal{V}(b_{j})}$, namely the length of
the longest confidence interval in $b_{i}$ is smaller than the length of the
longest confidence interval in $b_{j}$. This is the monotonic property of the
longest confidence interval.

We assume that $\mathcal{M}(b_{i})$ computes a
$\frac{1}{\gamma}$-approximation of the maximum variance in $b_{i}$, i.e.,
$\mathcal{M}(b_{i})\geq\frac{1}{\gamma}\mathcal{V}(b_{i})$. Let
$\mathcal{R}^{*}$ be the optimum partition and let $b^{*}$ be the bucket that
contains the query with the longest confidence interval in $\mathcal{R}^{*}$.
First, we notice that if $e\geq\sqrt{\mathcal{V}(b^{*})}$ then we always find
a partition with longest confidence interval at most $e$. We can show it by
induction on the right boundaries of the buckets (intervals) and the monotonic
property of confidence intervals. For the base case, let
$b_{1}^{*}=[t_{1},t_{2}]$ be the first bucket of partition $\mathcal{R}^{*}$.
The procedure $\mathcal{M}$ always underestimates the maximum variance in an
interval so the binary search in our procedure will consider the right
boundary to be greater than $t_{2}$. Let $t_{i}$ be the right boundary of the
$i$-th bucket in $\mathcal{R}^{*}$ and let assume that the $i$-th bucket in
our procedure has a right boundary $t_{j}\geq t_{i}$. We consider the
$(i+1)$-th bucket in $\mathcal{R}^{*}$ with boundaries $[t_{i+1},t_{r}]$. We
show that the $(i+1)$-th bucket in our procedure has a right boundary at least
$t_{r}$. Let $[t_{a},t_{b}]$ be the boundaries of the $(i+1)$-th bucket in our
procedure. We have $t_{a}\geq t_{i+1}$. If $t_{a}=t_{i+1}$ then $t_{b}\geq
t_{r}$ as in the basis case. If $t_{a}>t_{i+1}$ then because of the monotonic
property of the confidence intervals and the fact that the $\mathcal{M}$
procedure underestimates the maximum variance we also have that $t_{b}\geq
t_{r}$. Let $e^{\prime}$ be the smallest value in $E$ such that
$\sqrt{\mathcal{V}(b^{*})}\leq e^{\prime}$. Because of the previous
observation our algorithm always returns at least a valid partition for an
$e\leq e^{\prime}$. For every bucket $b$ of this partition,
$\sqrt{\mathcal{M}(b)}\leq e$. Let $b^{\prime}$ be the bucket in the returned
partition containing the query with the longest confidence interval in the
partition. We have,
$\sqrt{\mathcal{V}(b^{\prime})}\leq\sqrt{\gamma\mathcal{M}(b^{\prime})}\leq\sqrt{\gamma}e\leq\sqrt{\gamma}e^{\prime}\leq\rho\sqrt{\gamma}\sqrt{\mathcal{V}(b^{*})}$.

From Section C.1 we have that $\gamma=4$ for SUM and AVG queries queries. So
we get a partition where the maximum error is within $2\rho\sqrt{2}$ of the
optimum error for SUM queries and within $2\rho$ of the optimum error for AVG
queries.

Running time. We assume that $\mathcal{M}(\cdot)$ can be computed in $M$ time.
Since, $\mathcal{L},\mathcal{U}$ are polynomially bounded on $N$ we have that
$|E|=O(\log_{\rho}N)$ and it can be constructed in $O(\log_{\rho}N)$ time. The
binary search over $E$ takes at most $O(\log\log_{\rho}N)$ steps. For each
value $e\in E$ of the binary search we check if there is a partition with $k$
buckets and longest confidence interval at most $e$. For each possible bucket
we run a binary search over the samples $S$ and we run the procedure
$\mathcal{M}$ to get an approximation of the maximum variance. Hence, we can
decide if there exists a partition with confidence interval $e$ in $O(kM\log
m)$ time. Overall, our algorithm takes $O(kM\log m\log\log_{\rho}N)$. If
$\rho$ is a constant, for example $\rho=2$ then the running time is $O(kM\log
m\log\log N)$. From Section C.1 we have that in $1$-dimension $M=O(\log m)$
for SUM and $M=O(\log^{2}m)$ for AVG queries. Notice that if we skip the
$\log$ factors the running time depends only linearly on the number of buckets
$k$.

###  C.3. Partition for higher dimensions

We construct a partition by building a k-d tree using the dynamic procedures
$\mathcal{M}$ as shown in Section C.1. Using the results of [30] we could
construct a near optimum k-d tree in time $O(km)$ skipping the $\log$ factors.
Here, we use our new results from Section C.1 to construct a k-d tree faster
(roughly $O(k)$) with better approximation approximation guarantees.

We start by constructing a dynamic data structure from Section C.1 over the
initial set of samples $S$. Assume that after a number of insertions and
deletions in $S$ we want to (re)construct the tree structure $\mathcal{T}$
over $S$. We construct a partition tree on $S$ using ideas from the balanced
k-d tree construction. We pre-define an ordering of the dimensions. Each node
$u$ of the tree is associated with a rectangle $R_{u}$. We build the tree in
$k$ iterations in a top-down manner starting from the root $v$ such that
$\mathcal{D}\subset R_{v}$. In any iteration we store and maintain the
approximate maximum variance queries of every leaf node in a max heap $C$. In
the end of the $i$-th iteration we have a tree of $i$ leaf nodes. Let $u$ be
the leaf node with the maximum $\mathcal{M}(R_{u})$ value in $C$. We remove
its value from $C$. We find the medium coordinate with respect to the next
dimension in the ordering (in this branch of the tree) among the samples
$R_{u}\cap S$. We split $R_{u}$ on the median into two rectangles
$R_{u_{1}},R_{u_{2}}$ such that $R_{u_{1}}\cup R_{u_{2}}=R_{u}$ and we
construct the children $u_{1},u_{2}$ of the parent node $u$. Using the
algorithms from Section C.1 we compute
$\mathcal{M}(R_{u_{1}}),\mathcal{M}(R_{u_{2}})$ and we insert their values in
the max-heap $C$. We continue with the same way until we construct a tree
$\mathcal{T}$ with $k$ leaf nodes (buckets).

As we showed in [30] such a tree construction returns a partition which is
near optimal with respect to the optimum partition tree construction following
the same splitting criterion: split on the median of the leaf node with the
largest (real) maximum variance query. In our case we do not always split the
nod with the real largest error since we use the approximation function
$\mathcal{M}(\cdot)$.

For any query our data structure from Section C.1 can be updated in
$\tilde{O}(1)$ time. Given a (re-)partition activation query over a set $S$ of
$m$ samples we can construct a new $\mathcal{T}$ with the following
guarantees: For SUM queries, $\mathcal{T}$ can be constructed in
$O(k\log^{d}m)$ time with approximation factor $2\sqrt{k}$. For COUNT queries
we get the same construction time $O(k\log^{d}m)$ but the tree we construct is
optimum (with respect to the partition tree with same split criterion). For
AVG queries, $\mathcal{T}$ can be constructed in $O(k\log^{2d}m)$ time with
approximation factor $2\log^{d/2}m$. In all cases we can construct near-
optimum partitions in $\tilde{O}(k)$ time.

###  C.4. Re-Partitioning Triggers

A key contribution of JanusAQP is continuous re-optimization of the
partitioning. We describe how JanusAQP tracks the variances of the current
partitions and decides when to re-partition. We also propose two ways to re-
partition: partial or full re-partitioning.

Assume that the current partitioning is $\mathcal{R}$ and let
$\mathcal{M}(\mathcal{R})$ be the (approximate) maximum variance query with
respect to the current set of samples $S$. The automatic procedure first
checks the number of samples in each bucket (leaf node) of the current
$\mathcal{T}$. If there is a leaf node $i$ associated with partition $R_{i}$
such that $|S_{i}|<<\frac{1}{\alpha}\log m$ (recall that $\alpha$ is the
sampling rate) then there are not enough samples in $u$ to make robust
estimators. Hence, we need to find a new re-partition of $S$. Even if the
number of samples in each bucket is large our system might enable a re-
partition: For a partition $R_{i}$ in the leaf node layer of $\mathcal{T}$ let
$\mathcal{M}_{i}=\mathcal{M}(R_{i})$ be the (approximate) maximum variance at
the moment we constructed $\mathcal{T}$. Let $\beta>1$ be a parameter that
controls the maximum allowable change on the variance. It can either be
decided by the user or we can set it to $\beta=10$. Assume that an update
occurred in the leaf node associated with the partition $R_{i}$. After the
update we run the function $\mathcal{M}_{i}^{\prime}=\mathcal{M}(R_{i})$ and
we update $\mathcal{M}(\mathcal{R})$ if needed. If
$\frac{1}{\beta}\mathcal{M}_{i}\leq\mathcal{M}_{i}^{\prime}\leq\beta\mathcal{M}_{i}$
then the new maximum variance in partition $b_{i}$ is not very different than
before so we do not trigger a re-partition. Otherwise, the maximum variance in
bucket $b_{i}$ changed by a factor larger than $\beta$ from the initial
variance $\mathcal{M}_{i}$. In this case a re-partition might find a new tree
with smaller maximum error. We compute a new partitioning
$\mathcal{R}^{\prime}$ and hence a new tree $\mathcal{T}$. If
$\mathcal{M}(\mathcal{R}^{\prime})<\frac{1}{\beta}\mathcal{M}(\mathcal{R})$
then we activate a re-partition restarting the catch-up phase over the new
tree $\mathcal{T}$. On the other hand, if
$\mathcal{M}(\mathcal{R}^{\prime})\geq\frac{1}{\beta}\mathcal{M}(\mathcal{R})$
then our current partitioning $\mathcal{R}$ is good enough (its worst error is
close to the optimum one) so we can still use it. Of course, the user can also
manually trigger re-partitioning. For example, the user can choose to re-
partition once every hour, day, or after $\tau$ insertions and deletions have
occurred.

Next, we propose two ways to re-partition the index. In particular, the user
can select either partial re-partitioning or full re-partitioning. full re-
partitioning is easy; using the algorithms from the previous subsections we
can construct a new partitioning and a new tree structure in near-linear time
with respect to the samples. Hence, we focus on partial re-partitioning.
Instead of re-partitioning the entire space we can only re-partition the area
around the ”problematic” leaf node. Let $b_{i}$ be this leaf node with high
error or small number of samples. In order to define the neighboring area
around $b_{i}$ we propose either a predefined way or an automatic way. In both
cases, the neighboring area is defined by a parameter $\psi$, which is the
level of the tree above $b_{i}$ that the tree needs to be updated. In the
predefined way, the parameter $\psi$ is a known parameter. We find the node
$v$ which is defined as an ancestor of the leaf node $b_{i}$, $\psi$ levels
above $b_{i}$. Let $\mathcal{T}_{u}$ is the subtree with root node $u$ and let
$l_{u}$ be the number of leaf nodes in $\mathcal{T}_{u}$. Using the algorithms
from the previous subsections we find a near optimum partition starting from
node $u$ with $l_{u}$ leaf nodes. The running time is near-linear with respect
to the samples stored $\mathcal{T}_{u}$. In the automatic way, we do not know
the parameter $\psi$ upfront so we try different values of $\psi$ running a
binary search on the levels of the tree until we find a partition with low
enough error. For each different value of $\psi$ we try, we run the same
partial re-partitioning algorithm as in the static case starting from the node
$u$ we are considering in the binary search.

Generally, partial re-partitioning is faster than the full re-partitioning
since it only suffices to find a better partitioning in a small area of the
space. Furthermore, in partial re-partitioning we can still keep all the
current estimations in all nodes of $\mathcal{T}\setminus\mathcal{T}_{u}$,
i.e., the nodes of the tree that are not changed. Hence, the error of queries
after a partial re-partitioning is also lower than the error of the queries
immediately after a full re-partitioning. However, in both cases we need to
restart the catch-up phase over the new tree in order to get good estimators
to the nodes that were changed by the partial re-partitioning. Recall that we
cannot get samples from a particular area (ideally samples that stored in the
leaf nodes of $\mathcal{T}_{u}$) hence we run the catch-up phase getting
samples from the entire space. We finally note that while the catch-up phase
considers samples from the entire space, we only use these samples to improve
the estimators in the nodes that are still under-represented, i.e., the catch-
up phase time threshold for these nodes has not been completed.