fix: coerce int96 resolution inside of list, struct, and map types

[mbutrovich]

Which issue does this PR close?

• Closes Parquet: coerce_int96 does not work for int96 in nested types #15763.

Rationale for this change

What changes are included in this PR?

Stack-based DFS inspired by arrow-rs' Schema::normalize https://github.com/apache/arrow-rs/blob/1f15130414bdfc01c8989ec95702655bf553c5c5/arrow-schema/src/schema.rs#L464

We don't have a max depth argument here because you either complete the process or you fail. Anything else results in a schema that might yield unexpected data. I'm open to discussion on if we should include the max depth logic with some sort of reasonable cutoff, at which point we would have to return an error.

Are these changes tested?

New tests using Parquet schemas known to exercise the issue.

Are there any user-facing changes?

No.

[mbutrovich]

I'm going to punt on other nested types for this PR. If this approach is good, it should be straightforward to add other nested types.

[kazuyukitanimura]

Thanks, looks good, I will spend a little more time to make sure I understand this.

[alamb]

What seems strange to me about this PR is that DataFusion code is doing parquet specific type coercion when it otherwise uses the Arrow types

What I would expect is that DataFusion says "I want the data from parquet back as List(TimestampMicros)" and then he parquet crate would handle doing the nested conversion in the parquet schema

I am actually surprised it doesn't already work today, which is why I was hoping we could get an example parquet file to see what was going wrong

[mbutrovich]

What seems strange to me about this PR is that DataFusion code is doing parquet specific type coercion when it otherwise uses the Arrow types

We only want to coerce Timestamp(nanos) that originate as int96, so we need to reference the underlying Parquet schema. We don't want to touch Timestamps that aren't int96, hence the test coerce_int96_to_resolution_with_mixed_timestamps. Like I mention in the test, it's not actually clear to me that any system would ever write a Parquet file like that. For example, in Spark you set a config that determines how all timestamps are written. However, I don't think there's anything in the Parquet spec that prevents mixing timestamp type in a single file so I'd rather be conservative in the logic.

What I would expect is that DataFusion says "I want the data from parquet back as List(TimestampMicros)" and then he parquet crate would handle doing the nested conversion in the parquet schema

Maybe I'm misunderstanding, but that's essentially what's happening here. The API to request different type from the Parquet crate is by providing an Arrow schema with the desired types. This function builds that schema, but only modifies fields that are Timestamp(nanos) that are stored in the file as int96. This similar to the coercion functions that convert Utf8 fields to Utf8View.

I am actually surprised it doesn't already work today, which is why I was hoping we could get an example parquet file to see what was going wrong

It is working that way at the moment, but the type transformation just isn't digging into nested types to find nested int96 fields.

[mbutrovich]

Tested this PR with @andygrove's Comet branch for DF 48 and confirmed that we no longer need to set generateArray and generateStruct to false in CometFuzzTestSuite's "Parquet temporal types written as INT96" test.
https://github.com/apache/datafusion-comet/blob/main/spark/src/test/scala/org/apache/comet/CometFuzzTestSuite.scala#L227

[alamb]

Maybe I'm misunderstanding, but that's essentially what's happening here. The API to request different type from the Parquet crate is by providing an Arrow schema with the desired types. This function builds that schema, but only modifies fields that are Timestamp(nanos) that are stored in the file as int96. This similar to the coercion functions that convert Utf8 fields to Utf8View.

What I am reacting to is the fact that DataFusion code is directly manipulating the Parquet schema classes / structs, rather than, for example, setting some flag in the parquet read options and letting code in the parquet crate do it.

I can see your point however, how this transformation needs to have access to the parquet schema 🤔

[andygrove]

It is working that way at the moment, but the type transformation just isn't digging into nested types to find nested int96 fields.

My understanding is that PR is extending DataFusion's existing Parquet INT96 coercion to be recursive rather than just looking at the top-level types. It doesn't seem to be a change in overall approach.

Changes LGTM.

[andygrove]

Thanks @mbutrovich!

[andygrove]

@alamb I'd like to go ahead and merge this one if there are no objections