viirya commented on code in PR #59087:
URL: https://github.com/apache/spark/pull/59087#discussion_r4124800571
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaByteArrayReader.java:
##########
@@ -61,9 +61,21 @@ public class VectorizedDeltaByteArrayReader extends
VectorizedReaderBase
public void initFromPage(int valueCount, ByteBufferInputStream in) throws
IOException {
prefixLengthVector = new OnHeapColumnVector(valueCount, IntegerType);
prefixLengthReader.initFromPage(valueCount, in);
- prefixLengthReader.readIntegers(prefixLengthReader.getTotalValueCount(),
- prefixLengthVector, 0);
+ int prefixCount = prefixLengthReader.getTotalValueCount();
+ if (prefixCount > valueCount) {
+ throw new ParquetDecodingException("Corrupted DELTA_BYTE_ARRAY page: " +
prefixCount +
+ " prefix lengths in a page of " + valueCount + " values");
+ }
+ prefixLengthReader.readIntegers(prefixCount, prefixLengthVector, 0);
suffixReader.initFromPage(valueCount, in);
+ // Every value has both a prefix length and a suffix. Rows are
bounds-checked against the
Review Comment:
Done: `readValues`, `readGeoData` and `skipBinary` now call
`suffixReader.getBytes(currentRow)` before reading the prefix length, so an
undecoded row reports the row count, and I reworded the comment. The new test
builds a page whose prefix length header claims 0 values with a first value of
5. With the old order it failed with the misleading `prefix length 5 is larger
than the previous value's length 0`.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedPlainValuesReader.java:
##########
@@ -127,6 +123,38 @@ private ByteBuffer getBuffer(int length) {
}
}
+ /**
+ * Skips exactly `n` bytes. `in.skip` silently skips fewer bytes when the
page is too short,
+ * so a corrupt page must fail here, like `getBuffer` does on the read path.
+ */
+ private void skipFully(long n) {
Review Comment:
Filed SPARK-59831 for BYTE_STREAM_SPLIT and SPARK-59832 for RLE. Note that
SPARK-56894 was also backported to branch-4.3 and is in v4.3.0-rc1, so
SPARK-59831 should land before 4.3.0 is released.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VectorizedPlainValuesReaderSuite.scala:
##########
@@ -0,0 +1,135 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources.parquet
+
+import java.nio.{ByteBuffer, ByteOrder}
+import java.nio.charset.StandardCharsets
+
+import org.apache.parquet.bytes.ByteBufferInputStream
+import org.apache.parquet.io.ParquetDecodingException
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.execution.vectorized.OnHeapColumnVector
+import org.apache.spark.sql.types.{GeographyType, GeometryType, StringType}
+
+/**
+ * Tests that the vectorized PLAIN reader rejects corrupt pages on both the
read and the skip
+ * paths, instead of silently skipping too few bytes or moving the stream
position backwards.
+ */
+class VectorizedPlainValuesReaderSuite extends SparkFunSuite {
+
+ test("negative binary length is rejected when reading") {
+ val reader = newReader(page(4, "abcd", -1, "zzzz"))
+ val v = new OnHeapColumnVector(2, StringType)
+ val e = intercept[ParquetDecodingException] {
+ reader.readBinary(2, v, 0)
+ }
+ assert(e.getMessage.contains("negative binary length: -1"))
+ }
+
+ test("negative binary length is rejected when skipping") {
+ // Without the check, skipping the second value moved the position back to
the start of
+ // the page, so the next read returned "abcd" again.
+ val reader = newReader(page(4, "abcd", -12, "zzzz"))
+ val v = new OnHeapColumnVector(1, StringType)
+ reader.readBinary(1, v, 0)
+ val e = intercept[ParquetDecodingException] {
+ reader.skipBinary(1)
+ }
+ assert(e.getMessage.contains("negative binary length: -12"))
+ }
+
+ test("binary length larger than the rest of the page is rejected") {
+ val expected = "binary length 100 is larger than the 2 bytes left in the
page"
+ val e1 = intercept[ParquetDecodingException] {
+ newReader(page(100, "ab")).readBinary(1, new OnHeapColumnVector(1,
StringType), 0)
+ }
+ assert(e1.getMessage.contains(expected))
+ val e2 = intercept[ParquetDecodingException] {
+ newReader(page(100, "ab")).skipBinary(1)
+ }
+ assert(e2.getMessage.contains(expected))
+ }
+
+ test("geo paths reject invalid binary lengths") {
+ // The lengths are validated before the WKB is parsed, so the payload does
not matter.
+ Seq(GeometryType(0), GeographyType(4326)).foreach { geoType =>
+ def read(bytes: Array[Byte]): Unit = {
+ val reader = newReader(bytes)
+ val v = new OnHeapColumnVector(1, geoType)
+ geoType match {
+ case _: GeometryType => reader.readGeometry(1, v, 0)
+ case _: GeographyType => reader.readGeography(1, v, 0)
+ }
+ }
+ val e1 = intercept[ParquetDecodingException](read(page(-1, "abcd")))
+ assert(e1.getMessage.contains("negative binary length: -1"), geoType)
+ // Without the check, readNBytes returned a 2-byte array instead of
failing.
+ val e2 = intercept[ParquetDecodingException](read(page(100, "ab")))
+ assert(e2.getMessage.contains("binary length 100 is larger than the 2
bytes left"), geoType)
+ }
+ }
+
+ test("fixed-width skips past the end of the page are rejected") {
+ // An 8-byte page holds two 4-byte values or one 8-byte value.
+ val eightBytes = page(1, 2)
+ val cases: Seq[(String, VectorizedPlainValuesReader => Unit, Long)] = Seq(
+ ("skipBytes", _.skipBytes(3), 12L),
+ ("skipShorts", _.skipShorts(3), 12L),
+ ("skipIntegers", _.skipIntegers(3), 12L),
+ ("skipFloats", _.skipFloats(3), 12L),
+ ("skipLongs", _.skipLongs(2), 16L),
+ ("skipDoubles", _.skipDoubles(2), 16L),
+ ("skipFixedLenByteArray", _.skipFixedLenByteArray(3, 4), 12L),
+ // 72 booleans are 9 bytes.
+ ("skipBooleans", _.skipBooleans(72), 9L))
+ cases.foreach { case (name, skip, bytes) =>
+ val e = intercept[ParquetDecodingException] {
+ skip(newReader(eightBytes))
+ }
+ assert(e.getMessage.contains(s"Failed to skip $bytes bytes"), name)
+ }
+ }
+
+ test("skips within the page land on the next value") {
+ // Three binary values (including an empty one) followed by a sentinel int.
+ val binaryReader = newReader(page(2, "ab", 0, 3, "cde", 7))
+ binaryReader.skipBinary(3)
+ assert(binaryReader.readInteger() === 7)
+
+ val reader = newReader(page(1, 2, 3))
+ reader.skipIntegers(2)
+ assert(reader.readInteger() === 3)
+ }
+
+ private def newReader(bytes: Array[Byte]): VectorizedPlainValuesReader = {
+ val reader = new VectorizedPlainValuesReader
+ reader.initFromPage(0, ByteBufferInputStream.wrap(ByteBuffer.wrap(bytes)))
Review Comment:
Done. The negative tests in both suites now run on a single buffer and on
the same page split into 3-byte buffers. The split is not aligned with the
4-byte length prefixes, so reads and skips cross buffer boundaries. They also
include a length of `Int.MaxValue`.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaLengthByteArrayReader.java:
##########
@@ -48,21 +50,65 @@ public class VectorizedDeltaLengthByteArrayReader extends
VectorizedReaderBase i
public void initFromPage(int valueCount, ByteBufferInputStream in) throws
IOException {
lengthsVector = new OnHeapColumnVector(valueCount, IntegerType);
lengthReader.initFromPage(valueCount, in);
- lengthReader.readIntegers(lengthReader.getTotalValueCount(),
lengthsVector, 0);
+ lengthCount = lengthReader.getTotalValueCount();
+ if (lengthCount > valueCount) {
+ throw new ParquetDecodingException("Corrupted DELTA_LENGTH_BYTE_ARRAY
data: " +
+ lengthCount + " value lengths in a page of " + valueCount + "
values");
+ }
+ lengthReader.readIntegers(lengthCount, lengthsVector, 0);
this.in = in.remainingStream();
+ validateLengths();
+ }
+
+ /**
+ * The value lengths are read from the file, so validate all of them once,
before any of them
+ * is used: a corrupt page must fail the read instead of producing malformed
values. Values are
+ * consumed in order, so once every length is non-negative and they add up
to at most the bytes
+ * left in the page, no read or skip of a value can run past the end of the
page.
+ */
+ private void validateLengths() {
+ long totalLength = 0;
+ for (int i = 0; i < lengthCount; i++) {
+ int length = lengthsVector.getInt(i);
+ if (length < 0) {
+ throw new ParquetDecodingException(
+ "Corrupted DELTA_LENGTH_BYTE_ARRAY data: negative value length: "
+ length);
+ }
+ totalLength += length;
+ }
+ int available = in.available();
+ if (totalLength > available) {
+ throw new ParquetDecodingException("Corrupted DELTA_LENGTH_BYTE_ARRAY
data: value " +
+ "lengths add up to " + totalLength + " bytes, but only " + available
+ " are left");
+ }
+ }
+
+ /** Returns the number of values in the page, i.e. the number of decoded
value lengths. */
Review Comment:
Renamed to `getTotalValueCount()` with a matching Javadoc, and removed
`lengthCount` in favor of delegating to `lengthReader.getTotalValueCount()`.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetDeltaLengthByteArrayEncodingSuite.scala:
##########
@@ -105,6 +108,71 @@ class ParquetDeltaLengthByteArrayEncodingSuite
}
}
+ test("value lengths larger than the rest of the page are rejected") {
+ // The second value claims 100 bytes but only 2 bytes are left in the page.
+ val e = intercept[ParquetDecodingException] {
+ reader.initFromPage(2, craftPage(lengths = Array(2, 100), data = "abcd"))
+ }
+ assert(e.getMessage.contains("value lengths add up to 102 bytes, but only
4 are left"))
+ }
+
+ test("negative value length is rejected") {
+ val e = intercept[ParquetDecodingException] {
+ reader.initFromPage(3, craftPage(lengths = Array(2, -1, 3), data =
"abcd"))
+ }
+ assert(e.getMessage.contains("negative value length: -1"))
+ }
+
+ test("more value lengths than values in the page are rejected") {
+ val e = intercept[ParquetDecodingException] {
+ reader.initFromPage(2, craftPage(lengths = Array(1, 1, 1), data = "abc"))
+ }
+ assert(e.getMessage.contains("3 value lengths in a page of 2 values"))
+ }
+
+ test("rows without a decoded value length are rejected") {
+ // The page has 3 values but only 2 value lengths, so the third row has no
length.
+ def newReader(): VectorizedDeltaLengthByteArrayReader = {
+ val r = new VectorizedDeltaLengthByteArrayReader()
+ r.initFromPage(3, craftPage(lengths = Array(2, 2), data = "abcd"))
+ r
+ }
+ val expected = "reading 3 values from row 0, but only 2 value lengths were
decoded"
+ val e1 = intercept[ParquetDecodingException] {
+ newReader().readBinary(3, new OnHeapColumnVector(3, StringType), 0)
+ }
+ assert(e1.getMessage.contains(expected))
+ val e2 = intercept[ParquetDecodingException] {
+ newReader().skipBinary(3)
+ }
+ assert(e2.getMessage.contains(expected))
+
+ // getBytes is the path used by the DELTA_BYTE_ARRAY reader to read
suffixes.
+ val r = newReader()
+ r.getBytes(0)
+ r.getBytes(1)
+ val e3 = intercept[ParquetDecodingException] {
+ r.getBytes(2)
+ }
+ assert(e3.getMessage.contains(
+ "reading 1 values from row 2, but only 2 value lengths were decoded"))
+ }
+
+ testGeo("geo path rejects rows without a decoded value length") { geoType =>
+ val wkb = makePointWkb(1, 1)
+ val is = BytesInput.concat(
Review Comment:
Done: the geo test now uses the `writer` fixture, and `lengthsInput` is
folded into the page builder.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaLengthByteArrayReader.java:
##########
@@ -103,21 +150,26 @@ private void readGeoData(int total, WritableColumnVector
c, int rowId, int srid,
}
public ByteBuffer getBytes(int rowId) {
+ checkRows(rowId, 1);
int length = lengthsVector.getInt(rowId);
try {
return in.slice(length);
} catch (EOFException e) {
- throw new ParquetDecodingException("Failed to read " + length + "
bytes");
+ throw new ParquetDecodingException("Failed to read " + length + "
bytes", e);
}
}
@Override
public void skipBinary(int total) {
+ checkRows(currentRow, total);
+ long totalLength = 0;
for (int i = 0; i < total; i++) {
- int remaining = lengthsVector.getInt(currentRow + i);
- while (remaining > 0) {
- remaining -= in.skip(remaining);
- }
+ totalLength += lengthsVector.getInt(currentRow + i);
+ }
+ try {
Review Comment:
Moved it to a package-private static
`VectorizedReaderBase.skipFully(ByteBufferInputStream, long)` shared by both
readers.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaByteArrayReader.java:
##########
@@ -61,9 +61,21 @@ public class VectorizedDeltaByteArrayReader extends
VectorizedReaderBase
public void initFromPage(int valueCount, ByteBufferInputStream in) throws
IOException {
prefixLengthVector = new OnHeapColumnVector(valueCount, IntegerType);
prefixLengthReader.initFromPage(valueCount, in);
- prefixLengthReader.readIntegers(prefixLengthReader.getTotalValueCount(),
- prefixLengthVector, 0);
+ int prefixCount = prefixLengthReader.getTotalValueCount();
Review Comment:
Updated both the PR title and the JIRA title to include DELTA_BYTE_ARRAY.
--
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]
