# DynaHash: Efficient Data Rebalancing in Apache AsterixDB (Extended Version)

Chen Luo* Snowflake Inc.
<EMAIL_ADDRESS>* Work done while the author was at University of
California, Irvine. Michael J. Carey  University of California, Irvine
<EMAIL_ADDRESS>
###### Abstract

Parallel shared-nothing data management systems have been widely used to
exploit a cluster of machines for efficient and scalable data processing. When
a cluster needs to be dynamically scaled in or out, data must be efficiently
rebalanced. Ideally, data rebalancing should have a low data movement cost,
incur a small overhead on data ingestion and query processing, and be
performed online without blocking reads or writes. However, existing parallel
data management systems often exhibit certain limitations and drawbacks in
terms of efficient data rebalancing.

In this paper, we introduce DynaHash, an efficient data rebalancing approach
that combines _dyna_ mic bucketing with extendible _hash_ ing for shared-
nothing OLAP-style parallel data management systems. DynaHash dynamically
partitions the records into a number of buckets using extendible hashing to
achieve good a load balance with small rebalancing costs. We further describe
an end-to-end implementation of the proposed approach inside an open-source
Big Data Management System (BDMS), Apache AsterixDB. Our implementation
exploits the out-of-place update design of LSM-trees to efficiently rebalance
data without blocking concurrent reads and writes. Finally, we have conducted
performance experiments using the TPC-H benchmark and we present the results
here.

##  I Introduction

The coming end of Moore’s law and the information age have led data management
systems to exploit clusters of machines to process large amounts of fast
growing data. As a result, parallel shared-nothing data management systems
have become widely used due to their scalability. In a shared-nothing parallel
data management system, records are partitioned across a cluster of nodes that
communicate with each other via an interconnection network [26]. The shared-
nothing parallel architecture enables these systems to be horizontally scaled
as the number of nodes increases [26].

Early parallel data management systems [14, 25] generally assumed that the
cluster of nodes is relatively static. However, this assumption is no longer
true. It is desirable to dynamically adjust the cluster size for a number of
reasons. For example, it can be economical to dynamically scale the cluster in
and out as the workload changes, especially in the era of cloud computing.
Moreover, as stored data accumulates over time, the cluster also needs to be
scaled out to better serve the query workloads. In order to scale a cluster in
or out, the stored records must be rebalanced so that they can be
repartitioned to the new set of nodes. Ideally, rebalance operations should
result in a near-perfect load balance, impose a low overhead on regular
database operations, and involve a small data movement cost. Moreover,
rebalancing must be performed online so that reads and writes are not blocked.

In this paper, we focus on data rebalancing for shared-nothing parallel data
management systems aimed at analytical (OLAP) workloads. Even though many
parallel data management systems today have implemented various data
rebalancing functionalities, the existing implementations often exhibit
certain limitations or drawbacks for OLAP data management. Data management
systems that support OLTP workloads [2, 3, 18, 34, 54] often perform fine-
grained range partitioning to enable efficient data rebalancing. However, this
is not suitable for shared-nothing OLAP systems due to the potential query
load imbalance caused by range skews. Existing OLAP systems either rely on the
underlying shared storage for data rebalancing [19], incur a large data
movement cost [38], and/or block writes during rebalancing [33].

Our Contributions. In this paper, we present DynaHash, an efficient data
rebalancing approach for OLAP-style parallel data management systems with
local secondary indexes. The basic idea of DynaHash is to dynamically
partition the records into a set of buckets using extendible hashing [30] and
to move buckets for efficient rebalancing. By combining extendible hashing
with dynamic bucketing, DynaHash can greatly reduce the data movement cost
with a minimal impact on data ingestion and query processing performance.

As the second contribution, we describe an efficient rebalancing
implementation that avoids blocking concurrent reads and writes by exploiting
the out-of-place design offered by LSM-trees [48]. The techniques used include
bucketed LSM storage, lazy secondary index cleanup, concurrency control for
online rebalancing, and an effective approach to fault tolerance and recovery.
Even though some similar individual techniques have been implemented by other
systems, our contribution here is to show how to integrate them together to
yield an efficient and effective rebalancing implementation.

As the last contribution, we have implemented all of the proposed techniques
inside Apache AsterixDB [1] and have carried out extensive experiments on the
TPC-H benchmark [6] to evaluate the effectiveness of the techniques. The
experimental results show that the proposed rebalancing approach in DynaHash
significantly reduces the rebalance cost with just a small overhead on query
and ingestion performance. It should be noted that while our approach has been
implemented in an LSM-based row store, the design itself can be naturally
generalized to shared-nothing OLAP systems based on column-stores since those
systems have generally adopted the same out-of-place update design for their
data [31, 38, 52].

The remainder of this paper is organized as follows. Section II discusses
background information and related work. Section III presents an overview of
our proposed rebalancing approach. Section IV describes how to store buckets
efficiently on a single node. Section V presents the detailed design and
implementation of the rebalance operation. Section VI experimentally evaluates
the proposed techniques. Finally, Section VII concludes the paper.

##  II Background

###  II-A Data Rebalancing

To exploit the parallelism provided by a cluster of nodes, the records of a
dataset must be distributed to each node using a partitioning function. A
partitioning function deterministically assigns each record to a node based on
its partitioning key. Example partitioning functions include range
partitioning and hash partitioning111There could be other partitioning
functions in practice, such as round-robin partitioning and random
partitioning. However, we do not consider them here because those partitioning
functions are not deterministic.. Range partitioning divides the key space
into a set of ranges, each of which is assigned to a node. In contrast, hash
partitioning operates on the hashed keys to achieve a better load balance.

When the cluster needs to be scaled in or out, its datasets must be
repartitioned through a rebalance process. In general, rebalancing has three
important trade-offs, i.e., the load balance, the rebalance cost, and the
normal operation overhead. The load balance measures how evenly the data is
distributed across different nodes. This directly impacts query performance,
as in a shared-nothing system the query time is bottlenecked by the slowest
node. The rebalance cost measures how much of the data needs to be accessed
and moved during rebalancing. Finally, the normal operation overhead measures
any extra overhead for normal read and write operations in order to support
the needs of the rebalance operation.

Rebalancing data changes its partitioning function. Depending on how the
partitioning function changes, existing rebalancing schemes can be classified
as either _global_ or _local_. A global rebalancing scheme repartitions
(nearly) all records of a dataset when the cluster changes. This generally
leads to a near-perfect load balance and a small normal operation overhead but
at a very high rebalance cost. For example, with range partitioning, a global
rebalancing scheme can recompute the key range of each node based on the new
cluster size and then repartition all records based on the new ranges.

In contrast to global rebalancing, a local rebalancing scheme only changes the
partitioning function “locally” so that only a small portion of the records,
generally proportional to the affected nodes, are moved. This reduces the
rebalancing cost, but generally leads to a worse load balance and a higher
normal operation overhead. Commonly used local rebalancing schemes include
_static bucketing_ , _dynamic bucketing_ , and _consistent hashing_. In static
bucketing, the key space is pre-partitioned into a fixed number of buckets,
each of which is assigned to a node through a directory. During rebalancing,
only a small number of affected buckets are moved to new nodes, which
significantly reduces the rebalance cost. Dynamic bucketing further extends
the usability of static bucketing by dynamically splitting or merging buckets
as the dataset size grows or shrinks. Finally, consistent hashing eliminates
the overhead of the global directory by organizing the (hashed) key space into
a ring structure and letting each node serve a key range222The range
partitioning counterpart of consistent hashing is rarely used in practice
because of the potential for range skews. Thus, that scheme is not considered
here.. When a node is added or removed, its key range is adjusted locally
based on its next neighbor node. In general, consistent hashing is more
suitable for a large peer-to-peer architecture since it does not require a
global directory. In contrast, dynamic bucketing works naturally with a more
centralized (primary-secondary) architecture where the bucket assignment
information is managed by the coordinator.

###  II-B Log-Structured Merge Trees

The LSM-tree [48] is a persistent index structure optimized for write-
intensive workloads. The LSM-tree adopts an out-of-place update design by
always buffering writes into a memory component and appending records to a
transaction log for durability. Whenever the memory component is full, writes
are flushed to disk to form an immutable disk component. Multiple disk
components are periodically merged together to form a larger one, according a
pre-defined merge policy.

A query over an LSM-tree has to reconcile the entries with identical keys from
multiple components, as entries from newer components override those from
older components. A range query searches all components simultaneously using a
priority queue to perform reconciliation. A point lookup query simply searches
all components from newest to oldest until the first match is found. To speed
up point lookups, a common optimization is to build Bloom filters [13] over
the sets of keys stored in disk components.

###  II-C Apache AsterixDB

Apache AsterixDB [1, 11, 17] is an open-source Big Data Management System
(BDMS) that aims to manage massive amounts of semi-structured (e.g., JSON)
data efficiently. AsterixDB uses a shared-nothing parallel architecture with
local secondary indexes for OLAP-style workloads. An AsterixDB cluster
contains a Cluster Controller (CC) that serves as the coordinator and multiple
Node Controllers (NCs) that perform data processing tasks. Each NC has
multiple partitions to exploit the parallelism of modern hardware. A query in
AsterixDB is compiled and optimized by the CC into a Hyracks job [16] that is
then executed by the NCs. To support efficient data ingestion, AsterixDB
provides data feeds [32], which are long-running jobs that constantly ingest
external data into AsterixDB.

The records of an AsterixDB dataset are hash-partitioned based on their
primary keys across multiple NC partitions. Each dataset partition is managed
by an LSM-based storage engine [12], including a primary index, a primary key
index, and multiple local secondary indexes. The primary index stores records
indexed by primary keys, and the primary key index stores primary keys only.
The primary key index is built to support COUNT(*) style queries and
uniqueness checks efficiently [41] since it is much smaller than the primary
index. Secondary indexes use the composition of the secondary key and the
primary key as their index keys. AsterixDB supports LSM-based B+-trees,
R-trees, and inverted indexes using a generic LSM-ification framework that can
convert an in-place index into an LSM-based index. Each LSM-tree uses a
tiering-like merge policy to merge its disk components. AsterixDB uses a
record-level transaction model to ensure that indexes are kept consistent
within each partition.

AsterixDB currently uses a global rebalancing scheme with hash partitioning.
Given a cluster with $N$ partitions, AsterixDB assigns each record with key
$K$ to the $hash(K)\textrm{ mod }N$ partition. When the cluster size changes,
the partitioning function is recomputed so that the records of a dataset are
redistributed to the new set of nodes. This approach leads to a near-perfect
load balance with a minimum normal operation overhead, but the rebalance cost
is very high since nearly all records need to be moved during rebalancing. In
this work, we explore alternative data rebalancing schemes to make better
trade-offs among these three costs.

###  II-D Related Work

Rebalancing in Parallel Data Management Systems. Nearly all parallel data
management systems today have implemented some form of rebalancing. Here we
discuss rebalancing in some representative systems based on the taxonomy of
Section II-A.

