uros-b commented on code in PR #58418:
URL: https://github.com/apache/spark/pull/58418#discussion_r3927730735
##########
python/pyspark/sql/pandas/conversion.py:
##########
@@ -632,6 +639,16 @@ def createDataFrame( # type: ignore[misc]
selfcheck = arrowSafeTypeConversion == "true"
infer_pandas_dict_as_map = inferPandasDictAsMap == "true"
+ # Building a DataFrame from pandas/PyArrow data goes through Arrow,
whose nanosecond
+ # timestamp value conversion is a pending follow-up; fail
deterministically for an
+ # explicit nanosecond-typed schema rather than silently mis-handle the
values. Covers a
+ # bare atomic DataType schema as well as a StructType (a DDL string /
list of names cannot
Review Comment:
The pandas createDataFrame comment says a DDL string cannot carry these
types. Classic SparkSession.createDataFrame runs _parse_ddl first, so "ts
timestamp_ntz(9)" is already a StructType when the pandas mixin runs, and the
isinstance(schema, DataType) guard does fire.
##########
python/pyspark/sql/pandas/types.py:
##########
@@ -77,6 +78,45 @@
metadata_key = b"SPARK::metadata::json"
+def _contains_timestamp_nanos(dt: DataType) -> bool:
+ """True if ``dt`` is, or structurally contains, a nanosecond-capable
timestamp type.
+
+ The Arrow / pandas value conversion for :class:`TimestampNTZNanosType` /
+ :class:`TimestampLTZNanosType` is not implemented yet (planned follow-up).
Rather than let
+ these paths silently mis-handle the value (wrong time zone for LTZ, or a
leaked
+ ``pandas.Timestamp``), callers use this to fail deterministically; see
+ :func:`_reject_timestamp_nanos_conversion`.
+ """
+ if isinstance(dt, AnyTimestampNanoType):
+ return True
+ elif isinstance(dt, ArrayType):
+ return _contains_timestamp_nanos(dt.elementType)
+ elif isinstance(dt, MapType):
+ return _contains_timestamp_nanos(dt.keyType) or
_contains_timestamp_nanos(dt.valueType)
+ elif isinstance(dt, StructType):
+ return any(_contains_timestamp_nanos(f.dataType) for f in dt.fields)
+ elif isinstance(dt, UserDefinedType):
+ return _contains_timestamp_nanos(dt.sqlType())
+ else:
+ return False
+
+
+def _reject_timestamp_nanos_conversion(schema: DataType) -> None:
+ """Raise if ``schema`` involves a nanosecond timestamp type, for
Arrow/pandas value paths.
+
+ Keeps the not-yet-supported Arrow/pandas/Connect data paths failing
deterministically instead
+ of silently producing wrong values, consistent with :func:`to_arrow_type`,
which already
+ rejects these types with the same error condition.
+ """
+ from pyspark.errors import PySparkTypeError
+
+ if _contains_timestamp_nanos(schema):
+ raise PySparkTypeError(
+ errorClass="UNSUPPORTED_DATA_TYPE_FOR_ARROW_CONVERSION",
+ messageParameters={"data_type": str(schema)},
Review Comment:
_reject_timestamp_nanos_conversion reports str(schema) (the whole struct).
to_arrow_type reports the leaf. Same class of issue as the map-key message.
##########
python/pyspark/sql/types.py:
##########
@@ -484,6 +488,142 @@ def fromInternal(self, ts: int) -> datetime.datetime:
)
+class AnyTimestampNanoType(DatetimeType):
+ """
+ Super class of the nanosecond-capable timestamp data types
+ :class:`TimestampNTZNanosType` and :class:`TimestampLTZNanosType`.
+
+ .. versionadded:: 4.4.0
+ """
+
+ MIN_PRECISION: int = 7
+ MAX_PRECISION: int = 9
+ DEFAULT_PRECISION: int = 9
+ # Precision of the standard microsecond timestamp types, which the
parameterized DDL / JSON
+ # type names also accept (``timestamp_ntz(6)`` / ``timestamp_ltz(6)``).
+ MICROS_PRECISION: int = 6
+
+ # Set by each subclass to the SQL type name used in the DDL / JSON
representation, e.g.
+ # "timestamp_ntz". Also used, upper-cased, in the invalid-precision error
message.
+ _sqlTypeName: str = ""
+
+ def __init__(self, precision: int = DEFAULT_PRECISION):
+ # Reject non-integer precision (e.g. 7.5 or float("nan")), which would
otherwise slip
+ # through the range comparison below. operator.index accepts any
integer-like value
+ # (including a NumPy integer) and rejects the rest with a TypeError.
+ try:
+ precision = operator.index(precision)
+ except TypeError:
+ raise PySparkValueError(
+ errorClass="INVALID_TIMESTAMP_PRECISION",
+ messageParameters={
+ "precision": repr(precision),
+ "type": self._sqlTypeName.upper(),
+ },
+ )
+ if precision < self.MIN_PRECISION or precision > self.MAX_PRECISION:
+ raise PySparkValueError(
+ errorClass="INVALID_TIMESTAMP_PRECISION",
+ messageParameters={
+ "precision": str(precision),
+ "type": self._sqlTypeName.upper(),
+ },
+ )
+ self.precision = precision
+
+ def needConversion(self) -> bool:
+ return True
+
+ def simpleString(self) -> str:
+ return "%s(%d)" % (self._sqlTypeName, self.precision)
+
+ def jsonValue(self) -> str:
+ return "%s(%d)" % (self._sqlTypeName, self.precision)
+
+ def __repr__(self) -> str:
+ return "%s(%d)" % (type(self).__name__, self.precision)
+
+
+class TimestampNTZNanosType(AnyTimestampNanoType):
+ """Timestamp (datetime.datetime) data type without timezone information,
with
+ nanosecond-capable fractional-second precision (7 to 9 digits).
+
+ Parameters
+ ----------
+ precision : int, optional
+ Number of digits of fractional seconds, one of 7, 8 or 9 (default: 9).
+
+ Notes
+ -----
+ These types are behind the ``spark.sql.timestampNanosTypes.enabled``
preview flag (disabled
+ by default); using them while it is off raises an error.
+
+ ``datetime.datetime`` is microsecond-resolution, so values crossing the
Python boundary as
+ ``datetime.datetime`` -- :meth:`DataFrame.collect`,
:meth:`DataFrame.toLocalIterator`, and
+ Python UDF arguments -- are truncated to microseconds, as are
``datetime.datetime`` values
+ supplied to :meth:`SparkSession.createDataFrame` from Python lists/rows.
The value stored by
+ Spark keeps full precision; only this Python boundary is
microsecond-resolution. A ``map``
+ with keys of this type that differ only below a microsecond would collapse
to one entry, so
+ that conversion raises rather than silently dropping an entry.
+
+ Arrow- and pandas-based conversion for these types --
:meth:`DataFrame.toPandas`,
+ :meth:`SparkSession.createDataFrame` from a pandas ``DataFrame``,
Arrow-based UDFs, and the
+ Spark Connect data path -- is not yet supported and raises
+ ``UNSUPPORTED_DATA_TYPE_FOR_ARROW_CONVERSION``; it is planned as a
follow-up.
+
+ .. versionadded:: 4.4.0
+ """
+
+ _sqlTypeName = "timestamp_ntz"
+
+ def toInternal(self, dt: datetime.datetime) -> int:
+ # Mirrors TimestampNTZType.toInternal: the value is on the UTC grid
and carries no zone.
+ if dt is not None:
+ seconds = calendar.timegm(dt.timetuple())
+ return int(seconds) * 1000000 + dt.microsecond
+
+ def fromInternal(self, ts: int) -> datetime.datetime:
+ if ts is not None:
+ # using int to avoid precision loss in float
+ return datetime.datetime.fromtimestamp(ts // 1000000,
datetime.timezone.utc).replace(
+ microsecond=ts % 1000000, tzinfo=None
+ )
+
+
+class TimestampLTZNanosType(AnyTimestampNanoType):
+ """Timestamp (datetime.datetime) data type with local timezone semantics,
with
+ nanosecond-capable fractional-second precision (7 to 9 digits).
+
+ Parameters
+ ----------
+ precision : int, optional
+ Number of digits of fractional seconds, one of 7, 8 or 9 (default: 9).
+
+ Notes
+ -----
+ Carries the same microsecond-only Python boundary as
:class:`TimestampNTZNanosType`; see the
+ notes there.
+
+ .. versionadded:: 4.4.0
+ """
+
+ _sqlTypeName = "timestamp_ltz"
Review Comment:
Bare "timestamp_ltz" is still missing from _all_mappable_types (JVM maps it
to TimestampType). You already parse timestamp_ltz(6) that way; adding the bare
name would match DataType.parseDataType. Pre-existing, easy to include here.
--
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]
