Squall：Fine-Grained Live Reconfiguration for Partitioned Main Memory Databases

Squall:Fine-Grained Live Reconfiguration for Partitioned Main Memory Databases Aaron J.Elmore',Vaibhav Arora2,Rebecca Taft Andrew Pavlo,Divyakant Agrawal2.s,Amr El Abbadi wrcSgne8a8mag1Baansne aeimorecschcsb.. ABSTRACT vith the abilit udden shifts This whe e is de (OLTP) high performance and de spre ons.H o h In these o reco overall DBMS.M e ale databases in thi n,we introduce the Squall techniqu a node l ts fin ng of ohpgadinghan vare licated data having to shut down 1.INTRODUCTION disk-ha ar.they are pre The nced for scalable main-memory DBMSs ismo vated by de This me rent data ns.Such systems eschew leg s19,351 tional DBMSs 37]and alleviate the ntion on shared data the DBMS ep train o more,modern web/mobile applications require always-available data ches 137 f all ssor physical loggng full cit the f h or specific pe issions@acm.平 then instead on to a new nod a rights licensed to ACM ut)approach dynamic allows the system to continue to process it mirate

Squall: Fine-Grained Live Reconfiguration for Partitioned Main Memory Databases Aaron J. Elmore1 , Vaibhav Arora2 , Rebecca Taft3 Andrew Pavlo4 , Divyakant Agrawal2,5 , Amr El Abbadi2 1University of Chicago, 2University of California, Santa Barbara 3MIT CSAIL, 4Carnegie Mellon University, 5Qatar Computing Research Institute aelmore@cs.uchicago.edu, {vaibhavarora,agrawal,amr}@cs.ucsb.edu, rytaft@mit.edu, pavlo@cs.cmu.edu ABSTRACT For data-intensive applications with many concurrent users, mod￾ern distributed main memory database management systems (DBMS) provide the necessary scale-out support beyond what is possible with single-node systems. These DBMSs are optimized for the short-lived transactions that are common in on-line transaction pro￾cessing (OLTP) workloads. One way that they achieve this is to partition the database into disjoint subsets and use a single-threaded transaction manager per partition that executes transactions one-at￾a-time in serial order. This minimizes the overhead of concurrency control mechanisms, but requires careful partitioning to limit dis￾tributed transactions that span multiple partitions. Previous meth￾ods used off-line analysis to determine how to partition data, but the dynamic nature of these applications means that they are prone to hotspots. In these situations, the DBMS needs to reconfigure how data is partitioned in real-time to maintain performance ob￾jectives. Bringing the system off-line to reorganize the database is unacceptable for on-line applications. To overcome this problem, we introduce the Squall technique for supporting live reconfiguration in partitioned, main memory DBMSs. Squall supports fine-grained repartitioning of databases in the presence of distributed transactions, high throughput client workloads, and replicated data. An evaluation of our approach on a distributed DBMS shows that Squall can reconfigure a database with no downtime and minimal overhead on transaction latency. 1. INTRODUCTION The need for scalable main-memory DBMSs is motivated by de￾creasing memory costs and an increasing demand for high-throughput transaction processing systems. Such systems eschew legacy disk￾oriented concurrency control and recovery mechanisms of tradi￾tional DBMSs [37] and alleviate the contention on shared data￾structures [17,26,31,40]. But some OLTP databases are larger than the amount of DRAM that is available on a single node. Further￾more, modern web/mobile applications require always-available data Permission to make digital or hard copies of all or part of this work for personal or classroom use is granted without fee provided that copies are not made or distributed for profit or commercial advantage and that copies bear this notice and the full cita￾tion on the first page. Copyrights for components of this work owned by others than ACM must be honored. Abstracting with credit is permitted. To copy otherwise, or re￾publish, to post on servers or to redistribute to lists, requires prior specific permission and/or a fee. Request permissions from permissions@acm.org. SIGMOD’15, May 31–June 4, 2015, Melbourne, Victoria, Australia. Copyright is held by the owner/author(s). Publication rights licensed to ACM. ACM 978-1-4503-2758-9/15/05 ...$15.00. http://dx.doi.org/10.1145/2723372.2723726. with the ability to support sudden shifts in access patterns. The in￾ability to react to changes in usage or mitigate potential downtime can cause significant financial losses for service providers [7]. This argues for the use of a distributed DBMS architecture where the database is deployed in memory-only storage on a cluster of shared-nothing nodes. These scalable DBMSs, colloquially known as NewSQL [10], achieve high performance and scalability with￾out sacrificing the benefits of strong transactional guarantees by spreading the databases across nodes into disjoint partitions. Re￾cent examples of these systems include H-Store [24] (and its com￾mercial version VoltDB [6]), MemSQL [2], and SQLFire [5]. Even if a database resides entirely in memory across multiple partitions, the DBMS is not immune to problems resulting from changes in workload demands or access patterns. For example, sud￾den increases in the popularity of a particular item in the database can negatively impact the performance of the overall DBMS. Mod￾ern distributed systems can, in theory, add and remove resources dynamically, but in practice it is difficult to scale databases in this manner [23]. Increasing system capacity involves either scaling up a node by upgrading hardware or scaling out by adding additional nodes to the system in order to distribute load. Either scenario can involve migrating data between nodes and bringing nodes off-line during maintenance windows [18]. Previous work has shown how to migrate a database from one node to another incrementally to avoid having to shut down the system [8, 15, 19, 35]. These approaches, however, are not suitable for partitioned main-memory DBMSs. In particular, they are pred￾icated upon disk-based concurrency control and recovery mech￾anisms. This means that they rely on concurrent data access to migrate data using heavy-weight two-phase locking and snapshot isolation methods to ensure correctness [19, 35]. More proactive migration techniques rely on physical logging from standard repli￾cation and recovery mechanisms [8, 15]. But this dependency on the DBMS’s replication infrastructure places additional strain on partitions that may be already overloaded. Above all, these ap￾proaches are inappropriate for partitioned main-memory DBMSs that execute transactions serially [37], since the system does not support concurrent access or physical logging. Even if one adapted these approaches to move partitions between nodes, they are still not able to split one partition into multiple par￾titions. For example, if there is a particular entity in a partition that is extremely popular (e.g., the Wu Tang Clan’s Twitter account), then instead of migrating the entire partition to a new node it is better to move those individual tuples to their own partition [38]. A better (but more difficult) approach is to dynamically reconfig￾ure the physical layout of the database while the system is live. This allows the system to continue to process transactions as it migrates

H-Store is optimized for the efficient execution of workload H-Store Cluster Eacontain -Store Nod ind (2)contr e the I input pa ecutes is s its bas 1331.The base o that is itecture from 33 le th Figure 1:The H-Sto tis responsible for executing tran quenes for a-time hased on the order of the ival ti nce 201.But accomplishing this ise when the DBMS action tha granted the lock and (b)it ha 3 t lea ures that distributed t sume tha ard clock-skew s CP ach tage nt the Squall ese sys al DBMS e da of th approach.h ntigurationbyafeyint slow.If a tr hat reduc DBMS ed th system's perfoma ance usin then res re a DB in thro tion t for short-lived u ns 12 hitecture ctions' and persistent. of data s.the DBMS fault tol e is en d in S nd hich ition is fulb work in Section 8.and conclude in Section 9 2.BACKGROUND 2.2 Database Partitioning et in thiswork ms (i.e replication is 2.1 H-Store Architecture othe H-Store is a distributed,row-oriented DBMS that supports se partition from its that manages one or more partitions range.or round-robin partitioning 6).For this paper.we modi-

