This is an automated email from the ASF dual-hosted git repository.
viirya pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 5f4b183a0d39 [SPARK-59831][SQL] Validate bounds in the vectorized
BYTE_STREAM_SPLIT reader
5f4b183a0d39 is described below
commit 5f4b183a0d39a5a0c22bcdfc2a3b5387cffbc292
Author: Liang-Chi Hsieh <[email protected]>
AuthorDate: Wed Sep 30 18:23:03 2026 -0700
[SPARK-59831][SQL] Validate bounds in the vectorized BYTE_STREAM_SPLIT
reader
### What changes were proposed in this pull request?
Follow-up of SPARK-59812 (#59087), which made the vectorized PLAIN,
DELTA_LENGTH_BYTE_ARRAY and DELTA_BYTE_ARRAY readers fail with a
`ParquetDecodingException` on corrupt pages. This does the same for
`VectorizedByteStreamSplitValuesReader` (added by SPARK-56894), which did not
check that reads and skips stay within the page:
- `initFromPage` rejects a page whose length is not a multiple of the value
width, and a page with more encoded values than the page value count.
parquet-java 1.18.1's `ByteStreamSplitValuesReader` rejects both. For a column
without definition levels (max definition level 0), where every row has a
value, the page value count must match the encoded values exactly, for both v1
and v2 pages. For other columns it includes nulls and is only an upper bound.
- Batch reads (`readIntegers`, `readLongs`, `readFloats`, `readDoubles`,
their widening/narrowing variants, `readBytes`, `readShorts`, `readBinary`,
`readFixedLenByteArray`) and all skips check once per batch that the page has
enough values left. The per-value loops are unchanged.
- Single-value reads (`readInteger`, `readLong`, `readFloat`, `readDouble`,
`readByte`, `readShort`, `readBinary(len)`) check that a value is left.
- A negative count to read or skip, which can only come from the caller,
throws an `IllegalArgumentException`. So does `readBinary(int len)` with a
`len` other than the value width, before consuming a value. Its callers always
pass the type length of the column, so this is not reachable today, but it used
to fail with an `ArrayIndexOutOfBoundsException` for a larger `len` and return
a prefix of the value for a smaller one.
### Why are the changes needed?
On a corrupt page:
- The skip methods only did `offset += total`, so skipping past the end of
the page was never detected.
- Reading past the end (or reading after such a skip) failed with an
`ArrayIndexOutOfBoundsException` instead of a `ParquetDecodingException`.
- A page whose length is not a multiple of the value width was accepted.
BYTE_STREAM_SPLIT stores no value count, and the Parquet spec allows no padding
in a data page, so the stream boundaries cannot be derived for such a page: for
example, a page that lost a byte has every stream start at the wrong offset, so
every value is decoded from the wrong bytes. In addition, reads past the last
whole value silently returned values assembled from the trailing bytes instead
of failing.
- For a required column, a page with fewer encoded values than its value
count failed only when a read reached the end of the page, after returning
values decoded with the wrong stream stride.
For a column with definition levels, a data section that is off by whole
values is still not detected when the page is initialized, as in parquet-java,
and its values are decoded with the wrong stride. With fewer encoded values
than non-null values, a read or skip past the last encoded value fails. With
more, the page is not detected; checking that all encoded values are consumed
at the end of each page is left as a follow-up. The exact count of `DataPageV2`
(`num_values - num_nulls`) [...]
### Does this PR introduce _any_ user-facing change?
Only for corrupt Parquet files: reading such a BYTE_STREAM_SPLIT page with
the vectorized reader now fails with a `ParquetDecodingException` instead of an
`ArrayIndexOutOfBoundsException` or silently wrong values. Valid files are not
affected. For a required column, a page with fewer encoded values than its
value count now fails when the page is initialized, so a query that reads only
the first rows of such a page (e.g. with `LIMIT` or column-index row ranges)
fails instead of returni [...]
### How was this patch tested?
New tests in `ParquetByteStreamSplitEncodingSuite`, run for INT32, INT64,
FLOAT and DOUBLE, and for FIXED_LEN_BYTE_ARRAY of width 5:
- Negative: every applicable batch read and skip past the end of the page,
after a skip, and with a count that would overflow `offset + total`; a negative
count; every single-value read past the end; page lengths with 1 to W-1 extra
or missing bytes; more encoded values than the page value count; fewer encoded
values than an exact page value count; reads and skips on an empty page.
- Positive: every batch read and skip ending exactly at the end of the
page, checking the values read, followed by empty reads and skips; reading the
last value after a skip; fewer encoded values than the page value count
(nullable columns); an exact page value count that matches.
End-to-end tests:
- `ParquetVectorizedSuite`: a required BYTE_STREAM_SPLIT page whose value
count is one more than its encoded values, written with v1 and v2 pages to an
in-memory page store and read through `VectorizedParquetRecordReader`, fails
when the page is initialized. This covers the wiring of the exact check in
`VectorizedColumnReader`.
- `ParquetEncodingSuite`: a round trip with v1 and v2 pages, more than one
page per column, and required, optional and array columns of all
BYTE_STREAM_SPLIT types.
- `ParquetColumnIndexSuite`: unaligned BYTE_STREAM_SPLIT pages with v1 and
v2 pages, required and nullable columns, so that rows are skipped within the
BYTE_STREAM_SPLIT pages. The test asserts the encodings and that the page
boundaries of the filter column fall inside the pages of the unaligned columns.
All the new negative tests fail without this change. Making the skips throw
fails the `ParquetColumnIndexSuite` test for both page versions, making the
exact count check throw fails the `ParquetEncodingSuite` round trip for both,
passing `false` for the exact check in `VectorizedColumnReader` fails the
`ParquetVectorizedSuite` test, and rewriting the remaining count check as
`offset + total > valueCount` fails the overflow test. Existing suites also
pass: `ParquetEncodingSuite`, `Parq [...]
The single-value reads add a bounds check per value. An A/B microbenchmark
of master's reader against this one (per-value `readInteger`, `readLong`,
`readDouble`, `readBinary(12)` and batch `readIntegers`, `readDoubles`, 1M
values, JDK 21) showed no measurable difference.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5.5)
Closes #59118 from viirya/SPARK-59831-bss-bounds.
Authored-by: Liang-Chi Hsieh <[email protected]>
Signed-off-by: Liang-Chi Hsieh <[email protected]>
(cherry picked from commit f868de6914d0e6ede86309648614bb436b2b82e6)
Signed-off-by: Liang-Chi Hsieh <[email protected]>
---
.../VectorizedByteStreamSplitValuesReader.java | 121 ++++++--
.../parquet/VectorizedColumnReader.java | 3 +-
.../ParquetByteStreamSplitEncodingSuite.scala | 306 +++++++++++++++++++--
.../parquet/ParquetColumnIndexSuite.scala | 57 ++++
.../datasources/parquet/ParquetEncodingSuite.scala | 101 ++++++-
.../parquet/ParquetVectorizedSuite.scala | 45 ++-
6 files changed, 594 insertions(+), 39 deletions(-)
diff --git
a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedByteStreamSplitValuesReader.java
b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedByteStreamSplitValuesReader.java
index 4ca479ace79d..027e0bd9988e 100644
---
a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedByteStreamSplitValuesReader.java
+++
b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedByteStreamSplitValuesReader.java
@@ -20,6 +20,7 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import org.apache.parquet.bytes.ByteBufferInputStream;
+import org.apache.parquet.io.ParquetDecodingException;
import org.apache.parquet.io.api.Binary;
import org.apache.spark.sql.execution.vectorized.WritableColumnVector;
@@ -50,7 +51,10 @@ public class VectorizedByteStreamSplitValuesReader
/** Width of each value in bytes (4 for FLOAT/INT32, 8 for DOUBLE/INT64). */
private final int typeWidth;
- /** Total number of values in the current page. */
+ /**
+ * Number of values encoded in the current page. For a column with
definition levels this
+ * excludes nulls, so it can be smaller than the page value count.
+ */
private int valueCount;
/** Raw encoded page data: W streams of valueCount bytes each. */
@@ -59,8 +63,20 @@ public class VectorizedByteStreamSplitValuesReader
/** Current read position (number of values consumed so far). */
private int offset;
+ /**
+ * Whether the page value count is exactly the number of encoded values.
This holds for a
+ * column without definition levels (max definition level 0), where every
row has a value.
+ * Otherwise the page value count includes nulls and is only an upper bound.
+ */
+ private final boolean exactValueCount;
+
public VectorizedByteStreamSplitValuesReader(int typeWidth) {
+ this(typeWidth, false);
+ }
+
+ public VectorizedByteStreamSplitValuesReader(int typeWidth, boolean
exactValueCount) {
this.typeWidth = typeWidth;
+ this.exactValueCount = exactValueCount;
}
@Override
@@ -69,7 +85,30 @@ public class VectorizedByteStreamSplitValuesReader
// 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. 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.
+ if (encodedValues > valueCount) {
+ throw new ParquetDecodingException("Corrupted BYTE_STREAM_SPLIT page: "
+ encodedValues +
+ " encoded values exceed the page value count " + valueCount);
+ }
+ this.valueCount = encodedValues;
this.offset = 0;
this.pageData = new byte[totalBytes];
// Read the entire page into pageData. ByteBufferInputStream.slice()
returns a
@@ -81,6 +120,28 @@ public class VectorizedByteStreamSplitValuesReader
// --------------- helpers ---------------
+ /**
+ * Checks that the page has at least {@code total} values left. Batch reads
and skips call
+ * this once per batch, so the per-value loops need no bound check of their
own.
+ */
+ private void checkRemaining(int total) {
+ if (total < 0) {
+ throw new IllegalArgumentException("Cannot read or skip a negative
number of values: " +
+ total);
+ }
+ if (total > valueCount - offset) {
+ throw new ParquetDecodingException("Corrupted BYTE_STREAM_SPLIT page:
cannot read or " +
+ "skip " + total + " values, only " + (valueCount - offset) + " of " +
valueCount +
+ " values are left");
+ }
+ }
+
+ /** Returns the index of the next value for a single-value read and advances
past it. */
+ private int nextIndex() {
+ checkRemaining(1);
+ return offset++;
+ }
+
/** Assembles a single 4-byte little-endian int from 4 streams at the given
index. */
private int assembleInt(int idx) {
return (pageData[idx] & 0xFF)
@@ -115,31 +176,36 @@ public class VectorizedByteStreamSplitValuesReader
@Override
public int readInteger() {
- return assembleInt(offset++);
+ return assembleInt(nextIndex());
}
@Override
public long readLong() {
- return assembleLong(offset++);
+ return assembleLong(nextIndex());
}
@Override
public float readFloat() {
- return Float.intBitsToFloat(assembleInt(offset++));
+ return Float.intBitsToFloat(assembleInt(nextIndex()));
}
@Override
public double readDouble() {
- return Double.longBitsToDouble(assembleLong(offset++));
+ return Double.longBitsToDouble(assembleLong(nextIndex()));
}
@Override
public Binary readBinary(int len) {
- byte[] result = new byte[len];
- for (int b = 0; b < len; b++) {
- result[b] = pageData[b * valueCount + offset];
+ // Each value has one byte in each of the typeWidth streams.
+ if (len != typeWidth) {
+ throw new IllegalArgumentException(
+ "Cannot read a value of length " + len + " from values of width " +
typeWidth);
+ }
+ byte[] result = new byte[typeWidth];
+ int idx = nextIndex();
+ for (int b = 0; b < typeWidth; b++) {
+ result[b] = pageData[b * valueCount + idx];
}
- offset++;
return Binary.fromConstantByteArray(result);
}
@@ -147,6 +213,7 @@ public class VectorizedByteStreamSplitValuesReader
@Override
public void readBytes(int total, WritableColumnVector c, int rowId) {
+ checkRemaining(total);
for (int i = 0; i < total; i++) {
c.putByte(rowId + i, (byte) assembleInt(offset + i));
}
@@ -155,6 +222,7 @@ public class VectorizedByteStreamSplitValuesReader
@Override
public void readShorts(int total, WritableColumnVector c, int rowId) {
+ checkRemaining(total);
for (int i = 0; i < total; i++) {
c.putShort(rowId + i, (short) assembleInt(offset + i));
}
@@ -163,6 +231,7 @@ public class VectorizedByteStreamSplitValuesReader
@Override
public void readIntegers(int total, WritableColumnVector c, int rowId) {
+ checkRemaining(total);
for (int i = 0; i < total; i++) {
c.putInt(rowId + i, assembleInt(offset + i));
}
@@ -171,6 +240,7 @@ public class VectorizedByteStreamSplitValuesReader
@Override
public void readIntegersAsLongs(int total, WritableColumnVector c, int
rowId) {
+ checkRemaining(total);
for (int i = 0; i < total; i++) {
c.putLong(rowId + i, assembleInt(offset + i));
}
@@ -179,6 +249,7 @@ public class VectorizedByteStreamSplitValuesReader
@Override
public void readIntegersAsDoubles(int total, WritableColumnVector c, int
rowId) {
+ checkRemaining(total);
for (int i = 0; i < total; i++) {
c.putDouble(rowId + i, (double) assembleInt(offset + i));
}
@@ -187,6 +258,7 @@ public class VectorizedByteStreamSplitValuesReader
@Override
public void readLongs(int total, WritableColumnVector c, int rowId) {
+ checkRemaining(total);
for (int i = 0; i < total; i++) {
c.putLong(rowId + i, assembleLong(offset + i));
}
@@ -195,6 +267,7 @@ public class VectorizedByteStreamSplitValuesReader
@Override
public void readLongsAsInts(int total, WritableColumnVector c, int rowId) {
+ checkRemaining(total);
for (int i = 0; i < total; i++) {
c.putInt(rowId + i, (int) assembleLong(offset + i));
}
@@ -203,6 +276,7 @@ public class VectorizedByteStreamSplitValuesReader
@Override
public void readFloats(int total, WritableColumnVector c, int rowId) {
+ checkRemaining(total);
for (int i = 0; i < total; i++) {
c.putFloat(rowId + i, Float.intBitsToFloat(assembleInt(offset + i)));
}
@@ -211,6 +285,7 @@ public class VectorizedByteStreamSplitValuesReader
@Override
public void readFloatsAsDoubles(int total, WritableColumnVector c, int
rowId) {
+ checkRemaining(total);
for (int i = 0; i < total; i++) {
c.putDouble(rowId + i, (double) Float.intBitsToFloat(assembleInt(offset
+ i)));
}
@@ -219,6 +294,7 @@ public class VectorizedByteStreamSplitValuesReader
@Override
public void readDoubles(int total, WritableColumnVector c, int rowId) {
+ checkRemaining(total);
for (int i = 0; i < total; i++) {
c.putDouble(rowId + i, Double.longBitsToDouble(assembleLong(offset +
i)));
}
@@ -227,6 +303,7 @@ public class VectorizedByteStreamSplitValuesReader
@Override
public void readBinary(int total, WritableColumnVector c, int rowId) {
+ checkRemaining(total);
// Reuse a single scratch buffer to avoid per-value byte[] + Binary
allocation.
byte[] scratch = new byte[typeWidth];
for (int i = 0; i < total; i++) {
@@ -244,30 +321,36 @@ public class VectorizedByteStreamSplitValuesReader
}
// --------------- skip methods ---------------
- // All types share the same page layout, so skipping is just advancing the
offset.
+ // All types share the same page layout, so skipping is just advancing the
offset
+ // after checking the page has enough values left.
// skipBooleans is not overridden: BSS never encodes booleans; the base
class throws.
+ private void skipValues(int total) {
+ checkRemaining(total);
+ offset += total;
+ }
+
@Override
- public void skipBytes(int total) { offset += total; }
+ public void skipBytes(int total) { skipValues(total); }
@Override
- public void skipShorts(int total) { offset += total; }
+ public void skipShorts(int total) { skipValues(total); }
@Override
- public void skipIntegers(int total) { offset += total; }
+ public void skipIntegers(int total) { skipValues(total); }
@Override
- public void skipLongs(int total) { offset += total; }
+ public void skipLongs(int total) { skipValues(total); }
@Override
- public void skipFloats(int total) { offset += total; }
+ public void skipFloats(int total) { skipValues(total); }
@Override
- public void skipDoubles(int total) { offset += total; }
+ public void skipDoubles(int total) { skipValues(total); }
@Override
- public void skipBinary(int total) { offset += total; }
+ public void skipBinary(int total) { skipValues(total); }
@Override
- public void skipFixedLenByteArray(int total, int len) { offset += total; }
+ public void skipFixedLenByteArray(int total, int len) { skipValues(total); }
}
diff --git
a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
index 63aa0951313f..eff701082589 100644
---
a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
+++
b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
@@ -389,7 +389,8 @@ public class VectorizedColumnReader {
default -> throw new SparkUnsupportedOperationException(
"_LEGACY_ERROR_TEMP_3190", Map.of("typeName",
typeName.toString()));
};
- yield new VectorizedByteStreamSplitValuesReader(typeWidth);
+ yield new VectorizedByteStreamSplitValuesReader(
+ typeWidth, this.descriptor.getMaxDefinitionLevel() == 0);
}
case RLE -> {
PrimitiveType.PrimitiveTypeName typeName =
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetByteStreamSplitEncodingSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetByteStreamSplitEncodingSuite.scala
index da07bddc9723..6424b64d43a3 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetByteStreamSplitEncodingSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetByteStreamSplitEncodingSuite.scala
@@ -16,7 +16,7 @@
*/
package org.apache.spark.sql.execution.datasources.parquet
-import java.nio.ByteBuffer
+import java.nio.{ByteBuffer, ByteOrder}
import java.util.Random
import scala.reflect.ClassTag
@@ -24,7 +24,9 @@ import scala.reflect.ClassTag
import org.apache.parquet.bytes.{ByteBufferInputStream,
DirectByteBufferAllocator}
import org.apache.parquet.column.values.ValuesWriter
import
org.apache.parquet.column.values.bytestreamsplit.ByteStreamSplitValuesWriter._
+import org.apache.parquet.io.ParquetDecodingException
import org.apache.parquet.io.api.Binary
+import org.scalactic.source.Position
import org.apache.spark.SparkFunSuite
import org.apache.spark.sql.execution.vectorized.{OnHeapColumnVector,
WritableColumnVector}
@@ -36,9 +38,10 @@ import org.apache.spark.sql.types._
* Uses parquet-mr's ByteStreamSplitValuesWriter to encode data, then reads it
* back with the vectorized reader and verifies correctness. An abstract base
* covers the shared test matrix (batch reads, single-value reads, skip, direct
- * buffers, extreme values) for all numeric types; concrete sub-classes supply
- * only the type-specific writer/reader/comparison methods. FLBA is tested in a
- * standalone suite because its reader API differs (readBinary vs typed batch).
+ * buffers, extreme values, and bounds checks of every read and skip against
the
+ * page) for INT32, INT64, FLOAT, DOUBLE and FIXED_LEN_BYTE_ARRAY of width 5;
+ * concrete sub-classes supply only the type-specific writer/reader/comparison
+ * methods. A standalone suite covers FIXED_LEN_BYTE_ARRAY of other widths.
*/
abstract class ParquetByteStreamSplitEncodingSuite[T: ClassTag] extends
SparkFunSuite {
@@ -79,6 +82,9 @@ abstract class ParquetByteStreamSplitEncodingSuite[T:
ClassTag] extends SparkFun
/** A single representative value. */
protected def singleTestValue: T
+ /** The little-endian bytes of a value, as a reader of any width assembles
them. */
+ protected def toBytes(v: T): Array[Byte]
+
/** Override for types that need bitwise comparison (Float, Double). */
protected def assertEqual(expected: T, actual: T, msg: String): Unit = {
assert(expected === actual, msg)
@@ -88,8 +94,9 @@ abstract class ParquetByteStreamSplitEncodingSuite[T:
ClassTag] extends SparkFun
private def newReader(
page: Array[Byte], count: Int,
- useDirect: Boolean = false): VectorizedByteStreamSplitValuesReader = {
- val reader = new VectorizedByteStreamSplitValuesReader(typeWidth)
+ useDirect: Boolean = false,
+ exactValueCount: Boolean = false): VectorizedByteStreamSplitValuesReader
= {
+ val reader = new VectorizedByteStreamSplitValuesReader(typeWidth,
exactValueCount)
val buf = if (useDirect) {
val b = ByteBuffer.allocateDirect(page.length)
b.put(page); b.flip(); b
@@ -100,9 +107,12 @@ abstract class ParquetByteStreamSplitEncodingSuite[T:
ClassTag] extends SparkFun
reader
}
- private def readAndVerify(data: Array[T], useDirect: Boolean = false): Unit
= {
+ private def readAndVerify(
+ data: Array[T],
+ useDirect: Boolean = false,
+ exactValueCount: Boolean = false): Unit = {
val page = encode(data)
- val reader = newReader(page, data.length, useDirect)
+ val reader = newReader(page, data.length, useDirect, exactValueCount)
val cv = new OnHeapColumnVector(data.length, sparkType)
try {
readBatch(reader, data.length, cv, 0)
@@ -114,6 +124,85 @@ abstract class ParquetByteStreamSplitEncodingSuite[T:
ClassTag] extends SparkFun
}
}
+ private type Reader = VectorizedByteStreamSplitValuesReader
+ private type BatchOp = (Reader, Int, WritableColumnVector) => Unit
+ /** Checks the value at a row of the column vector against the little-endian
bytes. */
+ private type Check = (WritableColumnVector, Int, Array[Byte]) => Unit
+
+ private def littleEndian(bytes: Array[Byte]): Long =
+ bytes.indices.map(i => (bytes(i) & 0xFFL) << (8 * i)).sum
+
+ private def checkBinary: Check = (c, i, b) => assert(c.getBinary(i).toSeq
=== b.toSeq)
+
+ /**
+ * Every batch read and skip of the reader that applies to this value width,
with a column
+ * vector type it can write into and, for reads, a check of the values it
wrote. The 4-byte
+ * and 8-byte reads assemble exactly 4 and 8 streams, so they only apply to
pages of that
+ * width.
+ */
+ private lazy val batchOps: Seq[(String, DataType, BatchOp, Option[Check])] =
+ Seq[(String, DataType, Int, BatchOp, Option[Check])](
+ ("readBytes", ByteType, 4, (r, n, c) => r.readBytes(n, c, 0),
+ Some((c, i, b) => assert(c.getByte(i) === littleEndian(b).toByte))),
+ ("readShorts", ShortType, 4, (r, n, c) => r.readShorts(n, c, 0),
+ Some((c, i, b) => assert(c.getShort(i) === littleEndian(b).toShort))),
+ ("readIntegers", IntegerType, 4, (r, n, c) => r.readIntegers(n, c, 0),
+ Some((c, i, b) => assert(c.getInt(i) === littleEndian(b).toInt))),
+ ("readIntegersAsLongs", LongType, 4, (r, n, c) =>
r.readIntegersAsLongs(n, c, 0),
+ Some((c, i, b) => assert(c.getLong(i) ===
littleEndian(b).toInt.toLong))),
+ ("readIntegersAsDoubles", DoubleType, 4, (r, n, c) =>
r.readIntegersAsDoubles(n, c, 0),
+ Some((c, i, b) => assert(c.getDouble(i) ===
littleEndian(b).toInt.toDouble))),
+ ("readFloats", FloatType, 4, (r, n, c) => r.readFloats(n, c, 0),
+ Some((c, i, b) =>
+ assert(java.lang.Float.floatToRawIntBits(c.getFloat(i)) ===
littleEndian(b).toInt))),
+ ("readFloatsAsDoubles", DoubleType, 4, (r, n, c) =>
r.readFloatsAsDoubles(n, c, 0),
+ Some((c, i, b) => {
+ val expected =
java.lang.Float.intBitsToFloat(littleEndian(b).toInt).toDouble
+ assert(java.lang.Double.doubleToRawLongBits(c.getDouble(i)) ===
+ java.lang.Double.doubleToRawLongBits(expected))
+ })),
+ ("readLongs", LongType, 8, (r, n, c) => r.readLongs(n, c, 0),
+ Some((c, i, b) => assert(c.getLong(i) === littleEndian(b)))),
+ ("readLongsAsInts", IntegerType, 8, (r, n, c) => r.readLongsAsInts(n, c,
0),
+ Some((c, i, b) => assert(c.getInt(i) === littleEndian(b).toInt))),
+ ("readDoubles", DoubleType, 8, (r, n, c) => r.readDoubles(n, c, 0),
+ Some((c, i, b) =>
+ assert(java.lang.Double.doubleToRawLongBits(c.getDouble(i)) ===
littleEndian(b)))),
+ ("readBinary", BinaryType, typeWidth, (r, n, c) => r.readBinary(n, c, 0),
+ Some(checkBinary)),
+ ("readFixedLenByteArray", BinaryType, typeWidth,
+ (r, n, c) => r.readFixedLenByteArray(n, typeWidth, c, 0),
Some(checkBinary)),
+ ("skipBytes", ByteType, typeWidth, (r, n, _) => r.skipBytes(n), None),
+ ("skipShorts", ShortType, typeWidth, (r, n, _) => r.skipShorts(n), None),
+ ("skipIntegers", IntegerType, typeWidth, (r, n, _) => r.skipIntegers(n),
None),
+ ("skipLongs", LongType, typeWidth, (r, n, _) => r.skipLongs(n), None),
+ ("skipFloats", FloatType, typeWidth, (r, n, _) => r.skipFloats(n), None),
+ ("skipDoubles", DoubleType, typeWidth, (r, n, _) => r.skipDoubles(n),
None),
+ ("skipBinary", BinaryType, typeWidth, (r, n, _) => r.skipBinary(n),
None),
+ ("skipFixedLenByteArray", BinaryType, typeWidth,
+ (r, n, _) => r.skipFixedLenByteArray(n, typeWidth), None)
+ ).collect {
+ case (name, dt, width, op, check) if width == typeWidth => (name, dt,
op, check)
+ }
+
+ /** Every single-value read of the reader that applies to this value width.
*/
+ private lazy val singleOps: Seq[(String, Reader => Any)] = Seq[(String, Int,
Reader => Any)](
+ ("readByte", 4, _.readByte()),
+ ("readShort", 4, _.readShort()),
+ ("readInteger", 4, _.readInteger()),
+ ("readFloat", 4, _.readFloat()),
+ ("readLong", 8, _.readLong()),
+ ("readDouble", 8, _.readDouble()),
+ ("readBinary(len)", typeWidth, _.readBinary(typeWidth))
+ ).collect { case (name, width, op) if width == typeWidth => (name, op) }
+
+ private def assertCorrupted(clue: String)(f: => Any)(implicit pos:
Position): Unit = {
+ withClue(clue) {
+ val e = intercept[ParquetDecodingException](f)
+ assert(e.getMessage.contains("Corrupted BYTE_STREAM_SPLIT page"))
+ }
+ }
+
// --- tests ---
test("batch read - sequential values") {
@@ -168,6 +257,141 @@ abstract class ParquetByteStreamSplitEncodingSuite[T:
ClassTag] extends SparkFun
cv.close()
}
}
+
+ test("batch reads and skips that end exactly at the end of the page") {
+ val data = Array.fill(10)(nextRandom)
+ val page = encode(data)
+ for ((name, dt, op, check) <- batchOps) {
+ withClue(name) {
+ val cv = new OnHeapColumnVector(data.length, dt)
+ try {
+ val reader = newReader(page, data.length)
+ skipBatch(reader, 4)
+ op(reader, 6, cv)
+ check.foreach { f =>
+ for (i <- 0 until 6) withClue(s"row $i")(f(cv, i, toBytes(data(4 +
i))))
+ }
+ // Nothing is left, but an empty read or skip is still allowed.
+ op(reader, 0, cv)
+ assertCorrupted("single read")(readSingle(reader))
+ } finally {
+ cv.close()
+ }
+ }
+ }
+ }
+
+ test("batch reads and skips past the end of the page fail") {
+ val data = Array.tabulate(10)(i => sequentialValue(i))
+ val page = encode(data)
+ for ((name, dt, op, _) <- batchOps) {
+ val cv = new OnHeapColumnVector(data.length, dt)
+ try {
+ assertCorrupted(name)(op(newReader(page, data.length), 11, cv))
+ val reader = newReader(page, data.length)
+ skipBatch(reader, 4)
+ assertCorrupted(name)(op(reader, 7, cv))
+ // offset + total would overflow to a negative number.
+ val reader2 = newReader(page, data.length)
+ skipBatch(reader2, 1)
+ assertCorrupted(name)(op(reader2, Int.MaxValue, cv))
+ withClue(name) {
+ intercept[IllegalArgumentException](op(newReader(page, data.length),
-1, cv))
+ }
+ } finally {
+ cv.close()
+ }
+ }
+ }
+
+ test("single-value reads past the end of the page fail") {
+ val data = Array.tabulate(10)(i => sequentialValue(i))
+ val page = encode(data)
+ for ((name, op) <- singleOps) {
+ val reader = newReader(page, data.length)
+ skipBatch(reader, 4)
+ for (_ <- 0 until 6) op(reader)
+ assertCorrupted(name)(op(reader))
+ }
+ }
+
+ test("read the last value of the page after a skip") {
+ val data = Array.tabulate(10)(i => sequentialValue(i))
+ val reader = newReader(encode(data), data.length)
+ skipBatch(reader, 9)
+ assertEqual(data(9), readSingle(reader), "mismatch at index 9")
+ }
+
+ test("page length that is not a multiple of the value width fails") {
+ val page = encode(Array.tabulate(10)(i => sequentialValue(i)))
+ for (extra <- 1 until typeWidth) {
+ assertCorrupted(s"extra $extra bytes") {
+ newReader(page ++ Array.fill[Byte](extra)(1), 10)
+ }
+ }
+ for (missing <- 1 until typeWidth) {
+ assertCorrupted(s"missing $missing
bytes")(newReader(page.dropRight(missing), 10))
+ }
+ }
+
+ test("more encoded values than the page value count fails") {
+ val page = encode(Array.tabulate(10)(i => sequentialValue(i)))
+ for (exact <- Seq(false, true)) {
+ assertCorrupted(s"exactValueCount = $exact") {
+ newReader(page, 9, exactValueCount = exact)
+ }
+ }
+ }
+
+ test("fewer encoded values than the page value count (nulls) is allowed") {
+ // For a column with definition levels the page value count includes
nulls, which are
+ // not encoded.
+ val data = Array.tabulate(10)(i => sequentialValue(i))
+ val reader = newReader(encode(data), data.length + 5)
+ val cv = new OnHeapColumnVector(data.length, sparkType)
+ try {
+ readBatch(reader, data.length, cv, 0)
+ for (i <- data.indices) assertEqual(data(i), getFromVector(cv, i),
s"mismatch at $i")
+ assertCorrupted("read past the encoded values")(readBatch(reader, 1, cv,
0))
+ } finally {
+ cv.close()
+ }
+ }
+
+ test("fewer encoded values than the exact page value count fails") {
+ // Without definition levels every row has a value, so the page value
count is exact.
+ val data = Array.tabulate(10)(i => sequentialValue(i))
+ val page = encode(data)
+ assertCorrupted("one value short")(newReader(page, 11, exactValueCount =
true))
+ assertCorrupted("empty page")(newReader(Array.emptyByteArray, 1,
exactValueCount = true))
+ // A matching count is accepted.
+ readAndVerify(data, exactValueCount = true)
+ newReader(Array.emptyByteArray, 0, exactValueCount = true)
+ }
+
+ test("single-value readBinary with a length other than the value width
fails") {
+ val data = Array.tabulate(10)(i => sequentialValue(i))
+ val reader = newReader(encode(data), data.length)
+ for (len <- Seq(typeWidth - 1, typeWidth + 1)) {
+ withClue(s"len =
$len")(intercept[IllegalArgumentException](reader.readBinary(len)))
+ }
+ // The rejected reads do not consume a value.
+ assertEqual(data(0), readSingle(reader), "mismatch at index 0")
+ }
+
+ test("empty page") {
+ val reader = newReader(Array.emptyByteArray, 3)
+ val cv = new OnHeapColumnVector(1, sparkType)
+ try {
+ readBatch(reader, 0, cv, 0)
+ skipBatch(reader, 0)
+ assertCorrupted("skip")(skipBatch(reader, 1))
+ assertCorrupted("batch read")(readBatch(reader, 1, cv, 0))
+ assertCorrupted("single read")(readSingle(reader))
+ } finally {
+ cv.close()
+ }
+ }
}
// --- Concrete suites ---
@@ -180,11 +404,20 @@ private object BssWriterHelper {
writer.close()
bytes
}
+
+ def encodeFLBA(data: Array[Array[Byte]], typeWidth: Int): Array[Byte] =
+ encode(
+ new FixedLenByteArrayByteStreamSplitValuesWriter(
+ typeWidth, data.length, data.length * typeWidth, new
DirectByteBufferAllocator())
+ )(w => data.foreach(b => w.writeBytes(Binary.fromConstantByteArray(b))))
}
class ParquetByteStreamSplitEncodingIntegerSuite
extends ParquetByteStreamSplitEncodingSuite[Int] {
+ override protected def toBytes(v: Int): Array[Byte] =
+ ByteBuffer.allocate(4).order(ByteOrder.LITTLE_ENDIAN).putInt(v).array()
+
override protected def typeWidth: Int = 4
override protected def sparkType: DataType = IntegerType
@@ -219,6 +452,9 @@ class ParquetByteStreamSplitEncodingIntegerSuite
class ParquetByteStreamSplitEncodingLongSuite
extends ParquetByteStreamSplitEncodingSuite[Long] {
+ override protected def toBytes(v: Long): Array[Byte] =
+ ByteBuffer.allocate(8).order(ByteOrder.LITTLE_ENDIAN).putLong(v).array()
+
override protected def typeWidth: Int = 8
override protected def sparkType: DataType = LongType
@@ -253,6 +489,9 @@ class ParquetByteStreamSplitEncodingLongSuite
class ParquetByteStreamSplitEncodingFloatSuite
extends ParquetByteStreamSplitEncodingSuite[Float] {
+ override protected def toBytes(v: Float): Array[Byte] =
+ ByteBuffer.allocate(4).order(ByteOrder.LITTLE_ENDIAN).putFloat(v).array()
+
override protected def typeWidth: Int = 4
override protected def sparkType: DataType = FloatType
@@ -294,6 +533,9 @@ class ParquetByteStreamSplitEncodingFloatSuite
class ParquetByteStreamSplitEncodingDoubleSuite
extends ParquetByteStreamSplitEncodingSuite[Double] {
+ override protected def toBytes(v: Double): Array[Byte] =
+ ByteBuffer.allocate(8).order(ByteOrder.LITTLE_ENDIAN).putDouble(v).array()
+
override protected def typeWidth: Int = 8
override protected def sparkType: DataType = DoubleType
@@ -332,14 +574,49 @@ class ParquetByteStreamSplitEncodingDoubleSuite
}
}
+/** Runs the shared test matrix for FIXED_LEN_BYTE_ARRAY with an odd width. */
+class ParquetByteStreamSplitEncodingFLBAWidth5Suite
+ extends ParquetByteStreamSplitEncodingSuite[Array[Byte]] {
+
+ override protected def typeWidth: Int = 5
+ override protected def sparkType: DataType = BinaryType
+ override protected def toBytes(v: Array[Byte]): Array[Byte] = v
+
+ override protected def encode(data: Array[Array[Byte]]): Array[Byte] =
+ BssWriterHelper.encodeFLBA(data, typeWidth)
+
+ override protected def readBatch(
+ r: VectorizedByteStreamSplitValuesReader,
+ total: Int, cv: WritableColumnVector, rowId: Int): Unit =
+ r.readFixedLenByteArray(total, typeWidth, cv, rowId)
+
+ override protected def skipBatch(
+ r: VectorizedByteStreamSplitValuesReader, total: Int): Unit =
+ r.skipFixedLenByteArray(total, typeWidth)
+
+ override protected def readSingle(r: VectorizedByteStreamSplitValuesReader):
Array[Byte] =
+ r.readBinary(typeWidth).getBytes
+
+ override protected def getFromVector(cv: WritableColumnVector, rowId: Int):
Array[Byte] =
+ cv.getBinary(rowId)
+
+ override protected def nextRandom: Array[Byte] = {
+ val b = new Array[Byte](typeWidth); random.nextBytes(b); b
+ }
+ override protected def sequentialValue(i: Int): Array[Byte] =
+ ByteBuffer.allocate(8).order(ByteOrder.LITTLE_ENDIAN).putLong(i *
7L).array()
+ .take(typeWidth)
+ override protected def extremeValues: Array[Array[Byte]] = Array(
+ Array.fill[Byte](typeWidth)(0), Array.fill[Byte](typeWidth)(-1),
+ Array.fill[Byte](typeWidth)(Byte.MinValue),
Array.fill[Byte](typeWidth)(Byte.MaxValue))
+ override protected def singleTestValue: Array[Byte] = Array[Byte](1, 2, 3,
4, 5)
+}
+
class ParquetByteStreamSplitEncodingFLBASuite extends SparkFunSuite {
private val random = new Random(42)
private def writeFLBA(data: Array[Array[Byte]], typeWidth: Int): Array[Byte]
=
- BssWriterHelper.encode(
- new FixedLenByteArrayByteStreamSplitValuesWriter(
- typeWidth, data.length, data.length * typeWidth, new
DirectByteBufferAllocator())
- )(w => data.foreach(b => w.writeBytes(Binary.fromConstantByteArray(b))))
+ BssWriterHelper.encodeFLBA(data, typeWidth)
private def readAndVerifyFLBA(data: Array[Array[Byte]], typeWidth: Int):
Unit = {
val page = writeFLBA(data, typeWidth)
@@ -372,11 +649,6 @@ class ParquetByteStreamSplitEncodingFLBASuite extends
SparkFunSuite {
Array.fill(200)(Array.fill(3)(random.nextInt(256).toByte)), typeWidth =
3)
}
- test("read FLBA - width 5") {
- readAndVerifyFLBA(
- Array.fill(150)(Array.fill(5)(random.nextInt(256).toByte)), typeWidth =
5)
- }
-
test("read FLBA - width 7") {
readAndVerifyFLBA(
Array.fill(120)(Array.fill(7)(random.nextInt(256).toByte)), typeWidth =
7)
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetColumnIndexSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetColumnIndexSuite.scala
index e7d5583ee561..d799b501c48e 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetColumnIndexSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetColumnIndexSuite.scala
@@ -20,6 +20,7 @@ package org.apache.spark.sql.execution.datasources.parquet
import scala.jdk.CollectionConverters._
import org.apache.hadoop.fs.Path
+import org.apache.parquet.column.Encoding
import org.apache.parquet.hadoop.ParquetFileReader
import org.apache.parquet.hadoop.util.HadoopInputFile
@@ -193,4 +194,60 @@ class ParquetColumnIndexSuite extends ParquetTest with
SharedSparkSession {
}
}
}
+
+ test("SPARK-59831: reading unaligned pages - BYTE_STREAM_SPLIT") {
+ // BYTE_STREAM_SPLIT is enabled for the floating point columns, whose
pages are split by
+ // size. The pages of _1 are split by size with v1 pages, where it is
PLAIN encoded, and by
+ // row count with v2 pages, where its DELTA_BINARY_PACKED values are tiny.
With v1 pages
+ // the DOUBLE columns buffer as many bytes per row as _1 and get the same
pages, so only
+ // the FLOAT columns are unaligned with _1. With v2 pages all of them are.
In the unaligned
+ // columns, the rows outside the row ranges are skipped within the
BYTE_STREAM_SPLIT pages.
+ // _2 and _3 are required, and _4 and _5 have nulls.
+ val df = spark.range(0, 2000).selectExpr(
+ "id as _1",
+ "cast(id as float) as _2",
+ "cast(id as double) as _3",
+ "if(id >= 400 and id < 450, null, cast(id as float)) as _4",
+ "if(id % 3 = 0, null, cast(id as double)) as _5")
+ Seq("PARQUET_1_0", "PARQUET_2_0").foreach { version =>
+ withTempPath { file =>
+ df.coalesce(1)
+ .write
+ .option("parquet.writer.version", version)
+ .option("parquet.enable.dictionary", "false")
+ .option("parquet.enable.bytestreamsplit", "true")
+ .option("parquet.page.size", "1024")
+ .option("parquet.page.row.count.limit", "500")
+ .parquet(file.getCanonicalPath)
+
+ val parquetFile =
file.listFiles().filter(_.getName.startsWith("part")).head
+ val in = HadoopInputFile.fromPath(
+ new Path(parquetFile.getCanonicalPath),
spark.sessionState.newHadoopConf())
+ Utils.tryWithResource(ParquetFileReader.open(in)) { reader =>
+ val columns = reader.getFooter.getBlocks.get(0).getColumns.asScala
+ .map(c => c.getPath.toDotString -> c).toMap
+ def pageFirstRows(name: String): Set[Long] = {
+ val offsetIndex = reader.readOffsetIndex(columns(name))
+ (0 until
offsetIndex.getPageCount).map(offsetIndex.getFirstRowIndex).toSet
+ }
+ Seq("_2", "_3", "_4", "_5").foreach { name =>
+
assert(columns(name).getEncodings.contains(Encoding.BYTE_STREAM_SPLIT), name)
+ }
+ val filterPages = pageFirstRows("_1")
+ assert(filterPages.size > 1)
+ val unaligned =
+ if (version == "PARQUET_1_0") Seq("_2", "_4") else Seq("_2", "_3",
"_4", "_5")
+ unaligned.foreach { name =>
+ // Some row range starts or ends inside a page of this column.
+ assert(!filterPages.subsetOf(pageFirstRows(name)), s"$version
$name")
+ }
+ }
+
+ val parquetDf = spark.read.parquet(file.getCanonicalPath)
+ actions.foreach { action =>
+ checkAnswer(action(parquetDf), action(df))
+ }
+ }
+ }
+ }
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetEncodingSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetEncodingSuite.scala
index 9107be919b31..c8897708751b 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetEncodingSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetEncodingSuite.scala
@@ -26,8 +26,9 @@ import org.apache.hadoop.fs.Path
import org.apache.parquet.column.{Encoding, ParquetProperties}
import org.apache.parquet.column.ParquetProperties.WriterVersion.PARQUET_1_0
import org.apache.parquet.example.data.simple.SimpleGroup
-import org.apache.parquet.hadoop.ParquetOutputFormat
+import org.apache.parquet.hadoop.{ParquetFileReader, ParquetOutputFormat}
import org.apache.parquet.hadoop.example.ExampleParquetWriter
+import org.apache.parquet.hadoop.util.HadoopInputFile
import org.apache.parquet.io.api.Binary
import org.apache.parquet.schema.MessageTypeParser
import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName
@@ -38,6 +39,7 @@ import org.apache.spark.sql.Row
import org.apache.spark.sql.catalyst.util.DateTimeUtils
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.util.Utils
// TODO: this needs a lot more testing but it's currently not easy to test
with the parquet
// writer abstractions. Revisit.
@@ -454,6 +456,103 @@ class ParquetEncodingSuite extends
ParquetCompatibilityTest with SharedSparkSess
}
}
+ test("SPARK-59831: BYTE_STREAM_SPLIT round-trip with v1 and v2 pages, nulls
and arrays") {
+ // Required columns (max definition level 0) exercise the exact page value
count check,
+ // and optional and repeated columns the upper bound check, on many pages
of both page
+ // versions.
+ val schema = MessageTypeParser.parseMessageType(
+ """message root {
+ | required int32 int_col;
+ | required int64 long_col;
+ | required float float_col;
+ | required double double_col;
+ | required fixed_len_byte_array(5) flba_col;
+ | optional int32 int_nullable;
+ | optional double double_nullable;
+ | optional group arr (LIST) {
+ | repeated group list {
+ | optional float element;
+ | }
+ | }
+ |}
+ """.stripMargin)
+ val size = 5000
+ def flba(i: Int): Array[Byte] = {
+ val v = i * 7919L
+ Array.tabulate[Byte](5)(b => (v >> (8 * b)).toByte)
+ }
+ def arr(i: Int): Seq[java.lang.Float] = i % 7 match {
+ case 0 => null
+ case 1 => Seq.empty
+ case _ => Seq(i * 0.25f, if (i % 2 == 0) null else -i * 0.25f)
+ }
+ val expected = (0 until size).map { i =>
+ Row(i * 7919, i * 1000003L * (if (i % 2 == 0) 1 else -1), i * 0.1f, i *
-0.001, flba(i),
+ if (i % 3 == 0) null else i * 7, if (i % 5 == 0) null else i * 0.5,
arr(i))
+ }
+
+ Seq(ParquetProperties.WriterVersion.PARQUET_1_0,
+ ParquetProperties.WriterVersion.PARQUET_2_0).foreach { version =>
+ withTempDir { dir =>
+ val path = new Path(dir.toURI.toString, "bss.parquet")
+ val hadoopConf = spark.sessionState.newHadoopConf()
+ val builder = ExampleParquetWriter.builder(path)
+ .withType(schema)
+ .withDictionaryEncoding(false)
+ .withPageRowCountLimit(300)
+ .withWriterVersion(version)
+ .withConf(hadoopConf)
+ schema.getColumns.asScala.foreach { column =>
+ builder.withByteStreamSplitEncoding(column.getPath.mkString("."),
true)
+ }
+ val writer = builder.build()
+ try {
+ expected.foreach { row =>
+ val record = new SimpleGroup(schema)
+ record.add("int_col", row.getInt(0))
+ record.add("long_col", row.getLong(1))
+ record.add("float_col", row.getFloat(2))
+ record.add("double_col", row.getDouble(3))
+ record.add("flba_col",
Binary.fromConstantByteArray(row.getAs[Array[Byte]](4)))
+ if (!row.isNullAt(5)) record.add("int_nullable", row.getInt(5))
+ if (!row.isNullAt(6)) record.add("double_nullable",
row.getDouble(6))
+ if (!row.isNullAt(7)) {
+ val list = record.addGroup("arr")
+ row.getSeq[java.lang.Float](7).foreach { v =>
+ val element = list.addGroup("list")
+ if (v != null) element.add("element", v.floatValue())
+ }
+ }
+ writer.write(record)
+ }
+ } finally {
+ writer.close()
+ }
+
+ val in = HadoopInputFile.fromPath(path, hadoopConf)
+ Utils.tryWithResource(ParquetFileReader.open(in)) { reader =>
+ val columnChunks =
reader.getFooter.getBlocks.asScala.head.getColumns.asScala
+ assert(columnChunks.length === 8)
+ columnChunks.foreach { chunk =>
+ assert(chunk.getEncodings.contains(Encoding.BYTE_STREAM_SPLIT),
+ s"Column ${chunk.getPath} should use BYTE_STREAM_SPLIT encoding")
+ assert(reader.readOffsetIndex(chunk).getPageCount > 1,
+ s"Column ${chunk.getPath} should have multiple pages")
+ }
+ }
+
+ withMemoryModes { offHeapMode =>
+ withSQLConf(
+ SQLConf.COLUMN_VECTOR_OFFHEAP_ENABLED.key -> offHeapMode,
+ SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "true",
+ SQLConf.PARQUET_VECTORIZED_READER_NESTED_COLUMN_ENABLED.key ->
"true") {
+ checkAnswer(spark.read.parquet(path.toString), expected)
+ }
+ }
+ }
+ }
+ }
+
test("PLAIN-encoded FIXED_LEN_BYTE_ARRAY round-trip (dictionary disabled)") {
// Regression test: the FixedLenByteArrayUpdater batch read path must not
use
// the length-prefixed readBinary(total, c, rowId) method which is
designed for
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorizedSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorizedSuite.scala
index a78403a30716..916884580068 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorizedSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorizedSuite.scala
@@ -22,10 +22,12 @@ import java.util.{Optional, PrimitiveIterator}
import scala.collection.mutable.ArrayBuffer
import scala.language.implicitConversions
-import org.apache.parquet.column.{ColumnDescriptor, ParquetProperties}
+import org.apache.parquet.bytes.BytesInput
+import org.apache.parquet.column.{ColumnDescriptor, Encoding,
ParquetProperties}
import org.apache.parquet.column.impl.ColumnWriteStoreV1
import org.apache.parquet.column.page._
import org.apache.parquet.column.page.mem.MemPageStore
+import org.apache.parquet.column.statistics.Statistics
import org.apache.parquet.io.ParquetDecodingException
import org.apache.parquet.io.api.Binary
import org.apache.parquet.schema.{MessageType, MessageTypeParser}
@@ -481,6 +483,47 @@ 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)
+ // BYTE_STREAM_SPLIT stores byte b of value i at b * (number of values) +
i.
+ val values = Seq(1.0f, 2.0f, 3.0f)
+ val data = new Array[Byte](4 * values.length)
+ values.zipWithIndex.foreach { case (v, i) =>
+ val bits = java.lang.Float.floatToIntBits(v)
+ (0 until 4).foreach(b => data(b * values.length + i) = (bits >> (8 *
b)).toByte)
+ }
+
+ 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.RLE, Encoding.RLE, Encoding.BYTE_STREAM_SPLIT)
+ }
+ val recordReader = new VectorizedParquetRecordReader(
+ DateTimeUtils.getZoneId("EST"), "CORRECTED", "UTC", "CORRECTED",
"UTC", true, 2)
+ try {
+ recordReader.initialize(parquetSchema, parquetSchema,
+ TestParquetRowGroupReader(Seq(TestPageReadStore(memPageStore,
Seq(0L)))), 4)
+ val e =
intercept[ParquetDecodingException](recordReader.nextKeyValue())
+ assert(e.getMessage.contains("3 encoded values do not match the page
value count 4"),
+ s"page v2: $pageV2")
+ } finally {
+ recordReader.close()
+ }
+ }
+ }
+
truncateTypeTest("primitive type", IntegerType, LongType, IntegerType)
truncateTypeTest("basic struct",
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
