The S3 lake

queen-s3 writes a queue’s log into an object store in a format no Queen code has to read back. Objects are JSONL or Parquet under a Hive-partitioned layout, so DuckDB, Spark, Trino, Athena, ClickHouse and Snowflake query them with nothing in front of them. Running the connector is the operator page; this page is the contract a reader gets.

Two sentences first, because everything else assumes them.

The lake mirrors the log, not the outcome of processing. The sink reads through POST /api/v1/fetch, which takes no lease, moves no cursor and ignores consumer state entirely. A record that a consumer group later nacked, dead-lettered or never touched is in the lake exactly like one that was acked on the first delivery. That is the correct behaviour for a lake, and it has to be said out loud because the word “sink” makes people assume “what was consumed”.

The lake is plaintext. A fetch decrypts a payload when the broker has an encryption key configured, so a queue encrypted at rest inside Queen is readable text in the bucket. Bucket-side encryption is what restores the property, and it is not optional; the deploy page has the header and the key policy.

The bucket layout

<prefix>/queue=<queue>/dt=<date>/hour=<hour>/w-<k>-<tStart>-<tEnd>.<ext>
<prefix>/_queen/<queue>/windows/<k>.json
<prefix>/_queen/<queue>/checkpoint/<k>.json.zst

Real keys, at the default prefix queen and the default hourly alignment:

queen/queue=orders/dt=2026-09-04/hour=10/w-0000001842-1788516000000000-1788516300000000.jsonl.zst
queen/queue=orders/dt=2026-09-04/hour=10/w-0000001842-1788516000000000-1788516300000000.parquet
queen/_queen/orders/windows/0000001842.json
queen/_queen/orders/checkpoint/0000001840.json.zst

And under QUEEN_S3_LAYOUT=per-partition, one object per lane per window, with the offset range in the name, which is the Connect-shaped key a reader that wants one entity’s file can find by name:

queen/queue=orders/dt=2026-09-04/hour=10/w-0000001842-1788516000000000-1788516300000000-p-cust-0420-000000001400-000000001811.parquet

The parts, and what each one is for:

Three properties are pinned by tests, and something depends on each:

Deterministic. The key is a function of the window intent (k, tStart, tEnd) and nothing else: no wall clock, no attempt counter, no random suffix. That is what makes a retried upload an overwrite of the same key with the same bytes rather than a second object, and it is why exactly-once here needs no conditional PUT and no LIST.

Escaped. Everything outside [A-Za-z0-9._-] is percent-encoded with uppercase hex, so a partition named a/b, a queue with a space, and a name containing .. cannot escape their prefix. An ordinary name (orders, cust-0420, svc.billing) is never rewritten.

Ordered. Keys sort lexicographically in window order, because k is zero-padded and leads. A reader that lists a bucket gets the windows in commit order for free.

