Loading Lakes

If you read my previous blog post you'll know that we're using Delta Lake as the underlying protocol for our database. You may also recall that we're using it in some unconventional ways. One of these ways is a tenants data is split between two separate Delta Lake tables where data is moved in a one way direction from the primary table where everything is initially written, to the final tenant table that only contains data for that tenant.

This means that to be able to read from a tenant database you need to load two separate Delta Lake tables to get the full picture. Our entire database stack is written in Rust so naturally we were using a fork of the official delta-rs library to interact with our tables. If you'll recall this fork was due to the fact that we're using Vortex files instead of JSON or Parquet files in the table.

So the process to reading a table would go as follow:

1. Read the metadata tables _last_checkpoint file.
2. Perform a list operation from the last checkpoint offset.
3. Read all files returned from the list operation.
4. Rebuild the state from the operations read.
5. Perform steps 1-4 for the tenant table.
6. Build a DataFusion TableProvider from the list of files obtained from steps 1-5.
7. Run the DataFusion query using the TableProvider from step 6.

Steps 1-6 are all performed outside of DataFusion and most are performed by the delta-rs fork.

Delta Fusion

We've been huge fans of DataFusion since we first started using it. It's incredibly performant, easily extensible and keeps getting better with development. This naturally led us to ask the question, what if we used DataFusion to load the DeltaLake tables?

Instead of using the built in system in delta-rs to perform the reading of _last_checkpoint, listing the files and then reading all the data files (steps 1-4). What if we just had DataFusion perform all of these actions. We can leverage it's ability to parallelize reading and parsing across threads.

So we first built a DataFusion TableManager for the DeltaLake protocol. This table manager performs the reading of last checkpoint, then listing the files, and reads and processes the listed files in parallel. A scan of the DeltaLake TableManager returns a stream of Arrow records where each row contains an action from the DeltaLake protocol. Using this stream a reader can reconstruct the table state from the actions. This exposes the actions in a DeltaLake log as a simple table that is now queryable with standard SQL.

This is great if you're just reading a single DeltaLake table, however our tenants data is partitioned across two DeltaLake tables. We could simply perform two DataFusion queries to both tables, and perform the reconciliation of actions in the table outside of DataFusion, but we want to reap as many benefits of DataFusion as we can, so we can do better than that.

This is where we implemented the GreatLakes TableProvider. The GreatLake TableProvider is able to perform all of the reconciliation of DeltaLake actions from the DeltaLake table providers, it's aware of the layout we have on object storage so it's able to discover and query any tenant table and stitch the data together with the metadata table.

Now to get the full state of any given table we need only to write a simple SQL query

SELECT kind, file_path, file_size, segment_stats, partition_values, stats_json,
arrow_schema_json,
partition_values
FROM great_lakes
WHERE "schema" = '{schema_version}' AND "table" = '{table_name}';

Need for Speed

Our hypothesis was that because DataFusion is built for speed, it would be faster to query Delta Lakes with DataFusion than using the Rust library's built in loading. We turn to benchmarks to test this hypothesis.

At a quick glance this looks amazing, DataFusion is significantly faster at loading tables than the delta-rs library is. However there is a small but important wrinkle in this data. If you look closer you'll notice that at 10 writes DataFusion is actually slower than delta-rs. This is actually problematic because we checkpoint the Delta Log every 10 commits meaning that our performance for 10 commits is our normal use case.

The profile below is for the 10 commit benchmark. There's a lot going on in that profile, but one thing we can gleam from it is there is a static overhead cost of running DataFusion. Notably with the physical planning and optimization steps, we pay for that on every query, and when the amount of data we're querying is small enough, like in the 10 commits case, it ends up eating into any savings we make in the query path.

Conclusion

DataFusion is highly performant, it excels at querying large sets of data incredibly efficiently. We can utilize it to unlock the ability to perform arbitrary queries on our Detla Lake tables, to be able to ask questions about the table we were otherwise unable to do. But it's not always the best solution, we see that it has overhead to run a general purpose query engine like this, and for very small queries bespoke code still can have it's place.

We look forward to seeing if we can cut down on the physical planning and optimization stages using profiling data to make it competitive with delta-rs table loading.