H-Store Node BadAss Chip 3000 Squirrels. Yes, Squirrels BadAss Chip 3000 Squirrels. Yes, Squirrels ... Partition Data Partition Data Execution Engine Execution Engine Txn Coordinator H-Store Cluster Client Application Core Core Procedure Name Input Parameters BadAss Chip 3000 Squirrels. Yes, Squirrels BadAss Chip 3000 Squirrels. Yes, Squirrels ... BadAss Chip 3000 Squirrels. Yes, Squirrels BadAss Chip 3000 Squirrels. Yes, Squirrels ... BadAss Chip 3000 Squirrels. Yes, Squirrels BadAss Chip 3000 Squirrels. Yes, Squirrels ... BadAss Chip 3000 Squirrels. Yes, Squirrels BadAss Chip 3000 Squirrels. Yes, Squirrels ... BadAss Chip 3000 Squirrels. Yes, Squirrels BadAss Chip 3000 Squirrels. Yes, Squirrels ... BadAss Chip 3000 Squirrels. Yes, Squirrels BadAss Chip 3000 Squirrels. Yes, Squirrels ... Figure 1: The H-Store architecture from [33]. data and immediately relieves contention on hotspots. Some dis￾tributed NoSQL DBMSs, such as MongoDB [3], support splitting and migration of partitions to new nodes when the system needs to re-balance [20]. But accomplishing this is easier when the DBMS does not support atomic operations on multiple objects. Another approach is to pre-allocate multiple “virtual” partitions for each real partition at start-up and then migrate some of the virtual partitions to new nodes for re-balancing the load [34]. The downside of this is that the DBMS has no control of the contents of these virtual parti￾tions, and thus there is no way to know whether the migration will result in the desired change in performance until after the virtual partitions have been migrated. To the best of our knowledge, no DBMS today supports the fine-grained, tuple-level load balancing that is needed for the system to be truly autonomous. Given the lack of solutions for this problem, we present the Squall migration system for partitioned OLTP DBMSs. Our key contribu￾tion in this paper is an efficient mechanism for performing fine￾grained live reconfiguration by safely interleaving data migration with executing transactions. We also present several optimizations that reduce the migration costs for a variety of workloads. To evalu￾ate our work, we implemented Squall in the H-Store [1] DBMS and measured the system’s performance using two OLTP workloads. Our results demonstrate that Squall is able to reconfigure a DBMS with no downtime and a minimal decrease in throughput. The rest of the paper is organized as follows. We start in Sec￾tion 2 with an overview of the DBMS architecture targeted by our work and the reconfiguration scenarios that are important in this operating environment. We then present Squall in Section 3, dis￾cuss the details of data reconfiguration management in Section 4, and outline several optimizations in Section 5. Section 6 outlines how fault tolerance is enabled in Squall. We then provide a thor￾ough evaluation of Squall in Section 7. Finally, we describe related work in Section 8, and conclude in Section 9. 2. BACKGROUND We begin with an overview of the architecture of H-Store, an ex￾ample of the type of distributed DBMS that we target in this work. We then show how these DBMSs are susceptible to load imbalances and how a naïve migration approach to reconfiguring a database is insufficient. Although we use H-Store in our analysis, our work is applicable to any partitioned, main memory OLTP DBMS. 2.1 H-Store Architecture H-Store is a distributed, row-oriented DBMS that supports se￾rializable execution of transactions over main memory partitions. We define an H-Store instance as a cluster of two or more nodes de￾ployed within the same administrative domain. A node is a single physical computer system that contains a transaction coordinator that manages one or more partitions. H-Store is optimized for the efficient execution of workloads that contain transactions invoked as pre-defined stored procedures. Each stored procedure is comprised of (1) parameterized queries and (2) control code that contains application logic intermixed with invocations of those queries. Client applications initiate transac￾tions by sending the procedure name and input parameters to any node in the cluster. The partition where the transaction’s control code executes is known as its base partition [33]. The base par￾tition ideally will have most (if not all) of the data the transaction needs [32]. Any other partition involved in the transaction that is not the base partition is referred to as a remote partition. As shown in Fig. 1, each partition is assigned a single-threaded execution engine that is responsible for executing transactions and queries for that partition. A partition is protected by a single lock managed by its coordinator that is granted to transactions one-at￾a-time based on the order of their arrival timestamp [9, 13, 41]. A transaction acquires a partition’s lock if (a) the transaction has the lowest timestamp that is not greater than the one for the last trans￾action that was granted the lock and (b) it has been at least 5 ms since the transaction first entered the system [37]. This wait time ensures that distributed transactions that send their lock acquisition messages over the network to remote partitions are not starved. We assume that standard clock-skew algorithms are used to keep the nodes’ CPU clocks synchronized. Executing transactions serially at each partition has several ad￾vantages for OLTP workloads. In these applications, most transac￾tions only access a single entity in the database. That means that these systems are much faster than a traditional DBMS if the data￾base is partitioned in such a way that most transactions only access a single partition [32]. The downside of this approach, however, is that transactions that need to access data at two or more partitions are slow. If a transaction attempts to access data at a partition that it does not have the lock for, then the DBMS aborts that transaction (releasing all of the locks that it holds), reverts any changes, and then restarts it once the transaction re-acquires all of the locks that it needs again. This removes the need for distributed deadlock detec￾tion, resulting in better throughput for short-lived transactions [22]. All data in H-Store is stored in main memory. To ensure that transactions’ modifications are durable and persistent, each node writes asynchronous snapshots of the entire database to disk at fixed intervals [25,37]. In between these snapshots, the DBMS writes out a record to a redo-only command log for each transaction that com￾pletes successfully [27]. In addition to snapshots and command logging, main memory databases often use replication to provide durability and high availability. Each partition is fully replicated by another secondary partition that is hosted on a different node [27]. 2.2 Database Partitioning A partition plan for a database in a distributed OLTP DBMS is comprised of (1) partitioned tables, (2) replicated tables, and (3) transaction routing parameters [32]. A table can be horizontally partitioned into disjoint fragments whose boundaries are based on the values of one (or more) of the table’s columns (i.e., partitioning attributes). Alternatively, the DBMS can replicate non-partitioned tables across all partitions. This table-level replication is useful for read-only or read-mostly tables that are accessed together with other tables but do not partition in accordance with other tables. A transaction’s routing parameters identify the transaction’s base partition from its input parameters. Administrators deploy databases using a partition plan that min￾imizes the number of distributed transactions by collocating the records that are used together often in the same partition [14, 32]. A plan can be implemented in several ways, such as using hash, range, or round-robin partitioning [16]. For this paper, we modi-

