[
https://issues.apache.org/jira/browse/SPARK-58904?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jungtaek Lim updated SPARK-58904:
---------------------------------
Description:
h3. 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 the following 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.
h3. 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.
{code:java}
test("left outer range join with only right watermark evicts 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)),
// Fails because the query produces no output.
CheckNewAnswer(Row(1, null, 5, null))
)
}
{code}
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.
h3. 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:
{noformat}
Row(1, null, 5, null)
{noformat}
h3. 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.
h3. Notes
[PR #58238|https://github.com/apache/spark/pull/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:
h3. 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 the following 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.
h3. 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.
{code}
test("left outer range join with only right watermark evicts 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)),
// Fails because the query produces no output.
CheckNewAnswer(Row(1, null, 5, null))
)
}
{code}
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.
h3. 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:
{noformat}
Row(1, null, 5, null)
{noformat}
h3. 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.
h3. Notes
[PR #58238|https://github.com/apache/spark/pull/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.
> 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
>
> h3. 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 the following 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.
> h3. 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.
> {code:java}
> test("left outer range join with only right watermark evicts 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)),
> // Fails because the query produces no output.
> CheckNewAnswer(Row(1, null, 5, null))
> )
> }
> {code}
> 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.
> h3. 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:
> {noformat}
> Row(1, null, 5, null)
> {noformat}
> h3. 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.
> h3. Notes
> [PR #58238|https://github.com/apache/spark/pull/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]