For OLTP-style systems, Google’s Bigtable [18] and its open-source cousin
HBase [2] use dynamic bucketing with a shared-data architecture. Since their
underlying distributed storage systems already support rebalancing of
immutable data blocks, Bigtable and HBase only need to manage their in-memory
data during rebalancing. Dynamo [24] and its open-source cousin Cassandra [37]
are shared-nothing systems that use consistent hashing. Cassandra further
introduces the concept of virtual nodes to achieve a better load balance; the
basic idea is to let each node use multiple virtual nodes to manage multiple
key ranges. Couchbase [15] and Oracle NoSQL Database [4] are shared-nothing
systems that use static bucketing with hash partitioning. Both systems set the
number of buckets to a relatively high number. Couchbase sets this number to
1024 by default, while Oracle NoSQL Database recommends that each node (in the
expected largest cluster) should have 10 to 20 buckets. Minhas et al. [47]
applied a similar static bucketing approach to enable efficient scaling for
VoltDB [7]. MongoDB [3], TiDB [34], WattDB [51], and CockroachDB [54] each use
range-partitioned dynamic bucketing with a very small bucket size, e.g., 64MB.
Such range partitioning is suitable for OLTP workloads since each transaction
only accesses a small number of (usually one) partitions. However, it may not
be suitable for shared-nothing OLAP systems due to the potential range skews
and the fact that queries will often access most partitions. OLTP systems
typically use global secondary indexes due to the high selectiveness of OLTP
queries.

For OLAP-style systems, Snowflake [19] decouples storage from compute and
completely relies on the underlying shared storage for data rebalancing. It is
worth noting that Snowflake also uses consistent hashing to cache immutable
data files into compute nodes. However, this is much simpler than data
rebalancing considered here because the data files are immutable and one can
simply invalidate the cache when the cluster changes. Vertica [38] is a
shared-nothing system that uses global rebalancing with hash partitioning to
achieve a better load balance. Redshift [33] is shared-nothing and supports
both global rebalancing and static bucketing with hash partitioning. However,
Redshift does not support concurrent writes during rebalancing. Moreover, it
directly uses buckets (called “node slices” in Redshift) as its parallelism
unit, which can undesirably change the node parallelism after rebalancing.

Elastic OLTP Databases. Due to the importance and wide adoption of parallel
OLTP database systems, significant effort has been devoted to making them
elastic. Live migration techniques [20, 28, 29] enable OLTP databases to be
migrated without blocking ongoing transactions. E-Store [53] uses fine-grained
partitioning to elastically scale parallel databases. However, a key
difference is that these research efforts mainly focus on ACID transactions
with fine-grained range partitioning, global secondary indexes, and small
queries, while our work focuses on how to rebalance datasets efficiently for
OLAP-style (i.e., analytical query-oriented) systems.

Distributed Access Methods. To efficiently query data stored in a cluster of
nodes, a number of distributed access methods have been proposed as well. The
basic idea is to distribute an access method efficiently over a cluster of
nodes, potentially in a peer-to-peer setting, to support efficient read and
write operations [27, 39, 36, 8], The difference between these access methods
and our work is that we focus on rebalancing for OLAP systems rather than on a
single access method with simple key-value interfaces.

LSM-trees. For data storage in modern systems, a large number of improvements
have been proposed to optimize the LSM-tree [48], including [10, 21, 22, 23,
40, 41, 42, 43, 45, 46, 49, 50]. We refer readers to a recent survey [44] for
a more detailed description of these LSM-tree improvements. These LSM-tree
improvements have focused on single node settings. In contrast, in this work,
we focus on their role in a parallel shared-nothing architecture and exploit
the LSM-tree’s out-of-place design to support efficient data rebalancing with
concurrent reads and writes.

##  III Approach Overview

As mentioned in Section II-A, rebalancing involves three important trade-offs,
i.e., the load balance, the rebalance cost, and the normal operation overhead.
Our goal is to achieve a good load balance with a small rebalance cost and low
normal operation overhead. In this section, we provide a high-level overview
of DynaHash based on the following design choices.

Range Partitioning vs. Hash Partitioning. Most shared-nothing OLAP systems
prefer hash partitioning to achieve a good load balance because the data will
be nearly uniformly distributed and queries generally access most partitions.
In contrast, range partitioning is often unsuitable for shared-nothing OLAP
systems due to the potential for range skews. Thus, we choose hash
partitioning to use here as well.

Global Rebalancing vs. Local Rebalancing. Although global rebalancing achieves
a near-perfect load balance, it incurs a very large rebalance cost since most
records have to be moved during rebalancing. Since our goal is to reduce the
rebalance cost, we prefer to use a local rebalancing scheme. Among the three
local rebalancing schemes (Section II-A), dynamic bucketing dominates static
bucketing by elastically adjusting the number of buckets as data accumulates.
Dynamic bucketing is also preferable to consistent hashing since most parallel
OLAP systems (ours included) adopt a primary-secondary architecture. Based on
these considerations, a natural choice is to use dynamic bucketing here.

Combining Hash Partitioning with Dynamic Bucketing. The last choice we face is
how to combine hash partitioning with dynamic bucketing. One natural design
would be to range-partition the hashed key space into multiple buckets. Though
this solution works, hashing actually provides opportunities for a more
efficient design. Since hashed keys are generally uniformly distributed, one
can use an extendible hashing approach [30] to partition the key space into
multiple buckets. Figure 1 illustrates the resulting architecture based on
this idea with one Cluster Controller (CC) and two Node Controllers (NCs).
Each NC further has two storage partitions. In order to distribute the records
of a dataset to these four partitions, the hash key space is divided into
multiple buckets. A bucket is defined by taking the $d$ low-order bits of the
hash function, where $d$ is called the depth of this bucket. When a bucket
becomes too large, it is split into two smaller buckets by taking one more
hash bit, which thus increments the depth [30]. A rebalance operation can now
move only some affected buckets to new partitions, greatly reducing the
rebalance cost.

As shown in Figure 1, we use a _global directory_ stored at the CC to map
buckets to partitions. This directory has a global depth $D$, which is the
maximum number of bits used in all buckets. Thus, the size of this directory
is $2^{D}$. Note that in Figure 1 the two hash values $011$ and $111$
currently correspond to the same bucket $11$. To locate where a given key $K$
is stored, one simply needs to look in the global directory using the $D$ low-
order bits of $K$’s hash value, where $D$ is the depth of the global
directory. During query compilation, each query creates an immutable copy of
the global directory that it uses throughout query processing. Similarly, a
data feed, i.e., a data ingestion job, also employs an immutable copy of the
global directory in order to distribute the incoming records of a dataset to
the correct NC partitions.

We further use a _local directory_ at each partition to keep track of the
assigned buckets. To simplify bucket splits, the global directory can be
updated lazily before rebalancing is performed. For example, in Figure 1, the
bucket $00$ has already been split into two buckets $000$ and $100$ at
partition 0, but the global directory has not been updated yet. This does not
impact the correctness of the global directory since it can still correctly
route all keys to the right partitions.

Though the basic design in our rebalancing approach of relatively
straightforward, two key challenges must be addressed. First, how can we store
(i.e., physically organize) the multiple buckets within each partition to
enable efficient rebalancing with low normal operation overhead? Second, how
can we efficiently rebalance buckets while supporting both concurrent reads
and writes? In the next two sections we detail our solutions to these two key
challenges.

Figure 1: Example Architecture for DynaHash

##  IV LSM Storage for Buckets

In this section, we discuss how to efficiently store multiple buckets in each
partition. For efficient rebalancing, when a bucket needs to be moved out of a
partition, it is desirable to only access the records in this bucket. If range
partitioning were used and no secondary indexes were built, storing records in
their primary key order would naturally satisfy this property since records in
each bucket would be grouped together. However, with hash partitioning, the
primary key order is no longer the same as the bucket order since records are
bucketed based on hashed keys. Moreover, secondary indexes also complicate
this problem because their entries are ordered by secondary keys, not by
primary keys.

Storage Options. In general, since hash partitioning is used, there are three
storage options for buckets in each partition:

  * •

Option 1: Store entries in their original key order together in one LSM-tree
index.

  * •

Option 2: Store entries in their bucket order in one LSM-tree index. Within
each bucket, store its entries in their original key order.

  * •

Option 3: Store the entries in each bucket in a separate LSM-tree index
structure. Within each LSM-tree, store the entries in their original key
order.

Let us first consider the trade-offs for the primary index. Option 1 incurs no
overhead on reads and writes, but it incurs a large overhead on rebalancing
since moving a bucket must scan all entries, including those from other
buckets. Options 2 and 3 both reduce the rebalancing overhead since the
records within each bucket are physically grouped together. Moreover, Option 3
provides more flexibility for splitting buckets and deleting buckets after
rebalancing. However, Options 2 and 3 both induce some overhead on short
primary index scans since each query must search all buckets. Since short
primary key-order scans are not common in OLAP-style systems, we choose to
optimize the rebalancing performance by choosing Option 3 for the primary
index of a dataset. For secondary indexes, an important difference is that
they do not have to be read during rebalancing as they can be rebuilt on-the-
fly at their destination. In order not to incur too much normal runtime
overhead on secondary index queries, we choose to use Option 1 for secondary
indexes. Figure 2 shows an example of a dataset partition with a primary index
and a secondary index. Here we denote each record as a key-value pair, and the
secondary index is built on the value field. The primary index uses the
bucketed LSM-tree design, which is further described below, to store buckets
separately. The secondary index uses a traditional LSM-tree design to store
all buckets together.

Figure 2: An Example Partition with a Bucketed Primary Index and a Secondary
Index

Bucketed LSM-tree Design. Based on these basic design decisions, we introduce
a bucketed LSM-tree design for efficiently storing multiple buckets in the
primary index. As shown in Figure 2, each bucket can be viewed as a separate
LSM-tree with a memory component and multiple disk components. We use
reference counting for concurrency handling. That is, whenever a bucket, a
memory component, or a disk component is accessed, the reader or writer
increments a reference count so that the accessed entity cannot be destroyed
(reclaimed) until the access completes. All flushes and merges are performed
within each bucket. All buckets are coordinated using a local directory, as
mentioned in Section III. Note that in Figure 2, hashes $011$ and $111$
correspond to the same bucket $11$ with depth 2.

Data Ingestion and Query Processing. A bucketed LSM-tree provides the same set
of interfaces as a traditional LSM-tree. A write operation, including inserts,
deletes, and updates, first checks the local directory using the hash value of
the key to locate which bucket the entry belongs to and then adds the entry to
that bucket. Similarly, a point lookup query only searches its target bucket,
located via the local directory, to get the entry. A primary key range scan
query, however, must search all buckets. There are two approaches to process
such a range scan query. The first approach is to scan each bucket separately.
This will incur no additional overhead compared to the traditional LSM-tree
design, but the returned entries will no longer be sorted on the primary key.
The second approach is to use a priority queue to merge the entries returned
from all buckets together. This approach provides the same interface as the
traditional LSM-tree design by returning sorted results, but will incur a
larger search overhead due to the additional merge-sort step. To decide which
approach should be used, we have introduced an optimization rule in AsterixDB
as follows. By default, the first approach is used to avoid the merge-sort
overhead. However, if primary key order is required by subsequent query
operators, e.g., a user-specified order-by clause or a groupby operator on a
prefix of the primary key, the second approach will be used to avoid the
subsequent sort overhead. Finally, it should be noted that the bucketed LSM-
tree design does not change the processing of a secondary index query, which
simply searches the secondary index to fetch a list of primary keys and then
uses them to fetch records from the (bucketed) primary index.

Figure 3: Bucket Split Example

Efficient Bucket Splits. When a bucket becomes too large, it is split into two
smaller buckets by using one more hash bit. A straightforward implementation
of this would be to build two smaller LSM-trees based the original bucket.
However, this approach would not only cause additional write amplification,
but might also block reads and writes for a long time. Here we describe a more
efficient bucket splitting approach to address these issues.

