LuciferYang commented on code in PR #55932:
URL: https://github.com/apache/spark/pull/55932#discussion_r3903821101
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaLengthByteArrayReader.java:
##########
@@ -113,12 +113,28 @@ 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 += checkLength(lengthsVector.getInt(currentRow + i));
+ }
+ try {
+ in.skipFully(totalSkip);
+ } catch (IOException e) {
+ throw new ParquetDecodingException("Failed to skip " + totalSkip + "
bytes", e);
}
currentRow += total;
}
+
+ /**
+ * Validates a length decoded from the page. Lengths are file-supplied and
unverified, so a
+ * negative value (whether crafted or corrupt) must be rejected before it
reaches
+ * {@code in.slice}, {@code in.skipFully}, or the column vector, where it
would otherwise read
+ * stale bytes, rewind the stream, or move {@code elementsAppended}
backwards silently.
+ */
+ private static int checkLength(int length) {
Review Comment:
The checkLength guard only covers the DELTA_LENGTH_BYTE_ARRAY reader. For
DELTA_BYTE_ARRAY, suffix lengths get validated through suffixReader.getBytes,
but prefix lengths (VectorizedDeltaByteArrayReader :81/:122/:167) do not, so a
negative value flows into appendBytes and array allocation;
VectorizedPlainValuesReader's readBinary/skipBinary (:521/:534) likewise use
the decoded len unchecked, and a negative length or truncated page produces no
clean error. These are pre-existing lines, but they are the same class of bug
this PR is fixing.
Either apply checkLength symmetrically, or file a follow-up JIRA. Note the
PLAIN fix cannot be just swapping in.skip for skipFully: in 1.17.1, a
single-buffer stream rewinds the position and returns the negative n, while a
multi-buffer stream just returns 0, and skipFully stays silent in both cases;
the length has to be checked first, with skipFully's IOException wrapped. A
negative-suffix DELTA_BYTE_ARRAY test would also pin down the path that picks
up the new check via getBytes.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetDeltaLengthByteArrayEncodingSuite.scala:
##########
@@ -92,6 +95,61 @@ class ParquetDeltaLengthByteArrayEncodingSuite
}
}
+ test("skipBinary fails cleanly when the data region is truncated") {
+ // The length header is intact but the data region is short, so the
lengths decoded from
+ // the page declare more bytes than the stream actually holds. skipBinary
must surface a
+ // ParquetDecodingException instead of spinning forever on an exhausted
stream.
+ writeData(writer, values)
+ val fullBytes = writer.getBytes.toByteArray
+ val truncated = java.util.Arrays.copyOf(fullBytes, fullBytes.length - 3)
+ reader.initFromPage(values.length,
ByteBufferInputStream.wrap(ByteBuffer.wrap(truncated)))
+ val e = intercept[ParquetDecodingException] {
+ reader.skipBinary(values.length)
+ }
+ assert(e.getMessage.contains("Failed to skip"))
+ }
+
+ test("readBinary, getBytes and skipBinary reject a negative decoded length")
{
+ // Spark's own writer never emits a negative length, but a third-party or
corrupt file can:
+ // the page is concat(lengthHeader, data), so write the length header
directly with a
+ // negative entry. A negative length must be rejected before it reaches
in.slice/in.skipFully
+ // or the column vector, where it would otherwise read stale bytes, rewind
the stream, or move
+ // elementsAppended backwards silently.
+ def negativeLengthPage(): ByteBufferInputStream = {
+ val lengthWriter = new DeltaBinaryPackingValuesWriterForInteger(
+ 128, 4, 100, 200, new DirectByteBufferAllocator())
+ Seq(3, -6, 3).foreach(lengthWriter.writeInteger)
+ val data = "abcdefghi".getBytes()
+ val page = BytesInput.concat(lengthWriter.getBytes,
BytesInput.from(data)).toByteArray
+ ByteBufferInputStream.wrap(ByteBuffer.wrap(page))
+ }
+
+ // readBinary: the -6 entry is hit on the second value and must throw.
+ reader.initFromPage(3, negativeLengthPage())
+ val readVector = new OnHeapColumnVector(3, StringType)
+ val readError = intercept[ParquetDecodingException] {
+ reader.readBinary(3, readVector, 0)
Review Comment:
The test catches the throw but not two classes of bad change. First, a throw
that comes too early: readBinary(3, ...) succeeds on the first value (length 3)
and writes "abc" into readVector, yet nothing asserts it, so a misindexed check
still passes. Second, a check that rejects too much: with lengths [3, -6, 3]
there is no zero-length entry, so a <= 0 check that wrongly rejects valid empty
binaries still throws on the second value with a matching message and the test
stays green.
Add a zero-length entry (e.g. [3, 0, -6]); first read the two valid rows
with a successful call outside the intercept and assert them (row 0 "abc", row
1 empty), then expect the third value to throw. Asserting after the throw
cannot catch <= 0: a row that was never written also reads back as an empty
binary via getBinary. Also note negativeLengthPage is shared by all three
sections: with [3, 0, -6] the getBytes probe must move from getBytes(1) to
getBytes(2), or it stops throwing.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaLengthByteArrayReader.java:
##########
@@ -58,7 +58,7 @@ public void readBinary(int total, WritableColumnVector c, int
rowId) {
ByteBufferOutputWriter outputWriter =
ByteBufferOutputWriter::writeArrayByteBuffer;
int length;
for (int i = 0; i < total; i++) {
- length = lengthsVector.getInt(currentRow + i);
+ length = checkLength(lengthsVector.getInt(currentRow + i));
Review Comment:
The new skipBinary wrapper keeps the cause (:123), while the three
pre-existing catches in the same file (:65/:97/:110, "Failed to read N bytes")
still drop the EOFException. They predate this PR, but it touches every one of
those methods anyway, so passing e into the constructor would keep error
diagnosis consistent.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaLengthByteArrayReader.java:
##########
@@ -88,7 +88,7 @@ private void readGeoData(int total, WritableColumnVector c,
int rowId, int srid,
ByteBufferOutputWriter outputWriter =
ByteBufferOutputWriter::writeArrayByteBuffer;
int length;
for (int i = 0; i < total; i++) {
- length = lengthsVector.getInt(currentRow + i);
+ length = checkLength(lengthsVector.getInt(currentRow + i));
Review Comment:
readGeoData also goes through checkLength (:91), but neither new test covers
the geometry/geography path, and STUtils/GeometryType are already available in
the suite. A negative-length case there would pin this path down too; note that
readNBytes returns a short array instead of throwing on truncation, so do not
rely on it when writing the case.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetDeltaLengthByteArrayEncodingSuite.scala:
##########
@@ -92,6 +95,61 @@ class ParquetDeltaLengthByteArrayEncodingSuite
}
}
+ test("skipBinary fails cleanly when the data region is truncated") {
+ // The length header is intact but the data region is short, so the
lengths decoded from
+ // the page declare more bytes than the stream actually holds. skipBinary
must surface a
+ // ParquetDecodingException instead of spinning forever on an exhausted
stream.
+ writeData(writer, values)
+ val fullBytes = writer.getBytes.toByteArray
+ val truncated = java.util.Arrays.copyOf(fullBytes, fullBytes.length - 3)
Review Comment:
The file already imports java.util.Random, so the fully qualified
java.util.Arrays.copyOf here just needs java.util.Arrays added to the same
import block.
--
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]
