stevomitric opened a new pull request, #59780:
URL: https://github.com/apache/spark/pull/59780
### What changes were proposed in this pull request?
`ArrowColumnVector`'s `TimestampNTZNanosAccessor` /
`TimestampLTZNanosAccessor` now floor the decoded `timestamp[ns]` value to the
vector's declared precision (recovered from the
`SPARK::timestampNanos::precision` field metadata by
`ArrowUtils.fromArrowField`). Below p = 9 they call the shared
`DateTimeUtils.epochNanosToTimestampNanos(nanos, precision)` helper, the same
one the Parquet and Avro readers use; p = 9 keeps the existing decode unchanged.
### Why are the changes needed?
Spark keeps nanosecond timestamp values floored to the column precision (p =
7 is a 100 ns grid, p = 8 a 10 ns grid); casts, Parquet, Avro and the text
parsers all floor on the way in. The Arrow reader did not, so an Arrow batch
from a foreign producer with more sub-microsecond digits than the column allows
kept them. Values that are equal at the declared precision then compared,
hashed and grouped as different, and failed equality with a literal of the same
type. Reachable today, for example:
```python
pdf = pd.DataFrame({"ts": pd.to_datetime(["2020-01-02 03:04:05.123456789",
"2020-01-02 03:04:05.123456712"])})
df = spark.createDataFrame(pdf, "ts timestamp_ntz(7)")
df.distinct().count() # 2 before, 1 after
df.where("ts = CAST('2020-01-02 03:04:05.1234567' AS
TIMESTAMP_NTZ(7))").count() # 0 before, 2 after
```
The same happens for `createDataFrame` from a `pyarrow.Table`, over Spark
Connect (`LocalRelation` with Arrow data), and for Arrow-optimized Python UDTF
results.
This covers vectors whose Arrow field carries the precision tag. The PySpark
worker currently drops the tag on the top-level fields of UDF result batches,
so `mapInPandas`, `mapInArrow`, `applyInPandas`, `applyInArrow` and Python data
sources still read such a column as p = 9, and a scalar `pandas_udf` or
Arrow-optimized UDF with a p < 9 return type fails with `ARROW_TYPE_MISMATCH`.
That is a separate PySpark fix, left for a follow-up.
### Does this PR introduce _any_ user-facing change?
Yes, for the unreleased nanosecond timestamp types (behind
`spark.sql.timestampNanosTypes.enabled`): Arrow input to a `TIMESTAMP_NTZ(p)` /
`TIMESTAMP_LTZ(p)` column with p < 9 is now floored to p, matching every other
ingestion path. p = 9 is unchanged.
### How was this patch tested?
- `ArrowColumnVectorSuite`: new tests reading off-grid int64 epoch-nanos
(including pre-epoch values and `Long.MinValue`/`MaxValue`) at p = 7/8/9 for
NTZ and LTZ, and from a nested struct field; `ArrowConvertersSuite`: batches
written at p = 9 and read with a declared p = 7/8 schema (the `createDataFrame`
path); `ArrowWriterSuite`: the p = 7 round-trip now expects the floored value.
- PySpark `test_types.py`: `createDataFrame` from pandas and pyarrow with
NTZ(7), LTZ(8) and pre-epoch values, run in classic and via the Connect parity
suite.
- Existing `ArrowUtilsSuite`, `DateTimeUtilsSuite`,
`org.apache.spark.sql.execution.arrow.*`, `org.apache.spark.sql.vectorized.*`
and `ColumnarBatchSuite` pass.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
--
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]