The pseudocode for splitting a bucket $B$ is depicted in Algorithm 1. The
Split function first stops initiating new component merges for $B$ and waits
for all existing merges to finish. $B$’s memory component is then
asynchronously flushed to disk without blocking writes (line 5). After the
flush completes, the bucket $B$ is locked to temporarily block new readers and
writers so that $B$ can be safely split (lines 6 to 10). Since some writes may
have entered its memory component after the last asynchronous flush, $B$’s
memory component is now flushed synchronously to persist these writes. (It
should be noted that AsterixDB uses a no-steal buffer management policy,
meaning that a memory component is only flushed after all active writers have
completed.) Two new buckets $B_{1}$ and $B_{2}$, whose disk components refer
to the disk components of $B$, are then created. An example is shown in Figure
3. For each disk component of the splitting bucket $11$, we create two new
_reference disk components_ in buckets $011$ and $111$ respectively. A
reference disk component does not store any data; instead, it only points to a
real disk component. All queries accessing data through a reference disk
component must perform an additional filtering step based on the bucket’s hash
value to make sure that only the entries truly belonging to this bucket are
accessed. Thus, the actual creation of the new disk components of $B_{1}$ and
$B_{2}$ are effectively postponed until the next round of merges. Finally, a
directory metadata file that stores valid buckets is forced to disk,
indicating that the split operation is now complete (line 9), and the old
bucket $B$ is reclaimed automatically when its reference count becomes 0. Upon
recovery, the directory metadata file is used to determine the valid buckets.
All invalid (partially split) buckets will be cleaned up automatically.

1:$B$ $\leftarrow$ the bucket to be split

2:function Split(B)

3: Pause scheduling merges for $B$

4: Wait for $B$’s merges to finish

5: Asynchronously flush $B$’s memory component

6: Lock $B$

7: Synchronously flush $B$’s memory component

8: Create two buckets $B_{1}$ and $B_{2}$ that refer to $B$

9: Force a directory metadata file to disk

10: Unlock $B$

11: Resume scheduling merges for $B$

Algorithm 1 Pseudo Code for Bucket Split

To simplify synchronization with the CC, bucket splits are performed at each
partition locally without notifying the CC. Instead, the global directory at
the CC is only refreshed when a rebalance operation starts, as we will discuss
below. This design greatly simplifies the role of the CC since it does not
have to know about the existence of bucket splits.

##  V Efficient Data Rebalancing

Having considered how to store multiple buckets efficiently, we now discuss
how to efficiently rebalance data while supporting concurrent reads and
writes. In AsterixDB, data rebalancing is triggered via a system API after
some nodes have been added or before some nodes are removed. In general, a
rebalance operation contains three phases, namely initialization, data
movement, and finalization. During the initialization phase, all nodes perform
some preparation tasks for subsequent data movement. The data movement phase
transfers some of the records of a dataset, including concurrent writes, to
their new partitions. Finally, during the finalization phase, all nodes
unanimously commit or abort the rebalance operation depending on its outcome,
and some cleanup work is performed. It should be noted that a rebalance
operation may fail for various reasons. If a rebalance operation fails, the
produced intermediate results must be cleaned up correctly. In the remainder
of this section, we discuss the three phases in detail as well as how to
handle rebalance failures.

###  V-A Initialization Phase

When a rebalance operation starts, the CC forces a BEGIN log record indicating
that a rebalance operation has started. The CC further decides which buckets
should be moved to which partitions by computing a new global directory based
on the new set of nodes. In addition, all NCs must perform some preparation
tasks in order to support concurrent updates. The key challenge here is that
AsterixDB only supports a very simple record-level transaction model. If full
ACID transactions were supported by AsterixDB, then the rebalance operation
could potentially be implemented using a transaction, which would
automatically provide concurrency control for reads and writes. Instead, we
must design a customized concurrency control protocol.

Computing the Global Directory. Recall from Section IV that buckets splits are
performed at each node locally, without notifying the CC. Thus, in order to
compute the new global directory, the CC contacts all NCs to get their latest
local directories. Moreover, bucket splits for this dataset at each NC are
disabled until the rebalance completes. Note that since buckets may have
different sizes, it is straightforward to show that an optimal algorithm to
maximize the load balance is NP-hard by considering the partition
problem333The goal of the partition problem is to partition a multiset $S$ of
positive integers into two subsets $S_{1}$ and $S_{2}$ such that the
difference between the sum of elements in $S_{1}$ and the sum of elements in
$S_{2}$ is minimized [5]..

1:function Balance

2: for each unassigned bucket $B$ do

3: Assign $B$ to the least loaded partition

4: while $true$ do

5: $P_{max}\leftarrow$ the most loaded partition

6: $B\leftarrow$ the smallest bucket in $P_{max}$

7: $P_{min}\leftarrow$ the least loaded partition

8: if $abs((|P_{max}|-|B|)-(|P_{min}|+|B|))<|P_{max}|-|P_{min}|$ then

9: Assign $B$ to $P_{min}$

10: else

11: break

Algorithm 2 Pseudo Code for Computing Global Directory

To compute the new global directory efficiently, we use the greedy algorithm
shown in Algorithm 2. To describe this algorithm, we first introduce some
useful concepts. Given a directory with global depth $D$ and a bucket $B$ with
depth $d$, we define the _normalized size_ of the bucket $B$ (denoted as
$|B|$) as $2^{D-d}$. Given a partition $P$ or a node $N$, we denote $|P|$ or
$|N|$ as the sum of the normalized size of $P$’s buckets and $N$’s buckets,
respectively. Given two partitions, $P_{1}$ on node $N_{1}$ and $P_{2}$ on
node $N_{2}$, $P_{1}$ is said to be more loaded than if $|P_{1}|$ is larger
than $|P_{2}|$, or $|N_{1}|$ is larger than $|N_{2}|$ if $|P_{1}|$ equals
$|P_{2}|$. The Balance function first assigns the unassigned buckets (buckets
being displaced due to node removals) to the least loaded partitions (lines
2-4). After all such buckets are assigned, the algorithm balances the bucket
assignment using a series of iterations (lines 4-11). In each iteration, it
tries to assign the smallest bucket $B$ from the most loaded partition
$P_{max}$ to the least loaded partition $P_{min}$ (lines 8-11). If this
assignment reduces the difference between the normalized sizes of $P_{max}$
and $P_{min}$, the assignment is then performed; otherwise, the algorithm
terminates. It is possible to incorporate other heuristics to define the load
order among partitions. For example, one could further consider the total
storage size of a partition, including all datasets. We leave the exploration
of this direction as future work.

Preparing for Concurrent Writes. During the rebalance operation, which may
take a relatively long time to finish, some records may be updated by
concurrent writers. For each bucket that needs to be moved, these concurrent
writes must still be applied to its old partition since the rebalance
operation may fail. Moreover, the concurrent writes must also be applied to
the new partition to ensure that there are no lost writes if the rebalance
operation succeeds.

To ensure correctness with concurrent writes, we use a concurrency control
protocol that splits all the writes to a bucket based on the rebalance start
time. For all writes that happened before the rebalance operation starts, an
immutable bucket snapshot is created so that it can be safely scanned. For all
writes that happen after the rebalance operation starts, their log records are
replicated to the new partition so that the new partition will not miss any
writes. It should be noted that AsterixDB only supports a very simple record-
level transaction model without supporting snapshot scans. To implement the
required snapshot scan, we exploit the immutability of LSM disk components.
Specifically, the memory component of the moving bucket is flushed
synchronously during the initialization phase. This flush time is treated as
the rebalance start time, and the resulting disk components become the
immutable copy of all writes that happened before the rebalance operation
starts. To reduce the blocking of concurrent writes due to the synchronous
flush, the two-flush approach described in Algorithm 1 (lines 5-7) can be
used. Specifically, one can first flush the memory component asynchronously
and then use a synchronous flush to persist any leftover writes. In this case,
the rebalance start time becomes the time of the second (synchronous) flush.

###  V-B Data Movement Phase

After the initialization phase, the rebalance operation starts to move the
affected buckets to their new partitions. This involves adding scanned records
and replicated log records to both the primary index and secondary indexes at
their destination partitions. Moreover, queries must not be affected by the
rebalance operation.

Data Movement. By comparing the current global directory and the new global
directory, it is straightforward to determine the new partition of each
affected bucket. For ease of discussion, here we first describe how to move
one bucket $B$ from its old partition $P_{old}$ to its new partition
$P_{new}$, which is then extended to moving multiple buckets together.

Figure 4 shows the basic data movement process for a single bucket with a
primary index and one secondary index. At the old partition, the primary index
disk components of this bucket are scanned and the log records of any incoming
writes will be replicated. The scanned log records are used to load disk
components at the new partition, with the data represented by the replicated
log records being inserted into the memory components. In order to simplify
concurrency control and recovery, the moved records are always stored
separately from local user writes at the new partition. For a primary index
that uses the bucketed LSM-tree design, the received records are simply stored
in a new bucket. For a secondary index that stores all buckets together, the
received records are stored into a new list of components that are kept
invisible to queries. This design greatly simplifies concurrency control and
recovery. The new components that store moved records will be kept invisible
to queries until the rebalance completes. Moreover, in case the rebalance
operation fails, these new components can then be simply deleted to cleanup
the intermediate results. Finally, to ensure correctness, the scanned data
records must be treated as being strictly older than the replicated log
records. This is achieved by placing the loaded disk component after the disk
components storing replicated log records in the LSM disk component list.

It is straightforward to extend the basic data movement process to move
multiple buckets at the same time. One can simply scan multiple buckets at the
same time and repartition them using the new global directory so that the
scanned records can be sent to their new partitions. As an optimization, when
adding multiple buckets to a secondary index partition, the records can be
added to a single list of components instead of creating one list per bucket.
This will help to reduce the number of disk components present after the
rebalance operation completes.

Figure 4: Data Movement Process (One Bucket): _Scanned data records are loaded
into disk components and replicated log records are inserted into memory
components. These rebalance writes are stored separately from new user
writes._

Handling Concurrent Queries. As mentioned before, the data movement process
shown in Figure 4 greatly simplifies the required concurrency control for
queries. Since the moved records are stored separately from incoming user
writes, the destination’s partially loaded buckets are invisible to queries
until rebalancing completes. If a query starts before a rebalance operation
completes, the query accesses all buckets using a copy of the old global
directory. Otherwise, the query uses the new global directory, updated by the
rebalance operation, to access all buckets. Moreover, since accessed buckets
and LSM components are reference-counted, they can be accessed safely by the
query even if a rebalance operation completes in the middle of the query.

###  V-C Finalization Phase

After all the data records of moving buckets have been transferred to their
new partitions, the system is ready to commit or abort the rebalance operation
depending on its outcome. It should be noted that there could still be active
log replication activities, due to concurrent writes, at this stage. Thus, to
ensure that all nodes always reach a unanimous decision, we use a two-phase
commit protocol as follows.

Prepare Phase. After all data records have been moved to their new partitions,
the CC initializes the prepare phase, which will temporarily block incoming
queries and writes on the rebalancing dataset. The CC further waits for all
NCs to complete their log replication and to flush the memory components that
store rebalancing writes to disk. If all NCs succeed in doing so, i.e., they
all vote yes, the CC enters the _commit_ phase as discussed below. Otherwise,
the rebalance operation must be aborted and the rebalancing dataset will be
left unchanged. All incoming reads and writes will be blocked during this
finalization phase. However, this blocking is expected to be very short since
the CC only waits for existing writers to complete and since the number of log
records pending for replication is bounded.