_queen/ is deliberately not a Hive partition: a reader globbing queue=*/dt=*/hour=*/* walks straight past the sidecars without excluding anything by name.

The record envelope

Five fields, the same five in both formats. There are deliberately no headers: a stored frame has no header map and the broker refuses to fake one.

ts being the segment’s commit time is the one thing to internalise before building a table on it. Event time is a field inside the payload, and extracting it is the reader’s job. The lake is partitioned by when Queen committed a record, which is what makes a window a deterministic set.

The queue name is not one of the five. It is the queue=<queue> key of the path and nothing else, which is exactly where Spark and Hive put a key they wrote with partitionBy, and every reader with partition discovery hands it back as a column for free. Repeating it in the rows as well was the earlier shape and it was wrong in two concrete ways: a Hive or Athena table may not have a column and a partition key of the same name, and PyArrow’s dataset reader types the discovered key as dictionary<string> while the file’s own column is string, so pq.read_table('<prefix>') and pd.read_parquet('<prefix>'), the first thing a notebook types, failed to merge the schemas. A reader that opens one object with no path around it gets the queue from the Parquet footer (queen.queue, below); for a lone JSONL object, the key in the path it came from is the only record of it.

JSONL

One record per line, five fields, always in this order and with these spellings:

{"partition":"cust-0420","offset":1811,"transactionId":"ord-9f21","ts":"2026-09-04T10:03:41.918204Z","payload":{"type":"paid","amount":1290}}

payload is the broker’s own bytes, byte for byte. Nothing parses a payload into a tree, which means this writer never decides how a float or a large integer is spelled: whatever the producer sent is what the lake holds.

The bytes are a pure function of the records, and each of these is a pin with a test behind it: field order and spelling are fixed here rather than derived from a struct a refactor could reorder; ts is rendered by the same code every other surface of the connector prints a timestamp with; string escaping is serde_json’s, compared against it on a corpus rather than asserted in a comment; the zstd and gzip levels are constants, because a level is part of the bytes; and gzip headers carry mtime 0 with no filename and no comment, because GzBuilder’s default mtime is the wall clock and would put a timestamp inside the object.

Parquet

The low-level writer, with no arrow anywhere in the dependency tree. Five fixed columns:

message queen_record {
  required binary partition (STRING);
  required int64 offset;
  required binary transaction_id (STRING);
  required int64 ts (TIMESTAMP(MICROS,true));
  optional binary payload (STRING);
}

transaction_id is snake_case here and transactionId in JSONL, because each follows the convention its own format is read with. payload is the JSON text, so a reader extracts from it with json_extract or its local equivalent. It is the one nullable column.

Statistics are Chunk: min and max per row group, which is the pruning a reader can act on. Page statistics and the column index are deliberately off, because they would only add footer weight.

Everything that can move a byte is pinned rather than inherited from the crate’s defaults: created_by is the constant queen-s3 (the crate default embeds its own version, and the writer’s identity belongs in the manifest instead); key/value metadata is exactly two pairs, queen.envelope=1 and queen.queue=<queue>, both constant for the object, so the footer is still a pure function of the window; the writer version is PARQUET_1_0, so data pages are V1, which is what every reader below takes; compression is ZSTD level 3 or Snappy, at a fixed level; row groups close on a fixed record count rather than on bytes or time; and the data page size, dictionary page size, data page row count, write batch size and statistics truncation are all named constants, so a change to a crate default is a test failure rather than silently different bytes.

What is deliberately not pinned is the parquet crate version. A library upgrade may move the bytes, which is why the manifest records which writer produced each object and why the lockfile is bumped deliberately.

The sidecars

Neither is required to read the lake. The data objects tile the timeline on their own, and both sidecars live under _queen/ where a Hive glob does not see them.

The manifest

_queen/<queue>/windows/<k>.json, one per committed window, and the only place a wall-clock value is written. The data object carries none, which is what makes a retry byte-identical.

{
  "sink": "default",
  "queue": "orders",
  "k": 1842,
  "tStart": 1788516000000000,
  "tEnd": 1788516300000000,
  "format": "jsonl",
  "compression": "zstd",
  "layout": "merged",
  "objects": [
    { "key": "queen/queue=orders/dt=2026-09-04/hour=10/w-0000001842-1788516000000000-1788516300000000.jsonl.zst",
      "bytes": 44139012,
      "records": 918442,
      "sha256": "9f2c…" }
  ],
  "records": 918442,
  "bytes": 44139012,
  "partitions": 5183,
  "minTs": 1788516000121004,
  "maxTs": 1788516299904881,
  "lost": [],
  "writer": "queen-s3/1.5.0 jsonl+zstd",
  "committedAt": "2026-09-04T10:05:07.412000Z"
}

tStart, tEnd, minTs and maxTs are integers: microseconds since the epoch, the same clock the key encodes. committedAt is the only value on PostgreSQL’s clock’s opposite side, the sink’s own wall clock at commit, taken in milliseconds and rendered at microsecond precision, so its last three digits are always zeros. partitions counts the distinct lanes that contributed at least one record; each object’s sha256 is the SHA-256 of that object’s bytes, in hex. objects has one entry under merged and one per lane under per-partition, where each entry also carries partition, firstOffset and lastOffset. lost is normally empty and is the one thing to alert on. writer is what the object was written by, which is the field to consult if two objects of the same shape ever differ in bytes.

A manifest is what a verification pass reads, and it is also, deliberately, most of what an Iceberg manifest entry needs: row count, byte size and the time bounds. Nothing today writes a catalog.

The checkpoint

_queen/<queue>/checkpoint/<k>.json.zst, written every QUEEN_S3_CHECKPOINT_EVERY windows: for each tracked partition, the next offset the window after k would read from.

{ "k": 1840, "tEnd": 1788515700000000, "positions": [["cust-0420", 1812], ["cust-0421", 44]] }

It is a cache and never the commit truth. A stale entry costs re-read bytes that the window filter throws away; a missing one costs a backwards probe-seek, or a read from the log start. Nothing here can lose or duplicate a record, only make a restart slower. Positions are sorted by partition name before serialisation and the zstd level is a constant, so the same position map always encodes to the same object.

One rule the restart path follows and it is worth knowing: the checkpoint loaded is the newest one at or below the committed window. A checkpoint written for a window that was never committed names positions past the commit truth, and reading from it would skip records. That is the only way a position could cost more than a re-read, and it is closed by an inequality.

The window commit

The whole design is one idea, and everything a reader can rely on follows from it.

A window is the set of records whose segment ts lies in [T_{k-1}, T_k). Boundaries come from PostgreSQL’s clock and from nowhere else. The sink never reads its own wall clock to make one: a sink that compared its SystemTime to a ts would be wrong by construction, and the window engine is written so that the clock is not reachable from it at all.

safeTime is the ceiling. The broker answers, with every discovery call, a watermark below which no segment can still become visible: the oldest xact_start among in-transaction sessions, minus a guard, since a record’s created_at is stamped during the inserting transaction and so is at or after that transaction’s start. A window may close only at or below that value, minus QUEEN_S3_SAFE_GUARD_MS.

Given that, window k is a deterministic set:

1. Every segment inside the window is already visible, so nothing can be added to it later.
2. Nothing can be removed from it except by retention, which is the alarm case.
3. Within a partition, offsets and ts are co-monotone, so the window’s records for one lane are one contiguous offset range, findable by reading forward from any position at or below its start.
4. The writer sorts by (partition, offset) and puts no wall-clock value in the object.

So rebuilding window k from scratch produces the same records in the same order, and with the writer pins above, the same bytes. That is the trick: a retried upload is byte-identical without the object name having to enumerate a million offsets.

The commit sequence around it is three durable steps in Queen’s key/value store and the bucket:

What exactly-once means here, precisely

It holds for every reader, including one that only lists the bucket. Each window exists at most once, with one content, and the windows tile [T_0, committed.tEnd) with no overlap and no gap. No manifest is needed to read correctly; the manifest is for verification.

It is about the lake, not about delivery. Two things follow that people expect not to:

• A record a consumer group nacked, dead-lettered or evicted is in the lake, because the read path ignores consumer state.
• Duplicates by transactionId can exist in the log itself. An idempotency key reused outside the broker’s dedup window produces two messages, and the lake reproduces the log faithfully. A reader that wants one row per transactionId writes it: QUALIFY ROW_NUMBER() OVER (PARTITION BY transactionId ORDER BY ts, offset) = 1, or the local equivalent.

What the lake does not contain is a second copy of a window under a different name. That is the property the deterministic key and the deterministic bytes exist to give.

Two findings that shape when a window closes

Both were found while the engine was being built, both changed the code, and both are the kind of thing that would have been a silent hole rather than a visible bug.

The close is anchored on the last complete discovery pass, not on the newest safeTime seen. Incremental discovery only reports what moved, so a partition created (or written for the first time) after a pass is one the sink has never named. Anchoring on that pass’s watermark is what makes it safe: such a partition holds only records at or above that watermark, so they land at or above any boundary this close can choose, and the window that eventually ships them is a later one. Closing against a newer safeTime would let a window run past a partition discovery has never seen, and its records would then be dropped by the “already committed” filter, silently. The same value floors the frontier of caught-up partitions, which is what lets a queue of a million lanes close windows without polling every lane.

A partition whose fetch failed keeps blocking the close. A fetch error that is not one of the two structural answers is retried on the next tick, and until it answers, its partition holds the window open. Skipping it would be worse than slow: the records nobody read would arrive in a later window, where the “already committed” filter would drop them for being below its start. Lag is the correct symptom of a partition that cannot be read.

The crash matrix

The process killed at any point, and what a reader sees afterwards. These are the five fault points the connector can be told to inject, so this table is exercised rather than argued.

A single PUT is atomic per key, so no reader ever sees a partial object, and no reader ever sees two different versions of window k.

Retention overrun

This is the one failure that is data loss, and it is never silent.

If retention deletes a segment before the sink read it, the sink learns it in one of exactly two ways: a fetch below the log start is answered OFFSET_OUT_OF_RANGE, or discovery reports a log start above the sink’s position. Either way it records the gap and keeps committing, because a stalled sink loses more than a lagging one:

• the offset range goes into that window’s manifest, as "lost": [{ "partition": "cust-0420", "from": 1400, "to": 1811 }], where from and to are both inclusive;
• queen_s3_records_lost_total{queue} counts the missing offsets;
• one sampled line is logged;
• the sink resumes from the new log start.

So a lake with a hole says so, in the window where the hole is, in a file a query engine can read. A verification job over the manifests is one WHERE json_array_length(lost) > 0.

queen_s3_lag_seconds is the metric that was climbing for hours before any of that happened, and the retention rule is how it is prevented rather than detected.

Reader recipes

These are measured, not sketched. The connector’s compat lane builds a small lake with the sink’s own writers, opens it with each reader below, and judges every cell on exactness rather than on “it opened”: the record count per queue and per partition, the number of NULL payloads, and a digest of the partition|offset list in the order that reader returned it. The generated table, with the version of every library and the exact call, is connectors/queen-s3/compat/MATRIX.md. This is what it said on 2026-09-04, over 5 000 records in each of the five (format, compression) prefixes:

Which reader takes which compression is the part that varies, and it is the whole reason gzip is an option: two of the six readers cannot decompress a zstd JSONL object out of the box. Spark’s JSONL path goes through Hadoop’s ZStandardCodec, a JNI wrapper that needs a libhadoop built with zstd, and the official image does not carry one; the failure is at decompression rather than at codec lookup, and Parquet with zstd in the same JVM is unaffected because parquet-java has its own binding. pandas needs the zstandard package installed. Both read gzip with nothing extra, and Parquet sidesteps the question entirely by naming its codec inside the file, per column chunk, which is why the two Parquet columns are green everywhere.

Trino was stood up but never measured, so its row is an admission rather than a result. Two things that container did teach are in the DDL below.

DuckDB, straight off the bucket, with Hive partitions pruned:

SELECT partition, offset, ts, payload->>'type' AS type
FROM read_parquet('s3://my-lake/queen/queue=orders/dt=*/hour=*/*.parquet',
                  hive_partitioning = true)
WHERE dt = '2026-09-04' AND partition = 'cust-0420'
ORDER BY offset;

One entity’s exact history, in order, off an object store, with no Queen in the path. Scoping the glob to one queue= is the simple form; hive_partitioning = true over queue=* is what gives queue back as a column when several queues are read into one table.

The JSONL objects read the same way with read_json_auto, which is the form to reach for when the payload shape varies:

SELECT partition, offset, ts, payload
FROM read_json_auto('s3://my-lake/queen/queue=orders/dt=2026-09-04/hour=*/*.jsonl.zst',
                    format = 'newline_delimited', compression = 'zstd',
                    hive_partitioning = true);

