Skip to content
Toolforge

10. Joins, aggregation and larger than memory

Most people’s impression of DuckDB was formed when a query bigger than RAM simply failed. That has not been true for several years, and the current behaviour is worth knowing precisely — including the places where it is still true.

The three operators that need memory

Section titled “The three operators that need memory”
   STREAMING — constant memory, whatever the
   input size:


     filter, projection, the PROBE side of a
     hash join, most window functions over a
     sorted input, LIMIT


   BLOCKING — memory proportional to the data:


     1. HASH JOIN BUILD    the whole build
                           side, in a hash
                           table
     2. HASH AGGREGATE     one entry per
                           distinct group
     3. SORT               the whole input

Only these three matter. Every out-of-memory problem in DuckDB is one of them, and each has its own spilling strategy.

Hash joins, and how they spill

Section titled “Hash joins, and how they spill”
   PHASE 1 — BUILD, in memory


   read the build side, hash each key,
   insert into a partitioned hash table.


   ┌────┬────┬────┬────┬────┬────┬────┬────┐
   │ p0 │ p1 │ p2 │ p3 │ p4 │ p5 │ p6 │ p7 │
   └────┴────┴────┴────┴────┴────┴────┴────┘
      partitions by hash(key) high bits




   MEMORY LIMIT REACHED
     ↓
   PHASE 2 — SPILL


   write whole partitions to the temp
   directory. Keep as many as fit.


   ┌────┬────┬────┬────┬─ ─ ┬─ ─ ┬─ ─ ┬─ ─ ┐
   │ p0 │ p1 │ p2 │ p3 │ p4 │ p5 │ p6 │ p7 │
   └────┴────┴────┴────┴────┴────┴────┴────┘
     in memory          on disk




   PHASE 3 — PROBE, in passes


   pass 1: probe with the whole probe side.
           rows hashing to p0–p3 are joined
           now. rows hashing to p4–p7 are
           WRITTEN OUT to their own partition
           files.


   pass 2: load p4, p5 from disk into memory.
           read back the corresponding probe
           partitions. join. emit.


   pass 3: p6, p7. same.

This is the “grace hash join”, and the key property is that hash partitioning puts matching rows from both sides in the same partition number. Partition 5 of the build side can only match partition 5 of the probe side, so each pass is a self-contained join over a slice that fits.

   COST


   in memory        1 pass over each side
   spilled          1 write + 1 read of the
                    spilled fraction of BOTH
                    sides


   A join that spills half its data does
   roughly 2× the I/O. It gets slower.
   It does not fail.

Aggregation

Section titled “Aggregation”
   SELECT user_id, count(*), sum(amount)
   FROM events GROUP BY user_id;


   ─ 200 distinct users
     → each thread's hash table is 200
       entries, in L1 cache. Free.


   ─ 400 million distinct users
     → 400 million entries × (key + 2
       aggregates + overhead)
     → tens of gigabytes

The memory cost of a GROUP BY is the number of distinct groups, not the number of rows. Aggregating a billion rows into 50 groups is trivial; aggregating a billion rows into 800 million groups is the hard case, and it is what “high-cardinality group by” means.

The spilling works the same way as the join:

   ─ each thread builds a PARTITIONED local
     hash table
   ─ when memory is tight, partitions are
     written to the temp directory
   ─ the combine phase merges partition by
     partition, loading one set at a time

A COUNT(DISTINCT x) is a group-by in disguise — it must track every distinct value. On a high-cardinality column this is the most common surprise OOM in DuckDB, and approx_count_distinct (a HyperLogLog sketch, fixed small memory, ~2% error) is very often the right answer.

Sorting

Section titled “Sorting”
   ORDER BY revenue DESC


   in memory:
     radix sort on the key columns, then
     reorder the payload


   larger than memory:
     ─ sort chunks that fit
     ─ write each sorted RUN to disk
     ─ MERGE the runs

External merge sort, and it works. But note something people forget:

-- this needs to sort ALL rows
SELECT * FROM huge ORDER BY ts;


-- this does NOT — it is a TOP-N
SELECT * FROM huge ORDER BY ts LIMIT 100;

ORDER BY ... LIMIT n becomes a Top-N operator that keeps only n rows in a heap, in constant memory. If you are sorting a huge table to look at the first few rows, add the LIMIT and the whole problem disappears.

preserve_insertion_order — the setting nobody knows about

Section titled “preserve_insertion_order — the setting nobody knows about”
SET preserve_insertion_order = false;

Default is true, and it means DuckDB guarantees that results come back in the order rows were read, even though sixteen threads processed them out of order. Maintaining that guarantee requires buffering.

   preserve_insertion_order = true


     thread 3 finishes morsel 40 first, but
     morsel 12 has not finished, so morsel
     40's output must be HELD until 12 and
     everything between has been emitted.


     → a large result set can hold a great
       deal of memory just to preserve an
       order nobody asked for.
   WHEN TO TURN IT OFF


   ✓ a query with an explicit ORDER BY —
     the order is imposed anyway
   ✓ an aggregate — the output order is
     arbitrary regardless
   ✓ COPY ... TO a Parquet file, where you
     do not care about row order
   ✓ any large scan feeding a sink


   ✗ SELECT * FROM 'file.csv' where you
     genuinely want the file's order
     preserved