Commit Phase. Once the CC enters the commit phase, it forces a COMMIT log
record to disk indicating that the rebalance operation is committed. The CC
then updates the global directory of the rebalancing dataset and notifies all
NCs to install their received buckets and cleanup the moved buckets. To
install a received bucket at a partition, the NC simply needs to add (i.e.,
register) the loaded disk components to the component lists of the primary
index and secondary indexes. To cleanup a moved bucket from the primary index
of a partition, the bucket can be simply removed from the bucketed LSM-tree’s
local directory so that it cannot be accessed by new queries. It should be
noted that due to reference counting, the actual components of this bucket
will not be deleted until the last reader exits. To cleanup a secondary index,
we use a lazy delete approach that adds the hash value and the depth of this
bucket to the metadata of each LSM component. A query then performs an
additional validation check to ignore all invalid entries that belong to this
moved bucket. Thus, the cleanup of secondary index components is effectively
postponed to the next round of merges. All these operations, e.g., adding and
removing buckets, are made persistent by forcing metadata files to disk. After
all NCs have completed these tasks, the CC can resume query processing and
data ingestion on the rebalanced dataset. Finally, the CC produces a DONE log
record to indicate that no additional work is needed for this rebalance
operation.

Based on the two-phase commit protocol, the final outcome of the rebalance
operation is determined by whether the COMMIT log record has been forced to
disk successfully by the CC. In other words, the rebalance operation is
committed if the COMMIT log record has been successfully forced to disk.
Otherwise, the CC simply aborts the rebalance operation and leaves the
original dataset as is.

Figure 5: Rebalance Operation Timeline

###  V-D Handling Rebalance Failures

During the rebalance operation, some node(s) may potentially fail at any time.
Since node failures are expected to be rare, we simply abort the rebalance
operation if some node(s) fails before the rebalance operation commits.
However, the intermediate results produced by the rebalance operation must be
cleaned up carefully to ensure that the dataset remains in a consistent state.
Here we assume that the failed node(s) eventually recovers, i.e., no permanent
node failures. We plan to extend DynaHash to incorporate replication [9] to
handle permanent node failures as future work.

Before discussing how to handle various rebalance failures, we first summarize
the basic timeline of a rebalance operation, which is shown in Figure 5. The
CC first forces a BEGIN log record indicating that the rebalance operation has
started. It then requests all NCs to move their affected buckets to new
partitions. After all data movement is done, the CC enters the prepare phase
by waiting for all NCs to complete log replication. After all NCs have
successfully prepared, the CC enters the commit phase by forcing a COMMIT log
record and notifying all NCs to commit this rebalance operation. Finally,
after all NCs have committed, the CC produces a DONE log record indicating
that the rebalance operation can be safely forgotten. Based on this timeline,
we present a case analysis to discuss how to handle various possible rebalance
failures.

Case 1: NC fails before voting “prepared”. In this case, the CC simply aborts
the rebalance operation and asks all NCs (including the failed NC after its
recovery) to cleanup their received buckets. Recall from Figure 4 that the
received records are always added to a separate list of components. Thus, to
cleanup the received buckets, a partition can simply delete those lists of
components for both the primary index and secondary indexes of the dataset. It
should be noted that cleaning up a received bucket is idempotent since
cleaning up a non-existent bucket can be simply treated as a no-op. It is thus
safe to cleanup a received bucket from a partition multiple times. After all
NCs complete the cleanup task, the CC writes a DONE log record so that this
rebalance operation can be safely forgotten.

Case 2: NC fails after voting “prepared”. After the failed NC recovers, it
contacts the CC to report its presence. The NC will further receive
instructions about how to handle the pending rebalance operation. If the
rebalance operation is aborted, the NC simply cleans up the intermediate
results as in Case 1. Otherwise, the NC performs the commit tasks as in Case
4.

Case 3: CC fails before forcing the COMMIT log record. After the CC recovers
and sees the BEGIN log record for the rebalance operation, it aborts the
rebalance operation as in Case 1.

Case 4: NC fails before responding “committed”. The rebalance operation is
committed but the CC does not know whether the NC has committed the rebalance
operation or not. When the NC recovers, the CC requests this NC to commit the
rebalance operation by adding the received buckets and cleaning up the moved
buckets. Similar to case 3, both adding the received buckets and cleaning up
the moved buckets are idempotent operations, which means it is safe to apply
these operations multiple times.

Case 5: CC fails after forcing the COMMIT log record but not the DONE log
record. In this case, the rebalance operation is effectively committed but it
is possible that some NCs have not completed the commit tasks yet. Thus, after
the CC recovers, it notifies all NCs to add received buckets and cleanup moved
buckets as in Case 4. Finally, the CC writes a DONE log record as well.

Case 6: CC fails after the DONE log record is persisted. No additional task
needs to be performed in this case since the DONE log record indicates that
this rebalance operation has completed.

The two-phase commit protocol used in our rebalance operation has some subtle
differences from the traditional two-phase commit protocol used in distributed
transactions. For example, the CC forces a BEGIN log record when a rebalance
operation starts and NCs always contact the CC during recovery. This is
because in AsterixDB the rebalance operation is implemented as a metadata
transaction, and only the CC can produce metadata log records. In contrast, in
traditional distributed transactions each participant can produce log records.
Because of this difference, without forcing the BEGIN log record, the CC may
not know the existence of a rebalance operation if the entire cluster shuts
down before the rebalance operation completes. Similarly, the NC must always
contact the CC upon recovery since the NC cannot certainly know the status of
ongoing rebalance operations. (Contacting the CC upon recovery does not
involve additional overhead since the NC must register itself with the CC for
cluster management anyway.)

##  VI Experimental Evaluation

In this section, we experimentally evaluate the proposed rebalancing
techniques in the context of Apache AsterixDB [1]. Throughout the evaluation,
we mainly focus on the three aspects of different rebalancing approaches,
namely their rebalancing performance, ingestion performance, and query
performance. We first describe the general experimental setup followed by the
detailed evaluation results.

###  VI-A Experimental Setup

Hardware. All experiments were performed on a cluster of nodes with a single
CC and multiple NCs on AWS. The number of NCs ranged from 2 to 16. The CC ran
on a m5.xlarge node with 4 vCPUs, 16GB of memory, and a 500GB elastic block
store (EBS). Each NC ran on an i3.xlarge node with 4 vCPUs, 30.5GB of memory,
a 950GB SSD, and a 500GB EBS. We configured 4 partitions on each NC to exploit
the parallelism provided by multiple cores. The native SSD was used for LSM
storage and the EBS was used for transactional logging. Each NC used a thread
pool with 4 threads to execute LSM flush and merge operations. Each LSM-tree
used a size-tiered merge policy with a size ratio of 1.2 throughout the
experiments, which is similar to the setting used in other systems. This
policy merges a sequence of components when the total size of the younger
components is 1.2 times larger than that of the oldest component in the
sequence. Within this, we allocated 26GB per node of memory for the AsterixDB
instance. The buffer cache size was set at 8GB and the memory component budget
was set at 2GB. Each memory-intensive query operator [35], such as sort, hash
join, and hash group by, received a 128MB memory budget. Both the disk page
size and memory page size were set at 16KB.

Workload. To understand the performance impact of different rebalancing
approaches on OLAP-style workloads, we used the TPC-H [6] benchmark in our
evaluation. We built two secondary indexes, on LineItem and Orders, to enable
index-only plans for certain queries. The LineItem index contains l_shipdate,
l_partkey, l_suppkey, l_extendedprice, l_discount, and l_quantity. The Orders
index contains o_orderdate, o_custkey, o_shippriority, and o_orderpriority.
The scale factor of the TPC-H benchmark was set to 100 times the number of NCs
so that the total amount of data scales linearly as the cluster size
increases. Thus, each NC stored 100GB of TPC-H raw data. The primary index was
compressed for better storage efficiency. The total storage size at each NC,
including compressed primary indexes and uncompressed secondary indexes, was
about 130GB.

Evaluated Rebalancing Approaches. We evaluated three rebalancing approaches.
The first approach that we evaluated is AsterixDB’s current global rebalancing
approach with hash partitioning (called “Hashing”) as the baseline. This
approach simply creates a new dataset that is hash-partitioned based on the
new (target) set of nodes during rebalancing. Although Hashing achieves a near
perfect load balance, it incurs a very high rebalance cost and nearly doubles
the dataset’s disk usage during rebalancing. We further evaluated two
variations of the proposed rebalancing approach. The first variation, called
StaticHash, always splits a dataset into 256 buckets. The actual number of
buckets per partition ranged from 32 to 4 as the number of nodes varied from 2
to 16. Thus, evaluating StaticHash also shows the performance impact of the
number of buckets per partition. The second variation, called DynaHash,
dynamically splits a dataset by setting the maximum bucket size at 10GB. After
loading the TPC-H data, each partition always had 4 buckets.

###  VI-B Ingestion Performance

Figure 6: Ingestion Time

(a) Removing Nodes

(b) Adding Nodes

(c) Concurrent Data Ingestion

Figure 7: Rebalance Time

We first evaluated the ingestion performance of the different rebalancing
approaches. We used a TPC-H client that ran on a separate node to ingest all
TPC-H data into the AsterixDB cluster. The number of nodes of the AsterixDB
cluster varied from 2 to 16. The resulting ingestion time for each rebalancing
approach under different cluster sizes is shown in Figure 6. In general,
DynaHash incurs very a small overhead compared to AsterixDB’s Hashing
approach. Moreover, comparing DynaHash and StaticHash, we see that the
different numbers of buckets per partition also have just a small impact on
the ingestion performance. When the cluster size increases, the ingestion time
for all rebalancing approaches slightly increases because of the write stall
problem of LSM-trees: In general, we found that data ingestion is relatively
CPU-heavy in AsterixDB due to record parsing. When a node has active merges,
its ingestion rate will slow down due to the CPU contention caused by merges.
This in turn will slow down the entire cluster because the overall performance
of a shared-nothing system is bottlenecked by the slowest node, even though
other nodes may not have ongoing merges. Thus, when the number of nodes
increases, the write stall problem becomes more obvious, which increases the
overall ingestion time.

###  VI-C Rebalancing Performance

Next we evaluated the rebalancing performance of the alternative rebalancing
approaches, both for adding nodes and for removing nodes. We further evaluated
the impact of concurrent writes on the rebalance performance.

(a) Query Performance on 4 Nodes

(b) Query Performance on 16 Nodes

Figure 8: Query Performance on Original Cluster

Basic Rebalancing Performance. To understand the basic rebalancing performance
of the different rebalancing approaches, we conducted the following
experiments. We first loaded the TPC-H datasets into an AsterixDB cluster with
N nodes (N ranged from 2 to 16). We then rebalanced all datasets to N-1 nodes
to measure the time required to remove one node. Finally, we rebalanced all
datasets back to N nodes to measure the time required to add one node.

