HyukjinKwon commented on code in PR #59298:
URL: https://github.com/apache/spark/pull/59298#discussion_r4236503041
##########
python/pyspark/sql/tests/test_udf.py:
##########
@@ -809,6 +811,45 @@ def noop(x):
self.assertEqual(df.schema[0].dataType.typeName(), "timestamp_ntz")
self.assertEqual(df.first()[0], datetime.datetime(1970, 1, 1, 0,
0))
+ def test_udf_timestamp_dst_fold_round_trip(self):
+ # SPARK-60081: preserve datetime.fold in the repeated DST hour for
Python UDFs.
+ tz = "America/Los_Angeles"
+ tz_prev = os.environ.get("TZ", None)
+ try:
+ os.environ["TZ"] = tz
Review Comment:
**Non-blocking (P2):** `os.environ["TZ"]` plus `time.tzset()` only affects
this driver process. The UDF bodies and both changed converters run in Python
workers, which the JVM starts with its launch-time environment plus
`sc.environment`, so they never see America/Los_Angeles. In CI, PySpark jobs
set no TZ, so the containers run in UTC and the workers have no repeated hour.
All four assertions then also hold with the pre-fix code; I checked the old
expressions under TZ=UTC and they give 1636277400/1636273800.
Other TZ-sensitive suites set `cls.sc.environment["TZ"] = tz` (e.g.
`goldenutils.setup_timezone`, `test_arrow_map.py`,
`test_pandas_udf_scalar.py`). The Connect parity suite (`UDFParityTests`) has
no `sc`, so in-process tests of the two converters under a set TZ may be the
more reliable signal there. Also check how the inherited test behaves in
`ArrowPythonUDFLegacyTests`, which uses the legacy pandas conversion path.
See **Shared repair plan 2** in the review body.
##########
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:
**Blocking (P1):** For a naive datetime, `dt.timestamp()` already includes
`dt.microsecond / 1e6`, so `int(seconds) * 1000000 + dt.microsecond` counts the
fraction twice. `int()` also truncates toward zero, so every pre-1970 value
with non-zero microseconds comes out one second late, and far-future values
with large microseconds round up. For example, with TZ=UTC, `datetime(1969, 12,
31, 23, 59, 59, 500000)` used to give `-500000` and now gives `500000`. With
TZ=America/Los_Angeles, `datetime(1950, 5, 3, 12, 0, 0, 123000)` moves by
+1,000,000 us. The removed `time.mktime(dt.timetuple())` returned whole
seconds, so this was exact before.
This backs classic `createDataFrame` and non-Arrow UDF return values, so
those timestamps are silently corrupted. Dropping the microseconds before the
float conversion (`dt.replace(microsecond=0).timestamp()`) keeps fold and gives
integral seconds, but see the separate note about `datetime.min`, which that
form does not handle.
See **Shared repair plan 1** in the review body.
##########
python/pyspark/sql/tests/test_udf.py:
##########
@@ -809,6 +811,45 @@ def noop(x):
self.assertEqual(df.schema[0].dataType.typeName(), "timestamp_ntz")
self.assertEqual(df.first()[0], datetime.datetime(1970, 1, 1, 0,
0))
+ def test_udf_timestamp_dst_fold_round_trip(self):
+ # SPARK-60081: preserve datetime.fold in the repeated DST hour for
Python UDFs.
+ tz = "America/Los_Angeles"
+ tz_prev = os.environ.get("TZ", None)
+ try:
+ os.environ["TZ"] = tz
+ time.tzset()
+ with self.sql_conf(
+ {
+ "spark.sql.session.timeZone": tz,
+ "spark.sql.execution.pythonUDF.arrow.enabled": "true",
+ }
+ ):
+ second_fold = self.spark.sql("SELECT timestamp'2021-11-07
09:30:00Z' AS t")
+ first_fold = self.spark.sql("SELECT timestamp'2021-11-07
08:30:00Z' AS t")
+
+ for kwargs in [{}, {"useArrow": False}]:
+ to_epoch = udf(lambda x: int(x.timestamp()), LongType(),
**kwargs)
+ identity = udf(lambda x: x, TimestampType(), **kwargs)
+
+ self.assertEqual(1636277400,
second_fold.select(to_epoch("t")).first()[0])
Review Comment:
**Nit (P3):** The `convert_timestamp` change also changes what Spark Connect
`collect()` returns: a collected TIMESTAMP in the repeated hour now carries
fold=1. Every assertion here reads a Long computed in a worker (`to_epoch`) or
in the JVM (`unix_timestamp`), so reverting the conversion.py change would not
fail anything under `UDFParityTests`. This test already sets the driver TZ, so
an assertion such as `self.assertEqual(second_fold.first()[0].timestamp(),
1636277400)` would cover the client-side converter in Connect mode, and
`fromInternal` in classic mode.
See **Shared repair plan 2** in the review body.
##########
python/pyspark/sql/conversion.py:
##########
@@ -1429,7 +1429,7 @@ 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)
+ return datetime.datetime.fromtimestamp(value.timestamp())
Review Comment:
**Non-blocking (P2):** `datetime.datetime.fromtimestamp(value.timestamp())`
sends the instant through a float. Once |epoch seconds| goes past about 2^33
(roughly before 1685 or after 2255), a double can no longer hold microseconds;
for example, `3000-01-01 00:00:00.123457Z` comes back as `.123459`. Near the
top of the range it can also round across the year limit: with TZ=UTC,
`9999-12-31 23:59:59.999999Z` now raises `ValueError: year 10000 is out of
range`, and under America/Los_Angeles it becomes `16:00:00`. The previous
`astimezone()` path was exact.
This converter backs Arrow-optimized Python UDF inputs and Spark Connect
`collect()`/`toLocalIterator()`, so values that classic `collect()` returns
exactly can now drift or fail. `TimestampType.fromInternal` avoids this with
integer seconds. Something like
`datetime.datetime.fromtimestamp(calendar.timegm(value.utctimetuple())).replace(microsecond=value.microsecond)`
keeps fold, because `fromtimestamp` of an int sets it, without the precision
loss. I checked that this form returns 01:30 fold=1 for 09:30Z and stays exact
for 9999-12-31 23:59:59.999999 under UTC.
**Verification:**
- **Regression:** Far-range aware inputs with microseconds (e.g. 3000-01-01
00:00:00.123457Z, 0001-01-02 08:00:00.123457Z, 9999-12-31 23:59:59.999999Z)
convert to the exact local wall time, without drift or ValueError, under UTC
and America/Los_Angeles.
- **Behavior:** Under America/Los_Angeles, 2021-11-07 09:30:00Z converts to
01:30 fold=1 and 08:30:00Z to 01:30 fold=0, and .timestamp() of each returns
the original instant.
--
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]
