Opensource internals

ClickHouse Builds an Actions DAG Once and Runs It on Every Block

16 min read Jay Hubbert

ClickHouse compiles a SELECT into an ActionsDAG, topologically sorts it into a flat vector of actions with fixed slot numbers, and reuses that vector for every 65409-row block.

TL;DR: A SELECT in ClickHouse is not a loop over rows, it is a graph of actions. That graph — the ActionsDAG — is built once, flattened by a topological sort into a std::vector<Action> where every argument and every result has a fixed integer slot, and then executed once per 65 409-row block by simple array indexing. No per-row dispatch, no tree walking, no intermediate hash maps.

Every columnar database ends up with the same internal representation for a SELECT: a tree of expressions. The obvious way to run that tree is to walk it per row. SELECT a + b * c FROM t becomes, per row, a recursive descent through an expression tree, with a virtual call at every node.

ClickHouse never does that. It has three layers of insulation between the parsed query and the raw bytes:

  1. the query is compiled into a graph of actions (ActionsDAG),
  2. the graph is topologically sorted once into a flat list of actions with pre-assigned slot numbers,
  3. the list is replayed per block, and every intermediate result is a ColumnPtr in a fixed-size array indexed by slot number.

The entire design goal of ActionsDAG is stated in its own header comment:

cpp 
// src/Interpreters/ActionsDAG.h:60
/// Directed acyclic graph of expressions.
/// This is an intermediate representation of actions which is usually built from expression list AST.
/// Node of DAG describe calculation of a single column with known type, name, and constant value (if applicable).
///
/// DAG representation is useful in case we need to know explicit dependencies between actions.
/// It is helpful when it is needed to optimize actions, remove unused expressions, compile subexpressions,
/// split or merge parts of graph, calculate expressions on partial input.
///
/// Built DAG is used by ExpressionActions, which calculates expressions on block.

Read that last line again: built DAG is used by ExpressionActions, which calculates expressions on block. The DAG is the plan; ExpressionActions is the interpreter that runs it; the interpreter is run once per block.

The node types

A DAG node is “the calculation of a single column with known type, name, and constant value (if applicable)”. There are exactly six kinds:

cpp 
// src/Interpreters/ActionsDAG.h:73
    enum class ActionType : uint8_t
    {
        /// Column which must be in input.
        INPUT,
        /// Constant column with known value.
        COLUMN,
        /// Another one name for column.
        ALIAS,
        /// Function arrayJoin. Specially separated because it changes the number of rows.
        ARRAY_JOIN,
        FUNCTION,
        /// Placeholder node for correlated column
        PLACEHOLDER,
    };

Two of these are worth pausing on.

ARRAY_JOIN is separated out because it changes the number of rows. Every other action is a pure column-to-column map; arrayJoin is the one operation that can grow the block. That single bit of metadata is what lets the planner split a DAG at an arrayJoin boundary, and what lets removeUnusedActions keep an arrayJoin alive even when nothing downstream reads its result:

cpp 
// src/Interpreters/ActionsDAG.cpp:1030
        /// We cannot remove arrayJoin because it changes the number of rows.
        if (node.type == ActionType::ARRAY_JOIN)
            roots.push_back(&node);

PLACEHOLDER is the mark of a correlated column — a reference from a subquery to a column of an outer query. It is a node whose value is not known at plan time. It gets rewritten into an INPUT by decorrelate() once the correlated subquery is turned into a join:

cpp 
// src/Interpreters/ActionsDAG.h:412
    /// Replace all PLACEHOLDER nodes with INPUT nodes
    /// Not noexcept: appends to the inputs vector, which can throw on allocation.
    void decorrelate();

A node carries its children as raw pointers, its result name and type, and — for functions — both the overload (the IFunctionBase, which knows the signature and the monotonicity) and the prepared instance (the IExecutableFunction, which is what actually gets called at run time):