The rebalance times for removing and adding nodes are shown in Figure 7. In
general, both StaticHash and DynaHash substantially reduce the rebalancing
time compared with Hashing for both removing and adding nodes. Moreover, both
of the bucketing approaches have similar rebalance times, which shows that the
number of buckets per partition has a small impact on the rebalancing
performance. Interestingly, we see that Hashing has different performance
trends compared with StaticHash and DynaHash. Hashing has better rebalancing
performance for adding than for removing nodes since its rebalancing work when
adding is distributed across N nodes. When a node is removed, however, the
rebalancing work is largely distributed over N-1 nodes. In contrast, for
StaticHash and DynaHash, removing a node is more efficient than adding one
since the rebalancing work for node removal is distributed across the
remaining N-1 nodes. However, when a node is added, the new node becomes the
bottleneck because it receives data from all N-1 existing nodes.

Impact of Concurrent Writes. We further evaluated the impact of concurrent
writes on the rebalancing performance of DynaHash. In this experiment, we
rebalanced the datasets from 4 nodes to 3 nodes and inserted new records into
the LineItem dataset while rebalancing was active. We omitted StaticHash in
this experiment. These two approaches had identical behavior because they had
the same initial number of buckets and bucket splitting was disabled during
rebalancing. The resulting rebalancing time under different write rates is
shown in Figure 7(c). As one can see, rebalance takes longer to finish when
the write rate becomes larger. This is expected because these concurrent
writes compete for CPU and I/O resources with rebalance. Thus, it is desirable
to schedule rebalance operations during off-peak hours to minimize contention
with user workloads. However, as the result shows, even under high write
rates, the rebalance operation can still be completed in a reasonable amount
of time.

(a) Query Performance on 3 Nodes

(b) Query Performance on 15 Nodes

Figure 9: Query Performance on Downsized Cluster

###  VI-D Query Performance

Last but not least, we evaluated the OLAP query performance of the different
rebalancing approaches, focusing on the following three questions: First, what
is the overhead of the proposed bucketed LSM-tree structure on query
performance? Second, what is the impact on query performance of the load
balance resulting from the various rebalancing approaches? Finally, what is
the overhead due to lazy secondary index cleanup on query performance?

To answer these questions, we designed a series of experiments as follows.
First, we evaluated query performance on a cluster with 4 or 16 nodes without
rebalancing, which helps to answer the first question. We then rebalanced the
datasets down to 3 or 15 nodes so that we can evaluate the load balance impact
of the rebalancing approaches, which answers the second question. Finally for
DynaHash, we rebalanced the datasets back to 4 or 16 nodes so that we can
evaluate the performance impact of lazy secondary index cleanup (denoted as
“DynaHash-lazy-cleanup”).

The resulting query times on the original cluster size (4 or 16 nodes) are
shown in Figure 8. (Recall that the data size is scaled in proportion to the
cluster size.) Note that on 16 nodes, StaticHash and DynaHash are expected to
have similar behavior because they have the same number of buckets per
partition. In general, both StaticHash and DynaHash are seen to add a
negligible overhead on most TPC-H queries when compared with Hashing. This
shows that bucketed LSM-trees have a negligible overhead for OLAP-style
queries. Moreover, all bucketing approaches achieve very good scale-up because
the query times remain nearly constant when the number of nodes and dataset
size increase. Here one minor exception is q18, where StaticHash and DynaHash
incur a small overhead compared with Hashing. The reason is that q18 performs
a groupby on the prefix of LineItem’s primary keys, which requires the scanned
records to be ordered on the primary keys. In this case, the bucketed LSM-tree
incurs some additional overhead because it has to merge-sort more disk
components. Moreover, StaticHash also incurs a larger overhead on q18 under 4
nodes (Figure 8(a)) because it has 16 buckets per partition. Finally, we see
that lazy secondary index cleanup (DynaHash-lazy-cleanup) also has a
negligible overhead on TPC-H queries, which shows the effectiveness of this
technique. With lazy secondary index cleanup, queries need to access some
obsolete secondary index entries, but the overhead is small compared to the
overall query time.

The resulting query times on the resized cluster (3 or 15 nodes) are shown in
Figure 9. Since the number of buckets cannot be evenly divided by the number
of partitions, both StaticHash and DynaHash result in some load imbalance
where some partitions may have one more bucket than others. Despite this load
imbalance, both StaticHash and DynaHash only incur a very small overhead on
most TPC-H queries. This is because the load imbalance only impacts the data
scan time, while most TPC-H queries are relatively computation heavy. However,
for scan-heavy queries, such as q17, q18, and q21, the overhead caused by a
load imbalance is more noticeable. For example, q17 and q18 each perform a
full scan over the LineItem dataset to perform a groupby and aggregation, and
q21 scans the LineItem dataset multiple times. Moreover, as shown in Figure
9(a), using more buckets per partition as in the StaticHash approach slightly
reduces the overhead of the load imbalance for some queries, such as q21, but
doing so incurs some additional overhead for queries that require a sorted
order coming from primary index scans, such as q18.

###  VI-E Summary of Experimental Results

In general, our experimental results are consistent with the discussion in
Section II-A. Global rebalancing with hash partitioning achieves the best
ingestion and query performance, but results in a very large rebalance cost.
In contrast, DynaHash significantly reduces the rebalance time with only a
small overhead on the ingestion and query performance. The proposed bucketed
LSM-tree structure only incurs a small overhead for queries that require the
scanned records to be ordered by primary keys. Moreover, the load imbalance
caused by DynaHash mainly impacts the dataset scan performance. Overall,
DynaHash imposes a negligible overhead on computation-intensive queries and
just a small overhead on scan-heavy queries. By comparing StaticHash and
DynaHash, it can be seen that having more buckets per partition achieves a
better load balance, but it also leads to a larger overhead for queries that
require the scanned records to be ordered on primary keys. Moreover, DynaHash
requires less tuning since the number of buckets per partition is dynamically
adjusted as the cluster and dataset size scales, resulting in more stable
performance.

##  VII Conclusion

In this paper, we have described the design and implementation of DynaHash, an
efficient data rebalancing approach that combines dynamic bucketing with
extensible hashing in Apache AsterixDB. We first introduced a bucketed LSM-
tree design for efficiently storing multiple buckets. We further described an
efficient rebalancing implementation that exploits the LSM-tree’s out-of-place
update design to support concurrent reads and writes. An experimental
evaluation using the TPC-H benchmark has shown that the proposed techniques
significantly reduce the rebalance cost with negligible overheads for data
ingestion and query processing. In the future, we plan to extend DynaHash to
incorporate replication to provide better availability and fault tolerance.

## Acknowledgment

This work was supported by NSF awards CNS-1305430, IIS-1447720, IIS-1838248,
and CNS-1925610 along with industrial support from Amazon, Google, and
Microsoft and support from the Donald Bren Foundation (via a Bren Chair).

## References

  * [1] “AsterixDB,” https://asterixdb.apache.org/, 2020.
  * [2] “HBase,” https://hbase.apache.org/, 2020.
  * [3] “MongoDB,” https://www.mongodb.com/, 2020.
  * [4] “Oracle NoSQL database cloud service,” https://www.oracle.com/database/nosql-cloud.html, 2020.
  * [5] “Partition problem,” https://en.wikipedia.org/wiki/Partition_problem, 2020\.
  * [6] “Tpc-h,” http://www.tpc.org/tpch/, 2020.
  * [7] “Voltdb,” https://www.voltdb.com/, 2020.
  * [8] M. K. Aguilera _et al._ , “A practical scalable distributed B-Tree,” _PVLDB_ , vol. 1, no. 1, p. 598–609, Aug. 2008.
  * [9] M. M. Al Hubail, “Data replication and fault tolerance in asterixdb,” Master’s thesis, UC Irvine, 2016.
  * [10] W. Y. Alkowaileet _et al._ , “An LSM-based tuple compaction framework for Apache AsterixDB,” _PVLDB_ , vol. 13, no. 9, pp. 1388–1400, 2020\.
  * [11] S. Alsubaiee _et al._ , “AsterixDB: A scalable, open source BDMS,” _PVLDB_ , vol. 7, no. 14, pp. 1905–1916, 2014.
  * [12] ——, “Storage management in AsterixDB,” _PVLDB_ , vol. 7, no. 10, pp. 841–852, 2014.
  * [13] B. H. Bloom, “Space/time trade-offs in hash coding with allowable errors,” _CACM_ , vol. 13, no. 7, p. 422–426, 1970.
  * [14] H. Boral _et al._ , “Prototyping Bubba, a highly parallel database system,” _IEEE TKDE_ , vol. 2, no. 1, pp. 4–24, 1990.
  * [15] D. Borkar _et al._ , “Have your data and query it too: From key-value caching to big data management,” in _ACM SIGMOD_ , 2016, pp. 239–251.
  * [16] V. Borkar _et al._ , “Hyracks: A flexible and extensible foundation for data-intensive computing,” in _ICDE_ , 2011, pp. 1151–1162.
  * [17] M. J. Carey, “AsterixDB mid-flight: A case study in building systems in academia,” in _ICDE_ , 2019, pp. 1–12.
  * [18] F. Chang _et al._ , “Bigtable: A distributed storage system for structured data,” _ACM TOCS_ , vol. 26, no. 2, 2008.
  * [19] B. Dageville _et al._ , “The Snowflake elastic data warehouse,” in _ACM SIGMOD_ , 2016, pp. 215–226.
  * [20] S. Das _et al._ , “Albatross: Lightweight elasticity in shared storage databases for the cloud using live data migration,” _PVLDB_ , vol. 4, no. 8, pp. 494–505, May 2011.
  * [21] N. Dayan _et al._ , “Monkey: Optimal navigable key-value store,” in _ACM SIGMOD_ , 2017, pp. 79–94.
  * [22] N. Dayan and S. Idreos, “Dostoevsky: Better space-time trade-offs for LSM-tree based key-value stores via adaptive removal of superfluous merging,” in _ACM SIGMOD_ , 2018, pp. 505–520.
  * [23] ——, “The log-structured merge-bush & the Wacky continuum,” in _ACM SIGMOD_ , 2019, pp. 449–466.
  * [24] G. DeCandia _et al._ , “Dynamo: Amazon’s highly available key-value store,” in _SOSP_ , 2007, pp. 205–220.
  * [25] D. J. DeWitt _et al._ , “The Gamma database machine project,” _IEEE TKDE_ , vol. 2, no. 1, pp. 44–62, 1990.
  * [26] D. DeWitt and J. Gray, “Parallel database systems: The future of high performance database systems,” _CACM_ , vol. 35, no. 6, p. 85–98, 1992\.
  * [27] C. S. Ellis, “Extendible hashing for concurrent operations and distributed data,” in _ACM PODS_ , 1983, p. 106–116.
  * [28] A. J. Elmore _et al._ , “Zephyr: Live migration in shared nothing databases for elastic cloud platforms,” in _ACM SIGMOD_ , 2011, pp. 301–312.
  * [29] ——, “Squall: Fine-grained live reconfiguration for partitioned main memory databases,” in _ACM SIGMOD_ , 2015, pp. 299–313.
  * [30] R. Fagin _et al._ , “Extendible hashing—a fast access method for dynamic files,” _ACM TODS_ , vol. 4, no. 3, p. 315–344, Sep. 1979.
  * [31] F. Färber, S. K. Cha, J. Primsch, C. Bornhövd, S. Sigg, and W. Lehner, “SAP HANA database: Data management for modern business applications,” _SIGMOD Rec._ , vol. 40, no. 4, p. 45–51, 2012.
  * [32] R. Grover and M. Carey, “Data ingestion in asterixdb,” in _EDBT_ , 2015, pp. 605–616.
  * [33] A. Gupta _et al._ , “Amazon Redshift and the case for simpler data warehouses,” in _ACM SIGMOD_ , 2015, pp. 1917–1923.
  * [34] D. Huang _et al._ , “TiDB: A Raft-based HTAP database,” _PVLDB_ , vol. 13, no. 12, pp. 3072–3084, 2020.
  * [35] T. Kim _et al._ , “Robust and efficient memory management in Apache AsterixDB,” _Software: Practice and Experience_ , vol. 50, no. 7, pp. 1114–1151, 2020.
  * [36] B. Kröll and P. Widmayer, “Distributing a search tree among a growing number of processors,” in _ACM SIGMOD_ , 1994, p. 265–276.
  * [37] A. Lakshman and P. Malik, “Cassandra: A decentralized structured storage system,” _SIGOPS Oper. Syst. Rev._ , vol. 44, no. 2, pp. 35–40, Apr. 2010\.
  * [38] A. Lamb _et al._ , “The vertica analytic database: C-Store 7 years later,” _PVLDB_ , vol. 5, no. 12, pp. 1790–1801, 2012.
  * [39] W. Litwin _et al._ , “Lh: Linear hashing for distributed files,” in _ACM SIGMOD_ , 1993, p. 327–336.
  * [40] C. Luo, “Breaking down memory walls in LSM-based storage systems,” in _ACM SIGMOD_ , 2020, p. 2817–2819.
  * [41] C. Luo and M. J. Carey, “Efficient data ingestion and query processing for LSM-based storage systems,” _PVLDB_ , vol. 12, no. 5, pp. 531–543, 2019\.
  * [42] ——, “On performance stability in LSM-based storage systems,” _PVLDB_ , vol. 13, no. 4, pp. 449–462, 2019.
  * [43] ——, “Breaking down memory walls: Adaptive memory management in LSM-based storage systems,” _PVLDB_ , vol. 14, no. 3, pp. 241–254, 2020.
  * [44] ——, “LSM-based storage techniques: a survey,” _The VLDB Journal_ , vol. 29, no. 1, pp. 393–418, 2020.
  * [45] C. Luo _et al._ , “Umzi: Unified multi-zone indexing for large-scale HTAP,” in _EDBT_ , 2019, pp. 1–12.
  * [46] S. Luo _et al._ , “Rosetta: A robust space-time optimized range filter for key-value stores,” in _ACM SIGMOD_ , 2020, pp. 2071–2086.
  * [47] U. F. Minhas _et al._ , “Elastic scale-out for partition-based database systems,” in _ICDEW_ , 2012, pp. 281–288.
  * [48] P. O’Neil _et al._ , “The log-structured merge-tree (LSM-tree),” _Acta Inf._ , vol. 33, no. 4, pp. 351–385, 1996.
  * [49] M. A. Qader _et al._ , “A comparative study of secondary indexing techniques in LSM-based NoSQL databases,” in _ACM SIGMOD_ , 2018, pp. 551–566.
  * [50] S. Sarkar _et al._ , “Lethe: A tunable delete-aware LSM engine,” in _ACM SIGMOD_ , 2020, pp. 893–908.
  * [51] D. Schall and T. Härder, “Dynamic physiological partitioning on a shared-nothing database cluster,” in _ICDE_ , 2015, pp. 1095–1106.
  * [52] M. Stonebraker _et al._ , “C-Store: A column-oriented DBMS,” in _VLDB_ , 2005, p. 553–564.
  * [53] R. Taft _et al._ , “E-Store: Fine-grained elastic partitioning for distributed transaction processing systems,” _PVLDB_ , vol. 8, no. 3, pp. 245–256, 2014.
  * [54] ——, “CockroachDB: The resilient geo-distributed SQL database,” in _ACM SIGMOD_ , 2020, pp. 1493–1509.


