[
https://issues.apache.org/jira/browse/SPARK-58904?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jungtaek Lim updated SPARK-58904:
---------------------------------
Description:
### Summary
Analyzer accepts stream-stream range joins for which execution cannot evict
the required state
### Description
For stream-stream joins with event-time range conditions, the analyzer may
accept a one-sided watermark because it can derive a state-value watermark for
the opposite input from the join condition.
Execution does not always implement that accepted plan.
getOneSideStateWatermarkPredicate only creates a state-value eviction predicate
when the state side itself contains an attribute marked with watermark
metadata. When only the opposite input is watermarked, no
predicate is created for the state that should be evicted.
This affects these directions:
Join type Watermarked input State that should be evicted Effect
━━━━━━━━━━━━━ ━━━━━━━━━━━━━━━━━━━ ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
Left outer Right Left
Unmatched left rows are never emitted
───────────── ─────────────────── ──────────────────────────────
─────────────────────────────────────────────
Left semi Right Left
Never-matched left rows remain indefinitely
───────────── ─────────────────── ──────────────────────────────
─────────────────────────────────────────────
Right outer Left Right
Unmatched right rows are never emitted
───────────── ─────────────────── ──────────────────────────────
─────────────────────────────────────────────
Full outer Right Left
Unmatched left rows are never emitted
───────────── ─────────────────── ──────────────────────────────
─────────────────────────────────────────────
Full outer Left Right
Unmatched right rows are never emitted
For full outer joins, one-sided watermarking is intentional: it should
finalize unmatched rows from the opposite side. A watermark on both inputs is
needed to finalize both directions and clean up both state stores. The bug is
that the range-condition
implementation does not honor the supported one-sided case.
The state-key watermark path, where event time or a window is part of the
equi-join keys, is not covered by this issue.
### Reproducer
The following test uses a right-side watermark and a range condition. Once
the right watermark passes the latest possible matching time for the left row,
the left outer join should emit that row with a null right side.
test("left outer range join with only right watermark does not evict left
state") {
val leftInput = MemoryStream[(Int, Int)]
val rightInput = MemoryStream[(Int, Int)]
val left = leftInput.toDF().toDF("leftKey", "time")
.select(
$"leftKey",
timestamp_seconds($"time").as("leftTime"))
val right = rightInput.toDF().toDF("rightKey", "time")
.select(
$"rightKey",
timestamp_seconds($"time").as("rightTime"))
.withWatermark("rightTime", "10 seconds")
val joined = left.join(
right,
expr(
"""
|leftKey = rightKey AND
|leftTime BETWEEN
| rightTime - INTERVAL 5 SECONDS AND
| rightTime + INTERVAL 5 SECONDS
""".stripMargin),
"left_outer")
.select(
$"leftKey",
$"rightKey",
$"leftTime".cast("long"),
$"rightTime".cast("long"))
testStream(joined)(
AddData(leftInput, (1, 5)),
CheckNewAnswer(),
// Advances the right watermark to 20.
AddData(rightInput, (2, 30)),
CheckNewAnswer(),
// Runs another batch using watermark 20. A future right row that
// could match leftTime=5 would have rightTime in [0, 10], so the
// left row can now be finalized.
AddData(rightInput, (2, 40)),
// Actual: no output; the left row remains in state.
// Expected: Row(1, null, 5, null)
CheckNewAnswer()
)
}
Changing the join type to full_outer reproduces the same missing
unmatched-left output. Swapping the inputs and using right_outer reproduces the
symmetric problem.
### Expected behavior
For an accepted range join, execution should construct the state-value
eviction predicate implied by the opposite input’s watermark and the range
condition. In the reproducer, the left row should be emitted as:
Row(1, null, 5, null)
### Actual behavior
The query passes analysis, but no left-state eviction predicate is created.
The left row remains in state and its unmatched outer result is never emitted.
For left semi joins, the equivalent failure causes never-matched left rows to
accumulate rather than causing missing output.
### Notes
PR #58238 attempted a broader analyzer restriction but was closed for further
design review. The fix should preserve the documented one-sided watermark
semantics or explicitly redefine them; analysis and execution must agree.
was:
Stream-stream left semi and left outer joins rely on the left-side state being
evicted by the watermark, but the analyzer does not require the watermark to be
placed where eviction can happen. `checkForStreamStreamJoinWatermark` accepts a
right-only (or mis-placed) watermark, so at runtime no left-eviction predicate
is built: left outer silently drops its unmatched output, and left semi state
grows without bound. Left outer has a second issue: an already-emitted
unmatched row can be invalidated by a late row unless both sides are
late-filtered on the eviction dimension.
The analyzer must ensure that accepted watermark placement can drive the state
eviction and late-row filtering required by the join type. Otherwise, left
outer joins can omit unmatched output, left semi state can grow without bound,
and late rows can invalidate previously emitted results. The fix should also
audit the corresponding requirements for right/full outer and future left anti
joins.
> Enforce correct watermark placement for stream-stream left semi and left
> outer joins
> ------------------------------------------------------------------------------------
>
> Key: SPARK-58904
> URL: https://issues.apache.org/jira/browse/SPARK-58904
> Project: Spark
> Issue Type: Bug
> Components: Structured Streaming
> Affects Versions: 4.3.0
> Reporter: Ganesha S
> Priority: Critical
> Labels: correctness
>
> ### Summary
> Analyzer accepts stream-stream range joins for which execution cannot evict
> the required state
> ### Description
> For stream-stream joins with event-time range conditions, the analyzer may
> accept a one-sided watermark because it can derive a state-value watermark
> for the opposite input from the join condition.
> Execution does not always implement that accepted plan.
> getOneSideStateWatermarkPredicate only creates a state-value eviction
> predicate when the state side itself contains an attribute marked with
> watermark metadata. When only the opposite input is watermarked, no
> predicate is created for the state that should be evicted.
> This affects these directions:
> Join type Watermarked input State that should be evicted Effect
> ━━━━━━━━━━━━━ ━━━━━━━━━━━━━━━━━━━ ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
> ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
> Left outer Right Left
> Unmatched left rows are never emitted
> ───────────── ─────────────────── ──────────────────────────────
> ─────────────────────────────────────────────
> Left semi Right Left
> Never-matched left rows remain indefinitely
> ───────────── ─────────────────── ──────────────────────────────
> ─────────────────────────────────────────────
> Right outer Left Right
> Unmatched right rows are never emitted
> ───────────── ─────────────────── ──────────────────────────────
> ─────────────────────────────────────────────
> Full outer Right Left
> Unmatched left rows are never emitted
> ───────────── ─────────────────── ──────────────────────────────
> ─────────────────────────────────────────────
> Full outer Left Right
> Unmatched right rows are never emitted
> For full outer joins, one-sided watermarking is intentional: it should
> finalize unmatched rows from the opposite side. A watermark on both inputs is
> needed to finalize both directions and clean up both state stores. The bug is
> that the range-condition
> implementation does not honor the supported one-sided case.
> The state-key watermark path, where event time or a window is part of the
> equi-join keys, is not covered by this issue.
> ### Reproducer
> The following test uses a right-side watermark and a range condition. Once
> the right watermark passes the latest possible matching time for the left
> row, the left outer join should emit that row with a null right side.
> test("left outer range join with only right watermark does not evict left
> state") {
> val leftInput = MemoryStream[(Int, Int)]
> val rightInput = MemoryStream[(Int, Int)]
> val left = leftInput.toDF().toDF("leftKey", "time")
> .select(
> $"leftKey",
> timestamp_seconds($"time").as("leftTime"))
> val right = rightInput.toDF().toDF("rightKey", "time")
> .select(
> $"rightKey",
> timestamp_seconds($"time").as("rightTime"))
> .withWatermark("rightTime", "10 seconds")
> val joined = left.join(
> right,
> expr(
> """
> |leftKey = rightKey AND
> |leftTime BETWEEN
> | rightTime - INTERVAL 5 SECONDS AND
> | rightTime + INTERVAL 5 SECONDS
> """.stripMargin),
> "left_outer")
> .select(
> $"leftKey",
> $"rightKey",
> $"leftTime".cast("long"),
> $"rightTime".cast("long"))
> testStream(joined)(
> AddData(leftInput, (1, 5)),
> CheckNewAnswer(),
> // Advances the right watermark to 20.
> AddData(rightInput, (2, 30)),
> CheckNewAnswer(),
> // Runs another batch using watermark 20. A future right row that
> // could match leftTime=5 would have rightTime in [0, 10], so the
> // left row can now be finalized.
> AddData(rightInput, (2, 40)),
> // Actual: no output; the left row remains in state.
> // Expected: Row(1, null, 5, null)
> CheckNewAnswer()
> )
> }
> Changing the join type to full_outer reproduces the same missing
> unmatched-left output. Swapping the inputs and using right_outer reproduces
> the symmetric problem.
> ### Expected behavior
> For an accepted range join, execution should construct the state-value
> eviction predicate implied by the opposite input’s watermark and the range
> condition. In the reproducer, the left row should be emitted as:
> Row(1, null, 5, null)
> ### Actual behavior
> The query passes analysis, but no left-state eviction predicate is created.
> The left row remains in state and its unmatched outer result is never emitted.
> For left semi joins, the equivalent failure causes never-matched left rows
> to accumulate rather than causing missing output.
> ### Notes
> PR #58238 attempted a broader analyzer restriction but was closed for
> further design review. The fix should preserve the documented one-sided
> watermark semantics or explicitly redefine them; analysis and execution must
> agree.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
