[
https://issues.apache.org/jira/browse/SPARK-60609?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jungtaek Lim resolved SPARK-60609.
----------------------------------
Fix Version/s: 4.2.2
4.1.5
4.3.1
4.4.0
Resolution: Fixed
Issue resolved by pull request 59855
[https://github.com/apache/spark/pull/59855]
> Handle list returned by SimpleDataSourceStreamReader.read() in offset
> validation for backward compatibility
> -----------------------------------------------------------------------------------------------------------
>
> Key: SPARK-60609
> URL: https://issues.apache.org/jira/browse/SPARK-60609
> Project: Spark
> Issue Type: Bug
> Components: PySpark, Structured Streaming
> Affects Versions: 4.2.0, 4.1.1, 4.3.0, 5.0.0, 4.1.3, 4.2.1, 4.1.4, 4.2.2,
> 4.3.1
> Reporter: Vinod KC
> Assignee: Vinod KC
> Priority: Major
> Labels: pull-request-available
> Fix For: 4.2.2, 4.1.5, 4.3.1, 4.4.0
>
> Attachments: list_return_e2e.py
>
>
> SimpleDataSourceStreamReader.read() is documented to return an iterator of
> tuples together with the end offset. However, returning a list instead of an
> iterator has worked in practice: before 4.2.0, _SimpleStreamReaderWrapper
> only cached the result and later consumed it via copy.copy() and
> itertools.chain, both of which accept any iterable. Readers returning lists
> (e.g. ([], start) when there is no new data) therefore exist in the wild.
> SPARK-55416 (released in 4.2.0) added validation in
> _SimpleStreamReaderWrapper.add_result_to_cache() that calls next() on the
> returned batch when the end offset equals the start offset. Since next() only
> accepts iterators, such readers now fail with:
> {code}
> TypeError: 'list' object is not an iterator
> {code}
> Impact:
> * A reader that returns an empty list with end == start to signal "no new
> data" - accepted before 4.2.0, and equivalent to the empty-iterator case
> SPARK-55416 explicitly allows - now fails the streaming query on the first
> trigger without new data.
> * A reader that returns a non-empty list with end == start gets an unclear
> TypeError instead of the intended SIMPLE_STREAM_READER_OFFSET_DID_NOT_ADVANCE
> error.
> Reproduction:
> {code:python}
> from pyspark.sql.datasource import SimpleDataSourceStreamReader
> from pyspark.sql.datasource_internal import _SimpleStreamReaderWrapper
> from pyspark.sql.streaming.datasource import ReadAllAvailable
> class ListReader(SimpleDataSourceStreamReader):
> def initialOffset(self):
> return {"offset": 0}
> def read(self, start):
> return [], start # no new data
> _SimpleStreamReaderWrapper(ListReader()).latestOffset({"offset": 0},
> ReadAllAvailable())
> # TypeError: 'list' object is not an iterator
> {code}
> Proposed fix: keep the iterator contract unchanged, but for backward
> compatibility convert a list result to an iterator with iter() before the
> validation calls next(), so existing list-returning readers keep working as
> they did before 4.2.0.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
