Duckdb: Intra-Query Parallelism

[pdet]

[hannesmuehleisen]

Some thoughts about the design of the parallel query execution:

We split up queries into pipelines, which end for example at a hash table build or a result set. Those pipelines are then split into tasks on subsets of the input data (e.g. by adding an offset and count to the scan operator).

Not all tasks can be executed immediately, some will have to wait for a hash table to become available for example. Others can run, for example those who scan base tables. Executable tasks are scheduled for execution into a task queue.

A number of worker threads monitor the queue, take tasks and start executing them in an infinite loop. They will prefer tasks that belong to the same pipeline instance to help with data locality.

When all tasks on a pipeline finish, the scheduler finds the next pipelines that can now run given the availability of the input and schedules them. When the pipeline that ends in the overall result finishes, the query is done. Once a full chunk is available in the result set, it can be made available for the client to read.

Progress needs to be ensured for both long-running and short-running queries. This could be solved by maintaining two task queues, one for each class of queries. Queries start in the short-running queue. Once a query has been scheduled for execution for more than n times, it would move to the long-running queue. Worker threads could be split in preference for the queues. This is preferable to round robin, again to increase data locality.

An open issue is how to address the issue that a large query (e.g. SELECT * FROM lineitem) is executed, but the client only fetches a single result chunk from the result.
In this case, we have the threads executing a huge parallel scan with the CPU and RAM costs associated but a client that does no longer want these. In very simple cases, the client explicitly closes the result set or the connection, in which case execution can be terminated immediately by clearing the queue for a query and interrupting all threads. A solution might be to pause the pipeline once a result set buffer reaches a set watermark of unfetched result rows by pausing task execution for them. Once the client resumes fetching, a second watermark could trigger the scheduling of tasks again.

I found https://github.com/cameron314/concurrentqueue, which looks pretty good.

[Mytherin]

That concurrent queue is excellent, I have used it before. I think the general design is good.

Some considerations:

• Perhaps we should not queue all tasks immediately up front, but rather use a lazy queueing strategy, i.e. we only queue additional tasks as previous tasks are completed. Suppose we have a 100GB table and we are queueing parallel scans of 1MB/each, that would mean queueing 100,000 tasks up front just for that table scan. Instead we could queue only N tasks (where N is the amount of threads), and as those finish queue additional tasks.
• The chunk size of how we split up the scans is also something we should consider. Probably around 1MB would be a good size to split up the scans. Although maybe we can have a more dynamic solution as well. Either way we need to consider that we don't overload the task queue too much.
• Perhaps using a tasks for fixed parts of the table at all is not a good idea. For example, if we choose 1MB chunks, then if we have a very costly UDF running on a small table we would achieve zero parallelism. Instead we could also fetch vectors using some lock-free primitives, e.g. perform a CAS to fetch a chunk from the base table. This would work well on non-NUMA nodes because the scans of the tables would always be sequential, no matter how many threads there are.

[weijietong]

hi @Mytherin what's the progress of this issue ?

[Mytherin]

Currently in active development, and my priority at this moment. I suspect a basic version of this will be ready in 1~2 months, with more improvements coming as we go.

[Mytherin]

This is now implemented.