You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@texera.apache.org by gi...@apache.org on 2026/08/13 05:45:16 UTC
(texera) branch gh-readonly-queue/release/v1.2/pr-7529-e6bf52e877a19d2aef151c017c4de71953246523 created (now 2a8b858e51)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/release/v1.2/pr-7529-e6bf52e877a19d2aef151c017c4de71953246523
in repository https://gitbox.apache.org/repos/asf/texera.git
at 2a8b858e51 fix(workflow-operator, v1.2): Text Input operator using offset with an empty limit emits no rows (#7529)
This branch includes the following new commits:
new 2a8b858e51 fix(workflow-operator, v1.2): Text Input operator using offset with an empty limit emits no rows (#7529)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(texera) 01/01: fix(workflow-operator, v1.2): Text Input operator using offset with an empty limit emits no rows (#7529)
Posted by gi...@apache.org.
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 <eu...@outlook.com>
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 <xi...@uci.edu>
---
.../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