andygrove opened a new issue, #7048:
URL: https://github.com/apache/datafusion-comet/issues/7048
### Describe the bug
The native Parquet writer doesn't apply three Spark settings that change
what a reader of the file sees. They apply at the top level and inside structs,
arrays and maps:
1. **`spark.sql.parquet.outputTimestampType`.** The native writer always
writes `TimestampType` as `INT64 (TIMESTAMP(MICROS,true))`. Spark writes
`INT96` by default, and `INT64 (TIMESTAMP(MILLIS,true))` with
`TIMESTAMP_MILLIS`. With `TIMESTAMP_MILLIS`, Spark truncates to milliseconds
while the native writer keeps the microseconds, so the files read back
differently: `2023-11-14 22:13:20.123` from Spark's file and `2023-11-14
22:13:20.123456` from the native one. #3425 is the Spark SQL test the `INT96`
default breaks.
2. **`spark.sql.parquet.writeLegacyFormat=true`.** Spark writes the layout
that Hive 1.x and older Impala read: every decimal as `FIXED_LEN_BYTE_ARRAY`,
lists as `repeated group bag { ... array }`, and maps annotated
`MAP_KEY_VALUE`. The native writer writes the standard layout, with `INT32` and
`INT64` decimals.
3. **`org.apache.spark.sql.parquet.row.metadata`.** Spark stores the
Catalyst schema as JSON under this footer key, and its schema inference prefers
it over the Parquet types. The native writer writes `ARROW:schema` instead, so
a path-based read of a native file infers the schema from the Parquet types.
Field metadata such as column comments is lost, and with
`spark.sql.parquet.inferTimestampNTZ.enabled=false`, `TIMESTAMP_NTZ` columns,
nested ones included, come back as `TIMESTAMP`. Tables read through a catalog
are not affected. The other Spark footer key, `org.apache.spark.version`, is
#3427, and without it Spark rebases or rejects the native files' old dates
(#7047).
The native files also use arrow-rs's names where Spark uses the Parquet
specification's: `list`/`item` for list elements, `entries` for map entries and
`arrow_schema` for the root. Spark, parquet-mr and Trino read them, so the
names are cosmetic. `WriterProperties::set_coerce_types(true)` renames the list
and map fields to the specification's names.
`ParquetWriterExec` sets only the compression on its `WriterProperties`
(`native/core/src/execution/operators/parquet_writer.rs`).
### Steps to reproduce
Write a source with Comet disabled, then write it again natively under each
setting:
```scala
spark.conf.set("spark.sql.session.timeZone", "UTC")
spark.sql("""SELECT timestamp_micros(1700000000123456) AS t,
named_struct('t', timestamp_micros(1700000000654321)) AS s,
CAST(1.25 AS DECIMAL(9,2)) AS d9, array(CAST(1 AS DECIMAL(12,2))) AS
a""").write.parquet("/tmp/src")
spark.conf.set("spark.comet.write.parquet.enabled", "true")
// Spark 4.x. On Spark 3.x the key is
spark.comet.operator.DataWritingCommandExec.allowIncompatible
spark.conf.set("spark.comet.operator.WriteFilesExec.allowIncompatible",
"true")
spark.conf.set("spark.sql.parquet.outputTimestampType", "TIMESTAMP_MILLIS")
spark.read.parquet("/tmp/src").write.parquet("/tmp/millis")
spark.conf.unset("spark.sql.parquet.outputTimestampType")
spark.conf.set("spark.sql.parquet.writeLegacyFormat", "true")
spark.read.parquet("/tmp/src").write.parquet("/tmp/legacy")
```
On `main` at c230bf4e98 with Spark 4.1.3. The `TIMESTAMP_MILLIS` results are
the same on 3.4.3 and 3.5.9, and the `writeLegacyFormat` results on 3.5.9.
| Column | Spark's file | Native file |
| --- | --- | --- |
| `t`, default | `int96` | `int64 (TIMESTAMP(MICROS,true))` |
| `t`, `TIMESTAMP_MILLIS` | `int64 (TIMESTAMP(MILLIS,true))`, reads back as
`22:13:20.123` | `int64 (TIMESTAMP(MICROS,true))`, reads back as
`22:13:20.123456` |
| `d9`, `writeLegacyFormat` | `fixed_len_byte_array(4) (DECIMAL(9,2))` |
`int32 (DECIMAL(9,2))` |
| `a`, `writeLegacyFormat` | `repeated group bag { optional
fixed_len_byte_array(6) array (DECIMAL(12,2)); }` | `repeated group list {
optional int64 item (DECIMAL(12,2)); }` |
| Footer keys | `org.apache.spark.sql.parquet.row.metadata`,
`org.apache.spark.version` | `ARROW:schema` |
### Expected behavior
The native writer writes the physical types and layout these settings ask
for, or falls back when they are set to a value it can't write, and it writes
Spark's schema JSON into the footer. `INT96` is the default, so falling back on
it would stop every native write with a timestamp column. That case needs its
own decision: write `INT96`, or document the difference.
### Additional context
Found in a nested-type audit of the native writers. #5304 covers the other
Parquet writer properties the native writer ignores.
--
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]