cpp 
// src/Interpreters/ActionsDAG.h:92
    struct Node
    {
        NodeRawConstPtrs children;

        ActionType type{};

        std::string result_name;
        DataTypePtr result_type;

        /// Can be used to get function signature or properties like monotonicity.
        FunctionBasePtr function_base;
        /// Prepared function which is used in function execution.
        ExecutableFunctionPtr function;
        /// If function is a compiled statement.
        bool is_function_compiled = false;

The split between function_base and function matters more than it looks. function_base is the overload — it is shared, immutable, and knows things like “am I monotonic?” and “can I be constant-folded?”. function is the prepared instance, built from the concrete argument types and columns, and it is what executeAction calls.

Why std::list

The nodes are stored in a std::list, and the header is explicit that this is an implementation detail — but the reason is worth stating, because it explains the whole pointer-based style of the class:

cpp 
// src/Interpreters/ActionsDAG.h:140
    /// NOTE: std::list is an implementation detail.
    /// It allows to add and remove new nodes inplace without reallocation.
    /// Raw pointers to nodes remain valid.
    using Nodes = std::list<Node>;

Every optimisation in the system — pruning unused columns, splitting the DAG in half to push a filter into the storage, rewriting a subquery’s projection — hands out const Node * pointers into this list and expects them to stay valid. A std::vector<Node> would invalidate every pointer on every push_back. std::list gives O(1) insertion, O(1) erasure, and stable addresses.

Adding a node is therefore trivial:

cpp 
// src/Interpreters/ActionsDAG.cpp:546
ActionsDAG::Node & ActionsDAG::addNode(Node node)
{
    auto & res = nodes.emplace_back(std::move(node));
    if (res.type == ActionType::INPUT)
        inputs.emplace_back(&res);

    return res;
}

The DAG keeps three vectors of node pointers: nodes (all of them, in insertion order), inputs (the ones that must be present in the input block), and outputs (the ones that are the result).

Constant folding happens at plan time, not at run time

Before a function node is added, ClickHouse tries to evaluate it right now, at plan time, if all its arguments are constants:

cpp 
// src/Interpreters/ActionsDAG.cpp:122
void tryFoldFunctionToConstant(
    ActionsDAG::Node & node,
    const ColumnsWithTypeAndName & arguments,
    bool all_const,
    bool best_effort)
{
    /// If all arguments are constants, and function is suitable to be executed in 'prepare' stage - execute function.
    if (!node.function_base->isSuitableForConstantFolding())
        return;

    ColumnPtr column;
    try
    {
        if (all_const)
        {
            size_t num_rows = arguments.empty() ? 0 : arguments.front().column->size();
            column = node.function->execute(arguments, node.result_type, num_rows, true);
        }
        else
        {
            column = node.function_base->getConstantResultForNonConstArguments(arguments, node.result_type);
        }
    }
    catch (const Exception &)
    {
        if (!best_effort)
            throw;
        return;
    }

and the folded value is stored in the node:

cpp 
// src/Interpreters/ActionsDAG.cpp:175
    /// If the result is not a constant, just in case, we will consider the result as unknown.
    if (const auto * column_const = column ? typeid_cast<const ColumnConst *>(column.get()) : nullptr)
    {
        /// Functions may produce a ColumnConst sized to match the input block; in DAG nodes
        /// the size is meaningless and we keep them at zero.
        if (!column_const->empty())
            node.column = ColumnConst::create(column_const->getDataColumnPtr(), 0);
        else
            node.column = column_const->getPtr();
    }

Note the “size is meaningless, keep them at zero” detail: a ColumnConst in a DAG node carries a value, not a row count. The row count is recomputed at execution. That is why SELECT 1 + 1 costs nothing at run time — the whole subtree collapses into one COLUMN node.

The fold is deliberately conservative. Only functions that declare themselves foldable are folded, and the second stage — folding a predicate whose arguments are constants — is restricted to a short list of value-only operators:

cpp 
// src/Interpreters/ActionsDAG.cpp:1218
const std::unordered_set<std::string> & foldablePredicateFunctions()
{
    static const std::unordered_set<std::string> functions{
        "equals", "notEquals", "less", "greater", "lessOrEquals", "greaterOrEquals", "and", "or", "not",
        /// `xor` never short-circuits and reads its arguments by value only, the same shape as `not`
        "xor",
        /// `isNull` / `isNotNull` only look at the null map of their argument, so a `ColumnConst`
        /// and the materialized column it wraps give the same answer
        "isNull", "isNotNull"};
    return functions;
}

From graph to flat list: the topological sort

This is the heart of the whole design. ExpressionActions::linearizeActions takes the graph and produces a std::vector<Action> in which:

  • every action has a result slot — an integer index into a fixed-size array of columns;
  • every argument is a slot number too;
  • the order is a topological order, so an action’s arguments are always already computed when it runs.

The algorithm is a Kahn-style topological sort, but with one twist: it does not use a priority queue on “number of children”, it uses two FIFO queues — one for ordinary actions and one for arrayJoin — so that arrayJoins are deliberately delayed as long as possible:

cpp 
// src/Interpreters/ExpressionActions.cpp:557
    /// There are independent queues for arrayJoin and other actions.
    /// We delay creation of arrayJoin as long as we can, so that they will be executed closer to end.
    std::queue<const Node *> ready_nodes;
    std::queue<const Node *> ready_array_joins;

The slot allocation is a stack of freed positions:

cpp 
// src/Interpreters/ExpressionActions.cpp:568
    /// Every argument will have fixed position in columns list.
    /// If argument is removed, it's position may be reused by other action.
    std::stack<size_t> free_positions;

and each action claims a slot, freeing the slots of arguments that nobody else needs:

cpp 
// src/Interpreters/ExpressionActions.cpp:582
        /// Select position for action result.
        size_t free_position = num_columns;
        if (free_positions.empty())
            ++num_columns;
        else
        {
            free_position = free_positions.top();
            free_positions.pop();
        }

The interesting bit is the “is this argument still needed?” decision, which is a reference count over the DAG’s reverse edges:

cpp 
// src/Interpreters/ExpressionActions.cpp:605
            ++arg.num_created_parents;

            ExpressionActions::Argument argument;
            argument.pos = arg.position;
            argument.needed_later = arg_info.used_in_result || arg.num_created_parents != arg_info.parents.size();

            if (!argument.needed_later)
                free_positions.push(argument.pos);

needed_later == false means “this is the last consumer of that slot”, so the slot can be recycled by the next action. This is a register allocator for columns. A DAG with 50 nodes but only 4 live values at any point runs with a 4-element scratch array, not a 50-element one.

Executing the plan

The execution model is spelled out in the comment above ExecutionContext:

cpp 
// src/Interpreters/ExpressionActions.cpp:798
    /// This struct stores context needed to execute actions.
    ///
    /// Execution model is following:
    ///   * execution is performed over list of columns (with fixed size = ExpressionActions::num_columns)
    ///   * every argument has fixed position in columns list, every action has fixed position for result
    ///   * if argument is not needed anymore (Argument::needed_later == false), it is removed from list
    ///   * argument for INPUT is in inputs[inputs_pos[argument.pos]]
    ///
    /// Columns on positions `ExpressionActions::result_positions` are inserted back into block.

So the “intermediate results” are ColumnPtrs in a std::vector of fixed size. The action list is the program; the column vector is the register file. There is no map and no name lookup between the actions — an argument is addressed by its integer slot, and the only hash lookup in the whole execution is the one that binds the input block’s columns to the input slots, once per block.

Running one action is a switch on the node type. For a function:

cpp 
// src/Interpreters/ExpressionActions.cpp:868
            ColumnsWithTypeAndName arguments(action.arguments.size());
            for (size_t i = 0; i < arguments.size(); ++i)
            {
                if (!action.arguments[i].needed_later)
                    arguments[i] = std::move(columns[action.arguments[i].pos]);
                else
                    arguments[i] = columns[action.arguments[i].pos];
            }

Note the std::move: if this is the last use of the argument, the ColumnPtr is moved out of the register file, which releases a reference count. If the argument is needed later, it is copied — a refcount bump, not a data copy. The data itself is never copied; ColumnPtr is a COW handle to a column that was already materialised somewhere upstream.

The actual call is one line:

cpp 
// src/Interpreters/ExpressionActions.cpp:894
                res_column.column = action.node->function->execute(arguments, res_column.type, num_rows, dry_run);

function is the prepared IExecutableFunction stored on the node. There is no virtual dispatch per row — there is exactly one virtual call per block per node, and inside it the function loops over the column itself, with SIMD-friendly typed accessors.

Constants skip the call entirely:

cpp 
// src/Interpreters/ExpressionActions.cpp:855
            if (action.node->column)
            {
                /// Do not execute function if it's result is already known.
                res_column.column = action.node->column->cloneResized(num_rows);
                /// But still need to remove unused arguments.
                for (const auto & argument : action.arguments)
                {
                    if (!argument.needed_later)
                        columns[argument.pos] = {};
                }
                break;
            }

cloneResized(num_rows) materialises the constant to the block’s row count — a memset, not a computation.

The plan is built once

The ExpressionActions object is constructed once, at plan time, and then executed over and over. The constructor is where the whole plan-time work happens: constant folding has already happened in the DAG; here the DAG is sorted, the short-circuit analysis runs, and (if enabled) the JIT replaces hot function nodes with compiled ones:

cpp 
// src/Interpreters/ExpressionActions.cpp:61
ExpressionActions::ExpressionActions(ActionsDAG actions_dag_, const ExpressionActionsSettings & settings_, bool project_inputs_)
    : actions_dag(std::move(actions_dag_))
    , project_inputs(project_inputs_)
    , settings(settings_)
{
    /// It's important to determine lazy executed nodes before compiling expressions.
    std::unordered_map<const ActionsDAG::Node *, size_t> short_circuit_eager_arguments;
    std::unordered_set<const ActionsDAG::Node *> lazy_executed_nodes = processShortCircuitFunctions(
        actions_dag,
        settings.short_circuit_function_evaluation,
        settings.short_circuit_function_evaluation_reorder_arguments,
        short_circuit_eager_arguments);

#if USE_EMBEDDED_COMPILER
    if (settings.can_compile_expressions && settings.compile_expressions == CompileExpressions::yes)
        actions_dag.compileExpressions(settings.min_count_to_compile_expression, lazy_executed_nodes);
#endif

At run time, ExpressionActions::execute is a loop over the action vector:

cpp 
// src/Interpreters/ExpressionActions.cpp:1032
    execution_context.columns.resize(num_columns);

    for (const auto & action : actions)
    {
        try
        {
            executeAction(action, execution_context, dry_run, allow_duplicates_in_input, settings.enable_lazy_columns_replication);
            checkLimits(execution_context.columns);
        }
        catch (Exception & e)
        {
            e.addMessage(fmt::format("while executing '{}'", action.toString()));
            throw;
        }

The modern pipeline does not even build a Block per chunk. ExpressionTransform uses a positional fast path where the input column positions are resolved once, at construction:

cpp 
// src/Processors/Transforms/ExpressionTransform.cpp:18
ExpressionTransform::ExpressionTransform(
    SharedHeader header_, ExpressionActionsPtr expression_, RuntimeDataflowStatisticsCacheUpdaterPtr updater_)
    : ISimpleTransform(header_, std::make_shared<const Block>(transformHeader(*header_, expression_->getActionsDAG())), false)
    , expression(std::move(expression_))
    , input_positions(expression->getInputPositions(*header_))
    , updater(std::move(updater_))
{
    set_spillables.add(expression->getActionsDAG());
}

and then per chunk:

cpp 
// src/Processors/Transforms/ExpressionTransform.cpp:66
    /// Fast path: run positionally against the fixed input header, avoiding per-chunk Block name-index work.
    auto columns = expression->executeOnColumns(
        chunk.detachColumns(), getInputPort().getHeader(), input_positions, num_rows, false, &getCancellationFlag());

The comment on getInputPositions in the header explains exactly what was eliminated:

cpp 
// src/Interpreters/ExpressionActions.h:126
    /// Positional execution for callers whose input structure is fixed (e.g. ExpressionTransform).
    ///
    /// `getInputPositions(header)` precomputes, once, the mapping (required input slot -> position in
    /// `header`); the result is passed to `executeOnColumns` on every chunk. This avoids the per-chunk
    /// name lookups and the construction of a `Block` name index (`index_by_name`) on both the input and
    /// the output, which is significant when the header has very many columns.

This is the whole point of the design in one sentence: the query is analysed once; the data is only ever touched through integer indices.

The DAG is cloned per shard

Because the plan is a value-like object with stable node pointers, it can be copied cheaply and safely. ActionsDAG::clone copies the node list and then repairs the child pointers through a map:

cpp 
// src/Interpreters/ActionsDAG.cpp:2453
ActionsDAG ActionsDAG::clone(std::unordered_map<const Node *, const Node *> & old_to_new_nodes) const
{
    ActionsDAG actions;

    for (const auto & node : nodes)
    {
        auto & copy_node = actions.nodes.emplace_back(node);
        old_to_new_nodes[&node] = &copy_node;
    }

    for (auto & node : actions.nodes)
        for (auto & child : node.children)
            child = old_to_new_nodes[child];

    for (const auto & output_node : outputs)
        actions.outputs.push_back(old_to_new_nodes[output_node]);

    for (const auto & input_node : inputs)
        actions.inputs.push_back(old_to_new_nodes[input_node]);

    return actions;
}

The same pattern exists one level up, in ExpressionActions::clone, which also has to fix up the action list, because each action holds a const Node *:

cpp 
// src/Interpreters/ExpressionActions.cpp:91
ExpressionActionsPtr ExpressionActions::clone() const
{
    auto copy = std::make_shared<ExpressionActions>(ExpressionActions());

    std::unordered_map<const Node *, const Node *> copy_map;
    copy->actions_dag = actions_dag.clone(copy_map);
    copy->actions = actions;
    for (auto & action : copy->actions)
        action.node = copy_map[action.node];

Why clone at all? Because a distributed query sends the same plan to N shards, and because a plan can be executed with different settings (a FINAL variant, a count-only variant, a PREWHERE variant). Because the nodes are immutable once built, a clone is a shallow copy of a std::list plus a pointer fix-up — microseconds, and completely independent of the data.

Pruning: the DAG is edited, not thrown away

Because the DAG is an explicit graph, “remove the columns nobody asked for” is a graph reachability problem rather than a string-matching problem. The implementation is a two-stage iterative DFS:

cpp 
// src/Interpreters/ActionsDAG.cpp:1049
    enum class VisitStage { NonDeterministic, Required };

    for (auto stage : {VisitStage::NonDeterministic, VisitStage::Required})

The two stages matter: the first pass classifies which surviving nodes depend on non-deterministic functions (now(), rand()), and the second decides which of them may be treated as constants. A node whose children all became constants is collapsed into a COLUMN node:

cpp 
// src/Interpreters/ActionsDAG.cpp:1111
                /// Constant folding. A lambda that captures nothing has no children, but its folded value is a
                /// constant like any other, and a FUNCTION node left behind would cross plan steps as a column.
                /// An INPUT is never folded: `ExpressionActions` binds each entry of `inputs` to a required column as an INPUT.
                if (allow_constant_folding && node->column
                    && (!node->children.empty()
                        || (node->type == ActionType::FUNCTION && WhichDataType(node->result_type).isFunction())))
                {
                    node->type = ActionsDAG::ActionType::COLUMN;
                    node->children.clear();
                    node->is_deterministic_constant = !non_deterministic_nodes.contains(node);
                }

and the dead nodes are erased from the list:

cpp 
// src/Interpreters/ActionsDAG.cpp:1126
    const auto erased_count = std::erase_if(nodes, [&](const Node & node) { return !required_nodes.contains(&node); });
    std::erase_if(inputs, [&](const Node * node) { return !required_nodes.contains(node); });

std::erase_if on a std::list is O(1) per element and keeps the surviving nodes’ addresses valid — which is exactly why the nodes are in a list in the first place.

The picture

text 
   parsed SQL
       |
       v
   ActionsDAG  (std::list<Node>, children = Node*)
       |   addInput / addColumn / addAlias / addFunction / addArrayJoin
       |   constant folding at every add
       |   pruning by reachability
       |   splitting for filter push-down
       v
   ExpressionActions::linearizeActions()
       |   topological sort
       |   slot allocation (register allocator over ColumnPtr slots)
       |   per-action argument lists = {slot, needed_later}
       v
   std::vector<Action>   <-- the plan, built ONCE
       |
       |  execute()  per block (65 409 rows)
       v
   columns[num_columns]  (ColumnPtr slots)
       |
       +--> columns[action.result_position] = function->execute(args)

Trade-offs

What the DAG buys:

  • No per-row interpretation. One virtual call per node per block, not per row. A 65 409-row block amortises the dispatch 65 409-fold.
  • Cheap and safe optimisation. Because dependencies are explicit, pruning is exact, filter push-down is a graph split, and constant folding is local. No string matching on column names, no re-derivation of dependencies.
  • A reusable, clonable plan. The plan is a value; it can be cloned per shard, per replica, per variant (FINAL, count), without re-analysing the query.
  • A tiny register file. The slot allocator keeps the live-column array as small as the dataflow allows.

What it costs:

  • Pointer-based, not index-based. The class is full of const Node *, NodeRawConstPtrs, and std::unordered_map<const Node *, ...>. Copying requires a fix-up pass; you cannot serialise a node list and reload it without a relinking step. The alternative — index-based nodes in a vector — is faster to copy but makes every mutation a reindexing problem.
  • The topological sort is eager. linearizeActions must see the whole DAG before a single action can run. There is no streaming/partial compilation of an expression tree.
  • One DAG, one shape. The DAG is built for one specific set of inputs. splitActionsForFilter, splitActionsBeforeArrayJoin and friends exist to carve it into pieces that can live in different pipeline steps, but the pieces must still be derived by explicit graph surgery, and the surgery itself has to be correct for every corner case (that is why the codebase has pages of comments about arrayJoin offsets and prewhere inputs).

The one thing to remember: the performance of a ClickHouse expression is not “how fast is the function” but “how many nodes are in the plan and how many blocks do they run over”. A plan is built once and amortised over millions of rows; a node is a virtual call per block, not per row.

The part people get wrong: the DAG is not a lazily-evaluated tree. Every node in it is a slot in a flat array and every function in it is called once per block — unless it is explicitly marked lazy, which is a different mechanism entirely (see the short-circuit article).

ClickhouseInternalsQuery planPerformance