You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by vi...@apache.org on 2026/10/13 15:34:39 UTC
(spark) branch branch-4.0 updated: [SPARK-60211][4.1][PYTHON] Fix applyInPandasWithState failure with PyArrow 26 due to float slice offsets
This is an automated email from the ASF dual-hosted git repository.
viirya pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new 190d3110679b [SPARK-60211][4.1][PYTHON] Fix applyInPandasWithState failure with PyArrow 26 due to float slice offsets
190d3110679b is described below
commit 190d3110679b33d06355bef0f4190bf193b8d849
Author: Liang-Chi Hsieh <vi...@gmail.com>
AuthorDate: Tue Oct 13 08:34:13 2026 -0700
[SPARK-60211][4.1][PYTHON] Fix applyInPandasWithState failure with PyArrow 26 due to float slice offsets
### What changes were proposed in this pull request?
In `ApplyInPandasWithStateSerializer.load_stream`, explicitly convert `startOffset` / `numRows` read from the state struct to `int` before passing them to `RecordBatch.slice`.
### Why are the changes needed?
PyArrow 26.0.0 added `operator.index(...)` validation to `offset` / `length` in `RecordBatch.slice` (and `Table.slice` / `ChunkedArray.slice`). Before that, a float was silently truncated by the Cython `int64_t` conversion.
On this branch the state struct column is converted to pandas as a Series of dicts. When an Arrow batch contains padding state rows (`ApplyInPandasWithStateWriter.EMPTY_STATE_METADATA_ROW`, written whenever a batch has more data rows than state rows, i.e. a group with more than one row, or chunking), the struct has nulls, so its int32 fields are promoted to float64 and `startOffset` / `numRows` become Python floats. With PyArrow 26 this fails with:
```
File ".../python/pyspark/sql/pandas/serializers.py", line 1596, in gen_data_and_state
data_batch_for_group = data_batch.slice(data_start_offset, num_data_rows)
...
TypeError: 'float' object cannot be interpreted as an integer
```
This breaks `applyInPandasWithState` for users on PyArrow 26, and `FlatMapGroupsInPandasWithStateSuite` (5 tests) in the "sql - slow tests" job of the branch-4.0 / branch-4.1 scheduled builds since PyArrow 26.0.0 was released (that job installs an unpinned `pyarrow`; the last green branch-4.0 run had 25.0.1, the first failing one 26.0.0, with identical numpy / pandas).
branch-4.2 and later are not affected because SPARK-54962 converts nullable integers with `integer_object_nulls=True`, so the values stay ints there.
### Does this PR introduce _any_ user-facing change?
Yes. `applyInPandasWithState` works again with PyArrow 26. No change with earlier PyArrow versions.
### How was this patch tested?
Existing tests: `FlatMapGroupsInPandasWithStateSuite` covers this path and fails with PyArrow 26 in CI.
Locally:
- `build/sbt 'sql/testOnly *FlatMapGroupsInPandasWithStateSuite'` passes on JDK 17.
- Fed an Arrow IPC stream to `ApplyInPandasWithStateSerializer.load_stream` with and without a padding state row and recorded the argument types passed to `slice`: `float` before this change when padding is present, `int` after. `operator.index` (what PyArrow 26 calls) raises the exact error above on the former.
The change applies cleanly to branch-4.0 and branch-3.5, which have the same code.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (claude-opus-5-5)
Closes #59423 from viirya/SPARK-pyarrow26-applyinpandaswithstate.
Authored-by: Liang-Chi Hsieh <vi...@gmail.com>
Signed-off-by: Liang-Chi Hsieh <vi...@gmail.com>
(cherry picked from commit 0b46829a10a20943b8889a00ea4cadfaed240939)
Signed-off-by: Liang-Chi Hsieh <vi...@gmail.com>
---
python/pyspark/sql/pandas/serializers.py | 7 +++++--
1 file changed, 5 insertions(+), 2 deletions(-)
diff --git a/python/pyspark/sql/pandas/serializers.py b/python/pyspark/sql/pandas/serializers.py
index f47fe6ed754a..91d7cbef58f0 100644
--- a/python/pyspark/sql/pandas/serializers.py
+++ b/python/pyspark/sql/pandas/serializers.py
@@ -941,8 +941,11 @@ class ApplyInPandasWithStateSerializer(ArrowStreamPandasUDFSerializer):
# no more data with grouping key + state
break
- data_start_offset = state_info_col["startOffset"]
- num_data_rows = state_info_col["numRows"]
+ # The int fields of the state struct are converted to float when the batch
+ # contains padding (null) state rows. PyArrow >= 26 rejects float offsets
+ # in slice(), so convert them back to int.
+ data_start_offset = int(state_info_col["startOffset"])
+ num_data_rows = int(state_info_col["numRows"])
is_last_chunk = state_info_col["isLastChunk"]
if state_for_current_group:
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org