From 66efbf622e95c757e4b312ea9d05717ea04c027d Mon Sep 17 00:00:00 2001 From: Eugene Gu Date: Mon, 10 Aug 2026 02:51:33 +0000 Subject: [PATCH] fix(workflow-operator): Text Input operator using offset with an empty limit emits no rows (#7347) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ### What changes were proposed in this PR? `TextInputSourceOpExec` computed its line window as `slice(offset, offset + limit.getOrElse(Int.MaxValue))`. With an Offset set and the Limit left empty, the addition overflows `Int` to a negative bound, which Scala 2.13's `Iterator.slice` clamps to 0 and then returns an empty iterator — so the operator silently emitted **zero rows** while the workflow reported success. Any Offset ≥ 1 with an empty Limit is affected, and an explicit large Limit (e.g. `Int.MaxValue`) overflows the same way. This contradicts the Limit property's own description, "Leave empty to read all lines." This PR replaces the slice with `drop(offset)` + `take(limit)`, the same idiom the CSV, Arrow, and JSONL scan sources already use. There is no addition, so nothing can overflow; every configuration that previously worked is unchanged (verified case-by-case, including negative offsets and `isSingle` attribute types, which keep ignoring offset/limit as documented). **Before the fix (current `main`)** — Offset = 1, Limit left empty, five-line input `a b c d e`: the result is an empty set even though the workflow completes successfully. Expected: the four rows `b, c, d, e`. Screenshot 2026-08-05 at 2 21 56 PM **Control on the same build** — Offset = 0, Limit left empty returns all five rows: Screenshot 2026-08-05 at 2 22 05 PM ### Any related issues, documentation, discussions? Closes #7346. Same class of defect as #7245 (JSONL File Scan dropping rows when Offset is set), which was fixed by #7247. ### How was this PR tested? TDD: the regression tests were written first and confirmed to fail on the unfixed code — the offset-without-limit case and the offset-with-`Int.MaxValue`-limit case both produced empty output — then the fix was applied and all tests pass. Seven new cases were added to `TextInputSourceOpDescSpec` (the spec that already exercises `produceTuple()`): offset without limit, offset with an `Int.MaxValue` limit, offset+limit window, limit only, offset at/past the end of the input, negative offset treated as zero, and `SINGLE_STRING` ignoring offset/limit (documented behavior, pinned). ```bash sbt "WorkflowOperator/testOnly org.apache.texera.amber.operator.source.scan.text.TextInputSourceOpDescSpec" # 17 tests, all passed (10 pre-existing + 7 new) sbt "WorkflowOperator/testOnly org.apache.texera.amber.operator.source.scan.*" # 17 suites, 123 tests, all passed sbt "WorkflowOperator/scalafixAll --check" # passed, no lint issues sbt scalafmtCheckAll # passed, no mis-formatted files ``` Also verified manually in the UI with the same two-operator workflow shown in the screenshots above: Screenshot 2026-08-05 at 5 15 55 PM ### Was this PR authored or co-authored using generative AI tooling? Co-authored by: Claude Code (Claude Fable 5) (backported from commit 91235fbdb57768c880c0e043d4cb891283e225f1) Co-authored-by: Claude Fable 5 Co-authored-by: Xinyuan Lin --- .../scan/text/TextInputSourceOpExec.scala | 10 +- .../scan/text/TextInputSourceOpDescSpec.scala | 93 +++++++++++++++++++ 2 files changed, 99 insertions(+), 4 deletions(-) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/text/TextInputSourceOpExec.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/text/TextInputSourceOpExec.scala index 8ade443ef9b..ed314ce0fc6 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/text/TextInputSourceOpExec.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/text/TextInputSourceOpExec.scala @@ -34,10 +34,12 @@ class TextInputSourceOpExec private[text] ( (if (desc.attributeType.isSingle) { Iterator(desc.textInput) } else { - desc.textInput.linesIterator.slice( - desc.fileScanOffset.getOrElse(0), - desc.fileScanOffset.getOrElse(0) + desc.fileScanLimit.getOrElse(Int.MaxValue) - ) + // `slice(offset, offset + limit)` overflows Int when the limit is absent + // (it defaults to Int.MaxValue) or large, making `until <= from` and + // silently yielding no rows. + desc.textInput.linesIterator + .drop(desc.fileScanOffset.getOrElse(0)) + .take(desc.fileScanLimit.getOrElse(Int.MaxValue)) }).map(line => TupleLike(desc.attributeType match { case FileAttributeType.SINGLE_STRING => line diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/text/TextInputSourceOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/text/TextInputSourceOpDescSpec.scala index d1b5a5f94ad..e9dcf5b7610 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/text/TextInputSourceOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/text/TextInputSourceOpDescSpec.scala @@ -169,6 +169,99 @@ class TextInputSourceOpDescSpec extends AnyFlatSpec with BeforeAndAfter { textScanSourceOpExec.close() } + it should "read all lines after the offset when no limit is specified" in { + assert( + linesFrom(offset = Some(5)) == Seq("line6", "line7", "line8", "line9", "line10") + ) + } + + it should "read all lines after the offset when the limit is Int.MaxValue" in { + assert( + linesFrom(offset = Some(1), limit = Some(Int.MaxValue)) == + Seq("line2", "line3", "line4", "line5", "line6", "line7", "line8", "line9", "line10") + ) + } + + it should "read a window of lines when both offset and limit are specified" in { + assert(linesFrom(offset = Some(5), limit = Some(2)) == Seq("line6", "line7")) + } + + it should "read the first lines when only a limit is specified" in { + assert(linesFrom(limit = Some(3)) == Seq("line1", "line2", "line3")) + } + + it should "produce no tuples when the offset is at or past the end of the input" in { + assert(linesFrom(offset = Some(10)).isEmpty) + assert(linesFrom(offset = Some(99)).isEmpty) + } + + it should "treat a negative offset as zero" in { + assert( + linesFrom(offset = Some(-1)) == + Seq( + "line1", + "line2", + "line3", + "line4", + "line5", + "line6", + "line7", + "line8", + "line9", + "line10" + ) + ) + } + + it should "ignore the offset when reading the input text into a single output tuple" in { + val inputString: String = readFileIntoString(TestOperators.TestTextFilePath) + textInputSourceOpDesc.attributeType = FileAttributeType.SINGLE_STRING + textInputSourceOpDesc.textInput = inputString + textInputSourceOpDesc.fileScanOffset = Option(5) + val textScanSourceOpExec = + new TextInputSourceOpExec(objectMapper.writeValueAsString(textInputSourceOpDesc)) + textScanSourceOpExec.open() + val processedTuple: Iterator[Tuple] = textScanSourceOpExec + .produceTuple() + .map(tupleLike => + tupleLike + .asInstanceOf[SchemaEnforceable] + .enforceSchema(textInputSourceOpDesc.sourceSchema()) + ) + + assert( + processedTuple + .next() + .getField[String]("line") + .equals("line1\nline2\nline3\nline4\nline5\nline6\nline7\nline8\nline9\nline10") + ) + assertThrows[java.util.NoSuchElementException](processedTuple.next().getField("line")) + textScanSourceOpExec.close() + } + + /** + * Helper function collecting the "line" field of every tuple produced for + * the STRING attribute type with the given offset and limit. + */ + private def linesFrom(offset: Option[Int] = None, limit: Option[Int] = None): Seq[String] = { + textInputSourceOpDesc.attributeType = FileAttributeType.STRING + textInputSourceOpDesc.textInput = readFileIntoString(TestOperators.TestTextFilePath) + textInputSourceOpDesc.fileScanOffset = offset + textInputSourceOpDesc.fileScanLimit = limit + val exec = new TextInputSourceOpExec(objectMapper.writeValueAsString(textInputSourceOpDesc)) + exec.open() + try { + exec + .produceTuple() + .map( + _.asInstanceOf[SchemaEnforceable] + .enforceSchema(textInputSourceOpDesc.sourceSchema()) + .getField[String]("line") + ) + .toSeq + } finally exec.close() + } + /** * Helper function using UTF-8 encoding to read text file * into String