Skip to content

Commit df678aa

Browse files
kz930claudexuang7
authored
fix(jsonl-scan): count the worker slice from the offset, not the file (#7247)
### What changes were proposed in this PR? `JSONLScanSourceOpExec.open()` applies the requested offset twice. It drops `offset` lines from the reader, so the iterator it keeps already begins at the requested row; it then computes the worker's slice bounds as `offsetValue + …`, which are positions in the original file, and applies those bounds to that already-offset iterator. The rows in between are skipped, and when what remains is shorter than the offset the operator emits nothing at all. Neither case raises an error or writes a log line, so the operator reports success with rows missing. This PR resolves the bounds against the iterator they are actually applied to, by dropping the `offsetValue +` from both expressions. That is the first of the two options the issue describes; it keeps the reader-side `drop` doing the offset and leaves the slice to do only what it exists for, which is dividing the window across workers. ### Any related issues, documentation, discussions? Closes #7245 ### How was this PR tested? Two cases were added to `JSONLScanSourceOpExecSpec`, which previously covered the row limit but never set an offset: an offset on its own, and an offset together with a limit — the combination that used to yield no rows at all. Both fail against the current code and pass with the change, and the three existing cases are unaffected. Beyond those, the operator was driven over a five-line file across every combination of offset (unset, 0, 1, 2, 4, 5, 7), limit (unset, 0, 2, 3, 9) and worker count (1, 2, 3), with each result compared against `drop(offset).take(limit)` — the reading that `ArrowSourceOpExec` and this operator's own `sourceSchema()` already follow. 36 of the 105 combinations were wrong before the change and none after. Every one of the 36 had an offset of 1 or more, which is why an unset offset was unaffected and the gap went unnoticed. That sweep was a scratch harness for checking the fix and is not part of the diff; the two added cases are the representatives of it that are worth keeping. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 5) --------- Co-authored-by: Claude Opus 5 (1M context) <[email protected]> Co-authored-by: Xuan Gu <[email protected]>
1 parent 2ae9a97 commit df678aa

2 files changed

Lines changed: 43 additions & 3 deletions

File tree

‎common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpExec.scala‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -68,10 +68,11 @@ class JSONLScanSourceOpExec private[json] (
6868
val (it1, it2) = lines.duplicate
6969
val count: Int = it1.map(_ => 1).sum
7070

71-
val startOffset: Int = offsetValue + count / workerCount * idx
71+
// Bounds into `it2`, which already begins at `offsetValue`. Adding the offset
72+
// back in would count from the start of the FILE and skip those rows twice.
73+
val startOffset: Int = count / workerCount * idx
7274
val endOffset: Int =
73-
offsetValue + (if (idx != workerCount - 1) count / workerCount * (idx + 1)
74-
else count)
75+
if (idx != workerCount - 1) count / workerCount * (idx + 1) else count
7576

7677
rows = it2.iterator.slice(startOffset, endOffset)
7778
}

‎common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpExecSpec.scala‎

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -81,4 +81,43 @@ class JSONLScanSourceOpExecSpec extends AnyFlatSpec {
8181
val exec = new JSONLScanSourceOpExec(descString(uri, limit = Some(2)))
8282
assert(drain(exec).map(_.head) == Seq(0, 1))
8383
}
84+
85+
it should "start at the offset and keep every row after it" in {
86+
val uri = writeJsonl("""{"v":0}""", """{"v":1}""", """{"v":2}""", """{"v":3}""", """{"v":4}""")
87+
val exec = new JSONLScanSourceOpExec(descString(uri, offset = Some(2)))
88+
assert(drain(exec).map(_.head) == Seq(2, 3, 4))
89+
}
90+
91+
it should "apply the limit relative to the offset" in {
92+
val uri = writeJsonl("""{"v":0}""", """{"v":1}""", """{"v":2}""", """{"v":3}""", """{"v":4}""")
93+
// The window is shorter than the offset itself, which used to empty it out.
94+
val exec = new JSONLScanSourceOpExec(descString(uri, limit = Some(2), offset = Some(2)))
95+
assert(drain(exec).map(_.head) == Seq(2, 3))
96+
}
97+
98+
it should "split the offset window across workers, losing no row to either end" in {
99+
val uri = writeJsonl("""{"v":0}""", """{"v":1}""", """{"v":2}""", """{"v":3}""", """{"v":4}""")
100+
val desc = descString(uri, offset = Some(1))
101+
val worker0 = new JSONLScanSourceOpExec(desc, idx = 0, workerCount = 2)
102+
val worker1 = new JSONLScanSourceOpExec(desc, idx = 1, workerCount = 2)
103+
assert(drain(worker0).map(_.head) == Seq(1, 2))
104+
assert(drain(worker1).map(_.head) == Seq(3, 4))
105+
}
106+
107+
it should "give the last worker the remainder of the offset-and-limit window" in {
108+
val uri = writeJsonl(
109+
"""{"v":0}""",
110+
"""{"v":1}""",
111+
"""{"v":2}""",
112+
"""{"v":3}""",
113+
"""{"v":4}""",
114+
"""{"v":5}""",
115+
"""{"v":6}"""
116+
)
117+
// Four rows over three workers: one each, and the odd row goes to the last.
118+
val desc = descString(uri, limit = Some(4), offset = Some(2))
119+
val workers =
120+
(0 until 3).map(i => new JSONLScanSourceOpExec(desc, idx = i, workerCount = 3))
121+
assert(workers.map(drain(_).map(_.head)) == Seq(Seq(2), Seq(3), Seq(4, 5)))
122+
}
84123
}

0 commit comments

Comments
 (0)