Repository navigation
Native Parquet scan reads timestamp columns as string, int, double or decimal, and INT96 as bigint, where Spark rejects them #7214
Description
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:
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 330b29c, 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.
Activity