Jordan Goodman

Tracing a DuckDB Query

Today, we will be tracing a DuckDB query plan alongside its corresponding c++ code.

DuckDB’s repo is roughly designed into the below sections:

Below are the query plan operations:

I have created a test DuckDB file in a local repo.

Here are the results of the test query:

SELECT
    category,
    count(*) AS orders,
    sum(amount) AS revenue
FROM orders
WHERE ordered_at >= DATE '2024-10-01'
GROUP BY category
HAVING sum(amount) > 100000
ORDER BY revenue DESC
LIMIT 5;

| category | orders |      revenue |

| toys     | 31,418 | 4,358,717.50 |

| sports   | 31,418 | 4,319,445.00 |

| office   | 31,418 | 4,280,022.50 |

| home     | 31,418 | 4,240,750.00 |

| garden   | 31,418 | 4,201,827.50 |

DuckDB turns SQL into a series of internal operations, which we can obtain via the EXPLAIN command before a query.

TOP_N: Top 5, sum(amount) DESC                         5 rows out
  PROJECTION: category, orders, revenue                 8 rows
    FILTER: sum(amount) > 100000.00                    8 rows
      PROJECTION: __internal_decompress_string(#0), #1, #2
        HASH_GROUP_BY: group #0; sum_no_overflow, count_star()  8 rows
          PROJECTION: category, amount                  251,344 rows
            PROJECTION: __internal_compress_string_uhugeint(#0), #1
              SEQ_SCAN orders: ordered_at >= 2024-10-01 251,344 rows

Here is what happened in the query:

Now we will dive into how the code executes these operations.

DuckDB processes data in DataChunk objects, which is a small in-memory batch of rows.

DuckDB gives the date condition to the scan before it starts reading.

The scan then produces the next batch of rows.

The code calls the “table function” which reads the orders table.

// src/execution/operator/scan/physical_table_scan.cpp
auto filters = table_filters ? *table_filters : GetTableFilters(op);
TableFunctionInitInput input(op.bind_data.get(), op.column_ids, op.projection_ids, filters,
                             op.extra_info.sample_options, &op);
global_state = op.function.init_global(context, input);

// PhysicalTableScan::GetData
TableFunctionInput data(bind_data.get(), l_state.local_state.get(), g_state.global_state.get());
function.function(context.client, data, chunk);
return chunk.size() == 0 ? SourceResultType::FINISHED : SourceResultType::HAVE_MORE_OUTPUT;

Next, the columns and prepared via the projection operation.

This is achieved by evaluating the list of expressions for each batch:

// src/execution/operator/projection/physical_projection.cpp
OperatorResultType PhysicalProjection::Execute(ExecutionContext &context, DataChunk &input, DataChunk &chunk,
                                               GlobalOperatorState &gstate, OperatorState &state_p) const {
    auto &state = state_p.Cast<ProjectionState>();
    state.executor.Execute(input, chunk);
    return OperatorResultType::NEED_MORE_INPUT;
}

In the group by operation, DuckDB finds the entry for each category in a hash table and updates its running count and revenue total:

// src/execution/operator/aggregate/physical_hash_aggregate.cpp
aggregate_input_chunk.SetCardinality(chunk.size());
aggregate_input_chunk.Verify();

for (idx_t i = 0; i < groupings.size(); i++) {
    auto &grouping_global_state = global_state.grouping_states[i];
    auto &grouping_local_state = local_state.grouping_states[i];
    OperatorSinkInput sink_input {*grouping_global_state.table_state, *grouping_local_state.table_state,
                                  interrupt_state};

    auto &grouping = groupings[i];
    grouping.table_data.Sink(context, chunk, sink_input, aggregate_input_chunk, non_distinct_filter);
}

After it has seen all matching orders, DuckDB finishes the hash table and reads the category totals back out, reducing the problem set from 251,344 to 8.

// PhysicalHashAggregate::FinalizeInternal
grouping.table_data.Finalize(context, *grouping_gstate.table_state);

// PhysicalHashAggregate::GetData
auto res = radix_table.GetData(context, chunk, *grouping_gstate.table_state, source_input);
if (chunk.size() != 0) {
    return SourceResultType::HAVE_MORE_OUTPUT;
}

Above, you see DuckDB’s implementation of radix, which is essentially a method of building local aggregate hash tables, dividing their entries into buckets using bits from each entry’s hash, and then processing those buckets independently.

The filter operation in the having sum(amount) > 100000 is a straight forward check against the data chunk:

idx_t result_count = state.executor.SelectExpression(input, state.sel);
if (result_count == input.size()) {
    chunk.Reference(input);
} else {
    chunk.Slice(input, state.sel, result_count);
}

DuckDB orders and limits by sending the incoming batch to a local heap, which is a small ranked list that makes it cheap to retain current rows.

In this case, it holds 5 entries.

// src/execution/operator/order/physical_top_n.cpp
sink.heap.Sink(chunk, &gstate.boundary_value);
sink.heap.Reduce();

// TopNHeap::Sink
if (heap_size <= SMALL_HEAP_THRESHOLD) {
    AddSmallHeap(input, sort_keys_vec);
}

// TopNHeap::AddSmallHeap
if (!EntryShouldBeAdded(sort_key)) {
    continue;
}
AddEntryToHeap(entry);

I hope this post helped shed some light into how sql plans get executed under the hood.

I am not an expert c++ developer by any means, but DuckDB’s code base is pretty readable and organized, which made tracing this a bit easier than I expected!