[nhobin219]

Closes #159.

A stream can declare its identity columns, Stream.new(..., identity=("session_id", "line_number")). The writer then fills a derived _streamcast_id on every row, so a dedupe, a join or a lookup by the whole key reads one 16-byte column. This follows the design settled on the issue (2026-10-18 and 2026-10-20).

What it does

• The key (src/streamcast/_identity.py, the one implementation that the writer, ingest and readers all call): xxh3_128 over the identity columns' values only, in declaration order, with no domain tag.
  • booleans are 1 byte;
  • int32 and int64 are their little-endian two's complement;
  • a bytesN binary is its N bytes;
  • strings and binaries are a 4-byte little-endian length, then the bytes (UTF-8 for a string).
  • The column stores the canonical (big-endian) 16-byte digest, as fixed_size_binary(16).
  • Identity columns must be booleans, integers, strings or binaries, required and non-null. A float is refused (equality is not identity), and so are nested types and nulls (no encoding).
• The writer fills it on send, send_many and publisher frames (Stream._stored, inside _commit, which has no await, so I1 is unchanged) and on Stream.ingest (_ingest.Source, a column at a time). A row or a source that carries _streamcast_id is refused.
• It is never on the wire: not in frames, not in the greeting's schema, not in catch-up rows (_log.columns and _log.declared leave it out, through _log.owned).
• The declaration lives in two places:
  • on the log's schema, as field metadata on _streamcast_id (litelink keeps field metadata through new and open);
  • in each log's metadata entry, as identity, because an Iceberg schema keeps no field metadata, so readers and restore learn it there.
  • The key is written only when set, so a stream without an identity writes byte-identical metadata.
• Hidden from *, selectable by name:
  • Snapshot.scan, Stream.scan and Live.scan project * EXCLUDE ("_streamcast_id").
  • sql builds its view log with the column only when the query names it. A view has no hidden columns, so this is decided per query, with DuckDB's tokenizer: it counts the name quoted or not, in any case, and not inside a string or a comment. A query that names it also sees it in its own *.
  • A raw iceberg_scan outside streamcast shows it. SPEC says so.