Two DuckDB specifics the lane measured. Uncompressed JSONL is compression = 'uncompressed', not 'none': DuckDB’s enum has no none and passing it raises a NotImplementedException rather than doing nothing. And read_json_auto infers ts as a TIMESTAMP with the trailing Z dropped, so it is not timezone-aware, and payload as a STRUCT, which the JSON operator ->> does not apply to. Naming the columns keeps a query portable between the two formats:

FROM read_json('…/*.jsonl.zst', format = 'newline_delimited', compression = 'zstd',
               hive_partitioning = true,
               columns = {partition: 'VARCHAR', "offset": 'BIGINT', transactionId: 'VARCHAR',
                          ts: 'VARCHAR', payload: 'JSON'})

Spark, where the Hive layout is the native one. Point it at the layout root rather than at a glob of the files: that is what makes partition discovery hand back queue, dt and hour as columns, for JSON exactly as for Parquet.

spark = SparkSession.builder.config("spark.sql.session.timeZone", "UTC").getOrCreate()
df = spark.read.parquet("s3a://my-lake/queen/")
df.where("queue = 'orders' AND dt = '2026-09-04'").orderBy("partition", "offset")

The session timezone is not a detail. ts is Queen’s clock in UTC, and Spark’s to_timestamp on the JSONL text reads it in the session zone, so on a machine in Europe/Rome the same object comes back two hours out. Two more things the lane measured: Spark infers ts from JSON as a string (it does not infer timestamps from JSON unless asked) and payload as a nested struct, so a query written against the Parquet payload string does not run unchanged over the JSONL objects.

