Joins

A lookup join enriches the streaming rows against a dimension table. The dimension is declared with --dimension name=<uri>, where the URI is a file loaded once or a Kafka changelog followed live, and the name is what the SQL joins against.

A bounded file dimension

A file dimension is drained once at startup. Map orders to a region and aggregate over the enriched column.

cat > orders.ndjson <<'EOF'
{"merchant_id":10,"amount":250}
{"merchant_id":20,"amount":42}
{"merchant_id":10,"amount":980}
{"merchant_id":30,"amount":110}
EOF

cat > merchants.ndjson <<'EOF'
{"id":10,"region":"EU"}
{"id":20,"region":"US"}
{"id":30,"region":"EU"}
EOF

ubik --from file://orders.ndjson \
     --dimension merchants=file://merchants.ndjson \
  "SELECT m.region, count(*) AS n, sum(o.amount) AS total
   FROM orders o JOIN merchants m ON o.merchant_id = m.id
   GROUP BY m.region"

{"region":"EU","n":3,"total":1340}
{"region":"US","n":1,"total":42}

The join is a normal SQL JOIN, with one equi-key predicate. LEFT JOIN keeps a streaming row whose key misses the dimension.

ASOF join

A dimension that changes over time carries a validity timestamp per version. An ASOF JOIN picks, per key, the single version whose valid_from is the greatest one at or before the event's time. This is DuckDB's ASOF JOIN.

cat > txns.ndjson <<'EOF'
{"id":10,"amount":100,"event_time":"2026-07-18 10:01:00"}
{"id":10,"amount":100,"event_time":"2026-07-18 10:03:00"}
EOF

cat > fx.ndjson <<'EOF'
{"id":10,"rate":0.90,"valid_from":"2026-07-18 10:00:00"}
{"id":10,"rate":0.95,"valid_from":"2026-07-18 10:02:00"}
EOF

ubik --from file://txns.ndjson \
     --dimension fx=file://fx.ndjson \
  "SELECT t.event_time, t.amount, r.rate, t.amount * r.rate AS converted
   FROM txns t ASOF JOIN fx r
     ON t.id = r.id AND t.event_time >= r.valid_from"

The 10:01 transaction picks the 0.90 rate that took effect at 10:00; the 10:03 one picks 0.95, effective from 10:02.

{"event_time":"2026-07-18 10:01:00","amount":100,"rate":0.9,"converted":90.0}
{"event_time":"2026-07-18 10:03:00","amount":100,"rate":0.95,"converted":95.0}

The condition is one equi-key predicate plus one >= or > inequality on the time columns.

A live changelog dimension

Point --dimension at a Kafka topic instead of a file, and the dimension is followed live: new versions arriving on the topic update it as the stream runs.

ubik --from kafka://broker/orders \
     --dimension fx=kafka://broker/fx_rates --follow \
  "SELECT ... FROM orders o ASOF JOIN fx r ON o.ccy = r.ccy AND o.event_time >= r.valid_from"

The pick is watermark-gated: an order whose event time is ahead of the dimension's own watermark waits in a buffer until the matching version is known, rather than joining against a version that might still be superseded. That buffer, the dimension's full version chain, its per-partition watermark and the changelog offset are all in the checkpoint, so a kill -9 mid-run resumes the as-of picks exactly, even against a compacted topic that has since dropped the superseded versions. This is proven byte-identical to a DuckDB batch ASOF over the same data across crash cycles; see exactly-once.

In process

The Python binding takes a dimension as a host DataFrame through tables= instead of a URI, so a pandas, polars or pyarrow table becomes the join's build side; see the Python quickstart.

When a resume rejects a dimension

A bounded dimension is rebuilt from its file on resume and verified against the provenance the checkpoint pinned, so a join never enriches against a dimension that changed under it.