Sail at Scale: Derived TPC-DS Benchmark at 1 TB and 3 TB

Over two decades ago, Hadoop pioneered open-source distributed processing by following the ideas in Google’s MapReduce paper. The design was disk-centric. A computation was modeled as a sequence of map and reduce stages, and the data exchanged between them went through a distributed file system such as HDFS. Spark challenged that model with resilient distributed datasets (RDDs), which keep intermediate data in memory when possible, track lineage, and recover a lost partition by redoing the computation. The shuffle between stages stayed disk-first, though. Shuffle data is written either to an executor’s local disk or to a remote shuffle service that is itself disk-based.

Rethinking Distributed Compute

When we started Sail, we envisioned another shift in distributed compute, toward doing even more of the work in memory. In a cloud-native environment, it is easy to provision abundant compute for a short period, and network bandwidth inside a data center can be higher than what disks deliver. Based on those observations, we built the first version of Sail’s distributed architecture in November 2024, with the Arrow Flight protocol carrying shuffle data between workers. Earlier this year, we redesigned the control plane of our clusters to make the driver and workers more modular. The results in this post are the latest evidence of its effectiveness.

Benchmark Results

Sail completes all 99 queries of the derived TPC-DS benchmark in 444.3 seconds at 1 TB and 1,288.8 seconds at 3 TB on eight r8gn.8xlarge workers that never write shuffle data to disk.

The queries ran sequentially and were timed end to end from the Spark Connect client, so every number includes planning, execution across the cluster, and the transfer of the result rows. The chart below shows total query times, and the table summarizes the results. The two charts after them show the time for each query, first at 1 TB and then at 3 TB. The full setup and configuration are in the appendix.

At 1 TB, 78 of the 99 queries finish in under five seconds, and the median query takes 2.6 seconds. The two longest queries, q23 and q14, take 38.1 and 33.8 seconds.

At 3 TB, the median query takes 7.8 seconds, and 69 of the 99 queries finish in under ten seconds. Processing three times as much data required 2.9 times as much total query time. The three longest queries are q85, q23, and q14 at 137.6, 106.1, and 92.1 seconds.

Arrow Flight Shuffle at Scale

The scalability of Arrow Flight shuffle may be counterintuitive at first glance. For two stages that each have p partitions, there are p × p logical shuffle streams between them. A complex workload with dozens of stages and a few hundred partitions, which is typical of the TPC-DS queries, can easily have more than a million logical streams active in the cluster. That does not mean the cluster needs that many network connections. Arrow Flight is built on gRPC and HTTP/2, and thanks to gRPC multiplexing, one TCP connection per pair of workers is enough.

In our experience, a single connection per worker pair does well even for terabyte-scale jobs. We also support configuring the connection count for the Arrow Flight client to further boost performance if a single connection has a bandwidth limit in your environment. For example, in AWS, one TCP connection for EC2 has an upper limit of 5 Gbps by default if advanced capabilities such as Elastic Network Adapter (ENA) are not configured.

To our knowledge, Sail was the first project in the distributed data processing space to explore Arrow Flight–based shuffle, beginning two years ago. We’re proud to see this direction now being explored in the DataFusion Comet and Riffle communities as well.

Memory First, Not Memory Only

Pipelined shuffle over Arrow Flight completes the horizontal scaling story for memory-centric analytics. As we reported before, Sail is excellent at vertical scaling and runs 10 times as fast as Spark on a single machine. Enterprise applications may need to work through terabytes of data or more. With Arrow Flight shuffle, you can provision a number of powerful nodes on a high-bandwidth network and never store intermediate data on disk. A task failure does require retrying the entire job under pipelined shuffle, but the cost of a retry is not what it was in disk-centric systems, because the job itself finishes in a fraction of the time.

Sail is not a memory-only engine, though. Many operators, including sorting operators, spill to disk when they need to. Sail also provides blocking shuffle backed by remote storage and by Apache Celeborn. If you operate in a resource-limited environment or node preemption is frequent, blocking shuffle limits the impact of a failure. That is how Sail serves as one compute engine from your laptop to the cloud while leaving you the choice of how to trade performance against resiliency.

Maturing a Distributed Engine

A distributed compute engine does not mature overnight, and looking back at the journey, we are thrilled with how far it has come. The numbers reported here include our recent progress on bulk task scheduling, Arrow Flight transport configuration, and the query optimizer. These improvements are available to our open-source users and to LakeSail Platform customers, helping production workloads at scale run faster and more reliably.

Getting Started

