fix: prevent silent overflow when reading Parquet TIMESTAMP_MILLIS values

[peterxcli]

Rationale for this change

Spark's vectorized Parquet reader converts TIMESTAMP_MILLIS values with the checked millisToMicros (Math.multiplyExact), so overflow throws — independent of ANSI mode. Comet's reader used an unchecked v * 1000, silently wrapping the value. This change makes the conversion checked on the real Parquet read path, and delegates to Arrow kernels where Arrow's behavior matches Spark.

What changes are included in this PR?

• Add a checked Timestamp(Millisecond) -> Timestamp(Microsecond) conversion in parquet_convert_array, the path the native Parquet scan actually takes. Overflow now raises an error instead of silently wrapping, matching Spark's millisToMicros in every eval mode.
• Remove the millis-to-micros arm from array_with_timezone and its unit test: Spark logical timestamps are always microseconds (serde.rs, Utils.scala), so cast_array can never see a millisecond input. The arm was dead code, and the conversion is now implemented and tested where it actually runs.
• Reject a microsecond physical timestamp with a top-level millisecond Spark logical target during planning (CometCastColumnExpr::try_new), since Spark read schemas represent logical timestamps in microseconds. The error names it as a Comet bug and suggests spark.comet.scan.enabled=false as a workaround.
• Use Arrow's cast for Int32 -> Date32 in date_from_unix_date (a zero-copy reinterpret in arrow-cast), with DEFAULT_CAST_OPTIONS for scalar/array consistency.
• Add a Spark-level regression test (ParquetReadSuite."TIMESTAMP_MILLIS overflow fails in native scan") covering TimestampType and TimestampNTZType, positive and negative overflow, dictionary on/off, and ANSI on/off, asserting both Spark and Comet report overflow.

Note: an earlier revision also reworked cast_date_to_timestamp in temporal.rs; that hunk was dropped to avoid conflicting with #5457, which rewrites the same function as a safety fix.

Follow-up: #5517 tracks surfacing the overflow as a Spark-faithful ArithmeticException("long overflow") instead of a raw Arrow compute error.

How are these changes tested?

• New ParquetReadSuite regression test described above.
• cargo test --manifest-path native/Cargo.toml -p datafusion-comet-spark-expr --lib
• cargo test --manifest-path native/Cargo.toml -p datafusion-comet --lib parquet::cast_column::tests
• 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.

