Finding 886 rows in a billion-row join: making DuckDB 4× faster without touching the query

The problem: a needle, a haystack, and a query nobody wants to rewrite

Every Thunderduck query runs in its own container, running DuckDB over the customer's own data: Iceberg tables in S3, Postgres, MySQL, BigQuery and more. Before a query starts, we read its plan, estimate how much data it touches, and size the container for it, from 1 vCPU / 4 GB up to 16 vCPU / 64 GB.

One query kept showing up at the top of our slow list. It builds an audience list: take the contacts in one CRM segment, enrich them from a people table and an address table, and flag who has already bought, whose address is a duplicate, and whose address or email is on a suppression list. Here is its shape, with the names changed:

WITH segment AS (           -- under a thousand contacts from one CRM segment
  SELECT DISTINCT c.person_id, c.email
  FROM   crm_contacts c                        -- JSON properties, filtered by owner
  JOIN   segment_members m ON m.contact_id = c.contact_id
  WHERE  m.ref = '…'
),
people AS (                 -- 599 million rows, no selective filter
  SELECT person_id, first_name, last_name, primary_address_id AS address_id, primary_email AS email
  FROM   people
),
addresses AS (              -- 392 million rows
  SELECT address_id, line_1, line_2, city, region, zip, country
  FROM   addresses
  WHERE  line_1 IS NOT NULL AND country = 'US'
),
buyers AS (…),              -- JSON-extracted records from the e-commerce platform
blocked_addresses AS (SELECT DISTINCT address_id FROM blocked_addresses WHERE …),
blocked_emails    AS (SELECT DISTINCT email FROM suppression_entries
                      WHERE suppression_list_id IN (SELECT id FROM suppression_lists WHERE …))
SELECT people.*, addresses.*,
       row_number() OVER (PARTITION BY addresses.address_id
                          ORDER BY people.email DESC NULLS LAST) <> 1 AS is_duplicated_address,
       buyers.person_id IS NOT NULL             AS is_buyer,
       blocked_addresses.address_id IS NOT NULL AS is_address_blocked,
       blocked_emails.email IS NOT NULL         AS is_email_blocked
FROM segment
JOIN people                 ON people.person_id = segment.person_id
JOIN addresses              ON addresses.address_id = people.address_id
LEFT JOIN buyers            ON buyers.person_id = segment.person_id
LEFT JOIN blocked_addresses ON blocked_addresses.address_id = addresses.address_id
LEFT JOIN blocked_emails    ON blocked_emails.email = people.email
QUALIFY row_number() OVER (PARTITION BY people.person_id
                           ORDER BY people.email DESC NULLS LAST) = 1;

Eight tables, six CTEs, JSON extraction, three DISTINCTs, an IN subquery, five joins and two window functions. It returns 886 rows.

The two big tables are where the time goes: about 991 million rows between them, roughly 61 GB of Parquet in S3. Neither has a selective filter of its own. The only thing that makes them small is the join to a segment of fewer than a thousand people. That is a needle in a haystack: read nearly a billion rows to keep 0.0001% of them.

On our 8 vCPU / 30 GB tier it took 438 seconds. The duration wasn't the strange part. The CPU graph was: about 10% busy for seven minutes.

It could be written better. You could filter people by the segment's IDs up front instead of trusting the join to do it. But this query feeds a production application, and its exact semantics matter: which duplicate address wins, which email a person is reported with, how suppressions are flagged. Changing it means re-validating what the application does with the result, and the team that owns it was busy shipping something else.

The SQL that runs in production is the SQL that exists. So the engine has to make that query fast.

How Thunderduck solves it: tune DuckDB per query, from its plan

DuckDB's defaults are right for a laptop reading local files. They aren't right for a short-lived container reading tens of gigabytes from S3. Because Thunderduck already sizes a container for every query, it sets DuckDB for that size too. Every query now gets settings like these before it runs, and none of them change what the query means:

SET memory_limit = '24000MB';           -- below the container's RAM, on purpose
SET threads = 24;                       -- 3× the vCPUs, not 1×
SET dynamic_or_filter_threshold = 5000; -- push the segment's keys into the big scans

1. More threads than cores

DuckDB starts one worker thread per core by default. When those threads scan Parquet in S3, each one sends an HTTP range request and blocks until the bytes arrive. Eight threads means at most eight requests in flight, and the cores spend their time waiting on the network. Hence 10% CPU. We re-ran the same query on the same 8 vCPU container, changing only the thread count:

Going from 8 to 32 threads nearly tripled the read rate, and CPU was still under half used. We didn't ship 32. Every thread costs memory, so we settled on 3× the vCPUs for every tier. That gets most of the gain with room to spare.

