andygrove opened a new issue, #26393:
URL: https://github.com/apache/datafusion/issues/26393
### Describe the bug
With `datafusion.execution.parquet.coerce_int96` set, `Int96Coercer` coerces
an INT96 leaf inside a list only when the list uses the standard three-level
layout, whose repeated group is named `list`. In the legacy list layouts the
leaf stays `Timestamp(Nanosecond)`, so values outside the nanosecond range
(1677-09-21 to 2262-04-11) wrap to wrong timestamps without an error.
These layouts are common in data written for Hive compatibility:
- Spark writes `repeated group bag { optional int96 array; }` when
`spark.sql.parquet.writeLegacyFormat=true`. INT96 is still Spark's default
`spark.sql.parquet.outputTimestampType`.
- Hive writes `repeated group bag { optional int96 array_element; }`.
- Some writers use the two-level form, `repeated int96 array`.
The coercer builds each leaf's Parquet path by pushing a hard-coded
`".list."` between a list field and its child
([schema_coercion.rs#L650-L654](https://github.com/apache/datafusion/blob/cdc2fc041b6a812a24933abc40da6475dbbf5729/datafusion/datasource-parquet/src/schema_coercion.rs#L650-L654)),
then coerces a leaf only if that path is one of the file's INT96 column paths
([L706](https://github.com/apache/datafusion/blob/cdc2fc041b6a812a24933abc40da6475dbbf5729/datafusion/datasource-parquet/src/schema_coercion.rs#L706)).
For `repeated group bag { optional int96 array; }` it looks for
`at.list.array`, but the column path is `at.bag.array`. For the two-level form
the path is `at.array`. #16058, which fixed #15763, added the nested handling
for the standard layout only.
### To Reproduce
```rust
use std::{fs::File, sync::Arc};
use datafusion::arrow::util::pretty::print_batches;
use datafusion::prelude::*;
use parquet::data_type::{Int96, Int96Type};
use parquet::file::{properties::WriterProperties,
writer::SerializedFileWriter};
use parquet::schema::parser::parse_message_type;
fn int96(unix_seconds: i64) -> Int96 {
let nanos_of_day = unix_seconds.rem_euclid(86_400) * 1_000_000_000;
let julian_day = unix_seconds.div_euclid(86_400) + 2_440_588;
let mut v = Int96::new();
v.set_data(nanos_of_day as u32, (nanos_of_day >> 32) as u32, julian_day
as u32);
v
}
/// One record per value, each a one-element list.
fn write(path: &str, message: &str, max_def: i16, values: &[Int96]) {
let schema = Arc::new(parse_message_type(message).unwrap());
let props = Arc::new(WriterProperties::builder().build());
let mut writer = SerializedFileWriter::new(File::create(path).unwrap(),
schema, props).unwrap();
let mut row_group = writer.next_row_group().unwrap();
let mut column = row_group.next_column().unwrap().unwrap();
let defs = vec![max_def; values.len()];
let reps = vec![0; values.len()];
column.typed::<Int96Type>().write_batch(values, Some(&defs),
Some(&reps)).unwrap();
column.close().unwrap();
row_group.close().unwrap();
writer.close().unwrap();
}
#[tokio::main]
async fn main() -> datafusion::error::Result<()> {
// 9999-12-31T23:59:59Z, 1600-01-01T00:00:00Z, 2263-01-01T00:00:00Z,
2020-01-01T00:00:00Z
let values: Vec<Int96> = [253_402_300_799, -11_676_096_000,
9_246_182_400, 1_577_836_800]
.into_iter()
.map(int96)
.collect();
for (name, message, max_def) in [
("standard", "message m { optional group at (LIST) { repeated group
list { optional int96 element; } } }", 3),
("spark_legacy", "message m { optional group at (LIST) { repeated
group bag { optional int96 array; } } }", 3),
("hive", "message m { optional group at (LIST) { repeated group bag
{ optional int96 array_element; } } }", 3),
("two_level", "message m { optional group at (LIST) { repeated int96
array; } }", 2),
] {
let path = format!("/tmp/{name}.parquet");
write(&path, message, max_def, &values);
let ctx = SessionContext::new();
ctx.sql("SET datafusion.execution.parquet.coerce_int96 =
'us'").await?.collect().await?;
ctx.register_parquet("t", &path,
ParquetReadOptions::default()).await?;
let df = ctx.sql("SELECT at FROM t").await?;
println!("{name}: {}", df.schema().field(0).data_type());
print_batches(&df.collect().await?)?;
}
Ok(())
}
```
On DataFusion 55.2.0 with parquet 59.3.0 (the coercer code is unchanged on
`main` at cdc2fc041b):
| Layout | Column type | 9999-12-31T23:59:59 | 1600-01-01T00:00:00 |
2263-01-01T00:00:00 | 2020-01-01T00:00:00 |
| --- | --- | --- | --- | --- | --- |
| standard (`list` / `element`) | `List(Timestamp(µs))` |
`9999-12-31T23:59:59` | `1600-01-01T00:00:00` | `2263-01-01T00:00:00` |
`2020-01-01T00:00:00` |
| Spark legacy (`bag` / `array`) | `List(Timestamp(ns))` |
`1816-03-30T05:56:07.066277376` | `2184-07-20T23:34:33.709551616` |
`1678-06-12T00:25:26.290448384` | `2020-01-01T00:00:00` |
| Hive (`bag` / `array_element`) | `List(Timestamp(ns))` | same as Spark
legacy | same | same | same |
| two-level (`repeated int96 array`) | `List(non-null Timestamp(ns))` | same
as Spark legacy | same | same | same |
### Expected behavior
All four layouts coerce to `Timestamp(Microsecond)` and read the same values
as the standard layout. The coercer could match leaves against the Parquet
schema's own column paths, for example by walking the `SchemaDescriptor`
alongside the Arrow schema, instead of assuming the repeated group is named
`list`.
### Additional context
Found through Apache DataFusion Comet, which reads Spark-written Parquet
with `coerce_int96`: https://github.com/apache/datafusion-comet/issues/7029.
--
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]
