This is an automated email from the ASF dual-hosted git repository.
HeartSaVioR pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 64b89977087a [SPARK-60609][SS][PYTHON] Handle list returned by
SimpleDataSourceStreamReader.read() in offset validation for backward
compatibility
64b89977087a is described below
commit 64b89977087aaf01d898b640877a9a0c58de01c2
Author: Vinod KC <[email protected]>
AuthorDate: Thu Oct 29 16:24:09 2026 +0900
[SPARK-60609][SS][PYTHON] Handle list returned by
SimpleDataSourceStreamReader.read() in offset validation for backward
compatibility
### What changes were proposed in this pull request?
In `_SimpleStreamReaderWrapper.add_result_to_cache()`, when `read()`
returns `end == start`, convert a list result to an iterator with `iter()`
before calling `next()` on it, so that the offset validation added by
SPARK-55416 also works when `read()` returns a list rather than an iterator.
The documented contract of `read()` (an iterator of tuples) is unchanged;
the list handling is only for backward compatibility with readers that returned
a list, which worked before SPARK-55416 (4.1.1 and earlier).
### Why are the changes needed?
SPARK-55416 (#54237, released in 4.1.2 and 4.2.0) added a check in
`add_result_to_cache()` that calls `next(it)` when `read()` returns `end ==
start`. `next()` only accepts iterators, so a reader that returns a list now
fails with:
```
TypeError: 'list' object is not an iterator
```
Before SPARK-55416 (4.1.1 and earlier) a list return worked in every case:
the wrapper only cached the result and later consumed it via `copy.copy()` +
`itertools.chain`, both of which accept any iterable. So this is a regression
for existing readers, in particular:
- `read()` returning `([], start)` to signal "no new data" -- explicitly
allowed by SPARK-55416 for an empty iterator -- now fails the streaming query
with a `TypeError`.
- A buggy reader returning a non-empty list with `end == start` gets a
`TypeError` instead of the intended
`SIMPLE_STREAM_READER_OFFSET_DID_NOT_ADVANCE` error.
The `end != start` path is unaffected and the cached object is kept as
returned, matching the pre-SPARK-55416 behavior.
The regression affects 4.1.2, 4.1.3, 4.1.4, 4.2.0, 4.2.1 and 4.3.0 (RC3),
so this should also be backported to `branch-4.3`, `branch-4.2` and
`branch-4.1`.
### Does this PR introduce _any_ user-facing change?
Yes, it fixes a regression in 4.1.2 and 4.2.0. A
`SimpleDataSourceStreamReader.read()` that returns a list with `end == start`
no longer fails with `TypeError`: an empty list is accepted as "no new data",
and a non-empty list raises `SIMPLE_STREAM_READER_OFFSET_DID_NOT_ADVANCE`, the
same as for an iterator.
### How was this patch tested?
Added `test_simple_stream_reader_returns_list` in
`test_python_streaming_datasource.py`, covering a non-empty list with an
advanced end offset (cached and readable), an empty list with `end == start`
(allowed, not cached), and a non-empty list with `end == start` (raises
`SIMPLE_STREAM_READER_OFFSET_DID_NOT_ADVANCE`). Without the fix, the last two
cases fail with `TypeError: 'list' object is not an iterator`.
Also verified end-to-end with the released PyPI wheels and the 4.3.0 RC3
`pyspark-4.3.0.tar.gz` from dist.apache.org (JDK 17, `local[2]`, memory sink),
using a reader that returns 2 rows on some triggers and `([], start)` on the
others:
| Spark | `read()` returns iterator | `read()` returns list |
|---|---|---|
| 4.1.1 (before SPARK-55416) | OK, all 8 rows | OK, all 8 rows |
| 4.1.3 | OK, all 8 rows | query fails after 4 rows: `TypeError: 'list'
object is not an iterator` |
| 4.2.0 | OK, all 8 rows | query fails after 4 rows: `TypeError: 'list'
object is not an iterator` |
| 4.3.0 RC3 | OK, all 8 rows | query fails after 4 rows: `TypeError: 'list'
object is not an iterator` |
| 4.2.0 + this patch | OK, all 8 rows | OK, all 8 rows |
| 4.3.0 RC3 + this patch | OK, all 8 rows | OK, all 8 rows |
### Was this patch authored or co-authored using generative AI tooling?
Yes, Generated-by: Claude Code
Closes #59855 from vinodkc/simple-stream-reader-list.
Authored-by: Vinod KC <[email protected]>
Signed-off-by: Jungtaek Lim <[email protected]>
(cherry picked from commit cda7937a12542e0e9f412a8a00573626e84ad5e3)
Signed-off-by: Jungtaek Lim <[email protected]>
---
python/pyspark/sql/datasource_internal.py | 3 ++
.../sql/tests/test_python_streaming_datasource.py | 45 ++++++++++++++++++++++
2 files changed, 48 insertions(+)
diff --git a/python/pyspark/sql/datasource_internal.py
b/python/pyspark/sql/datasource_internal.py
index 932ea8c10194..06d98af1403a 100644
--- a/python/pyspark/sql/datasource_internal.py
+++ b/python/pyspark/sql/datasource_internal.py
@@ -105,6 +105,9 @@ class _SimpleStreamReaderWrapper(DataSourceStreamReader):
if end_str != start_str:
self.cache.append(PrefetchedCacheEntry(start, end, it))
return
+ # read() may return a list instead of an iterator.
+ if isinstance(it, list):
+ it = iter(it)
try:
next(it)
except StopIteration:
diff --git a/python/pyspark/sql/tests/test_python_streaming_datasource.py
b/python/pyspark/sql/tests/test_python_streaming_datasource.py
index 47f381e24930..6b6baf76eb73 100644
--- a/python/pyspark/sql/tests/test_python_streaming_datasource.py
+++ b/python/pyspark/sql/tests/test_python_streaming_datasource.py
@@ -559,6 +559,51 @@ class BasePythonStreamingDataSourceTestsMixin:
self.assertEqual(end, start)
self.assertEqual(len(wrapper.cache), 0)
+ def test_simple_stream_reader_returns_list(self):
+ """read() returning a list instead of an iterator is handled like an
iterator."""
+ from pyspark.sql.datasource_internal import _SimpleStreamReaderWrapper
+
+ class ListReader(SimpleDataSourceStreamReader):
+ def __init__(self, rows, advance):
+ self.rows = rows
+ self.advance = advance
+
+ def initialOffset(self):
+ return {"offset": 0}
+
+ def read(self, start: dict):
+ end = {"offset": start["offset"] + len(self.rows)} if
self.advance else start
+ return (list(self.rows), end)
+
+ def readBetweenOffsets(self, start: dict, end: dict):
+ return iter([])
+
+ def commit(self, end: dict):
+ pass
+
+ start = {"offset": 0}
+
+ # Non-empty list with an advanced end offset is cached and readable.
+ wrapper = _SimpleStreamReaderWrapper(ListReader([(0,), (1,)],
advance=True))
+ end = wrapper.latestOffset(start, ReadAllAvailable())
+ self.assertEqual(end, {"offset": 2})
+ self.assertEqual(list(wrapper.getCache(start, end)), [(0,), (1,)])
+
+ # Empty list with end == start is allowed and not cached.
+ wrapper = _SimpleStreamReaderWrapper(ListReader([], advance=False))
+ end = wrapper.latestOffset(start, ReadAllAvailable())
+ self.assertEqual(end, start)
+ self.assertEqual(len(wrapper.cache), 0)
+
+ # Non-empty list with end == start raises the offset validation error.
+ wrapper = _SimpleStreamReaderWrapper(ListReader([(0,)], advance=False))
+ with self.assertRaises(PySparkException) as cm:
+ wrapper.latestOffset(start, ReadAllAvailable())
+ self.assertEqual(
+ cm.exception.getCondition(),
+ "SIMPLE_STREAM_READER_OFFSET_DID_NOT_ADVANCE",
+ )
+
def test_stream_writer(self):
input_dir =
tempfile.TemporaryDirectory(prefix="test_data_stream_write_input")
output_dir =
tempfile.TemporaryDirectory(prefix="test_data_stream_write_output")
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
