iemejia commented on code in PR #55932:
URL: https://github.com/apache/spark/pull/55932#discussion_r3752926239
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaLengthByteArrayReader.java:
##########
@@ -113,11 +119,12 @@ public ByteBuffer getBytes(int rowId) {
@Override
public void skipBinary(int total) {
+ long totalSkip = 0;
for (int i = 0; i < total; i++) {
- int remaining = lengthsVector.getInt(currentRow + i);
- while (remaining > 0) {
- remaining -= in.skip(remaining);
- }
+ totalSkip += lengthsVector.getInt(currentRow + i);
+ }
+ while (totalSkip > 0) {
+ totalSkip -= in.skip(totalSkip);
Review Comment:
Good catch, fixed. `skipBinary` now uses `in.skipFully(totalSkip)` and wraps
the resulting `IOException` in a `ParquetDecodingException`, so an exhausted
stream fails fast with a diagnostic instead of spinning (and remains
interruptible). Each length is validated non-negative before being summed, so
`totalSkip` can never be negative and trigger a rewind. Added a test
(`skipBinary fails cleanly when the data region is truncated`) that decodes a
full length header over a truncated data region and asserts the exception.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaLengthByteArrayReader.java:
##########
@@ -54,17 +54,26 @@ public void initFromPage(int valueCount,
ByteBufferInputStream in) throws IOExce
@Override
public void readBinary(int total, WritableColumnVector c, int rowId) {
- ByteBuffer buffer;
- ByteBufferOutputWriter outputWriter =
ByteBufferOutputWriter::writeArrayByteBuffer;
- int length;
+ // Compute total data length across all values so we can read everything
in a single
+ // bulk slice instead of allocating one ByteBuffer per value.
+ int totalDataLen = 0;
for (int i = 0; i < total; i++) {
- length = lengthsVector.getInt(currentRow + i);
- try {
- buffer = in.slice(length);
- } catch (EOFException e) {
- throw new ParquetDecodingException("Failed to read " + length + "
bytes");
- }
- outputWriter.write(c, rowId + i, buffer, length);
+ totalDataLen += lengthsVector.getInt(currentRow + i);
Review Comment:
Addressed by reverting `readBinary` to the per-value `in.slice(length)`
implementation, as suggested in your summary comment below. There is no longer
any `int` summation of file-supplied lengths in `readBinary`, so the
overflow-to-zero path is gone. The only remaining summation is in `skipBinary`,
which now accumulates into a `long`.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaLengthByteArrayReader.java:
##########
@@ -113,11 +119,12 @@ public ByteBuffer getBytes(int rowId) {
@Override
public void skipBinary(int total) {
+ long totalSkip = 0;
for (int i = 0; i < total; i++) {
- int remaining = lengthsVector.getInt(currentRow + i);
- while (remaining > 0) {
- remaining -= in.skip(remaining);
- }
+ totalSkip += lengthsVector.getInt(currentRow + i);
Review Comment:
Fixed in `skipBinary`: each length is validated `>= 0` before being added to
the running total, so a batch like `[100, -100]` now throws
`ParquetDecodingException` rather than summing to 0 and letting `currentRow`
drift. The `readBinary` side of this concern is resolved by reverting that
method to `master` (no batch summation there).
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaLengthByteArrayReader.java:
##########
@@ -54,17 +54,26 @@ public void initFromPage(int valueCount,
ByteBufferInputStream in) throws IOExce
@Override
public void readBinary(int total, WritableColumnVector c, int rowId) {
- ByteBuffer buffer;
- ByteBufferOutputWriter outputWriter =
ByteBufferOutputWriter::writeArrayByteBuffer;
- int length;
+ // Compute total data length across all values so we can read everything
in a single
Review Comment:
Agreed, and done. The `readBinary` bulk-slice change has been dropped
entirely and reverted (along with `readGeoData`) to `master`. The PR now
contains only the measurable `skipBinary` win plus the validation fixes. The
description and the umbrella issue have been updated to drop the `readBinary`
table; a fresh AMD EPYC 7763 benchmark run on the reverted code confirms
`readBinary` now measures identical to the baseline (a true no-op), while
`skipBinary` stays at a consistent 2.00x-2.33x.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaLengthByteArrayReader.java:
##########
@@ -54,17 +54,26 @@ public void initFromPage(int valueCount,
ByteBufferInputStream in) throws IOExce
@Override
public void readBinary(int total, WritableColumnVector c, int rowId) {
- ByteBuffer buffer;
- ByteBufferOutputWriter outputWriter =
ByteBufferOutputWriter::writeArrayByteBuffer;
- int length;
+ // Compute total data length across all values so we can read everything
in a single
+ // bulk slice instead of allocating one ByteBuffer per value.
+ int totalDataLen = 0;
for (int i = 0; i < total; i++) {
- length = lengthsVector.getInt(currentRow + i);
- try {
- buffer = in.slice(length);
- } catch (EOFException e) {
- throw new ParquetDecodingException("Failed to read " + length + "
bytes");
- }
- outputWriter.write(c, rowId + i, buffer, length);
+ totalDataLen += lengthsVector.getInt(currentRow + i);
+ }
+
+ ByteBuffer allData;
+ try {
+ allData = in.slice(totalDataLen);
+ } catch (EOFException e) {
+ throw new ParquetDecodingException("Failed to read " + totalDataLen + "
bytes");
+ }
+ byte[] dataArray = allData.array();
Review Comment:
Resolved by reverting `readBinary` - the unguarded `array()` access is gone
with the bulk-slice removal.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaLengthByteArrayReader.java:
##########
@@ -54,17 +54,26 @@ public void initFromPage(int valueCount,
ByteBufferInputStream in) throws IOExce
@Override
public void readBinary(int total, WritableColumnVector c, int rowId) {
- ByteBuffer buffer;
- ByteBufferOutputWriter outputWriter =
ByteBufferOutputWriter::writeArrayByteBuffer;
- int length;
+ // Compute total data length across all values so we can read everything
in a single
+ // bulk slice instead of allocating one ByteBuffer per value.
+ int totalDataLen = 0;
for (int i = 0; i < total; i++) {
- length = lengthsVector.getInt(currentRow + i);
- try {
- buffer = in.slice(length);
- } catch (EOFException e) {
- throw new ParquetDecodingException("Failed to read " + length + "
bytes");
- }
- outputWriter.write(c, rowId + i, buffer, length);
+ totalDataLen += lengthsVector.getInt(currentRow + i);
+ }
+
+ ByteBuffer allData;
+ try {
+ allData = in.slice(totalDataLen);
Review Comment:
Resolved by the revert - `readBinary` no longer takes a whole-batch slice,
so the `MultiBufferInputStream.slice` allocate+memcpy path is no longer on the
hot path.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaLengthByteArrayReader.java:
##########
@@ -85,19 +94,16 @@ public void readGeography(int total, WritableColumnVector
c, int rowId) {
private void readGeoData(int total, WritableColumnVector c, int rowId, int
srid,
WKBConverterStrategy converter) {
- ByteBufferOutputWriter outputWriter =
ByteBufferOutputWriter::writeArrayByteBuffer;
- int length;
for (int i = 0; i < total; i++) {
- length = lengthsVector.getInt(currentRow + i);
+ int length = lengthsVector.getInt(currentRow + i);
byte[] physicalValue;
try {
// Converts WKB into a physical representation of geometry/geography.
physicalValue = converter.convert(in.readNBytes(length), srid);
} catch (IOException e) {
throw new ParquetDecodingException("Failed to read " + length + "
bytes");
}
-
- outputWriter.write(c, rowId + i, ByteBuffer.wrap(physicalValue),
physicalValue.length);
+ c.putByteArray(rowId + i, physicalValue, 0, physicalValue.length);
Review Comment:
Since we reverted `readGeoData` as well, both
`ByteBufferOutputWriter::writeArrayByteBuffer` call sites remain, so the
interface still has callers and this PR no longer makes it dead code. Removing
it is therefore out of scope here - happy to file a separate follow-up cleanup
for it if you'd like.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaLengthByteArrayReader.java:
##########
@@ -113,11 +119,12 @@ public ByteBuffer getBytes(int rowId) {
@Override
public void skipBinary(int total) {
+ long totalSkip = 0;
Review Comment:
Added `skipBinary fails cleanly when the data region is truncated`: it
writes a valid page, truncates the data region so the decoded lengths declare
more bytes than the stream holds, and asserts `skipBinary` throws
`ParquetDecodingException`. This directly exercises the dry-stream case (the
infinite-loop fix). The overflow case no longer applies now that `readBinary`
is reverted and `skipBinary` sums into a `long`; a negative-length page can't
be produced through the parquet writer, but that path is guarded by the
explicit `length < 0` check.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
