From 472db3244592d34e6ce50f61db78d0d5442c67df Mon Sep 17 00:00:00 2001 From: David Mollitor Date: Thu, 17 Sep 2026 13:07:51 +0000 Subject: [PATCH] [SPARK-59615][SQL] Avoid an intermediate buffer in VectorizedPlainValuesReader.readGeoData `VectorizedPlainValuesReader.readGeoData` (the vectorized PLAIN-encoding read path for GEOMETRY/GEOGRAPHY columns) buffered every converted value for the whole page into a `ByteBufferOutputStream` created with no initial capacity, then copied the result out via `toByteArray()` into `WritableColumnVector.arrayData().appendBytes(...)`. It also wrote a redundant 4-byte length prefix before each value. Rewrite it to append each converted value directly to the column vector's `arrayData()` and record it with `putArray`, matching `VectorizedDeltaByteArrayReader.readGeoData`. This removes the 32-byte-default `ByteBufferOutputStream` and its repeated grow-and-copy, the `toByteArray()` and bulk-append copies, and the redundant length prefix (the element length is already stored by `putArray`, which pointed past the prefix). `WritableColumnVector.appendBytes` already reserves and auto-grows the array data, so no intermediate buffer is needed. Behavior-preserving: the decoded column is byte-for-byte identical. Verified with `ParquetGeoSuite`, `ParquetDeltaByteArrayEncodingSuite`, and `ParquetDeltaLengthByteArrayEncodingSuite`. Co-authored-by: Isaac --- .../parquet/VectorizedPlainValuesReader.java | 25 +++++-------------- 1 file changed, 6 insertions(+), 19 deletions(-) diff --git a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedPlainValuesReader.java b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedPlainValuesReader.java index 4376d526e6629..d31980f311bfe 100644 --- a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedPlainValuesReader.java +++ b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedPlainValuesReader.java @@ -31,7 +31,6 @@ import org.apache.spark.sql.execution.vectorized.WritableColumnVector; import org.apache.spark.sql.types.GeographyType; import org.apache.spark.sql.types.GeometryType; -import org.apache.spark.util.ByteBufferOutputStream; /** * An implementation of the Parquet PLAIN decoder that supports the vectorized interface. @@ -592,29 +591,17 @@ public void readGeography(int total, WritableColumnVector v, int rowId) { private void readGeoData(int total, WritableColumnVector v, int rowId, int srid, WKBConverterStrategy converter) throws IOException { - // Go through the input stream and convert the WKB to the internal representation - // writing it to the output buffer and putting the (offset, length) in the vector. - // Finally, append all data from the output buffer in a single operation. - int base = v.arrayData().getElementsAppended(); - int dataLen = 0; - final int intSize = 4; - ByteBuffer lenBuffer = ByteBuffer.allocate(intSize); - ByteBufferOutputStream out = new ByteBufferOutputStream(); - + // Convert each WKB value to its physical representation and append it directly to the + // vector's array data, recording the (offset, length) of each element. + WritableColumnVector arrayData = v.arrayData(); for (int i = 0; i < total; i++) { int len = readInteger(); // Converts WKB into a physical representation of geometry/geography. byte[] physicalValue = converter.convert(in.readNBytes(len), srid); - v.putArray(rowId + i, base + dataLen + intSize, physicalValue.length); - - lenBuffer.putInt(0, physicalValue.length); - out.write(lenBuffer.array()); - out.write(physicalValue); - - dataLen += intSize + physicalValue.length; + int offset = arrayData.getElementsAppended(); + arrayData.appendBytes(physicalValue.length, physicalValue, 0); + v.putArray(rowId + i, offset, physicalValue.length); } - out.close(); - v.arrayData().appendBytes(dataLen, out.toByteArray(), 0); } }