---
description: https://arxiv.org/pdf/2002.05837.pdf
---

# PushdownDB: Accelerating a DBMS using S3 Computation



1. **What is the problem that is being solved?**

* Study the effectiveness of pushing parts of DMBS analytics queries into the Simple Storage Service (S3) engine of AWS, using a capability called S3 select.&#x20;
  * Primitive (filter, projection, aggregation): can always be cost effectively moved into S3
  * Complex (join, top-K, group-by) require reimplementation to take advantage of S3 Select and are often candidates of pushdown.&#x20;

&#x20;   2\. **What are the metrics of success?**

* Cost&#x20;
* Query latency&#x20;

&#x20;  3\. **What are the key innovations over prior work? (motivations**)&#x20;

* Cloud: disaggregated architecture, separate compute and storage&#x20;
  * Problem: network can be a major performance bottleneck&#x20;
  * Intuitive solutions
    * _caching:_ compute server loads data from the remote storage once, caches it in main memory or local storage, and reuses it across multiple queries, thereby amortizing the network transfer cost. &#x20;
    * _computation pushdown:_  push functionality as close to storage as possible&#x20;
      * E.x. push computation into specialized processors that are closer to storage&#x20;
* AWS: S3 select -- limited computation can be pushed onto S3&#x20;
  * Problem
    * Limited computational interface of S3 allows only simple query operators to be pushed down&#x20;
      * Select, project, aggregation&#x20;
    * Other: requires new implementation&#x20;
    * S3 Select pricing can be more expensive than EC2 nodes&#x20;
* This paper:&#x20;
  * Contribution: first extensive study of pushdown computing for database operators in a disaggregated architecture.&#x20;

&#x20;  4\. **Specific things about the paper (notes)**&#x20;

#### Data management in the cloud&#x20;

* Computing services: Elastic Compute Cloud (EC2)&#x20;
  * instances&#x20;
  * Locally-attached storage&#x20;
* Storage services: Simple Storage Service (S3)&#x20;
  * Highly available&#x20;
  * Virtually infinite storage capacity with relatively low cost&#x20;
  * Can be shared across multiple computing instances&#x20;
  * Cheaper than locally-attached and/or block-based alternatives (Elastic Block Store (EBS))&#x20;
* S3 select&#x20;
  * 2018
  * Implement operators by scanning the rows in the table and returning qualifying rows to the compute node&#x20;
  * Not support: join, group by, top-K&#x20;
    * Challenging to redesign these operators to use S3 select&#x20;
      * E.x. Join: data shuffling among storage nodes&#x20;
* Computing query cost&#x20;
  * AWS: cost varies based on the region where the users data and computation are located&#x20;
  * **Storage cost**
    * Charged monthly based on the amount of space used: $0.022/GB/Month&#x20;
    * Only depends on data size and not on frequency of access&#x20;
  * **Data transfer cost**
    * S3 users are charged for only the outgoing traffic and the prices is based on the destination of data&#x20;
    * S3 Select not used: free (transferring data within the same region) to $0.09/GB (transferring data out of AWS)&#x20;
  * **S3 Select cost**
    * Based on the amount of data scanned ($0.002/GB) in processing an S3 Select query and the amount of data returned ($0.0007/GB)
    * Depends on the selectivity of the query&#x20;
    * Data scan and transfer cost is the major component in overall query cost&#x20;
  * **Network request cost**&#x20;
    * Issuing HTTP requests to S3 is based on the request type and the number of requests&#x20;
    * Cost paid for both S3 Select requests and conventional table read requests&#x20;
  * **Computation cost**&#x20;
    * EC2 memory-optimized instances&#x20;
    * Query execution time --> computational cost based on the hourly price of the host EC instance&#x20;
      * r4.8xlarge instance costs $2.128 per hour&#x20;

#### PushdownDB database&#x20;

* row-based DMBS testbed&#x20;
  * Minimal optimizer and an executor&#x20;
* Query plan: directed acyclic graph, and executes queries in either serial or parallel mode&#x20;
  * Serial: single CPU executes one operator at a time
    * Some can benefit. I.e. projection followed by a filter (avoids inter-process data transfers)&#x20;
  * Parallel: multiple processes and passes batches of tuples from producer and consumer using a queue&#x20;
    * Most can achieve better performance&#x20;

#### Operators&#x20;

* Filter ("where")
  * Indexing with S3 Select&#x20;
    * Problem: Hash indexes, tree-based indexes are both not a good fit for cloud storage environment because single lookup requires multiple accesses to the index. Multiple S3 requests that incur long network delays.&#x20;
    * Design: Index table
      * Contains&#x20;
        * Values of the columns to be indexed
        * Byte offsets of indexed records in that table&#x20;
      * Accessing a table through an index (two phases)
        * **Phase 1**: predicate on the indexed columns is sent to the index table using an S3 Select request. Then the byte offsets of selected rows are returned to the compute server.&#x20;
        * **Phase 2**: byte offsets are used to directly load the corresponding rows from the data table, by sending an HTTP request for each individual row.&#x20;
          * Note: not using S3 select&#x20;

