This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch
gh-readonly-queue/release/v1.2/pr-7316-2a8b858e5155c5230f4dfb866838a0f985730a8f
in repository https://gitbox.apache.org/repos/asf/texera.git
commit ea5d00fd2742c250d414ae88699f1864b506b449
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Thu Aug 13 05:39:23 2026 +0000
fix(jsonl-scan, v1.2): count the worker slice from the offset, not the file
(#7316)
### What changes were proposed in this PR?
Automated backport of #7247 to `release/v1.2`.
Source: df678aa8a9a992bfeaae6f63f4fac3237d036701 ยท [automation
run](https://github.com/apache/texera/actions/runs/30962737205)
### Any related issues, documentation, discussions?
Backport of #7247. Originally linked #7245.
### How was this PR tested?
Release-branch CI runs on this branch once the conflicts are resolved
and this PR is marked ready for review.
### Was this PR authored or co-authored using generative AI tooling?
No.
Co-authored-by: Kary Zheng <[email protected]>
Co-authored-by: Claude Opus 5 (1M context) <[email protected]>
Co-authored-by: Xuan Gu <[email protected]>
Co-authored-by: Xinyuan Lin <[email protected]>
---
.../source/scan/json/JSONLScanSourceOpExec.scala | 7 +-
.../scan/json/JSONLScanSourceOpExecSpec.scala | 123 +++++++++++++++++++++
2 files changed, 127 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
new file mode 100644
index 0000000000..a90dcd3640
--- /dev/null
+++
b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpExecSpec.scala
@@ -0,0 +1,123 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.amber.operator.source.scan.json
+
+import org.apache.texera.amber.operator.source.scan.FileDecodingMethod
+import org.apache.texera.amber.util.JSONUtils.objectMapper
+import org.scalatest.flatspec.AnyFlatSpec
+
+import java.net.URI
+import java.nio.charset.StandardCharsets
+import java.nio.file.Files
+
+class JSONLScanSourceOpExecSpec extends AnyFlatSpec {
+
+ private def writeJsonl(lines: String*): URI = {
+ val path = Files.createTempFile("jsonl-scan-", ".jsonl")
+ path.toFile.deleteOnExit()
+ Files.write(path, lines.mkString("\n").getBytes(StandardCharsets.UTF_8))
+ path.toFile.toURI
+ }
+
+ private def descString(
+ uri: URI,
+ flatten: Boolean = false,
+ limit: Option[Int] = None,
+ offset: Option[Int] = None
+ ): String = {
+ val desc = new JSONLScanSourceOpDesc
+ desc.setResolvedFileName(uri)
+ desc.fileEncoding = FileDecodingMethod.UTF_8
+ desc.flatten = flatten
+ desc.limit = limit
+ desc.offset = offset
+ objectMapper.writeValueAsString(desc)
+ }
+
+ private def drain(exec: JSONLScanSourceOpExec): List[Seq[Any]] = {
+ exec.open()
+ try exec.produceTuple().map(_.getFields.toSeq).toList
+ finally exec.close()
+ }
+
+ "JSONLScanSourceOpExec" should "read each JSON line, ordering fields by
sorted attribute name" in {
+ // keys are written name-then-id; the output must be reordered to
id-then-name
+ val exec = new JSONLScanSourceOpExec(
+ descString(writeJsonl("""{"name":"a","id":1}""",
"""{"name":"b","id":2}"""))
+ )
+ val rows = drain(exec)
+ assert(rows.size == 2)
+ assert(rows.head == Seq(1, "a"))
+ assert(rows(1) == Seq(2, "b"))
+ }
+
+ it should "partition rows across workers" in {
+ val uri = writeJsonl("""{"v":0}""", """{"v":1}""", """{"v":2}""",
"""{"v":3}""")
+ val worker0 = new JSONLScanSourceOpExec(descString(uri), idx = 0,
workerCount = 2)
+ val worker1 = new JSONLScanSourceOpExec(descString(uri), idx = 1,
workerCount = 2)
+ assert(drain(worker0).map(_.head) == Seq(0, 1))
+ assert(drain(worker1).map(_.head) == Seq(2, 3))
+ }
+
+ it should "apply the row limit" in {
+ val uri = writeJsonl("""{"v":0}""", """{"v":1}""", """{"v":2}""",
"""{"v":3}""", """{"v":4}""")
+ 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)))
+ }
+}
