HyukjinKwon commented on code in PR #59298:
URL: https://github.com/apache/spark/pull/59298#discussion_r4237058691
##########
python/pyspark/sql/types.py:
##########
@@ -484,6 +484,29 @@ def __repr__(self) -> str:
return "TimeType(%d)" % (self.precision)
+def _naive_dt_to_epoch_seconds(dt: datetime.datetime) -> int:
+ """Convert a naive local datetime to whole POSIX seconds, honoring
``fold``.
+
+ Shared by :meth:`TimestampType.toInternal` and
:meth:`TimestampLTZNanosType.toInternal`
+ so both local-time types agree on the repeated DST hour (SPARK-60081).
+
+ ``dt.replace(microsecond=0).timestamp()`` is used instead of
``time.mktime(dt.timetuple())``
+ so the result honors ``datetime.fold``. Dropping the microseconds first
keeps the float
+ integer-valued (and the ``int()`` conversion exact) across the full
0001..9999 range, so
+ the caller can add ``dt.microsecond`` back without the double-counting /
truncation that
+ plain ``int(dt.timestamp())`` produces for pre-1970 and far-future values.
+
+ ``datetime.timestamp()`` probes the UTC offset one day earlier on
0001-01-01 and raises
+ ``OverflowError`` ("year 0 is out of range"); fall back to ``time.mktime``
for that
Review Comment:
**Nit (P3):** On CPython, naive `datetime.min.timestamp()` raises
`ValueError('year 0 is out of range')`, not `OverflowError`, so the fallback
runs through the `ValueError` arm. `datetime.min` also doesn't round-trip:
`TimestampType().fromInternal(TimestampType().toInternal(datetime.datetime.min))`
raises the same `ValueError`. Wording such as "raises ValueError ('year 0 is
out of range'); fall back to time.mktime so toInternal keeps the previous value
for these range-edge inputs" would match the behavior.
**Verification:**
- **Inspection:** The docstring's stated trigger and guarantee match the
runtime behavior: naive datetime.min.timestamp() raises ValueError, and
toInternal(datetime.min) returns the mktime value.
##########
python/pyspark/sql/tests/test_types.py:
##########
@@ -3887,6 +3889,121 @@ def test_row_with_duplicated_fields(self):
self.assertEqual(list(r), [1, 2, 3])
self.assertEqual([v for v in r], [1, 2, 3])
+ # SPARK-60081: naive-datetime handling in TimestampType.toInternal and
+ # TimestampLTZNanosType.toInternal, and in the Arrow-to-row TIMESTAMP
converter. The
+ # production code only runs on POSIX systems where ``time.tzset`` works;
skip elsewhere.
Review Comment:
**Nit (P3):** The converters under test run on every platform. Only these
tests need `time.tzset()`, which is what the `skipUnless` guards check, so
"These tests need time.tzset() (POSIX only); skip elsewhere." would be accurate.
In `test_naive_timestamp_to_internal_range_extremes`, "must still
round-trip" overstates the check. The assertion only compares
`toInternal(datetime.min)` with the `time.mktime` value, and `fromInternal` of
that value raises.
The 'pre-PR' wording in that test also won't mean anything after merge; "the
previous time.mktime-based path" would hold up.
**Verification:**
- **Inspection:** The test comments match what the decorators and assertions
actually require and check.
##########
python/pyspark/sql/types.py:
##########
@@ -484,6 +484,29 @@ def __repr__(self) -> str:
return "TimeType(%d)" % (self.precision)
+def _naive_dt_to_epoch_seconds(dt: datetime.datetime) -> int:
+ """Convert a naive local datetime to whole POSIX seconds, honoring
``fold``.
+
+ Shared by :meth:`TimestampType.toInternal` and
:meth:`TimestampLTZNanosType.toInternal`
+ so both local-time types agree on the repeated DST hour (SPARK-60081).
+
+ ``dt.replace(microsecond=0).timestamp()`` is used instead of
``time.mktime(dt.timetuple())``
+ so the result honors ``datetime.fold``. Dropping the microseconds first
keeps the float
+ integer-valued (and the ``int()`` conversion exact) across the full
0001..9999 range, so
+ the caller can add ``dt.microsecond`` back without the double-counting /
truncation that
+ plain ``int(dt.timestamp())`` produces for pre-1970 and far-future values.
+
+ ``datetime.timestamp()`` probes the UTC offset one day earlier on
0001-01-01 and raises
+ ``OverflowError`` ("year 0 is out of range"); fall back to ``time.mktime``
for that
+ range-edge case so ``datetime.min`` still round-trips.
+ """
+ try:
+ seconds = dt.replace(microsecond=0).timestamp()
Review Comment:
**Blocking (P1):** This call dispatches to subclass overrides.
`pandas.Timestamp.timestamp()` treats a naive value as UTC, and `replace()`
returns another `pd.Timestamp`, so a naive `pd.Timestamp` is now converted as
if its wall time were UTC.
Under `TZ=America/Los_Angeles`,
`TimestampType().toInternal(pd.Timestamp('2021-01-01 12:00'))` returns
`1609502400000000` (12:00Z). The equivalent stdlib datetime and the old
`time.mktime` path both give `1609531200000000` (12:00 PST). Under Asia/Seoul
the value is off by 9 hours.
These values reach `toInternal` without type verification on two paths:
- non-Arrow Python UDF return values (`worker.py`);
- Spark Connect `F.lit(...)` and column comparisons such as `col('t') >=
pd.Timestamp(...)` (`connect/expressions.py`).
The result is silently wrong data in any non-UTC process.
The `dt.replace(microsecond=0).timestamp()` form came from my earlier
suggestion, and the previous revision's `dt.timestamp()` had the same problem.
Calling the stdlib implementation explicitly,
`datetime.datetime.timestamp(dt.replace(microsecond=0))`, keeps the local
interpretation and still honors fold. I checked that it gives `1609531200` here
and `1636277400` for a fold=1 `pd.Timestamp` at 2021-11-07 01:30.
**Verification:**
- **Regression:** Under a non-UTC process zone, toInternal of a naive
pandas.Timestamp equals toInternal of the equivalent stdlib datetime for both
local-time types.
- **Behavior:** Naive repeated-hour values still map fold=0 and fold=1 to
the earlier and later instants, for stdlib and pandas inputs.
- **Compatibility:** Existing stdlib naive-datetime expectations for
pre-1970 microseconds and datetime.min/max are unchanged.
##########
python/pyspark/sql/conversion.py:
##########
@@ -1429,7 +1430,13 @@ def convert_timestamp(value: Any) -> Any:
# lossless path), so drop any sub-microsecond digits to
match classic collect().
if hasattr(value, "to_pydatetime"):
value = value.to_pydatetime(warn=False)
- return value.astimezone().replace(tzinfo=None)
+ # SPARK-60081: route through integer POSIX seconds so
fromtimestamp honors
+ # datetime.fold on the local wall time, then re-attach
microseconds. Going
+ # through float timestamps would drop microseconds past
~2255 and overflow at
+ # 9999-12-31 under UTC; keeping the microseconds separate
avoids both.
+ return datetime.datetime.fromtimestamp(
Review Comment:
**Non-blocking (P2):** For instants in roughly the first day of year 1, the
local `fromtimestamp` probes the UTC offset 24 hours earlier and raises
`ValueError: year 0 is out of range`.
Under `TZ=UTC`, `0001-01-01 00:00:00Z` and `0001-01-01 12:00:00.000005Z` now
raise here. The replaced `value.astimezone().replace(tzinfo=None)` returned
`0001-01-01 00:00` and `0001-01-01 12:00:00.000005`. Under America/Los_Angeles,
the failing window runs until about `0001-01-02 08:00Z`.
This converter backs Spark Connect `collect()`/`toLocalIterator()` and Arrow
Python UDF/UDTF inputs, so a `TIMESTAMP'0001-01-01 00:00:00'` sentinel now
fails the whole action on those paths.
`test_arrow_convert_timestamp_range_extremes` starts at `0001-01-02 08:00Z`,
just past the window, so it doesn't catch this.
The integer-second form I suggested earlier missed this edge, and the
previous float form failed the same way. A fallback like the one in
`_naive_dt_to_epoch_seconds` would fix it:
`value.astimezone().replace(tzinfo=None)` on `(OverflowError, ValueError)`.
That restores these values and still keeps fold=1 at 2021-11-07 09:30Z and
exact microseconds at 9999-12-31.
**Verification:**
- **Regression:** First-day-of-year-1 instants convert without ValueError
and match the previous astimezone() results under UTC and America/Los_Angeles,
for both timestamp types.
- **Behavior:** Repeated-hour fold handling and far-range microsecond
exactness from the existing converter tests still hold.
##########
python/pyspark/sql/types.py:
##########
@@ -492,9 +492,7 @@ def needConversion(self) -> bool:
def toInternal(self, dt: datetime.datetime) -> int:
if dt is not None:
- seconds = (
- calendar.timegm(dt.utctimetuple()) if dt.tzinfo else
time.mktime(dt.timetuple())
- )
+ seconds = calendar.timegm(dt.utctimetuple()) if dt.tzinfo else
dt.timestamp()
Review Comment:
Thanks, the double counting is gone and the pre-1970 test covers it. One
follow-up on the form I suggested: `dt.replace(microsecond=0).timestamp()`
dispatches to `pandas.Timestamp.timestamp()` when a naive `pd.Timestamp` comes
in (for example from a non-Arrow UDF return or a Connect `lit`), and pandas
reads naive values as UTC. Under America/Los_Angeles,
`toInternal(pd.Timestamp('2021-01-01 12:00'))` now lands 8 hours off. Calling
the stdlib method explicitly,
`datetime.datetime.timestamp(dt.replace(microsecond=0))`, keeps the local
interpretation and still honors fold.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4236503037","thread_id":"inline:4236503037","verdict_sha256":"12a6526df95d6060523b1e644a6680422edc34e5bbf73a806263de56014de191"}
-->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