[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.

[andygrove]

Thanks for the rework. Rejecting the pair outright is a better answer than either version of the division fix, and the ParquetSchemaConverter link makes the invariant checkable. CI is fully green, including Iceberg 1.11, which is the PR-gated Iceberg job, so the scan paths that matter here did get exercised.

A few things I confirmed independently so you do not have to re-derive them.

Your reachability claim is correct. to_arrow_datatype at native/core/src/execution/serde.rs:93-96 maps Spark timestamps to microseconds only, and the JVM side agrees at Utils.scala:157-160. IcebergScanExec builds its schema from the same convert_spark_types_to_arrow_schema at planner.rs:1651. A millisecond logical target genuinely cannot occur.

Deleting the three tests costs no coverage. types_differ_only_in_field_names returns false for a flat (Timestamp, Timestamp) pair, so relabel_array was never on the timestamp path, and its List, Map, and Struct coverage is intact.

temporal.rs and date_from_unix_date.rs are now consistent on DEFAULT_CAST_OPTIONS, and both look right to me.

Nothing needed on docs. No serde or expression registration changed, so the compatibility pages and expressions.md stay as they are.

The one thing I would like to sort out before this merges is whether the overflow fix actually reaches the Parquet reader. I do not think it does. Details inline.

[peterxcli]

@andygrove thanks for another round of review, addressed all of your review. please take another look. TIA!

The overflow fix may not reach the actual Parquet reader.

Moved the checked millis -> micros conversion into parquet_convert_array, the real Parquet reader path. It now uses try_unary with mul_checked(1_000), so overflow returns an error instead of null.

Please verify the fix end to end against Spark behavior.

Added a native Parquet scan regression covering TimestampType and TimestampNTZType, positive and negative overflow, dictionary on/off, and ANSI on/off. It verifies both Spark and Comet report overflow.

Is the millis→micros arm in array_with_timezone reachable?

Confirmed it is unreachable for Parquet reads. Removed the dead arm, its imports, and its unit test. The conversion is now tested where it actually runs.

The millisecond-target rejection only checks top-level timestamps.

Narrowed the comment to explicitly state that the guard applies to top-level timestamp columns, so it does not imply nested timestamp validation.

[andygrove]

Note on this review: this was generated by an LLM (Claude Code) at my request while I worked through a review backlog. I have not verified the individual findings myself. Please treat everything below as suggestions to evaluate rather than as authoritative review feedback, and push back on anything that is wrong or already handled.

Turning the silent v * 1000 wrap into a checked multiply is the important part here, and linking to ParquetVectorUpdaterFactory and SparkDateTimeUtils.millisToMicros in the comment makes it easy to verify against Spark. Deleting the duplicate cast_timestamp_micros_to_millis_* helpers is a good cleanup.

Several things.

This conflicts directly with #5457

#5457 rewrites cast_date_to_timestamp in temporal.rs as well, replacing the chrono region-zone path with fixed-offset arithmetic and a plan-time fallback. This PR keeps that path and adds an Arrow fast path for the NTZ case. They cannot both land as written. Worth coordinating with @sunchao on which goes first.

date_from_unix_date loses a zero-copy path

The old code was:

Date32Array::new(int_array.values().clone(), int_array.nulls().cloned())

which is O(1): cloning an Arrow Buffer is a refcount bump. The new code calls cast_with_options(arr, &DataType::Date32, ...). Depending on the Arrow version, Int32 -> Date32 may or may not take a reinterpret fast path. If it does not, this turns a free operation into a full array copy on every batch.

Could you confirm which it is? If Arrow copies, the old code was better and I would keep it. The general principle of delegating to Arrow is right, but not when the hand-written version is asymptotically cheaper.

Separately, the new version accepts any type Arrow can cast to Date32, where the old one rejected anything that was not Int32. Spark's DateFromUnixDate only accepts IntegerType, so this should not matter, but it does remove a guard against a planner bug. Worth a comment saying the input type is guaranteed by the serde.

Removing the millis-to-micros arm from array_with_timezone

That arm is gone, so the same conversion in array_with_timezone now falls to Err("Not supported"). array_with_timezone is called from cast_array, not just the Parquet reader. Is there any path where a Comet CAST sees a Timestamp(Millisecond, None) input, for example from an Iceberg scan or a UDF return value? If there is, this changes a working conversion into an error. If there is not, it would be good to say so in the commit message, since the deletion looks unrelated to the Parquet fix at first glance.

What does the new overflow error look like to a user?

try_unary(... mul_checked ...) produces an ArrowError, which as far as I know surfaces as CometNativeException rather than the ArithmeticException("long overflow") Spark's Math.multiplyExact throws. #5169 is doing exactly this kind of error-fidelity work for the decimal paths. Is it worth making this one a typed SparkError from the start, rather than adding it to the list of raw Arrow errors that need converting later?

try_new returning a plan error

The new microsecond-physical to millisecond-target check returns DataFusionError::Plan from inside the schema adapter. If a file ever does present that shape, the query fails rather than falling back to Spark. The comment argues Spark's schema converter makes it unreachable, and the Spark link supports that. Could the message say what a user should do if they somehow hit it, or would it be better to fall back rather than fail?

[peterxcli]

@andygrove thanks for the review! Addressed three points and pushing back on two, details below.

This conflicts directly with #5457

Good catch. Since #5457 rewrites cast_date_to_timestamp as a safety fix and my change there was behavior-neutral (Arrow's NTZ arm compiles to the same unchecked days * MICROS_PER_DAY), I dropped the temporal.rs hunk from this PR entirely. #5457 can own that function; this PR now touches only the Parquet overflow paths, which also fits the fix: title better.

date_from_unix_date loses a zero-copy path

Confirmed on arrow-cast 58.4.0: (Int32, Date32) takes cast_reinterpret_arrays::<Int32Type, Date32Type> (mod.rs#L1631), which is a zero-copy buffer reinterpret — same cost as the old Date32Array::new, no regression. (Also matches your July 31 review note.) Added a comment documenting that the input is guaranteed Int32 by Signature::exact and the serde.

Removing the millis-to-micros arm from array_with_timezone

This was settled in the previous round — the arm was removed at your suggestion after we verified it unreachable: Spark logical timestamps are exclusively microseconds (serde.rs:95-98, Utils.scala:157-160), and both Iceberg scans and UDF return types flow through the same schema conversion, so cast_array can never see a millisecond input. I updated the PR description to note the rationale so the deletion doesn't look unrelated in the squashed commit.

What does the new overflow error look like to a user?

I'd rather not convert it in this PR. Spark's exception here is an untyped java.lang.ArithmeticException("long overflow") from Math.multiplyExact — there is no Spark error class to map to — and reusing SparkError::ArithmeticOverflow would emit "set spark.sql.ansi.enabled to false to bypass this error", which is actively misleading since this path throws regardless of ANSI (the new test pins that with ANSI both on and off). Proper fidelity needs a new native variant plus ShimSparkErrorConverter mappings per Spark version, which belongs with the #5169 error-fidelity effort. Filed #5517 to track it.

try_new returning a plan error

Fallback isn't reachable from there — by the time the native schema adapter runs, the JVM has already committed to the native scan. I extended the message to state it indicates a Comet bug, link the issue tracker, and suggest spark.comet.scan.enabled=false as a workaround.

[sunchao]

Worth coordinating with @sunchao on which goes first.

Feel free to merge this first!

[sunchao]

Reviewed afa3673 with five specialist agents; no blocking correctness findings. The inline dictionary fixture note does not block approval.

Local validation passed: 82 native Parquet tests, 643 native expression tests, and the Spark 4.1.3 overflow regression. A temporary 16-row variant also passed after verifying dictionary pages for both timestamp types and both signs. Native validation excluded HDFS; the full Spark suite was not run locally.

[sunchao]

Re-reviewed 5c8de42 with five specialist agents. One additional regression is described inline; the earlier dictionary fixture issue is fixed.

Local validation: 82 native Parquet tests, 660 native expression tests, and the Spark 4.1.3 overflow regression passed. The additional filtered-read regression passes with base c067e4e and fails on this head. Native builds excluded HDFS; the full Spark suite was not rerun locally.

[sunchao]

[P2] Preserve timestamp pruning before checked conversion

With a TimestampType column stored as TIMESTAMP_MILLIS and containing 9223372036854776 milliseconds, WHERE ts < TIMESTAMP '1970-01-01 00:00:00' returns no rows in Spark 4.1.3 and the base native build (c067e4e), but 5c8de42 throws Overflow happened on: 9223372036854776 * 1000. I reproduced this with plain/dictionary encoding, ANSI on/off, and Comet row-filter pushdown on/off: all eight Spark/base cases succeed and all eight head cases fail.

The existing CometCastColumnExpr prevents DataFusion from recognizing the timestamp statistics predicate, so Comet converts values from row groups Spark skips. This new error turns that pruning limitation into a query failure. Please preserve pruning before the checked conversion and add this filtered case as a regression test, while retaining overflow errors for values actually read.

[peterxcli]

Confirmed and fixed — thanks for the thorough repro. The predicate's column was wrapped in CometCastColumnExpr, which is opaque to DataFusion's pruning analyzer, so the row group Spark prunes from millisecond statistics was being read and converted. The fix rewrites predicate comparisons over a TIMESTAMP_MILLIS file column into the millisecond domain (exact integer rescaling of the literal, like Spark's ParquetFilters pushing predicates in the file's physical unit), plus IS NULL/IS NOT NULL unwrapping. Pruning works again, predicate evaluation never converts file values, and the scan output conversion stays checked, so values actually read still fail like millisToMicros.

Added your filtered case as a regression test over dictionary × ANSI × rowFilterPushdown (8 configs), and native unit tests pinning the rounding table for all six comparison operators, both operand orders, and negative/sub-millisecond literals.

One deliberate divergence to note: with row-filter pushdown on and a non-pruned row group, rows the filter discards are no longer converted, so Comet can succeed where Spark (which converts the whole row group during decode) throws — the benign direction of "errors only for values actually read." (d465192)