Duckherder Adds Distributed Query Execution to DuckDB

Duckherder - DuckDB Remote and Distributed Execution Extension

Duckherder is a DuckDB extension built upon storage extension that enables (certain) distributed query execution across multiple worker nodes using Arrow Flight for efficient data transfer. It allows you to seamlessly work with remote tables and execute queries in parallel across distributed workers while maintaining DuckDB's familiar SQL interface.

Warning

This is a personal project and has nothing to do with DuckLabs. It is not affiliated with, endorsed by, or associated with DuckDB Labs in any way.

WIP Disclaimer

This repository is currently a work in progress, and not yet for production usage. Feel free to play around with it, give me feedback, and ping me for feature request and collaboration!

Table of Contents

Overview

Architecture

Object-Storage-Backed Deployment Model

Distributed Execution System

Installation

Object Storage

Usage

Roadmap

Overview

Duckherder implements a client-server architecture with distributed query execution:

Client (Duckherder) : Coordinates queries, manages remote table references, and initiates distributed execution

Server (Duckling) : Consists of two components: Driver Node : Analyzes queries, creates partition plans, and coordinates worker execution Worker Nodes : Execute partitioned queries on local data and return results via Arrow Flight

Driver Node : Analyzes queries, creates partition plans, and coordinates worker execution

Worker Nodes : Execute partitioned queries on local data and return results via Arrow Flight

Distributed Executor : Runs on the driver node, partitions queries based on DuckDB's physical plan analysis, and distributes tasks to workers

The extension transparently handles query routing, allowing you to run CREATE, SELECT, INSERT, DELETE, and ALTER operations on remote tables through DuckDB's storage extension interface.

Architecture

High-Level System Architecture

┌─────────────────────────────────────────┐ │ Client DuckDB Instance │ │ ┌───────────────────────────────────┐ │ │ │ Duckherder Catalog (dh) │ │ │ │ - Remote table references │ │ │ │ - Query routing logic │ │ │ └───────────────────────────────────┘ │ │ │ │ │ │ Arrow Flight Protocol │ │ │ │ └──────────────┼──────────────────────────┘ │ ┌──────────────┴──────────────────────────┐ │ Server DuckDB Instance(s) │ │ ┌───────────────────────────────────┐ │ │ │ Driver Node │ │ │ │ - Distributed executor │ │ │ │ - Query plan analysis │ │ │ │ - Task coordination │ │ │ └───────────┬───────────────────────┘ │ │ │ │ │ ┌───────────┴───────────────────────┐ │ │ │ Worker Nodes │ │ │ │ - Actual table data │ │ │ │ - Partitioned execution │ │ │ │ - Result streaming │ │ │ └───────────────────────────────────┘ │ └─────────────────────────────────────────┘

Object-Storage-Backed Deployment Model

The target architecture for integrating distributed execution with an object-storage-backed DuckDB database follows a single-writer, multiple-reader model:

The Driver/Coordinator owns the only write authority . In the initial design, the Writer runs inside the Driver process. It owns the read-write database handle and writer lease, applies DDL/DML, and publishes new checkpoints or snapshots to object storage.

All worker nodes are readers . They execute partitioned read tasks against the snapshot selected by the coordinator and never publish database changes.

The client is a SQL and session endpoint . It registers storage and cluster configuration, attaches the database, and submits queries without coordinating individual workers or writing database files directly.

Object storage is the source of truth for database data, manifests, and committed snapshots.

+------------------+ SQL / transaction +------------------------------------+ | Client CLI / App | -----------------------> | Driver / Coordinator | | - ATTACH | <----------------------- | | | - Session | results | +------------------------------+ | +------------------+ | | Gateway + Result Merger | | | +------------------------------+ | | | Catalog + Query Planner | | | | table ID + schema + snapshot | | | +------------------------------+ | | | Transaction Coordinator | | | | BEGIN / COMMIT / ROLLBACK | | | +------------------------------+ | | | Writer (READ-WRITE) | | | | writer lease + epoch | | | | DDL/DML + checkpoint/publish | | | +------------------------------+ | +----------+-------------+-----------+ | | read tasks + snapshot ID | | write + | | publish v v +----------------+ +------------------+ | Worker Pool | | Object Storage | | READ-ONLY | | data + manifests | | Worker 1 ... N |->| + checkpoints | +-------+--------+ +------------------+ | ^ | | +-------------+ read pinned snapshot

