[SPARK-59352][SQL] Validate prefix lengths in the vectorized DELTA_BYTE_ARRAY reader

[holdenk]

What changes were proposed in this pull request?

Validate and fast fail invalid prefix lengths while reading parquet data

Why are the changes needed?

The prefix length in a DELTA_BYTE_ARRAY page is read from the file and was used to copy bytes out of the previous value without validation. Reject negative prefix lengths and prefix lengths larger than the previous value with a ParquetDecodingException, so corrupt pages fail the read deterministically instead of producing malformed values or hard-to-diagnose low-level exceptions.

Does this PR introduce any user-facing change?

No

How was this patch tested?

New unit test

Was this patch authored or co-authored using generative AI tooling?

Yes, claude and cursor

[HyukjinKwon]

0 blocking, 1 non-blocking, 0 nits.
Correct, well-targeted hardening; the validation bound matches every copy site and all prefix-consumption paths are guarded. Approve with one optional test-coverage suggestion.

Suggestions (1)

• sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaByteArrayReader.java:143: The geo read path (readGeoData) guard has no corrupt-prefix test; only readBinary and skipBinary are covered. -- see inline

Verification

Confirmed the guard's bound is right at each consumption site: previous.remaining() equals the bytes read by appendBytes(...) in readValues/skipBinary (from previous.position()) and by previous.get(wkb, 0, prefixLength) in readGeoData, so a valid prefix always passes and any over-read is rejected first. All three loops that call prefixLengthVector.getInt(...) now call checkPrefixLength before copying, and the previous == null case maps to length 0 rather than dereferencing null.

[HyukjinKwon]

The new tests cover the readBinary and skipBinary paths but not this geo path (readGeometry/readGeography), which copies the prefix with previous.get(wkb, 0, prefixLength) rather than the appendBytes(...) the other two use. A corrupt-prefix geometry/geography case would exercise that distinct copy behind the same guard. Consider a geo craft-page test, or parametrizing craftPage/the existing tests over the read method. Non-blocking.

[dongjoon-hyun]

Thank you, @holdenk. The bound (previous.remaining()) matches all three copy sites, and the change looks correct to me.

1. Could you add a boundary case, e.g. prefixLengths = Array(0, 3), suffixes = Array("ab", "")? With on-heap vectors, previous.array() is the whole child byteData, so an off-by-one prefix within capacity silently copied stale bytes before this fix (65536 already failed with AIOOBE). A passing Array(0, 2) case would also pin the valid boundary.
2. Nit: please use 4-space continuation indentation for the throw new ParquetDecodingException( lines, consistent with 701d5e9.
3. Please use the Generated-by: <tool name and version> format in the AI tooling section of the PR description.