This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-7247-2ae9a97363b8fb5f4c29fa8a7fa88fb7c8fbd27a in repository https://gitbox.apache.org/repos/asf/texera.git
commit df678aa8a9a992bfeaae6f63f4fac3237d036701 Author: Kary Zheng <[email protected]> AuthorDate: Tue Aug 4 17:04:44 2026 -0700 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]> --- .../source/scan/json/JSONLScanSourceOpExec.scala | 7 ++-- .../scan/json/JSONLScanSourceOpExecSpec.scala | 39 ++++++++++++++++++++++ 2 files changed, 43 insertions(+), 3 deletions(-) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpExec.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpExec.scala index 3c47796892..98a6a0bdcc 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpExec.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpExec.scala @@ -68,10 +68,11 @@ class JSONLScanSourceOpExec private[json] ( val (it1, it2) = lines.duplicate val count: Int = it1.map(_ => 1).sum - val startOffset: Int = offsetValue + count / workerCount * idx + // Bounds into `it2`, which already begins at `offsetValue`. Adding the offset + // back in would count from the start of the FILE and skip those rows twice. + val startOffset: Int = count / workerCount * idx val endOffset: Int = - offsetValue + (if (idx != workerCount - 1) count / workerCount * (idx + 1) - else count) + if (idx != workerCount - 1) count / workerCount * (idx + 1) else count rows = it2.iterator.slice(startOffset, endOffset) } diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpExecSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpExecSpec.scala index 2338e48f7c..a90dcd3640 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpExecSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpExecSpec.scala @@ -81,4 +81,43 @@ class JSONLScanSourceOpExecSpec extends AnyFlatSpec { val exec = new JSONLScanSourceOpExec(descString(uri, limit = Some(2))) assert(drain(exec).map(_.head) == Seq(0, 1)) } + + it should "start at the offset and keep every row after it" in { + val uri = writeJsonl("""{"v":0}""", """{"v":1}""", """{"v":2}""", """{"v":3}""", """{"v":4}""") + val exec = new JSONLScanSourceOpExec(descString(uri, offset = Some(2))) + assert(drain(exec).map(_.head) == Seq(2, 3, 4)) + } + + it should "apply the limit relative to the offset" in { + val uri = writeJsonl("""{"v":0}""", """{"v":1}""", """{"v":2}""", """{"v":3}""", """{"v":4}""") + // The window is shorter than the offset itself, which used to empty it out. + val exec = new JSONLScanSourceOpExec(descString(uri, limit = Some(2), offset = Some(2))) + assert(drain(exec).map(_.head) == Seq(2, 3)) + } + + it should "split the offset window across workers, losing no row to either end" in { + val uri = writeJsonl("""{"v":0}""", """{"v":1}""", """{"v":2}""", """{"v":3}""", """{"v":4}""") + val desc = descString(uri, offset = Some(1)) + val worker0 = new JSONLScanSourceOpExec(desc, idx = 0, workerCount = 2) + val worker1 = new JSONLScanSourceOpExec(desc, idx = 1, workerCount = 2) + assert(drain(worker0).map(_.head) == Seq(1, 2)) + assert(drain(worker1).map(_.head) == Seq(3, 4)) + } + + it should "give the last worker the remainder of the offset-and-limit window" in { + val uri = writeJsonl( + """{"v":0}""", + """{"v":1}""", + """{"v":2}""", + """{"v":3}""", + """{"v":4}""", + """{"v":5}""", + """{"v":6}""" + ) + // Four rows over three workers: one each, and the odd row goes to the last. + val desc = descString(uri, limit = Some(4), offset = Some(2)) + val workers = + (0 until 3).map(i => new JSONLScanSourceOpExec(desc, idx = i, workerCount = 3)) + assert(workers.map(drain(_).map(_.head)) == Seq(Seq(2), Seq(3), Seq(4, 5))) + } }