ter.We postpone the details of these workloads and the execution E-Storeus s system-level statistics (e.g sustained high E-Store data.showing WAREHOUSE and CUSTOMER PC-C te ed me es experien ughput due to co fied H-Store to use range partitioning.We discuss how to suppor tive partitioning se In building E-Stor W ID).S The Wapi SE table is parti ioned by the lly pull large of tuples.As s and CUSTO)co LTP sy paramter.Since there sforeign-key coordination betw apmoednemgioe a. tem that secks to minimize this simplified example throughout the paper for exposition 2.3 The Need for Live Reconfiguration ease data Ir hat use a heavyweight the same (from one partition partition). manc anges in workload access pattems 32 ions or pa 3.OVERVIEW OF SQUALL o demonstrate the detr d partition clus ized and fault-toler r to creat n the pr BMS h sume hat a tuple doe oes)or false post tives (i.e the system as

10467 W_ID City Zip 1 Miami 33132 2 Seattle 98101 C_ID Name W_ID 14 Ron 1 2 Wyatt 2 12 Jack 1 Warehouse Customer Partition 1 W_ID City Zip 3 New York 10467 4 Chicago 60414 C_ID Name W_ID 1 Mike 3 1004 Gabriel 3 3 Dean 3 Warehouse Customer Partition 2 W_ID City Zip 5 Los Angeles 90001 7 San Diego 92008 C_ID Name W_ID 21 Snake 5 7 R.J. 5 4 Stephen 7 Warehouse Customer Partition 3 W_ID City Zip 10 Austin 78702 C_ID Name W_ID 9 Todd 10 Warehouse Customer Partition 4 Figure 2: Simple TPC-C data, showing WAREHOUSE and CUSTOMER parti￾tioned by warehouse IDs. 0% 20% 40% 60% 80% Percent of New Orders for Warehouse 1-3 0 5000 10000 15000 TPS Figure 3: As workload skew increases, the number of new order transac￾tions increasingly access 3 warehouses in TPC-C and the collocated ware￾houses experience reduced throughput due to contention. fied H-Store to use range partitioning. We discuss how to support alternative partitioning schemes in Appendix C. Fig. 2 shows a simplified TPC-C database partitioned by the plan in Fig. 5a. The WAREHOUSE table is partitioned by its id col￾umn (W_ID). Since there is a foreign key relationship between the WAREHOUSE and CUSTOMER tables, the CUSTOMER table is also parti￾tioned by its W_ID attribute. Hence, all data related to a given W_ID (i.e., both WAREHOUSE and CUSTOMER) are collocated on a single par￾tition. Any stored procedure that reads or modifies either table will use W_ID as its routing parameter. Since there is a foreign-key re￾lationship between CUSTOMER and WAREHOUSE, the CUSTOMER table does not need an explicit mapping in the partition plan. We will use this simplified example throughout the paper for exposition. 2.3 The Need for Live Reconfiguration Although partitioned DBMSs like H-Store execute single-partition transactions more efficiently than systems that use a heavyweight concurrency control scheme, they are still susceptible to perfor￾mance degradations due to changes in workload access patterns [32]. Such changes could either cause a larger percentage of transactions to access multiple partitions or partitions to grow larger than the amount of memory available on their node. As with any distributed system, DBMSs need to react to these situations to avoid becom￾ing overloaded; failing to do so in a timely manner can impact both performance and availability in distributed DBMSs [20]. To demonstrate the detrimental effect of overloaded partitions on performance, we ran a micro-benchmark using H-Store with a three-node cluster. For this experiment, we used the TPC-C bench￾mark with a 100 warehouse database evenly distributed across 18 partitions. We modified the TPC-C workload generator to create a hotspot on one of the partitions by having a certain percentage of transactions access one of three hot warehouses instead of a uniformly random warehouse. Transaction requests are submitted from up to 150 clients running on a separate node in the same clus￾ter. We postpone the details of these workloads and the execution environment until Section 7. As shown in Fig. 3, as the warehouse selection moves from a uniform to a highly skewed distribution, the throughput of the system degrades by ∼60%. This shows that overloaded partitions have a significant impact on the throughput of a distributed DBMS like H-Store. The so￾lution is for the DBMS to respond to these adverse conditions by migrating data to either re-balance existing partitions or to offload data to new partitions. Some of the authors designed E-Store [38] for automatically identifying when a reconfiguration is needed and to create a new partition plan to shuffle data items between par￾titions. E-Store uses system-level statistics (e.g., sustained high CPU usage) to identify the need for reconfiguration, and then uses tuple-level statistics (e.g., tuple access frequency) to determine the placement of data to balance load across partitions. E-Store relies on Squall to execute the reconfiguration. Both components view each other as a black-box; E-Store only provides Squall with an updated partition plan and a designated leader node for a reconfig￾uration. Squall makes no assumptions on the plans generated by a system controller other than that all tuples must be accounted for. 0 20 40 60 80 100 120 Elapsed Time (seconds) 0 5000 10000 15000 TPS Figure 4: A Zephyr-like migration on two TPC-C warehouses to alleviate a hot-spot effectively causes downtime in a partitioned main-memory DBMS. In building E-Store we evaluated a Zephyr-like migration for load-balancing (cf. Section 7 for a detailed description) where des￾tination partitions reactively migrate tuples as needed and period￾ically pull large blocks of tuples. As shown by Fig. 4, however, this approach results in downtime for the system, and is therefore not an option for modern OLTP systems. This disruption is due to migration requests blocking transaction execution and a lack of coordination between partitions involved in the migration. These issues highlight the demand for a live reconfiguration sys￾tem that seeks to minimize performance impact and eliminate down￾time. We define a live reconfiguration as a change in the assignment of data to partitions in which data is migrated without any part of the system going off-line. A reconfiguration can cause the number of partitions in the cluster to increase (i.e., data from existing parti￾tions are sent to a new, empty partition), decrease (i.e., data from a partition being removed is sent to other existing partitions), or stay the same (i.e., data from one partition is sent to another partition). 3. OVERVIEW OF SQUALL Squall is a technique for efficiently migrating fine-grained data in a strongly consistent distributed OLTP DBMS. The key advantage of Squall over previous approaches is that it does not require the DBMS to halt execution while all data migrates between partitions, thereby minimizing the impact of reconfiguration. The control of data movement during a reconfiguration is completely decentral￾ized and fault-tolerant. Squall is focused on the problem of how to perform this reconfig￾uration safely. In particular, Squall ensures that during the reconfig￾uration process the DBMS has no false negatives (i.e., the system assumes that a tuple does not exist at a partition when it actually does) or false positives (i.e., the system assumes that a tuple exists

figuration and does not migrate any data.Since the reconfiguration se(D)( on.the 3.2 Data Migration The easiest way to safely transfer data in a distributed DBMS is at a partition when it does no). tions.For ing th system is still ex cuting tran this dat any lost or duplicated dat o over n of m This 3.1 Initialization If the s data is d pull me od intr uce ency for tr otified by an exter e.)the d on ation,(2)acti I data in al tra whether it is allowed to start the nfiguration s.This lock the tatus of the uple it is migrat ous nother partition then theshes 3.3 Termination ntralized tor of dat nce all of the partitions agree tos has sent nd/o art the reco tion (out on (in ning attri As we in S Each and exits the re 4.MANAGING DATA MIGRATION fter a partition compl The migration of data between partitions in a transactionally safe hen the le all on

plan:{ "warehouse (W_ID)": { "Partition 1" : [0-3) "Partition 2" : [3-5) "Partition 3" : [5-9) "Partition 4" : [9- ) }} (a) Old Plan plan:{ "warehouse (W_ID)": { "Partition 1": [0-2) "Partition 2": [3-5) "Partition 3": [2-3),[5-6) "Partition 4": [6- ) }} (b) New Plan Figure 5: An example of an updated partition plan for the TPC-C database shown in Fig. 2. at a partition when it does not). Determining when a reconfigu￾ration should occur and how the partition plan should evolve are addressed in the E-Store project [38]. We assume that a separate system controller monitors the DBMS and then initiates the recon- figuration process by providing the system with the new partition plan when appropriate [18, 38]. Squall processes a live reconfiguration in three stages: (1) ini￾tializing the partitions’ tracking data structures, (2) migrating data between partitions, and (3) identifying when the reconfiguration process has terminated. In this section, we provide an overview of these three steps. We discuss Squall’s data migration protocol in further detail in Section 4. We then present optimizations of this process in Section 5, such as eager data requests and a divide-and￾conquer technique for reconfiguration plans. 3.1 Initialization The initialization phase synchronously engages all of partitions in the cluster to start a new reconfiguration. As part of this step, each partition prepares for reconfiguration and identifies the tuples it will be exchanging with other partitions. A reconfiguration starts when the DBMS is notified by an exter￾nal controller that the system needs to re-balance. This notification identifies (1) the new partition plan for the database and (2) the des￾ignated leader node for the operation. The leader is any node in the cluster that contains a partition affected by the reconfiguration. If the reconfiguration calls for a new node to be added to the cluster, then that node must be on-line before the reconfiguration can begin. To initiate the reconfiguration, the leader invokes a special trans￾action that locks every partition in the cluster and checks to see whether it is allowed to start the reconfiguration process. This lock￾ing is the same as when a normal transaction needs to access every partition in the system. The request is allowed to proceed if (1) the system has terminated all previous reconfigurations and (2) the DBMS is not writing out a recovery snapshot of the database to disk. If either of these conditions is not satisfied, then the transac￾tion aborts and is re-queued after the blocking operation finishes. This ensures that all partitions have a consistent view of data owner￾ship and prevents deadlocks caused by concurrent reconfigurations. Once all of the partitions agree to start the reconfiguration, they then enter a “reconfiguration mode” where each partition examines the new plan to identify which tuples are leaving the partition (out￾going) and which tuples will be moving into the partition (incom￾ing). These incoming and outgoing tuples are broken into ranges based on their partitioning attributes. As we discuss in Section 5, this step is necessary because Squall may need to split tuple ranges into smaller chunks or split the reconfiguration into smaller sub￾reconfigurations for performance reasons. After a partition completes this local data analysis, it notifies the leader and waits to learn whether the reconfiguration will proceed. If all of the partitions agree to proceed with the reconfiguration, then the leader sends out acknowledgement to all of the partitions to begin migrating data. Squall only uses this global lock during the initialization phase to synchronize all partitions to begin recon- figuration and does not migrate any data. Since the reconfiguration transaction only modifies the meta-data related to reconfiguration, the transaction is extremely short and has a negligible impact on performance. For all our trials in our experimental evaluation, the average length of this initialization phase was ∼130 ms. 3.2 Data Migration The easiest way to safely transfer data in a distributed DBMS is to stop executing transactions and then move data to its new loca￾tion. This approach, known as stop-and-copy, ensures that transac￾tions execute either before or after the transfer and therefore have a consistent view of the database. But shutting down the system is unacceptable for applications that cannot tolerate downtime, so stop-and-copy is not an option. It is non-trivial, however, to transfer data while the system is still executing transactions. For example, if half of the data that a transaction needs has already been mi￾grated to a different partition, it is not obvious whether it is better to propagate changes to that partition or restart the transaction at the new location. The challenge is in how to coordinate this data movement between partitions without any lost or duplicated data, and with minimal impact to the DBMS’s performance. To overcome these problems, Squall tracks the location of mi￾grating tuples at each partition during the reconfiguration process. This allows each node’s transaction manager to determine whether it has all of the tuples that are needed for a particular transaction. If the system is uncertain of the current location of required tu￾ples, then the transaction is scheduled at the partitions where the data is supposed to be according to the new plan. Then when the transaction attempts to access the tuples that have not been moved yet, Squall will reactively pull data to the new location [19, 35]. Although this on-demand pull method introduces latency for trans￾actions, it has four benefits: (1) it always advances the progress of data migration, (2) active data is migrated earlier in the reconfigu￾ration, (3) it requires no external coordination, and (4) it does not incur any downtime to synchronize ownership metadata. In addition to the on-demand data pulls, Squall also asynchronously migrates additional data so that the reconfiguration completes in a timely manner. All of these migration requests are executed by a partition in the same manner as regular transactions. Each partition is responsible for tracking the progress of migrating data between itself and other partitions. In other words, each partition only tracks the status of the tuples it is migrating. This allows it to identify whether a particular tuple is currently stored in the local partition or whether it must retrieve it from another partition. 3.3 Termination Since there is no centralized controller in Squall that monitors the process of data migration, it up to each partition to indepen￾dently determine when it has sent and/or received all of the data that it needs. Once a partition recognizes that it has received all of the tuples required for the new partition plan, it notifies the current leader that the data migration is finished at that partition. When the leader receives acknowledgments from all of the partitions in the cluster, it notifies all partitions that the reconfiguration process is complete. Each partition removes all of its tracking data structures and exits the reconfiguration mode. 4. MANAGING DATA MIGRATION The migration of data between partitions in a transactionally safe manner is the most important feature of Squall. As such, we now discuss this facet of the system in greater detail. We first describe how Squall divides each partition’s migrating tuples into ranges and tracks the progress of the reconfiguration

take for a partition difficult Since these tables are not partitic 是→ f cus tracke The pproach is that artniononyhd amount of s al state that Squall mai ntains during the recon d on these We then de ribe the two ways that Squal data 1L This alows the Section5.including how to split rangesmoun o e the am 4.2 Tracking Reconfiguration Progress are th o help explain this pro partition to record the curent status of these range that tupl NOT STARTED:The data as ociated with the range has n that rec PARTIAL:Son ome of tween the 4.1 Identifying Migrating Data COMPLETE:All of the data for the range has migrated to the des- When a new reconfiguration begins.Squall calculates the differ tination partition and ou This Each partition locally anizes the ranges into incor SELECT FROM MAREHOUSE WHERE W_ID 6 AND W_ID 7 ing warehouses for partition3 are noted as edfor this tra ers with wI WAREHOUSE,W_ID [2.3).13) )into the following ve areh E,1D=6,o.3→4 in these That is.an mple.the sends all customers with greater to partition4: the key which p C5 CHER.W_ID=6,6o.3→ of transacti ions that All of the tables in TPC-C that are partitioned on their WAREHOUSE time than scanning the partition plan entres to de

