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-7529-e6bf52e877a19d2aef151c017c4de71953246523 in repository https://gitbox.apache.org/repos/asf/texera.git
commit 2a8b858e5155c5230f4dfb866838a0f985730a8f Author: Eugene Gu <[email protected]> AuthorDate: Thu Aug 13 05:38:02 2026 +0000 fix(workflow-operator, v1.2): Text Input operator using offset with an empty limit emits no rows (#7529) ### What changes were proposed in this PR? Backport of #7347 to `release/v1.2`, cherry-picked from main commit 91235fbdb (clean, no conflicts). `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." The fix 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. ### Any related issues, documentation, discussions? Backport of #7347 (originally closed #7346). ### How was this PR tested? The 7 regression tests from #7347 come along with the cherry-pick. On this branch: ```bash sbt "WorkflowOperator/testOnly org.apache.texera.amber.operator.source.scan.text.TextInputSourceOpDescSpec" # 15 tests, all passed (8 pre-existing on release/v1.2 + 7 new) # (main has 17: two getPhysicalOp/propagateSchema coverage tests were added # to this spec after v1.2 branched and are unrelated to this fix) sbt "WorkflowOperator/scalafmtCheck" "WorkflowOperator/Test/scalafmtCheck" # passed sbt "WorkflowOperator/scalafixAll --check" # passed ``` ### Was this PR authored or co-authored using generative AI tooling? Co-authored by: Claude Code (Claude Fable 5) Co-authored-by: Xinyuan Lin <[email protected]> --- .../source/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 8ade443ef9..ed314ce0fc 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 d1b5a5f94a..e9dcf5b761 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
