fix: check nested TIMESTAMP_MILLIS overflow in unfiltered scans

[peterxcli]

Which issue does this PR close?

Partially addresses #5553. The remaining filtered-scan mismatch is tracked in #5739 and depends on apache/datafusion#20871.

Rationale for this change

An unfiltered Parquet read currently returns NULL when a TIMESTAMP_MILLIS value inside a struct, list, or map overflows during conversion to microseconds. Spark raises an overflow error. This change gives unfiltered nested reads the same checked conversion as top-level timestamps.

What changes are included in this PR?

Apply checked conversion recursively, carrying ancestor validity and list/map offsets so null containers and slices cannot expose hidden overflow values. Preserve the original validity of required children and map keys.

Build visibility masks only for subtrees that may need checked timestamp conversion. This avoids scanning unchanged integer-array and map siblings. Add regression coverage for nested overflow, null containers, slices, and buffer sharing, plus a focused conversion benchmark.

Filtered scans retain safe conversion while #5739 tracks the pruning behavior needed for Spark-compatible errors.

How are these changes tested?

After rebasing onto upstream/main at 92ad99e97, the three native timestamp regressions pass. Cargo metadata confirms that both upstream's explode benchmark and this PR's parquet_timestamp_conversion benchmark are registered. Formatting and diff checks pass.

Before this rebase, workspace Clippy with warnings denied and all three Spark 4.1 TIMESTAMP_MILLIS regressions passed. The rebase changed only the placement of the added benchmark registration; the conversion code and tests are identical.

Native regression command, from native with the JDK library directory on DYLD_LIBRARY_PATH:

cargo test -p datafusion-comet --profile ci test_millis_to_micros --lib --offline

The pre-rebase benchmark compares 156fe535 with 1229854c2 using the same harness on macOS arm64 with Rust 1.96.0 and the optimized ci profile. Each before/after pair ran back-to-back with 30 samples, a one-second warmup, and two seconds of measurement. Inputs contain 1,024 parent rows, with every eighth container null and its offset range empty. Timestamp containers have width 32; slices retain the middle half of the parent rows. Input construction is outside the timed section.

Case	Before (µs)	After (µs)
Integer-array sibling, width 8	13.91	1.23
Integer-array sibling, width 1,024	1307.28	1.25
Timestamp list	59.93	59.64
Timestamp list, sliced	40.70	40.50
Timestamp map	60.57	59.87
Timestamp map, sliced	41.06	40.77

These are median times per conversion. The wide sibling avoids a temporary 917,504-entry visibility vector and its bitmap. Timestamp-container medians differ by at most 1.2%; the measurements do not establish query-level throughput.

[sunchao]

Correctness

Reviewed 156fe535 against 7e198439. The previous checked conversion covered only top-level timestamps, so an unfiltered read of overflowing TIMESTAMP_MILLIS inside a struct, list, or map could return NULL. Applying the check recursively matches the maintained Spark 3.5 and 4.0 readers: both direct and dictionary decoding multiply milliseconds with overflow checking before rebasing, independently of ANSI mode. TimestampNTZ uses the checked conversion without calendar rebasing. Their existing differences in permitted LTZ-to-NTZ reads are outside this patch.

The visibility mask is needed because null ancestors and list/map slices can hide backing values. Combining validity before checking, then restoring the original child validity, preserves required children without raising on hidden overflow. I found no additional correctness issue in that handling. Filtered scans retain the existing safe cast, so the remaining mismatch tracked in #5739 is not fixed here.

[P2] Rust CI is blocked by the new test’s reference-counted .clone() calls. Spark 3.5 and 4.0 scan CI passed the expanded overflow test and existing filtered timestamp tests. Both checked out 44d18f09, whose parents are the reviewed base/head and whose tree equals the head. The new Rust visibility test has not run in that CI job: Rust CI fails during Clippy on 15 clone_on_ref_ptr errors in the added test. Please replace those calls with explicit Arc::clone so the Rust tests can run. The author's two passing focused Rust tests are separate evidence. I did not rerun a local build. Maintained Spark 3.4 and 4.1 source branches were unavailable for source verification.

Performance

[P2] Avoid visibility work for unchanged sibling subtrees. The unfiltered check also propagates visibility through selected sibling fields that need no timestamp conversion. For example, adapting a struct containing a millisecond timestamp and a nullable ARRAY<INT> reaches repeated_visibility for the integer array. When a list row is null, it allocates a byte per backing item and constructs an Arrow bitmap, although the unchanged integer cast ignores that visibility. This adds a linear pass to a sibling conversion that otherwise reuses its values buffer. The same concern applies to map children.

