Vinod KC created SPARK-60609:
--------------------------------
Summary: 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.3.0, 4.3.1, 5.0.0
Reporter: Vinod KC
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]