2. Leave memory headroom, or more threads will get you killed

The 32-thread run peaked at 29.0 GB of a 30 GB container, even though DuckDB's memory_limit was 24 GB. DuckDB's limit covers DuckDB's own buffers. It doesn't cover everything the process allocates: per-thread HTTP and decompression buffers and allocator overhead live outside it. We measured roughly 150 MB per thread on top of a fixed ~900 MB.

DuckDB can spill to disk when it hits its own limit, but the container's limit is fatal: the kernel OOM-kills the process, and the query fails with nothing to show for minutes of work. With two window functions and three DISTINCTs, this query is exactly the kind that needs to spill. So every tier sets memory_limit below the container's RAM, with room for its threads:

3. Push the needle into the haystack

This is the step that matters most for this query. The segment is under a thousand person IDs. If DuckDB knew those IDs while scanning the 599-million-row people table, it could skip every Parquet row group whose statistics show it can't contain any of them. Then the people it finds give it a similar handful of address IDs to do the same against the 392-million-row address table.

DuckDB does exactly this, by pushing the join's actual keys into the scan of the other side. The catch is the default: it pushes the key list only when there are at most 50 distinct keys (dynamic_or_filter_threshold). Above that it pushes only the minimum and maximum, and for random IDs the min-to-max range covers almost every row group, so the filter skips nothing. Our segment was well over 50 keys, so DuckDB was effectively scanning blind.

We raised the threshold to 5,000 and measured it at the HTTP layer. On a sorted file with 4,000 row groups, a join with 1,000 keys went from reading 684 MB to 183 MB. On unsorted data it skips nothing, because every row group still spans almost every key. But it costs nothing either. Which leads to the last part.

4. Sort the haystack, so statistics can skip something

Row-group skipping only works when each row group holds a narrow range of the key, which means the data has to be sorted on that key. Both big tables were bucketed but not sorted within their buckets, and bucketing doesn't help here: DuckDB can't prune bucket partitions using join keys that only exist at run time. So we rewrote people sorted by person_id and addresses sorted by address_id.

With both tables sorted, the same query on the same 8 vCPU container dropped from about 166 s to 101 s. The bytes read barely changed, though: 46.7 GB, down from 47.3. The sorted files were faster to scan, but this scan applied the pushed keys row by row instead of skipping whole row groups. We're still investigating why, so treat the sorting gain on this query as real but not yet fully explained.

The result

For the very largest plans (250 million+ estimated rows) we also added a 16 vCPU / 64 GB tier. On a separate, heavier analytics query it halved the runtime, from 100.6 s on 8 vCPU to 51.5 s on 16, for almost the same cost per run ($0.0132 → $0.0134). The same query on BigQuery took 40 s and cost about $0.14 in slot time: still faster, at about ten times the price.

A bug we shipped along the way

Raising dynamic_or_filter_threshold made DuckDB's query profiles much larger, because a profile now lists every pushed key. Our container entrypoint passed the execution plan to the reporting step through an environment variable, and Linux limits a single environment string to 128 KB (MAX_ARG_STRLEN). A large profile made every later exec fail with E2BIG: the query had finished, but its result was never reported. We now pass plans through files. A performance setting can change the size of your telemetry, and the telemetry path needs limits too.

What this means on Thunderduck

• Nothing to configure: threads, memory_limit and the join-key threshold are set for every query from the size of its container, which is chosen from the query's plan before it starts. See https://docs.thunderduck.ai/reference/limits/
• Your SQL stays as it is: none of these settings change a query's results, so the application team doesn't have to rewrite anything to get the speed.
• Sort the tables you join by: rebuild a big Iceberg table once, sorted on the key you join or filter by, and pushed join keys get row groups to skip. A CREATE TABLE … AS SELECT … ORDER BY does it, shown below.
• Every query is priced before it runs, so a heavy query like this one shows its tier and cost up front instead of on the bill.

CREATE TABLE lake.default.people_sorted AS
SELECT * FROM lake.default.people
ORDER  BY person_id;

If you run DuckDB over object storage yourself

• Set threads above your core count. S3 scans are I/O-bound; 3× vCPUs worked well for us.
• Lower memory_limit as you raise threads. Budget ~150 MB per thread outside DuckDB's accounting, so heavy windows and DISTINCTs spill instead of getting OOM-killed.
• Raise dynamic_or_filter_threshold for needle-in-a-haystack joins. The default of 50 keys is far below a real segment.
• Sort big tables on the keys you join by. Bucketing alone doesn't help with run-time join keys; sorted row groups do.
• Measure bytes at the HTTP layer. DuckDB's own byte counters miss reads served from local files and its file cache.