A new class of special functions arising from the solution of differential
equations involving multiple proportional delays
Jayvant Patadea111Corresponding author, Sachin Bhalekarb
aAshokrao Mane group of InstitutionKolhapur - 416112, India.
bDepartment of Mathematics, Shivaji University, Kolhapur - 416004, India.
Email<EMAIL_ADDRESS><EMAIL_ADDRESS>

###### Abstract

Proportional delay is a particular case of time dependent delay. In this
article, we consider differential equations involving multiple delays. The
series solution of this equation leads to a class of special functions. This
class of special functions is independent from all the existing special
functions obtained as a solution of differential equations. We analyze the
basic properties of this class and discuss various identities and relations.

Keywords: Special functions, Successive approximation, proportional delay.

##  1 Introduction

Special functions are plying a vital role in the solutions of differential
equations. Exponential, sine, cosine, hypergeometric and Mittag-Leffler are
important classes of special functions arising as solutions of various
classical and fractional differential equations. Though there is a huge
literature devoted to the special functions arising from ordinary differential
equations (ODEs), there is a lack of corresponding literature in delay
differential equations (DDE). In [1, 2] we discussed the solutions of a class
of DDE viz. proportional delay differential equations and provided the
solutions in terms of new special functions. In this article, we generalize
the DDEs considered in [2] to involve multiple delays. Since the differential
equations without delay are inequivalent to the differential equations
involving delay, the special functions proposed in this article are
independent from the existing special functions.

##  2 Preliminaries

In this section, we discuss some basic definitions and results [3, 4].

######  Definition 2.1.

The upper and lower incomplete gamma functions are defined as

$\Gamma(n,x)=\int_{x}^{\infty}t^{n-1}e^{-t}dt\quad\textrm{and}$ (2.1)
$\gamma(n,x)=\int_{0}^{x}t^{n-1}e^{-t}dt\quad\textrm{respectively}.$ (2.2)

######  Definition 2.2.

Kummer’s confluent hyper-geometric functions ${}_{1}F_{1}(a;c;x)$ and
$U(a;c;x)$ are defined as below

${}_{1}F_{1}(a;c;x)=\sum_{n=0}^{\infty}\frac{(a)_{n}}{(c)_{n}}\frac{x^{n}}{n!},\quad
c\neq 0,-1,-2,\cdots\textrm{and}$ (2.3) $U(a;c;x)=\frac{\pi}{\sin(\pi
c)}\left(\frac{{}_{1}F_{1}(a;c;x)}{\Gamma(c)\Gamma(1+a-c)}-x^{1-c}\frac{{}_{1}F_{1}(1+a-c;2-c;x)}{\Gamma(a)\Gamma(2-c)}\right),$
(2.4) $-\pi<arg(x)\leq\pi.$

######  Definition 2.3.

The generalized Laguerre polynomials are defined as

$\displaystyle L_{n}^{(\alpha)}(x)$ $\displaystyle=$
$\displaystyle\sum_{m=0}^{n}(-1)^{m}\dbinom{n+\alpha}{n-m}\frac{x^{m}}{m!}$
(2.5) $\displaystyle=$
$\displaystyle\dbinom{n+\alpha}{n}{}_{1}F_{1}(-n;\alpha+1;x).$ (2.6)

######  Definition 2.4.

A real function $f(x)$, $x>0$, is said to be in space $C_{\alpha}$,
$\alpha\in\mathbb{R}$, if there exists a real number $p(>\alpha)$, such that
$f(x)=x^{p}f_{1}(x)$ where $f_{1}(x)\in C[0,\infty)$.

######  Definition 2.5.

A real function $f(x)$, $x>0$, is said to be in space $C^{m}_{\alpha}$,
$m\in\mathbb{N}\cup\\{0\\}$, if $f^{(m)}\in C_{\alpha}$.

######  Definition 2.6.

Let $f\in C_{\alpha}$ and $\alpha\geq-1$, then the (left-sided) Riemann-
Liouville integral of order $\mu,\mu>0$ is given by

$I^{\mu}f(t)=\frac{1}{\Gamma(\mu)}\int_{0}^{t}(t-\tau)^{\mu-1}f(\tau)d\tau,\quad
t>0.$ (2.7)

######  Definition 2.7.

The (left sided) Caputo fractional derivative of $f,f\in
C_{-1}^{m},m\in\mathbb{N}\cup\\{0\\}$, is defined as:

$\displaystyle D^{\mu}f(t)$ $\displaystyle=$
$\displaystyle\frac{d^{m}}{dt^{m}}f(t),\quad\mu=m$ (2.8) $\displaystyle=$
$\displaystyle I^{m-\mu}\frac{d^{m}}{dt^{m}}f(t),\quad{m-1}<\mu<m,\quad
m\in\mathbb{N}.$

Note that for $0\leq m-1<\alpha\leq m$ and $\beta>-1$

$\displaystyle I^{\alpha}(x-b)^{\beta}$ $\displaystyle=$
$\displaystyle\frac{\Gamma{(\beta+1)}}{\Gamma{(\beta+\alpha+1)}}(x-b)^{\beta+\alpha},$
$\displaystyle\left(I^{\alpha}D^{\alpha}f\right)(t)$ $\displaystyle=$
$\displaystyle f(t)-\sum_{k=0}^{m-1}f^{(k)}(0)\frac{t^{k}}{k!}.$ (2.9)

######  Definition 2.8.

Mittag-Leffler function of order $\alpha>0$ is defined by the series

$E_{\alpha}(x)=\sum_{n=0}^{\infty}\frac{x^{n}}{\Gamma{(\alpha n+1)}}$ (2.10)

###  2.1 Daftardar-Gejji and Jafari Method

Daftardar-Gejji and Jafari Method (DJM) [5] is one of the popular methods
applied to solve nonlinear equations of form

$u=f+L(u)+N(u),$ (2.11)

where $L$ and $N$ are linear and nonlinear operators respectively and $f$ is
known function.
In this case, the DJM provides the solution in the form of series

$u=\sum_{i=0}^{\infty}u_{i}=f+\sum_{i=0}^{\infty}L(u_{i})+\sum_{i=0}^{\infty}G_{i}$
(2.12)

where $G_{0}=N(u_{0})$ and
$G_{i}=\left\\{N\left(\sum_{j=0}^{i}u_{j}\right)-N\left(\sum_{j=0}^{i-1}u_{j}\right)\right\\}$,
$i\geq 1$.
From Eq.(2.12), the DJM series terms are generated as bellow:

$u_{0}=f,\quad u_{m+1}=L(u_{m})+G_{m},\quad m=0,1,2,\cdots.$ (2.13)

##  3 Existence, uniqueness and convergence:Nonlinear Case

First, we consider the nonlinear equation

$y^{\prime}(x)=f\left(x,y(q_{0}x),y(q_{1}x),\cdots,y(q_{n}x)\right),q_{0}=1\,\textrm{and}\,0<q_{i}<1,i=1,2,\cdots,n.$
(3.1)

The Eq. (3.1) is a particular case of time dependent delay differential
equation (DDE)

$y^{\prime}(x)=f\left(x,y\left(x-\tau_{0}(x)\right),y\left(x-\tau_{1}(x)\right),y\left(x-\tau_{2}(x)\right)\cdots
y\left(x-\tau_{n}(x)\right)\right)$

with $\tau_{i}(x)=(1-q_{i})x,\quad i=1,2,3,\cdots,n$. The DJM series solution
of Eq. (3.1) is of the form

$y=\sum_{i=0}^{\infty}y_{i}.$ (3.2)

We present convergence result of this series solution motivated from [6].

######  Theorem 3.1.

