Inside QuestDB's Query Engine: Tracing Three Queries

I’ve been exploring the QuestDB codebase, a time series columnar database, and, specifically, trying to learn how its query engine works. And whenever we discuss a query engine execution, certain kinds of dimensions and options within those dimensions arise in the conversation. For example:

This taxonomy is not necessarily something that is agreed on in the literature or in the database industry. It’s just how I usually think about it based on my research, and also for the purposes of this blog post we are mostly focused on the execution part of a query engine, so we’ll ignore the parsing, optimization, and most of the planning.

You might think that those dimensions are orthogonal to each other and a query engine is essentially composed of one option of each dimension. But in reality, things are much more nuanced, and a query engine can express a unique mix of those options.

In this blog post, I explore how QuestDB expresses these dimensions and try to build a strong mental model of its query engine execution. This blog post assumes some understanding of these dimensions. Also, I try not to get too distracted by details of the codebase and implementation and focus more on what techniques are applied and in which context they are applied.

I ended up with 3 queries that I think are good enough to get a gist of what is going on:

SELECT symbol, SUM(amount) FROM trades GROUP BY symbol;
SELECT symbol, side, SUM(amount) FROM trades GROUP BY symbol, side;
SELECT symbol, side, SUM(amount) FROM trades WHERE side != 'SELL' GROUP BY symbol, side;

Dataflow

When I started, one of the first things that I noticed was the presence of classes extending AbstractRecordCursorFactory everywhere. For example, AsyncFilteredRecordCursorFactory, GroupByRecordCursorFactory , and AsyncGroupByRecordCursorFactory.

SqlCompiler returns a cursor in which one record at a time is returned. From that we infer that QuestDB is implementing a pull-based tuple-at-a-time dataflow à la Volcano. You might be thinking, how in the world could there be vectorization, or JIT, or parallelism underneath that? That’s what we explore next.

First query

Tracing our first query, SELECT symbol, SUM(amount) FROM trades GROUP BY symbol, we get the following trace log:

[shared-network_7] > vect.GroupByRecordCursorFactory.getCursor  [vectorized parallel aggregate, aggregates=1 shards=14]
[shared-network_7] < vect.GroupByRecordCursorFactory.getCursor
[shared-network_7] > vect.GroupByRecordCursorFactory.getCursor  [vectorized parallel aggregate, aggregates=1 shards=14]
[shared-network_7] < vect.GroupByRecordCursorFactory.getCursor
[shared-network_7] . vect.GroupByRecordCursorFactory.buildMaps  [frames=5 aggregates=1 => total tasks dispatched=5]
[shared-network_7] . vect.GroupByRecordCursorFactory.runWhatsLeft  [dispatcher (workerId=7) drained queue cursor=12 after dispatch finished]
[shared-network_7] . SumDoubleVectorAggregateFunction.aggregate  [native keyedIntSumDouble kernel processing 100000 rows (one SIMD-friendly C++ loop over key + value columns)]
[shared-query_3] . GroupByVectorAggregateJob.doRun  [worker=3 cursor=13 (executes Vect.sumDouble etc on a page-frame column)]
[shared-query_3] . SumDoubleVectorAggregateFunction.aggregate  [native keyedIntSumDouble kernel processing 100000 rows (one SIMD-friendly C++ loop over key + value columns)]
[shared-query_8] . GroupByVectorAggregateJob.doRun  [worker=8 cursor=14 (executes Vect.sumDouble etc on a page-frame column)]
[shared-query_12] . GroupByVectorAggregateJob.doRun  [worker=12 cursor=15 (executes Vect.sumDouble etc on a page-frame column)]
[shared-query_12] . SumDoubleVectorAggregateFunction.aggregate  [native keyedIntSumDouble kernel processing 100000 rows (one SIMD-friendly C++ loop over key + value columns)]
[shared-query_8] . SumDoubleVectorAggregateFunction.aggregate  [native keyedIntSumDouble kernel processing 100000 rows (one SIMD-friendly C++ loop over key + value columns)]
[shared-query_1] . GroupByVectorAggregateJob.doRun  [worker=1 cursor=16 (executes Vect.sumDouble etc on a page-frame column)]
[shared-query_1] . SumDoubleVectorAggregateFunction.aggregate  [native keyedIntSumDouble kernel processing 100000 rows (one SIMD-friendly C++ loop over key + value columns)]
[shared-network_7] . vect.GroupByRecordCursorFactory.merge  [vaf=0 fold pRosti[3] (size=10) into pRostiBig (size=10) via native kernel]
[shared-network_7] . vect.GroupByRecordCursorFactory.merge  [vaf=0 fold pRosti[7] (size=10) into pRostiBig (size=10) via native kernel]
[shared-network_7] . vect.GroupByRecordCursorFactory.merge  [vaf=0 fold pRosti[8] (size=10) into pRostiBig (size=10) via native kernel]
[shared-network_7] . vect.GroupByRecordCursorFactory.merge  [vaf=0 fold pRosti[12] (size=10) into pRostiBig (size=10) via native kernel]

