viirya commented on code in PR #59117:
URL: https://github.com/apache/spark/pull/59117#discussion_r4232804741
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala:
##########
@@ -2480,6 +2484,93 @@ class ParquetIOSuite extends ParquetTest with
SharedSparkSession {
}
}
}
+
+ /**
+ * Writes 100 rows of `column` to a single uncompressed data page, lets
`corrupt` modify the
+ * RLE section that starts with `03 00 00 00 C8 01 01` (the 4-byte length 3,
then one RLE run
+ * with header 100 << 1 and value 1), and returns the error of reading the
file with the
+ * vectorized reader. For a nullable INT32 column that is the definition
levels; for a
+ * BOOLEAN column written as a V2 page it is the RLE-encoded values.
+ */
+ private def readCorruptedRleSection(
+ column: String,
+ writerVersion: String = "PARQUET_1_0")(
+ corrupt: (Array[Byte], Int) => Unit): Throwable = {
+ var error: Throwable = null
+ withTempPath { dir =>
+ val path = dir.getCanonicalPath
+ spark.range(100).selectExpr(column).coalesce(1)
+ .write
+ .option(ParquetOutputFormat.ENABLE_DICTIONARY, "false")
+ .option(ParquetOutputFormat.WRITER_VERSION, writerVersion)
+ .option("compression", "uncompressed")
+ .parquet(path)
+ val file = dir.listFiles().filter(_.getName.endsWith(".parquet")).head
+ val bytes = Files.readAllBytes(file.toPath)
+ val section = Array[Byte](3, 0, 0, 0, 0xC8.toByte, 0x01, 1)
+ val offsets = bytes.indices.filter { i =>
Review Comment:
Done, with `indexOfSlice` and the uniqueness check you suggested.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala:
##########
@@ -2480,6 +2484,93 @@ class ParquetIOSuite extends ParquetTest with
SharedSparkSession {
}
}
}
+
+ /**
+ * Writes 100 rows of `column` to a single uncompressed data page, lets
`corrupt` modify the
+ * RLE section that starts with `03 00 00 00 C8 01 01` (the 4-byte length 3,
then one RLE run
+ * with header 100 << 1 and value 1), and returns the error of reading the
file with the
+ * vectorized reader. For a nullable INT32 column that is the definition
levels; for a
+ * BOOLEAN column written as a V2 page it is the RLE-encoded values.
+ */
+ private def readCorruptedRleSection(
+ column: String,
+ writerVersion: String = "PARQUET_1_0")(
+ corrupt: (Array[Byte], Int) => Unit): Throwable = {
+ var error: Throwable = null
+ withTempPath { dir =>
+ val path = dir.getCanonicalPath
+ spark.range(100).selectExpr(column).coalesce(1)
+ .write
+ .option(ParquetOutputFormat.ENABLE_DICTIONARY, "false")
+ .option(ParquetOutputFormat.WRITER_VERSION, writerVersion)
+ .option("compression", "uncompressed")
+ .parquet(path)
+ val file = dir.listFiles().filter(_.getName.endsWith(".parquet")).head
+ val bytes = Files.readAllBytes(file.toPath)
+ val section = Array[Byte](3, 0, 0, 0, 0xC8.toByte, 0x01, 1)
+ val offsets = bytes.indices.filter { i =>
+ bytes.slice(i, i + section.length).sameElements(section)
+ }
+ assert(offsets.size == 1)
+ corrupt(bytes, offsets.head)
+ Files.write(file.toPath, bytes)
+ dir.listFiles().filter(_.getName.endsWith(".crc")).foreach(_.delete())
+
+ withSQLConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "true") {
+ // Before SPARK-59832 a truncated read never returned, so run it with
a timeout.
+ // scalastyle:off awaitresult
+ error = intercept[SparkException] {
+ Await.result(
+
Future(spark.read.parquet(path).collect())(ExecutionContext.global), 1.minute)
+ }
+ // scalastyle:on awaitresult
+ }
+ }
+ error
+ }
+
+ private def causes(e: Throwable): Seq[Throwable] =
Review Comment:
Done.
--
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]