partial results return to Gateway + Result Merger

Each query is pinned to one committed snapshot before tasks are sent to workers. A newly committed write is visible to new queries, while already running queries continue reading their original snapshot. The coordinator remains the commit authority even if the Writer is later moved from the Driver process into a dedicated writer process.

This section describes the intended object-storage-backed distributed architecture. Transaction coordination, snapshot propagation, and direct worker reads from shared object storage are still work in progress.

Distributed Execution Flow

┌─────────────┐ │ COORDINATOR │ Extract query plan └──────┬──────┘ │ ├─ Phase 1: Extract & validate logical plan ├─ Phase 2: Analyze DuckDB's natural parallelism ├─ Phase 3: Create partition plans ├─ Phase 4: Prepare result schema │ ├──────────┬──────────┬──────────┐ ▼ ▼ ▼ ▼ ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐ │WORKER 0│ │WORKER 1│ │WORKER 2│ │WORKER N│ └───┬────┘ └───┬────┘ └───┬────┘ └───┬────┘ │ │ │ │ │ Execute │ Execute │ Execute │ Execute │ partition│ partition│ partition│ partition │ (Local) │ (Local) │ (Local) │ (Local) │ │ │ │ └──────────┴──────────┴──────────┘ │ ▼ ┌─────────────┐ │ COORDINATOR │ Merge results │ Combine │ Smart aggregation │ Finalize │ Return final result └─────────────┘

Distributed Execution System

Partitioning Strategy

The distributed executor analyzes DuckDB's physical plan to create optimal partitions:

Queries EstimatedThreadCount() from DuckDB's physical plan

Understands how many parallel tasks DuckDB would naturally create

Extracts cardinality and operator information

The system uses a three-tier hierarchy to select the optimal partitioning strategy:

Row group-based partitioning (preferred):

Aligned with DuckDB's storage structure (122,880 rows per group)

Tasks process complete row groups (e.g., Task 0: groups [0,1], Task 1: groups [2,3])

Optimal cache locality and I/O efficiency

Respects DuckDB's natural parallelism boundaries

Range-based partitioning (fallback):

Used when row groups can't be detected but cardinality is known

Contiguous row access

Not aligned with row group boundaries

Modulo partitioning (final fallback):

Scattered row access (poor cache locality)

Works for any table size and operator type

Simple and always correct

When row groups are detected (the common case), the system:

Calculates total row groups: cardinality / 122,880

Assigns whole row groups to tasks (e.g., Task 0 gets groups [0,1], Task 1 gets groups [2,3])

Converts row group ranges to rowid ranges: rowid BETWEEN (rg_start * 122880) AND (rg_end * 122880 - 1)

Workers scan complete row groups for optimal performance

DuckDB Thread Model → Distributed Model Mapping

Installation

Building from Source

Install VCPKG for dependency management:

git clone https://github.com/Microsoft/vcpkg.git ./vcpkg/bootstrap-vcpkg.sh export VCPKG_TOOLCHAIN_PATH= ` pwd ` /vcpkg/scripts/buildsystems/vcpkg.cmake

Build the extension:

Clone the repo. git clone --recurse-submodules https://github.com/dentiny/duckdb-distributed-execution.git cd duckdb-distributed-execution # Build with release mode. export VCPKG_TOOLCHAIN_PATH=/path/to/vcpkg/scripts/buildsystems/vcpkg.cmake CMAKE_BUILD_PARALLEL_LEVEL= $( nproc ) make

The build produces:

./build/release/duckdb - DuckDB shell with extension pre-loaded

./build/release/test/unittest - Test runner

./build/release/extension/duckherder/duckherder.duckdb_extension - Loadable extension

./build/release/extension/duckdb_object_storage/duckdb_object_storage.duckdb_extension - Object-storage filesystem extension

./build/release/distributed_server - Standalone distributed driver node

./build/release/distributed_worker - Standalone distributed worker node

Running Standalone Server and Workers

For production deployments or testing distributed execution across multiple machines, you can run standalone server and worker processes:

Start the driver server on default host (0.0.0.0) and port (8815) without workers ./build/release/distributed_server # Start the driver server on a specific host and port ./build/release/distributed_server 192.168.1.100 8815 # Start the driver server with 4 local workers (for single-machine distributed execution) ./build/release/distributed_server 0.0.0.0 8815 4