Reconfiguration Pull W_ID=2 Pull W_ID>5 Warehouse W_ID: 3-4 Customer W_ID: 3-4 Txns Partition 2 Warehouse W_ID: 0-2 Customer W_ID: 0-2 Txns Partition 1 Warehouse W_ID: 5-8 Customer W_ID: 5-8 Txns Partition 3 Warehouse W_ID: 9- Customer W_ID: 9 Txns Partition 4 Warehouse W_ID: 3-4 Customer W_ID: 3-4 Txns Partition 2 Warehouse W_ID: 0-1 Customer W_ID: 0-1 Txns Partition 1 Warehouse W_ID: 2,5 Customer W_ID: 2,5 Txns Partition 3 Warehouse W_ID: 6- Customer W_ID: 6- Txns Partition 4 Squall State Squall State Squall State Squall State Incoming: WHouse: (2) Customer: (2) Outgoing: WHouse: [6,9) Customer: [6,9) Figure 6: As a system’s partition plan changes, Squall tracks the progress of the reconfiguration at each node to ensure correct data ownership. based on these ranges. We then describe the two ways that Squall moves data: reactive migration and asynchronous migration. The former moves tuples when transactions need them. This allows the system to migrate hot tuples early in the reconfiguration without the use of complex usage modeling [36]. The latter is when the system periodically sends data to ensure that the reconfiguration eventually completes with minimal impact on the DBMS’s performance. To help explain this process, we will refer to Figs. 5 and 6 as our running example. For a particular tuple that is migrating from one partition to another, we refer to the partition that is losing that tuple as the source partition and the partition that receives the tuple as the destination partition. Although a partition can be both a source and destination during a reconfiguration, we refer to a partition as either a source or destination for a particular tuple. 4.1 Identifying Migrating Data When a new reconfiguration begins, Squall calculates the differ￾ence between the original partition plan and the new plan to deter￾mine the set of incoming and outgoing tuples per partition. This allows a partition to determine when it can safely execute a trans￾action that accesses migrating data without any external metadata. Migrating tuples are organized into reconfiguration ranges that specify a table name, the partitioning attribute(s) of the table, the minimum-inclusive key, maximum-exclusive key, and the old/new partition IDs. Tables in distributed OLTP DBMSs are partitioned by one or more columns [14, 32]. Without loss of generality, how￾ever, we discuss reconfiguration ranges for the single column case. Each partition locally organizes the ranges into incoming and outgoing ranges based on the previous and new partition IDs. For example, in the reconfiguration shown in Figs. 5 and 6, the incom￾ing warehouses for partition 3 are noted as: (WAREHOUSE, W_ID = [2, 3), 1 → 3) This means that partition 3 receives warehouse 2 from partition 1. Likewise, the following range identifies that partition 3 sends all warehouses with an ID of 6 or greater to partition 4: (WAREHOUSE, W_ID = [6, ∞), 3 → 4) These rules cascade for all tables that are not explicitly listed in these partition plan entries. That is, any table with a foreign￾key relationship to one of the tables identified in an entry will have its tuples migrated based on these rules as well. In our TPC-C example, the CUSTOMER table is partitioned by its WAREHOUSE ID, thus partition 3 would also have the following implicit rule that sends all customers with a W_ID of 6 or greater to partition 4: (CUSTOMER, W_ID = [6, ∞), 3 → 4) All of the tables in TPC-C that are partitioned on their WAREHOUSE id, such as DISTRICT, ORDERS, and STOCK, are handled similarly. We note that this makes predicting how long the migration will take for a partition difficult. Since these tables are not partitioned by a primary or unique key, the number of tuples associated with a range can be far larger than the cardinality of the partition keys en￾capsulated by the range (e.g., there can be thousands of customers associated with a single W_ID). Each of the above ranges is derived deterministically from the original and new partition plans. This means that each partition can independently calculate its local set of incoming and outgoing ranges from the updated plan that it received in the initialization phase. The advantage of this approach is that a partition only has to track the list of ranges migrating to or from itself. This reduces the amount of global state that Squall maintains during the reconfig￾uration and facilitates several other enhancements to improve per￾formance. We discuss these additional optimizations for this phase in Section 5, including how to split ranges to reduce the amount of data associated with each of them. 4.2 Tracking Reconfiguration Progress Squall tracks the progress of data migration for all incoming and outgoing ranges at a particular partition. It maintains a table at each partition to record the current status of these ranges: NOT STARTED: The data associated with the range has not yet mi￾grated to/away from this partition, and therefore all data as￾sociated with the range is located at the source partition. PARTIAL: Some of the data for the range has been migrated, and some of the tuples may be currently in-flight between the source and destination partitions. COMPLETE: All of the data for the range has migrated to the des￾tination partition. Continuing with our example from Section 4.1, a status of NOT STARTED for the CUSTOMER table range indicates that all customers with a W_ID of 6 or greater are present only at partition 3. This means that any transaction that needs to access the tuples for these customers will do so at partition 3. Since a transaction could execute a query that contains a predi￾cate at a different granularity than the reconfiguration range, Squall allows for the initial reconfiguration ranges to be split into sub￾ranges by a query. For the same WAREHOUSE range example with a NOT STARTED status, assume that a transaction arrives at partition 4 that executes the following query: SELECT * FROM WAREHOUSE WHERE W_ID >= 6 AND W_ID 7, thus the entire range is not needed for this transaction. In this scenario, partition 4 would split the original range [6, ∞) into the following two ranges both with the status of NOT STARTED: (WAREHOUSE, W_ID = [6, 8), 3 → 4) (WAREHOUSE, W_ID = [8, ∞), 3 → 4) When partition 4 requests the data associated with the first range, partition 3 similarly splits its original range to reflect the requested range. Once the data for the sub-range is migrated, both partitions update the status of the first range to COMPLETE. For transactions with queries that access an individual key through an equality predicate (e.g., W_ID = 7), Squall must find the range that the key belongs to in its tracking table to determine which par￾tition has that data. Since many OLTP workloads are comprised of transactions that access tuples through single keys, Squall also supports recording the movement of individual tuples at the key level through its tracking table. This enables faster lookups at run￾time than scanning the partition plan entries to determine whether

