Landing the changelog offline

A windowed or continuous query publishes a changelog: rows keyed by their (group, window) tuple, upserted, so the topic's latest-per-key state is the current aggregate (source and sink). Offline reads and backfills come from a lake table, so the same changelog has to reach Iceberg, Hudi or Delta. ubik emits the changelog and, in process, the Arrow result. The host commits the table. ubik owns no lake writer, no catalog client, and no file layout.

What ubik emits

Two feeds, both host-committable, both carrying the same rows.

The upsert rule

The offline table is an upsert keyed by (group, window), never a plain append.

A resume re-emits a window that was emitted before the crash, byte-identical, and that re-emission is what the idempotent key exists for (exactly-once). A latest-per-key consumer folds the duplicate away for free. A lake table that appends every record keeps both copies. So the committer has to MERGE on the message key (or dedupe on read), which is what every lake sink already does for a changelog topic. Configure it that way and a re-emitted window overwrites its own row instead of doubling it.

Schema and partitioning

The changelog value is JSON, so a Connect committer infers the column types. In process the Arrow result carries the types exactly, and ubik.validate(sql, ...) returns the result schema before a row is read, so the host can create the table up front.

The key is the (group, window) tuple. Lake tables partition by the window column for time-pruned reads, so the query must project it:

SELECT merchant,
       TUMBLE(event_time, INTERVAL '1 MINUTE') AS w,   -- project the window column
       count(*) AS n
FROM orders
GROUP BY merchant, TUMBLE(event_time, INTERVAL '1 MINUTE')

The changelog leg

Point a Kafka Connect Iceberg, Hudi or Delta sink at the --to topic, in upsert mode keyed by the message key, partitioned by the window column. Run the sink in transactional emit mode and set the connector to read committed so a partial window is never landed; run it in idempotent mode for lower latency when the committer dedupes by key regardless (emit modes).

{
  "connector.class": "io.tabular.iceberg.connect.IcebergSinkConnector",
  "topics": "orders_agg",
  "iceberg.tables": "warehouse.orders_agg",
  "iceberg.tables.upsert-mode-enabled": "true",
  "iceberg.tables.default-id-columns": "merchant,w",
  "consumer.override.isolation.level": "read_committed"
}

The (group, window) tuple is the identity, so id-columns is the group columns plus the window column. This is the same seam the online landing uses: one topic feeds a low-latency KV store and an offline job lands the same topic into the lake.

The Arrow leg

In process there is no broker. Iterate the pipeline, buffer the batches, commit them to the lake with your own writer, then checkpoint. Commit first, checkpoint second: a crash between the two replays the uncommitted tail into the same idempotent upsert, so the table never has a hole.

import ubik, pyarrow as pa
from deltalake import DeltaTable

Q = ("SELECT merchant, TUMBLE(ts, INTERVAL '1 MINUTE') AS w, count(*) AS n "
     "FROM orders GROUP BY merchant, TUMBLE(ts, INTERVAL '1 MINUTE')")
dt = DeltaTable("s3://lake/orders_agg")

with ubik.pipeline(Q, from_="kafka://localhost:9092/orders",
                   checkpoint="~/.ubik/orders") as p:
    buf = []
    for window in p:                    # each closed window, as a pyarrow RecordBatch
        buf.append(window)
        if len(buf) >= 256:
            (dt.merge(pa.Table.from_batches(buf),         # upsert on the (group, window) key
                      predicate="t.merchant = s.merchant AND t.w = s.w",
                      source_alias="s", target_alias="t")
               .when_matched_update_all()
               .when_not_matched_insert_all()
               .execute())
            buf.clear()
            p.checkpoint()              # commit to the lake THEN checkpoint

A one-shot backfill over a bounded range reads the whole result at once:

tbl = ubik.stream(Q, from_="kafka://localhost:9092/orders").arrow()
write_deltalake("s3://lake/orders_agg", tbl, mode="overwrite")

What ubik does not do

ubik does not own the table. It writes no data files, holds no catalog client, speaks to no object store, and picks no partition layout. The DuckDB Parquet extension is trimmed out of the build on purpose, so --to accepts a kafka:// topic and nothing else: there is no --to file://out.parquet. The moment an engine names files it inherits manifests, compaction and catalogs, which is a lake writer, which is storage. The changelog and the Arrow result are the seam; the host commits on its side.