For large COPY operations and large exports this setting alone frequently turns an OOM into a completed query. It is the single highest-value obscure setting in DuckDB, and chapter 13 lists it with the others.

Where it still cannot cope

Section titled “Where it still cannot cope”

Being honest, because “DuckDB handles larger than memory” is stated more absolutely than it deserves.

1. In-memory databases with no temp directory

Section titled “1. In-memory databases with no temp directory”
   duckdb.connect()      ← :memory:


   There is no database file, so by default
   there is nowhere to spill TO.


   FIX:
     SET temp_directory = '/fast/scratch';
     SET max_temp_directory_size = '200GB';

Set temp_directory on every in-memory connection that will do real work. Persistent databases default to spilling next to the database file; in-memory ones need to be told.

2. A single enormous transaction

Section titled “2. A single enormous transaction”

Chapter 3’s MVCC point. UPDATE or DELETE over a hundred million rows accumulates undo information in memory that cannot be spilled.

   ✗ DELETE FROM events WHERE ts < '2020-01-01';
     -- 300 million rows, one transaction


   ✓ delete in batches, committing between
   ✓ or, better: rewrite the table
     CREATE TABLE new AS
       SELECT * FROM events WHERE ts >= '2020-01-01';
     DROP TABLE events;
     ALTER TABLE new RENAME TO events;

For a column store, rewriting is usually cheaper than deleting. The same logic applies in ClickHouse and to Iceberg’s copy-on-write mode.

3. Window functions over huge partitions

Section titled “3. Window functions over huge partitions”
SELECT *, sum(amount) OVER (PARTITION BY user_id
                            ORDER BY ts
                            ROWS BETWEEN UNBOUNDED PRECEDING
                                     AND CURRENT ROW)
FROM events;

A window function’s partition is materialised. If one user_id has 200 million rows, that partition must be held. Window spilling has improved considerably, but it remains the operator most likely to be the thing that fails, and a heavily skewed partition key is the trigger.

4. list() and string_agg() building giant values

Section titled “4. list() and string_agg() building giant values”
-- one row, one list, containing 400 million elements
SELECT list(user_id) FROM events;

A single value cannot be spilled. It is one value.

5. Very wide result sets pulled into Python

Section titled “5. Very wide result sets pulled into Python”

The query may complete comfortably and then .df() materialises the whole thing as a pandas DataFrame in the same process. Use .fetchmany(), .record_batch(), or write to Parquet instead of returning it. Chapter 12 has the patterns.

Diagnosing a spill

Section titled “Diagnosing a spill”
-- did it spill, and by how much?
SELECT * FROM duckdb_temporary_files();


-- current memory picture
SELECT * FROM duckdb_memory();


-- the settings in force
SELECT current_setting('memory_limit'),
       current_setting('temp_directory'),
       current_setting('threads');
   SYMPTOM               LIKELY CAUSE


   slow, disk busy,      a spill. Working as
   query completes       designed. Raise
                         memory_limit or accept
                         it.


   "Out of Memory Error: ran out of memory in a
    could not allocate"  place that cannot spill,
                         or temp_directory is
                         unset/full


   the OS killed the     memory_limit is higher
   process               than the memory actually
                         available. See below.


   one core busy         skew, not memory.
                         Chapter 8.

The last one is worth stressing. memory_limit defaults to about 80% of system RAM, and DuckDB’s accounting does not cover everything — Python objects, Arrow buffers you are holding, the OS page cache. In a container, set memory_limit well below the container’s limit, because a cgroup OOM kill gives you no error message at all, just a dead process. Chapter 13 has the numbers.

What to take from this chapter

Section titled “What to take from this chapter”
  • Only three operators need memory proportional to the data: hash-join build, hash aggregate, and sort. Everything else streams in constant memory.
  • Hash joins spill with a grace/partitioned strategy — matching rows from both sides land in the same partition number, so each pass is a self-contained join. A spilled join does roughly 2× the I/O; it gets slower, it does not fail.
  • A GROUP BY’s memory is the number of distinct groups, not rows. COUNT(DISTINCT x) on a high-cardinality column is the commonest surprise OOM — approx_count_distinct is usually the right answer.
  • ORDER BY ... LIMIT n is a Top-N in constant memory. Adding the LIMIT you meant makes the problem vanish.
  • preserve_insertion_order = false is the highest-value obscure setting — the default holds output in memory to preserve an order you usually did not ask for.
  • Five things it still cannot do: in-memory databases with no temp_directory, one enormous transaction (rewrite instead of deleting), window functions over a skewed partition key, a single giant list() value, and materialising a huge result into pandas.
  • Diagnose with duckdb_temporary_files() and duckdb_memory().
  • In a container, set memory_limit well below the cgroup limit — a cgroup kill produces no error, just a dead process.

Next: concurrency and transactions — what “one writer” actually means, and the four patterns for working around it.

Written by Sepuri Sai Krishna

© 2026CC BY 4.0Source