a key is in a range.It also reduces the amount of work required for partitions.As both partitions are locked during the migration of e uaicn herey rventin anyation csdrcohcecoiguion artition with after y to update the WAREHOUSE Squall the deadlock detection topr the tracki Ouce the the blocking cd by disrupi ve for ac 4.3 Identifying Data Locatior 4.5 Asynchronous Migration the DBMS.the on 3.1).it then dure tha ons to tha 29.32 s this p Thi ples and th chronous pull re e【o execut task did to e hepam each nk to the dest nalionpartition are uncertain,then it for the requ st to the table.This process repeats uled at a all tra g rtition to 10 the ed tuple mproves perf as only ma ing data at the source ine a tuple the ti is para 4.4 Reactive Migration chu d by Squ Squall schedule transaction at its Tspondinedcstina We explore the opti of th when it executes Like 5.OPTIMIZATIONS pe This 5.1 Range Splitting ng) As described in Section 4.1.Squall computes the initial transfe ranges of tuples for a n isqueued just like a transaction sacquires ex

a key is in a range. It also reduces the amount of work required for range splitting and simplifies the tracking of keys with non-discrete domains (e.g., strings or floating-point values). When a transaction reads or updates tuples, Squall ensures that all required data is present. Using our TPC-C example from above, assume that no data has yet been migrated (i.e., all of the entries in the tracking table are set to NOT STARTED). A transaction then arrives at partition 4 that executes a query to update the WAREHOUSE tuple with W_ID = 7. After failing to find a matching key entry in the tracking table and checking that the corresponding range is not marked as COMPLETE, partition 4 issues a request for (WAREHOUSE, W_ID = 7). Once the warehouse is migrated, both partitions set the corresponding WAREHOUSE range in the status table to PARTIAL and add a key-based entry for (WAREHOUSE, W_ID = 7) in the tracking table with the status of COMPLETE. 4.3 Identifying Data Location Under normal operation, when a new transaction request arrives at the DBMS, the system evaluates the routing parameters for the stored procedure that the transaction wants to execute with the cur￾rent plan to determine the base partition for that request [29, 32]. But during a reconfiguration, Squall intercepts this process and uses its internal tracking metadata to determine the base partition. This is because the partition plan is in transition as tuples move and thus the location of a migrating tuple is uncertain. The mechanisms in Section 4.2 ensure safe execution of transactions on migrating data, whereas the following techniques for determining where to execute transactions are for improving the DBMS’s performance. If the data that a transaction needs is moving and either the source or destination partition is local to the node receiving the transac￾tion, then Squall will check with the local partition to determine whether the required tuple(s) are present (i.e., the source partition has the range marked NOT STARTED or the destination partition has the range marked COMPLETE). If Squall deems that the tuples’ loca￾tions are uncertain, then it forwards the request to the destination partition. Since a transaction could have been (correctly) sched￾uled at a source partition before tuples were migrated, Squall traps transaction execution before it starts to verify that those required tuples were not migrated out while the transaction was queued. In this scenario, the transaction is restarted at the destination partition. Scheduling the transaction at the partition that always has the re￾quired tuples improves performance by avoiding a migration when a transaction has already arrived at the same local site. But in all other cases, it is better to send the transaction to the destination partition and then use Squall’s reactive migration mechanism to pull data as it is needed. This alleviates overloaded partitions more quickly by migrating hot tuples first and minimizes the communi￾cation overhead of partitions trying to determine a tuple’s location. 4.4 Reactive Migration Squall schedules each transaction at its corresponding destina￾tion partition, even if the system has not completed the migration of the data that the transaction will access when it executes. Like￾wise, when a transaction invokes a query, Squall examines the re￾quest to determine whether the data that it accesses has migrated to its destination partition. In this scenario, the destination partition blocks the query while the data is reactively pulled from the source partition. This blocks all transactions that access these partitions during this time, thereby avoiding consistency problems (e.g., two transactions modify the same the tuple while it is moving). To perform the reactive migration, Squall issues a pull request to the source partition. This request is queued just like a transaction and thus acquires exclusive locks on both source and destination partitions. As both partitions are locked during the migration of data items, no other transaction can concurrently issue a read or update query at these partitions, thereby preventing any transaction anomalies due to the reconfiguration. The pull request is scheduled at the source partition with the highest priority so that it executes immediately after the current transaction completes and any other pending reactive pull requests. Squall relies on the DBMS’s standard deadlock detection to pre￾vent cyclical reactive migrations from stalling the system, since all transactions are blocked while the source partition is extracting the requested data. Additionally, transaction execution is also blocked at the destination partition when it loads the requested data. Since the blocking caused by reactive migrations can be disruptive for ac￾tive transactions, Squall also employs an additional asynchronous migration mechanism to minimize the number of reactive pulls. 4.5 Asynchronous Migration After Squall initializes the list of incoming data ranges for each partition (cf. Section 3.1), it then generates a series of asynchronous migration requests that pull data from the source partitions to their new destination. During the reconfiguration, Squall periodically schedules these requests one-at-a-time per partition with a lower priority than the reactive pull requests. When the source partition receives an asynchronous pull request, it first marks the target range as PARTIAL in its tracking table. It then sub-divides the request into tasks that each retrieve a fixed-size chunk to prevent transactions from blocking for too long if Squall migrates a large range of tuples. If the last task did not extract all of the tuples associated with the range, then another task for the asynchronous pull request is rescheduled at the source partition. As Squall sends each chunk to the destination partition, it includes a flag that informs the destination whether it will send more data for the current range. For the first chunk that arrives at the destination for the range, the destination lazily loads this data and marks the range as PARTIAL in its tracking table. This process repeats until the last chunk is sent for the current range. Squall will not initiate two concurrent asynchronous migration requests from a destination partition to the same source. Addition￾ally, before issuing an asynchronous pull request, Squall checks if the entire range has been pulled by a prior reactive migration (i.e., marked as COMPLETE). If so, the request is discarded; otherwise the full request is issued as only matching data at the source will be returned. Additionally, if a transaction attempts to access partially migrated data, then this forces a reactive pull to migrate the remain￾ing data or flush pending responses (if any). Since limiting the amount data per extraction and interleaving regular transaction execution is paramount to minimizing perfor￾mance impact, the time between asynchronous data requests and the asynchronous chunk size limit are both controlled by Squall. These settings allow Squall to balance the trade-off between time to completion and impact of reconfiguration. We explore the opti￾mal setting for these parameters in Section 7.6. 5. OPTIMIZATIONS We next present several optimizations for improving the runtime performance of the DBMS during a reconfiguration with Squall. 5.1 Range Splitting As described in Section 4.1, Squall computes the initial transfer ranges of tuples for a new reconfiguration by analyzing the differ￾ence between the previous and new partition plans. If the new parti￾tion plan causes a large number of tuples to move to a new location, Squall may schedule transactions on the destination partition when

