fix: prevent silent overflow when reading Parquet TIMESTAMP_MILLIS values

[peterxcli]

Which issue does this PR close?

Closes #5090.

Rationale for this change

Several native temporal conversions duplicate Arrow kernels with hand-written per-row loops. This change delegates conversions to Arrow where its behavior matches Spark and keeps Spark-specific handling where it does not. It also makes millisecond-to-microsecond overflow return an error instead of silently wrapping, matching Spark's Parquet reader.

What changes are included in this PR?

• Use Arrow casts for Int32-to-Date32, Date32-to-Timestamp(Microsecond) without a timezone, and Timestamp(Millisecond)-to-Timestamp(Microsecond).
• Keep the custom timezone-aware Date32-to-Timestamp path for Spark's DST overlap and gap behavior.
• Use DEFAULT_CAST_OPTIONS consistently, so millisecond-to-microsecond overflow errors in every eval mode like Spark's checked millisToMicros conversion.
• Reject a microsecond physical timestamp with a millisecond Spark logical target during planning because Spark read schemas represent logical timestamps in microseconds.
• Add focused coverage for overflow, nulls, timezone behavior, and the invalid millisecond logical target.

How are these changes tested?

• cargo test --manifest-path native/Cargo.toml -p datafusion-comet-spark-expr --lib (595 passed)
• cargo test --manifest-path native/Cargo.toml -p datafusion-comet --lib parquet::cast_column::tests (5 passed)
• cargo clippy --manifest-path native/Cargo.toml -p datafusion-comet-spark-expr -p datafusion-comet --lib --tests -- -D warnings
• cargo fmt --manifest-path native/Cargo.toml --all -- --check

[andygrove]

Thanks for working on this. Replacing the hand-written loops with Arrow kernels is a good direction, and I especially like that you spotted Arrow's timezone adjustment in cast_with_options and worked around it with a metadata-only relabel. The comment explaining why the relabel has to happen first is really helpful, and the new +07:00 test cases genuinely guard it, since without the relabel they would be off by seven hours.

I have a few things I would like to work through before this goes in.

1. Micros to millis: Arrow truncates, Spark floors

native/core/src/parquet/cast_column.rs

Arrow's timestamp downscale is plain integer division (time_array.unary(|o| o / divisor) in arrow-cast), so it truncates toward zero. Spark's SparkDateTimeUtils.microsToMillis is Math.floorDiv(micros, MICROS_PER_MILLIS), and there is a comment there specifically about pre-1970 timestamps needing that adjustment. So for -1_500_001 micros Spark produces -1501 and we produce -1500.

The old unary(|v| v / 1000) had the same behavior, so this is not something the PR introduces. My concern is that the new test asserts -1_500_001 -> -1500, which bakes the divergence into a test as though it were intended. Moving to Arrow's cast also takes away the easy fix, since the hand-written closure could have simply become v.div_euclid(1000).

Would it make sense to keep a small kernel for this one case so the floor semantics can be matched? If you would rather keep the Arrow cast, could we file an issue and reference it next to the negative assertion so the expected value does not read as deliberate?

2. CastOptions::default() is safe: true

native/spark-expr/src/utils.rs

Could the millis to micros cast use safe: false instead of CastOptions::default()? Arrow branches on that flag for the upscale path. With safe: true it takes unary_opt(|o| o.checked_mul(mul)), which allocates a fresh null buffer and checks every element. With safe: false it takes try_unary, which reuses the input null buffer. So the default is doing an extra pass per batch compared to the unary this replaces.

There is a behavior argument too. Spark's millisToMicros is Math.multiplyExact and throws on overflow, so raising an error is closer to Spark than silently producing NULL. It would also line up with DataFusion's DEFAULT_CAST_OPTIONS, which is what cast_column.rs uses in this same PR.

3. Inconsistent cast options in date_from_unix_date

native/spark-expr/src/datetime_funcs/date_from_unix_date.rs

