fix: preserve Spark errors for Parquet timestamp overflow

[peterxcli]

Which issue does this PR close?

Closes #5517.

Rationale for this change

Parquet TIMESTAMP_MILLIS overflow currently surfaces as a raw Arrow error. Spark reports a file-read SparkException naming the failing file, caused by ArithmeticException("long overflow"), even with ANSI disabled.

What changes are included in this PR?

Emit a Parquet-specific overflow error and carry the original URL-encoded file path through native per-file planning and decoding. Reuse Spark's file-read wrapper and keep LongOverflow unwrapped for casts, consistent with #5457.

How are these changes tested?

The regression puts a healthy file and an overflowing file in one task, including a fused shuffle, and checks exception class, condition, SQLSTATE, exact file path, and arithmetic cause. It covers both overflow signs, dictionary pages, nested timestamps, and both ANSI settings. Native Parquet tests and focused Spark 3.4, 3.5, and 4.1 tests pass.

[andygrove]

This gives users a bare java.lang.ArithmeticException for a failure that happens inside the Parquet reader, but Spark wraps that one. On 4.x, FileScanRDD.hasNext routes the reader's exception through FileDataSourceV2.attachFilePath, and on 3.4 and 3.5 it goes through the NonFatal(e) branch at the end of nextIterator. Both land on QueryExecutionErrors.cannotReadFilesError, so what a Spark user actually sees is a SparkException with FAILED_READ_FILE naming the file, caused by ArithmeticException: long overflow. After this change Comet throws the ArithmeticException on its own with no SparkThrowable anywhere in the chain, so intercept[SparkException] and anything reading getCondition still diverge.

That is also the opposite of what the other scan-path errors in this converter do. ParquetSchemaConvert, ParquetMissingFieldIds and CannotReadFile each wrap in cannotReadFilesError in ShimSparkErrorConverter so the outer exception is a SparkException, and SparkErrorConverterSuite has a test asserting exactly that. The new test does not catch the difference because isLongOverflow searches the whole cause chain, so a missing wrapper is invisible to it.

Would it make sense to wrap this case the same way, using the taskFilePaths already threaded into convertToSparkException to supply the path? One wrinkle is that #5457 raises the same LongOverflow from a cast, where a bare ArithmeticException really is what Spark throws, so the scan case probably needs its own error type or a wrap on the scan side rather than a blanket change in the converter. CometTestBase.checkSparkError would hold the line on class parity if you want the test to cover it.

[rich7420]

@peterxcli +1,LGTM thanks for the patch

[sunchao]

Correctness

The existing checked TIMESTAMP_MILLIS conversion reports an Arrow overflow error. This patch gives Parquet overflow its own structured error, reconstructs Spark's file-read exception in the version-specific shims, and keeps the plain LongOverflow conversion separate. It also recovers task file paths from the injected native plan when a fused scan does not pass them directly. This addresses the missing-wrapper concern in the earlier review.

I compared the reader and exception paths with the maintained Spark 3.5 and 4.0 branches. Both use checked millisecond-to-microsecond multiplication for direct and dictionary timestamp values. Spark wraps the resulting ArithmeticException("long overflow") with cannotReadFilesError, using the current file's URL-encoded path. The error condition is _LEGACY_ERROR_TEMP_2064 on 3.5 and FAILED_READ_FILE.NO_HINT on 4.0. The multiplication is independent of ANSI mode. Timestamp and timestamp-without-time-zone reader paths both use it.

One P2 remains: a multi-file task substitutes the complete comma-separated task file list for the failing file. The inline comment covers the source path and reproduction. The error class and arithmetic cause now match, but the path parameter does not identify the failed file in this ordinary case.

The patch leaves the existing overflow boundary, ancestor-null visibility, nested array/map handling, timezone metadata, and filtered-scan conversion policy in place. The revised tests check positive and negative overflow, actual dictionary pages, both ANSI settings, nested types, and a scan feeding repartition. They now compare structured exception class, condition and SQLSTATE and assert the path and arithmetic cause. Each fixture still contains only one file.

Validation

