DuckDB API: Joining Databases Without Moving Them First

The problem: the join that had nowhere to happen

Rohlik Group runs online grocery stores in five countries, and behind each one is a set of operational databases, roughly one per service: the e-shop, the warehouse system, the delivery fleet, pricing, workforce management, and so on. Mostly MySQL, some Postgres. They are separate servers, deliberately — a service owns its data, and nobody wants the warehouse system's load landing on the e-shop's replica.

That is good service design and an awkward reporting story. Plenty of questions we are asked every day need two of those databases at once. Which warehouses have orders stuck in a zone that is currently express-disabled? The orders are in rohlik. The zone settings are in wapi. Different servers, so there is no view, no JOIN, no query that can answer it in one place.

We had two ways around it, and we did not like either.

Go the long way round. Give each source table its own extract job into Snowflake, join them there, and send the result back down to the MySQL reporting database that Metabase reads. It works, and it is where most of our modelled data belongs. But count the steps behind a single operational report: one extract per source database, a transform that unions and joins them in the warehouse, and a write-back that pushes the result down again. Four scheduled hops, each waiting on the one before it, and the report can never be fresher than the slowest of them.

The chain is also longer than its parts suggest. Extract jobs run on a shared pool of slots, alongside every other extraction in the company, so a job starts when a slot frees up, not when it is triggered.

We measured that gap over ten days of production jobs. Off-peak it is nothing: a median of 4 seconds, never more than 7. At the top of the hour, where the hourly schedules collide, the same job waited a median of 16 seconds and as much as 33 — four times longer, on every single run. Those are small numbers, and we are not going to pretend otherwise. But they are paid per hop, they land on the busiest minute of the hour, and they appear in no job's runtime. They appear in the number someone is looking at.

Keep a small engine on the side. We had one, internally called quackaroni: a cron in our reporting repo that ran DuckDB queries against the replicas and wrote the results into the reporting databases. The idea was right, and that is the part worth saying plainly — it is what the service in this post grew out of. But it was a cron. No retries, no dependencies, no backfills, no task logs next to the rest of our pipelines, and its own deployment path. Anything that lived there was invisible to the orchestration everything else used.

So we built the thing the second option was reaching for, properly: a service that does the federated read, does the computation in memory, and writes the result where it is needed — driven from Airflow like every other pipeline we run.

What the DuckDB API is

A Python service (FastAPI) that takes SQL over HTTP, runs it through DuckDB against whichever databases the query touches, and either returns the result or writes it to a destination. It has been in production since February 2026, running on Kubernetes and autoscaling with load.

DuckDB is doing the load-bearing work here, through its database extensions. mysql_query('rohlik', '...') sends that SQL to the rohlik MySQL server and streams the result back, and because the statement is pushed down as written, the source's indexes are used. Two such calls in one query land in the same DuckDB session, in memory, where joining them is an ordinary local join. The federation is DuckDB's; what we built around it is everything that makes it a production data path — the routing, the five-country fan-out, the type handling, the write modes, and the operators that drive it.

Three things come out of that:

	Endpoint	Where the result goes
Query	POST /duckdb/query	Back to the caller as JSON
Transfer to Snowflake	POST /duckdb/transfer	A Snowflake table, written by the service
Transfer to MySQL	POST /duckdb/transfer/mysql_to_mysql	A MySQL table, written by the service

The two transfer endpoints matter more than they look. On that path the rows never travel through the Airflow worker — the service reads them, computes and writes them, and Airflow gets back a row count. A report rebuild moves no data through the orchestrator at all.

Country is a parameter, not a copy of the query

Every request carries a country. A single code ("cz") runs the query against that country's databases. A list, or the keyword "international", runs the whole query independently per country and unions the results with a country column added.

That is one entry, not five. The same query text, five sets of connection details, one result table. It is the single most-used feature of the service, because almost every report we have is a five-country report.

Database aliases are resolved per country from one config file: rohlik means a different host in Czechia than it does in Hungary, and the query never says which. Only the aliases a query actually mentions get attached — the parser reads them out of the mysql_query(...) calls, so a query that touches one database does not open connections to five.