5.3 Pull Prefetching a pull request requests a key that is (parti and ()has split re s (ef.Section 5.D).Squal plan split into three sub-pans. 5.4 Splitting Reconfigurations Executing a reconfiguration as a single step can cause perfor tention an ccur when a single partition is the sour same s on the large amount of al part ting a large r d be better served running on the so rition plan It th 0NA3H0U5E.id=[L.100000).3→4) rtition in each sub-plan.This Although Squall'ssyronous would split entire range is marked as PARTIA in the source and destination sub-plan that ach mig range will pot respo ring that all p within this entially or this rea hema 1371 Sauall can als of cnecomlaeeoL.100100.20j-00.100m0. el keys tha in TPC-C are typ y partitioned by the w ac 5.2 Range Merging with it.Eac ea hotspot has formed execution for extended periods by throttling migrations. 6 FAULT TOLERANCE buted DBMS like H-Stor ailabil n each ests in ugh replic ge small rang merged range is capped to half of the chunk size limit. ion

plan:{ "warehouse (W_ID)":{ "Partition 1->2" : [1-2), "Partition 1->3" : [2-3), "Partition 1->4" : [3-4) }} % → & plan:{ "warehouse (W_ID)":{ "Partition 1->2" : [1-2) }} plan:{ "warehouse (W_ID)":{ "Partition 1->3" : [2-3) }} plan:{ "warehouse (W_ID)":{ "Partition 1->4" : [3-4) }} Figure 7: A sample reconfiguration plan split into three sub-plans. District'1' District'10' Warehouse'1' Order'1' Customer'1' Customer'3,000' Order'5' …' …' …' Par$$on'1' Order'YYY1' Customer'XXX' …' Order'YYY8'' District'2' Customer'YYY' Par$$on'2' Figure 8: During reconfiguration in TPC-C, Squall uses secondary parti￾tioning to split the DISTRICT table to avoid moving an entire WAREHOUSE entity all at once. While migration is in progress, the logical warehouse is split across two partitions, causing some distributed transactions. they would be better served running on the source partition. For ex￾ample, the following range entry requires movement of 100k tuples from the WAREHOUSE table from partition 3 to partition 4: (WAREHOUSE, id = [1, 100000), 3 → 4) Although Squall’s asynchronous migration mechanism would split this range into smaller chunks, once the first chunk is extracted the entire range is marked as PARTIAL in the source and destination partitions’ tracking tables. This will cause all transactions that ac￾cess data within this range to be routed to the destination partition and any query that accesses data within this range will potentially cause a reactive pull, even if the data was already migrated. For this reason, during the initialization phase, Squall splits con￾tiguous ranges into smaller sub-ranges, each with an expected size equal to the chunk size. Assuming that the average size of each tu￾ple is 1 KB in our above example, with a chunk size limit of 1 MB the range would be set to [1, 1000),[1000, 2000),...,[99000, 100000). Such smaller ranges result in fewer partial ranges, and thus reduce the likelihood that transactions are blocked unnecessarily. 5.2 Range Merging In addition to splitting large contiguous ranges, Squall also com￾bines small, non-contiguous ranges together to reduce the number of pull requests. For example, suppose a hotspot has formed on partition 1 for keys [1, 10). The controller’s load-balancing algo￾rithm may distribute keys to other partitions in a round-robin man￾ner. In this example, assume keys (1, 3, 5, 7, 9) are migrated from partition 1 to partition 2. This would result in five migration pull requests between the partitions for a small amount of data per re￾quest. Since there is overhead and disruption to service associated with each reactive pull, issuing these small requests individually is sub-optimal. Therefore, if a table is partitioned on a unique key and has a fixed tuple size, Squall will merge small ranges into a single pull request that is composed of multiple ranges. The size of this merged range is capped to half of the chunk size limit. 5.3 Pull Prefetching We also extend Squall’s reactive pull mechanism to eagerly re￾turn more data from the source partition than is requested each time. When a pull request requests a key that is (1) partitioned on a unique column, (2) has fixed-size tuples (e.g., no varchar fields), and (3) has split reconfiguration ranges (cf. Section 5.1), Squall eagerly returns the entire range instead of the single requested key. In our experience, the additional time to extract additional tuples is substantially less than the time to schedule and process additional pull requests for the remaining of tuples in the range. 5.4 Splitting Reconfigurations Executing a reconfiguration as a single step can cause perfor￾mance problems due to contention on a single partition. Such con￾tention can occur when a single partition is the source for many destination partitions (e.g., when moving tuples out of a hotspot partition). In this scenario, multiple destination partitions make concurrent migration requests to the same source partition. Each of these requests blocks transaction execution, thus increasing the load on the already-overloaded source partition. This contention can also occur for certain workloads, such as TPC-C, when there is a large amount of data associated with an individual partition key. These request convoys greatly degrade the DBMS’s performance beyond what normally occurs during a reconfiguration. To avoid this problem, Squall throttles data movement by split￾ting a large reconfiguration into smaller units. Squall first identifies the ranges of keys that need to move in order to transition to the new partition plan. It then splits these ranges into a fixed number of sub￾plans where each partition is a source for at most one destination partition in each sub-plan. This is different than the splitting of re￾configuration ranges described in Section 5.1. because it reduces the number of partitions that retrieve data from a single partition. As shown in the example in Fig. 7, the plan on the left moves data from partition 1 to partitions 2, 3, and 4. The plan is then di￾vided into three separate sub-plans that each migrate data from par￾tition 1 to just one partition at a time. The reconfiguration leader is responsible for generating these sub-plans and ensuring that all par￾titions move through the sub-plans together. The advantage of this approach is that it does not require additional coordination from a partition that is already overloaded. For databases with a tree-based schema [37], Squall can also split reconfigurations at a finer granularity using secondary partitioning attributes. This allows the system to split up the migration of root￾level keys that would otherwise require a significant number of re￾lated tuples to be migrated simultaneously. For example, the tables in TPC-C are typically partitioned by the WAREHOUSE id. But each warehouse in TPC-C may have over one million tuples associated with it. Each warehouse contains 10 DISTRICT records, however, so by partitioning the tables using their DISTRICT ids, Squall can split a warehouse into 10 pieces to limit the overhead of each data pull (cf. Fig. 8). Splitting the ranges in this way increases the number of distributed transactions in TPC-C, but avoids blocking execution for extended periods by throttling migrations. 6. FAULT TOLERANCE Distributed DBMS like H-Store ensure high availability and fault tolerance through replicating partitions on other nodes [37]. Each partition is fully replicated at another node and must remain in sync with the primary during reconfiguration. Squall fully integrates with these master-slave replication schemes. Squall schedules any reconfiguration operation at both the pri￾mary and its secondaries, but all data movement is done through

