dongjoon-hyun commented on code in PR #59118:
URL: https://github.com/apache/spark/pull/59118#discussion_r4147219003
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetColumnIndexSuite.scala:
##########
@@ -136,4 +136,35 @@ 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.
Either way they
+ // are unaligned with the BYTE_STREAM_SPLIT pages, so the rows outside the
row ranges are
+ // skipped within those 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 parquetDf = spark.read.parquet(file.getCanonicalPath)
+ actions.foreach { action =>
+ checkAnswer(action(parquetDf), action(df))
Review Comment:
This test only runs `checkAnswer`, so it keeps passing if its premises stop
holding:
- If `parquet.enable.bytestreamsplit` did not reach the writer, the float
and double columns would be PLAIN with the same page layout, and none of the
new skip checks would run.
- Without `parquet.page.row.count.limit`, `_1` would be a single page with
v2, and nothing would be skipped.
On master, SPARK-59837 added
`assert(reader.readOffsetIndex(column).getPageCount > 1)` to
`checkUnalignedPages` for this reason, but this test does not go through that
helper (and cannot as is, since the helper takes no extra options).
Could we assert here that `_2` to `_5` contain `Encoding.BYTE_STREAM_SPLIT`
and that `_1` has more than one page? After a rebase onto master, most of the
imports for this are already in the file. The new round-trip test in
`ParquetEncodingSuite` asserts the encoding but not the `many pages` part of
its comment, so a page count assertion would help there too.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedByteStreamSplitValuesReader.java:
##########
@@ -69,7 +82,23 @@ 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();
+ // 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 length " +
+ totalBytes + " is not a multiple of the value width " + typeWidth);
+ }
this.valueCount = totalBytes / typeWidth;
+ // Without definition levels the page value count must match the encoded
values exactly.
+ // Otherwise it includes nulls and is only an upper bound, so a page that
is off by whole
+ // values is not detected here: its values are decoded with the wrong
stream stride, and
+ // only a read or skip past the last encoded value fails.
+ if (exactValueCount ? this.valueCount != valueCount : this.valueCount >
valueCount) {
Review Comment:
Not a blocker, and fine as a follow-up. For a column with definition levels
the remaining checks are one-directional. Let D be the number of non-null
definition levels and E the number of encoded values. E < D fails once a read
or skip runs past the last encoded value, as the comment says. But E > D is
never detected, even when the page is read to its end.
For example, an optional FLOAT page with `num_values = 10`, 5 non-null
levels and 24 data bytes (E = 6) passes `initFromPage`, 5 values are read with
a stride of 6, and the page ends with `offset = 5 < valueCount = 6` without any
error. If the data section is the corrupt part, those 5 values are assembled
from the wrong bytes.
This could be closed without `num_nulls`. When
`VectorizedColumnReader.readBatch` has consumed all levels of a page
(`readState.valuesToReadInPage == 0` right after the `defColumn.readBatch(...)`
/ `repColumn.readBatchRepeated(...)` calls), all encoded values must have been
consumed too. A no-op default hook on `VectorizedValuesReader` that this reader
implements as `offset == valueCount` would catch it with one call per page.
Checking only before the next `readPage()` would miss the last page of a column
chunk.
As far as I can tell from reading `VectorizedRleValuesReader`, this always
holds for valid files: rows outside the row ranges are consumed through
`skipValues`, and the repeated path flushes its pending definition levels
before returning. I have not tested it, though. It is still a late check, like
the existing one, so a query that stops early would not reach it.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java:
##########
@@ -389,7 +389,8 @@ private ValuesReader getValuesReader(Encoding encoding) {
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);
Review Comment:
This line is the only production wiring of the exact check, and I think no
test pins it. Replacing the second argument with `false`, or going back to the
1-arg constructor (which still exists with the lenient default), keeps every
test green. The unit tests construct the reader themselves, and the end-to-end
tests only read valid files, which behave the same in both modes. The opposite
mistakes (`true` or `!= 0`) would be caught by the nullable columns, so only
this direction is open.
Could we add one negative test that goes through `VectorizedColumnReader`?
`ParquetVectorizedSuite` already has an in-memory harness (`MemPageStore` +
`TestParquetRowGroupReader`), and writing a page directly with
`memPageStore.getPageWriter(cd).writePage(...)` should allow a required
BYTE_STREAM_SPLIT page whose value count is off by one.
One caveat: with a small page read in a single batch, the lenient mode also
fails with a `Corrupted BYTE_STREAM_SPLIT page` message from `checkRemaining`.
So the test would need to assert the init message (`encoded values do not match
the page value count`) or use a batch size smaller than the page.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetColumnIndexSuite.scala:
##########
@@ -136,4 +136,35 @@ 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.
Either way they
Review Comment:
nit: `Either way they are unaligned` does not seem to hold for the double
columns with v1 pages.
With v1, `_1` (PLAIN INT64) and `_3` (BYTE_STREAM_SPLIT DOUBLE) are both
required and buffer 8 bytes per row, and parquet-java checks all columns at the
same row counts (every 100 rows here), so both get 200-row pages. `_5` also
ends up with 200-row pages, and only `_2`/`_4` get 300-row pages. So in the v1
iteration no rows are skipped inside a double page, and a skip over the nulls
of `_4` (rows 400 to 449) only happens with v2, where `_1` has 500-row pages.
This is from reading `ColumnWriteStoreBase.sizeCheck`, not from running the
test.
The test still does its job, since the float columns are skipped into with
v1 and all four columns with v2. Maybe the comment could say that with v1 pages
only the FLOAT columns are unaligned.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedByteStreamSplitValuesReader.java:
##########
@@ -69,7 +82,23 @@ 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();
+ // 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 length " +
+ totalBytes + " is not a multiple of the value width " + typeWidth);
+ }
this.valueCount = totalBytes / typeWidth;
Review Comment:
nit: The check below compares the field `this.valueCount` (number of encoded
values) with the parameter `valueCount` (page value count, including nulls),
and the field is assigned before the check can throw. Computing a local first
would make the condition easier to read and would leave the reader untouched
when the page is rejected:
```java
int encodedValues = totalBytes / typeWidth;
if (exactValueCount ? encodedValues != valueCount : encodedValues >
valueCount) {
throw ...
}
this.valueCount = encodedValues;
```
The field Javadoc (`Total number of values in the current page`) could then
say that it is the number of encoded (non-null) values. There is no behavior
change, since a reader whose `initFromPage` failed is never used again.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedByteStreamSplitValuesReader.java:
##########
@@ -115,38 +166,39 @@ public short readShort() {
@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];
+ int idx = nextIndex();
for (int b = 0; b < len; b++) {
- result[b] = pageData[b * valueCount + offset];
+ result[b] = pageData[b * valueCount + idx];
Review Comment:
nit (pre-existing, not from this PR): `readBinary(int len)` uses the caller
`len` as the number of streams. With `len > typeWidth` this fails with a raw
`ArrayIndexOutOfBoundsException` after `nextIndex()` has already consumed the
value, and with `len < typeWidth` it silently returns a prefix. The current
callers always pass the type length of the same column descriptor, so this is
not reachable today.
Since this PR makes the other reads fail cleanly, maybe use `typeWidth` here
like the batch `readBinary` does, or reject `len != typeWidth` in the same way
as a negative count.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetByteStreamSplitEncodingSuite.scala:
##########
@@ -332,6 +563,52 @@ 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] =
Review Comment:
nit (optional): This body is identical to `writeFLBA` in
`ParquetByteStreamSplitEncodingFLBASuite` below. A shared
`BssWriterHelper.encodeFLBA(data, typeWidth)` would avoid the second copy.
Also, `read FLBA - width 5` in the standalone suite now takes the same path as
the shared `single-value read` test at width 5.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetByteStreamSplitEncodingSuite.scala:
##########
@@ -332,6 +563,52 @@ 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.encode(
+ new FixedLenByteArrayByteStreamSplitValuesWriter(
+ typeWidth, data.length, data.length * typeWidth, new
DirectByteBufferAllocator())
+ )(w => data.foreach(b => w.writeBytes(Binary.fromConstantByteArray(b))))
+
+ 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)
+
+ override protected def assertEqual(
Review Comment:
nit: This override does not seem to be needed. The base `assert(expected ===
actual, msg)` already compares arrays element by element. ScalaTest default
equality handles arrays structurally, and `ParquetEncodingSuite` relies on it
in `assert(row.getAs[Array[Byte]](4) === expectedFlba, ...)`.
If you prefer to keep it for the failure message, the comment on the base
hook (`Override for types that need bitwise comparison (Float, Double)`) could
be widened.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetByteStreamSplitEncodingSuite.scala:
##########
@@ -332,6 +563,52 @@ class ParquetByteStreamSplitEncodingDoubleSuite
}
}
+/** Runs the shared test matrix for FIXED_LEN_BYTE_ARRAY with an odd width. */
+class ParquetByteStreamSplitEncodingFLBAWidth5Suite
Review Comment:
nit: With this subclass, the class scaladoc of
`ParquetByteStreamSplitEncodingSuite` (L39-43) is out of date. It says the
shared matrix is `for all numeric types` and that `FLBA is tested in a
standalone suite because its reader API differs`. Could we update it to mention
FLBA of width 5 and the new bounds tests? (Commenting here because those lines
are not part of the diff.)
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetByteStreamSplitEncodingSuite.scala:
##########
@@ -168,6 +253,140 @@ 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 empty reads and skips are still allowed.
+ op(reader, 0, cv)
+ skipBatch(reader, 0)
+ readBatch(reader, 0, cv, 0)
+ assertCorrupted(name)(readSingle(reader))
Review Comment:
nits in these tests:
- `assertCorrupted(name)` is already inside `withClue(name)` (L261), so the
clue is printed twice on failure. Maybe `assertCorrupted("single read")` as in
the `empty page` test.
- `skipBatch(reader, 0)` and `readBatch(reader, 0, cv, 0)` (L272-273) repeat
what `op(reader, 0, cv)` does on the iteration of the corresponding entry, in
the same reader state, so they could be dropped.
- In `page length that is not a multiple of the value width fails`, the
`extra` loop passes a count of 11. So like the `missing` loop, it is a page
that is short for its declared count. A count of 10 would make it a real
trailing-bytes case.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetByteStreamSplitEncodingSuite.scala:
##########
@@ -168,6 +253,140 @@ 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 empty reads and skips are still allowed.
+ op(reader, 0, cv)
+ skipBatch(reader, 0)
+ readBatch(reader, 0, cv, 0)
+ assertCorrupted(name)(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), 11)
+ }
+ }
+ 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))
+
+ val reader = newReader(page, data.length, exactValueCount = true)
Review Comment:
nit (optional): These lines repeat the body of `readAndVerify`. Since the
reader only looks at `exactValueCount` in `initFromPage`, `readAndVerify` with
an extra `exactValueCount` parameter, or just `newReader(page, data.length,
exactValueCount = true)` like the empty-page line below, would cover the same
thing.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetEncodingSuite.scala:
##########
@@ -454,6 +454,101 @@ 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),
Review Comment:
nit (optional): The value formulas and null conditions are written twice,
here and in the writer loop below (only `flba` and `arr` are shared). Building
the records from `expected`, or sharing small helper functions, would keep a
single source. A mismatch would fail `checkAnswer`, so this is only about
maintenance.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedByteStreamSplitValuesReader.java:
##########
@@ -81,6 +110,28 @@ public void initFromPage(int valueCount,
ByteBufferInputStream in) throws IOExce
// --------------- 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);
Review Comment:
Thank you for applying this. One small thing on the PR description: the
sentence about the per-value A/B microbenchmark (`one compare per value ... no
measurable difference`) was written for the previous form of `nextIndex()`.
`checkRemaining` now has two branches. The C2 compiler should fold the negative
check for the constant 1 once it is inlined, so I do not expect a difference.
Could you confirm that the result still holds on this commit?
As a side note, `VectorizedByteStreamSplitReaderBenchmark` has no per-value
case of the Spark reader for `readInteger`/`readLong`/`readDouble` (only
`readBinary` for FLBA), so this path is not tracked there.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetByteStreamSplitEncodingSuite.scala:
##########
@@ -79,6 +81,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]
Review Comment:
nit (optional, feel free to ignore): `toBytes` has a single call site
(L268), and `littleEndian` does what `BytesUtils.bytesToInt` / `bytesToLong`
from parquet already do for 4 and 8 bytes. `encode(Array(v))` would give the
same bytes as `toBytes(v)`, since a page with one value is the little-endian
bytes of that value.
That said, the explicit `ByteBuffer` version keeps the expected bytes
independent of the parquet writer, so keeping it is fine too.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedByteStreamSplitValuesReader.java:
##########
@@ -69,7 +82,23 @@ 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();
+ // 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 length " +
Review Comment:
nit on the wording of the two messages:
- `page length`: for a column with definition levels and v1 pages,
`totalBytes` is what is left after the level bytes, so the number does not
match a size in the page header. Maybe `page data length`, since the existing
comment above already says `the page data section`.
- `do not match` (L100): when the count is only an upper bound, the rule is
that the encoded values must not exceed it, but the message reads as if they
had to be equal. The previous `exceed` wording was more precise for that mode.
Unrelated to this PR: like in the sibling readers, these unchecked
exceptions bypass the `IOException` wrappers in `VectorizedColumnReader`, so
the message has no column name. That could be a separate improvement.
--
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]
