countblanklinestrue[t]#1

# Odyssey: A Journey in the Land of Distributed Data Series Similarity Search

Manos Chatzakis
EPFL
<EMAIL_ADDRESS>
and Panagiota Fatourou
FORTH, ICS & University of Crete, CSD
<EMAIL_ADDRESS>
and Eleftherios Kosmas
FORTH, ICS & Hellenic Mediterranean University
& University of Crete, CSD
<EMAIL_ADDRESS>
and Themis Palpanas
Université Paris Cité & IU
<EMAIL_ADDRESS>
and Botao Peng
Institute of Computing Technology
Chinese Academy of Sciences
<EMAIL_ADDRESS>
Work conducted in collaboration with ICS-FORTH and Université Paris CitéPart
of work performed while P. Fatourou was working as a Marie Sklodowska-Curie
Individual Fellow (project PLATON, GA No 101031688) at Université Paris Cité,
France.

###### Abstract

This paper presents Odyssey, a novel distributed data-series processing
framework that efficiently addresses the critical challenges of exhibiting
good speedup and ensuring high scalability in data series processing by taking
advantage of the full computational capacity of modern distributed systems
comprised of multi-core servers. Odyssey addresses a number of challenges in
designing efficient and highly-scalable distributed data series index,
including efficient scheduling, and load-balancing without paying the
prohibitive cost of moving data around. It also supports a flexible partial
replication scheme, which enables Odyssey to navigate through a fundamental
trade-off between data scalability and good performance during query
answering. Through a wide range of configurations and using several real and
synthetic datasets, our experimental analysis demonstrates that Odyssey
achieves its challenging goals.

This paper appeared in PVLDB 2023, Volume 16.

##  1 Introduction