the primary since it cannot assume that there is a one-to-one match base to the state that it was in before the crash.The DBMS's coordi in seri For data loading.the prir o 7.EXPERIMENTAL EVALUATION node or an entire cluster during a reconfiguration. nteuH-Sre th ench 6.1 Failure Handling We set Squall's chunk s There re the r of odkwihaoucepaionilgam&elamnodewihdi sof rec on.they are ab any node f node. Sto s complete lls singl the last mes sage sent by ting data.th Since mig are syn titions resend any pending requests to the recently failed site 6.2 Crash recovery ennco thy epbyr 9)mgron on the ronowingv s all of its check sed mig ures ach.Second. yr can only migr a pu tim es after a reco tion completes bu the itialize) ces In o ctive techni nd o d the starting pe never ace ed)and pu ulling single keys at a time crea after the che ther rrent plar enge for each p rtition then read nts measure incl ing overloade partitions ory from the file or disk,the DBMS then replays the command log to restore the data-

the primary since it cannot assume that there is a one-to-one match between the number of source replicas and destination replicas. For the data extraction requests, the primary notifies secondaries when a range is partially or fully extracted so the replica can remove tu￾ples from its local copy. Using fixed-size chunks enable the repli￾cas to deterministically remove the same tuples per chunk as their primary without needing to send a list of tuple ids. For data loading, the primary forwards the pull response to its secondary replicas for them to load newly migrated tuples. Before the primary sends an acknowledgement to Squall that it received the new data, it must receive an acknowledgement from all of its replicas. This guarantees strong consistency between the replicas and that for each tuple there is only one primary copy at any time. We now describe how Squall handles either the failure of a single node or an entire cluster during a reconfiguration. 6.1 Failure Handling The DBMS sends heartbeats between nodes and uses watchdog processes to determine when a node has failed. There are three cases that Squall handles: (1) the reconfiguration leader failing, (2) a node with a source partition failing, and (3) a node with destina￾tion partition failing. A node failure can involve all three scenarios (e.g., the leader fails and has in- and out-going data). Since replicas independently track the progress of reconfiguration, they are able to replace a failed primary replica during reconfiguration if needed. If any node fails during reconfiguration, it is not allowed to rejoin the cluster until the reconfiguration has completed. Afterwards it recovers the updated state from its primary node. During the data migration phase, the leader’s state is synchronously replicated to secondary replicas. If the leader fails, a replica is able to resume managing the reconfiguration. After fail-over, the new leader replica sends a notification to all partitions to announce the new leader. Along with this notification, the last message sent by the leader is rebroadcast in case of failure during transmission. When a node fails while migrating data, the secondary replica replacing the partition’s primary replica reconciles any potentially lost reconfiguration messages. Since migration requests are syn￾chronously executed at the secondary replicas, Squall is only con￾cerned with requests sent to the failed primary that have not yet been processed. When a secondary replica is promoted to a pri￾mary, it broadcasts a notification to all partitions so that other par￾titions resend any pending requests to the recently failed site. 6.2 Crash Recovery During a reconfiguration, the DBMS suspends all of its check￾point operations. This ensures that the partitions’ checkpoints stored on disk are consistent (i.e., a tuple does not exist in two partitions at the same time). The DBMS continues to write transaction entries to its command log during data migration. If the entire system crashes after a reconfiguration completes but before a new snapshot is taken, then the DBMS recovers the data￾base from the last checkpoint and performs the migration process again. The DBMS scans the command log to find the starting point after the last checkpoint entry and looks for the first reconfiguration transaction that started after the checkpoint. If one is found, then the DBMS extracts the partition plan from the entry and uses that as the current plan. The execution engine for each partition then reads its last snapshot. For each tuple in a snapshot, Squall determines what partition should store that tuple, since it may not be the same partition that is reading in the snapshot. Once the snapshot has been loaded into memory from the file on disk, the DBMS then replays the command log to restore the data￾base to the state that it was in before the crash. The DBMS’s coordi￾nator ensures that these transactions are executed in the exact order that they were originally executed the first time. Hence, the state of the database after this recovery process is guaranteed to be correct, even if the number of partitions changes due to the reconfiguration. This is because (1) transactions are logged and replayed in serial order, so the re-execution occurs in exactly the same order as in the initial execution, and (2) replay begins from a transactionally￾consistent snapshot that does not contain any uncommitted data, so no rollback is necessary at recovery time [21, 27]. 7. EXPERIMENTAL EVALUATION We now present our evaluation of Squall. For this analysis, we integrated Squall with H-Store [1] with an external controller that initiates the reconfiguration procedure at a fixed time in the bench￾mark trial. We used the April 2014 release of H-Store with com￾mand logging enabled. We set Squall’s chunk size limit to 8 MB and set a minimum time between asynchronous pulls to 200 ms. We also limit the number of reconfiguration sub-plans to be between 5 and 20, with a 100 ms delay between them. The experiments in Section 7.6 show our justification for this configuration. The experiments were conducted on a cluster where each node has a Intel Xeon E5620 CPU running 64-bit CentOS Linux with OpenJDK 1.7. The nodes are in a single rack connected by a 1GB switch with an average RTT of 0.35 ms. As part of this evaluation, we also implemented three different reconfiguration approaches in H-Store: Stop-and-Copy: A distributed transaction locks the entire clus￾ter and then performs the data migration. All partitions block until this process completes. Pure Reactive: The system only pulls single tuples from the source partition to the destination when they are needed. This is the same as Squall but without the asynchronous migration or any of the optimizations described in Section 5. Zephyr+: This technique combines the pure-reactive migration with chunked asynchronous pulls (cf. Section 4.5) and pull prefetching (cf. Section 5.3) to simulate pulling data in pages instead of individual keys. The purely reactive reconfiguration approach is semantically equiv￾alent to the Zephyr [19] migration system based on the following observations. First, Zephyr relies on a transaction execution com￾pleting at the source to begin a push-phase in addition to the reac￾tive pull-based migration. In general for a reconfiguration, transac￾tion execution at the source does not complete, so we stick to a pure pull-based approach. Second, Zephyr can only migrate a disk-page of tuples at a time (typically 4 KB). Squall on the other hand can migrate both single tuples and ranges. Third, there is no need to copy a “wireframe” between partitions to initialize a destination, as all required information (e.g., schema or authentication) is already present. In our experiments, the pure reactive technique was not guaranteed to finish the reconfiguration (because some tuples are never accessed) and pulling single keys at a time created significant coordination overhead. Thus, we also implemented the Zephyr+ approach that uses chunked asynchronous pulls with prefetching. Our experiments measure how well these approaches are able to reconfigure a database in H-Store under a variety of situations, including re-balancing overloaded partitions, cluster contraction, and shuffling data between partitions. For experiments where Pure Reactive and Zephyr+ results are identical, we only show the latter

Sapp-and-Copy a)Th b)T ut (TPC-C 一2pr+ Elaosed Time (seconds) e)Mean Latensy (YCSB id)Mean Latensy (PC-C) 7.1 Workloads mark framew 7.2 Load Balancing and thn me partition and the rate hot tuples aw m10 lit ac YCSB:The Yahoo!Cloud Serving Benchmark is a collection ot workloads that are repre tition. d-aanc ing a singk The YCSB 10 colum mly ger rated string it takes longe olete the p cess pattern or with Zipfian-skewed hotspots. ases as the over uniform TPC-C:This The DBMS's performance for the other meth nine tables and five procedures that simulate a warehouse-centric

