Skip to content

Native Parquet scan reads timestamp columns as string, int, double or decimal, and INT96 as bigint, where Spark rejects them #7214

Open

Description

@andygrove

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.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    area:scanParquet scan / data readingbugSomething isn't workingpriority:lowMinor issues, test failures, tooling, cosmetic

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions