LuciferYang commented on code in PR #55932:
URL: https://github.com/apache/spark/pull/55932#discussion_r3749258790
##########
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:
`allData.array()` has no `hasArray()` fallback. The other readers in this
package all check first: `VectorizedPlainValuesReader.java` does it in 13
places, and `VectorizedDeltaBinaryPackedReader.java:461` has the same shape.
- Today: Spark uses parquet's default `HeapByteBufferAllocator`, so this is
latent rather than broken; the assumption is pre-existing and has only moved
into this method.
- Suggestion: follow `VectorizedPlainValuesReader.java:523-529` and fall
back to the ByteBuffer overload of `putByteArray`
(`WritableColumnVector.java:451`), which already handles direct and read-only
buffers.
##########
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:
The cost of `slice` depends on the stream. `SingleBufferInputStream.slice`
is always a zero-copy `duplicate()`, while `MultiBufferInputStream.slice` is
zero-copy only when `n <= current.remaining()`; otherwise it does
`ByteBuffer.allocate(n)` plus a full memcpy, before the EOF check.
- Impact: the old per-value slices copied at most one value, at a boundary.
The new one scales that copy to a whole batch.
- Scope: the default vectored path gives one buffer per range, so
multi-buffer only comes from the `ConsecutivePartList.readAll` fallback (8 MiB
chunks).
- Suggestion: take the bulk slice only when `in instanceof
SingleBufferInputStream`.
##########
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:
The two `ByteBufferOutputWriter::writeArrayByteBuffer` sites this PR removes
were the last uses in the repo. Only the declaration at
`VectorizedValuesReader.java:191` remains; `writeArrayByteBuffer` and
`skipWrite` have no callers, and `skipWrite` has had none since SPARK-37974
introduced it.
- Impact: a reader of `VectorizedValuesReader` will take it for the current
output convention, while both readers now call `putByteArray` directly.
- Suggestion: delete the interface here; `MimaExcludes.scala:161` already
excludes `org.apache.spark.sql.execution.*`.
##########
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:
All 14 cases in `ParquetDeltaLengthByteArrayEncodingSuite` write with the
writer and read back for comparison, so lengths and data always agree. None of
the three inputs that matter here are covered: an overflowing sum, a negative
length, and a stream that runs dry.
- Impact: all three fail silently, so a regression surfaces only in
production on a third-party file, and these methods serve nothing else since
Spark's own writer never emits DELTA_LENGTH_BYTE_ARRAY.
- Suggestion: add one case with a page whose lengths disagree with its data
region; truncating `writer.getBytes.toInputStream` is enough.
--
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]
