andygrove opened a new issue, #7214:
URL: https://github.com/apache/datafusion-comet/issues/7214
### Describe the bug
With an explicit read schema, the native Parquet scan accepts reads of a
timestamp column that Spark's vectorized reader rejects with
`[FAILED_READ_FILE.PARQUET_COLUMN_DATA_TYPE_MISMATCH]`. Reading a
`TIMESTAMP(MICROS)`, INT96 or `TIMESTAMP(NANOS)` column as `string`, `double`
or `decimal(20,0)` returns the converted value, and reading it as `int` returns
NULL. Reading an INT96 column as `bigint` returns microseconds since the epoch.
Reading an INT64 timestamp as `bigint`, or any of the three as `date`, matches
Spark.
`spark_has_updater` in `native/core/src/parquet/schema_adapter.rs` rejects
`Timestamp -> Date32`, but it has no rule for a timestamp leaf read as a
string, an integer, a float or a decimal, so the adapter casts it. In Spark's
`ParquetVectorUpdaterFactory`, the INT96 branch only accepts `TimestampType`
and `TimestampNTZType`. The INT64 branch accepts `LongType`, a timestamp with a
matching unit, `DayTimeIntervalType` and `TimeType`. It accepts a decimal only
for a decimal or unsigned integer annotation. INT96 and INT64 timestamps reach
the adapter as the same Arrow type, so the INT96 `bigint` case needs the
Parquet type of the leaf. #7089 adds that for `TIMESTAMP(NANOS)`.
### Steps to reproduce
Start a session with the configs from the installation guide
(`spark.plugins=org.apache.spark.CometPlugin`, the Comet shuffle manager,
off-heap memory), then:
```scala
spark.conf.set("spark.sql.parquet.outputTimestampType", "TIMESTAMP_MICROS")
spark.sql("SELECT 1 AS id, TIMESTAMP'2020-01-01 01:02:03.123456' AS
t").write.parquet("/tmp/ts_micros")
spark.conf.set("spark.sql.parquet.outputTimestampType", "INT96")
spark.sql("SELECT 1 AS id, TIMESTAMP'2020-01-01 01:02:03.123456' AS
t").write.parquet("/tmp/ts_int96")
spark.read.schema("id int, t string").parquet("/tmp/ts_micros").collect()
spark.read.schema("id int, t int").parquet("/tmp/ts_micros").collect()
spark.read.schema("id int, t bigint").parquet("/tmp/ts_int96").collect()
```
Build each DataFrame again after setting `spark.comet.enabled=false` to see
Spark's result, since a DataFrame keeps the plan from its first action. For
`TIMESTAMP(NANOS)`, write the file with pyarrow, which uses nanoseconds by
default from version 13.
On `main` at 330b29cf10, Spark 4.1.3 (Spark 3.4.3 gives the same results):
| File column | Read as | Spark | Comet native scan |
| --- | --- | --- | --- |
| `TIMESTAMP(MICROS)`, INT96, `TIMESTAMP(NANOS)` | `string` |
`PARQUET_COLUMN_DATA_TYPE_MISMATCH` | the timestamp as text |
| `TIMESTAMP(MICROS)`, INT96, `TIMESTAMP(NANOS)` | `int` | the same error |
NULL |
| `TIMESTAMP(MICROS)`, INT96, `TIMESTAMP(NANOS)` | `double` | the same error
| the raw value as a double |
| `TIMESTAMP(MICROS)`, INT96, `TIMESTAMP(NANOS)` | `decimal(20,0)` | the
same error | the raw value |
| INT96 | `bigint` | the same error | microseconds since the epoch |
The plans show `CometNativeScan parquet`, and adding a filter on `id` gives
the same results.
### Expected behavior
The native scan rejects these reads the way Spark does, deferred to decode
time like the other updater rejections (`RejectOnNonEmpty`).
### Additional context
Found while reviewing #7089. #4297 covered the numeric pairs, and #5188
covers ANSI interval targets. Like #7034, Comet returns values where Spark
fails, but the `int` read returns NULL rather than a plausible value.
--
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]
