1 of 17

Reducing query latency in DataFusion via a caching object store layer

Artjoms Iškovs,�Principal Engineer �27th September 2024

2 of 17

Before EDB: Splitgraph and Seafowl

Seafowl: “analytics at the edge”
Released in 2022, uses DataFusion
Kind of like PostgREST + Varnish for OLAP

Now, EDB, or EnterpriseDB has always been known for its Postgres work, there’s a bunch of people from EDB who are on the Postgres Core team or are PG committers. So what are we doing with DataFusion? It all started in 2022 when Miles and Marko and I were working on a startup called Splitgraph, which was like a data platform with some analytics stuff included, and it heavily relied on Postgres. But in 2022 we also discovered DataFusion and on top of that we built Seafowl, which was like a database designed for interactive data applications on the Web. So this is an illustration I pulled from an old slide deck.

The idea was that you could have this read-only SQL API, you deploy that somewhere, you can even colocate it with data, so you’d have Parquet files living in the same Docker image, and then your client-side application could send analytics queries to it straight from the browser. And we designed the API to be cache-friendly, so that you could put it behind Cloudflare or Varnish.

And it was all powered by DataFusion, so we got to learn a lot about its pieces and mainly Marko started contributing to it then.

3 of 17

At EDB: Postgres Lakehouse + HTAP

In 2023 we got acquired by EDB as part of their bigger strategy to expand from transactional databases to analytics. And the thing we’re building at EDB is HTAP: hybrid transactional-analytical processing. The idea is that something like Postgres is great at transactional workloads, consistency, indexing, locking etc. But it’s never going to be as fast at analytical queries, queries that have to aggregate through a large amount of data, as a dedicated vectorized query engine like DataFusion that uses a columnar format.

So we want to couple the two. This HTAP product will replicate data and changes from Postgres to a columnar store and let the user accelerate analytics-shaped queries, on offloaded data, or even let you bring your own data from some third-party writer. And here, Seafowl has moved away from its Web roots. We use DataFusion on both sides, reading and writing. Marko will talk more about writing and replicating changes in his talk, but I’ll talk about reading.

We use a so-called lakehouse architecture where we store columnar data in cheap object storage and then you can spin up compute nodes on demand or scale them up and down. The lakehouse format we chose right now is Delta Tables just because that’s what Seafowl used to use, but we’re also looking at Iceberg.

4 of 17

5 of 17

So, how does this interact with DataFusion? Well, here’s a DataFusion query plan for a TPC-H query. The Delta Table is a DataFusion TableProvider and at plan time, we can go to the Delta library and say “OK, I have these filters and this table path. What Parquet files do I need to read through?” In its transaction log, Delta also stores some metadata that lets us do partition pruning, we can look at the filter and discard some files altogether. And all the Delta stuff gets replaced by a Parquet Scan through a set of files, with a certain set of filters and columns, with these files being backed by an ObjectStore provider.

The ObjectStore is another abstraction where it lets you get a file, or get a certain chunk of a file, or write a file. And it can be backed by an actual object store, or by a filesystem.

6 of 17

Multiple files
Small byte ranges
Big byte ranges
(partially) overlapping byte ranges
Byte ranges requested multiple times

…from different parts of the query tree
…from different queries
…concurrently?!

So what’s the problem? Well, if you’re reading a Parquet file, it has a layout that’s optimized for this kind of access pattern. So what a Parquet reader will do is first go to the very end of the file and find out where its footer is located. Then it’ll read the footer and look at the row group metadata, since each Parquet file is sliced into row groups. It’ll use the filters to find out which row groups to scan through. And because we only need specific columns, we don’t need to read the entire row group either. So we can surgically read just the precise bits of the file that we really need for the query.

But this is the kind of thing can happen multiple times. You might have multiple DataFusion workers reading through this file. You might have subqueries that hit the same table, so some of these things might be overlapping. Or you could have concurrent queries, or someone running the same query twice.

