Streaming joins: bounded state, temporal semantics and skew

Streaming join превращает время и ordering в часть relational semantics. Без bound состояние растёт бесконечно; без versioned dimension «исторический» replay может соединиться с сегодняшним значением.

Корректная модель

• Stream-stream interval joins retain both inputs; buffering only one side is incorrect.
• Watermarks/TTL bound state at the cost of explicitly late/unmatched results.
• Temporal joins require versioned/as-of dimension semantics for reproducible replay.
• Hot-key mitigation must alter physical partition keys or isolate the workload; more generic workers do not split one key.

Границы и компоненты

Сценарии

Symmetric interval join

Order and payment for one key match when their event times satisfy the configured interval. Either side may arrive first, so both sides need retained state until watermarks make further matches too late.

Проверяемый исход: One stable joined result is emitted regardless of arrival order.

Late match after cleanup

A matching right event arrives after the left state was legally cleaned by watermark/lateness policy. It cannot be joined locally and goes to correction/audit policy.

Проверяемый исход: The system reports a late miss; it does not claim perfect completeness after bounded retention.

Temporal as-of enrichment

An event joins the dimension version effective at its event time, not whichever value is current during processing/replay.

Проверяемый исход: Historical replay remains reproducible when version history is retained.

Hot-key state skew

One join key owns a disproportionate number of records, so one partition/state shard dominates. A local cache or adding workers alone does not split that key.

Проверяемый исход: Only an explicit shard subkey plus semantically valid second-stage merge changes placement; otherwise isolate/cap the tenant.

Failure, concurrency и replay checklist

• Persist/checkpoint both join states and timers consistently.
• Use stable pair/result IDs so replay does not duplicate joined output.
• Bound dimension cache with version/TTL/invalidation and source of truth fallback.
• Monitor state bytes and rates per key/partition, not only total records.

Формулы, units и допущения

• Expected retained records per side ≈ arrival rate × join retention span, then adjust for key skew and overlapping windows.
• A naive many-to-many hot key with L left and R right records can emit up to L×R matches; cardinality controls are correctness/capacity requirements.
• Watermark cleanup reduces state but defines a completeness cutoff; it cannot be called lossless for arbitrarily late events.

Числа выше — учебные inputs или размерностные формулы. Их нельзя выдавать за benchmark или SLA конкретного продукта.

Связанные темы

[CONCEPT]windowing-watermarks

[CONCEPT]hot-key-mitigation

Первичные источники

• https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/table/sql/queries/joins/
• https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/datastream/operators/joining/
• https://www.vldb.org/pvldb/vol8/p1792-Akidau.pdf
• https://kafka.apache.org/41/streams/developer-guide/dsl-api.html#joining

Scope note

Диаграмма показывает причинные границы и recovery contracts, а не скрытую реализацию конкретного managed-сервиса. Любая stronger guarantee действует только в явно названной transaction/checkpoint/acknowledgement boundary.