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 18d6d218e6b0 [SPARK-55299][PS] Infer the correct unit for calculated
timedeltas
18d6d218e6b0 is described below
commit 18d6d218e6b05c746a55021aa8883405445f1fe6
Author: Fangchen Li <[email protected]>
AuthorDate: Tue Jun 23 07:42:02 2026 +0900
[SPARK-55299][PS] Infer the correct unit for calculated timedeltas
### What changes were proposed in this pull request?
This PR adds `_with_inferred_unit` to `TimedeltaOps.sub`/`rsub` to set the
result dtype from the operand units (numpy `timedelta64` dtype, or
`pd.Timedelta.unit` for scalars)
### Why are the changes needed?
To match pandas 3 behavior.
### Does this PR introduce _any_ user-facing change?
Yes. On pandas 3, the dtype of a timedelta produced by subtraction now
follows the finer resolution of the operands (capped at microseconds)
instead
of always timedelta64[us]. Behavior on pandas < 3.0.0 is unchanged.
### How was this patch tested?
Unitetests added.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 4.8
Closes #56624 from fangchenli/SPARK-55299-timedelta-unit.
Authored-by: Fangchen Li <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>
---
.../pyspark/pandas/data_type_ops/timedelta_ops.py | 39 +++++++++++++++++++-
.../tests/data_type_ops/test_timedelta_ops.py | 43 ++++++++++++++++++++++
2 files changed, 80 insertions(+), 2 deletions(-)
diff --git a/python/pyspark/pandas/data_type_ops/timedelta_ops.py
b/python/pyspark/pandas/data_type_ops/timedelta_ops.py
index aa3e8f64801f..6f17474f61d8 100644
--- a/python/pyspark/pandas/data_type_ops/timedelta_ops.py
+++ b/python/pyspark/pandas/data_type_ops/timedelta_ops.py
@@ -18,6 +18,7 @@
from datetime import timedelta
from typing import Any, Union
+import numpy as np
import pandas as pd
from pandas.api.types import CategoricalDtype
@@ -72,6 +73,38 @@ class TimedeltaOps(DataTypeOps):
else:
return col.astype(self.dtype)
+ def _with_inferred_unit(
+ self, result: SeriesOrIndex, left: IndexOpsLike, right:
Union[IndexOpsMixin, timedelta]
+ ) -> SeriesOrIndex:
+ # pandas 3.0.0+ promotes timedelta arithmetic to the finer resolution
of the
+ # operands; before that, pandas-on-Spark represented timedelta as
nanoseconds.
+ if LooseVersion(pd.__version__) < "3.0.0":
+ return result
+
+ def unit_of(obj: Union[IndexOpsMixin, timedelta]) -> str:
+ if isinstance(obj, IndexOpsMixin):
+ dtype = obj.dtype
+ if isinstance(dtype, np.dtype) and np.issubdtype(dtype,
np.timedelta64):
+ return np.datetime_data(dtype)[0]
+ elif isinstance(obj, pd.Timedelta):
+ return obj.unit
+ # datetime.timedelta scalars and object-backed interval columns
map to microseconds.
+ return "us"
+
+ promoted = np.promote_types(
+ np.dtype(f"timedelta64[{unit_of(left)}]"),
+ np.dtype(f"timedelta64[{unit_of(right)}]"),
+ )
+ # DayTimeIntervalType stores microseconds and cannot represent finer
resolutions.
+ if np.datetime_data(promoted)[0] == "ns":
+ promoted = np.dtype("timedelta64[us]")
+
+ field = result._internal.data_fields[0]
+ if field.dtype == promoted:
+ # Already the right resolution (the common us case); avoid
rebuilding the field.
+ return result
+ return result._with_new_scol(result.spark.column,
field=field.copy(dtype=promoted))
+
def sub(self, left: IndexOpsLike, right: Any) -> SeriesOrIndex:
_sanitize_list_like(right)
@@ -80,7 +113,8 @@ class TimedeltaOps(DataTypeOps):
and isinstance(right.spark.data_type, DayTimeIntervalType)
or isinstance(right, timedelta)
):
- return pyspark_column_op("__sub__", left, right)
+ result = pyspark_column_op("__sub__", left, right)
+ return self._with_inferred_unit(result, left, right)
else:
raise TypeError("Timedelta subtraction can only be applied to
timedelta series.")
@@ -88,7 +122,8 @@ class TimedeltaOps(DataTypeOps):
_sanitize_list_like(right)
if isinstance(right, timedelta):
- return pyspark_column_op("__rsub__", left, right)
+ result = pyspark_column_op("__rsub__", left, right)
+ return self._with_inferred_unit(result, left, right)
else:
raise TypeError("Timedelta subtraction can only be applied to
timedelta series.")
diff --git a/python/pyspark/pandas/tests/data_type_ops/test_timedelta_ops.py
b/python/pyspark/pandas/tests/data_type_ops/test_timedelta_ops.py
index 868850a27c27..34190920017a 100644
--- a/python/pyspark/pandas/tests/data_type_ops/test_timedelta_ops.py
+++ b/python/pyspark/pandas/tests/data_type_ops/test_timedelta_ops.py
@@ -17,10 +17,12 @@
from datetime import timedelta
+import numpy as np
import pandas as pd
from pandas.api.types import CategoricalDtype
import pyspark.pandas as ps
+from pyspark.loose_version import LooseVersion
from pyspark.testing.pandasutils import PandasOnSparkTestCase
from pyspark.pandas.tests.data_type_ops.testing_utils import OpsTestBase
@@ -73,6 +75,47 @@ class TimedeltaOpsTestsMixin:
pdf, psdf = self.timedelta_pdf, self.timedelta_psdf
self.assert_eq(pdf["that"] - pdf["this"], psdf["that"] - psdf["this"])
+ def test_sub_unit(self):
+ # pandas 3.0.0+ takes the finer resolution of the operands instead of
always
+ # microseconds (SPARK-55299); before that timedelta64 is always
nanoseconds.
+ if LooseVersion(pd.__version__) < "3.0.0":
+ return
+
+ for left_unit, right_unit in [("s", "s"), ("ms", "ms"), ("us", "us"),
("s", "ms")]:
+ pser1 = pd.Series(np.array([3, 5, 8],
dtype=f"timedelta64[{left_unit}]"))
+ pser2 = pd.Series(np.array([1, 2, 3],
dtype=f"timedelta64[{right_unit}]"))
+ psser1, psser2 = ps.from_pandas(pser1), ps.from_pandas(pser2)
+ self.assert_eq(pser1 - pser2, psser1 - psser2)
+ self.assert_eq(pser1 - timedelta(seconds=1), psser1 -
timedelta(seconds=1))
+ self.assert_eq(timedelta(seconds=1) - pser1, timedelta(seconds=1)
- psser1)
+
+ # pd.Timedelta is a datetime.timedelta subclass that carries its
own resolution.
+ for unit in ["s", "ms"]:
+ scalar = pd.Timedelta(1, unit=unit)
+ self.assert_eq(pser1 - scalar, psser1 - scalar)
+ self.assert_eq(scalar - pser1, scalar - psser1)
+
+ pidx = pd.Index(np.array([3, 5, 8], dtype="timedelta64[s]"))
+ psidx = ps.from_pandas(pidx)
+ self.assert_eq(pidx - pidx, psidx - psidx)
+
+ # DayTimeIntervalType cannot represent nanoseconds, so nanosecond
operands are
+ # capped at microseconds rather than matching pandas' nanosecond
result.
+ pser_ns = pd.Series(np.array([3, 5, 8], dtype="timedelta64[ns]"))
+ pser_s = pd.Series(np.array([3, 5, 8], dtype="timedelta64[s]"))
+ psser_ns, psser_s = ps.from_pandas(pser_ns), ps.from_pandas(pser_s)
+ self.assertEqual((psser_ns - psser_ns)._to_pandas().dtype,
np.dtype("timedelta64[us]"))
+ self.assertEqual(
+ (psser_s - pd.Timedelta(1, unit="ns"))._to_pandas().dtype,
np.dtype("timedelta64[us]")
+ )
+
+ # object-backed interval columns carry no numpy resolution; must not
fail.
+ pser_obj = pd.Series([timedelta(days=1), timedelta(seconds=2)],
dtype=object)
+ psser_obj = ps.from_pandas(pser_obj)
+ expected = pser_obj.astype("timedelta64[us]")
+ self.assert_eq(expected - timedelta(0), psser_obj - timedelta(0))
+ self.assert_eq(expected - expected, psser_obj - psser_obj)
+
def test_mul(self):
self.assertRaises(TypeError, lambda: self.psser * "x")
self.assertRaises(TypeError, lambda: self.psser * 1)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
