ClickHouse Query Execution Pipeline

Chain of Independent Computations

SELECT avg(length(URL)) FROM hits WHERE URL != ''

• Reading URL column
• Computing URL != ''
• Filtering
• Computing length(URL)
• Aggregation

Each operation can be executed independently

Pipeline Execution Tasks

What benefits does using a computational pipeline provide?

• Abstraction of computations from operation execution order
• Parallelization of computations
• Balancing computational resources (CPU, GPU, FPGA)

Volcano — An Extensible and Parallel Query Evaluation System

Pipeline represented as a graph of operators (Exchange operator)

Each operator (of relational algebra) supports the interface:

• open() - prepare for execution
• next() - return next row (or EOF)
• close() - finish execution

Row-by-row query execution (tuple-at-time)

Volcano

Advantages

• Simple interface
• Old proven method
• Supports multitasking

Disadvantages

• Many instructions per row
• Low IPC (Instructions Per Cycle) value
• No support for vector operations

Column Computations

MonetDB/X100: Hyper-Pipelining Query Execution

Idea

• All computations are performed on column chunks (size 1024)
• Developed MIL language for writing operations on columns
• Fast vectorized X100 engine

Column Computations

Result:

• Reduced number of operations per row
• Increased IPC through vector operations
• Queries became memory-bound instead of CPU-bound

Disadvantages

• Increased memory consumption

Query Pipeline Execution

Pipeline can be executed differently

• push: "forward" execution order. Push data into the beginning.
• pull: "backward" execution order. Pull data from the end.
• arbitrary order within topological sort constraints

Query Pipeline Execution

Let's consider the task in a generalized form

• Operation knows how to execute itself
• Only need to be able to get the next pipeline operation
• Represent pipeline as a data structure that needs to be traversed

Data structure iteration

• Selecting next element
• Operation on data

Types of Data Iteration

External

• Selecting next element is described externally
• Example: for loop, iterators
• Control over iteration order and termination moment

for (element : array)  /// Outer iteration.
{
    foo(element);

    if (!element)  /// Early exit.
        break;
}

Types of Data Iteration

Internal

• Selecting next element is determined by data structure
• Cannot choose order or terminate iteration early
• Has call stack, code is simplified

void Tree::inOrder(auto & function)
{
    if (left_child)
        left_child->inOrder(function);

    function(this);

    if (right_child)
        right_child->inOrder(function);
}

Query Pipeline Execution

push and pull approaches are implemented with internal iteration

• difficult to achieve vertical parallelization
• special handling of premature query termination
• execution flow tied to data flow (since stack is used)

Graph representation allows external iteration

Example: LocustDB

• All data stored in memory
• Query execution in topological sort order
• Each operation processes a chunk of data

Pipeline Execution in ClickHouse

Mixed scheme

• IBlockInputStream - data read stream (pull)
• IBlockOutputStream - data write stream (push)
• BlockIO - stream connection point

Query execution - transferring from InputStream to OutputStream

Some block operations are the same for InputStream and OutputStream

Parallel Query Execution

Approaches:

• by operations (intra-operator): execute slow operations
  (IO, sort, group by) in multiple threads
• by data (bushy): copy pipeline per thread
• by pipeline parts (vertical): different parts in different threads

Vertical Pipeline Parallelization

Volcano: exchange operator

• open() - create new process
• next() - RPC call

ClickHouse: AsynchronousBlockInputStream

• Creates new thread
• Increases query latency
• Used in "special cases"

Parallel Execution in ClickHouse

Data parallelism

• Data sources return multiple streams for reading
• Pipeline is built simultaneously for all streams
• Streams are merged into one in some operations (order by, group by)

Example

SELECT length(URL) AS l, count() FROM hits GROUP BY l ORDER BY l

Parallel Execution in ClickHouse

Let's look at query execution result

URL length distribution has a heavy tail

Parallel Execution in ClickHouse

Data is processed at different speeds

Parallel Execution in ClickHouse

Solution - "stealing" tasks from neighboring thread during reading

Parallel Execution in ClickHouse

Task stealing doesn't solve all problems

• Still can't parallelize vertically
• Arbitrary data reading order

Example: reading in primary key order (optimization)

SELECT value FROM table LIMIT 10                 -- Any 10 rows
SELECT value FROM table ORDER BY value LIMIT 10  -- All data

What if data is already partially sorted (within a part)?

• Can optimize if reading correctly
• Cannot steal tasks

Processors in ClickHouse

Representing query processing pipeline as a graph

Components:

• Processor (processsor): arbitrary operation on a block
• Port (port): input or output for processor (stores data block)
• Ports (input and output) can be connected with connect() function

Processors in ClickHouse

Types of processors

• source, sink - files, tables, table functions, …
• simple operation (function call, filtering, adding column)
• resize - arbitrary change in number of data streams

