[
https://issues.apache.org/jira/browse/SPARK-56907?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Ismaël Mejía updated SPARK-56907:
---------------------------------
Description:
h2. Summary
Reduce per-value overhead in the {{skipBinary}} path of the
DELTA_LENGTH_BYTE_ARRAY vectorized Parquet reader
({{VectorizedDeltaLengthByteArrayReader}}) and harden it against malformed
input.
{{skipBinary}} replaces the per-value skip loop (N separate {{in.skip()}}
calls) with a single bulk skip: lengths are summed into a {{long}}, each length
is validated as non-negative, and the skip is performed via
{{in.skipFully(...)}} wrapped in a {{ParquetDecodingException}}. Besides the
speedup, this fixes three latent issues that a naive bulk form would introduce:
* a non-terminating CPU spin when the stream is exhausted
({{ByteBufferInputStream.skip}} returns -1), which would ignore killTask and
hang the stage;
* silent row drift when a negative length cancels positive ones in the same
batch;
* {{int}} overflow of the summed length.
An earlier revision also rewrote {{readBinary}} (bulk slice) and
{{readGeoData}}; those were reverted to the master implementations after
review, since the {{readBinary}} wall-clock gain was within run-to-run noise.
The reverted {{readBinary}} measures identical to the baseline, confirming it
is a no-op on that path.
h2. Benchmarks
GHA on AMD EPYC 7763, Best Time in ms (lower is better),
DELTA_LENGTH_BYTE_ARRAY skipBinary:
||payloadLen||JDK17 Base||JDK17 PR||JDK21 Base||JDK21 PR||JDK25 Base||JDK25 PR||
|8|7|3|7|3|6|3|
|32|7|3|7|3|6|3|
|128|7|3|7|3|6|3|
|512|7|3|7|3|6|3|
Consistent 2.00x-2.33x improvement across all payload sizes and JDKs.
h2. Tests
Existing {{ParquetDeltaLengthByteArrayEncodingSuite}} and
{{ParquetEncodingSuite}} pass. Added {{skipBinary fails cleanly when the data
region is truncated}}, which decodes a valid length header over a truncated
data region and asserts {{skipBinary}} raises {{ParquetDecodingException}}
instead of spinning on an exhausted stream.
was:
Reduces object allocation in the DELTA_LENGTH_BYTE_ARRAY vectorized Parquet
reader ({{VectorizedDeltaLengthByteArrayReader}}) by applying three targeted
changes:
* *readBinary*: Replace per-value {{in.slice(length)}} (one ByteBuffer
allocation per value) with a single bulk {{in.slice(totalDataLen)}} that reads
the entire batch at once. Individual values are then written to the column
vector via {{putByteArray}} from the shared backing array.
* *skipBinary*: Replace the per-value skip loop (N separate {{in.skip()}}
calls) with a single bulk skip by summing all value lengths upfront.
* *readGeoData*: Remove the {{ByteBuffer.wrap()}} + {{ByteBufferOutputWriter}}
indirection per value and call {{putByteArray}} directly.
h3. Benchmark Results (GHA, AMD EPYC 7763, PR vs committed upstream baseline,
Best Time in ms, speedup = base/PR)
*skipBinary* (consistent win):
||Case||JDK 17||JDK 21||JDK 25||
|payloadLen=8|2.33x|1.75x|2.00x|
|payloadLen=32|2.33x|1.75x|2.00x|
|payloadLen=128|2.33x|1.75x|2.00x|
|payloadLen=512|2.33x|1.75x|2.00x|
*readBinary* (allocation-bound, wall-clock roughly neutral / within noise):
||Case||JDK 17||JDK 21||JDK 25||
|payloadLen=8|1.13x|0.94x|1.14x|
|payloadLen=32|1.13x|0.94x|1.07x|
|payloadLen=128|1.06x|0.79x|1.00x|
|payloadLen=512|1.05x|0.95x|1.00x|
{{skipBinary}} is the primary measurable win (1.75x-2.33x across all
sizes/JDKs) from replacing N per-value stream skips with a single bulk skip.
{{readBinary}} removes N-1 ByteBuffer allocations per batch (reduced GC
pressure), but Best Time is roughly neutral and within run-to-run variance at
these sizes (small gains on JDK 17/25, noise-level dips on JDK 21; 16-24 ms
measurements at 1 ms integer granularity).
Full committed results: JDK 17
https://github.com/iemejia/spark/actions/runs/31322679242 , JDK 21
https://github.com/iemejia/spark/actions/runs/31322682640 , JDK 25
https://github.com/iemejia/spark/actions/runs/31322686206
PR: https://github.com/apache/spark/pull/55932
Parent issue: https://github.com/apache/spark/issues/56011
Summary: Reduce per-value skip in DELTA_LENGTH_BYTE_ARRAY Parquet
vectorized reader (was: Reduce per-value allocation in DELTA_LENGTH_BYTE_ARRAY
Parquet vectorized reader)
> Reduce per-value skip in DELTA_LENGTH_BYTE_ARRAY Parquet vectorized reader
> --------------------------------------------------------------------------
>
> Key: SPARK-56907
> URL: https://issues.apache.org/jira/browse/SPARK-56907
> Project: Spark
> Issue Type: Sub-task
> Components: SQL
> Affects Versions: 5.0.0
> Reporter: Ismaël Mejía
> Priority: Major
> Labels: pull-request-available
>
> h2. Summary
> Reduce per-value overhead in the {{skipBinary}} path of the
> DELTA_LENGTH_BYTE_ARRAY vectorized Parquet reader
> ({{VectorizedDeltaLengthByteArrayReader}}) and harden it against malformed
> input.
> {{skipBinary}} replaces the per-value skip loop (N separate {{in.skip()}}
> calls) with a single bulk skip: lengths are summed into a {{long}}, each
> length is validated as non-negative, and the skip is performed via
> {{in.skipFully(...)}} wrapped in a {{ParquetDecodingException}}. Besides the
> speedup, this fixes three latent issues that a naive bulk form would
> introduce:
> * a non-terminating CPU spin when the stream is exhausted
> ({{ByteBufferInputStream.skip}} returns -1), which would ignore killTask and
> hang the stage;
> * silent row drift when a negative length cancels positive ones in the same
> batch;
> * {{int}} overflow of the summed length.
> An earlier revision also rewrote {{readBinary}} (bulk slice) and
> {{readGeoData}}; those were reverted to the master implementations after
> review, since the {{readBinary}} wall-clock gain was within run-to-run noise.
> The reverted {{readBinary}} measures identical to the baseline, confirming it
> is a no-op on that path.
> h2. Benchmarks
> GHA on AMD EPYC 7763, Best Time in ms (lower is better),
> DELTA_LENGTH_BYTE_ARRAY skipBinary:
> ||payloadLen||JDK17 Base||JDK17 PR||JDK21 Base||JDK21 PR||JDK25 Base||JDK25
> PR||
> |8|7|3|7|3|6|3|
> |32|7|3|7|3|6|3|
> |128|7|3|7|3|6|3|
> |512|7|3|7|3|6|3|
> Consistent 2.00x-2.33x improvement across all payload sizes and JDKs.
> h2. Tests
> Existing {{ParquetDeltaLengthByteArrayEncodingSuite}} and
> {{ParquetEncodingSuite}} pass. Added {{skipBinary fails cleanly when the data
> region is truncated}}, which decodes a valid length header over a truncated
> data region and asserts {{skipBinary}} raises {{ParquetDecodingException}}
> instead of spinning on an exhausted stream.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
