dongjoon-hyun commented on code in PR #59360:
URL: https://github.com/apache/spark/pull/59360#discussion_r4247340459
##########
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:
**[2] Minor: `append(c: Char)` is now used only by tests.**
`fetchLine` only calls the range overload. That leaves two copies of the
overflow and release logic, and only one of them runs in production. It also
means the SPARK-59804 test `Replay line buffer stops retaining oversized
content` now checks the path that replay no longer uses. Removing `append(c)`
and moving that test to the range overload, or making `append(c)` delegate to
`append(Array(c), 0, 1)`, would keep a single implementation. The equivalence
test would then only be needed if both methods stay.
##########
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:
**[3] Minor: each chunk is scanned twice.**
This loop looks for `\n`, and then `BoundedLineBuffer.append(chars, start,
end)` walks the same range again to add up `utf8Width`. Counting the width in
this loop (or passing it to the buffer) would make it a single pass. After the
line goes over the limit, only the `\n` search is needed.
##########
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:
**[1] Minor: the read pattern of codec streams is not exercised.**
`ByteArrayInputStream.available()` is always positive, so `StreamDecoder`
keeps decoding until the 8192-char array is full, and every
`reader.read(chars)` here returns a full chunk. With LZ4/ZStd streams,
`available()` is often 0, so reads come back short and uneven. A chunk can then
end right after a `\r`, right before a `\n`, or be a single `\n`. Wrapping the
input in a stream that returns a random 1..N bytes per `read` and reports
`available() == 0` would cover the `fill()`/`pos`/`len` handling under those
reads.
##########
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:
**[4] Minor: optional fast path for lines that fit in one chunk.**
Each line still gets a new `StringBuilder` (initial capacity 16) that grows
by doubling, about 10 copies for a 10 KB line, and is copied once more by
`toString`/`substring`. Typical event lines fit in one 8192-char chunk, so when
the `\n` is found in the same chunk and the line is within the limit, `new
String(chars, pos, i - pos)` would skip the builder. This is optional, since
the PR's numbers already match `Source.getLines()`.
##########
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:
**[5] Design question (non-blocking): splitting on bytes.**
In UTF-8 the byte `0x0A` never appears inside a multi-byte sequence, so
lines could be split and limited on raw bytes, and only the accepted lines
decoded strictly. That gives the exact byte count without the per-char
`utf8Width` rule, and an over-long line is skipped without being decoded. The
trade-off is that malformed UTF-8 inside a skipped line would no longer fail
the replay. I'm fine with the current approach. I'm only mentioning it in case
the remaining overhead matters later.
##########
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:
**[6] Minor: line indices across read buffers are not covered.**
This test sees only line contents through `eventsFilter`, not the physical
index that `boundedLines` returns. The existing `Replay reports the physical
location of skipped lines` input (about 4 KB) fits in one read buffer. So no
test checks that `lineIndex` stays right when a skipped line spans several
`fill()` calls. An example is a 20000-char skipped line followed by a malformed
line, where the `Malformed line #N` and `first skipped line` numbers should
still be the physical ones.
--
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]