Pipeline Execution

Processors are in different states

• NeedData, PortFull, Unneeded - need to transfer data
• Ready, Finished - can continue/finish work
• Async, Wait - asynchronous execution

Pipeline execution - also a processor

• Single-threaded execution - in topological traversal order
• Multi-threaded - graph traversal in multiple threads using heuristics

Pipeline Execution

Processors

• White - needs input data
• Yellow - ready for execution
• Red - output port is full

Ports

• White - free, yellow - occupied

Pipeline Execution

Processors

• White - needs input data
• Yellow - ready for execution
• Red - output port is full

Ports

• White - free, yellow - occupied

Pipeline Execution

Processors

• White - needs input data
• Yellow - ready for execution
• Red - output port is full

Ports

• White - free, yellow - occupied

Pipeline Execution

Processors

• White - needs input data
• Yellow - ready for execution
• Red - output port is full

Ports

• White - free, yellow - occupied

Pipeline Execution

Processors

• White - needs input data
• Yellow - ready for execution
• Red - output port is full

Ports

• White - free, yellow - occupied

Pipeline Execution

Processors

• White - needs input data
• Yellow - ready for execution
• Red - output port is full

Ports

• White - free, yellow - occupied

Pipeline Execution

Processors

• White - needs input data
• Yellow - ready for execution
• Red - output port is full

Ports

• White - free, yellow - occupied

Pipeline Execution

Processors

• White - needs input data
• Yellow - ready for execution
• Red - output port is full

Ports

• White - free, yellow - occupied

Pipeline Execution

Processors

• White - needs input data
• Yellow - ready for execution
• Red - output port is full

Ports

• White - free, yellow - occupied

Pipeline Execution

Processors

• White - needs input data
• Yellow - ready for execution
• Red - output port is full

Ports

• White - free, yellow - occupied

Pipeline Execution

Processors

• White - needs input data
• Yellow - ready for execution
• Red - output port is full

Ports

• White - free, yellow - occupied

Pipeline Execution

Processors

• White - needs input data
• Yellow - ready for execution
• Red - output port is full

Ports

• White - free, yellow - occupied

Pipeline Execution

Processors

• White - needs input data
• Yellow - ready for execution
• Red - output port is full

Ports

• White - free, yellow - occupied

Pipeline Execution

Processors

• White - needs input data
• Yellow - ready for execution
• Red - output port is full

Ports

• White - free, yellow - occupied

Pipeline Execution

Processors

• White - needs input data
• Yellow - ready for execution
• Red - output port is full

Ports

• White - free, yellow - occupied

Pipeline Execution

Processors

• White - needs input data
• Yellow - ready for execution
• Red - output port is full

Ports

• White - free, yellow - occupied

Pipeline Execution

Processors

• White - needs input data
• Yellow - ready for execution
• Red - output port is full

Ports

• White - free, yellow - occupied

Pipeline Execution

Processors

• White - needs input data
• Yellow - ready for execution
• Red - output port is full

Ports

• White - free, yellow - occupied

Shared Pipeline Execution

Example from QueryLog

SELECT sum(value) FROM table WHERE foo(x) = 1;
SELECT sum(value) FROM table WHERE foo(x) = 5;
SELECT sum(value) FROM table WHERE foo(x) = 7;

Can compose shared pipeline for multiple queries

Pipeline Optimizations

Query optimizations can be performed at pipeline construction stage

• Based on data flows and their properties (reliable, corresponds to essence)
• Alternative to AST optimizations (laborious, error-prone)

Examples:

• Information about column monotonicity
• Statistics: number of rows, number of distinct values, distribution
• Merging pipeline parts with the same result

SELECT value IN (SELECT x FROM table) WHERE (value + 1) IN (SELECT x FROM table)

Resource Allocation

Unified pipeline allows transparent CPU sharing between users

• Processor is created for individual user (possibly multiple)
• Block processing by processor — user's resource consumption
• Can limit resources only under high load
• Choice of operation execution priority

Pipeline Description Language

Idea: create language for describing query execution pipeline

• Describes nodes and their connections
• Allows pipeline (de)serialization

Additional capabilities

• Transfer ready pipelines to shards: savings on query analysis
• Low-level interface
• Pipelines not describable through SQL syntax

Introspection

External

• Query plan as tree (already implemented)
• Step-by-step optimization results

Internal

• Debug processors with statistics and logs
• Information about creation process
• More understandable call stack

What We Expect from New Pipeline

• Vertical parallelization
• Introspection (EXPLAIN, global plan) of query
• Cheap query cancellation
• Serialization (description language) of pipeline
• Data statistics (ORDER BY optimization on PK)
• Resource sharing and user priorities
• Processing multiple queries simultaneously
• Reducing number of threads
• Simplifying shuffle distributed join implementation