Polars, PyArrow and pandas, the three a notebook has installed. For Parquet the plain call over the prefix works and gives the path keys as columns, dictionary<string> in PyArrow and category in pandas:

pq.read_table("queen/")                         # pyarrow, keys discovered
pd.read_parquet("queen/")                       # pandas, same reader underneath
pl.read_parquet("queen/")                       # polars, over the ROOT: a glob discovers nothing

For JSONL none of the three discovers a partition key over a glob, so read one queue’s prefix at a time and the queue is the directory asked for. PyArrow is the exception: its dataset reader does JSON too, pads.dataset(prefix, format="json", partitioning="hive"), which gives the keys back, though its compression detection stops at gzip and a .jsonl.zst still wants a CompressedInputStream per file.

Trino and Athena, with partition projection, so no crawler and no MSCK REPAIR ever runs. One table per queue, pointed at that queue’s prefix, which is the natural modelling and also the only one a Hive table allows: a table may not have a column and a partition key of the same name.

CREATE EXTERNAL TABLE queen_orders (
  `partition`    string,
  `offset`       bigint,
  transaction_id string,
  ts             timestamp,
  payload        string
)
PARTITIONED BY (dt string, hour string)
STORED AS PARQUET
LOCATION 's3://my-lake/queen/queue=orders/'
TBLPROPERTIES (
  'projection.enabled'      = 'true',
  'projection.dt.type'      = 'date',
  'projection.dt.format'    = 'yyyy-MM-dd',
  'projection.dt.range'     = '2026-01-01,NOW',
  'projection.dt.interval'  = '1',
  'projection.dt.interval.unit' = 'DAYS',
  'projection.hour.type'    = 'integer',
  'projection.hour.range'   = '0,23',
  'projection.hour.digits'  = '2'
);