Cross-database filtering: duck_cte

Pushing a statement down to MySQL is what makes this fast, but it also means one mysql_query() call cannot reference another — each one is a separate statement on a separate server. Filtering one database by the contents of another needs the values to make the trip.

That is what the duck_cte syntax does:

duck_cte_zones:
    SELECT id, id_restaurant
    FROM mysql_query('wapi', '''SELECT id, id_restaurant FROM wapi.zones WHERE express = 0''');

duck_query:
    SELECT o.id, o.id_restaurant, o.created_at, z.id AS zone_id
    FROM mysql_query('rohlik', '''SELECT id, id_restaurant, created_at
                                  FROM rohlik.orders
                                  WHERE id_restaurant IN @duck_cte_zones.id_restaurant''') o
    JOIN duck_cte_zones z ON z.id_restaurant = o.id_restaurant;

Each duck_cte_<name>: block runs first and its result is kept in the DuckDB session. @duck_cte_<name>.<column> is then substituted into later statements as a literal value list, so the IN (...) is evaluated by MySQL, on its own index, before any rows cross the wire. The named CTE is also an ordinary table in the final duck_query:, so you can join against it.

The practical effect is that the expensive side of the join is filtered by the cheap side, at the source. CTEs can be chained, and in multi-country mode the whole chain runs independently per country — a Czech order list is filtered by Czech zones, never by Hungarian ones.

Driving it from Airflow

Nothing in that service knows about scheduling. The whole orchestrating side lives in our Airflow repo, as four operators over one hook, and that is where a report becomes a pipeline.

Operator	What it does
DuckDBToSnowflakeLightweightOperator	Source or CTE query → Snowflake, in replace, upsert, truncate_insert or union_replace mode
DuckDBMySQLTransferOperator	Source or CTE query → a MySQL destination table, swapped in atomically
DuckDBAPIOperator	Run a query and push the result into XCom, for alerts and checks
DuckDBIcebergQueryOperator	Query S3- or GCS-backed Iceberg tables through read_iceberg()

And the pipelines themselves are YAML. A config file holds the destination and a list of entries; the DAG loops over them and builds one task each:

country: international
snowflake_database: DWH_PROD
snowflake_schema: DUVO

tables:
  - destination_table: r_int_live_warranty_claims
    load_mode: replace
    duckdb_timeout: 300
    query: |
      duck_cte_rohlik:
          SELECT ... FROM mysql_query('rohlik', '''SELECT ...''');
      duck_query:
          SELECT ... FROM duck_cte_rohlik ...

Adding a report is appending an entry. No new DAG, no new deployment, and — the reason the format is YAML and the SQL is plain — no shape an agent has to guess at. This is the same bet as the one behind RDF's config format: more and more of this work is drafted by agents, and a flat file of SQL plus a handful of options is something they get right the first time.

Today that is more than 200 table and report definitions across 43 DAGs, from a three-minute operational alert to a daily rebuild. The service autoscales behind them; an Airflow pool caps concurrent tasks at 40, against observed daily peaks of 28 to 36. One of those DAGs rebuilds 25 federated reports every 15 minutes, straight from the live country databases, and together these pipelines let us retire 55 of the old cron queries in one go.

Cancellation, which is less boring than it sounds

Airflow killing a task does not stop work that is already running inside a service. So each operator derives a task id by hashing the DAG id, task id, run id, try number and map index — deterministic, so the same attempt always produces the same id — sends it with the request, and calls the API's cancel endpoint from on_kill.

We know the rest of that story, and it is in the limits below.

Performance: where the Parquet engine actually wins

The Snowflake transfer path originally built a pandas DataFrame and wrote it with write_pandas. We added a second engine that has DuckDB write Parquet files directly and loads those with PUT + COPY INTO, and measured the two head to head on production replicas and production Snowflake — each write mode running the same source through both engines as a concurrent pair, so a pair always shares replica and pod conditions.

On the load-bearing measurement, rohlik.warehouse_timeslots, 1,595,362 rows across five countries:

	pandas	Parquet	Change
