From c0d82234c773e3f0b4a22721db5391d391eac1bf Mon Sep 17 00:00:00 2001 From: eugenegujing Date: Wed, 5 Aug 2026 16:38:38 -0700 Subject: [PATCH] fix(workflow-operator): text input operator using offset with an empty limit emits no rows 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 Iterator.slice clamps to 0, so the operator silently emitted zero rows while the workflow reported success. An explicit large limit overflows the same way. This contradicts the Limit property's own description, "Leave empty to read all lines." Replace the slice with drop(offset) + take(limit), the same idiom the CSV, Arrow, and JSONL scan sources already use. Every configuration that previously worked is unchanged. Add regression tests covering offset-without-limit, offset with an Int.MaxValue limit, the offset+limit window, limit-only, offset at and past EOF, a negative offset, and SINGLE_STRING ignoring offset/limit. Closes #7346 Co-Authored-By: Claude Fable 5 --- .../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 200da75dfb9..ef8a824849a 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