A stream-stream join is one of those operations that sounds straightforward until you actually try to run it in production. Two event streams arrive, you want to correlate events that share a key, and you express that as a SQL JOIN. The concept is obvious. The implementation surface is enormous.
We have built and debugged several stream join pipelines across different use cases while developing Feldera, and the problems we encountered cluster into a few categories. This post covers the ones that were not obvious from the documentation and that required real operational experience to understand.
State Unboundedness Is the Fundamental Problem
In a batch join, both sides are bounded. The engine reads the left table, builds a hash map, probes it with the right table, and produces a result. Memory usage is proportional to the smaller table. When the query finishes, the memory is released.
In a stream join, both sides are unbounded. The engine must buffer events from each side while waiting for a matching event from the other side. If events from the left stream arrive but their matching events from the right stream have not arrived yet, the left events must be held in state. If events from the right stream arrive for keys that the left stream has not yet emitted events for, those events must also be held.
Without bounds, state grows indefinitely. A stream join with no time constraint is equivalent to a full outer join on an ever-growing table. Memory usage grows linearly with the number of distinct join keys and the arrival rate differential between the two sides.
The practical implication: every stream-stream join needs an explicit time bound or a state expiry policy. If you do not set one, you will either run out of memory or build up gigabytes of stale join state that will never be used.
The Time Bound Trade-Off
Setting a time bound on a stream join means: match events from the left and right streams only if their event times are within some window of each other. Events outside the window are dropped from the join consideration.
In Feldera, you express this as a predicate on the join condition:
-- Join click events with purchase events within a 30-minute session window
CREATE MATERIALIZED VIEW click_to_purchase AS
SELECT
c.user_id,
c.item_id,
c.click_ts,
p.purchase_ts,
p.purchase_ts - c.click_ts AS conversion_latency
FROM clicks c
JOIN purchases p
ON c.user_id = p.user_id
AND c.item_id = p.item_id
AND p.purchase_ts BETWEEN c.click_ts AND c.click_ts + INTERVAL '30 minutes';The time bound determines what you are willing to miss. A 30-minute bound means: any purchase that happens more than 30 minutes after the click will not be matched. If your real conversion pattern has a long tail (users who click and then purchase hours or days later), a tight bound will miss those conversions.
The opposite is also true: a very wide bound (say, 7 days) means the engine must hold 7 days of click state waiting for potential matching purchases. For high-volume click streams, this can mean hundreds of gigabytes of state. The right bound for your use case is a function of the actual time distribution of your matching pairs, not a guess.
We recommend spending time up front with a batch analysis of your historical data to compute the actual distribution of time deltas between matching events. Plotting the CDF of match latency tells you what bound you need to capture 90%, 95%, 99% of matches. Setting your window to the p99 instead of the p99.9 often reduces state by an order of magnitude with minimal impact on match quality.
Late Arrivals and the Watermark Problem
Time bounds help with state size, but they introduce a different problem: what do you do when events arrive out of order? A purchase event with a timestamp of 14:15 might arrive in the Kafka topic at 14:22 if there was a processing delay upstream. If your watermark for the click stream has already advanced past 14:15 when this purchase arrives, the system needs to decide whether to attempt a late match or drop the event.
The answer depends on your correctness requirements. For feature engineering use cases, the most common approach is to allow late arrivals within a configurable tolerance window (say, 2 minutes) and drop events that arrive later than that. The tolerance window sets how long the engine holds state for potential late matches before considering a time period closed.
A tighter late-arrival tolerance reduces state and decreases the lag between event occurrence and feature update, but it increases the fraction of late events that are dropped and never contribute to features. A looser tolerance handles more late events but increases the state footprint and feature update latency.
This is a calibration exercise, not a configuration choice you can make a priori. Measure the actual late arrival distribution in your pipeline before setting the tolerance. In our experience, late arrival distributions are highly bimodal: most events arrive within a few seconds, and the tail consists of a small fraction of events that are minutes or hours late due to mobile buffering, batch log collection, or upstream retry logic. Often the right answer is to set a tight tolerance for the common case and handle the late-arriving edge cases in a separate reconciliation pipeline rather than trying to accommodate them in the real-time join.
Join Cardinality and State Explosion
A join that seems bounded from the time window perspective can still produce unbounded state if the join key has high cardinality and many events accumulate per key within the window.
Consider joining a user click stream with a user session stream on user_id. If a single user can generate thousands of click events and hundreds of session events within a 30-minute window, the join state for that user is the cross-product of those events. For most users this is fine. But if your data includes bot-like users or high-frequency API clients, a small number of users can dominate your join state.
The practical mitigation is to add a secondary filter or limit before the join to cap the number of events per key within the window. Alternatively, pre-aggregate one side of the join before joining, so the join is not row-to-row but aggregate-to-stream.
Exactly-Once vs At-Least-Once Output Semantics
Stream joins produce output records when matching pairs are found. In an at-least-once delivery scenario, a single matching pair might produce duplicate output records if the engine processes the same input event twice due to a restart or retry.
For feature aggregation downstream, duplicates in the join output can corrupt running aggregates. A duplicate "click-to-purchase match" row will inflate conversion counts and revenue sums. The standard mitigation is deduplication in the downstream consumer using the join keys and event timestamps as a composite deduplication key.
Feldera's incremental operators are designed to handle the retraction model correctly: a duplicate event is treated as a retraction of the first occurrence followed by a re-addition, which is idempotent for additive aggregates. But this requires that the input event stream carries the semantics of being a change log rather than a pure append log. If your Kafka topic is an append-only log without retraction events, duplicate detection and deduplication must happen before the join operator.
What Stream Joins Are Not a Good Fit For
Stream joins work well when you have two correlated event types with a defined time relationship and a bounded time window. They are the wrong approach when the relationship is not time-bounded, when one side of the join is a slowly-changing reference (use a CDC lookup join instead), or when the join logic requires accessing historical state beyond the current window (use a feature store lookup instead of a join).
A stream-stream join is also not the right tool for detecting absence: "users who clicked but did not purchase within 30 minutes" is an anti-join, which requires holding state for clicks that have not yet matched and emitting a record when the window closes without a match. This pattern is possible but significantly more complex to operate than a standard equi-join. If your use case is primarily about detecting absence or non-events, a different approach using time-based state expiry and explicit absence signals is usually simpler.