This is an automated email from the ASF dual-hosted git repository.
HyukjinKwon pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 26660a9f92eb [SPARK-57989][CONNECT][PYTHON] Raise NOT_IMPLEMENTED for
year-month interval in Connect collect, matching Classic's default
26660a9f92eb is described below
commit 26660a9f92ebadd601d87f2e34a71a9a1f34c64b
Author: Hyukjin Kwon <[email protected]>
AuthorDate: Tue Jul 7 16:49:26 2026 +0900
[SPARK-57989][CONNECT][PYTHON] Raise NOT_IMPLEMENTED for year-month
interval in Connect collect, matching Classic's default
### What changes were proposed in this pull request?
This PR makes the PySpark Spark Connect client raise a clean
`PySparkNotImplementedError` (error class `NOT_IMPLEMENTED`) when a result
containing a `YearMonthIntervalType` is collected, instead of surfacing an
opaque PyArrow error.
Concretely:
- `from_arrow_type` (in `pyspark/sql/pandas/types.py`) now maps the Arrow
`YEAR_MONTH` interval type to `YearMonthIntervalType`. The JVM serializes
Spark's `YearMonthIntervalType` to an Arrow `YEAR_MONTH` interval, but PyArrow
exposes no `is_*()` helper or factory for it (only `MONTH_DAY_NANO` is in
`pyarrow.types`), so the branch matches on the stable Arrow type id
(`Type::INTERVAL_MONTHS == 21`).
- `ArrowTableToRowsConversion.convert` (in `pyspark/sql/conversion.py`)
checks the result schema before materializing and raises `NOT_IMPLEMENTED` if
any field (including nested array/map/struct/UDT element types) is a
`YearMonthIntervalType`. PyArrow cannot materialize such a column:
`to_pylist()` raises an opaque `KeyError: 21` from `get_array_class_from_type`,
and that lookup fails even for an empty column, so the check is intentionally
unconditional in the row count.
- `_has_type` (in `pyspark/sql/types.py`) now recurses into
`UserDefinedType.sqlType()` so a year-month interval hidden inside a UDT is
detected too.
This matches the default behavior of the classic PySpark path, where
`YearMonthIntervalType.fromInternal` raises `NOT_IMPLEMENTED`.
### Why are the changes needed?
Collecting a year-month interval through the Spark Connect client
previously failed with confusing, low-level errors rather than the intended
`NOT_IMPLEMENTED`:
- `PySparkTypeError: [UNSUPPORTED_DATA_TYPE_FOR_ARROW_CONVERSION]
month_interval is not supported`, or
- `pyarrow.lib.ArrowNotImplementedError: No known equivalent Pandas block
for Arrow data of type month_interval`, or
- an opaque `KeyError: 21` from PyArrow when materializing rows.
Classic PySpark already raises a clear `NOT_IMPLEMENTED` error for this
unsupported operation (`YearMonthIntervalType.fromInternal`). Spark Connect
should behave the same so users get an actionable message and the two clients
stay consistent.
### Does this PR introduce _any_ user-facing change?
Yes, for the Spark Connect Python client. Collecting a year-month interval
value (`df.collect()`/`first()`/`take()`/`head()`) now raises
`PySparkNotImplementedError` with error class `NOT_IMPLEMENTED` instead of a
`PySparkTypeError`/`ArrowNotImplementedError`/`KeyError`. Collecting a
year-month interval was never supported; only the surfaced error changes.
Two behaviors remain specific to Spark Connect and differ from classic
PySpark:
- `PYSPARK_YM_INTERVAL_LEGACY=1` (which makes classic return the internal
integer months) is not honored; collect always raises.
- An empty result (e.g. `.limit(0).collect()`) raises rather than returning
`[]`, because PyArrow cannot build the `INTERVAL_MONTHS` array class regardless
of row count.
### How was this patch tested?
- Updated
`test_connect_error.SparkConnectErrorTests.test_ym_interval_in_collect` to
expect `PySparkNotImplementedError` and added coverage for a year-month
interval nested inside an array.
- Added
`test_connect_error.SparkConnectErrorTests.test_ym_interval_empty_collect`
covering the empty-result case.
- Updated the skip reason on `test_parity_types` for
`test_ym_interval_in_collect` to explain why the inherited classic contract
(which asserts the `PYSPARK_YM_INTERVAL_LEGACY=1` integer-months behavior)
cannot be satisfied by Spark Connect.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
Closes #57068 from HyukjinKwon/connect-ym-interval-collect.
Authored-by: Hyukjin Kwon <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>
---
python/pyspark/sql/conversion.py | 21 +++++++++++++++++++-
python/pyspark/sql/pandas/types.py | 8 ++++++++
.../sql/tests/connect/test_connect_error.py | 23 +++++++++++++++++++---
.../pyspark/sql/tests/connect/test_parity_types.py | 8 +++++++-
python/pyspark/sql/types.py | 2 ++
5 files changed, 57 insertions(+), 5 deletions(-)
diff --git a/python/pyspark/sql/conversion.py b/python/pyspark/sql/conversion.py
index 9110a6382725..bfa0d4a559a5 100644
--- a/python/pyspark/sql/conversion.py
+++ b/python/pyspark/sql/conversion.py
@@ -21,7 +21,7 @@ import decimal
from typing import TYPE_CHECKING, Any, Callable, List, Optional, Sequence,
Union, overload
import pyspark
-from pyspark.errors import PySparkRuntimeError, PySparkValueError
+from pyspark.errors import PySparkNotImplementedError, PySparkRuntimeError,
PySparkValueError
from pyspark.sql.pandas.types import (
_dedup_names,
_deduplicate_field_names,
@@ -62,6 +62,7 @@ from pyspark.sql.types import (
VariantType,
VariantVal,
_create_row,
+ _has_type,
)
if TYPE_CHECKING:
@@ -1276,6 +1277,24 @@ class ArrowTableToRowsConversion:
assert schema is not None and isinstance(schema, StructType)
+ # YearMonthIntervalType is serialized by the JVM as an Arrow
YEAR_MONTH interval, which
+ # PyArrow cannot materialize into Python values: `to_pylist()` raises
an opaque
+ # `KeyError: <Arrow type id>` from `get_array_class_from_type`. That
lookup fails for an
+ # empty column too (it resolves the array class before reading any
element), so the check
+ # below is intentionally unconditional in the row count -- it covers
empty results as well,
+ # surfacing a clean NOT_IMPLEMENTED instead of the opaque KeyError.
Collecting such a value
+ # is therefore not supported in the Spark Connect client; raise the
same NOT_IMPLEMENTED
+ # error as the classic PySpark path
(YearMonthIntervalType.fromInternal). Note that, unlike
+ # classic, PYSPARK_YM_INTERVAL_LEGACY (returning the integer months)
cannot be honored here,
+ # and an empty result raises rather than returning [] as classic would.
+ if any(_has_type(f.dataType, YearMonthIntervalType) for f in
schema.fields):
+ raise PySparkNotImplementedError(
+ errorClass="NOT_IMPLEMENTED",
+ messageParameters={
+ "feature": "Collecting a year-month interval value in
Spark Connect"
+ },
+ )
+
fields = schema.fieldNames()
if len(fields) > 0:
diff --git a/python/pyspark/sql/pandas/types.py
b/python/pyspark/sql/pandas/types.py
index 4c1ae1a0fedf..1a5c7a33c240 100644
--- a/python/pyspark/sql/pandas/types.py
+++ b/python/pyspark/sql/pandas/types.py
@@ -47,6 +47,7 @@ from pyspark.sql.types import (
TimestampType,
TimestampNTZType,
DayTimeIntervalType,
+ YearMonthIntervalType,
ArrayType,
MapType,
StructType,
@@ -410,6 +411,13 @@ def from_arrow_type(
spark_type = TimestampType()
elif types.is_duration(at):
spark_type = DayTimeIntervalType()
+ elif at.id == 21: # Arrow Type.INTERVAL_MONTHS
+ # The JVM serializes Spark's YearMonthIntervalType to an Arrow
YEAR_MONTH interval
+ # (an integer number of months); see ArrowUtils.scala /
ArrowWriter.scala. Unlike
+ # DayTimeIntervalType (sent as an Arrow Duration), PyArrow exposes no
factory or
+ # is_*() helper for this type -- only MONTH_DAY_NANO is in
pyarrow.types -- so match
+ # on the stable Arrow type id (Type::INTERVAL_MONTHS == 21).
+ spark_type = YearMonthIntervalType()
elif types.is_list(at):
spark_type = ArrayType(
elementType=from_arrow_type(at.value_type, prefer_timestamp_ntz),
diff --git a/python/pyspark/sql/tests/connect/test_connect_error.py
b/python/pyspark/sql/tests/connect/test_connect_error.py
index 50c6900cabc0..8f142dd55632 100644
--- a/python/pyspark/sql/tests/connect/test_connect_error.py
+++ b/python/pyspark/sql/tests/connect/test_connect_error.py
@@ -21,7 +21,7 @@ from pyspark.errors import PySparkAttributeError
from pyspark.errors.exceptions.base import SessionNotSameException
from pyspark.sql.types import Row
from pyspark.sql import functions as F
-from pyspark.errors import PySparkTypeError
+from pyspark.errors import PySparkNotImplementedError, PySparkTypeError
from pyspark.testing.connectutils import ReusedConnectTestCase
from pyspark.util import is_remote_only
@@ -261,10 +261,27 @@ class SparkConnectErrorTests(ReusedConnectTestCase):
)
def test_ym_interval_in_collect(self):
- # YearMonthIntervalType is not supported in python side arrow
conversion
- with self.assertRaises(PySparkTypeError):
+ # PyArrow cannot materialize Arrow YEAR_MONTH intervals, so collecting
a year-month
+ # interval over Spark Connect raises NOT_IMPLEMENTED (matching the
default of the classic
+ # PySpark path). Unlike classic, PYSPARK_YM_INTERVAL_LEGACY is not
honored here.
+ with self.assertRaises(PySparkNotImplementedError):
self.spark.sql("SELECT INTERVAL '10-8' YEAR TO MONTH AS
interval").first()
+ # A year-month interval nested inside an array is rejected the same
way (the schema-level
+ # check recurses into array/map/struct element types).
+ with self.assertRaises(PySparkNotImplementedError):
+ self.spark.sql("SELECT array(INTERVAL '10-8' YEAR TO MONTH) AS
interval").first()
+
+ def test_ym_interval_empty_collect(self):
+ # Even an empty result raises NOT_IMPLEMENTED rather than returning
[]. PyArrow cannot
+ # build the INTERVAL_MONTHS array class at all -- `to_pylist()` raises
`KeyError: 21` from
+ # get_array_class_from_type regardless of row count -- so the
schema-level check covers
+ # empty results too, giving a clean error instead of an opaque PyArrow
KeyError. This is
+ # one place Spark Connect diverges from classic PySpark, which returns
[] for an empty
+ # result (it never reaches PyArrow materialization).
+ with self.assertRaises(PySparkNotImplementedError):
+ self.spark.sql("SELECT INTERVAL '10-8' YEAR TO MONTH AS
interval").limit(0).collect()
+
if __name__ == "__main__":
from pyspark.testing import main
diff --git a/python/pyspark/sql/tests/connect/test_parity_types.py
b/python/pyspark/sql/tests/connect/test_parity_types.py
index 3f7417b36247..42346c4fc4a1 100644
--- a/python/pyspark/sql/tests/connect/test_parity_types.py
+++ b/python/pyspark/sql/tests/connect/test_parity_types.py
@@ -98,7 +98,13 @@ class TypesParityTests(TypesTestsMixin,
ReusedConnectTestCase):
def test_schema_with_collations_json_ser_de(self):
super().test_schema_with_collations_json_ser_de()
- @unittest.skip("This test is dedicated for PySpark Classic.")
+ @unittest.skip(
+ "The inherited Classic contract also asserts that
PYSPARK_YM_INTERVAL_LEGACY=1 returns "
+ "the integer months (Row(interval=128)), which Spark Connect cannot
satisfy: PyArrow has "
+ "no INTERVAL_MONTHS array support, so the legacy flag is not honored
and collect raises "
+ "NOT_IMPLEMENTED regardless. The default-raise behavior Connect does
match is covered by "
+
"test_connect_error.SparkConnectErrorTests.test_ym_interval_in_collect."
+ )
def test_ym_interval_in_collect(self):
super().test_ym_interval_in_collect()
diff --git a/python/pyspark/sql/types.py b/python/pyspark/sql/types.py
index 6b9020b3b104..2757964bcb9c 100644
--- a/python/pyspark/sql/types.py
+++ b/python/pyspark/sql/types.py
@@ -2882,6 +2882,8 @@ def _has_type(dt: DataType, dts: Union[type, Tuple[type,
...]]) -> bool:
return _has_type(dt.elementType, dts)
elif isinstance(dt, MapType):
return _has_type(dt.keyType, dts) or _has_type(dt.valueType, dts)
+ elif isinstance(dt, UserDefinedType):
+ return _has_type(dt.sqlType(), dts)
else:
return False
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
