[viirya]

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)

[viirya]

Note that this needs to backport to both branch-4.0 and branch-3.5.

[dongjoon-hyun]

+1, LGTM. Thank you, @viirya and @HyukjinKwon .

cc @uros-b as the release manager of Apache Spark 4.1.4.

[viirya]

Merge Summary:

• merged into branch-4.1 0b46829
• merged into branch-4.0 190d311
• merged into branch-3.5 2378a96

Posted by merge_spark_pr.py

[viirya]

Thanks @HyukjinKwon @dongjoon-hyun !

[viirya]

The branch-3.5 cherry-pick of this (2378a96) was reverted in d7355a7 (#59455): branch-3.5 is in an extended LTS that takes security fixes only (https://spark.apache.org/versioning-policy.html). It remains in branch-4.0 and branch-4.1.