Jonathan Gao created SPARK-58979:
------------------------------------
Summary: ColumnPruning disables late-event filtering when event
time is dropped after dropDuplicates
Key: SPARK-58979
URL: https://issues.apache.org/jira/browse/SPARK-58979
Project: Spark
Issue Type: Bug
Components: Structured Streaming, Optimizer
Affects Versions: 5.0.0
Reporter: Jonathan Gao
Consider the following streaming queries with identical input and watermark
configuration.
PRUNE-1 projects the event-time column away after dropDuplicates:
{code:scala}
val input = MemoryStream[(String, Long)]
val result = input.toDF()
.selectExpr("_1 AS id", "CAST(_2 AS TIMESTAMP) AS ts")
.withWatermark("ts", "10 seconds")
.dropDuplicates("id")
.select("id")
testStream(result, Append)(
AddData(input, ("a", 1000000L)),
CheckNewAnswer("a"),
AddData(input, ("b", 1000L)),
CheckNewAnswer())
{code}
PRUNE-2 retains the event-time column above dropDuplicates:
{code:scala}
val input = MemoryStream[(String, Long)]
val result = input.toDF()
.selectExpr("_1 AS id", "CAST(_2 AS TIMESTAMP) AS ts")
.withWatermark("ts", "10 seconds")
.dropDuplicates("id")
.selectExpr("id", "CAST(ts AS LONG) AS tsl")
testStream(result, Append)(
AddData(input, ("a", 1000000L)),
CheckNewAnswer(("a", 1000000L)),
AddData(input, ("b", 1000L)),
CheckNewAnswer())
{code}
PRUNE-2 correctly drops the late row. PRUNE-1 currently emits id "b". The final
projection therefore changes the late-event behavior of the same streaming
deduplication operation.
The root cause is that Deduplicate.references contains only the deduplication
keys. ColumnPruning consequently inserts a Project below Deduplicate that
removes the watermark-tagged event-time attribute. StreamingDeduplicateExec
searches its child output for that attribute to construct the late-event
predicate; after pruning, no predicate is created and the late row is emitted.
The proposed fix is to include watermark-tagged child attributes in references
for streaming Deduplicate, preventing ColumnPruning from removing an implicit
input dependency. This follows the established fix for
DeduplicateWithinWatermark in SPARK-50492. Batch Deduplicate continues to
reference only its keys.
Related: https://issues.apache.org/jira/browse/SPARK-50492
Proposed patch branch:
https://github.com/nahtonaj/spark/tree/fix/spark-dedup-watermark-pruning
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