union_replace, service-side total	47.1 s	~31 s	−34%
upsert, service-side total	56.8 s	~47 s	−17%
Extract & transform (both modes)	12.7 s	9.9–10.2 s	−19% to −22%

Those are service-side timings. At the Airflow task level, where the service's work sits inside the usual orchestration overhead, the deltas were a more modest −11% to −17%.

The headline is not that Parquet is a faster way to do the same work. It is that it does less work. We expected the staging step to be the win, and it was not — an earlier change to snappy compression and 100,000-row chunks had already removed most of the write_pandas penalty. What is left is structural:

• No transposing. Both engines read MySQL identically and in parallel. The gap is what happens after: pandas does a concat plus a Series.apply per *_id column to build the country-unique keys, about 4.0 seconds; DuckDB emits a SQL concat straight into the Parquet file, about 1.77. That gap grows with row count times the number of id-like columns.
• No timestamp repair. On upsert, pandas rewrites every datetime column twice — once on the destination, once on the temp table — for 11.14 seconds at this size. Parquet's files carry proper logical types, so the columns arrive correct and that step is zero. This one is worth dwelling on: Parquet was actually 4.7 seconds slower in the staging step of the upsert run and still won the mode outright, purely by not doing a repair the other path needed.
• No counting pass. union_replace on pandas runs a serial COUNT(*) per country first, re-executing the caller's own predicate. Cheap on an unfiltered source — 4.80 seconds here. On a filtered one it is not: on a 26 GB table, the probe pass alone cost 1,432 seconds against 1,732 for the extract it was sizing up.

Two honest caveats, because the numbers above are more uneven than a table makes them look. First, this is n = 1 per cell. The extract-phase and repair-phase numbers are stable and reproducible; the Snowflake phase is not — we measured the same engine loading the same five Parquet files 57 seconds apart at 12.74 s and 23.3 s, a 1.8× spread. Any staging-phase difference smaller than that is noise, so we do not claim one.

Second, Parquet is not free to start. Five sequential PUTs, a schema inference and a temp-table creation are a fixed cost, and on an 18-row lookup table Parquet finished about 3.5 seconds behind pandas. Small tables stay on the default path.

What it is, and is not, for

On the transfer path, a bounded WHERE predicate does not reduce the work: extract time tracks the size of the source table rather than the number of rows that match. Incremental extraction of a large table is therefore not viable here today, and it is the first thing we check before putting a table on the service.

That makes the division of labour easy. Big incremental tables go to RDF, which reads in id ranges and is built for exactly that. The DuckDB API is for the federated and the operational: cross-database joins, five-country unions, reports measured in thousands or low millions of rows, rebuilt often. Neither tool is the general answer, and knowing which one a table belongs on takes about one question.

Three smaller limits, all known and all with the same character — things the service does not yet do, rather than things it does wrong:

• The Parquet engine does not serve the paginated path. Set source_chunk_size or sql_page_size and the request silently falls back to pandas. Large tables are exactly where pagination is used, so today the slower path is also the one still carrying the timestamp repair.
• Cancellation cannot always reach the pod that owns the task. The peer-pod lookup can fail with a 403, which the service swallows, and the cancel then answers 404 — indistinguishable from "no such task". We have watched the consequence: Airflow's timeout killed a task at 13:18:35, the server-side work carried on and wrote its target table at 13:36:18, while the retry had already started a second scan of the same 26 GB table.
• We have not located the crossover. We know 18 rows loses and 1.59 million wins. Between those, the engine choice is a guess with a sensible default.

What's next

Making bounded predicates actually reduce the work is the first item, because it decides how much more of our data can live on this path. After that, bringing the Parquet engine to the paginated path, and locating the crossover so the engine choice stops being a judgement call.

The more interesting direction is Iceberg. We are testing it internally across the data platform, and a DuckDB session that can join a live MySQL replica to an Iceberg table in the same query is a genuinely different shape of answer than either side gives alone. That is the one we are most curious about.

Thanks

To the Data Engineering team, for the migrations, the reviews and the reports that kept finding the edges — and to Devin, our coding agent, which wrote a good share of the service's code alongside us.

If you are solving federated reporting somewhere else, we would like to hear how. Especially if you have made predicate pushdown work.