This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/texera.git
The following commit(s) were added to refs/heads/main by this push:
new 91235fbdb5 fix(workflow-operator): Text Input operator using offset
with an empty limit emits no rows (#7347)
91235fbdb5 is described below
commit 91235fbdb57768c880c0e043d4cb891283e225f1
Author: Eugene Gu <[email protected]>
AuthorDate: Sun Aug 9 19:39:03 2026 -0700
fix(workflow-operator): Text Input operator using offset with an empty
limit emits no rows (#7347)
### 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`.
<img width="1349" height="839" alt="Screenshot 2026-08-05 at 2 21 56 PM"
src="https://github.com/user-attachments/assets/107a06a0-f0f3-4d4c-bf13-0357420ca6cb";
/>
**Control on the same build** — Offset = 0, Limit left empty returns all
five rows:
<img width="1344" height="841" alt="Screenshot 2026-08-05 at 2 22 05 PM"
src="https://github.com/user-attachments/assets/9acdd769-1762-44d4-8386-a6ca16a78d0d";
/>
### 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:
<img width="1131" height="760" alt="Screenshot 2026-08-05 at 5 15 55 PM"
src="https://github.com/user-attachments/assets/28e639df-0888-4793-a3a3-ffe9c37d07c8";
/>
### Was this PR authored or co-authored using generative AI tooling?
Co-authored by: Claude Code (Claude Fable 5)
Co-authored-by: Claude Fable 5 <[email protected]>
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 200da75dfb..ef8a824849 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
@@ -171,6 +171,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
