Thread by @ghalimi

Now that we have a pretty good idea of what the UI will look like, let's focus on the runtime side of things, especially when data gets too large to fit within a single host (2 TB on a host with 4 TB of RAM). In other words, how will the distributed query planner work?

https://twitter.com/ghalimi/status/1615065408599654401

The assumption is that source data is partitioned as a set of Parquet files (with or without Iceberg). From there, if your query is map-reducible, you can use @DuckDB's COMBINE to map it on a fleet of Lambdas, then reduce it on an EC2 VM. That's what #STOIC does today.

Things start getting more complicated if you have a SELECT within a FROM clause, a HAVING clause, or inner queries (correlated or uncorrelated). Let's review these scenarios one by one.

SELECT in FROM Clause

In that case, you want to distribute the inner SELECT across Lambda functions, and push down into the Lambdas as many parts of the outer query as possible. After that, you reduce everything on the monostore (single EC2 VM). No big deal.

HAVING in GROUP BY

In that case, you want to execute the inner query within the Lambdas executing the map parts of the GROUP BY (if you can), before sending filtered binary results to the monostore for reduction (with COMBINE).

Uncorrelated Query

This one is simple: run it once on a Lambda if it's dealing with small data, or on the monostore if it's not, then make its result available to all Lambdas running the map part of the outer query.

Correlated Query

This one is where things get really interesting. But before we dive deep into it, let's try to set some boundaries: in some extreme cases, it simply won't work in a distributed fashion, no matter what you try.

Here is an example of such an extreme case: if the denormalized table for the inner query is large (or very large) and cannot be partitioned across the same dimension(s) as the denormalized table for the outer query and the later produces a large result, you're out of luck.

In that case, no matter what you do, each inner query will be expensive, and you'll have to execute as many of them as you have rows in the result of the outer query.

What that means is that if a large query takes 1s on your fleet of Lambdas and your outer query has 1B rows, the full query will take at least 1 billion seconds, or more than 11 days...

With that out of the way, let's review some scenarios that will work a lot better. For example, if the result of your outer query is small, things are easy. And if the dataset of the inner query is small, you can replicate it on the Lambdas running the outer query's map part.

If the outer query generates a large result and the inner query is based on an equally large dataset, your only option is to align their respective partitions. Unfortunately, this is not always possible, and even if it is, it might not be easy.

But there is one more trick that you can pull: you can try to include in the results of the outer query the dimension(s) used for partitioning the inner query's dataset. If you can do that, you can then partition the outer query's result by this/these dimension(s).

From there, you pass each outer result partition to the Lambda handling the corresponding inner dataset partition, hoping that the former are not too skewed. And if they are, you just assign multiple Lambdas to the same inner dataset partition.

That, in a nutshell, will be the strategies implemented by the first version of our distributed query planner. It will handle all 22 TPC-H queries fairly well, and many more. What is not covered in this design is things like CUBE, GROUPING SETS, and WINDOW, so stay tuned...