viirya commented on code in PR #59360:
URL: https://github.com/apache/spark/pull/59360#discussion_r4250458249
##########
core/src/test/scala/org/apache/spark/scheduler/ReplayListenerSuite.scala:
##########
@@ -207,6 +207,56 @@ class ReplayListenerSuite extends SparkFunSuite with
BeforeAndAfter with LocalSp
assert(diagnostic.length < 1200)
}
+ test("SPARK-60158: Line buffer range appends match single-character
appends") {
+ val alphabet = Array('x', '\r', 0x00e9.toChar, 0x4e2d.toChar,
0xd83d.toChar, 0xde00.toChar)
+ val random = new scala.util.Random(60158)
+ for (_ <- 0 until 2000) {
+ val chars =
Array.fill(random.nextInt(64))(alphabet(random.nextInt(alphabet.length)))
+ val limit = random.nextInt(80)
+ val single = new ReplayListenerBus.BoundedLineBuffer(limit)
+ chars.foreach(single.append)
+ val ranged = new ReplayListenerBus.BoundedLineBuffer(limit)
+ var start = 0
+ while (start < chars.length) {
+ val end = math.min(chars.length, start + random.nextInt(8))
+ ranged.append(chars, start, end)
+ start = end
+ }
+ withClue(s"chars=${chars.map(_.toInt).mkString(",")} limit=$limit: ") {
+ assert(ranged.result() == single.result())
+ assert(ranged.length == single.length)
+ assert((ranged.capacity == 0) == (single.capacity == 0))
+ }
+ }
+ }
+
+ test("SPARK-60158: Replay splits lines across read buffers like a reference
splitter") {
+ val alphabet = Seq("x", "\r", 0x00e9.toChar.toString,
0x4e2d.toChar.toString,
+ new String(Character.toChars(0x1f600)))
+ val random = new scala.util.Random(60158)
+ for (_ <- 0 until 50) {
+ val lines = Seq.fill(random.nextInt(20)) {
+ val length = if (random.nextInt(4) == 0) random.nextInt(20000) else
random.nextInt(100)
+ Seq.fill(length)(alphabet(random.nextInt(alphabet.length))).mkString
+ }
+ val data = lines.mkString("\n") + (if (random.nextBoolean()) "\n" else
"")
+ val limit = 1 + random.nextInt(30000)
+ // Reference: split on '\n', drop one trailing CR, keep lines within the
byte limit.
+ val expected = (if (data.isEmpty) Seq.empty else data.split("\n",
-1).toSeq)
+ .dropRight(if (data.endsWith("\n")) 1 else 0)
+ .map(line => if (line.endsWith("\r")) line.dropRight(1) else line)
+ .filter(_.getBytes(StandardCharsets.UTF_8).length <= limit)
+ val seen = new ArrayBuffer[String]
+ assert(new ReplayListenerBus(limit).replay(
+ new ByteArrayInputStream(data.getBytes(StandardCharsets.UTF_8)),
"reference",
Review Comment:
Added in 714c458a1f4. The reference splitter test now also runs each input
through a stream that returns 1 to N bytes per `read` (N is 1, 3, 64 or 8192)
and reports `available() == 0`, so chunks end at arbitrary points, including
right after a `\r` and right before or at a `\n`. A mutation that does not
treat a `\n` at the start of a new chunk as a line end passes all the existing
tests but fails this one.
##########
core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala:
##########
@@ -275,19 +294,44 @@ private[spark] object ReplayListenerBus {
def append(c: Char): Unit = {
Review Comment:
Done in 714c458a1f4, together with [3]. `append(c)` and the range `append`
are replaced by a single `appendUntilNewline(chars, start, end)`, which is what
`fetchLine` calls, so there is one copy of the limit and release logic. The
SPARK-59804 buffer test now uses it, and the equivalence test is replaced by
one that splits random lines at random points and compares the result with a
reference (drop one trailing CR, keep the line if its UTF-8 length fits), and
checks that the buffer is released once the line exceeds the limit plus the CR
allowance.
##########
core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala:
##########
@@ -116,15 +130,20 @@ private[spark] class ReplayListenerBus(
@tailrec private def fetchLine(): (String, Int) = {
val buffer = new BoundedLineBuffer(effectiveMaxLineLength)
- var c = reader.read()
- if (c == -1) {
+ if (!fill()) {
null
} else {
val index = lineIndex
lineIndex += 1
- while (c != -1 && c != '\n') {
- buffer.append(c.toChar)
- c = reader.read()
+ var endOfLine = false
+ while (!endOfLine && fill()) {
+ var i = pos
+ while (i < len && chars(i) != '\n') {
Review Comment:
Changed in 714c458a1f4: `appendUntilNewline` finds the `\n` and adds up the
UTF-8 width in the same loop, and once the line goes over the limit it only
searches for `\n`. The full-replay numbers stay the same (lz4 0.22 s, zstd 0.19
s, vs 0.20 s / 0.17 s for `Source.getLines()`).
##########
core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala:
##########
@@ -116,15 +130,20 @@ private[spark] class ReplayListenerBus(
@tailrec private def fetchLine(): (String, Int) = {
val buffer = new BoundedLineBuffer(effectiveMaxLineLength)
Review Comment:
I'd prefer to leave this out. The numbers are already on par with
`Source.getLines()`, and the fast path would need its own handling of the
trailing CR and the limit, which would be a second place to keep in sync with
`BoundedLineBuffer`.
##########
core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala:
##########
@@ -89,13 +89,27 @@ private[spark] class ReplayListenerBus(
val decoder = StandardCharsets.UTF_8.newDecoder()
.onMalformedInput(CodingErrorAction.REPORT)
.onUnmappableCharacter(CodingErrorAction.REPORT)
- val reader = new BufferedReader(new InputStreamReader(logData, decoder))
+ // Read decoded characters in bulk and scan them for '\n', instead of
calling read() per
+ // character on a BufferedReader, which takes a lock on every call.
+ val reader = new InputStreamReader(logData, decoder)
Review Comment:
Thanks. I'd keep decoding every line for now: SPARK-59804 kept failing the
replay on malformed UTF-8 inside a skipped line (`Replay still rejects
malformed UTF-8 in skipped lines`), and splitting on raw bytes would drop that.
Raw-byte splitting also comes up for an incomplete UTF-8 sequence at EOF
(SPARK-60114), so I'll look at it there if the remaining overhead matters.
##########
core/src/test/scala/org/apache/spark/scheduler/ReplayListenerSuite.scala:
##########
@@ -207,6 +207,56 @@ class ReplayListenerSuite extends SparkFunSuite with
BeforeAndAfter with LocalSp
assert(diagnostic.length < 1200)
}
+ test("SPARK-60158: Line buffer range appends match single-character
appends") {
+ val alphabet = Array('x', '\r', 0x00e9.toChar, 0x4e2d.toChar,
0xd83d.toChar, 0xde00.toChar)
+ val random = new scala.util.Random(60158)
+ for (_ <- 0 until 2000) {
+ val chars =
Array.fill(random.nextInt(64))(alphabet(random.nextInt(alphabet.length)))
+ val limit = random.nextInt(80)
+ val single = new ReplayListenerBus.BoundedLineBuffer(limit)
+ chars.foreach(single.append)
+ val ranged = new ReplayListenerBus.BoundedLineBuffer(limit)
+ var start = 0
+ while (start < chars.length) {
+ val end = math.min(chars.length, start + random.nextInt(8))
+ ranged.append(chars, start, end)
+ start = end
+ }
+ withClue(s"chars=${chars.map(_.toInt).mkString(",")} limit=$limit: ") {
+ assert(ranged.result() == single.result())
+ assert(ranged.length == single.length)
+ assert((ranged.capacity == 0) == (single.capacity == 0))
+ }
+ }
+ }
+
+ test("SPARK-60158: Replay splits lines across read buffers like a reference
splitter") {
Review Comment:
Added `SPARK-60158: Replay reports physical line numbers for lines spanning
read buffers` in 714c458a1f4: a skipped 20000-char line followed by a malformed
line, with full reads and with 1-byte and up-to-100-byte reads. It checks
`first skipped line: 2` and `Malformed line #4`, and fails if the line index is
advanced per chunk instead of per line.
--
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]