Let f be a continuous function defined on a $(n+2)$ dimensional rectangle
$R=\\{(x,y_{0},y_{1},\cdots,y_{n})|0\leq x\leq b,-\delta_{i}\leq
y_{i}\leq\delta_{i},i=0,1,2,\cdots,n\\}$ and $\mid f\mid\leq M$ on $R$.
Suppose that f satisfies Lipschitz type condition

$\mid
f\left(x,y_{0},y_{1},\cdots,y_{n}\right)-f\left(x,u_{0},u_{1},\cdots,u_{n}\right)\mid\leq\sum_{i=0}^{n}L_{i}\mid
y_{i}-u_{i}\mid.$ (3.3)

Then the DJM series solution (3.2) of DDE (3.1) converges uniformly in the
interval [0,b].

###### Proof.

Without loss of gerenality, assume that $y(0)=1$. The equivalent integral
equation of (3.1) is

$y(x)=1+\int_{0}^{x}f\left(x,y(q_{0}t),y(q_{1}t),y(q_{2}t),\cdots,y(q_{n}t)\right)dt.$

Using DJM, we get

$\displaystyle y_{0}(x)$ $\displaystyle=$ $\displaystyle 1,$ $\displaystyle
y_{1}(x)$ $\displaystyle=$
$\displaystyle\int_{0}^{x}f\left(t,y_{0}(q_{0}t),y_{0}(q_{1}t),y_{0}(q_{2}t),\cdots,y_{0}(q_{n}t)\right)dt.$
$\displaystyle\Rightarrow\mid y_{1}(x)\mid$ $\displaystyle\leq$ $\displaystyle
Mx.$

$\textrm{Since}\;q_{0}=1,0<q_{i}<1,i=1,2,3,\cdots,n.,\;q_{i}b\leq b.$

$\displaystyle\Rightarrow\mid y_{1}(q_{i}x)\mid$ $\displaystyle\leq$
$\displaystyle Mq_{i}x,\quad\forall x\in[0,b].$ $\displaystyle y_{2}(x)$
$\displaystyle=$
$\displaystyle\int_{0}^{x}\left(f\left(t,y_{1}(q_{0}t)+y_{0}(q_{0}t),y_{1}(q_{1}t)+y_{0}(q_{1}t),y_{1}(q_{2}t)+y_{0}(q_{2}t),\cdots,y_{1}(q_{n}t)+y_{0}(q_{n}t)\right)\right.$
$\displaystyle\left.-f\left(t,y_{0}(q_{0}t),y_{0}(q_{1}t),y_{0}(q_{2}t),\cdots,y_{0}(q_{n}t)\right)\right)dt.$
$\displaystyle\Rightarrow\mid y_{2}(x)\mid$ $\displaystyle\leq$
$\displaystyle\int_{0}^{x}\left(\sum_{i=0}^{n}L_{i}\mid
y_{1}(q_{i}t)\mid\right)dt$ $\displaystyle\leq$ $\displaystyle
M\left(\sum_{i=0}^{n}L_{i}q_{i}\right)\frac{x^{2}}{2!}$ $\displaystyle\leq$
$\displaystyle M\left(\sum_{i=0}^{n}L_{i}\right)\frac{x^{2}}{2!}.$
$\displaystyle\Rightarrow\mid y_{2}(q_{i}x)\mid$ $\displaystyle\leq$
$\displaystyle
Mq_{i}^{2}\left(\sum_{i=0}^{n}L_{i}q_{i}\right)\frac{x^{2}}{2!},\quad
x\in[0,b]$ $\displaystyle\leq$ $\displaystyle
M\left(\sum_{i=0}^{n}L_{i}\right)\frac{x^{2}}{2!}.$

Using induction, we get

$\displaystyle\mid y_{m}(x)\mid$ $\displaystyle\leq$ $\displaystyle
M\prod_{j=1}^{m-1}\left(\sum_{i=0}^{n}L_{i}q^{j}_{i}\right)\frac{x^{m}}{m!}$
$\displaystyle\leq$ $\displaystyle
M\left(\sum_{i=0}^{n}L_{i}\right)^{m-1}\frac{x^{m}}{m!},\quad m=1,2,3\cdots.$

Taking summation over $m$, we get

$\displaystyle\left|\sum_{m=0}^{\infty}y_{m}\right|$ $\displaystyle\leq$
$\displaystyle\frac{M}{\sum_{i=0}^{n}L_{i}}\left(e^{\sum_{i=0}^{n}L_{i}x}-1\right)+1$
$\displaystyle\leq$
$\displaystyle\frac{Me^{\sum_{i=0}^{n}L_{i}b}}{\sum_{i=0}^{n}L_{i}}+1,\quad
x\in[0,b].$

Thus, by [7], we can conclude that the series solution of (3.1) converges
uniformly in the interval [0,b]. Hence existence of (3.1) is proved. The
uniqueness is of solution is obvious by condition (3.3). ∎

##  4 Stability analysis

The following definitions and theorems are generalization of corresponding
definition and theeorems given in [8].

######  Definition 4.1.

Consider the autonomus time-dependent delay differential equation (DDE),

$y^{\prime}(x)=g(y\left(x\right),y\left(x-\tau_{1}(x)\right),y\left(x-\tau_{2}(x)\right)\cdots
y\left(x-\tau_{n}(x)\right)),$ (4.1)

where $g:\mathbb{R}^{n+1}\rightarrow\mathbb{R}$. The flow $\phi_{x}(x_{0})$ is
the solution $y(x)$ of (4.1) with initial condition
$y(x)=x_{0},\,x\leq 0$. The point $y^{*}$ is called equilibrium solution of
(4.1) if $g(y^{*},y^{*},\cdots,y^{*})=0$.
(a) If, for any $\epsilon>0$, there exist $\delta>0$ such that
$|x_{0}-y^{*}|<\delta\Rightarrow|\phi_{x}(x_{0})-y^{*}|<\epsilon,$ then the
system (4.1) is stable (in the Lyapunov sense) at the equilibrium $y^{*}$.
(b) If the system (4.1) is stable at $y^{*}$ and
moreover,$\lim\limits_{x\rightarrow\infty}|\phi_{x}(x_{0})-y^{*}|=0$ then the
system (4.1) is said to be asymptotically stable at $y^{*}$.
(c) If the system (4.1) is not stable then it is called unstable.

######  Theorem 4.1.

Assume that the equilibrium solution $y^{*}$ of the equation

$y^{\prime}=g(y(x),y(x-\tau_{1}^{*}),y(x-\tau_{2}^{*}),\cdots,y(x-\tau_{n}^{*})),\quad\tau_{1}^{*}=\tau_{1}(x_{0}),\tau_{2}^{*}=\tau_{2}(x_{0}),\cdots,\tau_{n}^{*}=\tau_{n}(x_{0})$

is stable at equilibrium $y^{*}$ and

$\|g(y(x),y(x-\tau_{1}(x)),\cdots,y(x-\tau_{n}(x)))-g(y(x),y(x-\tau_{1}(x_{1})),\cdots,y(x-\tau_{n}(x_{n})))\|<\sum_{i=0}^{n}\epsilon_{i}|x-x_{i}|,$

for some $\epsilon_{i}>0$ and $x,x_{i}\in[x_{0},x_{0}+c)$$(i=1,2,\cdots,n)$, c
is a positive constant, then there exists $\bar{x}>0$ such that the
equilibrium solution $y^{*}$ of Eq. (4.1) is stable at equilibrium $y^{*}$ on
finite time interval $[x_{0},\bar{x})$.

######  Corollary 1.

If the real parts of all roots of
$\lambda-\sum_{i=0}^{n}a_{i}e^{-\lambda\tau_{i}^{*}}$ are negative, where
$a_{i}=\partial_{i}f$, $i=1,2,\cdots,n$ evaluated at equilibrium $y^{*}$, then
Eq. (4.1) is stable at $y^{*}$ on finite time interval $[x_{0},\bar{x})$.

##  5 Linear equation: Exact solution

Consider the differential equation involving multiple proportional delays,

$y^{\prime}(x)=\sum_{i=0}^{n}a_{i}y(q_{i}x),\quad y(0)=1,$ (5.1)

where $q_{0}=1$, $0<q_{i}<1$ and $a_{0},a_{i}\in\mathbb{R}$,
$i=1,2,3,\cdots,n$. The equation (5.1) has applications in Science and
Engineering [9, 10].
Integrating (5.1), we get

$y(x)=1+\int_{0}^{x}\left(\sum_{i=0}^{n}a_{i}y(q_{i}t)\right)dt$

Using successive approximation, we obtain

$\displaystyle y_{0}(x)$ $\displaystyle=$ $\displaystyle 1,$ $\displaystyle
y_{1}(x)$ $\displaystyle=$
$\displaystyle\int_{0}^{x}\left(\sum_{i=0}^{n}a_{i}y_{0}(q_{i}t)\right)dt$
$\displaystyle=$ $\displaystyle\sum_{i=0}^{n}a_{i}\frac{x}{1!},$
$\displaystyle y_{2}(x)$ $\displaystyle=$
$\displaystyle\int_{0}^{x}\left(\sum_{i=0}^{n}a_{i}y_{1}(q_{i}t)\right)dt$
$\displaystyle=$
$\displaystyle\int_{0}^{x}\sum_{i=0}^{n}a_{i}\left(\sum_{i=0}^{n}a_{i}q_{i}t\right)dt$
$\displaystyle=$
$\displaystyle\left(\sum_{i=0}^{n}a_{i}\right)\left(\sum_{i=0}^{n}a_{i}q_{i}\right)\frac{x^{2}}{2!},$
$\displaystyle y_{3}(x)$ $\displaystyle=$
$\displaystyle\int_{0}^{x}\left(\sum_{i=0}^{n}a_{i}y_{2}(q_{i}t)\right)dt$
$\displaystyle=$
$\displaystyle\left(\sum_{i=0}^{n}a_{i}\right)\left(\sum_{i=0}^{n}a_{i}q_{i}\right)\left(\sum_{i=0}^{n}a_{i}q^{2}_{i}\right)\frac{x^{3}}{3!},$
$\displaystyle\vdots$

$\displaystyle y_{m}(x)$ $\displaystyle=$
$\displaystyle\frac{x^{m}}{m!}\prod_{j=0}^{m-1}\left(\sum_{i=0}^{n}a_{i}q^{j}_{i}\right),\quad
m=1,2,3\cdots.$

$\therefore$ The exact solution of (5.1) is

$\displaystyle y(x)$ $\displaystyle=$ $\displaystyle
y_{0}(x)+y_{1}(x)+y_{2}(x)+\cdots$ (5.2) $\displaystyle=$ $\displaystyle
1+\sum_{m=1}^{\infty}\frac{x^{m}}{m!}\prod_{j=0}^{m-1}\left(\sum_{i=0}^{n}a_{i}q^{j}_{i}\right).$

If we define

$\prod_{j=0}^{m-1}\left(\sum_{i=0}^{n}a_{i}q^{j}_{i}\right)=1\quad\textrm{for}\quad
m=0,$

then

$y(x)=\sum_{m=0}^{\infty}\frac{x^{m}}{m!}\prod_{j=0}^{m-1}\left(\sum_{i=0}^{n}a_{i}q^{j}_{i}\right)$
(5.3)

This solution of (5.1) provides a novel special function

$\mathcal{R}(\bar{a};\bar{q};x)=\sum_{m=0}^{\infty}\frac{x^{m}}{m!}(\bar{a};\bar{q})_{m}.$
(5.4)