Install the pysail package and point your existing Spark SQL and DataFrame code at a Sail server. The Getting Started guide covers a single machine, and the Kubernetes deployment guide covers a cluster like the one in this post. You can follow along on GitHub or by joining our Slack Community.

Managed Sail in Your Cloud

Want to run Sail with managed infrastructure? The LakeSail Platform offers fully managed Sail in your own cloud account, with built-in governance, observability, and enterprise controls. Get started with a free trial and see how it can improve performance and reduce costs for your team. You can also schedule a free call with one of our engineers to ask any questions you might have.

Appendix

A. Benchmark Setup

TPC-DS is an industry-standard decision-support benchmark of 99 SQL queries over a retail data warehouse, covering multi-way joins, window functions, rollups, and multi-step common table expressions. We run a derived version using data that follows the standard schema and the SQLBench-DS query set, timed directly without an official TPC audit. These are not official TPC-DS results.

We ran the experiment on the LakeSail Platform, which uses Amazon EKS to manage Sail servers and clusters. The Sail server and each worker run on separate EC2 instances, and we fixed the number of workers so that resource allocation was predictable.

The data is stored on S3. We generated it with the Rust-based tpchgen-rs generator, with our own modification to write partitioned data. TPC-DS data generation is still a work in progress upstream, so that partitioning logic is not generally available yet. The dataset is stored in Parquet format with Snappy compression and 128 MB row groups.

Partitioning applies to the fact tables and the large dimension tables (catalog_sales, catalog_returns, store_sales, store_returns, web_sales, web_returns, inventory, and customer). The remaining small dimension tables have a single partition each.

• Software: We used a Sail Docker image built from commit 99ee46f69a97342e91bf7d4eaedb4509f1d8a9c2 of the main branch.
• Queries: We ran the 99 queries of SQLBench-DS, derived from TPC-DS under the TPC fair-use policy. Queries q14, q23, q24, and q39 each consist of two statements, so a full run comprises 103 statements.
• Hardware: We used eight worker nodes on AWS EC2 r8gn.8xlarge instances. Each worker provided 32 Graviton4 vCPUs, 256 GiB of memory, and 100 Gbps of network bandwidth, for a total of 256 vCPUs and 2 TiB of memory across the workers. The driver ran on an r8g.xlarge instance with 4 vCPUs and 32 GiB of memory.
• Disk: Each node had one 16 GiB gp3 root volume with 3,000 IOPS and 125 MiB/s of throughput. We used no instance storage and no separate volume for temporary files.
• Network: All nodes were in one availability zone with ENA Express enabled. We used no placement group.
• Client: We used PySpark over Spark Connect from a notebook on an r8g.xlarge instance in the same availability zone.
• Timing: We timed each statement at the client from submission to the arrival of its result rows, fetched with toArrow(). We report the best complete run with that client setup at each scale.
• Metrics: We measured client-side wall-clock time only. We collected no memory, disk, or network telemetry in this run.

Sail ran with the following settings, including explicitly configured values that match the defaults. See the configuration reference for details.

Here are four methodology notes.

• Cost-based join reordering and window top-N optimization are both marked experimental. We plan to enable join reordering by default once the optimizer implementation stabilizes.
• Parquet metadata and directory listings were cached between queries, per the cache settings above. Table data was read from S3 for every query.
• q85 took 1.9 seconds at 1 TB and 137.6 seconds at 3 TB. The query plan selected at 3 TB with experimental join reordering is the likely cause of this difference, although we have not yet confirmed it.
• This run measured Sail only. No other engine was run on the same cluster.

B. Spark and Spark Accelerators

We were not able to find authoritative TPC-DS results for Spark and Spark accelerators at these scales. The reports available online contradict one another, so we cannot draw a definitive comparative conclusion from them. We do note that on similar hardware configurations, Sail outperforms JVM Spark and Spark accelerators on the 1 TB and 3 TB TPC-DS datasets in similar settings.

A few long-running queries contribute substantially to the total time and narrow Sail’s overall performance gain. Spark and Spark accelerators benefit from query optimizers that have matured over many years, and that could inspire future work in Sail for even better performance. We recommend running controlled experiments on your own large-scale production workloads, and we would love to hear about your findings.

• Spark on Graviton R-series instances on EKS
• Spark with Celeborn on EKS
• Spark with Gluten and Velox on EKS
• Spark with Gluten, Velox, and Comet on EKS
• Spark RAPIDS on GPUs on EKS
• Spark with DataFusion Comet on EKS
• DataFusion Comet TPC-DS benchmark results