Motivation. Processing large collections of real-world data series is nowadays
one of the most challenging and critical problems for a wide range of diverse
application domains, including finance, astrophysics, neuroscience,
engineering, and others Palpanas (2015); Zoumpatianos and Palpanas (2018);
Palpanas and Beckmann (48(3). Such applications produce big collections of
ordered sequences of data points, called data series. When data series
collections are generated, they need to be analyzed in order to extract useful
knowledge Kashino et al. (1999); Ye and Keogh (2009); Huijse et al. (2014);
Raza et al. (2015); Boniol et al. (2020); Boniol and Palpanas (2020); Linardi
et al. (2020); Wang et al. (2023a). This analysis usually encompasses
answering similarity search queries Palpanas (2015); Echihabi et al. (2018,
2019), which are useful in a variety of downstream analysis tasks Echihabi et
al. (2020, 2021a, 2021b). Moreover, several applications across domains are
very sensitive to the accuracy of the results Palpanas and Beckmann (48(3);
Bagnall et al. (9(7), and thus, require exact query answering Echihabi et al.
(2018), which is our focus.

As the size of the data series collections grows larger Palpanas (2015, 2017);
Palpanas and Beckmann (48(3), recently proposed State-of-the-Art (SotA) data
series indexes exploit parallelism through the use of multiple threads and the
utilization of the SIMD capabilities of modern hardware Peng et al. (2018,
2020a); Echihabi et al. (2022). However, the unprecedented growth in size that
data series collections experience nowadays, renders even SotA parallel data
series indexes inadequate Palpanas (2017); Echihabi et al. (2018, 2019);
Palpanas and Beckmann (48(3); Bagnall et al. (9(7); Gogolou et al. (2019);
Echihabi et al. (2023), mainly due to the large number of random disk page
reads required for exact query answering Echihabi et al. (2018). To address
these issues, fast in-memory solutions have been proposed Peng et al. (2020b,
2021a, 2021b). However, these solutions do not take advantage of distributed
systems, and hence, are limited by the amount of memory of a single machine.
This is the limitation we address, thus allowing the above SotA solutions to
handle datasets that far exceed the main memory capacity of any single node.

Challenges. In the context of data series similarity search, exact query
answering is very demanding in terms of resources, even when using a data
series index. We need to either prune, or visit every leaf of the index.
Previous works Echihabi et al. (2018); Gogolou et al. (2019) though, have
shown that pruning is not very effective, especially for some hard datasets.

The main goal we need to satisfy is (naturally) _scalability_. That is,
increasing the available hardware resources (e.g., the number of nodes) should
decrease the time cost, ideally by an equivalent amount, or should enable to
process an equivalent amount of additional data (at about the same time cost).
In order to meet this goal, we need to ensure that all nodes of the
distributed system equally contribute to completing the work, during the
entire duration of the execution. In turn, this translates to producing
effective solutions to the following two problems: (i) query scheduling: given
a query workload, decide which queries to assign to each system node; and (ii)
load-balancing: devise mechanisms so that system nodes that have finished
their work can help other system nodes finish theirs.

The challenges in this context are the following. First, to achieve effective
query scheduling, we need to come up with mechanisms for estimating the
execution cost of data series similarity search queries, which do not
currently exist. Second, this observation renders a load balancing scheme
necessary, yet, this also means that we need to replicate data in order to
make such a mechanism viable, as moving big volumes of data series around
would be prohibitively expensive. Data replication works against data
scalability and is more costly in whatever regards index creation time, but
results in better query answering times, thus leading in interesting trade-
offs through which an effective solution should navigate. Third, along with
all the above considerations, we also need to ensure that our solutions will
still maintain their good parallelization properties for efficient execution
in multi-core CPUs inside each system node, and also achieve high pruning
power during query answering.

Our Approach. We propose a novel distributed data-series (DS) indexing and
processing framework, called Odyssey, that efficiently addresses the high
scalability objective by taking advantage of the full computational capacity
of the computing platform.

To come up with an appropriate scheduling scheme for Odyssey, we performed a
query analysis that shows correlation between the total execution time and a
parameter of the category of the single-node data series indexes we consider.
This analysis drove the design of efficient scheduling schemes, by generating
an execution time prediction for each query of the input query batch.

To achieve Load Balancing (LB) even in settings where predictions may not be
accurate, Odyssey provides a LB mechanism, which ensures that nodes sitting
idle can take away (or steal) work from other nodes which have still work to
do (provided that these nodes store similar data). Combining Odyssey scheduler
with this LB technique results in very good performance and high scalability
for all query batches we experiment with.

Ensuring data scalability and, at the same time, good performance for query
answering are contradicting goals. A scheme where data are not replicated
would result in the lowest space overhead, but experiments show that this
technique does not ensure the best performance during query answering, because
no data replication means that Odyssey’s LB mechanism cannot be used.

Odyssey manages to effectively unify these two contradicting goals by
supporting a flexible partial replication scheme. This way, it navigates
through the fundamental trade-off between data scalability and good
performance during query answering. The degree of replication is one of
Odyssey’s parameters. By specifying it appropriately, users can choose the
time-space trade-off that best suits their application and setting.
Experiments show that Odyssey achieves good performance even for small
replication degrees.

Supporting the components for efficient distributed computation that Odyssey
provides, on top of an index that exploits the computation power of a single
node as efficiently as SotA parallel indexes Peng et al. (2020b, 2021a,
2021b), was one more challenging task we undertook while designing Odyssey. A
simple approach of using an instance of the SotA MESSI index Peng et al.
(2021a) in each node did not result in good performance mainly due to two
reasons. First, different data series queries may exhibit variable degrees of
locality (revealed only at runtime), resulting in low pruning in some of the
nodes, and thus, in severe load balancing problems and performance
degradation. Second, supporting load-balancing on top of such a simple
approach would require moving data around, which is often prohibitively
expensive. Odyssey single-node indexing scheme borrows some techniques from
SotA indexes Peng et al. (2018, 2020a, 2020b, 2021a, 2021b), and couples those
with new components and mechanisms, to achieve load balancing and come up with
a scheme in which work from an overloaded node can be given away to idle nodes
without having to pay the prohibited cost of moving any data around.

Odyssey is innovative in different ways. First, it employs a different pattern
of parallelism from all existing approaches in traversing the index tree to
produce the set of data series that cannot be pruned. Second, it presents new
implementations for populating and processing the data structures needed for
efficient query answering. To achieve load balancing among the threads, it is
critical to choose an appropriate threshold on the size of these data
structures, and Odyssey proposes an effective mechanism for predicting a good
threshold. Additionally, Odyssey provides efficient communication and book-
keeping mechanisms, to enable fast exchange of information among nodes to
ensure good pruning degrees in all of them.

Odyssey is up to $6.6$x faster than its competitors and more than $3.5$x
better than its best competitor. Additionally, Odyssey’s index creation
perfectly scales with both the dataset size and the number of node. Moreover,
Odyssey’s best performing scheduling strategy is more than $2.5$x faster than
its initial one.

Contributions. The main contributions of the paper are as follows:

$\bullet$ We describe Odyssey, a scalable framework for distributed data
series similarity search in clusters with multi-core servers. This makes our
approach the first customized data series solution that exploits
parallelization both inside and across system nodes.

$\bullet$ We develop a scheduling algorithm for assigning queries to the nodes
of the cluster, which tries to balance the workload across the nodes by
computing a (good-enough) estimation of the execution time of each query.

$\bullet$ We present a novel exact search algorithm that supports work-
stealing between nodes that share the same index (full replication). Thus, our
approach leads to high performance, even when the work is not (or cannot be)
equally distributed over the nodes of the cluster. We further extend our
solution to work even when only a part of the index is shared among nodes
(partial replication).

$\bullet$ Our approach supports different replication degrees among the nodes,
allowing users to navigate the entire spectrum of solutions, trading space
(replication degree) for speed (query answering time).

$\bullet$ We also present a density-aware data partitioning method that can
efficiently partition data in a way that improves the work balancing
capabilities of our approach.

$\bullet$ Finally, we conduct an experimental evaluation (code and data
available online cod (2022)) with a wide range of configurations, using real
and synthetic datasets. The evaluation demonstrates the efficiency of Odyssey,
which exhibits an almost linear scale-up, and up to 6.6x times faster exact
query answering times than the competitors.

##  2 Preliminaries and Related Work

Data Series. A data series, denoted as $S=\\{p_{1},...,p_{n}\\}$, is a
sequence of points, where each point $p_{i}$ is a pair $(u_{i},t_{i}),1\leq
i\leq n$, of a real value $u_{i}$ and the position $t_{i}$ of $p_{i}$ in the
sequence; $n$ is the size (or dimensionality) of the sequence. When $t_{i}$
represents time, we talk about time series. In several cases, we omit the
$t_{i}$, e.g., when they are equally spaced, or only play the role of an index
for the values $u_{i}$ Echihabi et al. (2018); for simplicity, we omit them,
as well.

iSAX Summary. The iSAX summary Shieh and Keogh (2008) of a data series splits
the x-axis in equal segments and represents each segment with the mean value
of the points of the data series that it contains (see Figure 1). Then it
partitions the y axis into regions of sizes determined by the normal
distribution and represents each region using a number of bits (cardinality).
The number of bits can be different for each region, and this enables the
creation of a hierarchical index tree (iSAX-based index tree Palpanas (2020);
see Figure 1).

(a) Data Series

(b) PAA Summary

(c) iSAX Summary

(d) iSAX Tree

Figure 1: From data series to iSAX index

Similarity Search. Given a collection of data series $\mathcal{C}$ and an
input data series $S$, called the query, similarity search is the task of
finding the data series in $\mathcal{C}$ which are most similar to $S$. We
focus on finding a single best answer, known as the 1-NN problem. We also
focus on Euclidean Distance (ED). The euclidean distance (or real distance)
between two time series $T=\\{t_{1},...,t_{n}\\}$ and
$S=\\{s_{1},...,s_{n}\\}$ is defined as
$ED(T,S)=\sqrt{\sum_{i=1}^{n}(t_{i}-s_{i})^{2}}$. We call the distance between
the iSAX summaries of $T$ and $S$, lower-bound distance. The lower-bound
distance between any two data series is always smaller than or equal to the
real distance between them.

Single-Node Parallel Summary-Based DS Indexing. Such indexes Peng et al.
(2018, 2020a, 2020b, 2021a, 2021b); Echihabi et al. (2022); Azizi et al.
(2023) exploit multiple threads (and SIMD) to create an index tree and answer
queries on top of this tree. They are usually comprised of two main phases,
the index tree construction and the query answering phases. In the index tree
construction phase, they first calculate, in parallel, summarizations of all
data series in the collection. If the summarizations are iSAX summaries, we
talk about iSAX-based DS indexing. To achieve a good degree of locality and
low synchronization overheads, they store these summaries into a set of
summarization buffers. Data series that have similar summarizations are placed
into the same buffer. Subsequently, the data series of each of these buffers
are stored into each of the subtrees of the index tree that they construct.
These design decisions allow them to build the index tree in an almost
embarrassingly parallel way (thus, without incurring synchronization
overheads), and achieve locality in accessing the data during tree
construction. They thus respect crucial principles for achieving good
performance that should be respected when designing a parallel index.

To answer a query, these indexes first calculate the summarization of the
query. Subsequently, they traverse the index tree to find the most appropriate
data series based on the iSAX summary lower bound distances. The distance of
these data series from the query series is stored in a variable called best-
so-far (BSF), and serves as an initial approximate answer to the active query.
Then, BSF is used to prune data series from the initial collection. A data
series $S$ is pruned when the lower bound distance between $S$ and the query
is higher than the current value of the BSF. This process outputs a hopefully
small subset of the initial DS collection, containing series that need to be
further examined. These series are often stored in (one or more) priority
queues Peng et al. (2020b, 2021a, 2021b). Multiple threads process,
concurrently, the elements of the priority queues, calculating real distances
(if needed), and updating the BSF each time a new minimum is met (see Figure
2). Once this process completes, the distance to the answer is contained in
BSF.

Figure 2: Algorithm Outline of Parallel DS Indexes.

Multi-node Systems and Query Processing. The system consists of a number of
asynchronous nodes which communicate by exchanging messages. Each node is a
multi-core machine, capable to support multiple threads (and possibly SIMD
computation). Threads communicate by accessing shared variables. A shared
variable can be atomically read and written. Stronger primitives, such as
Fetch&Add may also be provided. Fetch&Add($V,val$) atomically adds the value
$val$ to the current value of variable $V$ and returns the value that $V$ had
before this update.

An arbitrarily large batch of queries is provided in the system as input. The
goal is to utilize the system’s computational power to execute these queries
in a way that minimizes the makespan, i.e., the length of time that elapses
from the time that any node starts processing a query of the batch to the
first point that all nodes have completed their computation. Our techniques
can easily be adjusted to work with queries that arrive in the system
dynamically.

The data series in the initial collection can be stored in all nodes (full
replication), or may be scattered to the different nodes so that nodes store
disjoint subsets of the data (no replication). A partial replication scheme is
also possible, where nodes store subsets of the data which are not necessarily
pairwise disjoint (e.g., more than one node may store the same subset of data
series). A data partitioning mechanism determines how to split and distribute
the data of the initial data-series collection to nodes.

Query scheduling algorithms aim to schedule the input queries to nodes in a
way that each node has approximately the same amount of work to do.
Considering full replication, a Static Query Scheduler (SQS) partitions the
sequence of queries into $N$ subsequences and each node gets one of these
subsequences to answer. A Dynamic Query Scheduler (DQS) employs a coordinator
node, and has other nodes requesting queries to execute from the coordinator.
The coordinator may serve requests by assigning the next unprocessed query to
a worker when it receives its request, or it may preprocess the sequence of
queries (e.g., by re-arranging the queries based on some property) before it
starts assigning queries to nodes. To avoid loosing computational power, the
coordinator can answer queries itself between serving requests from other
nodes.

###  2.1 Related Work

Data series similarity search queries require the use of specialized index
structures in order to be executed fast on very large collections of data
sequences. In general, data series indexes operate by pruning the search space
based on the summarizations of the series and corresponding lower bounds, and
only use the raw data of the series in order to filter out the false
positives.

Data Series Indexes. Agrawal et al. Agrawal et al. (1993) presented the first
work that argued for the use of a spatial indexing structure for indexing data
sequences, based on the R-Tree Sellis et al. (1987), and was later optimized
Rafiei and Mendelzon (1998). Various indices, specific to data sequences, have
been proposed in the literature Echihabi et al. (2021a). DSTree Wang et al.
(2013) is an index based on the APCA summarization Keogh et al. (2001). The
DSTree can adaptively perform split operations by increasing the detail of
APCA as needed. The iSAX index is based on the SAX summarization, and its
extension, iSAX Shieh and Keogh (2008). In this case, the data series
summarization is bitwise, leading to a concise representation and overall
index. Several other iSAX-based indices have been proposed in the literature
Camerra et al. (2014); Zoumpatianos et al. (2014, 2016); Linardi and Palpanas
(2019, 2020); Palpanas (2020); Wang and Palpanas (2021); Wang et al. (2023b).
These indexes are among the SotA solutions in this area Echihabi et al.
(2018), including MESSI Peng et al. (2020b, 2021a), an in-memory, multi-core
and SIMD-enabled version of the iSAX index.

Data Series Management Systems. Several data series management systems have
been developed in the last few years Jensen et al. (2017); Echihabi et al.
(2021a).Beringei Pelkonen et al. (2015) has a custom in-memory storage engine.
It compresses and organizes data in a series per series scheme. CrateDB Crate
(2018) partitions data in chunks, stores them in a distributed file system,
and indexes them using Apache Lucene. InfluxDB InfluxDB (2018) uses Time-
Structured Merge Trees (LSM tree variant). Prometheus Prometheus (2018) is
based on the Beringei ideas. QuasarDB QuasarDB (2018) utilizes either RocksDB
or Hellium Hellium (2018). Riak TS RiakTS (2018) supports both LevelDB or
Bitcask, which is a custom log structured hash table. Timescale Timescale
(2018) is a Postgres extension. IoTDB Wang et al. (2020) is geared towards
streaming data series. Finally, various systems such as OpenTSDB OpenTSDB
(2015), Timely Timely (2018) (concentrated on security) and Warp10 Warp10
(2018) are developed on top of HBase. All the aforementioned systems support
range scans in the positions, aggregation functions and filtering. InfluxDB
supports queries like moving averages, prediction, transformations, etc, and
Timescale supports gap filling. Nevertheless, none of the above systems
supports exact whole-matching similarity search queries.

Distributed Data Series Indexes. KV-Match Wu et al. (2019) and its
improvement, L-Match Feng et al. (2020), are index structures that can support
similarity search. These indices can be implemented on top of Apache HBase,
and operate in a distributed fashion within Apache Spark. We note that these
solutions only support subsequence similarity search, and not whole-matching
Echihabi et al. (2018), which is the focus of our paper. TARDIS Zhang et al.
(2019) is an Apache Spark system for similarity search. It supports
approximate queries, as well as _exact match_ queries, where we want to know
if the query appears _exactly the same_ within the dataset, or not. This query
type is much easier than the exact queries we consider in our work, and cannot
be efficiently transformed to exact querying. Finally, DPiSAX Yagoubi et al.
(2017, 2020) is a distributed solution for data series similarity search,
developed for Apache Spark using Scala. It was designed for answering batches
of approximate search queries, but also supports exact search. DPiSAX exploits
the iSAX summaries of a small sample of the dataset, in order to distribute
the data to the nodes equally. Then, an iSAX index is built in each node on
the local data, and is used to perform query answering. In order to produce
the exact search results, all nodes need to send their partial results to the
coordinator, which merges them and produces the final, exact answer. Note that
DPiSAX was not explicitly designed for intra-node parallelization, but is the
only distributed data series index in the literature that supports exact
search.

Work-stealing was employed in the Cilk framework Cil (1996). The work-stealing
approach was formally studied and analyzed in Blumofe and Leiserson (1999);
Fatourou and Spirakis (1999, 2000). Lots of work has been done on this topic
(e.g., Blelloch et al. (1999, 1997); Fatourou (2001)).

##  3 The Odyssey Framework

Figure 3: Odyssey flowchart.

We start with a high level overview of the Odyssey flowchart, which comprises
of five stages (see Figure 3).

In the first stage, a coordinator node partitions the raw data-series
collection to as many chunks as the number of system nodes, and assigns a
chunk to each node (including itself). (Section 3.4 details Odyssey’s
partitioning schemes.) In the second stage, each node (i) loads its chunk of
data in memory, (ii) computes their iSAX summaries and stores them into a
number of summarization buffers, for achieving locality, and (iii) builds its
index tree. To enhance performance at query answering, Odyssey employs data
replication. It forms groups of nodes (replication groups, described in
Section 3.3), where all nodes of each group store the same chunk of data. Each
replication group has a coordinator node, called group coordinator, which
schedules queries to the group’s nodes. A batch of queries (e.g., originating
from a _k-NN_ classification task) to execute is submitted to all group
coordinators (as different groups store different data chunks). In the third
stage, the group coordinators start by estimating the execution time of each
query, then sort queries in descending order of estimated execution times, and
dynamically schedule them to the group’s nodes (Section 3.1 describes query
scheduling). In the fourth stage, each node processes the queries assigned to
it. It first calculates an initial BSF, and then prunes the index tree using
this BSF, populating the priority queues with leaves that cannot be pruned.
Finally, it processes the elements of the priority queues to find the best
local answer (corresponding to its data chunk). In this stage, Odyssey
supports BSF-sharing and work-stealing (detailed in Section 3.2). In the last
stage, the coordinator node collects the local answers from the group
coordinators, and produces the final answers.

###  3.1 Query Scheduling

To correctly answer a query, it should be forwarded to at least one set of
system nodes that collectively store all the data. We call such sets node
_clusters_ in Section 3.3. Thus, in the no-replication case this set contains
all system nodes, so a scheduling algorithm should forward all queries to all
nodes. Other replication settings (and especially full replication) are more
interesting, as they enable the utilization of different scheduling
techniques.

To come up with Odyssey scheduler, we experimented with a collection of
scheduling techniques, including the simple static and dynamic schemes (SQS
and DQS) for full replication settings, discussed in Section 2. Unfortunately,
these schemes suffer from severe load imbalance problems for many categories
of query batches. For the static case, consider for example, a query sequence
which consists of progressively more difficult queries (i.e., of queries that
each requires less time to run than the next one). SQS will assign to the
first system nodes easy queries, while the last nodes will get more work to
do. The dynamic method (DQS) may also result in load imbalances: even in
simple cases where e.g., a query batch includes a single difficult query at
the end, most nodes may be sitting idle, while a single node is running the
difficult query. This may significantly degrade performance.

Some of these load imbalances could be avoided, if we knew the execution time
of each query. Recent work Gogolou et al. (2020); Echihabi et al. (2023)
illustrated that there exists a correlation between the initial BSF and the
number of vertices visited in a single-node index tree. We performed a
corresponding query analysis which showed that similarity search queries, for
which the initial BSF is high, tend to also have high execution times. In this
work, we use a linear regression model (other pediction schemes can be used,
as well) to produce estimates for each query. An example of this outcome is
shown in Figure 4 (for Seismic; we follow the same process for the other
datasets).

These observations led us to design two scheduling algorithms. The first,
static prediction-based scheduling, statically allocates the queries to nodes
based on their estimations. Each node maintains a $\mathit{load}$ variable,
which stores the sum of the estimations of the queries that are assigned to
it. The algorithm uses a greedy approach to assign queries to nodes so that
load balancing is achieved. There are two variations of the algorithm: the
first (unsorted) schedules the queries using their order in the sequence, and
the second (sorted) sorts the sequence based on decreasing execution time
estimations. . The second scheduling algorithm, called dynamic prediction-
based scheduling, is an enhanced version of DQS, where queries are assigned to
nodes after sorting the entire query batch, based on estimations (in
decreasing order).

Figure 4: Linear regression for Seismic queries prediction.

Consider a system of two nodes, $\mathit{sn_{1}}$ and $\mathit{sn_{2}}$, and
let $\mathcal{Q}=\\{q_{1},q_{2},q_{3},q_{4},q_{5}\\}$ be a query batch to
execute. Assume that $\mathcal{ES}=\\{100,50,200,250,80\\}$ is the set of the
estimated execution times, where the $i$-th element of $\mathcal{ES}$ is the
estimated execution time for $q_{i}$, $1\leq i\leq 5$. Unsorted static
prediction-based scheduling, with load variables $l_{1}$ and $l_{2}$ (for
$\mathit{sn_{1}}$ and $\mathit{sn_{2}}$, respectively), proceeds as follows:
$q_{1}$ is assigned to $\mathit{sn_{1}}$ (so, $l_{1}=100$), and $q_{2}$ is
assigned to $\mathit{sn_{2}}$ (so, $l_{2}=50$). Since $l_{2}<l_{1}$, $q_{3}$
is assigned to $sn_{2}$ (thus, $l_{2}=250$). Following a similar strategy,
$q_{4}$ is assigned to $\mathit{sn_{1}}$, and $q_{5}$ is assigned to
$\mathit{sn_{2}}$. So, $sn_{1}$ receives $\\{q_{1},q_{4}\\}$ and $sn_{2}$
receives $\\{q_{2},q_{3},q_{5}\\}$. In sorted static prediction-based
scheduling, the queries of $\mathcal{Q}$ are first sorted in decreasing order
of their estimated times, resulting in
$Q^{\prime}=\\{q_{4},q_{3},q_{1},q_{5},q_{2}\\}$ (which corresponds to
$\mathcal{ES}^{\prime}=\\{250,200,100,80,50\\}$). After applying the static
prediction-based scheduling algorithm (as above) on these sets,
$\\{q_{4},q_{5}\\}$ is assigned to $sn_{1}$ and $\\{q_{3},q_{1},q_{2}\\}$ is
assigned to $sn_{2}$. Finally, dynamic prediction-based scheduling also sorts
the queries of $\mathcal{Q}$. In this case, $q_{4}$ is assigned to
$\mathit{sn_{1}}$, $q_{3}$ to $\mathit{sn_{2}}$, while the rest of the queries
are dynamically assigned to nodes (in order) upon request (thus, based on
actual execution times).

The Odyssey framework supports all of the above scheduling algorithms. The
Odyssey index utilizes dynamic prediction-based scheduling, which turned out
to be the best approach in most cases.

###  3.2 Load Balancing

Odyssey provides a load balancing (LB) mechanism, which can be applied on top
of any of the scheduling schemes described in Section 3.1. Specifically, idle
nodes can steal work from other nodes which still have work to do (provided
that they store similar data).

This is necessary as predictions may not always be accurate, or the query
batch may be produced dynamically at run time, in which case sorting of the
entire query batch is not possible. It is also necessary for achieving high
scalability. As the number of utilized nodes increases, the number of batch
queries that each node has to process becomes smaller and smaller. Thus,
problematic scenarios as those described in Section 3.1, may appear, where
just one or a few nodes work on difficult queries, while others are sitting
idle.

Overview of our approach. We performed a number of experiments to get a break-
down of the query answering time. This break-down illustrated that the biggest
part of the time for query answering goes to priority queues’ processing. We
thus focus on designing a method that allows nodes to steal work during the
execution of that phase. For simplicity, we first focus on the full-
replication case, where the initial collection of data is available in every
node; partial replication is then discussed in Section 3.3.

A simple work-stealing scheme Blumofe and Leiserson (1999); Cil (1996) would
not work, mainly because moving data (stored in priority queues) around from
one node to another is expensive and should be avoided. Thus, the main
challenge in our setting is to take work away from one node and assign it to
another without ever moving any data around.

Odyssey’s load-balancing mechanism works as follows. An idle system node $sn$
randomly chooses another node $sn^{\prime}$ and sends it a steal request. If
$sn^{\prime}$ has still work to do, it chooses a number of priority queues to
give away to $sn$. To avoid paying the cost of transferring data around,
Odyssey employs a technique that informs $sn$ on how to locally build the
priority queues to work on, based on its own index. Node $sn$ traverses the
identified part of its index tree and re-constructs these priority queues. As
the time to create the priority queues is relatively small in comparison to
that for processing them, this scheme works quite well.

Note that the approaches followed by existing SotA indexes Peng et al. (2020b,
2021a, 2021b) for creating and processing the priority queues are too naive to
support work-stealing without moving any data around. In Odyssey, we propose
(in Section 3.2.1) a new implementation of a single-node, multi-threaded
index, which respects the good design principles described for parallel
indexes in Section 2, while it simultaneously copes with the problem mentioned
above.

####  3.2.1 Single-Node Query Answering

Consider any system node $\mathit{sn}$ and assume that an iSAX-based index
tree has been created and an initial value for the BSF has been computed in
$sn$. An outline of the single-node query answering algorithm of Odyssey is
depicted in Figure 5. The pseudocode is provided in Algorithms 1 and 2.

Figure 5: Outline of the Odyssey single-node query-answering process.

Description. Node $sn$ executes each of the queries in the query batch
assigned to it one by one (Algorithm 1). For each such query $Q$, it creates a
number of search workers to execute it (line 1). As soon as, all queries in
$sn$’s query batch have been processed, $sn$ informs other nodes that it has
completed (line 1). Then, it tries to help other active nodes by executing
PerformWorkStealing (line 1). Each node allocates a thread to play the role of
the work-stealing manager (line 1). This thread simply processes all work-
stealing requests that the node will receive (Algorithm 3). (Work-stealing is
discussed in Section 3.2.2.)

The query answering algorithm in $\mathit{sn}$ splits the tree into root
subtree (RS) batches, i.e., sets of consecutive root subtrees (see Figure 5),
and allocates a number of threads to work on them. Each thread begins by
getting an RS-batch to work on using Fetch&Add (Algorithm 2). Then, the thread
executes the $\mathit{ProcessBatch}$ routine, which traverses the tree
recursively and inserts the leaves that cannot be pruned into one of a set of
priority queues that belong to the RS-batch.

For every RS-batch, there exists one active priority queue at each point in
time. When the size of this priority queue surpasses a threshold, this queue
is abandoned and another one is initialized for the RS-batch.

As soon as an idle thread $\mathit{th}$ discovers that all RS-batches have
been assigned for processing, it tries to help some other still active thread,
$\mathit{th^{\prime}}$, to complete processing its assigned RS-batch (lines
2-2, Algorithm 2). To reduce the synchronization cost, there is a threshold,
$\mathit{HelpTH}$, on the number of threads that help on each RS-batch (line
2). This phase ends when the subtrees of all RS-batches have been traversed
and all priority queues have been populated. Experiments showed that we get
the best performance when the number of RS-batches, $N_{sb}$, equals the
number of worker threads.

As soon as this tree traversal phase is over, we have a set of priority queues
for each RS-batch, stored in an array. For performance reasons, this array is
sorted by the priority of the top element of each priority queue. This
comprises the priority queue preprocessing phase (lines 2-2). This way, the
algorithm processes the priority queues with the smallest lower bound
distances to the query first. These queues contain data series that are more
probable to be in closer real distance to the query, thus enabling further
pruning.

Then, the priority queue processing phase starts (lines 2-2). Every thread
gets a priority queue from the PQueues array to process (using Fetch&Add).
Routine $\mathit{ProcessPriorityQueue}$ processes those data series stored in
the priority queue, which cannot be pruned. Whenever a lower real time
distance between any of these series and the query series is calculated, the
BSF is updated to contain this distance. This improved BSF is submitted to all
nodes of the system. Finally, all answers are transmitted to the coordinator
node, and the globally smallest value of the BSF is the response to the query.

1 Shared Variables: Shared PointerToArray $PQueues=NULL$

2

Input: QuerySeriesBatch $QBatch$, Index $Index$, Integer $NThreads$

3

4Array $BSFArray[]$ $\triangleright$ with size $|QBatch|$

5

6for  _every query series id $Q$ in $\mathit{QBatch}$ _ do

7

8 $iSAX_{Q}$ = calculate iSAX summary for $\mathit{Q}$

9 $BSF$ = approxSearch($iSAX_{Q}$, $Index$)

10 create a thread to execute an instance of WorkStealingManager(Q)

11 for  _$i$ $\leftarrow$ $0$ _to_ $NThreads-1$_ do

12 create a thread to execute an instance of SearchWorker($Q$, $Index$,
$N_{sb}$, $i$, $PQueues$ )

13

14 Wait for all threads to finish

15 $\mathit{FinishFlag[Q]}$ := TRUE

16 BSFArray[Q] := $BSF$;

17send(DONE, $sn$) to all nodes

18 PerformWorkStealing()

19 return ($BSFArray$)

Algorithm 1 Odyssey Single-Node Query Answering - Code for node $sn$

1

2$\triangleright$ Shared Variables

3 Shared Integers $BCnt$ = 0, $PQCnt$ = 0, $TotPQ$ = 0;

4

Input: QuerySeries $Q$, Index $Index$, Integer $N_{sb}$, Integer $tid$, Queue
$PQueues[]$

5

6Integer $bindex$, $pqindex$;

7

8$\triangleright$ Tree Traversal Phase

9while  _$(TRUE)$_ do

10 $bindex$ $\leftarrow$ Fetch&Add($BCnt$,1);

11 if  _$bindex\geq N_{sb}$_ then

12 break;

13 $ProcessRSBatch(Q,bindex,Index.RSBatches)$;

14 $Index.RSBatches[bindex].complete\leftarrow TRUE$;

15for  _$bindex\leftarrow 0$ to $N_{sb}$_ do

16 if  _$!Index.RSBatches[bindex].complete$ AND
Fetch&Add($Index.RSBatches[bindex].helped,1)<HelpTH$_ then

17 $ProcessBatch(Q,bindex,Index.RSBatches)$;

18 $Index.RSBatches[bindex].complete\leftarrow TRUE$;

19

20Barrier for all threads;

21$\triangleright$ Priority Queue Preprocessing Phase

22if  _$tid==0$_ then

23 Traverse all RS-batches and put their priority queues into $PQueues[]$;

24 $SortByRootPriority(PQueues)$;

25 $TotPQ\leftarrow$ number of valid elements of $PQueues$;

26Barrier for all threads;

27$\triangleright$ Priority Queue Processing Phase

28while  _$(TRUE)$_ do

29 $pqindex$ $\leftarrow$ Fetch&Add($PQCnt,1)$;

30 if  _$pqindex\geq TotPQ$_ then

31 break;

32 if  _$PQueues[pqindex].stolen$_ then

33 continue;

34 $ProcessPriorityQueue(PQueues[pqindex])$;

35

Algorithm 2 SearchWorker - Code for thread $tid$

Input: Integer $N_{B}$

1 Upon Receiving a message of type StealingRequest from node $sn^{\prime}$:

2 S := Set of at most $N_{send}$ ids of RS-batches that satisfy the Take-Away
Property

3 send($S$, $Q$ of $sn$, $Q$’s current BSF) to $sn^{\prime}$

4 Mark the priority queues of the RS-batches with ids in $S$ as stolen

5

6 $\triangleright$ Always-enabled event: it is executed repeatedly

7 Upon receiving no message:

8 if  _$\mathit{FinishFlag[Q]}$ in $sn$ is set_ then

9 Terminate

10

Algorithm 3 WorkstealingManager - Code for node $sn$

Input: Index $index$, Function $exact\\_search\\_workstealing\\_func$,
QuerySeries $queries[]$, Integer $total\\_nodes\\_per\\_nodegroup$

1 Upon Receiving a DONE message from node $sn^{\prime}$:

2 add $sn^{\prime}$ in set $DoneNds$

3 if  _$DoneNds$ contains all system’s nodes_ then

4 Terminate

5

6 Upon Receiving a $msg=\langle S,Q_{s},BSF_{s}\rangle$ from node
$sn^{\prime}$:

7 if  _|S| > 0_ then

8 Create threads to traverse the RS-batches with ids in $S$

9 Populate and process the corresponding priorities queues

10 BSFArray[$Q_{s}$] := $BSF_{s}$; $\triangleright$ computed by threads above

11 Wait all threads to complete

12

13ResponseFlag := 0

14 $\triangleright$ Always-enabled event: it is executed repeatedly

15 Upon receiving no message:

16 if  _!(ResponseFlag)_ then

17 $sn^{\prime}$ := choose randomly a node not in $DoneNds$

18 send(StealingRequest, $sn$) to $sn^{\prime}$

19 ResponseFlag := 1

20

Algorithm 4 PerformWorkStealing - Code for node $sn$

Size of Priority Queues. The size of each priority queue cannot be larger than
a specific threshold, $\mathit{TH}$. If adding an element in a priority queue
results the size of the queue to reach $\mathit{TH}$, then the thread gives up
this priority queue and initiates a new one for the RS-batch. This way, each
priority queue does not contain leaves from more than one RS-batch, and
contains at most $\mathit{TH}$ leaves from the tree part that corresponds to
the RS-batch.

(a) Sigmoid function fitting for determining $\mathit{TH}$.

(b) Performance for different Threshold division factors.

Figure 6: Odyssey Single-Node Query-Answering Algorithm Configuration.

Choosing the appropriate value for the threshold, $\mathit{TH}$, is important
for achieving load balancing among the different threads. Our goal is to
develop a method for determining a threshold value which will result with a
set of priority queues that have about the same size. The threshold is
determined and configured for every dataset we use, based on the queries we
run. We explain the process of determining $\mathit{TH}$ for the Seismic real
dataset iri (2016), but the process is similar for all other datasets (real or
synthetic) we experimented with. After running multiple queries of varying
difficulty, we figured out that there exists again a correlation between the
initial BSF that is computed for the query and the median size of the priority
queues produced for answering it. Then we performed a sigmoid function fitting
using the following parameterized formula:

$f(Z)=m+(M-m)\frac{1}{1+b\cdot exp(-c(Z-d))}$

where $M\in[0,1],m\leq M,b,c\in\mathbf{R}^{*},$ and $d\in\mathbf{R}$ are the
parameters of the sigmoid function (Figure 6(a)). The final threshold value
for each query is the median value estimation as it comes from the sigmoid
function, divided by a factor (e.g. for seismic this factor has to be 16,
based on the diagram shown in Figure 6(b)).

Experiments show that after the tree traversal phase is completed, we end up
with a set of RS-batches that have a number of priority queues with most of
them being the same size. This results in load balancing among the threads
when processing priority queues.

####  3.2.2 Work-Stealing Algorithm

If a system node $\mathit{sn}$ becomes idle, $\mathit{sn}$ initiates the work-
stealing protocol (Algorithm 4, lines 4-4). It randomly chooses a system node
$\mathit{sn^{\prime}}$ from the set of those nodes that $sn$ knows to be still
active and sends a steal request to it111The codes for Algorithms 3 and 4 are
written in an event-driven style Attiya and Welch (2004); Lynch (1996). A
thread in each node acts as the work-stealing manager (Algorithm 3). As soon
as the work-stealing manager of $\mathit{sn^{\prime}}$ receives the request,
it tries to give away work to $\mathit{sn}$ (lines 3-3 of Algorithm 3).

Earlier work has demonstrated that a large amount of the query answering
execution time is devoted to verifying that there is no better answer after
the correct answer has been processed Gogolou et al. (2019, 2020); Echihabi et
al. (2023). Based on these findings, Odyssey’s work-stealing mechanism chooses
to give away an RS-batch $B$ which satisfies the Take-Away Property, namely
that $B$ is not yet stolen and its first priority queue is located in the
rightmost possible index of the $\mathit{PQueue}$ array. This priority queue
is then marked as stolen. If more than one batches are to be given away, this
process is applied repeatedly to choose additional RS-batches. Recall that the
$PQueue$ array is sorted by the priority of the top element of each priority
queue. Thus, by giving away batches in this way, $sn^{\prime}$ assigns to
helpers priority queues that may still contain work. Additionally, it gives
away RS-batches that have the highest probability to be unprocessed.
Throughout the process, the current $BSF$ is shared among the nodes, every
time it is updated, as a helper may steal a priority queue that contains a
better answer (or the owner may compute a better BSF later).

The number, $N_{send}$, of RS-bathes that a node gives-away during stealing
affects performance. Theoretically, we would like to give away a number of RS-
batches which on the one hand, it will enable the stealing node to do a
noticeable amount of work, but on the other, the work to be given away should
not result in higher query answering times. Experiments show that fixing
$N_{send}$ to 4 was the best choice (so $N_{send}=4$ in Odyssey).

###  3.3 Data Replication

Odyssey aims at ensuring data scalability and, at the same time, good
performance for query answering. Optimal data scalability requires to follow a
no replication approach, but experiments show that the best query answering
performance is noticed for fully replicated settings. Odyssey manages to
effectively navigate through this trade-off between data scalability and good
performance during query answering, by providing a flexible partial
replication scheme.

The idea is to split the set of system nodes into clusters, where each cluster
collectively stores the entire dataset (see Figure 7). Each cluster node
stores (and indexes) a chunk of the dataset. The chunks stored in each node of
a cluster are mutually disjoint. A replication group is a group of nodes such
that each node stores the same dataset as every other node in the group. (We
experimented with replication groups of the same size, but Odyssey can operate
with replication groups of different sizes, as well.) The nodes of a
replication group build their iSAX indices from the same data chunk. Thus,
inside every replication group, we can apply the scheduling and load-balancing
schemes described in Sections 3.1 and 3.2, respectively. We call the number of
clusters the replication degree of the system.

Consider a system with $N_{sn}$ system nodes. We call partial-$k$,
$k\in\\{1,2,4,\ldots,N_{sn}\\}$, a replication setting with $k$ replication
groups and $N_{sn}/k$ clusters. Observe that partial-$N_{sn}$, or equally-
split corresponds to no replication (each node stores a disjoint chunk of the
dataset), and partial-$1$, or full corresponds to full replication (each node
stores the full dataset). Note that Odyssey’s data replication scheme supports
$1+\log N_{sn}$ different replication degrees. Smaller replication degrees
lead to smaller space overheads (and thus better data scalability). Thus,
Odyssey’s data replication scheme allows us to tackle memory limitation
problems. Moreover, more replication groups lead to scalability in index
creation.

A system with $8$ nodes supports $1+\log 8=4$ different replication degrees:
full (partial-$1$), partial-$2$, partial-$4$, and equally-split (partial-$8$).
Figure 7 illustrates the case of partial-$4$: we have 4 replication groups,
organized in 2 clusters; replication degree is 2.

Figure 7: Data replication partial-$4$ ($N_{sn}=8$, data size $80$GB).

###  3.4 Data Partitioning

Odyssey framework supports more than one partitioning schemes. Under equally-
split, each system node is assigned a discrete chunk and builds the
corresponding index, resulting in a scheme where each node keeps a local index
on its own part of the data. Queries are forwarded to all nodes. Each node
produces an answer based on its local index and data. The minimum among them
is the final answer. Before distributing the data, random shuffling (RS) can
be applied to randomly rearrange the series of the initial collection.

To answer a query batch using partial data replication (or no replication),
each query is sent to every replication group. Each node answers queries using
its local data, and the partial answers for each query are gathered in the end
to find the smallest answer. Very often for real data, the close answers to a
query could be located into a small part of the dataset. The group that has
these data will get a good initial answer, it will prune more and it will
answer each query really fast, while other groups, will not necessarily
compute good initial BSF values. Thus, they will have more work to do leading
to imbalances. For this reason, we enhance our distributed index with a book-
keeping method that supports BSF sharing. When a node is processing a query
and finds an improved value for BSF, it shares this value through a common
BSF-Sharing channel (as illustrated in Figure 7). Every node periodically
checks this channel to see if an answer for a query has arrived. Because this
process runs in parallel, a node may receive a better answer for a query that
will be encountered later on. Odyssey’s book-keeping method solves such
synchronization problems. Each node holds an array that stores the
improvements received from the channel for the BSF of each query, and before
answering a query it checks the data held in this array. Thus, each node has
the best answers extracted from all nodes, and our experimental evaluation
shows that the use of this method is critical for performance.

In addition to these simple techniques, Odyssey also provides a sophisticated
data partitioning scheme, based on preprocessing of the initial data series
collection, which provides a density-aware distribution of the data among the
available nodes. The required preprocessing incurs some time overhead.
However, it occurs only once for answering as many queries as needed, and
thus, as the number of queries to process increases, this overhead is
amortized. We describe this scheme in Section 3.4.1.

####  3.4.1 density-aware Data Partitioning

We observe that a good partitioning strategy should not assign all similar
series to the same system node. In such a case, we risk to create work
imbalance for the following reason. Assume that we need to answer a similarity
search query, for which all candidate series from the dataset that are similar
to the query are stored in one of the system nodes, while all other nodes are
storing series that are not similar to the query. Then, during query
answering, the node with the similar series will need to perform many (lower
bound and real distance) computations in order to determine which of the
candidate series is the nearest neighbor to the query, with essentially little
pruning (if at all). On the other hand, all the other nodes that store
dissimilar series will be able to prune aggressively, and therefore, finish
their part of the computations much faster.

The above observations led us to the design of the density-aware partitioning
strategy, whose goal is to partition similar series across all system nodes,
without incurring a high computational cost. This is achieved by exploiting
Gray Code Gardner (1986) ordering for effectiveness (since it helps us split
the similar series), and the summarization buffers of our index for efficiency
(since we have to operate at the level of buffers, rather than individual
series).

Figure 8 shows an example of partitioning the data series in the summarization
buffers according to a simple strategy using binary code, and to a strategy
based on Gray Code. In the former case, the buffers that end up in the same
node contain similar series: their iSAX representations (the iSAX word of the
buffer) are very close to one another, e.g., node 1 stores buffers "000" and
"100", so series whose iSAX summaries only differ in one bit. In the latter
case, this problem is addressed. The Gray Code ordering places similar buffers
close to one another (by definition, two neighboring buffers in this order
differ in only one bit), so it is then easy to assign them to different system
nodes in a round-robin fashion.

Figure 8: Examples of partitioning the iSAX buffers’ data to 4 system nodes,
based on (a) simple iSAX and (b) Gray Code.

We depict the flowchart of the density-aware partitioning strategy in Figure
9. We start by computing the iSAX summaries of the data series collection, and
assigning each summary to the corresponding summarization buffer. These
buffers are ordered according to Gray Code, and then the actual data
partitioning starts (using round-robin scheduling). We first partition the
series inside the $\lambda$ largest buffers; this is necessary, since often
times a small number of buffers will contain an unusually large number of
series (that we do not want to assign them all to the same system node). Then,
we partition the remaining buffers, and we check if the partitioning is
balanced. If it is not, then we select the largest buffer of the largest node,
and we partition the series inside this buffer. Our experiments with several
real datasets (omitted for brevity) showed that density-aware exhibits a very
stable behavior as we vary $\lambda$ from a few hundred to several thousands.
In this study, we use $\lambda=400$.

Figure 9: Flowchart of the density-aware data partitioning.

##  4 Extensions

We now discuss two extensions of Odyssey, in order to support _k-NN_ search
and the Dynamic Time Warping (DTW) distance.

_k-NN_ Search. Extending Odyssey to support _k-NN_ similarity search is
straight-forward. Instead of computing a single BSF value, we simply need to
keep track of the $k$ smallest BSF values.
DTW Distance. We also extend Odyssey to perform similarity search using
Dynamic Time Warping (DTW), which is an elastic distance measure Keogh and
Ratanamahatana (2005). Note that no changes are required in the index
structure for this: the index we build can answer both Euclidean and DTW
similarity search queries. Supporting DTW queries requires modifying the query
answering algorithm only, and using LB_Keogh Keogh and Ratanamahatana (2005),
which is a tight lower bound of the DTW distance. We note that a lower bound
for the DTW distance between the query and a candidate series can be computed
by considering the distances between the corresponding points of the candidate
series and the points of the LB_Keogh envelope of the query.

##  5 Experimental Evaluation

Setup. Experiments conducted on a cluster of 16 SR645 nodes, connected through
an HDR 100 Infiniband network. Each node has 128 cores (with no hyper-
threading), 200GB RAM (available to users out of the 256GB physical memory),
and runs Red Hat Enterprise Linux r8.2. All evaluated algorithms written in C
and compiled using MPICC, Intel(R) MPI Library for Linux OS, v2021.2.

Algorithms. Our experimental analysis includes the entire range of Odyssey’s
data distribution strategies with $k$ replication groups, partial-$k$,
$k\in\\{1,2,4,\ldots,N_{sn}\\}$, as well as the density-aware data
partitioning algorithm (density-aware). Recall that partial-$N_{sn}$, or
equally-split corresponds to no replication, and partial-$1$, or full
corresponds to full replication. Additionally, our analysis evaluates
Odyssey’s queries scheduling algorithms: (i) static scheduling assigning
equally sized query sets to nodes (static); (ii) dynamic scheduling using a
coordinator (dynamic); and (iii) predictions-based scheduling, including:
static without ordering (predict-st-unsorted), static with ordering (predict-
st), and dynamic (predict-dn). Moreover, we evaluate Odyssey’s work-stealing
mechanism using both dynamic and predict-dn, resulting in algorithms work-
steal and work-steal-predict, respectively. The latter is our best scheduling
algorithm (cf. paragraph “Queries scheduling”). We note that that Odyssey’s
query scheduling and work-stealing mechanisms can be used together only with
the full or partial data distribution strategies that provide some
replication.

We compare Odyssey to: (i) MESSI Peng et al. (2021a), where we run the MESSI
index independently in each system node; (ii) MESSI SW BSF, where we extend
the previous solution by enabling system-wide sharing of the BSF values; and
(iii) DPiSAX Yagoubi et al. (2020), where we implement (in C) the DPiSAX data
partitioning strategy, and (for fair comparison) implement query answering in
each node using MESSI.

Datasets. We evaluated Odyssey’s strategies and algorithms using real and
synthetic datasets, of varying sizes (refer to Table 1). The synthetic data
series, called _Random_ , were generated as random-walks (i.e., cumulative
sums) of steps that follow a Gaussian distribution (0,1). This type of data
has been extensively used in the past Faloutsos et al. (1994); Camerra et al.
(2014); Zoumpatianos et al. (2015, 2018); Echihabi et al. (2018, 2019), and
models the distribution of stock market prices Faloutsos et al. (1994). Our
five real datasets come from the domains of seismology (_Seismic_), astronomy
(_Astro_), deep learning (_Deep_), image processing (_Sift_), and information
retrieval (_Yan-TtI_). _Seismic_ contains seismic instrument recordings and
consists of 100M data series of size 256 for Seismology with Artificial
Intelligence (2018). _Astro_ represents celestial objects and consists of 100M
data series of size 256 Soldi et al. (2014). _Deep_ Vision (2018) contains 1B
Deep vectors of size 96 extracted from the last layers of a convolutional
neural network. _Sift_ Jégou et al. (2011) is comprised of image descriptors
and Yandex Text-to-Image (_Yan-TtI_) Simhadri et al. (2022) contains 1B
vectors that include image- and textual-embeddings in the same space; it
represents typical cross-modal information retrieval tasks.

Table 1: Details of datasets used in experiments.

Dataset # of series Length (floats) Size (GB) Description Seismic 100M 256 100
seismic records Astro 270M 256 265 astronomical data Deep 1B 96 358 deep
embeddings Sift 1B 128 477 image descriptors Yan-TtI 1B 200 800 image and text
Random 100M-1600M 256 100-1600 random walks

Evaluation Measures. During each experiment, $E$, and for each node,
$\mathit{sn}$, we measure (i) the buffer time required to calculate the iSAX
summaries and fill-in the receive buffers, (ii) the tree time required to
insert the items of the receive buffers in the index tree, and (iii) the query
answering time required to answer the queries assigned to $\mathit{sn}$. The
sum of these times constitute the total time that $\mathit{sn}$ works during
$E$; also, buffer and tree times constitute the time required to create the
index, called index time. To compute all the above times during $E$, we take
the maximum among the corresponding times of each node participating in $E$.
We report the average times of $10$ experiments.

Query scheduling. To compare Odyssey’s queries scheduling algorithms, the full
replication strategy is selected, to avoid measuring any overheads resulting
from the partial replicated strategies. Recall that scheduling algorithms
can’t be used together with the no replication strategies. We experimented
with both Random (synthetic dataset) and Seismic (real dataset), and all of
our algorithms positively affected performance in comparison with static.
Moreover, for the synthetic dataset, we have seen no remarkable differences
between all our scheduling algorithms, since the randomness when producing the
data series of both the dataset and the queries set, results in queries with
almost the same effort to be answered. We present the results for Seismic,
where the effort for answering queries varies. Specifically, Figure 10 shows
that as the number of nodes increases, predict-dn is the best scheduling
policy in all cases and it is up to $150\%$ better than static.

(a) full replication

(b) partial-$2$ replication

Figure 10: Odyssey’s scheduling algorithms (Seismic).

Work-stealing. Figure 10(a) shows that work-steal-predict greatly outperforms
(up to almost 2x) predict-dn for large number of nodes when using full
replication, i.e. our work-stealing technique positively affects performance
on these cases. The same is true for partial-$2$ replication, but to a lesser
extent. Recall (from Section 3.2.2) that this happens since all the algorithms
that do not use the work-stealing technique suffer from load-imbalance issues.
Specifically, when a query set contains a few (significantly less than the
number of nodes) queries that require significantly more effort to get
answered (than the majority of queries), then as the number of nodes increases
more nodes remain idle at the end of the corresponding query answering phase,
since no such difficult query is assigned to them.

Query Scalability. To evaluate the scalability of Odyssey’s algorithms with
increasing number of queries, we conducted experiments with work-steal using
synthetic and real datasets. In Figure 11(a), we present the results for the
Random dataset (results with the other tested datasets are similar) with full
replication, for a total of $100$, $200$, $400$, and $800$ queries. As we can
see, work-steal scales almost perfectly with the increasing number of queries,
since the time to execute $100$ queries in $1$ node is the same with the time
to execute $j*100$ queries in $j$ nodes, $j\in\\{2,4,8\\}$. We have observed
the same trend for the partial scheduling algorithms (Figure 11(b)). Note that
partial replication can be applied only with two or more nodes. Additionally,
we present in Figure 12 scalability experiments, by increasing the dataset
size, for Random (between 100-1600GB) and Yan-TtI (between 100-800GB). We
measure the total query answering time for 100 queries, when using 8 nodes.
Note that we could not execute all replication strategies for all dataset
sizes, due to the memory capacity of our nodes. The results show that query
answering time scales gracefully as we increase the dataset size, while
increasing the replication degree leads to better performance. Moreover, we
observe that Odyssey’s query answering algorithm achieves good scalability as
the number of nodes increases. This is better illustrated in Figure 14, which
presents the work-steal throughput on the Random dataset.

(a) full replication

(b) partial-$2$ replication

Figure 11: Query answering scalability as the number of queries increase
(Random).

(a) Random

(b) Yan-TtI

Figure 12: Query time for 100 queries vs data size (8 nodes).

Figure 13: Query throughput (Random, full replication).

Figure 14: Index size.

Replication. We study now Odyssey’s different replication strategies using the
Seismic dataset and work-steal-predict that is our best scheduling algorithm,
to avoid any overhead incurred by load-imbalances between nodes. Specifically,
we test equally-split, partial-$4$, partial-$2$ and full, for varying number
of queries. Figures 15(a)-15(b) present the query answering time222We report
results with 16 nodes only for the small workload, because the scheduler of
our cluster does not allow long-running jobs on more than 8 nodes., where we
observe that the more a dataset is replicated, the less time is required to
answer queries, and this is consistent for all number of queries. So, the full
replication strategy has the smaller queries answering time. On the other
hand, Figures 15(c)-15(d) present the total execution time, which includes
also the time for index tree construction. Interestingly, for small query
numbers ($100$), we observe exactly the opposite: a larger amount of data
replication, results in bigger total time, with full having now the bigger
index tree construction time. This happens because the increased index tree
construction time dominates in the total time. However, as the number of
queries increases, the differences between the total execution time of
algorithms become smaller. Remarkably, for large enough number of queries
(e.g., $800$), the increased index tree construction cost is amortized by the
smaller query answering time, having full replication strategy performing
better than equally-split. This analysis reveals an interesting trade-off
(regarding the level of replication) between the query answering cost and the
index tree construction cost, while the latter can be amortized using a large
enough set of queries. Figure 16 shows the results of the query answering
experiment with 100 queries for the rest of the real datasets. We observe
similar trends to those of Seismic (Figure 15(a)). Overall, when query
answering needs to be optimized, we recommend that Odyssey is used with the
highest possible replication degree (given the dataset size and compute-
cluster characteristics).

(a) 100 Queries

(b) 800 Queries

(c) 100 Queries

(d) 800 Queries

Figure 15: Comparison of Odyssey’s replication strategies, using work-steal-
predict with Seismic.

(a) Astro

(b) Deep

(c) Sift

(d) Yan-TtI

Figure 16: Comparison of Odyssey’s replication strategies, using work-steal-
predict with real datasets, using 100 queries.

Index Scalability. We present in Figure 14 the total index size in GBs, for
every replication strategy when using 8 nodes, for all real datasets we used
and for Random 100GB (Ran.100). In all cases, the index size is very small
compared to the size of the dataset. Figures 17(a) and 17(b) illustrate the
index creation time of Odyssey for our 1B series Deep dataset using equally-
split, as the dataset size increases on a system with $16$ nodes and as the
number of nodes increases (while using the full size datasets), respectively.
In both cases, we observe optimal speedup regarding index creation.
Additionally, Figure 17(c) presents the scalability of Odyssey on the Random
dataset as both the dataset size and the number of nodes increase linearly,
again using equally-split. As shown, Odyssey achieves perfect scalability
since the corresponding buffer times and index times remain almost constant.

(a)

(b)

(c)

(d)

Figure 17:  (a) Index scalability on Deep using equally-split, as the dataset
size increases, with $16$ nodes. (b) Index scalability on Deep using equally-
split, as the number of nodes increases. (c) Index scalability on the Random
dataset as both the dataset size and the number of nodes increase linearly,
using equally-split. (d) Comparison of work-steal-predict with Odyssey’s
different data partitioning schemes, against other implementations, using
Seismic.

Data partitioning and comparison to competitors. Figure 17(d) presents (i) a
comparison of work-steal-predict Odyssey’s best performing algorithm, against
dmessi, dmessi-sw-bsf, and dpisax; and (ii) the performance of Odyssey’s
different data partitioning schemes, i.e., equally-split and density-aware, as
well as the full replication strategy, using Seismic. Interestingly, dmessi
performs significantly worse that all the other implementations, showing that
by simply executing multiple instances of a SotA single-node algorithm like
MESSI on a multi-node system (in order to scale its applicability on larger
dataset sizes) does not perform well on real datasets; thus, more
sophisticated approaches are required. On the other hand, Odyssey’s work-
steal-predict with full replication strategy is significantly better than all
its competitors. Specifically, it is up to $6.6$x, $3.7$x and $3.8$x faster
than dmessi, dmessi-sw-bsf, and dpisax, respectively. Moreover, regarding
Odyssey’s data partitioning techniques, Figure 17(d) shows that work-steal-
predict with the density-aware partitioning performs better than equally-
split.

Extensions to _k-NN_ and DTW. Finally, we present experiments with _k-NN_
queries, and the DTW distance, where we measure the query answering time for
100 queries as we increase the number of nodes, when using different
replication strategies. We evaluated all replication strategies when varying
$k$ between 1 and 20 for _k-NN_ , and when varying the warping window size
between 1%-15% of the series length for DTW. Figure 19 shows the _k-NN_
results for $k=10$, and Figure 19 shows the DTW results for 5% warping
(results with the rest of parameter values are similar). As expected, query
answering times are in both cases higher than before, while using more nodes
and higher replication degrees improves performance in the same way we have
observed in previous experiments. Results with Seismic exhibit similar trends
and are omitted for brevity.

Figure 18: _10-NN_ query
answering (Random 100GB)

Figure 19: DTW with 5% warping (Random 100GB).

##  6 Conclusions

In this work, we presented Odyssey, a novel distributed data-series processing
framework that takes advantage of the full computational capacity of modern
clusters comprised of multi-core servers. Odyssey addresses a number of
challenges in designing an efficient and highly-scalable distributed data
series index, including efficient scheduling, load-balancing, and flexible
partial replication, and successfully navigates the fundamental trade-off
between data scalability and good performance during query answering. In
future work, we plan to extend Odyssey to support subsequence similarity
search Linardi and Palpanas (2020), as well as approximate similarity search.

## Acknowledgments

Work supported by NSFC Grant No. 62202450, EU Horizon 2020 Marie Sklodowska-
Curie project No 101031688, and Hellenic Foundation for Research and
Innovation (HFRI) under the “Second Call for HFRI Research Projects to support
Faculty Members and Researchers” No 3684. Numerical computations performed on
the S-CAPAD/DANTE platform, IPGP, France.

## References

  * Palpanas [2015] Themis Palpanas.  Data series management: The road to big sequence analytics.  _SIGMOD Record_ , 2015.
  * Zoumpatianos and Palpanas [2018] Kostas Zoumpatianos and Themis Palpanas.  Data series management: Fulfilling the need for big sequence analytics.  In _ICDE_ , 2018.
  * Palpanas and Beckmann [48(3] Themis Palpanas and Volker Beckmann.  Report on the first and second interdisciplinary time series analysis workshop (ITISA).  _SIGREC_ , 48(3), 2019.
  * Kashino et al. [1999] Kunio Kashino, Gavin Smith, and Hiroshi Murase.  Time-series active search for quick retrieval of audio and video.  In _ICASSP_ , 1999.
  * Ye and Keogh [2009] Lexiang Ye and Eamonn Keogh.  Time series shapelets: a new primitive for data mining.  In _SIGKDD_. ACM, 2009.
  * Huijse et al. [2014] Pablo Huijse, Pablo A Estevez, Pavlos Protopapas, Jose C Principe, and Pablo Zegers.  Computational intelligence challenges and applications on large-scale astronomical time series databases.  _CIM_ , 2014.
  * Raza et al. [2015] Usman Raza, Alessandro Camerra, Amy L Murphy, Themis Palpanas, and Gian Pietro Picco.  Practical data prediction for real-world wireless sensor networks.  _TKDE_ , 2015.
  * Boniol et al. [2020] Paul Boniol, Michele Linardi, Federico Roncallo, and Themis Palpanas.  Automated Anomaly Detection in Large Sequences.  In _ICDE_ , 2020.
  * Boniol and Palpanas [2020] Paul Boniol and Themis Palpanas.  Series2Graph: Graph-based Subsequence Anomaly Detection for Time Series.  _PVLDB_ , 2020.
  * Linardi et al. [2020] Michele Linardi, Yan Zhu, Themis Palpanas, and Eamonn J. Keogh.  Matrix Profile Goes MAD: Variable-Length Motif And Discord Discovery in Data Series.  In _DAMI_ , 2020.
  * Wang et al. [2023a] Qitong Wang, Stephen Whitmarsh, Vincent Navarro, and Themis Palpanas.  iEDeaL: A Deep Learning Framework for Detecting Highly Imbalanced Interictal Epileptiform Discharges.  _PVLDB_ , 16(2), 2023a.
  * Echihabi et al. [2018] Karima Echihabi, Kostas Zoumpatianos, Themis Palpanas, and Houda Benbrahim.  The Lernaean Hydra of Data Series Similarity Search: An Experimental Evaluation of the State of the Art.  _PVLDB_ , 2018.
  * Echihabi et al. [2019] Karima Echihabi, Kostas Zoumpatianos, Themis Palpanas, and Houda Benbrahim.  Return of the Lernaean Hydra: Experimental Evaluation of Data Series Approximate Similarity Search.  _PVLDB_ , 2019.
  * Echihabi et al. [2020] Karima Echihabi, Kostas Zoumpatianos, and Themis Palpanas.  Scalable machine learning on high-dimensional vectors: From data series to deep network embeddings.  In _WIMS: The 10th International Conference on Web Intelligence, Mining and Semantics_ , pages 1–6. ACM, 2020.
  * Echihabi et al. [2021a] Karima Echihabi, Kostas Zoumpatianos, and Themis Palpanas.  Big sequence management: Scaling up and out.  In _Proceedings of the 24th International Conference on Extending Database Technology, EDBT_ , pages 714–717. OpenProceedings.org, 2021a.
  * Echihabi et al. [2021b] Karima Echihabi, Themis Palpanas, and Kostas Zoumpatianos.  New trends in high-d vector similarity search: Ai-driven, progressive, and distributed.  _Proc. VLDB Endow._ , 14(12):3198–3201, 2021b.
  * Bagnall et al. [9(7] Anthony J. Bagnall, Richard L. Cole, Themis Palpanas, and Konstantinos Zoumpatianos.  Data series management.  _Dagstuhl Reports_ , 9(7), 2019.
  * Palpanas [2017] Themis Palpanas.  The parallel and distributed future of data series mining.  In _HPCS_ , 2017.
  * Peng et al. [2018] Botao Peng, Themis Palpanas, and Panagiota Fatourou.  Paris: The next destination for series indexing and query answering.  _IEEE BigData_ , 2018.
  * Peng et al. [2020a] Botao Peng, Themis Palpanas, and Panagiota Fatourou.  Paris+: Data series indexing on multi-core architectures.  _TKDE_ , 2020a.
  * Echihabi et al. [2022] Karima Echihabi, Panagiota Fatourou, Kostas Zoumpatianos, Themis Palpanas, and Houda Benbrahim.  Hercules Against Data Series Similarity Search.  _PVLDB_ , 2022.
  * Gogolou et al. [2019] Anna Gogolou, Theophanis Tsandilas, Themis Palpanas, and Anastasia Bezerianos.  Progressive similarity search on time series data.  In _EDBT_ , 2019.
  * Echihabi et al. [2023] Karima Echihabi, Theophanis Tsandilas, Anna Gogolou, Anastasia Bezerianos, and Themis Palpanas.  ProS: Data Series Progressive k-NN Similarity Search and Classification with Probabilistic Quality Guarantees.  _VLDBJ_ , 2023.
  * Peng et al. [2020b] Botao Peng, Themis Palpanas, and Panagiota Fatourou.  Messi: In-memory data series indexing.  In _ICDE_ , 2020b.
  * Peng et al. [2021a] Botao Peng, Panagiota Fatourou, and Themis Palpanas.  Fast data series indexing for in-memory data.  _VLDB J._ , 30(6):1041–1067, 2021a.
  * Peng et al. [2021b] Botao Peng, Panagiota Fatourou, and Themis Palpanas.  SING: Sequence Indexing Using GPUs.  In _Proceedings of the International Conference on Data Engineering (ICDE)_ , 2021b.
  * cod [2022] Odyssey code and datasets.  https://helios2.mi.parisdescartes.fr/~themisp/odyssey/, 2022.
  * Shieh and Keogh [2008] Jin Shieh and Eamonn Keogh.  Isax: Indexing and mining terabyte sized time series.  2008\.  doi:10.1145/1401890.1401966.  URL https://doi.org/10.1145/1401890.1401966.
  * Palpanas [2020] Themis Palpanas.  Evolution of a Data Series Index - The iSAX Family of Data Series Indexes.  In _Communications in Computer and Information Science (CCIS)_ , volume 1197, 2020.
  * Azizi et al. [2023] Ilias Azizi, Karima Echihabi, and Themis Palpana.  Elpis: Graph-Based Similarity Search for Scalable Data Science.  _PVLDB_ , 2023.
  * Agrawal et al. [1993] Rakesh Agrawal, Christos Faloutsos, and Arun N. Swami.  Efficient similarity search in sequence databases.  In David B. Lomet, editor, _FODO_ , 1993.
  * Sellis et al. [1987] Timos K. Sellis, Nick Roussopoulos, and Christos Faloutsos.  The r+-tree: A dynamic index for multi-dimensional objects.  In _VLDB_ , 1987.
  * Rafiei and Mendelzon [1998] Davood Rafiei and Alberto O. Mendelzon.  Efficient retrieval of similar time sequences using DFT.  In Katsumi Tanaka and Shahram Ghandeharizadeh, editors, _FODO_ , 1998.
  * Wang et al. [2013] Yang Wang, Peng Wang, Jian Pei, Wei Wang, and Sheng Huang.  A data-adaptive and dynamic segmentation index for whole matching on time series.  _PVLDB_ , 6(10), 2013.
  * Keogh et al. [2001] Eamonn J. Keogh, Kaushik Chakrabarti, Sharad Mehrotra, and Michael J. Pazzani.  Locally adaptive dimensionality reduction for indexing large time series databases.  In Sharad Mehrotra and Timos K. Sellis, editors, _SIGMOD_ , 2001\.
  * Camerra et al. [2014] Alessandro Camerra, Jin Shieh, Themis Palpanas, Thanawin Rakthanmanon, and Eamonn J. Keogh.  Beyond one billion time series: indexing and mining very large time series collections with isax2+.  _Knowl. Inf. Syst._ , 39(1):123–151, 2014.  URL http://dblp.uni-trier.de/db/journals/kais/kais39.html#CamerraSPRK14.
  * Zoumpatianos et al. [2014] Kostas Zoumpatianos, Stratos Idreos, and Themis Palpanas.  Indexing for interactive exploration of big data series.  In _SIGMOD_ , 2014.
  * Zoumpatianos et al. [2016] Kostas Zoumpatianos, Stratos Idreos, and Themis Palpanas.  Ads: the adaptive data series index.  _VLDB J._ , 2016.
  * Linardi and Palpanas [2019] Michele Linardi and Themis Palpanas.  Scalable, variable-length similarity search in data series: The ulisse approach.  _PVLDB_ , 2019.
  * Linardi and Palpanas [2020] Michele Linardi and Themis Palpanas.  Scalable data series subsequence matching with ulisse.  _VLDBJ_ , 2020.
  * Wang and Palpanas [2021] Qitong Wang and Themis Palpanas.  Deep learning embeddings for data series similarity search.  In _KDD ’21: The 27th ACM SIGKDD Conference on Knowledge Discovery and Data Mining, Virtual Event, Singapore, August 14-18, 2021_ , pages 1708–1716. ACM, 2021.
  * Wang et al. [2023b] Zeyu Wang, Qitong Wang, Peng Wang, Themis Palpanas, and Wei Wang.  Dumpy: A compact and adaptive index for large data series collections.  In _SIGMOD_ , 2023b.
  * Jensen et al. [2017] Søren Kejser Jensen, Torben Bach Pedersen, and Christian Thomsen.  Time series management systems: A survey.  _IEEE Trans. Knowl. Data Eng._ , 29(11):2581–2600, 2017.
  * Pelkonen et al. [2015] Tuomas Pelkonen, Scott Franklin, Paul Cavallaro, Qi Huang, Justin Meza, Justin Teller, and Kaushik Veeraraghavan.  Gorilla: A fast, scalable, in-memory time series database.  _VLDB_ , 2015.
  * Crate [2018] Crate.  CrateDB: Real-time SQL Database for Machine Data & IoT, 2018.  URL http://crate.io/.
  * InfluxDB [2018] InfluxDB.  InfluxDB - Open Source Time Series, Metrics, and Analytics Database (http://influxdb.com/), 2018.  URL http://influxdb.com/.
  * Prometheus [2018] Prometheus.  Prometheus – Monitoring system & time series database, 2018.  URL http://prometheus.io/.
  * QuasarDB [2018] QuasarDB.  QuasarDB: high-performance, distributed, time series database, 2018\.  URL https://www.quasardb.net/.
  * Hellium [2018] Hellium.  Hellium: Ultra high performance key/value storage, 2018.  URL https://www.levyx.com/helium.
  * RiakTS [2018] RiakTS.  Riak TS – Basho Technologies, 2018.  URL http://basho.com/products/riak-ts/.
  * Timescale [2018] Timescale.  Timescale - an open source time series management system, 2018.  URL http://timescale.com/.
  * Wang et al. [2020] Chen Wang, Xiangdong Huang, Jialin Qiao, Tian Jiang, Lei Rui, Jinrui Zhang, Rong Kang, Julian Feinauer, Kevin Mcgrail, Peng Wang, Diaohan Luo, Jun Yuan, Jianmin Wang, and Jiaguang Sun.  Apache iotdb: Time-series database for internet of things.  _Proc. VLDB Endow._ , 13(12):2901–2904, 2020\.
  * OpenTSDB [2015] OpenTSDB.  OpenTSDB - A Distributed, Scalable Monitoring System (http://opentsdb.net/), 2015.  URL http://opentsdb.net/.
  * Timely [2018] Timely.  Timely – A secure time series database based on Accumulo and Grafana, 2018.  URL https://code.nsa.gov/timely/.
  * Warp10 [2018] Warp10.  Warp 10 – The Most Advanced Time Series Platform., 2018.  URL https://www.warp10.io/.
  * Wu et al. [2019] Jiaye Wu, Peng Wang, Ningting Pan, Chen Wang, Wei Wang, and Jianmin Wang.  Kv-match: A subsequence matching approach supporting normalization and time warping.  In _35th IEEE International Conference on Data Engineering, ICDE 2019, Macao, China, April 8-11, 2019_ , pages 866–877. IEEE, 2019.
  * Feng et al. [2020] Kefeng Feng, Peng Wang, Jiaye Wu, and Wei Wang.  L-match: A lightweight and effective subsequence matching approach.  _IEEE Access_ , 8:71572–71583, 2020.
  * Zhang et al. [2019] Liang Zhang, Noura Alghamdi, Mohamed Y. Eltabakh, and Elke A. Rundensteiner.  TARDIS: distributed indexing framework for big time series data.  In _ICDE_ , 2019.
  * Yagoubi et al. [2017] Djamel Edine Yagoubi, Reza Akbarinia, Florent Masseglia, and Themis Palpanas.  Dpisax: Massively distributed partitioned isax.  pages 1135–1140, 2017.  doi:10.1109/ICDM.2017.151.
  * Yagoubi et al. [2020] Djamel-Edine Yagoubi, Reza Akbarinia, Florent Masseglia, and Themis Palpanas.  Massively distributed time series indexing and querying.  _IEEE Transactions on Knowledge and Data Engineering_ , 32(1):108–120, 2020.  doi:10.1109/TKDE.2018.2880215.
  * Cil [1996] Cilk: An efficient multithreaded runtime system.  _Journal of Parallel and Distributed Computing_ , 37(1):55–69, 1996.
  * Blumofe and Leiserson [1999] Robert D. Blumofe and Charles E. Leiserson.  Scheduling multithreaded computations by work stealing.  _J. ACM_ , 46(5):720–748, sep 1999.  ISSN 0004-5411.  doi:10.1145/324133.324234.  URL https://doi.org/10.1145/324133.324234.
  * Fatourou and Spirakis [1999] Panagiota Fatourou and Paul Spirakis.  A new scheduling algorithm for general strict multithreaded computations.  In Prasad Jayanti, editor, _Distributed Computing_ , pages 297–311, Berlin, Heidelberg, 1999. Springer Berlin Heidelberg.  ISBN 978-3-540-48169-0.
  * Fatourou and Spirakis [2000] Panagiota Fatourou and Paul Spirakis.  Efficient scheduling of strict multithreaded computations.  _Theory of Computing Systems_ , 33:173–232, 2000.
  * Blelloch et al. [1999] Guy E. Blelloch, Phillip B. Gibbons, and Yossi Matias.  Provably efficient scheduling for languages with fine-grained parallelism.  _J. ACM_ , 46(2):281–321, mar 1999.  ISSN 0004-5411.  doi:10.1145/301970.301974.  URL https://doi.org/10.1145/301970.301974.
  * Blelloch et al. [1997] Guy E. Blelloch, Phillip B. Gibbons, Yossi Matias, and Girija J. Narlikar.  Space-efficient scheduling of parallelism with synchronization variables.  In _Proceedings of the Ninth Annual ACM Symposium on Parallel Algorithms and Architectures_ , SPAA ’97. Association for Computing Machinery, 1997\.  ISBN 0897918908.
  * Fatourou [2001] Panagiota Fatourou.  Low-contention depth-first scheduling of parallel computations with write-once synchronization variables.  In _Proceedings of the Thirteenth Annual ACM Symposium on Parallel Algorithms and Architectures_ , SPAA ’01, New York, NY, USA, 2001. Association for Computing Machinery.  ISBN 1581134096.  doi:10.1145/378580.378639.  URL https://doi.org/10.1145/378580.378639.
  * Gogolou et al. [2020] Anna Gogolou, Theophanis Tsandilas, Karima Echihabi, Anastasia Bezerianos, and Themis Palpanas.  Data Series Progressive Similarity Search with Probabilistic Quality Guarantees.  In _Proceedings of the 2020 ACM SIGMOD International Conference on Management of Data_ , 2020.
  * iri [2016] Incorporated Research Institutions for Seismology – Seismic Data Access.  http://ds.iris.edu/data/access/, 2016.
  * Attiya and Welch [2004] Hagit Attiya and Jennifer Welch.  _Distributed Computing: Fundamentals, Simulations and Advanced Topics_.  John Wiley and Sons, Inc., Hoboken, NJ, USA, 2004.  ISBN 0471453242.
  * Lynch [1996] Nancy A. Lynch.  _Distributed Algorithms_.  Morgan Kaufmann Publishers Inc., San Francisco, CA, USA, 1996.  ISBN 9780080504704.
  * Gardner [1986] Martin Gardner.  _Knotted Doughnuts and Other Mathematical Entertainments_.  W. H. Freeman, 1986.
  * Keogh and Ratanamahatana [2005] Eamonn Keogh and Chotirat Ann Ratanamahatana.  Exact indexing of dynamic time warping.  _KIS_ , 2005.
  * Faloutsos et al. [1994] Christos Faloutsos, M. Ranganathan, and Yannis Manolopoulos.  Fast subsequence matching in time-series databases.  In _SIGMOD_ , pages 419–429, New York, NY, USA, 1994. ACM.  doi:10.1145/191839.191925.  URL http://doi.acm.org/10.1145/191839.191925.
  * Zoumpatianos et al. [2015] Kostas Zoumpatianos, Yin Lou, Themis Palpanas, and Johannes Gehrke.  Query workloads for data series indexes.  In _Proceedings of the 21th ACM SIGKDD International Conference on Knowledge Discovery and Data Mining, Sydney, NSW, Australia, August 10-13, 2015_ , pages 1603–1612, 2015.  doi:10.1145/2783258.2783382.  URL http://doi.acm.org/10.1145/2783258.2783382.
  * Zoumpatianos et al. [2018] Kostas Zoumpatianos, Yin Lou, Ioana Ileana, Themis Palpanas, and Johannes Gehrke.  Generating data series query workloads.  _VLDB J._ , 2018.
  * for Seismology with Artificial Intelligence [2018] Incorporated Research Institutions for Seismology with Artificial Intelligence.  Seismic Data Access.  http://ds.iris.edu/data/access/, 2018.
  * Soldi et al. [2014] S Soldi, Volker Beckmann, WH Baumgartner, Gabriele Ponti, Chris R Shrader, P Lubiński, HA Krimm, F Mattana, and Jack Tueller.  Long-term variability of agn at hard x-rays.  _Astronomy & Astrophysics_, 563:A57, 2014.
  * Vision [2018] Skoltech Computer Vision.  Deep billion-scale indexing.  http://sites.skoltech.ru/compvision/noimi, 2018.
  * Jégou et al. [2011] Hervé Jégou, Romain Tavenard, Matthijs Douze, and Laurent Amsaleg.  Searching in one billion vectors: Re-rank with source coding.  In _Proceedings of the IEEE International Conference on Acoustics, Speech, and Signal Processing, ICASSP 2011, May 22-27, 2011, Prague Congress Center, Prague, Czech Republic_ , pages 861–864. IEEE, 2011\.  doi:10.1109/ICASSP.2011.5946540.  URL https://doi.org/10.1109/ICASSP.2011.5946540.
  * Simhadri et al. [2022] Harsha Vardhan Simhadri, George Williams, Martin Aumüller, Matthijs Douze, Artem Babenko, Dmitry Baranchuk, Qi Chen, Lucas Hosseini, Ravishankar Krishnaswamny, Gopal Srinivasa, Suhas Jayaram Subramanya, and Jingdong Wang.  Results of the neurips?21 challenge on billion-scale approximate nearest neighbor search.  In _Proceedings of the NeurIPS 2021 Competitions and Demonstrations Track_ , pages 177–189, 2022.