Please restrict visibility propagation to subtrees that need checked millisecond-to-microsecond conversion. Add a focused before/after microbenchmark covering a wide unchanged array sibling, nullable timestamp lists/maps, and slices. There are no benchmark results in the PR, and these CI correctness tests establish no scan-throughput result. I am not claiming a measured query slowdown.

Design

Keeping the overflow rule in the schema conversion path makes its relationship to the physical Parquet unit clear. The scan-level option preserves the existing filtered-scan policy while this patch handles unfiltered nesting. Restoring child validity after checking is also preferable to exposing ancestor nulls as new child nulls, which would violate required-field contracts.

The mask should follow the conversion that consumes it. Skipping unchanged sibling subtrees would retain this design while avoiding the performance issue above. Removing the filtered-scan safeguard should remain tied to the pruning and conversion work tracked in #5739.

Abstraction & complexity

One shared repeated_visibility helper is appropriate for list and map offsets. The recursive parent mask avoids separate implementations for every nesting combination, and retaining the original offsets keeps slice reconstruction straightforward. No additional abstraction is needed. The requested narrowing should keep mask construction local to affected conversion subtrees, with a small test showing that unchanged siblings preserve the existing path.

[sunchao]

Correctness

Re-reviewed 4d3e37a4 against 92ad99e9, focusing on the changes since 156fe535. Both previous P2 findings are addressed, and I found no new P1/P2 issue. The reference-counted clones now use Arc::clone. The current Rust CI job passes workspace Clippy with warnings denied and runs 1,190 tests successfully, including all three timestamp regressions. Four other tests are skipped.

The new eligibility check does not remove the visibility needed by a millisecond-to-microsecond conversion. Every supported recursive ancestor containing that conversion has both units, and each selected child is checked again after field matching. Null ancestors and sliced list/map offsets still hide overflowing backing values. Original child validity is restored after conversion, preserving required children and map keys. This retains the checked-overflow behavior verified against the maintained Spark 3.5 and 4.0 readers. The new regression also checks unchanged integer-array and map values, offsets, parent validity and shared value-buffer pointers.

The inspected Spark 3.5/4.0/4.1 scan jobs pass the unfiltered overflow regression and both filtered-scan regressions, with 472/479/487 successful tests and zero failures. There are 15/8/4 cancellations and one ignored test in those jobs. All eight inspected jobs use Comet merge 210e8bfc, whose raw parents are the exact B/H pair and whose entire tree equals H. The scan jobs' native artifact is traced to the matching upload from the Spark 3.5 native/JVM build job. At the 2026-09-08 04:52 UTC refresh, 66 checks passed and 8 were skipped, with no failed or pending checks. I ran no local product tests. Maintained Spark 3.4/4.1 source branches remain unavailable for source comparison.

Performance

The prior sibling-allocation issue is fixed at the right point: the unit check runs before both NullBuffer::union and repeated-mask expansion. An unchanged integer array or map therefore avoids the backing-length boolean vector and bitmap, while preserving its values buffers. For affected timestamp containers, Arrow's null union drops all-valid masks, allowing a full unsliced container to keep the existing no-mask path. Slices and actual null parents still construct the visibility required for checked conversion.

The added benchmark covers the requested narrow and wide unchanged array sibling, nullable timestamp lists/maps, and their slices. The author reports median conversion times of 13.91 to 1.23 microseconds for width 8 and 1307.28 to 1.25 microseconds for width 1,024. Timestamp-container medians stay within 1.2%. These are author-reported component timings on one macOS arm64 runner, not independently reproduced measurements or query-throughput evidence. The measured candidate 1229854c2 has byte-identical conversion code, tests and benchmark source to H. The rebase only adds the upstream explode registration beside this benchmark in Cargo.toml. The benchmark CI check compiles and lints the benchmarks but does not measure them.

Design

The subtree predicate is deliberately conservative until struct fields are matched by name or ID. Rechecking each selected child then narrows the work without moving visibility handling into the schema matcher. The existing filtered-scan policy is unchanged. This addresses the earlier request without adding a second conversion path or changing error semantics.

Abstraction & complexity

One small type-tree predicate is proportionate to this change. It avoids introducing cached conversion plans or another masking abstraction. The added native test checks observable buffer sharing and semantic preservation, while the benchmark measures the conversion boundary directly with input construction outside the timed section. No further change is requested from this follow-up.