The CI run is green: 54 successful and 10 skipped head checks. Logs confirm the overflow regression passed in the Spark 3.4, 3.5, 4.0, 4.1 and 4.2 scan jobs. The native job reports 1,441 passed and five skipped tests, including the strengthened nested-overflow test and existing visibility controls. These jobs checked out a253dedb, a merge of 76e0f702 into f69c4c81. All nine authored files and the relevant scan, JNI and test helpers are identical to the reviewed head. The complete trees differ in ten unrelated hash and benchmark files, so this is qualified merge-CI evidence.

Locally, I compiled the exact current converter, full 4.x shim and Comet exception classes against cached Spark 4.0.4 dependencies. The JVM component check reproduced the two-path mismatch. Single-file, explicit-native-path and plain-overflow controls passed, along with checked-multiplication boundary controls. This did not execute a native scan. Maintained Spark 3.4 and 4.1 sources are unavailable, so their CI passes do not constitute maintained-source compatibility verification.

Performance

The successful conversion path still performs one checked multiply per visible value and retains the existing Arrow array conversion. The new native error variant changes the failure result without adding a per-row string or path allocation. Plan parsing and recursive path collection happen after execution fails, so they do not add work to successful batches. I found no new material hot-path overhead in this diff. No performance benchmark was run, and the correctness checks are not a speedup claim.

Design

Separating Parquet overflow from plain arithmetic overflow is the right boundary because the surrounding operation determines Spark's exception wrapper. Reusing QueryExecutionErrors.cannotReadFilesError also keeps version-specific conditions in Spark's own API. The remaining design issue is preserving the identity of the file that raised the error. Once the converter receives only a unit error and a task-wide list, it cannot recover that identity. Attaching the path while the reader still knows the active file would resolve the inline finding without changing successful conversion behavior.

Abstraction & complexity

The two error variants and the small shim cases fit the existing structured-error mechanism. Keeping plan metadata recovery on the failure path avoids adding a second successful-execution traversal. The recursive scan-path helper is small, but it describes task inputs rather than the failing input, which is why it cannot complete the required error context by itself. Beyond that path issue, I found no additional abstraction or simplification change worth requiring in this patch.

[sunchao]

Correctness

[P2] Preserve the failing file path for multi-file tasks

Could we carry the active Parquet file's path with ParquetTimestampOverflow instead of filling it from the whole task list? CometNativeScanExec populates that list from every file in the FilePartition, and normal bin packing can put several files in one task. With one healthy file and one overflowing file, this branch sets Spark's path parameter to file:///data/good.parquet,file:///data/bad.parquet, rather than the single failing file that Spark's FileScanRDD supplies. I reproduced this with the exact current converter and 4.x shim against Spark 4.0.4. The new native error has no path parameter, so it always needs this fallback. The existing tests use one file and cannot catch this case. Please preserve the failing path at the native reader boundary and add a two-file, single-task regression, including the fused shuffle path.

[peterxcli]

@sunchao addressed review! ptal again, thanks!

[rich7420]

Please update the ParquetSource check in try_attach_parquet_reader_filter to recognize this wrapper. Otherwise wrapped scans skip reader-filter pushdown, causing the Rust CI failure.

I reproduced the failure locally; recognizing ParquetErrorContext in that check makes the same test pass.

[sunchao]

Re-reviewed 2b05584f against fad62309. The previous failing-file-path P2 is addressed: the native per-file context supplies the original URI, and the converter no longer substitutes every task input. The updated two-file regression passed in the Spark 3.4, 3.5, 4.0, 4.1 and 4.2 scan jobs; a local probe of the current converter and 4.x shim also preserved the encoded failing URI and the separate bare arithmetic-overflow behavior.

One P2 performance regression remains, already covered by the reader-filter comment. I independently confirmed that the wrapper hides ParquetSource from runtime reader-filter detection. This disables reader pruning for eligible unfiltered probes when the default-off join dynamic-filter option is enabled, while batch filtering remains. The existing native regression fails on this head's exact CI tree, including an Int32-only file. I have not added a duplicate inline comment. The Spark 3.5 shuffle job also has a separate Celeborn test failure; CI is not green.