The two branches end up with different options. The array path gets CastOptions::default(), which is safe: true, while scalar.cast_to(...) resolves to cast_to_with_options(target, &DEFAULT_CAST_OPTIONS), which is safe: false. Int32 -> Date32 goes through cast_reinterpret_arrays, so it cannot fail either way today. But if the Signature::exact(vec![Int32]) is ever widened, the two paths would quietly disagree, one nulling and one erroring. Worth making them match while it is cheap to do?

For what it is worth, I checked the two things in this file that looked riskiest and both are fine. Int32 -> Date32 stays zero-copy, so there is no regression versus the manual Date32Array::new. And dropping the explicit ScalarValue::Null arm is safe, because can_cast_types has (Null, _) => true and the cast returns new_null_array.

4. Test coverage in cast_column.rs

Both evaluate tests moved from Timestamp(ms, None) to Timestamp(ms, Some("+07:00")), and the three deleted unit tests were the ones covering target_tz = None. I think that leaves the no-timezone case uncovered, which is the branch where relabel_array early-returns because the types already match. Could one of these keep a None target?

It would also be good to have a case where the input array already carries a timezone, say Timestamp(us, Some("UTC")) to Timestamp(ms, Some("America/New_York")). relabel_array overwrites whatever timezone the input has, and "relabel, do not shift" is exactly the property the workaround is protecting, so having that pinned down would help.

A note, not a request

In cast_date_to_timestamp, Arrow's Date32 -> Timestamp(us) is a plain unary(|x| (x as i64) * MICROSECONDS_IN_DAY) that ignores the safe flag, so a large Date32 wraps silently. Spark's daysToMicros uses Math.multiplyExact. That is identical to the old (d as i64) * 86_400 * 1_000_000, so I am not asking for anything here. I only wanted to note it so it is not mistaken for something the Arrow cast fixed.

[peterxcli]

@andygrove thanks for the review!

1. Micros → millis rounding

Changed this one conversion back to a small custom kernel because Arrow truncates negative values toward zero, while Spark uses floor division.

• Before: -1_500_001 / 1_000 = -1500
• Now: -1_500_001.div_euclid(1_000) = -1501

2. Millis → micros cast options for array_with_timezone in utils.rs

Changed to use DEFAULT_CAST_OPTIONS.

I also added a regression assertion using i64::MAX to confirm overflow returns an error from spark: https://github.com/apache/spark/blob/v4.2.0/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/DateTimeUtilsSuite.scala#L969-L972

3. date_from_unix_date option consistency

Changed to use DEFAULT_CAST_OPTIONS.

4. cast_column.rs test coverage

Expanded the array test to cover all three relevant timezone layouts:

• No timezone → no timezone
• No timezone → +07:00
• UTC → America/New_York

[andygrove]

Thanks for the quick turnaround. All four points from the last round look addressed, and pinning the new test values to Spark's own DateTimeUtilsSuite example is a nice touch. I ran the touched tests locally and they pass, and I re-checked the claims against arrow-cast 58.4.0 and Spark master.

A couple of things I confirmed so nobody has to re-derive them. SparkDateTimeUtils.microsToMillis really is Math.floorDiv, with a comment about pre-1970 timestamps, so div_euclid is the right call. millisToMicros really is Math.multiplyExact, so safe: false in utils.rs is the Spark-faithful choice. And there is no performance regression anywhere: the millis-to-micros upscale reinterprets to Int64 zero-copy and then uses try_unary, which reuses the input null buffer, so it is the same single allocation the old unary did. Date32 -> Timestamp(us) is one unary plus a same-type cast that early-returns, and Int32 -> Date32 is still cast_reinterpret_arrays. I also grepped for other micros-to-millis sites that might share the truncation bug, and the only ones are the two this PR fixes.

