Major clouds operators like AWS, Azure, and GCP have embraced disaggregated storage architectures. Under such paradigms, raw data is uploaded to a highly-available object store such as AWS S3 or Azure Storage. When needed, data is fetched into a compute cluster to run compute-intensive jobs. While storage disaggregation helps improve resource utilization and elasticity in public clouds, it can hurt the performance of ad-hoc, interactive analytics queries on large objects. Prior work [47,93,99] shows that object stores are dominated by large objects and large object reads are prevalent. In one storage production trace from Microsoft [47], over 60% of objects have size greater than 1GB. Meanwhile, real-world workloads are highly selective. BigQuery [59,85] reports that about 50% of queries return less than 1% of the data. Transferring a large object over the network while eventually using only a small portion of its data not only wastes valuable network bandwidth but also signicantly degrades query performance.
To improve the performance of ad-hoc, interactive queries, analytics systems deploy a classic technique called computation pushdown, which ships compute closer to storage to avoid unnecessary data movement. Figure 1 depicts the highlevel architecture of computation pushdown for big data analytics in the cloud. Users can submit simple SQL-like queries to be executed closer to the object stores. Upstream database query engines can also push down parts of query plans to the object stores, such as highly-selective predicates and aggregates, thereby reducing the data volume sent to the data warehouse. In recent years, computation pushdown is embraced by data warehousing platforms [5, 55,72,105,109] and storage systems [7,11,16].
Despite its popularity, we argue that existing computation pushdown solutions on cloud object stores are inecient in the very common scenario of (a) erasure-coded data, and (b) data stored in analytics le formats. First, most cloud object stores [1, 35,47,76,87,93,103] use erasure coding due to its signicantly lower cost compared to replication. Objects are striped into xed-sized data and parity blocks and distributed across multiple storage nodes for fault tolerance. Data placement and recovery are done at the granularity of these data and parity blocks. Second, popular analytics le formats [3,4,10,39] allow queries to execute only on the relevant parts of a le without having to read the entire le. For example, Apache Parquet [4] partitions data into a collection of row groups and then each row group is partitioned column-wise into column chunks. A column chunk comprises values from a single attribute and directly supports common analytical operations, such as lters, projections, and aggregations. We refer to the granularity of a data unit capable of executing a part of the query (e.g., lter operations) on its own as the smallest computable unit.
Since erasure codes treat data as a blob of bytes, they are unaware of the internal semantics of the highly structured objects. Moreover, the smallest computable units tend to have variable sizes inuenced by several factors, such as the data type, cardinality and compression. Today, when an analytics le object is erasure coded into xed-sized blocks across storage nodes, each of its smallest computable units may be split across multiple nodes. Consequently, existing object stores such as MinIO [35] and Ceph [103] must rst reassemble these smallest computable units across storage nodes before executing a query. This wastes precious network bandwidth and network processing CPU in the storage cluster and also incurs extra network latency. The root cause of this sub-optimality is that erasure coding and computation pushdown in modern cloud storage operate in distinct layers of the storage stack, functioning within isolated silos.
We present Fusion, a new object store for analytics that is optimized for computation pushdown. It is designed to eciently handle ad-hoc, interactive analytics queries against object-based storage, akin to popular analytics systems like Amazon Athena [32] and BigQuery [34]. Fusion co-designs the erasure coding topology to take into account popular analytics le formats. One of our main contributions is a novel technique called le-format-aware coding (FAC), which prevents the splitting of the smallest computable units across nodes during erasure coding. Prior work [36] pads the original objects to enforce the alignment of these computable units with storage blocks, sharing a similar goal of preventing data splits. However, padding incurs a signicant storage overhead, negating the cost advantages of erasure coding.
FAC employs an intelligent stripe construction algorithm that departs from the conventional practice of using a xed block size per stripe [76,91,94]. Instead, it uses variable block sizes and stripe sizes to align erasure code blocks seamlessly with the boundaries of the smallest computable units. This creates a new challenge, as naively utilizing erasure coding with variable-sized data blocks dramatically increases storage overhead. To solve this storage overhead challenge, the stripe construction algorithm eciently bin-packs the smallest computable units, creating data blocks of similar sizes within a stripe, and dynamically adjusts stripe sizes based on the size distribution of the smallest computable units. FAC then utilizes this layout to erasure code the object and store each smallest computable unit intact among nodes.
Fusion uses a ne-grained pushdown mechanism to optimize query pushdown on FAC-encoded objects. Modern analytics le formats employ aggressive compression and encoding techniques on the smallest computable units to reduce storage costs. For instance, the smallest computable units in the TPC-H lineitem Parquet object can have compression ratios of up to 63. For queries involving such highly compressed smallest computable units, the pushdown operation, after decoding these smallest computable units locally on the storage node, may transfer a substantial portion of the uncompressed data as the query result over the network. This leads to higher network latency and degrades query performance. In such scenarios, it is advantageous to transfer the smallest computable unit over the network in its compressed form rather than processing it in-situ on the storage node. We nd that pushdown performance is directly inuenced by both the query's selectivity and the compressibility of the smallest computable units it operates on. We design a cost model to capture this relationship and Fusion uses it to determine which smallest computable units benet the most from query pushdown and enables pushdown only for those smallest computable units.
In summary, this paper makes the following contributions: 1. Novel coding technique (FAC) that co-designs erasure coding with analytics le format to maximize pushdowns.
2. Bin-packing algorithm for variable-sized blocks that minimizes storage overhead and prevents fragmentation of the smallest computable units. 3. A ne-grained, cost-based pushdown mechanism that maximizes pushdown performance. 4. Fusion, a new analytics object store, improves tail latency by up to 81% on TPC-H and 48% on real-world SQL queries, with only a 1.2% storage overhead w.r.t. the optimal. 2 Background Computation pushdown. Computation pushdown is a decades-old technique where parts of a query are "pushed down" closer to where the data is stored. It is prevalent in 1 MB data blocks parity blocks stripe 1 stripe 2 1 MB 1 MB 1 MB 1 MB 1 MB 1 MB 1 MB 1 MB 3 MB 0.5 MB 1 MB 0.5 MB 0.5 MB 0.5 MB 3 MB 3 MB 3 MB 12 MB erasure coding Figure 2. A (9, 6) erasure code partitions a 12MB object into two 6MB data stripes, each consisting of 6 data blocks and 3 parity blocks. Stripe 1 uses a xed block size of 1MB. Stripe 2 uses variable data block sizes. big data analytics systems that push down segments of SQL queries, such as lters (e.g., , WHERE clauses) and aggregations (e.g., SUM, AVG), to the storage layer. Some notable examples include AWS S3 Select [7] and Azure Data Lake Storage query acceleration [5]. Computation pushdown signicantly improves query latency by reducing the amount of data transferred over the network, thereby saving both network bandwidth and CPU resources for users.
Erasure-coded object stores. Distributed storage systems [1, 35,47,62,73,76,87,92,93,103,104] widely adopt systematic erasure codes to tolerate node failures with lower storage overhead compared to replication. An (=, :) systematic erasure code divides the original data into stripes, each containing : plaintext data blocks and (= :) coded parity blocks. Each stripe is stored across distinct nodes for maximal fault tolerance. This guarantees that the original stripe can be reconstructed as long as no more than (= :) of its blocks are lost. The conventional approach stripes data into uniformly sized data blocks and then generates parity blocks of the same size. This ensures the optimal storage overhead of (= : )
: . If a stripe contains variable-sized data blocks, erasure coding requires padding all other data blocks in that stripe to match the size of its largest data block and generates parity blocks of the same size. However, this can lead to increased storage overhead. Figure 2 illustrates the relationship between data block size and storage overhead. A (9, 6) erasure code encodes a 12MB object into two stripes, each containing 6MB of data. Stripe 1 divides the data into xed-sized blocks of 1MB, achieving the lowest storage overhead of 0.5⇥ (i.e., 3•1"⌫ 6"⌫ ). In stripe 2, data blocks have variable sizes and must be padded implicitly to match the largest data block size, which is 3MB, and parity blocks are generated accordingly. This results in a higher storage overhead of 1.5⇥ (i.e., 3•3"⌫ 6"⌫ ). Therefore, existing storage systems [76,91,94] typically partition data into xed-sized blocks before applying erasure coding to minimize storage overhead.
Systematic Reed-Solomon (RS) is the most prevalent erasure code used in modern storage systems, with two most [39], due to their ecient data storage and retrieval. These columnar formats use a hybrid data layout called Partition Attributes Across (PAX) [38] that horizontally partitions a table into a set of row groups and then vertically partitions a row group column-wise into column chunks laid out contiguously on disk. Figure 3 depicts the PAX data layout of Table 1, using a row group size of 3 rows. Compared to unstructured and semi-structured data formats such as CSV and JSON [28], PAX achieves better compression by grouping values of the same data type into column chunks. Parquet, by default, employs a variety of encoding and compression techniques (e.g., dictionary encoding, bit packing, and run-length encoding) on column chunks before writing them to disk. Each column chunk is a selfcontained data unit for encoding and compression. As shown in Figure 3, the employee names from the rst column chunk are rst mapped into a sequence of unique integer codes: "0", "1" , "2". Since there are only three distinct values, each integer code can be represented using only two bits. The names are further bit packed into a bit stream: "00", "01", "10". Finally, this column chunk is stored as a dictionary page followed by a number of encoded data pages on disk.
Moreover, PAX is ecient for columnar scans, lters, and projections as it retrieves only the necessary column chunks from disk. Consequently, a column chunk is the smallest { "Alice": 0, "B e": 2 } 00 01 10 ob": 1, "Charli node 1 Bob Alice Charlie network node 2 node 3 Figure 5. Processing a column chunk incurs non-trivial datareassembly cost over the network.
computable unit of PAX-based formats. For brevity, the remainder of the paper uses column chunks to denote the smallest computable units. In this work, we focus on the popular Parquet format as a case study, but our techniques are generalizable to other structured analytics le formats.
In this section, we discuss why computation pushdown is fundamentally limited in existing erasure-coded systems and highlight the challenges involved in improving its eciency.
Existing storage system architectures typically implement the analytics le formats and erasure coding in separate layers of the software stack. Consequently, erasure codes operate without any awareness of the internal semantics of the highly structured objects, treating them merely as a blob of bytes. They divide objects into xed-sized blocks, whose sizes are often pre-congured by the storage system, ranging from a few MBs to 100MBs to evenly distribute I/O [18,29,33,74,76,81,91,94]. As a result, column chunks of an object may be split across many nodes. Figure 4a depicts the percentage of column chunks in two representative Parquet les, namely the TPC-H lineitem table [31] and the NYC taxi rides dataset [13], that get split in existing storage systems. We experiment over a range of erasure code block sizes as reported by real production systems. Notably, even with a large block size of 100MB, the percentage of column chunk splits remains signicant, reaching up to 40% and 24% for the two datasets, respectively. As a result, the storage system must rst reassemble a column chunk across storage nodes before processing.
Running example. We use the following query as a motivating example to illustrate this problem. Assume the company wants to know Bob's current salary from Table 1:
As shown in Figure 5, column chunk 0 of row group 0 from Table 1 is split across three nodes. Although node 3 has the encoded data, it cannot directly run the WHERE clause to search for the name, Bob, since it cannot decode the encoded data without the dictionary. Unfortunately, the dictionary page is split across two other nodes. Therefore, the system must reassemble this column chunk across all three storage nodes before it can run the lter operation to nd matching rows whose name is Bob. Similar steps are repeated for the salary column. This can incur non-trivial network latency.
We measure the cost of data reassembly in current storage systems by executing the microbenchmark query outlined in §6 on the TPC-H lineitem Parquet le. The SQL query retrieves about 1% of column values. Figure 4b shows the latency breakdown of this query. We nd that the interested column chunks span across 5 storage nodes on average. Despite the large chunk sizes, disk reads contribute only a small fraction to the total query time. But 50% of the time is spent on network operations for transferring and reassembling fragmented parts of the column chunks.
The column chunk sizes are inuenced by various factors such as the data type and data cardinality (i.e., the number of column values). In addition, most columnar data formats integrate compression and encoding to signicantly reduce storage costs when writing column chunks to disk. Figure 6 shows the average compression ratio of column chunks from each column in the TPC-H lineitem Parquet le, with a median of 9.3 and a maximum of 63.5, respectively. The exact size of a column chunk also depends on the compressbility of its contents. Thus, column chunks within a PAX object can have variable sizes. Figure 4c shows the normalized size distribution of column chunks in four real-world Parquet les with compression enabled. Dierent datasets demonstrate distinct size distributions. For instance, in the TPC-H lineitem Parquet le, the majority of column chunks are either extremely small, representing highly repetitive integer values, or extremely large, representing diverse arbitrary string types. In contrast, the NYC taxi Parquet le exhibits a more uniform chunk size distribution since the dataset records user trip information that is more diverse in nature including pickup timestamp, trip duration, and distance.
To prevent splitting variable-sized column chunks across xed-sized erasure code blocks, prior work [36] proposes a padding solution that inserts additional padding into the original objects to enforce the alignment of column chunks with the underlying storage blocks. If placing a column chunk in the current block would lead to splitting, the padding solution lls the remaining space in the current block with extra padding and relocates that column chunk to the next erasure code block. However, the extra padding results in increased storage overhead. As depicted in Figure 4d, we measure the storage overhead of the padding approach w.r.t. the optimal strategy when storing real-world Parquet les. Surprisingly, the additional storage overhead can be very high, in some cases exceeding 100%.
To summarize, the mismatch between variable column chunk sizes and xed erasure code block sizes fundamentally limits the eciency of computation pushdown in existing systems. This sub-optimality arises because le formats and erasure coding operate in separate layers of the storage stack, functioning within isolated silos. This motivates our design of Fusion, a new analytics object store that maximizes the benets of computation pushdown, by co-designing its erasure coding to take into account the specic le format characteristics of the underlying data. Yet, incorporating le-format awareness into erasure coding is challenging, as seen in the padding solution, which incurs signicant storage overhead. Fusion departs from the conventional xed block-size-per-stripe approach used in most industry systems today. Instead, it utilizes a variable block-size-per-stripe
Table 2. Terminology. Term Description column chunk The smallest computable data unit in Parquet. data/parity block The smallest storage unit for data placement. stripe A set of data and parity blocks. The basic unit for data recovery and redundancy. bin Equivalent to a data block. binset Equivalent to all data blocks in a stripe. FAC Fusion's le-format-aware coding technique. approach that dynamically aligns erasure code blocks to column chunk boundaries within an object while minimizing storage overhead.
This section discusses Fusion's system components. Table 2 summarizes the key terminology.
We outline the main design principles that underpin Fusion's architecture and provide an overview of its design.
Use variable block sizes to prevent fragmentation of column chunks. While it is possible to resize column chunks to t some xed block sizes (e.g., by adjusting the row group size in Parquet), this alternative approach has several drawbacks. First, it incurs high CPU overhead. Resizing a column chunk (e.g., rewriting a large chunk into smaller ones) requires decoding and re-encoding the original data, which is computationally expensive [82,110]. Second, it is inexible. If storage block sizes change in the future due to system upgrades or optimizations, all column chunks must be regenerated to align with the new sizes. Third, altering the data format (e.g., with a dierent row group size) may create inconsistencies and compatibility issues, as clients expect to retrieve objects in the same format in which they were originally written to the storage service. To avoid these complications, Fusion stores column chunks in their original sizes without modication, and dynamically adjusts storage block sizes during erasure coding to prevent splitting any variable-sized column chunks.
Push down granular computations at the column chunk level to reduce network latency. A SQL query typically scans only a subset of table data (e.g., specic columns) and performs granular operations like lters and projections, which can run independently and in parallel on individual column chunks. Since Fusion stores each column chunk intact on a single storage node, its data layout is inherently optimized for computation pushdown at the column chunk level. Fusion utilizes a ne-grained pushdown mechanism that executes operations directly on the relevant column chunks and aggregates the computation results over the network. In contrast to existing object stores that read and reconstruct entire objects over the network, Fusion signicantly reduces network latency by minimizing I/O and network usage.
Figure 7 depicts the high-level architecture of Fusion. It consists of a coding module (FAC) and a query pushdown module. When an object is uploaded to Fusion, FAC identies the semantic boundaries of individual column chunks within the object based on its specied data format. It dynamically adjusts data block sizes per stripe to accommodate these variable-sized column chunks. It then constructs erasure code stripes and distribute them across storage nodes. FAC ensures that no column chunk is partitioned across data blocks, while simultaneously minimizing any additional storage overhead. When a SQL query arrives in Fusion, the query pushdown module transforms the original query into nergrained operations (e.g., lters and projections). It estimates the pushdown cost of individual operations and selectively pushes down those operations to nodes where the relevant column chunks reside, provided that doing so leads to reduced latency. Finally, it orchestrates the execution of individual operations and constructs the nal result back to clients. In Fusion, storage nodes are identical to each other and run both modules. Fusion does not have a dedicated coordinator. Each node acts as a coordinator for handling requests. Next, we dive deeper into the two modules.
Drawing insights from §3, we introduce the novel concept behind the coding module called le-format-aware coding (FAC). FAC leverages knowledge of an object's internal structure during erasure coding, making sure that column chunks are never split across data blocks. However, since column chunks exhibit variable sizes (Figure 4c), a naive way to construct erasure code stripes may lead to uneven data block sizes, resulting in higher storage overhead. Figure 8 illustrates this problem. The stripe has a total data size of 6MB. The larger column chunks are placed in the rst data block, data blocks parity blocks 5MB 5MB d3 d1 d2 p1 d6 d4 d5 p2 p3 Figure 8. A stripe of variable-sized data blocks under RS(9,6). Green boxes are column chunks. Dashed grey boxes are padding.
d1, with a total size of 5MB. In this case, an erasure code must pad all data blocks to the size of the largest data block and generate parity blocks of the same size. Its storage overhead is 2.5⇥, whereas the optimal storage overhead is 0.5⇥, achieved by dividing the stripe into 6 1MB data blocks and generating 3 1MB parity blocks. Hence, the primary objective of FAC is to minimize this additional storage overhead in the presence of variable-sized data blocks. Next, we present a formal denition of this problem.
Problem Formulation. Based on Figure 8, we make a key observation: the size of parity blocks in a stripe depends solely on the largest data block size within the same stripe. Therefore, minimizing storage overhead is equivalent to minimizing the sum of the largest data block size from all stripes. We model this challenge as a variant of the bin packing problem. Specically, we dene each data block as a bin with capacity ⇠, and all data blocks belonging to the same stripe as a bin set. Under a (=, :) erasure code, each bin set contains : bins. Given a list of # column chunks with sizes {B 1 , . . . , B # } and a total number of < bin sets, we want to assign column chunks into these bin sets such that the total sum of the largest bin size from all bin sets is minimized. Formally, we express the optimization objective as follows, where G 8 9; are binary variables 2 {0, 1} with 1 indicating that the 8-th column chunk is assigned to the 9-th bin of the ;-th bin set, and 0 otherwise: min (1) The rst constraint ensures that the total size of column chunks assigned to each bin does not exceed its capacity. The second constraint requires that each column chunk must be assigned to one and only one bin within some bin set. We set the bin capacity ⇠ as the maximum column chunk size such that each bin has sucient space to accommodate any individual column chunk. Since each bin hosts at least one column chunk, we set < to be # /:, the maximum number of required bin sets. Note that classic bin packing problems typically focus on minimizing the total number of bins used. After an extensive review of various bin packing variants, we found no exact match for our problem. The closest variant is minimizing the maximum bin size across all bins, also known as the multiprocessor scheduling problem [70]. However, this formulation diers from our objective, which aims to minimize the sum of the maximum bin sizes across all bin sets. To the best of our knowledge, our formulation introduces a new variant of the bin packing problem. §7 presents more related work on bin packing.
The Oracle. The above problem formulation belongs to the integer linear programming (ILP) class, which is known to be NP-complete. We implemented a solution using the Gurobi Optimizer [8], a popular commercial ILP solver. To assess its performance, we generated synthetic datasets by randomly sampling column chunk sizes between 1MB to 100MB. Figure 10a depicts its solve time as the number of column chunks increases. The ILP solver becomes impractically slow as problem size increases. Notably, with just 35 column chunks, Gurobi requires over 3 hours to compute an optimal solution. For context, real-world analytics les commonly comprise hundreds to thousands of column chunks, as illustrated in Table 3. Given that FAC operates on the critical path of Put operations, such excessive latency is unacceptable if this solver were to be integrated.
FAC Stripe Construction Algorithm. FAC uses an efcient, lightweight algorithm based on the insight that a stripe's storage overhead is determined once its largest data block is nalized. The algorithm constructs one stripe at a time and chooses the largest unassigned column chunk as the largest data block. It adheres to two principles when constructing the rest of the stripe. First, it proactively reduces overhead for future stripes by selecting larger column chunks for the current stripe. Doing so prevents these sizeable column chunks from becoming the largest data blocks in future stripes, incurring higher overhead. Second, it reduces wasted padding space within the current stripe by lling gaps in each data block, excluding the largest one, with smaller column chunks. This aligns the sizes of the remaining data blocks close to the largest block size.
Algorithm 1 Stripe Construction Algorithm 1: Procedure CS(:, 8C4<B = {B 1 , B 2 , . . . , B # }) 2: B83 0, BCA8?4B ; 3: Sort(8C4<B) // Sort item sizes in descending order 4: while 8C4<B not empty do 5: 18=B InitABinSet(: ) // Open an empty bin set 6: <0G Pop(8C4<B ) 7: Append(18=B [0],<0G) // Add the largest item to the rst bin 8: for 8C4< in 8C4<B do 9: 183 Return the bin id of the least occupied bin where 8C4< ts, or 1 otherwise 10: // Exclude the rst bin 11: if 183 > 0 then 12: Append(18=B [183 ], 8C4<) 13: end if 14: end for 15: BCA8?4B [B83 ] 18=B // Seal the current bin set 16: B83 B83 + 1 17: end while 18: Return BCA8?4B
Algorithm 1 presents the detailed algorithm. It starts with a queue of # unassigned column chunks from an object and sorts them in descending order based on size. In each iteration, it opens a bin set of : empty bins, pops the largest column chunk from the head of the queue, and places it into the rst bin. At this point, it seals the rst bin and sets the bin capacity ⇠ to be its size. This ensures that all other bins cannot grow larger than the rst bin. Next, it iterates over the queue. For each unassigned column chunk, it checks whether this column chunk can t into any of the bins, excluding the rst one. Among all bins with adequate space to accommodate the column chunk, it allocates the column chunk to the least occupied bin, aiming to achieve a more balanced distribution of load within the bin set. After a full scan of the queue, it seals the current bin set and proceeds to the next iteration. The above steps are repeated until all column chunks are assigned. At the end, the algorithm outputs < bin sets, each containing : bins, with each bin containing one or more column chunks. Figure 9 provides a visualization of the algorithm. During erasure coding, data blocks, along with parity blocks, are distributed across = randomly chosen storage nodes for each stripe.
The time complexity of the algorithm is $ (< • # ). It is linearly proportional to the number of column chunks within an object. We nd that FAC stripe construction algorithm runs extremely fast for real-world analytics le objects (i.e., 10s to 100s of microseconds). In contrast, the total put latency of analytics le typically ranges between 10s to 100s of seconds. As an example, uploading an 11GB TPC-H Parquet le to the baseline storage system takes 34 seconds. In comparison, FAC's algorithm completes in just 500 microseconds, constituting a negligible 0.0015% of the overall put latency. We provide more details on the runtime of our algorithm under realistic workloads in §6.3. The theoretical worst-case storage overhead of the algorithm under a (=, :) erasure code is (= :), which is the same as the replication approach that tolerates the same number of failures. This extreme scenario occurs in a stripe when one data block is very large, and the rest of the data blocks are negligibly small. Since the parity block size of a stripe always equals the size of its largest data block, the storage overhead of this stripe becomes comparable to that of replication. We introduce a system-level hyperparameter in Fusion, allowing users to specify the maximum additional storage overhead they can tolerate compared to the optimal. If the algorithm cannot construct stripes within the specied storage budget, Fusion defaults to erasure coding the object into xed-sized blocks, similar to existing storage systems. §6.3 conducts a comprehensive evaluation of storage overhead across diverse synthetic and real-world datasets. Empirically, we nd that FAC incurs an additional storage overhead of no more than 1.24% compared to the optimal scenario.
The query pushdown module manages the execution of SQL queries on FAC-encoded objects. It decomposes a query into sub-queries for individual column chunks and uses a cost model to determine whether to enable or disable pushdown at the column chunk level, ensuring maximum pushdown eciency. The pushed-down operations will run in-situ on the storage nodes where the relevant column chunks reside. Fusion employs this ne-grained adaptive pushdown mechanism because we nd query pushdown does not always reduce the network trac and improve query latency. In fact, pushdown eciency depends on query selectivity (i.e., the amount of data a query returns) and the compression ratios of column chunks. To better understand this, we rst describe how a SQL query is executed in Fusion with pushdown always enabled.
When a query arrives at a coordinator node in Fusion, it breaks down the query into ne-grained operations and executes them in two stages: a lter stage followed by a projection stage. In the lter stage, lter operations are rst forwarded to storage nodes hosting the column chunks to be ltered. A storage node will read the column chunk from disk, decompress and decode it, and then run the lter(s) on the decoded values. It returns a bitmap of the ltered results to the coordinator. The coordinator waits for all lter bitmaps and consolidates them into a nal bitmap. In the projection stage, the coordinator forwards the nal bitmap to all storage nodes hosting the column chunks to be projected. Each storage node reads the column chunk from disk, decompresses and decodes it, and then selects the values whose corresponding bits are set in the bitmap. Each sends back all selected values in the uncompressed form to the coordinator. Finally, the coordinator assembles all projection replies and sends the nal query result back to the client.
Our measurement on real-world datasets in Figure 6 shows that some columns are highly compressed, due to a combination of encoding and compression techniques used in PAX-based le formats. When query selectivity is high (i.e., returning many rows), the size of uncompressed values may greatly exceed the compressed column chunk size during the projection stage. This degrades query performance of Fusion since it sends more data over the network compared to existing systems that reassemble compressed column chunks on the coordinator. Figure 10b shows the parameter space where pushdown is eective. We dene the compressibility of a column chunk as the ratio of its uncompressed size to its compressed size. For queries with low selectivity (common in cloud analytics), pushdown improves query latency significantly. However, for queries with high selectivity or column chunks with high compressibility, pushdown degrades the latency.
Therefore, Fusion adopts a ne-grained adaptive approach to query pushdown. Rather than pushing down the entire query to storage nodes, Fusion estimates which column chunks will benet the most from query pushdown and enables pushdown only for those column chunks. Column chunks with unfavorable selectivity and compressibility values may incur high network costs. The coordinator will fetch them in compressed form and process them locally. The Pushdown Cost Estimator sub-module performs the cost estimation for column chunks. It decides to push down projections of a column chunk only when B4;42C8E8C~⇥ 2><?A4BB818;8C~< 1 , which we refer to as the Cost Equation.
The compressibility of a column chunk is estimated from the metadata of PAX-based objects (e.g., Parquet le footer). At the end of the the lter stage, the coordinator, knowing the exact query selectivity when constructing the nal lter bitmap, can assess whether the benet of pushing down projections on a column chunk outweighs the network overhead of transferring the uncompressed projection results. Specically, it uses query selectivity and the compressibility of column chunks to estimate the size of uncompressed projection results and pushes down projection operations only when the size of uncompressed results is smaller than the encoded column chunk size. Otherwise, Fusion disables projection pushdown for this column chunk and reassembles it at the coordinator instead.
Fusion is written in the Go language and uses Apache Arrow's Parquet library [2]. It supports three APIs, Put, Get and Query. Put writes an object to Fusion. Get retrieves the contents of the object with a given oset and size. Query runs a SQL query on the stored objects. We focus on Put and Query in this section as they are more interesting. Figure 11 shows the execution ow of Put and Query in Fusion. To achieve better load balancing, Fusion's implementation does not use a dedicated coordinator node for handling client requests. Instead, client requests are routed to a storage node based on the hash value of the object's name. Each node in Fusion acts as a coordinator for both Put and Query.
Storing Objects. When a Put request arrives at a coordinator node, the FAC module on the coordinator rst parses the object's metadata and identies the byte osets and sizes of all column chunks within the object. The stripe constructor then uses this information to nd an intelligent stripe layout, while keeping the storage overhead under a systemcongurable threshold. If no layout satises the threshold, it outputs the default layout that may split column chunks. Finally, FAC constructs erasure code stripes based on the selected layout, using some conventional erasure coding algorithm (e.g., Reed-Solomon), and persists the data and parity blocks across storage nodes. Note that even though FAC is designed to keep individual column chunks intact, the object is still distributed across the storage nodes, preserving the benets of IO and compute parallelism. Updates in Fusion are treated as fresh inserts.
Querying Objects. Fusion supports basic SQL query expressions on objects. When a Query request arrives at a coordinator, the SQL executor on the coordinator executes this query in two stages: the lter stage and projection stage. In the lter stage, the coordinator uses built-in object metadata to perform coarse-grained ltering and skips column chunks that do not satisfy the lter condition. For example, the Parquet le footer contains the min and max values for each column chunk within a row group, which speeds up ltering at the row group level. The coordinator then looks up Fusion's chunk location map to identify the storage nodes hosting the relevant column chunks and pushes down lter operations to those nodes. Each storage node constructs a bitmap for each column chunk, indicating the rows that satisfy the lter condition. It uses Snappy to compress bitmaps before sending them back to the coordinator. At the end of the lter stage, the coordinator aggregates all lter bitmaps and knows the exact query selectivity. The pushdown cost estimator uses the Cost Equation described in §4.3 to evaluate the potential benets of pushing down projection operations. Projections on column chunks that meet the criteria are pushed to the relevant storage nodes. The remaining column chunks are fetched and processed locally at the coordinator. At the end, the coordinator consolidates all projection results and returns the nal query result to the client.
Metadata Management Fusion keeps a chunk location map per object, tracking the storage node that hosts each column chunk. Every map entry is 8 bytes in size with 4 bytes for the column chunk oset within the object and 4 bytes for the storage node ID. The size of the location map is much smaller (a few KBs) compared to the overall object size, typically in GBs, as shown Table 3. Fusion must ensure the same durability for the location map as for the object itself because the system would be unable to retrieve the object if losing its map. In the current implementation, the coordinator replicates the location map to : + 1 storage nodes to tolerate the same number of failures as RS(=, :). In the future, Fusion will store location maps in a distributed storage service like ZooKeeper [23] or etcd [27].
Recovery and Fault Tolerance. Fusion does not modify the erasure code algorithms and follows the conventional recovery procedures for data reconstruction when one or more data blocks are lost. Therefore, Fusion provides the same level of fault tolerance as existing erasure codes. SQL Support. Fusion supports the SELECT SQL command and standard ANSI clauses i.e., SELECT, FROM and WHERE (similar to S3 Select [25]). It currently lacks support for aggregate pushdown such as SUM and AVG, which we aim to implement in the future to further enhance Fusion's performance.
Note that Fusion is designed as an analytics-optimized object store for ad-hoc, interactive queries. It is not a data warehouse solution like Snowake [55] or Amazon Redshift [72], which are specialized for operational workloads. Therefore, complex queries such as joins are excluded from Fusion, as they better run in a data warehouse. Meanwhile, data warehouses can still leverage Fusion by pushing down parts of the query plans to accelerate performance. We envision Fusion as a object store that eciently handles simple ad-hoc analytics queries and as pushdown engine that complements data warehousing systems.
We evaluate Fusion by answering the following questions:
• How does le-format-aware coding impact latency? ( §6.1) • How does Fusion perform on real-world SQL queries? ( §6.2)
• What are the overheads of FAC? ( §6.3) the incoming and outgoing bandwidth on each machine to 25Gbps for all experiments, except in Figure 14c.
Datasets. We use the TPC-H [31] and the NYC yellow taxi [13] datasets and convert them to Parquet format. TPC-H lineitem
table Parquet le is 10GB in size. It consists of 16 columns and 10 row groups. Each row group is 1GB and contains 30M rows. NYC yellow taxi Parquet le uses dataset from years 2015-2017 and is 8.4GB. It has 20 columns and 16 row groups, each of size 525MB and 25M rows. See Table 3 for details. All parquet les are generated using pyarrow [14] v13.0.0 and Parquet format 2.4, and have dictionary encoding and Snappy compression enabled. We create two datasets of 100GB and 84GB, respectively, by duplicating each Parquet le ten times. While cloud datasets can reach terabytes in size, individual les are typically tens of gigabytes in size. The Parquet le sizes we use are consistent with the real-world. Workloads. We run two sets of workloads: microbenchmark queries and real-world SQL queries. The microbenchmark executes a SQL query on the TPC-H lineitem table to retrieve a single column with a lter condition. Query selectivity by default is 1%, as observed in production systems [59, 85]. The only exception is the selectivity sweep evaluation in §6.1, where we vary query selectivity by changing the value eld: SELECT column FROM lineitem WHERE column < value
We also use four real-world SQL queries based on the TPC-H and taxi datasets (Table 4). Q1 and Q2 are the pricing summary report query and forecasting revenue change query from TPC-H [15]. Q3 and Q4 are two analytics queries from Timescale [12] on the taxi dataset, as shown below.
(1) How many rides took place every day in 2015? Baseline. We implement a baseline system representative of state-of-the-art systems such as MinIO [35] and Ceph [103], which erasure codes an object into xed-sized blocks and may split its column chunks. The baseline assembles the Parquet object on a coordinator node before executing the query on it. It also uses the optimization that leverages Parquet le metadata (i.e., the le footer) to retrieve only those column chunks that match the SQL query lter condition.
Performance metrics. We focus on median and tail latency in the performance analysis. Fusion targets at ad-hoc, interactive analytics queries where latency is the primary metric of interest. Leading industrial analytics systems such as Athena and BigQuery often report query completion time in their performance blogs [20][21][22]. Similarly, data analytics frameworks like Spark [109] and MapReduce [56] prioritize job completion time over throughput in their evaluations. In the context of data analytics, latency is a more meaningful metric due to the high variance in query runtime. and 81% for median and tail latency reduction. As shown in Figure 12, these column chunks are bigger in size and are more likely to get split in the baseline. As a result, the baseline must perform parallel reads from multiple storage nodes and transfer a substantial amount of data over the network. We closely examine one column as a case study to better understand why Fusion outperforms the baseline in this scenario. Figure 13c shows the query latency breakdown of column 5. Disk read involves the time spent reading the raw data from disk, while chunk processing encompasses the total time required for decoding data from Parquet format and evaluating SQL operations. Network overhead constitutes the combined latency from network transfer and the additional overhead introduced by remote procedure calls (RPCs). The chunks of column 5 are large in size (i.e., 165MB) and span across 2.6 storage nodes on average. Both system spend approximately the same amount of time on disk read and chunk processing. However, the baseline spends about 57% of its total time to reassemble column chunks over the network compared to Fusion, which processes the column chunks in-situ on the storage nodes and takes less than 4% of its total time to transfer the nal ltered result (i.e., 1% of all column values). In this case, the baseline transfers 2827GB of data over the network compared to Fusion, which transfers just 44GB, i.e., a 64.2⇥ reduction in network trac.
On the other hand, the latency improvements for columns 3, 4, 9, 10, and 11 are modest. These column chunks are smaller in size and exhibit higher compression ratio (see Figure 6). Consequently, the baseline system can read these chunks over the network, often from a single storage node, without incurring substantial network trac. Figure 13d shows the latency breakdown for column 9. Both systems spend no more than 3% of their total time on network operations. We nd that Fusion consistently achieves better performance gains under low query selectivity. In contrast, the performance of the baseline system varies signicantly and depends on a number of factors, including the column type and individual chunk sizes.
Query selectivity. Next, we vary query selectivity to understand its impact on latency. We conduct a selectivity sweep for column 5 and column 9, representing one of the better-performing columns and one of the worst-performing columns, respectively, in Fusion. Figure 14a and 14b shows the latency reduction of Fusion compared to the baseline. Recall the Cost Equation from §4.3. Fusion is expected to achieve the most performance gains when query selectivity and/or the compressibility of column chunks is low. The results support this intuition, as queries with lower selectivity exhibit the highest latency gains. Under high selectivity, such as 75% and 100%, Fusion identies that the size of uncompressed projection result signicantly exceeds that of the original compressed chunk to be projected. Hence, Fusion disables projection pushdown for the column chunk and falls back to the baseline mode, transferring the compressed chunk to the coordinator for further processing. It's worth noting that even under these extreme cases, Fusion still benets from the pushdown of lter operations.
Network. Figure 14c shows the median and tail latency reduction of Fusion compared to the baseline under various network congurations. Fusion achieves higher latency gains under a more constrained network (e.g., 10Gbps). Due to its high data transfer volume, the baseline experiences more performance degradation on a slower network. CPU utilization. Figure 14d compares the CPU utilization of both systems for dierent columns under a xed system load of 10 queries per sec. The CPU utilization is averaged across all storage nodes. Despite performing the same amount of computation work, Fusion utilizes less CPU since it transfers less data over the network. The ecient computation pushdown in Fusion enables it to save valuable CPU resources while delivering the same system throughput as the baseline.
Figure 15a illustrates the median and tail latency reduction of Fusion on real-world SQL queries. For Q1 and Q2, Fusion achieves a median latency reduction of up to 48% and a tail latency reduction of up to 40%. Both queries involve lter and projection operations across multiple columns. Fusion's le-format-aware coding approach (FAC) enables the direct pushdown of lter and projection operations onto the storage nodes hosting the column chunks, eectively reducing network overheads. On taxi-based queries, Fusion reduces the median latency by up to 32% and tail latency by up to 48%. For Q3, despite the high selectivity (i.e., 37.5%), the date column has a low compression ratio of 1.6. Fusion still oers latency gains since the selectivity and compressibility product is 0.6, which is less than 1, making it suitable for pushdown according to the Cost Equation. Q4 involves two projection columns, namely date and fare. However, the fare column has a high compression ratio of 152. The product of selectivity and compressibility for this column far exceeds 1, leading Fusion to disable the projection pushdown. Despite this, Fusion still outperforms the baseline due to its eective pushdown of the date column. Figure 15b shows the total network trac usage in Fusion compared to the baseline system. Fusion generates upto 8.9⇥ lower network trac due to its pushdown friendly data layout.
Next we evaluate the storage and runtime overhead of FAC on both synthetic and real-world datasets. The optimal storage overhead under RS(9,6) is 0.5⇥ (i.e., 9 6 6 ). We report the storage overhead of FAC as a percentage, representing its additional overhead relative to the optimal. The synthetic datasets simulate a wide range of le congurations. allowing for a comprehensive evaluation of storage overhead. Specically, we vary the number of column chunks and the skewness of the chunk size distribution. Each synthetic dataset consists of chunk sizes between 1MB to 100MB, chosen from a Zipan distribution. Each data point in Figure 16a is the averaged storage overhead over 100 dataset runs. Surprisingly, we nd that the chunk size distribution has little impact on FAC's storage overhead. When the chunk sizes are highly skewed (e.g., Zipan 0.99), the FAC algorithm can eectively group the majority of chunks with similar sizes together. The remaining are packed into a small number of uneven data blocks, with negligible storage overhead. When chunk sizes are random (e.g., Zipan 0), the algorithm benets from high variance and can distribute chunks more uniformly across data blocks, increasing the likelihood of bin-packing into similar-sized blocks. We nd that the storage eciency of FAC is directly inuenced by the number of chunks. A higher number of chunks results in a larger search space, oering more satisfactory solutions and increasing the likelihood of nding a good one. As illustrated in Figure 16a, FAC's storage overhead decreases to 3% with 100 chunks, drops to 0.8% with 500 chunks and approaches closer to 0 for higher values. For large objects with more column chunks, the storage overhead of FAC tends to converge towards the optimal. FAC's storage overhead under RS(14, 10) exhibits a similar pattern, which we omit here due to space limits. Next, we measure the storage overhead of FAC on four real-world Parquet les, with the number of column chunks ranging from 84 to 320 (Table 3). We compare FAC against two alternative approaches: Oracle, the Gurobi-based ILP solution, and Padding, proposed by Adams et al. [36], which inserts extra padding into the original object to enforce data alignment with erasure code blocks. Figure 16b shows the additional storage overhead. Figure 16c shows the runtime overhead relative to the the total time of the Put operation when writing the le to the baseline storage system. While Oracle achieves minimal storage overhead, its runtime is prohibitively high, up to 3.91⇥ the overall Put latency for uk pp. Padding runs faster but incurs a substantial storage overhead of up to 83.8% for recipeNLG. In contrast, FAC's stripe construction algorithm strikes a more favorable balance between storage overhead and runtime. It incurs a storage overhead of no more than 1.24% relative to the optimal and a runtime overhead of no more than 0.0027%, respectively. Previous studies [47,93,99] suggest that large le objects in gigabytes dominate cloud storage. Our empirical ndings lead us to believe that Fusion would result in little extra storage overhead when deployed in production.
Computation Pushdown. Both systems and database research has actively explored computation pushdown techniques. Recent systems [43,79,107] enable the pushdown of arbitrary functions to storage nodes, reducing network round trips, particularly for multi-hop traversal queries like B-tree index lookups. Unlike Fusion, they are in-memory systems and do not consider erasure coding. New computational storage devices (CSDs) [30,36,58] allow application code to run directly inside the disk. Adams et al. [36] propose an application-level solution that pads objects to align their internal data units with erasure code blocks but may incur signicant storage overhead. In contrast, Fusion operates within the storage layer and minimizes storage overhead. Cloud databases also employ predicate pushdown to speed up queries and reduce network trac [7, 24,26,72,105,106,108]. FlexPushdownDB [105] takes a hybrid approach that combines selective caching and query pushdown. It does not address pushdown eciency in storage systems. To the best of our knowledge, Fusion is the rst analytics object store that improves computation pushdown on structured analytics les stored in an erasure-coded form.
Analytics le formats. Analytics le formats can be categorized as unstructured (plaintext, binary les), semi-structured (CSV, JSON [28]), and structured ( [3, 4, 39]). PAX-based [38] structured le formats, such as Parquet [4] and ORC [39], are widely used in data analytics. Albis [102] is another structured format that disables compression and encoding for better query performance. Fusion leverages the structural knowledge embedded in these formats to erasure code data into a computation-friendly layout.
Erasure coding. Systematic erasure codes are widely deployed in public clouds [1, 19,35,42,62,76,80,87,100,103] due to their advantage of storing data blocks in plaintext. Fusion currently supports systematic Reed-Solomon (RS) and can easily incorporate other systematic codes. Nonsystematic codes are not supported since they store data blocks in encoded form, making direct computations on storage nodes infeasible. Prior work on erasure codes primarily focuses on improving repair performance through new classes of codes, such as regenerating codes [57] and locally repairable codes [69,75,76,90,95], which are orthogonal to the focus of Fusion.
Bin packing and its variants. The bin packing problem has been extensively studied for decades and has numerous variants, including heterogeneous bin types [65], variable bin sizes [44,53,64,77,96,97], item fragmentation [61,84,86,98], and dierent optimization objectives [40,51,54,68] and constraints [37,41,63,66,78]. The problem space can be categorized into one-dimensional [50,52,67] and multidimensional [46,48,49,60]. In the multidimensional case, item sizes and bin capacities span multiple dimensions and must satisfy constraints across all dimensions. The bin packing problem has many important applications in the cloud and data centers, such as resource-constrained scheduling [45,66,71,88], and virtual machine placement [83,89,101]. These applications usually use online approximation algorithms since items arrive on-demand, and future items are not known in advance. Fusion tackles a new variant of the one-dimensional bin packing problem. It utilizes a greedy-based oine algorithm, as all items (i.e., column chunks) become available once an analytics object is uploaded.
SummaryModern erasure-coded cloud object stores are not optimized for computation pushdown. By co-designing erasure coding and le placement topology, Fusion signicantly reduces query latency for big data analytics with low storage overhead. We believe Fusion's co-design approach can benet other system areas that use erasure codes for data storage.