dongjoon-hyun commented on code in PR #59118:
URL: https://github.com/apache/spark/pull/59118#discussion_r4148793592
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorizedSuite.scala:
##########
@@ -481,6 +484,44 @@ class ParquetVectorizedSuite extends ParquetTest with
SharedSparkSession {
}
}
+ test("SPARK-59831: BYTE_STREAM_SPLIT page of a required column with a wrong
value count") {
+ // A required column has no definition levels, so its page value count
must match the
+ // encoded values exactly. The page has 3 encoded values but claims 4. The
batch size is
+ // smaller than the page, so that the page is rejected when it is
initialized rather than
+ // when a read runs past its last encoded value.
+ val parquetSchema = MessageTypeParser.parseMessageType(
+ "message root { required float a; }")
+ val ty = parquetSchema.asGroupType().getType("a").asPrimitiveType()
+ val cd = new ColumnDescriptor(Array("a"), ty, 0, 0)
+ val valuesWriter = new FloatByteStreamSplitValuesWriter(3, 12, new
DirectByteBufferAllocator())
+ Seq(1.0f, 2.0f, 3.0f).foreach(v => valuesWriter.writeFloat(v))
+ val data = valuesWriter.getBytes.toByteArray
+ valuesWriter.close()
+
+ Seq(false, true).foreach { pageV2 =>
+ val memPageStore = new MemPageStore(4)
+ val pageWriter = memPageStore.getPageWriter(cd)
+ if (pageV2) {
+ pageWriter.writePageV2(4, 0, 4, BytesInput.empty(), BytesInput.empty(),
+ Encoding.BYTE_STREAM_SPLIT, BytesInput.from(data),
Statistics.createStats(ty))
+ } else {
+ pageWriter.writePage(BytesInput.from(data), 4, 4,
Statistics.createStats(ty),
+ Encoding.BIT_PACKED, Encoding.BIT_PACKED, Encoding.BYTE_STREAM_SPLIT)
Review Comment:
nit: `Encoding.BIT_PACKED` is deprecated in parquet-java, so these two
references add two deprecation warnings to the build, and no other code in the
repository names it. Since both max levels are 0, `readPageV1` does not look at
the level encodings (its `RLE` check only applies when `getMaxDefinitionLevel()
!= 0`), so `Encoding.RLE` would behave the same here.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedByteStreamSplitValuesReader.java:
##########
@@ -69,7 +85,29 @@ public void initFromPage(int valueCount,
ByteBufferInputStream in) throws IOExce
// but valueCount includes nulls. Use the actual bytes available in the
stream
// to derive the real number of encoded values.
int totalBytes = in.available();
- this.valueCount = totalBytes / typeWidth;
+ // The page stores no value count and the Parquet spec allows no padding
in a data
+ // page, so the page length must be an exact multiple of the value width.
Otherwise
+ // the page is truncated or has trailing bytes, and the stream boundaries
derived from
+ // its length may be wrong.
+ if (totalBytes % typeWidth != 0) {
+ throw new ParquetDecodingException("Corrupted BYTE_STREAM_SPLIT page:
page data length " +
+ totalBytes + " is not a multiple of the value width " + typeWidth);
+ }
+ int encodedValues = totalBytes / typeWidth;
+ // Without definition levels the page value count must match the encoded
values exactly.
+ if (exactValueCount && encodedValues != valueCount) {
+ throw new ParquetDecodingException("Corrupted BYTE_STREAM_SPLIT page: "
+ encodedValues +
+ " encoded values do not match the page value count " + valueCount);
+ }
+ // Otherwise the page value count includes nulls and is only an upper
bound, so a page that
+ // is off by whole values is not detected here, and its values are decoded
with the wrong
+ // stream stride. With fewer encoded values than non-null rows, a read or
skip past the
+ // last encoded value fails. With more, the page is not detected.
Review Comment:
nit: `is not detected here` and `With more, the page is not detected` read
as if such a page always went undetected, right above the check that rejects a
page with more encoded values than the page value count. Only a page that is
off by whole values within that bound goes undetected. Maybe something like:
```java
// Otherwise the page value count includes nulls and is only an upper
bound. A page with
// more encoded values than that is rejected below, but a page that is
off by whole values
// within the bound is not detected here, and its values are decoded
with the wrong stream
// stride. With fewer encoded values than non-null rows, a read or skip
past the last
// encoded value fails. With more, the page is not detected.
```
--
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]
