Query Engines: Push vs. Pull (2021)

At Polars I developed a new query engine which uses a hybrid of push and pull. I gave a short (and not very technical) talk about the engine at our first meetup recently, which can be viewed here: https://www.youtube.com/watch?v=Ndil-eLynh4.

Each operator is a (set of) async functions which are connected to its input(s) and output(s) through capacity-1 spsc async channels. An operator pulls input by reading from a channel, and pushes output by writing into a channel. For an oversimplified example, consider a simple select operator:

    while let Ok(morsel) = input.recv().await {
        let result = select(morsel);
        if output.send(result).await.is_err() {
            break;
        }
    }

Note how it has two await points: on receive and send. The nice part about this is that Rust will automatically transform these asynchronous functions to state machines which can pause execution when either a send or receive blocks, returning control to our scheduler. In the above example the operator is able to pause both to wait for more data, or to wait until the receiver is able to consume more data. This also makes for example pausing execution in the middle of a hash join probe to send some partial results onwards in the computational graph trivial.

> In a setting where every operator has exactly one output, when to run an operator to produce some output is obvious: when your consumer needs it. This becomes, at the very least, messier with multiple outputs, since “requests for rows” and “computations to produce rows” are no longer one-to-one.

Don't push-based systems have the same issue for inputs? If you have multiple inputs then you may be handled one of the inputs without the other. The classic example with iterators is a `zip` operator, though maybe this is probably not so common in database queries.

Interesting reading!

Does this explain a big inefficiency in BigQuery where it only ever does hash-joins? Is it because it is push so it never does merge-joins even when the inputs are all sorted or clustered etc?

Although tbh can't see why a system can't combine them both; some of the edges being push+buffer and some being buffer+pull.