• The WHERE rewrite (_snapshot._keyed): when filters= (on scan, sql, and live's where=) fix every identity column with ==, the reader adds "_streamcast_id" = '<key>'::BLOB, computed from the literals and naming the column explicitly.
  • It is added beside the natural terms, not in place of them. This is the one place I departed from "turns X into Y". The natural columns carry the statistics: the manifest prunes whole logs on them, and DuckDB prunes files and row groups on the Iceberg bounds, while a hash's bounds rule nothing out. Dropping the natural terms would read every file, which contradicts "the natural columns' statistics still prune files".
  • Every row matching the natural terms has exactly that key, so the term changes no answer. A literal the key can't encode (e.g. a string for an int column) adds no term.
  • A where= SQL string is not rewritten, for the same reason SQL is not mined for pruning terms (Prune retired logs from where= and the SQL WHERE clause, soundly #57).
• Broker tail rows (Live, as_of_offset=LATEST) are keyed by the reader exactly as the writer keyed them (_tail_table).
• Fixed for the stream's life:
  • Stream.new on an existing log refuses an identity= other than the log's, in either direction.
  • Stream.migrate(..., identity=) refuses a changed identity (added, dropped or reordered), a schema that removes an identity column, and one that retypes one.
  • restore and revive carry the identity over from the metadata.
  • Whether migrate may add an identity to an existing stream is left open (SPEC says so).
• Not a primary key: duplicates are appended, and a test asserts it.
• SPEC gets "A declared identity" and "Library columns: what is visible, and what is plumbing" in §5, which states the naming rule as decided on the issue. It also describes the metadata file's identity key and the snapshot behaviour. API.md and the CHANGELOG are updated.
• xxhash>=3.6.0 becomes a direct dependency. It was already installed through litelink, at the same floor.

Breaking? Version?

New API, so a minor release, 0.24.0 (not cut, as asked). It is breaking in one narrow way, stated in the CHANGELOG: _streamcast_id is now a reserved name. A schema declaring it is refused, and so is reopening a log that an earlier release created with an application column of that name (rename it with migrate first). The critical reviewer found this; I reproduced it and chose to document it rather than make column ownership vary per log.

Mixed versions, measured against 0.23.0 from PyPI in its own virtualenv, on a stream this branch keyed:

• 0.23.0's Stream.new refuses to open the log ("has columns [..., '_streamcast_id'], and this stream declares ..."), so an old server fails closed.
• 0.23.0's readers show _streamcast_id in SELECT *, and its catch-up rows carry the key, which live frames don't.
• So upgrade a stream's readers before declaring an identity on it. Streams without an identity are unchanged for every version.
• I did not add 0.23.0 to tests/test_interop.py; the matrix there covers streams without an identity, which this doesn't change.

No consumer pin needs to move until someone wants the feature. Per the shared rules, this should not be released until a consumer pull request uses it.

Cost, measured

Same session, same 2-vCPU box. The box was under load (load average 1.3–2.6), so I report the minimum over alternating rounds; the medians were noise (plain was slower than keyed by median).

	without identity	with identity
send_many, 1000-row batches, 40 alternating rounds	12.74 µs/row	15.66 µs/row
ingest, 200k rows, 5 rounds	632 ns/row	1707 ns/row
Identity.key() alone		1.36 µs/row

The ingest-side keys() was first written with a dict per row, at 5085 ns/row; the column-wise version measures 1150 ns/row. A stream that declares no identity takes the old path.

Tests: tests/test_identity.py, 34 tests, ~11 s

The ones the step asked for:

• Golden vectors for each encoding, written out byte by byte, with their keys:
  • ('ab','c') → ef2fcd6c…, against ('a','bc') → 6e496efc…;
  • (string, int64), (int32, bool), (bytes4, binary), UTF-8 byte length, and that declaration order matters.
  • The digest's byte order is anchored on xxHash's published XXH3-128 of the empty input (99aa06d3…).
• A join across two streams on _streamcast_id (test_two_streams_join_on_the_key).
• The rewrite returns the same rows as the natural predicate (test_a_key_filter_returns_the_rows_the_natural_predicate_does). Two more prove the key term is really applied:
  • a row written past the writer with a forged key matches the natural predicate and not the keyed filter;
  • the key term alone returns the same rows.
• SELECT * leaves it out, SELECT _streamcast_id returns it, through sql and scan, on published rows, the broker tail, and a Live view.
• migrate refuses an identity change: removing an identity column, retyping one, and changing or dropping the declaration. A migration that keeps the identity keys the next log alike.

Plus: refused declarations (float, nullable, nested, unknown column, the reserved name); a row or an ingest source supplying the key; reopening with another identity or gaining one; ingest keying as send does (including a float64 source cast to int64); frames and greeting not carrying the key.

Falsified: 25 mutations, each run against the tests that should catch it, then restored. The first pass missed two, and I fixed the tests for both:

• "no rewrite" (_select not adding the key term): the answer is identical by design, so I added the forged-key test.
• "batch not cast" in ingest: an int32 source gives the same Python ints, so the test now uses a float64 source.

All 25 are caught now. The mutations include: no length prefix, a reversed digest, big-endian ints, ints coerced from bool/float/str, * not hidden, the name never detected, the rewrite on the wrong column order, the tail unkeyed, the tail key left in catch-up rows, the writer storing a constant key, a row allowed to supply the key, the open ignoring identity, each migrate refusal disabled, migrate dropping the identity, metadata or shape() forgetting it, ingest unkeyed, a source allowed to supply the key, and the wire carrying the key.

Critical review

It ran once, as a general-purpose subagent with the repository's rubric. Its bars: a reader silently given a short or wrong answer by the key filter; one row getting two keys depending on how it entered; a stream without an identity behaving differently or failing to open after the upgrade; sql binder errors from the hiding.

• No critical finding. It attacked non-ASCII and quote strings, int32 bounds, bool, wrong-type literals, duplicate == terms, and send vs ingest vs tail keys.
• Plausible, fixed by documenting: a pre-existing application column named _streamcast_id no longer opens (above).
• Minor, fixed: a quoted "_STREAMCAST_ID" was not recognised (DuckDB resolves quoted identifiers without case), which caused a binder error. It now is, with a test.
• Outside this change: it also noted that _term_sql on main already mis-renders bytes values and NUL-containing strings in filters=. Keying makes neither worse. I have not filed it.

Gates

The full suite, run with every tier required (STREAMCAST_REQUIRE_S3=1 STREAMCAST_REQUIRE_NODE=1 STREAMCAST_REQUIRE_INTEROP=1 STREAMCAST_REQUIRE_POSTGRES=1 uv run pytest, as check-all sets them; rustfs up). The log was read, not re-run:

FAILED tests/test_snapshot_readers.py::TestAReader::test_the_event_loop_runs_while_it_reads
ERROR tests/test_node.py::test_a_browser_websocket_reads_and_writes_a_stream
== 1 failed, 989 passed, 1 skipped, 6 warnings, 1 error in 1013.51s (0:16:53) ==

• The error is the environment: Failed: node is not installed, and STREAMCAST_REQUIRE_NODE is set. This box has no Node, so the browser-client tier did not run here; CI runs it.
• The failure is a timing assertion under load: AssertionError: 223 ticks in 4.98s (it needs 249). The load average was 4.7 on 2 vCPUs. Run alone afterwards it passed 3 of 3 (1.7–4.7 s). I'm not counting that as a pass of the full run; CI's run is the one that says.
• The skip is structural: tests/test_store.py:129: a SQLite file has no connection to lose.
• The S3, interop and Postgres tiers ran, because they were required.

CI on this PR passed in full (run 39506588699): Tests (3.11) and (3.14), the type check, lint and format, the wheel job, and CI success.

• ruff check, ruff format --check and the blank-line check pass.
• ty check reports 2 possibly-missing-submodule warnings in tests/test_limits.py, a file this branch doesn't touch. They are on main too and fail just typecheck (and so just check-all) locally. CI's type check passed on the last PRs, so its ty version presumably differs.

Note on other open PRs

This branches from main (26e903d). #177 (startup adopts or refuses a published stream) also edits _open_or_create in src/streamcast/_stream.py, so whichever merges second needs a small rebase.

🤖 Generated with Claude Code

[nhobin219]

It is added beside the natural terms, not in place of them. This is the one place I departed from "turns X into Y". The natural columns carry the statistics: the manifest prunes whole logs on them, and DuckDB prunes files and row groups on the Iceberg bounds, while a hash's bounds rule nothing out. Dropping the natural terms would read every file, which contradicts "the natural columns' statistics still prune files".

Did we actually measure the impact on query performance here? Like, is this even worth doing? The entire point of this change was not having to select the natural columns to filter the result (saves having to read those columns back fromobject storage). Is DuckDB's optimizer smart enough to not read the natural columns up front?