Min Yang created SPARK-60359:
--------------------------------
Summary: EvictionIterator removes state rows in hasNext instead of
next
Key: SPARK-60359
URL: https://issues.apache.org/jira/browse/SPARK-60359
Project: Spark
Issue Type: Bug
Components: Structured Streaming
Affects Versions: 5.0.0
Reporter: Min Yang
EvictionIterator (used by StreamingAggregationStateManager.evictionIterator and
StatefulStreamlineAggregateExec) removes an evictable row from the state store,
and increments numRowsRemovedSoFar, inside hasNext. A caller that calls hasNext
without a matching next() therefore deletes a row it never received.
Consequences:
- Append mode emits evicted rows and counts numRowsRemovedSoFar as
numOutputRows. If the consumer stops after a hasNext probe, the probed row is
deleted from state but never emitted, and is still counted as output.
- Update mode incremental cleanup has to re-check its removal budget before
every hasNext to avoid removing one extra row per input.
The iterator should only find the next evictable row in hasNext, and remove it
(and update the metric) when next() returns it.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