One thing on packaging that I would like to sort out before this merges. The div_euclid change is a real behavior fix rather than a refactor, since any pre-1970 timestamp with a sub-second component now yields a different value than before. Our changelog is generated from PR titles, so as refactor: this lands with no signal that timestamp results changed. Could we retitle to fix:, or split the floor fix out so it gets its own entry? Either way it would help to have the divergence tracked in an issue we can link from the code comment.

The description needs a refresh too. The first bullet still says we use Arrow casts for the Parquet micros-to-millis conversion, and that is the one case that ended up keeping a hand-written kernel. It would also be good to mention that millis-to-micros overflow went from silently wrapping to raising an error, since that is user visible as well.

Nothing needed on docs. No serde or expression registration changed, so the compatibility pages and expressions.md stay as they are. date_from_unix_date already has good SQL test coverage including the Spark min and max date boundaries, and both of its branches are now consistent, so I have nothing to raise there.

The rest of my comments are inline.

[andygrove]

Do you know whether this arm is reachable from a Spark or Iceberg query? I traced it back to the DataFusion 52 Iceberg migration and could not find anything on the JVM side that asks for a Timestamp(Millisecond) logical field, so I could not tell whether the floor fix here is observable end to end.

If it is reachable, a Spark-level test reading a pre-1970 sub-second timestamp would be worth having, since that is the case that was wrong before. If it is only defensive for a schema shape we do not currently generate, a comment saying that would be just as useful.

[andygrove]

The array test above now loops over three timezone layouts, which is a good improvement. This scalar test is still only target_tz = None, and test_cast_timestamp_micros_to_millis_scalar, which got deleted, was the one checking that a target timezone lands on the scalar.

Could this test loop over the same timezone cases as the array one, so we pin down that the resulting ScalarValue::TimestampMillisecond actually carries target_tz?

[andygrove]

This one is still CastOptions::default() while the other two sites moved to DEFAULT_CAST_OPTIONS. Could we make it three for three?

I know it makes no behavioral difference today, since Arrow's Date32 -> Timestamp(us) arm is a plain unary that ignores the safe flag. But leaving one call site on safe: true reads like it was a deliberate choice, and if Arrow ever adds the overflow check then safe: false is what we want anyway, because Spark's daysToMicros goes through Math.multiplyExact.

[andygrove]

I want to make sure I understand what the new is_err() assertion in the test is locking in. Over in cast.rs, array_with_timezone is called on line 262, one line before native_cast_options is built with safe: !matches!(eval_mode, EvalMode::Ansi). So this cast now hard errors on overflow even under try_cast or legacy mode, where the rest of cast_array would produce NULL.

My guess is that this is fine, because the arm looks like it serves Iceberg schema adaptation rather than a user-written cast, and Spark's Parquet reader calls millisToMicros unconditionally and throws regardless of ANSI. If that is right, could we add a short comment saying so? It would stop someone from later "fixing" this to respect eval mode.

[peterxcli]

sorry posted comment to wrong pr, nvm

[peterxcli]

@andygrove thanks for the review! review change is pushed. please take another look, thanks!

1. div_euclid is a behavior fix, not a refactor.

Retitled the PR. The final patch rejects that invalid read-schema pair instead of using div_euclid.

2. Refresh the PR description and mention overflow.

Updated it to describe the planning rejection and that millis→micros overflow now errors instead of wrapping.

3. Is micros→millis reachable from Spark or Iceberg?

No. Spark read schemas use microsecond logical timestamps. Construction now returns DataFusionError::Plan, with a Spark source link.

4. Keep scalar target-timezone coverage.

The pair is now rejected before scalar or array evaluation. The planning test covers timezone-free and timezone-bearing fields.

5. Use DEFAULT_CAST_OPTIONS at the third site.

Done. The timezone-free Date32 -> Timestamp path now uses DEFAULT_CAST_OPTIONS.

6. Explain why millis→micros always errors on overflow.

Added links to Spark’s Parquet call site and checked millisToMicros implementation.