The first thing to notice is the thread name. That already hints to us the parallelism going on under the hood. shared-network_7 is one worker, from the shared-network pool of workers, that is responsible for starting the execution of the query and returning the results. The shared-query workers are part of another worker pool executing tasks in parallel.

The operator that represents our query is RostiRecordCursor created by vect.GroupByRecordCursorFactory and when hasNext (we don’t trace the hasNext call because that would pollute our log), it tries to build the internal hash map by calling buildMaps. The operator has access to the storage through PageFrameCursor that returns chunks of the data called frames.

From my research on query parallelism, these techniques resemble a lot of morsel-driven parallelism; frames (morsels) being the unit of dispatch, and workers working in parallel on those units.

Second query

Before tracing our next query, SELECT symbol, side, SUM(amount) FROM trades GROUP BY symbol, side;, let’s compare its query plan to the query plan of the former query.

We can see that AsyncGroupByRecordCursor is the cursor that is built, and after the buildMap call a dispatchAndAwait call on UnorderedPageFrameSequence is called. From the thread names we can see the execution happening in parallel: shared-query_12 got frame 1, shared-query_3 got frame 2, shared-query_5 got frame 3, and shared-query_8 got frame 4.

Within aggregate computeKeyedBatch from SumDoubleGroupByFunction is called on a batch of 2048 rows. And you can see it’s a for-loop over the rows updating the value stored in the address of the hash map.

Third query

Now let’s take a look at our last query: SELECT symbol, side, SUM(amount) FROM trades WHERE side != 'SELL' GROUP BY symbol, side;

The trace log is similar to the last query, except for filterAndAggregate. What I found interesting here is that, since we added a new operator to our query, I was expecting some kind of cursor composition where the group-by cursor receives the filter cursor as a child. I found out that what is happening here is something called filter stealing. The group-by cursor “steals” the filter, and the filter operation becomes embedded in the group-by cursor without the need for another cursor.

Now to the JIT. From what I could verify, only WHERE clauses are JIT’d. If JIT is enabled and the expression is JIT-able, the expression is compiled to native code, and a function pointer is stored in CompiledFilter.fnAddress. And we saw how the compiled filter was passed to AsyncGroupByRecordCursorFactory, AsyncGroupByRecordCursorFactory.filterAndAggregate calls that function to filter the rows of a frame before aggregating them.

Wrapping up

In this blog post, we investigated 3 queries that gave us a good sense of what is going on in QuestDB’s engine in terms of the dimensions we talked about. We clearly see it’s an engine that applies different techniques according to the context of the query. Although it is a pull-based tuple-at-a-time engine, internally, leaf nodes can parallelize the query execution using a morsel-driven-like design. The data is broken into frames, and jobs are queued for workers to process. In some cases it outsources operations to native code, as in the case of SUM, using SIMD to calculate the aggregate, and WHERE, using JIT’d expressions to filter rows.