Note: We use a brief notations $\mathcal{R}(\bar{a};\bar{q};x)$ for the
special function
$\mathcal{R}(a_{0},a_{1},\cdots,a_{n};q_{0},q_{1},\cdots,q_{n},x)$ and
$(\bar{a};\bar{q})_{m}$ for
$\prod_{j=0}^{m-1}\left(\sum_{i=0}^{n}a_{i}q^{j}_{i}\right)$.

##  6 Analysis

######  Theorem 6.1.

The power series

$\mathcal{R}(\bar{a};\bar{q};x)=\sum_{m=0}^{\infty}\frac{x^{m}}{m!}(\bar{a};\bar{q})_{m},$
(6.1)

has infinite radius of convergence.

###### Proof.

Suppose

$A_{m}=\frac{1}{m!}(\bar{a};\bar{q})_{m},\quad m=0,1,2,\cdots.$

If $R$ is radius of convergence of (6.1) then by using ratio test [11]

$\displaystyle\frac{1}{R}=\lim_{m\to\infty}\left|\frac{A_{m+1}}{A_{m}}\right|$
$\displaystyle=$
$\displaystyle\lim_{m\to\infty}\left|\frac{\frac{1}{{(m+1)!}}\prod_{j=0}^{m}\left(\sum_{i=0}^{n}a_{i}q^{j}_{i}\right)}{\frac{1}{m!}\prod_{j=0}^{m-1}\left(\sum_{i=0}^{n}a_{i}q^{j}_{i}\right)}\right|$
$\displaystyle=$
$\displaystyle\lim_{m\to\infty}\left|\frac{\left(\sum_{i=0}^{n}a_{i}q^{m}\right)}{(m+1)}\right|$
$\displaystyle\Rightarrow R$ $\displaystyle=$ $\displaystyle\infty.$

Thus the series has infinite radius of convergence. ∎

######  Corollary 2.

The power series (5.4) is absolutely convergent for all $x$ and hence it is
uniformly convergent on any compact interval on $\mathbb{R}$.

######  Theorem 6.2.

For $r\in\mathbb{N}\cup\\{0\\}$, we have

$\frac{d^{r}}{dx^{r}}\mathcal{R}(\bar{a};\bar{q};x)=\mathcal{R}^{(r)}(\bar{a};\bar{q};x)=\sum_{m=r}^{\infty}\frac{x^{m-r}}{(m-r)!}(\bar{a};\bar{q})_{m}.\\\
$

######  Theorem 6.3.

(Addition Theorem)

$\mathcal{R}(\bar{a};\bar{q};x+y)=\sum_{r=0}^{\infty}\frac{x^{r}}{r!}\mathcal{R}^{(r)}(\bar{a};\bar{q};y)$

###### Proof.

We have

$\displaystyle\mathcal{R}(\bar{a};\bar{q};x+y)$ $\displaystyle=$
$\displaystyle\sum_{m=0}^{\infty}\frac{(x+y)^{m}}{m!}(\bar{a};\bar{q})_{m}$
$\displaystyle=$
$\displaystyle\sum_{m=0}^{\infty}\sum_{r=0}^{m}\frac{x^{r}}{r!}\frac{y^{m-r}}{(m-r)!}(\bar{a};\bar{q})_{m}$
$\displaystyle=$
$\displaystyle\sum_{r=0}^{\infty}\sum_{m=r}^{\infty}\frac{x^{r}}{r!}\frac{y^{m-r}}{(m-r)!}(\bar{a};\bar{q})_{m}$
$\displaystyle=$
$\displaystyle\sum_{r=0}^{\infty}\frac{x^{r}}{r!}\sum_{m=r}^{\infty}\frac{y^{m-r}}{(m-r)!}(\bar{a};\bar{q})_{m}$
$\displaystyle\mathcal{R}(\bar{a};\bar{q};x+y)$ $\displaystyle=$
$\displaystyle\sum_{r=0}^{\infty}\frac{x^{r}}{r!}\mathcal{R}^{(r)}(\bar{a};\bar{q};y).$

∎

Note: If $x+iy=z\in\mathbb{C}$, then

$\mathcal{R}(\bar{a};\bar{q};z)=\sum_{r=0}^{\infty}\frac{(-ix)^{r}}{r!}\mathcal{R}^{(r)}(\bar{a};\bar{q};iy)$

######  Theorem 6.4.

If $z=re^{i\theta}\in\mathbb{C}$, then

$\mathcal{R}(\bar{a};\bar{q};z)=\mathcal{R}_{c}(\bar{a};\bar{q};z)+i\mathcal{R}_{s}(\bar{a};\bar{q};z),$

where,

$\displaystyle\mathcal{R}_{c}(\bar{a};\bar{q};z)$ $\displaystyle=\textrm{Real
part of\, }\mathcal{R}(\bar{a};\bar{q};z)$
$\displaystyle=\sum_{m=0}^{\infty}\frac{r^{m}\cos\theta}{m!}(\bar{a};\bar{q})_{m}$
$\displaystyle\mathcal{R}_{s}(\bar{a};\bar{q};z)$
$\displaystyle=\textrm{Imaginary part of\, }\mathcal{R}(\bar{a};\bar{q};z)$
$\displaystyle=\sum_{m=0}^{\infty}\frac{r^{m}\sin\theta}{m!}(\bar{a};\bar{q})_{m},$

######  Theorem 6.5.

For $q_{0}=1$, $0<q_{i}<1$ and $a_{0},a_{i}\geq 0$, $i=0,1,2,3,\cdots,n$. The
function $\mathcal{R}(\bar{a};\bar{q};x)$ satisfies the following inequality

$e^{a_{0}x}\leq\mathcal{R}(\bar{a};\bar{q};x)\leq
e^{\left(\sum_{i=0}^{n}a_{i}\right)x},\quad 0\leq x<\infty.$

###### Proof.

Since $q_{0}=1$, $0<q_{i}<1$ and $a_{0},a_{i}\geq 0$, $i=0,1,2,3,\cdots,n$, we
have

$\displaystyle\prod_{j=0}^{m-1}\left(\sum_{i=0}^{n}a_{i}q^{j}_{i}\right)$
$\displaystyle\leq$ $\displaystyle\left(\sum_{i=0}^{n}a_{i}\right)^{m}$
$\displaystyle\Rightarrow\frac{x^{m}}{m!}\prod_{j=0}^{m-1}\left(\sum_{i=0}^{n}a_{i}q^{j}_{i}\right)$
$\displaystyle\leq$
$\displaystyle\frac{x^{m}\left(\sum_{i=0}^{n}a_{i}\right)^{m}}{m!}.$ Taking
summation over m, we get $\displaystyle\mathcal{R}(\bar{a};\bar{q};x)\leq
e^{\left(\sum_{i=0}^{n}a_{i}\right)x},\quad 0\leq x<\infty.$ (6.2)

Similarly, we have

$\displaystyle a_{0}^{m}$ $\displaystyle\leq$
$\displaystyle\prod_{j=0}^{m-1}\left(\sum_{i=0}^{n}a_{i}q^{j}_{i}\right)$
$\displaystyle\Rightarrow e^{a_{0}x}$ $\displaystyle\leq$
$\displaystyle\mathcal{R}(\bar{a};\bar{q};x),\quad 0\leq x<\infty.$ (6.3)

From (6.2) and (6.3), we get

$e^{a_{0}x}\leq\mathcal{R}(\bar{a};\bar{q};x)\leq
e^{\left(\sum_{i=0}^{n}a_{i}\right)x},\quad 0\leq x<\infty.$

∎

##  7 Generalization to fractional order DDE

Consider the fractional delay differential equation involving multiple
proportional delays,

$D^{\alpha}_{0}y(x)=\sum_{i=0}^{n}a_{i}y(q_{i}x),\quad y(0)=1,$ (7.1)

where $q_{0}=1$, $0<q_{i}<1$ and $a_{0},a_{i}\in\mathbb{R}$,
$i=1,2,3,\cdots,n$.

The exact solution of (7.1) is

$\mathcal{R}_{\alpha}(\bar{a};\bar{q};x)=\sum_{m=0}^{\infty}\frac{x^{\alpha
m}}{\Gamma{(\alpha m+1)}}(\bar{a};\bar{q})_{\alpha,m}$

where
$(\bar{a};\bar{q})_{\alpha,m}=\prod_{j=0}^{m-1}\left(\sum_{i=0}^{n}a_{i}q^{\alpha
j}_{i}\right)$

######  Theorem 7.1.

The power series (7) is convergent for all finite values of $x$.

######  Theorem 7.2.

For $q_{0}=1$, $0<q_{i}<1$ and $a_{0},a_{i}\geq 0$, $i=0,1,2,3,\cdots,n$. The
function $\mathcal{R}_{\alpha}(\bar{a};\bar{q};x)$ satisfies the following
inequality

$E_{\alpha}{(a_{0}x^{\alpha})}\leq\mathcal{R}_{\alpha}(\bar{a};\bar{q};x)\leq
E_{\alpha}{\left(\left(\sum_{i=0}^{n}a_{i}\right)x^{\alpha}\right)},\quad
0\leq x<\infty.$

##  8 Conclusions

In this paper, we have obtained a new special function arising from
differential equation involving multiple proportional delays. The solution is
obtained by applying the successive approximation method. The existence,
uniqueness, stability and convergence results for the time dependent delay
differential equations are presented in this paper. The new special function
exhibit different properties and relations. The generalization to fractional
order case is also presented.

Acknowledgements:
S. Bhalekar acknowledges the Science and Engineering Research Board (SERB),
New Delhi, India for the Research Grant (Ref. MTR/2017/000068) under
Mathematical Research Impact Centric Support (MATRICS) Scheme.

## References

  * [1] Bhalekar, S., Patade, J. Series Solution of the Pantograph Equation and Its Properties. Fractal and Fractional, 1(1) (2017), p.16.
  * [2] Bhalekar, S., Patade, J. Analytical Solution of Pantograph Equation with Incommensurate Delay. Physical Sciences Reviews, (2017) 2(9).
  * [3] Magnus, Wilhelm, Fritz Oberhettinger, Raj Soni.: Formulas and theorems for the special functions of mathematical physics. Springer Science & Business Media 52, (2013)
  * [4] Kilbas, A.A., Srivastava, H.M., Trujillo, J.J.: Theory and Applications of Fractional Differential Equations. Elsevier, Amsterdam (2006)
  * [5] V. Daftardar-Gejji and H. Jafari, An iterative method for solving non linear functional equations, J. Math. Anal. Appl., 316 (2006) 753–763.
  * [6] Bhalekar, S., Patade, J. Analytical solutions of nonlinear equations with proportional delays, Appl. Comput. Math., 15(3) (2016) 331–345.
  * [7] Rudin, W. Principles of mathematical analysis, New York, McGraw-Hill, 1964.
  * [8] Deng W, Wu Y, Li C. Stability analysis of differential equations with time-dependent delay. Int J Bifurcation Chaos. 2006;16(02):465–472.
  * [9] J. Patade, S. Bhalekar, On Analytical Solution of Ambartsumian Equation. Natl. Acad. Sci. Lett. 40(4) (2017) 291–293.
  * [10] Ockendon. J., Tayler A. B. The dynamics of a current collection system for an electric locomotive.” Proceedings of the Royal Society of London A: Mathematical, Physical and Engineering Sciences, The Royal Society. 322(1551) (1971)
  * [11] Apostal, T. M.: Mathematical analysis. Reading: Addison-Wesley (1964)