dongjoon-hyun commented on code in PR #59087:
URL: https://github.com/apache/spark/pull/59087#discussion_r4124253166
##########
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:
nit: The row check happens inside `suffixReader.getBytes(currentRow)`, but
`readValues`, `readGeoData` and `skipBinary` call
`prefixLengthVector.getInt(currentRow)` and `checkPrefixLength` before it. When
both counts are 0, `readIntegers(0, ...)` still writes the DELTA_BINARY_PACKED
header's first value into slot 0, so a corrupt page whose definition levels
claim a non-null value fails with a misleading `prefix length N is larger than
the previous value's length 0` instead of the row-count error. It still throws
`ParquetDecodingException` and the ordering predates this PR, so only the
message is affected. Could we call `suffixReader.getBytes(currentRow)` before
`checkPrefixLength`, or reword this comment (e.g. "would otherwise read a
prefix length that was never decoded")?
##########
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:
Not a blocker for this PR, but the same class of issue remains in the
sibling readers on the same skip path:
- `VectorizedByteStreamSplitValuesReader`: `skip*` is just `offset += total`
without a bound, and `initFromPage` does not reject `totalBytes % typeWidth !=
0`, so a short page silently mis-strides every value.
- `VectorizedRleValuesReader`: when `readNextGroup()` returns `false` at the
end of the stream, callers silently `break` (partial dictionary-id
reads/skips), `readBooleans` ignores the return value and can loop forever, and
the level path can spin in `VectorizedColumnReader.readBatch`. The
`sliceStream(length)` length and the PACKED `numGroups * bitWidth` are not
validated either.
parquet-java 1.18.1 rejects most of these (`ByteStreamSplitValuesReader`,
and `RunLengthBitPackingHybridDecoder` with "Reading past RLE/BitPacking
stream."). Could you file follow-up JIRAs for them? The BYTE_STREAM_SPLIT
reader (SPARK-56894) is master-only, so it would be easy to fix before a
release.
##########
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:
nit: All the new negative tests use a `SingleBufferInputStream` (here, and
in `craftPage` since `BytesInput.concat(...).toInputStream` materializes into
one buffer), so the `MultiBufferInputStream` path is not covered. It behaves
differently: `skip(n)` with `n <= 0` is a no-op, and `slice(len)` allocates
`len` bytes before it detects EOF. Since bounding that allocation is one of the
reasons for the `len > in.available()` check, could we also run these tests on
a split page, e.g. `ByteBufferInputStream.wrap(ByteBuffer.wrap(a),
ByteBuffer.wrap(b))`, including a length close to `Int.MaxValue`?
##########
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:
nit: This is a valid one-value DELTA_LENGTH_BYTE_ARRAY page (only the
`valueCount = 2` passed to `initFromPage` is inconsistent), so the existing
`writer` fixture produces the same bytes:
```scala
writeBinaryData(writer, Array(wkb))
reader.initFromPage(2, writer.getBytes.toInputStream)
```
Then `lengthsInput` could be folded into `craftPage`.
##########
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:
nit: This returns the number of decoded (non-null) value lengths, but "the
number of values in the page" reads like the `valueCount` of `initFromPage`,
which includes nulls. `VectorizedDeltaBinaryPackedReader` already names the
same concept `getTotalValueCount()` ("True value count. May be less than
valueCount because of nulls"), and
`VectorizedDeltaByteArrayReader.initFromPage` now uses both names a few lines
apart. How about renaming this to `getTotalValueCount()` with a matching
Javadoc? `lengthCount` could also simply delegate to
`lengthReader.getTotalValueCount()`.
##########
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:
nit (optional): This is the same wrapper as the new private
`VectorizedPlainValuesReader.skipFully(long)` (same call and message), and it
can no longer fire after `validateLengths()`. If you want to share it, a
package-private static helper in `VectorizedReaderBase`, which already hosts
helpers used by non-subclasses such as `unsignedLongToBytesBigEndian`, could
serve 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:
nit: Since this PR now also validates DELTA_BYTE_ARRAY pages (the prefix
length count against the value count and the suffix count), how about
mentioning DELTA_BYTE_ARRAY in the PR title (and the JIRA title)? The PR title
becomes the commit title, so this new `ParquetDecodingException` would
otherwise be missing from the commit log.
--
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]