And this is a problem with actual networked object stores: there’s a latency to making every request, I think S3 in the same region is something between 15-30ms. You also pay actual money to make requests and if you’re in a different region, you pay bandwidth costs to egress data.

7 of 17

8 of 17

Caching logic

Alright. How does it work? I transposed the diagram a bit and the data flow is left to right. So, DataFusion wants us to read a tiny slice of a file. The first thing we do is completely ignore what it wants. We consider each file as being subdivided into a chunks of a certain size, in this case 1MB. Conceptually, if it costs us 15ms to read 5kB and 17ms to read 1MB, we might as well read the 1MB. So internally we think that we need to read the 0..1MB chunk.

Then we go to our actual cache core. We use a library called moka for this. It handles concurrent access and cache maintenance logic, like eviction, because we want to maintain a certain disk cache utilization and not go beyond that. We ask moka if we have this 1MB chunk cached. If not, we go to the actual object store and call get_range with this expanded chunk. If DataFusion requested a range that spans multiple chunks, we still go through this splitting, but we coalesce the request back into one get_range. So if DF asks for 0..4500kB, we’ll request 0..5MB and then split it up into chunks instead of making 5 requests.

Now, the request has returned. We used to write it to disk at this point before returning it, but that would slow us down a lot because we’re sitting there waiting on IO while DataFusion could have been doing CPU work with this data. So instead the cache key can point to either an on-disk path or an actual in-memory sequence of bytes. We insert the latter into the cache and enqueue a Tokio task to move the in-memory chunk to disk in the background. And immediately return to DataFusion.

If there’s another request that among other things hits this chunk, we can now read it from the cache, it’ll be either in memory or on-disk and we can read it from disk like that.

One last thing is eviction. We set a least-recently-used policy on Moka and we have an eviction listener that we register with it. So here we have a 256MB cache and say we’re reading through 500MB of data, then the oldest chunks will get evicted and when Moka tells us that, we delete the file from disk.

9 of 17

Benchmarking

22 TPC-H queries, SF10 (3GiB in Delta)

cache off
cache on (256MiB)
Inject 30ms latency with toxiproxy

10 of 17

11 of 17

12 of 17

Demo

TPC-H Q6

SELECT

sum(l_extendedprice * l_discount) AS revenue

FROM

lineitem

WHERE

l_shipdate >= CAST('1994-01-01' AS date)

AND l_shipdate < CAST('1995-01-01' AS date)

AND l_discount BETWEEN 0.05

AND 0.07

AND l_quantity < 24;

13 of 17

14 of 17

15 of 17

Q&A

https://github.com/splitgraph/seafowl/blob/main/src/object_store/cache.rs

16 of 17

Backup slides

17 of 17

Future work

This version of this slide features an image at right and copy to the left. When using more than two such slides in a row, be sure to alternate these layouts.

Concurrency issues!

Reading a file chunk that just got evicted and deleted
Deleting a chunk that already got deleted
Two threads downloading the same chunk

Possibility of deadlocks!

Writes?

So what’s broken? Well, one thing we couldn’t figure out, and Marko spent a few weeks on this, is concurrency. When you’re getting hit by a lot of DataFusion workers, concurrent queries, subqueries, a lot of weird things can happen. We have some mitigations for that and we know it doesn’t impact correctness (we have a TPC-H test suite) but there’s performance impact.

You read from the cache, it’s a cache hit, it’s on disk, but because of the pressure on the cache that thing gets evicted and deleted from disk. In this case we say “ok, I guess it’s not there, we’ll just initiate a refetch”.

Double evictions. We get told that some chunk has been evicted and we try to delete it, but it’s already been deleted. Who did it? No idea, we can ignore it.

Concurrent downloads. This is a fun one, you can have two threads requesting the same chunk, so do you download it once and tell the other thread to wait? What if it fails? We had a setup with tokio channels but then we found a deadlock on TPC-H Q18, so we stopped doing that. Instead, we just let both threads do the download.

And we don’t treat this cache as a write-through. Marko will speak more about writes in his talk, but we don’t do the thing where we write to the cache as well as to the object store to speed up reading something we just wrote.