[
https://issues.apache.org/jira/browse/SPARK-58904?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jungtaek Lim updated SPARK-58904:
---------------------------------
Summary: Execution fails to construct a state-value watermark predicate for
valid one-sided stream-stream range joins (was: Analyzer accepts stream-stream
range joins that cannot evict the required state)
> Execution fails to construct a state-value watermark predicate for valid
> one-sided stream-stream range 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]
