dongjoon-hyun commented on code in PR #59074:
URL: https://github.com/apache/spark/pull/59074#discussion_r4117057071
##########
core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala:
##########
@@ -104,35 +107,54 @@ private[spark] class ReplayListenerBus(
line
}
- @tailrec private def fetchLine(): String = {
+ @tailrec private def fetchLine(): (String, Int) = {
val sb = new java.lang.StringBuilder()
var overLong = false
+ var byteLength = 0L
var c = reader.read()
if (c == -1) {
null
} else {
+ val index = lineIndex
+ lineIndex += 1
while (c != -1 && c != '\n') {
- if (sb.length() * 2 < maxLineLength) {
- sb.append(c.toChar)
- } else {
- overLong = true
+ if (!overLong) {
+ // The strict decoder guarantees valid surrogate pairs. Count
their four UTF-8
+ // bytes on the high surrogate and zero on the low surrogate.
+ byteLength += (if (c < 0x80) {
+ 1
+ } else if (c < 0x800) {
+ 2
+ } else if (Character.isHighSurrogate(c.toChar)) {
+ 4
+ } else if (Character.isLowSurrogate(c.toChar)) {
+ 0
+ } else {
+ 3
+ })
+ // Allow one extra byte until we know whether the line ends in
CRLF.
+ if (byteLength <= maxLineLength.toLong + 1) {
+ sb.append(c.toChar)
+ } else {
+ overLong = true
Review Comment:
Test coverage: because L148 now re-decides from the full `byteLength`, this
cap is no longer observable by any test. A mutant that always appends (never
sets `overLong`), or one that uses `2 * maxLineLength + 1`, produces the same
lines, indices, warnings and exceptions for every input in the suite; only the
buffered size of skipped lines changes (e.g. 8193 -> 9216 chars in "Over-long
event log lines are skipped instead of materialized"). Before this PR, removing
the cap made that test fail. A test that observes the buffer bound would guard
SPARK-59407's main guarantee, e.g. by extracting the reader into a helper and
asserting the max buffered length, or by streaming a huge line from a lazy
`InputStream` with a small limit.
##########
core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala:
##########
@@ -104,35 +107,54 @@ private[spark] class ReplayListenerBus(
line
}
- @tailrec private def fetchLine(): String = {
+ @tailrec private def fetchLine(): (String, Int) = {
val sb = new java.lang.StringBuilder()
var overLong = false
+ var byteLength = 0L
var c = reader.read()
if (c == -1) {
null
} else {
+ val index = lineIndex
+ lineIndex += 1
while (c != -1 && c != '\n') {
- if (sb.length() * 2 < maxLineLength) {
- sb.append(c.toChar)
- } else {
- overLong = true
+ if (!overLong) {
+ // The strict decoder guarantees valid surrogate pairs. Count
their four UTF-8
+ // bytes on the high surrogate and zero on the low surrogate.
+ byteLength += (if (c < 0x80) {
+ 1
+ } else if (c < 0x800) {
+ 2
+ } else if (Character.isHighSurrogate(c.toChar)) {
+ 4
+ } else if (Character.isLowSurrogate(c.toChar)) {
+ 0
+ } else {
+ 3
+ })
+ // Allow one extra byte until we know whether the line ends in
CRLF.
+ if (byteLength <= maxLineLength.toLong + 1) {
Review Comment:
This changes the per-line memory bound, not only the unit. Previously the
builder stopped growing at `maxLineLength / 2` chars; now it can hold
`maxLineLength + 1` chars, and a single char >= U+0100 inflates it to UTF-16 (2
bytes per char). With the unchanged default of `512m`:
- Default SHS heap (`-Xmx1g`), one line of U+0101 followed by 600,000,000
ASCII chars: the current code stops at capacity 301,989,886 chars and skips the
line. With this PR the builder grows to capacity 603,979,774 chars (a ~1.2 GB
array) and fails with `OutOfMemoryError: Java heap space` before the line is
rejected. `case e: Exception` does not catch it.
- For configured limits in [576m, 1152m), a Latin-1 builder that has grown
past 603,979,774 chars has capacity 1,207,959,550.
`AbstractStringBuilder.inflate()` sizes the UTF-16 array from the capacity, so
the first char >= U+0100 fails with `UTF16 String size is 1207959550, should be
less than 1073741823` regardless of the heap size.
- On JDK 8 (branch-3.5) there are no compact strings, so even pure-ASCII
lines need about twice the heap of the current code.
The docs still say the setting bounds "the memory replay can use". Could we
default to `256m` (keeps the current bound for ASCII lines), keep a separate
heap-oriented cap for the buffer, or document that heap usage can reach about
2-4x the limit?
##########
core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala:
##########
@@ -104,35 +107,54 @@ private[spark] class ReplayListenerBus(
line
}
- @tailrec private def fetchLine(): String = {
+ @tailrec private def fetchLine(): (String, Int) = {
val sb = new java.lang.StringBuilder()
var overLong = false
+ var byteLength = 0L
var c = reader.read()
if (c == -1) {
null
} else {
+ val index = lineIndex
+ lineIndex += 1
while (c != -1 && c != '\n') {
- if (sb.length() * 2 < maxLineLength) {
- sb.append(c.toChar)
- } else {
- overLong = true
+ if (!overLong) {
+ // The strict decoder guarantees valid surrogate pairs. Count
their four UTF-8
+ // bytes on the high surrogate and zero on the low surrogate.
+ byteLength += (if (c < 0x80) {
+ 1
+ } else if (c < 0x800) {
+ 2
+ } else if (Character.isHighSurrogate(c.toChar)) {
+ 4
+ } else if (Character.isLowSurrogate(c.toChar)) {
+ 0
+ } else {
+ 3
+ })
+ // Allow one extra byte until we know whether the line ends in
CRLF.
+ if (byteLength <= maxLineLength.toLong + 1) {
+ sb.append(c.toChar)
+ } else {
+ overLong = true
+ }
}
c = reader.read()
Review Comment:
Performance (pre-existing since SPARK-59407, kept by this rewrite):
`BufferedReader.read()` takes a lock on every call (on JDK 21 an
`InternalLock`/`ReentrantLock`, whose CAS survives C2 inlining). On ~100 MB of
TaskEnd-like ASCII lines this loop splits at ~73 MB/s (13.6 s/GB) vs ~1,039
MB/s (0.96 s/GB) for the previous `Source.getLines()`; together with Jackson
`readTree` it is 60.9 MB/s vs 293 MB/s. That is roughly +12.6 s of CPU per GB
for UI rebuilds and compaction, and ~8 s to drain a 600 MiB line. The UTF-8
width branching added here is not measurable. Keeping the same per-char logic
but reading into a local `char[]` via `InputStreamReader.read(char[], ...)` is
~5x faster, and bulk-scanning the buffer for `\n` with slice appends is ~15x
faster. Maybe a follow-up?
##########
core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala:
##########
@@ -104,35 +107,54 @@ private[spark] class ReplayListenerBus(
line
}
- @tailrec private def fetchLine(): String = {
+ @tailrec private def fetchLine(): (String, Int) = {
val sb = new java.lang.StringBuilder()
var overLong = false
+ var byteLength = 0L
var c = reader.read()
if (c == -1) {
null
} else {
+ val index = lineIndex
+ lineIndex += 1
while (c != -1 && c != '\n') {
- if (sb.length() * 2 < maxLineLength) {
- sb.append(c.toChar)
- } else {
- overLong = true
+ if (!overLong) {
+ // The strict decoder guarantees valid surrogate pairs. Count
their four UTF-8
+ // bytes on the high surrogate and zero on the low surrogate.
+ byteLength += (if (c < 0x80) {
+ 1
+ } else if (c < 0x800) {
+ 2
+ } else if (Character.isHighSurrogate(c.toChar)) {
+ 4
+ } else if (Character.isLowSurrogate(c.toChar)) {
+ 0
+ } else {
+ 3
+ })
+ // Allow one extra byte until we know whether the line ends in
CRLF.
+ if (byteLength <= maxLineLength.toLong + 1) {
+ sb.append(c.toChar)
+ } else {
+ overLong = true
+ }
}
c = reader.read()
}
- if (overLong) {
+ if (sb.length() > 0 && sb.charAt(sb.length() - 1) == '\r') {
+ sb.setLength(sb.length() - 1)
+ byteLength -= 1
+ }
+ if (overLong || byteLength > maxLineLength) {
Review Comment:
nit: `overLong ||` is redundant. `overLong` is only set when `byteLength >=
maxLineLength + 2`, `byteLength` no longer changes afterwards, and the CR strip
subtracts at most 1, so `byteLength > maxLineLength` already holds (also for
`Int.MaxValue`, 0 and negative limits; checked exhaustively for small limits
plus random inputs). Either use `if (byteLength > maxLineLength)`, or drop the
flag and keep only `if (byteLength <= maxLineLength + 1L) sb.append(c.toChar)`
in the loop, which removes one nesting level.
##########
core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala:
##########
@@ -148,15 +170,21 @@ private[spark] class ReplayListenerBus(
sourceName: String,
maybeTruncated: Boolean,
eventsFilter: ReplayEventsFilter): Boolean = {
+ replayEntries(lines.zipWithIndex, sourceName, maybeTruncated, eventsFilter)
Review Comment:
This overload is also what the end-event reparse in `FsHistoryProvider` uses
(FsHistoryProvider.scala L1109-L1127), fed by an unbounded
`Source.fromInputStream(in)(Codec.UTF8).getLines()`. So on that path there is
no `maxLineLength` and line numbers are relative to the skip point. With the
defaults (compressed event logs, `endEventReparseChunkSize=1m`), completed apps
always take this path, and because `target = lastFile.getLen -
reparseChunkSize` is based on the compressed length while `skip` runs on the
decompressed stream, most of the log is read there. In a local repro (156.6 MB
decompressed / 16.4 MB compressed), 141 MB went through `getLines()`, and a
12.6 MB line over a 4 MiB limit was fully materialized. This predates the PR,
but since the PR defines the limit and line-number semantics, it may be worth
reusing the bounded reader there (skip, drop the partial first line, then
replay through `boundedLines`) or filing a follow-up.
##########
core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala:
##########
@@ -104,35 +107,54 @@ private[spark] class ReplayListenerBus(
line
}
- @tailrec private def fetchLine(): String = {
+ @tailrec private def fetchLine(): (String, Int) = {
val sb = new java.lang.StringBuilder()
var overLong = false
+ var byteLength = 0L
var c = reader.read()
if (c == -1) {
null
} else {
+ val index = lineIndex
+ lineIndex += 1
while (c != -1 && c != '\n') {
- if (sb.length() * 2 < maxLineLength) {
- sb.append(c.toChar)
- } else {
- overLong = true
+ if (!overLong) {
+ // The strict decoder guarantees valid surrogate pairs. Count
their four UTF-8
+ // bytes on the high surrogate and zero on the low surrogate.
+ byteLength += (if (c < 0x80) {
Review Comment:
nit: since the strict decoder only emits complete surrogate pairs, counting
2 bytes per surrogate half gives identical results (the running total matches
at every code point boundary), so this could be `if (c < 0x80) 1 else if (c <
0x800 || Character.isSurrogate(c.toChar)) 2 else 3`, and the comment above
becomes simpler.
##########
core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala:
##########
@@ -67,22 +66,26 @@ private[spark] class ReplayListenerBus(
maybeTruncated: Boolean = false,
eventsFilter: ReplayEventsFilter = SELECT_ALL_FILTER): Boolean = {
val lines = boundedLines(logData, sourceName)
- replay(lines, sourceName, maybeTruncated, eventsFilter)
+ replayEntries(lines, sourceName, maybeTruncated, eventsFilter)
}
/**
- * Reads '\n'-terminated lines like Source.getLines(), but never
materializes more than
- * `maxLineLength` bytes of a single line. An over-long line is drained and
skipped
- * with a warning instead of being turned into a String.
+ * Reads '\n'-terminated lines and retains their original zero-based
indices. The limit
+ * measures UTF-8 content bytes, excluding the line ending, rather than JVM
heap usage.
+ * The character buffer is bounded by the limit plus one possible trailing
CR. An over-long
+ * line is drained and skipped with a warning instead of being turned into a
String.
*/
- private def boundedLines(logData: InputStream, sourceName: String):
Iterator[String] = {
+ private def boundedLines(
Review Comment:
Since `boundedLines` is private, compaction reads the same files with two
different readers: `EventLogFileCompactor.initializeBuilders` replays through
this bounded reader, while `EventFilter.applyFilterToFile` (EventFilter.scala
L79) rewrites them with an unbounded `Source.getLines()`. If a skipped
over-limit line is the start event of a still-live entity (e.g.
`StageSubmitted` of a running stage, `JobStart` of a running job,
`SQLExecutionStart` of a running execution without jobs), the builders never
mark it live, the rewrite materializes the line and rejects it together with
its dependent task events, and `cleanupCompactedFiles` deletes the originals,
so raising the limit later cannot recover them. A port of the builder/filter
logic lost 23 events with a 1 MiB limit and 2 MiB lines. The inconsistency came
with SPARK-59407 and this PR only moves the threshold (e.g. CJK-heavy lines
between limit/3 and limit/2 chars are newly skipped), but sharing one bounded
reader for both passe
s, and passing over-long lines through unchanged in the rewrite, would make
them consistent.
##########
core/src/main/scala/org/apache/spark/internal/config/History.scala:
##########
@@ -167,8 +167,8 @@ private[spark] object History {
val EVENT_LOG_MAX_LINE_LENGTH =
ConfigBuilder("spark.history.fs.eventLog.maxLineLength")
- .doc("Maximum length of a single event log line during replay. Lines
longer than " +
- "this are skipped with a warning instead of being read into memory,
bounding the " +
+ .doc("Maximum UTF-8 byte length of a single event log line during
replay, excluding " +
+ "the line ending. Longer lines are skipped with a warning, bounding
the " +
"memory replay can use when an event log is corrupt or unexpectedly
large. Setting " +
"this to 0 or a negative value disables the limit. " +
Review Comment:
With the new `Long` arithmetic, "disables the limit" is no longer strictly
true. `ReplayListenerBus.maxLineLength` maps `<= 0` (and values above
`Int.MaxValue`) to `Int.MaxValue`, and lines with more than 2,147,483,647 UTF-8
bytes are now skipped with a warning (e.g. 2^30 x U+00E9 is 2,147,483,648 bytes
but only a 1 GiB Latin-1 builder), whereas the previous `sb.length() * 2 <
Int.MaxValue` overflowed and never skipped. Spark 3.4+ writers cannot produce
such a line (one `ByteArrayOutputStream` per event), so this is mostly
theoretical, but skipping the check when the limit is disabled, or documenting
the 2^31-1 byte cap, would keep the doc accurate.
##########
core/src/test/scala/org/apache/spark/scheduler/ReplayListenerSuite.scala:
##########
@@ -109,6 +109,85 @@ class ReplayListenerSuite extends SparkFunSuite with
BeforeAndAfter with LocalSp
assert(eventMonster.loggedEvents(1) ===
JsonProtocol.sparkEventToJsonString(applicationEnd))
}
+ test("Replay line limit uses UTF-8 bytes") {
Review Comment:
minor, about the PR description: it says "Three new regression tests fail
against the original implementation", but four of the five new tests fail on
the current code (this one, the boundary grid, the physical-line-number test
and the truncated/filtered test); only "Replay still rejects malformed UTF-8 in
skipped lines" passes there. Since the description becomes the commit message,
it may be worth correcting.
##########
core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala:
##########
@@ -148,15 +170,21 @@ private[spark] class ReplayListenerBus(
sourceName: String,
maybeTruncated: Boolean,
eventsFilter: ReplayEventsFilter): Boolean = {
+ replayEntries(lines.zipWithIndex, sourceName, maybeTruncated, eventsFilter)
+ }
+
+ private def replayEntries(
Review Comment:
The physical line number only surfaces for non-IOException failures.
`JsonParseException` (and `JsonMappingException`s such as
`MismatchedInputException`) extend `IOException`, so for a malformed JSON line
in the middle of a file, `throw jpe` (L219) is caught by `case ioe: IOException
=> throw ioe` (L233) before the `Malformed line #N` log (L237). For example,
`[end, 2048 x's, 2048 x's, {bad, end]` with a 1024 limit throws without any
line number, and `FsHistoryProvider` only logs Jackson's `line: 1, column: N`,
which is relative to the single-line string. `{}` in the new test goes through
the NPE path. The case ordering is old (SPARK-2261), but logging
`sourceName`/`lineNumber` before rethrowing would make this fix effective for
the most common corruption.
##########
core/src/test/scala/org/apache/spark/scheduler/ReplayListenerSuite.scala:
##########
@@ -109,6 +109,85 @@ class ReplayListenerSuite extends SparkFunSuite with
BeforeAndAfter with LocalSp
assert(eventMonster.loggedEvents(1) ===
JsonProtocol.sparkEventToJsonString(applicationEnd))
}
+ test("Replay line limit uses UTF-8 bytes") {
+ val start = SparkListenerApplicationStart("x" * 6000, None, 125L, "user",
None)
+ val json = JsonProtocol.sparkEventToJsonString(start)
+ assert(json.getBytes(StandardCharsets.UTF_8).length < 8 * 1024)
+ val listener = new EventBufferingListener
+ val bus = new ReplayListenerBus(8 * 1024)
+ bus.addListener(listener)
+ val input = new ByteArrayInputStream(json.getBytes(StandardCharsets.UTF_8))
+ assert(bus.replay(input, "ascii"))
+ assert(listener.loggedEvents.toSeq == Seq(json))
+ }
+
+ test("Replay line limit handles UTF-8 boundaries and line endings") {
+ val end =
JsonProtocol.sparkEventToJsonString(SparkListenerApplicationEnd(1000L))
+ // scalastyle:off nonascii
+ val names = Seq("x", "\u00e9", "\u4e2d", "\ud83d\ude00")
Review Comment:
The U+1F600 case does not exercise the 4-byte path. `JsonProtocol`
serializes through Jackson's `UTF8JsonGenerator`, which writes supplementary
characters as the ASCII escape `\ud83d\ude00`
(`COMBINE_UNICODE_SURROGATES_IN_UTF8` is off by default in 2.22.2). So that
JSON line is pure ASCII, the surrogate branches (ReplayListenerBus.scala
L128-L131) are never taken, and mutants that count 3+3, 3+0 or 0+0 bytes for a
pair all pass. Also, the boundary always falls on the trailing ASCII
`","Timestamp":125,"User":"user"}`, so no multi-byte character straddles the
limit, and since the name starts at byte offset 110, the 8192-byte decoder
boundary never splits a U+00E9/U+4E2D sequence (8192 - 110 = 8082 is divisible
by 2 and 3). Could we build raw UTF-8 input directly (e.g. lines whose last
code point is 2, 3 or 4 bytes) and put the limit at -1/0/+1/+2 bytes around it?
##########
core/src/test/scala/org/apache/spark/scheduler/ReplayListenerSuite.scala:
##########
@@ -109,6 +109,85 @@ class ReplayListenerSuite extends SparkFunSuite with
BeforeAndAfter with LocalSp
assert(eventMonster.loggedEvents(1) ===
JsonProtocol.sparkEventToJsonString(applicationEnd))
}
+ test("Replay line limit uses UTF-8 bytes") {
+ val start = SparkListenerApplicationStart("x" * 6000, None, 125L, "user",
None)
+ val json = JsonProtocol.sparkEventToJsonString(start)
+ assert(json.getBytes(StandardCharsets.UTF_8).length < 8 * 1024)
+ val listener = new EventBufferingListener
+ val bus = new ReplayListenerBus(8 * 1024)
+ bus.addListener(listener)
+ val input = new ByteArrayInputStream(json.getBytes(StandardCharsets.UTF_8))
+ assert(bus.replay(input, "ascii"))
+ assert(listener.loggedEvents.toSeq == Seq(json))
+ }
+
+ test("Replay line limit handles UTF-8 boundaries and line endings") {
+ val end =
JsonProtocol.sparkEventToJsonString(SparkListenerApplicationEnd(1000L))
+ // scalastyle:off nonascii
+ val names = Seq("x", "\u00e9", "\u4e2d", "\ud83d\ude00")
+ // scalastyle:on nonascii
+ for {
+ name <- names
+ repetitions <- Seq(100, 4097)
+ ending <- Seq("\n", "\r\n", "")
+ excess <- Seq(0, 1)
+ } {
+ val start = SparkListenerApplicationStart(name * repetitions, None,
125L, "user", None)
+ val json = JsonProtocol.sparkEventToJsonString(start)
+ val limit = json.getBytes(StandardCharsets.UTF_8).length - excess
+ val listener = new EventBufferingListener
+ val bus = new ReplayListenerBus(limit)
+ bus.addListener(listener)
+ // Put a short event first, including when the final line has no
terminator.
+ val input = new ByteArrayInputStream((end + "\n" + json + ending)
+ .getBytes(StandardCharsets.UTF_8))
+ assert(bus.replay(input, "utf8"))
+ val expected = if (excess == 0) Seq(end, json) else Seq(end)
+ assert(listener.loggedEvents.toSeq == expected, s"name=$name
ending=$ending excess=$excess")
Review Comment:
nit: (1) The clue omits `repetitions` and prints `ending` raw, so `\n` and
`\r\n` failures look identical, e.g. `s"name=$name repetitions=$repetitions
ending=${ending.map(_.toInt).mkString(",")} excess=$excess"`; the
`assert(bus.replay(input, "utf8"))` above has no clue at all. (2) `repetitions`
100 vs 4097 gives the same outcome for every (name, ending, excess)
combination, so it only doubles the grid (see the decoder-boundary note on
L127). (3) "Replay line limit uses UTF-8 bytes" is subsumed by the `name =
"x"`, `excess = 0` iterations here.
--
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]