Projection is what makes an hourly layout cheap at a year’s scale: the engine computes the partitions it needs from the predicate instead of listing the bucket to discover them. Two things the lane learned from a Trino 483 container before giving up on measuring it: partition and offset are reserved words and have to be quoted everywhere, and the Hive connector’s hive.timestamp-precision defaults to MILLISECONDS and then refuses a timestamp(6) column, so the catalog needs hive.timestamp-precision=MICROSECONDS or the table has to declare timestamp(3) and drop three digits of Queen’s clock.

ClickHouse, over the JSONL objects. Pass the structure: left to itself, schema inference types payload as a Tuple of the keys it sampled, which is not a JSON document any more, and payload IS NULL is then false for every row, so a null payload silently becomes an empty tuple instead of a NULL.

SELECT partition, `offset`, ts, JSONExtractString(payload, 'type') AS type
FROM s3('https://s3.eu-central-1.amazonaws.com/my-lake/queen/queue=orders/dt=2026-09-04/hour=*/*.jsonl.zst',
        'JSONEachRow',
        'partition String, `offset` Int64, transactionId String, ts String, payload Nullable(String)');

Recent ClickHouse turns hive partitioning on by itself, so queue, dt and hour arrive as columns with no argument at all, even alongside an explicit structure, and WHERE hour = 10 prunes on them. Over the Parquet objects ts arrives as DateTime64(6, 'UTC'): the TIMESTAMP(MICROS,UTC) annotation survives.

Snowflake, through an external stage, where the format is the whole configuration:

COPY INTO queen_orders
FROM @my_lake/queen/queue=orders/
FILE_FORMAT = (TYPE = PARQUET);

-- or, for the JSONL objects
COPY INTO queen_orders_raw
FROM @my_lake/queen/queue=orders/
FILE_FORMAT = (TYPE = JSON COMPRESSION = ZSTD);

Two habits pay for themselves whichever engine reads the lake. Order by (partition, offset) and never by ts alone, because every record of one segment shares a timestamp and only the offset is unique. And prune on dt and hour before anything else: they are exact, they cost nothing, and they are the difference between reading one hour and reading a year.