![The schema of the index table ](https://2097630930-files.gitbook.io/~/files/v0/b/gitbook-legacy-files/o/assets%2F-MVORxAomcgtzVVUqmws%2F-MXSXPiRGIRh_u054Vov%2F-MXSgijfxD7LZXvLyp10%2Fimage.png?alt=media\&token=c8ba2853-d663-4dcf-bfcf-0a7057790e74)

* Join
  * Not support to push a join operator in its entirety into S3&#x20;
  * Why difficult to pushdown processing for joins&#x20;
    * Two tables to be joined are partitioned across multiple S3 objects (can load data in parallel)&#x20;
    * If not partitioned on the join key, needs to shuffling data across partitions. Challenging to support at the storage layer.&#x20;
  * Hash join&#x20;
    * Build phase: loads smaller table in parallel and sends each tuple to the appropriate partition to build a hash table&#x20;
    * Probe phase: loads bigger table in parallel and sends tuples to the correct partition to join matching tuples by probing the hash table&#x20;
  * Algorithm that PushdownDB supports&#x20;
    * Baseline Join
      * the server loads both tables from S3 and executes the hash join locally, without using S3 Select
    * Filtered Join
      * pushes down selection and projection using S3 Select, and executes the rest of the query in the same way as baseline join.
    * Bloom Join (\*)
      * after the build phase, a Bloom filter is constructed based on the join keys in the first table; the Bloom filter is then sent as an S3 Select request to load a filtered version of the second table
      * Support only integer join attributes&#x20;
        * A more general support for hashing in S3 Select API would enable bloom joins on arbitrary attributes&#x20;
        * ... Needs extension of S3 Select interface&#x20;
* Group-by&#x20;
  * **Server-side group-by**: perform at the server-side by loading all data from S3 directly&#x20;
  * **Filtered group-by**: loading S3 data using a predicate&#x20;
  * **S3-side group-by**: pushes the group-by logic entirely into S3 thus minimize the amount of network traffic&#x20;
    * First phase: collects the values for the groups in the group-by clause&#x20;
    * Second phase: requests S3 to perform aggregation for each individual group that the first phase identified.&#x20;
  * **Hybrid group-by**&#x20;
    * Practice: data sets highly skewed (a few large groups contains majority of the rows)&#x20;
      * Can lead to long S3 Select queries in S3-side group=-by&#x20;
    * Propose: distinguish groups based on their size
      * pushes the aggregation on large groups to S3, eliminating the need for transferring large amount of data&#x20;
      * small groups: are aggregated by the query execution nodes&#x20;
    * Phases&#x20;
      * First: only a sample of rows to capture the populous groups (scan the first 1% of data from table)&#x20;
      * Second: two queries sent to S3. Q1 runs remote aggregation for the large groups. Q2 is sent for loading rows belonging to the rest of the groups from S3.&#x20;
* Top-K&#x20;
  * Sampling-Based Top-K Algorithm
    * First phase: samples the records from the table and decides what subset of records to load in the second phase&#x20;
    * Second phase: query execution node loads this subset of records and performs the top-K computation on it&#x20;

&#x20;  5\. **What are the key results?**

#### Filter&#x20;

![Server-side filter loads the entire table from S3 and performs filtering on the compute node. S3-side filter sends the filtering predicate to S3 in an S3 Select request. S3-side indexing uses the index table implementation.](https://2097630930-files.gitbook.io/~/files/v0/b/gitbook-legacy-files/o/assets%2F-MVORxAomcgtzVVUqmws%2F-MXSXPiRGIRh_u054Vov%2F-MXShsNT5bpdrva-wqP3%2Fimage.png?alt=media\&token=1e5b7cfa-e5bc-424e-bfcc-bcfd654c4758)

Take-away

* Runtime: S3-side indexing has similar performance as S3-side filter when the filter is highly selective, but performance of indexing degrades as the filter selects more than 10^{-4} of the rows.&#x20;
  * In this case, most of the execution time is spent requesting and receiving individual byte ranges from the data table&#x20;
    * Though in parallel, they incur excessive CPU computation that become a performance bottleneck&#x20;
* Cost:
  * S3-side filter is 24% more expensive than server-side filter
    * S3-side: data scanning and loading
    * Server-side: computation  &#x20;
  * S3-side indexing is cheaper than server-side filter by 2.7x when filter is very selective&#x20;
    * Reason: index table reduces the amount of data being scanned and transferred&#x20;
* In conclusion, S3-side indexing is the best approach with highly selective queries, whereas S3-side filter achieves a good balance between performance and cost for queries with any selectivity.

#### Join

1. Customer Selectivity&#x20;

![](https://2097630930-files.gitbook.io/~/files/v0/b/gitbook-legacy-files/o/assets%2F-MVORxAomcgtzVVUqmws%2F-MXSXPiRGIRh_u054Vov%2F-MXSqQt_8GsOpQVkOQLB%2Fimage.png?alt=media\&token=eaf3d4c1-a0f6-4885-a53c-01f46b48cff5)

![](https://2097630930-files.gitbook.io/~/files/v0/b/gitbook-legacy-files/o/assets%2F-MVORxAomcgtzVVUqmws%2F-MXSXPiRGIRh_u054Vov%2F-MXSqUQ3HXbNNkg5CpM8%2Fimage.png?alt=media\&token=661a4107-f606-4023-b80c-21a4016d68c4)

Take-away

* Runtime:&#x20;
  * baseline and filtered joins perform similarly, which is expected since they only apply selection to the smaller customer table and load the entire orders table, which incurs the same large amount of network traffic&#x20;
  * Bloom join: significantly better than either&#x20;
    * High selectivity on the first table is encapsulated by the Bloom filter, which significantly reduces the number of returned rows for the larger orders table&#x20;

2\. Orders Selectivity&#x20;

![](https://2097630930-files.gitbook.io/~/files/v0/b/gitbook-legacy-files/o/assets%2F-MVORxAomcgtzVVUqmws%2F-MXSXPiRGIRh_u054Vov%2F-MXSrqnYK_4tbuJUDJ9C%2Fimage.png?alt=media\&token=1d467ec7-e7e4-4b7e-ad78-64db79ba757e)

Take-away:

* Filtered join performs significantly better than baseline join when the filter on the orders table is selective&#x20;
  * Performance advantage disappeared when the filter is less selective&#x20;
* Bloom join: performs better and remains fairly constant as the number of records returned from the orders table remains small due to the Bloom filter
  * Cost: comparable or cheaper than alternatives&#x20;

![](https://2097630930-files.gitbook.io/~/files/v0/b/gitbook-legacy-files/o/assets%2F-MVORxAomcgtzVVUqmws%2F-MXSXPiRGIRh_u054Vov%2F-MXSsVCjQTykpaGARBau%2Fimage.png?alt=media\&token=8ac7a3a7-2c1a-4447-a45b-75a622b79cfd)

* We can see that the best performance and cost numbers can be achieved when the false positive rate is 0.01. When the false positive rate is low, the Bloom filter is large in size, increasing the computation requirement in S3 Select. When the false positive rate is high, the Bloom filter is less selective, meaning more data will be returned from S3.&#x20;

#### Group-by

![](https://2097630930-files.gitbook.io/~/files/v0/b/gitbook-legacy-files/o/assets%2F-MVORxAomcgtzVVUqmws%2F-MXSXPiRGIRh_u054Vov%2F-MXSutaYVztsZnzGlMOf%2Fimage.png?alt=media\&token=7c2d1811-e8fd-4dfd-a5a4-25c52d478024)

![](https://2097630930-files.gitbook.io/~/files/v0/b/gitbook-legacy-files/o/assets%2F-MVORxAomcgtzVVUqmws%2F-MXSXPiRGIRh_u054Vov%2F-MXSuxIHE52ilsTcGLJh%2Fimage.png?alt=media\&token=09811ad0-890a-4655-a6d0-9f47632a0c88)

* S3-side group-by performs 4.1× better than filtered group-by when there are only a few unique groups. Performance degrades, however, when more groups exist. This is due to the increased computation overhead that is performed by the S3 servers
* The server-side group-by pays more for compute, but the other two algorithms pay more for scanning and transferring S3 data

![](https://2097630930-files.gitbook.io/~/files/v0/b/gitbook-legacy-files/o/assets%2F-MVORxAomcgtzVVUqmws%2F-MXSXPiRGIRh_u054Vov%2F-MXSvdFMweNHaNxXaIaD%2Fimage.png?alt=media\&token=4d3ef8ae-ccc5-4d47-8641-bcb4adb9c73e)

#### Top-K

![](https://2097630930-files.gitbook.io/~/files/v0/b/gitbook-legacy-files/o/assets%2F-MVORxAomcgtzVVUqmws%2F-MXSXPiRGIRh_u054Vov%2F-MXSz9Xa-37uYG-t6MUC%2Fimage.png?alt=media\&token=f2f4f534-fe22-4119-9c06-7f9d30b47b86)

![](https://2097630930-files.gitbook.io/~/files/v0/b/gitbook-legacy-files/o/assets%2F-MVORxAomcgtzVVUqmws%2F-MXSXPiRGIRh_u054Vov%2F-MXSzBWSAJZndtZyQEFj%2Fimage.png?alt=media\&token=d9adeb45-4fab-4e1f-b651-9f3e17530c4f)

Take-away:

* As the sample size increases
  * the execution time of the sampling phase also increases. This is expected because more data needs to be sampled and returned
  * the execution time of the scanning phase decreases. This is because a larger sample leads to a more stringent threshold, and therefore fewer qualified rows in the scanning phase.
* The amount of data returned from S3 first decreases due to the dropping S3 traffic in the scanning phase, and later increases due to the growing traffic of the sampling phase.

Figure 9:

* for both algorithms, runtime increases as K increases. This is because a larger K requires a bigger heap and also more computation at the server side. The sampling-based top-K algorithm is consistently faster than the server-side top-K due to the reduction in the amount of data loaded from S3.
* the sampling-based top-K algorithm is also consistently cheaper than server-side top-K.&#x20;

#### Overall

![](https://2097630930-files.gitbook.io/~/files/v0/b/gitbook-legacy-files/o/assets%2F-MVORxAomcgtzVVUqmws%2F-MXSXPiRGIRh_u054Vov%2F-MXT-7dApY4kqH-KQaT2%2Fimage.png?alt=media\&token=575222a2-4066-4e77-ac74-11ecc1b50b5a)

* Baseline: PushdownDB implementation but not include S3 Select features. The server loads the entire table from S3 and performs computation locally.&#x20;
* Optimized: optimizations discussed in the paper.&#x20;
* On average, the optimized PushdownDB outperforms the baseline PushdownDB by 6.7× and reduces the cost by 30%.

#### Experiment with Parquet&#x20;

![](https://2097630930-files.gitbook.io/~/files/v0/b/gitbook-legacy-files/o/assets%2F-MVORxAomcgtzVVUqmws%2F-MXSXPiRGIRh_u054Vov%2F-MXT-_dvU4PpI97K38Zm%2Fimage.png?alt=media\&token=cadd8939-2c40-4663-8637-5ddcc3460497)

* Parquet substantially outperforms CSV in the 10 and 20 column cases, where the query requests a small fraction of columns.
  * our query scans only a single column of Parquet data but has to scan the entire CSV file — Parquet outperforms CSV due to less IO overhead on S3 Select servers.
* the performance advantage of Parquet over CSV is more prominent when the filter is more selective — when more data passes through, data transfer becomes the bottleneck so CSV and Parquet achieve similar performance. This

&#x20;  6\. **What are some of the limitations and how might this work be improved?**

* Limitations and suggestions to improve S3 Select&#x20;
  * **Suggestion 1: multiple byte ranges for GET requests**
    * Now: current GET request to S3 supports only a single byte range. This means that a large number of GET requests have to be sent if many records are selected by a query.&#x20;
    * Want to: allow a single GET request to contain multiple byte ranges can significantly reduce to cost of HTTP request processing in both the server and S3 &#x20;
  * **Suggestion 2: Index inside S3**&#x20;
    * Build index structure entirely inside S3 avoids network messages between S3 and the server that are caused by accesses to the index data structure during an index loopup&#x20;
  * **Suggestion 3: More efficient Bloom filters**&#x20;
    * Current S3 Select does not support bit-wise operators&#x20;
    * Support efficient bit-wise operators can improve efficiency of bloom join&#x20;
  * **Suggestion 4: Partial group-by**
    * Now, CASE clause to implement S3-side group-by, not efficient&#x20;
    * Suggest adding partial group-by queries to S3 to resolve this performance issue&#x20;
  * **Suggestion 5: Computation-aware pricing**&#x20;
    * Now: S3 Select Pricing model fixes the amount of data scanning costs regardless of what computation is being performed&#x20;
    * Suggest that the data scan cost should depend on the workload&#x20;

&#x20;  7\. **How might this work have long term impact?**

* presents PushdownDB, a data analytics engine that accelerates common database operators by performing computation in S3 via S3 Select. PushdownDB reduces both runtime and cost for a wide range of operators, including filter, project, join, group-by, and top-K.&#x20;
* First extensive study of pushdown computing for database operators in a disaggregated architecture.&#x20;



#### Some notes on related works: Near-Data Processing (NDP)&#x20;

* Processing-in-Memory (PIM) exploits computation near or inside DRAM devices to reduce data transfer between CPU and main memory&#x20;
* The development of FPGAs and SSDs in recent years has made near storage computing more practical
  * Studies: push computation to both near-storage FPGAs and the processor within an SSD device&#x20;
    * But focus on simple operators like filter or projection, but did not study the effect of more complex operators &#x20;
* Hybrid shipping techniques execute some query operators at the client side, where the query is invoked, and some at the server side, where data is stored
  * not consider how to push down only some of the steps involved in the implementation of a single operator&#x20;