0 2000 4000 6000 TPS Stop-and-Copy 0 2000 4000 6000 TPS Pure Reactive 0 2000 4000 6000 TPS Zephyr+ 0 20 40 60 80 100 120 Elapsed Time (seconds) 0 2000 4000 6000 TPS Squall (a) Throughput (YCSB) 0 5000 10000 15000 TPS Stop-and-Copy 0 5000 10000 15000 TPS Zephyr+ 0 50 100 150 200 250 300 Elapsed Time (seconds) 0 5000 10000 15000 TPS Squall (b) Throughput (TPC-C) 0 50 100 150 200 Lat. (ms) Stop-and-Copy 0 2000 4000 6000 Lat. (ms) Pure Reactive 0 50 100 150 200 Lat. (ms) Zephyr+ 0 20 40 60 80 100 120 Elapsed Time (seconds) 0 50 100 150 200 Lat. (ms) Squall (c) Mean Latency (YCSB) 0 500 1000 1500 2000 Lat. (ms) Stop-and-Copy 0 500 1000 1500 2000 Lat. (ms) Zephyr+ 0 50 100 150 200 250 300 Elapsed Time (seconds) 0 500 1000 1500 2000 Lat. (ms) Squall (d) Mean Latency (TPC-C) Figure 9: Load Balancing – Due to a skewed workload, one partition distributes hot tuples to cold partitions. YCSB distributes 90 tuples across 14 partitions and TPC-C distributes all tuples associated with 2 warehouses to 2 different partitions. 7.1 Workloads We now describe the two workloads from H-Store’s built-in bench￾mark framework that we used in our evaluation. Transaction re￾quests are submitted from 180 client threads running on separate nodes. Each client submits transactions to any DBMS node in a closed loop (i.e., it blocks after it submits a request until the result is returned). In each trial, the DBMS “warms-up” for 30 seconds and then measurements are collected for five minutes. Latencies are measured as the time from when the client submits a request to when it receives a response. The dashed vertical-line in our time￾series graphs denotes the start of a reconfiguration and the light dotted line is the end of the reconfiguration. YCSB: The Yahoo! Cloud Serving Benchmark is a collection of workloads that are representative of large-scale services created by Internet-based companies [12]. For all of the YCSB experi￾ments in this paper, we use a YCSB database containing a single table with 10 million records. Each YCSB tuple has a primary key and 10 columns each with 100 bytes of randomly generated string data. The workload consists of two types of transactions; one that updates a single record for 15% of operations and one that reads a single record for the remaining 85% of operations. Our YCSB workload generator supports executing transactions with either a uniform access pattern or with Zipfian-skewed hotspots. TPC-C: This benchmark is the current industry standard for evaluating the performance of OLTP systems [39]. It consists of nine tables and five procedures that simulate a warehouse-centric order processing application. A key feature of this benchmark is that roughly 10% of all transactions touch multiple warehouses, which typically results in a multi-partition transaction. We use a database with 100 warehouses partitioned across three nodes. 7.2 Load Balancing We first evaluate how well the reconfiguration approaches are able to reconfigure a skewed workload. In these tests, we create a hotspot on a single partition and then migrate hot tuples away from the overloaded partition. For YCSB, we evenly partition 10 million tuples between the four nodes and create a hotspot on 100 tuples. For TPC-C, we use a 100 warehouse database split across three nodes and then create a three-warehouse hotspot on one par￾tition. We used a simple load-balancing algorithm that distributes hot tuples to other partitions in a round-robin manner [38]. The YCSB results in Figs. 9a and 9c show that Squall has a lower impact on the DBMS’s performance than the other methods, but it takes longer to complete the reconfiguration. Initially, Squall causes the DBMS’s throughput to drop by ∼30% in the first few seconds when the reconfiguration begins. This is because transac￾tions are redirected to the new destination partition for the hotspot’s tuples and then that data is pulled from their source partition. After this initial dip, the throughput steadily increases as the overloaded partition becomes less burdened until the reconfiguration completes after ∼20 seconds. The DBMS’s performance for the other meth￾ods is strikingly different: they all halt transaction execution for 5– 15 seconds and increase response latency significantly even though

Sicp-ano.Coo 7.3 Cluster Consolidation the with a YCSB data base that is pa toned acro our node The tuples the DrMs the data from ther the thus eveyrsn 0 psedTreaon 20 MS'sthrougputoo (a)Throughput (YCSB) Stop-and-Copy uration thr ugh splitting the reconf ration into many Thi at the py is not do th Slep-and-Copy t of data hei Elapsed Tin (a)Throughput (YCSB Stop-and-Copy all at once.It takes the 2 the data and update indexes for these t es to get ked up and naller sub-pan es the an disruptive a large pull can be for method

0 2000 4000 6000 TPS Stop-and-Copy 0 2000 4000 6000 TPS Pure Reactive 0 2000 4000 6000 TPS Zephyr+ 0 50 100 150 200 250 300 Elapsed Time (seconds) 0 2000 4000 6000 TPS Squall (a) Throughput (YCSB) 0 100 200 300 400 Lat. (ms) Stop-and-Copy 0 2000 4000 Lat. (ms) Pure Reactive 0 5000 10000 Lat. (ms) Zephyr+ 0 50 100 150 200 250 300 Elapsed Time (seconds) 0 100 200 300 400 Lat. (ms) Squall (b) Latency (YCSB) Figure 10: Cluster Consolidation – Contracting from four nodes to three nodes, with all remaining partitions receiving an equal number of tuples from the contracting node. the amount of data being migrated is small. For Stop-and-Copy, this downtime results in thousands of aborted transactions. The Pure Reactive and Zephyr+ migrations cause the DBMS to “hold” transactions during the migration, which results in clients waiting indefinitely for their requests to be processed. This demonstrates that the difficulties in re-balancing an overloaded partition are not only dependent on the amount of data that is moved. For TPC-C, Figs. 9b and 9d again show that Stop-and-Copy and Zephyr+ block execution for 24 and 30 seconds, respectively. These results also show the DBMS’s performance to oscillate dur￾ing reconfiguration with Squall. This is because the transactions access two large tables that cause Squall to send on-demand pull requests that retrieve a lot of data all at once. It takes the system 500–2000 ms to move the data and update indexes for these tables, during which the partitions are unable to process any transactions. This causes their queues to get backed up and results in short peri￾ods during the reconfiguration when no transaction completes. It is these distributed transactions in TPC-C that increase the likelihood that a partition will block during data migration. As described in Section 5.4, splitting a single reconfiguration into smaller sub-plans minimizes contention on a single partition and reduces the amount of data pulled in a single request by splitting up large ranges. These results demonstrate how disruptive a large pull can be for methods that lack this functionality. 7.3 Cluster Consolidation We next measure the performance impact of the reconfiguration methods when the number of nodes in the cluster contracts for a fixed workload. For this experiment, we start with a YCSB data￾base that is partitioned across four nodes. The tuples are evenly di￾vided among the partitions and clients access tuples with a uniform distribution. After 30 seconds, the system initiates a reconfigura￾tion that removes one of the nodes. This causes the DBMS to move the data from the removed node to the other three remaining nodes. The results in Fig. 10 show that Pure Reactive never completes the reconfiguration and the DBMS’s throughput is nearly zero. This is because transactions access tuples uniformly, and thus every trans￾action causes another on-demand pull request to retrieve a sin￾gle tuple at a time. The DBMS’s throughput also drops to nearly zero with Zephyr+ during the reconfiguration because the parti￾tions on the remaining nodes all try to retrieve data at the same time. This causes the system to take longer to retrieve the data for each pull request. Squall alleviates this bottleneck by limiting the number of concurrent partitions actively involved in the reconfig￾uration through splitting the reconfiguration into many steps. This results in Squall’s reconfiguration taking approximately 4× longer than Stop-and-Copy. We contend that this trade-off is acceptable given that the DBMS is not down for almost 50 seconds during the reconfiguration. Likewise, we believe that Squall’s consistent per￾formance impact shows that it is well-suited for both load balancing and contraction reconfigurations that do not have a tight deadline. 0 2000 4000 6000 TPS Stop-and-Copy 0 2000 4000 6000 TPS Pure Reactive 0 2000 4000 6000 TPS Zephyr+ 0 20 40 60 80 100 120 Elapsed Time (seconds) 0 2000 4000 6000 TPS Squall (a) Throughput (YCSB) 0 100 200 300 400 Lat. (ms) Stop-and-Copy 0 100 200 300 400 Lat. (ms) Pure Reactive 0 100 200 300 400 Lat. (ms) Zephyr+ 0 20 40 60 80 100 120 Elapsed Time (seconds) 0 100 200 300 400 Lat. (ms) Squall (b) Latency (YCSB) Figure 11: Data Shuffling – Every partition either loses 10% of its tuples to another partition or receives tuples from another partition