This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-7348-88ca47f2f4d4ae26535012951b9c64cb9af3c91c in repository https://gitbox.apache.org/repos/asf/texera.git
commit 8982803165bbf2ace67ec9635b258a16649e17ce Author: Eugene Gu <[email protected]> AuthorDate: Sun Aug 9 19:26:36 2026 -0700 fix(workflow-operator): File Scan operator using offset with an empty limit emits no rows (#7348) ### What changes were proposed in this PR? `FileScanUtils.createTuplesFromFile` computed the end of its line slice as `offset + limit.getOrElse(Int.MaxValue)`. With Offset >= 1 and Limit left empty, the addition overflows `Int` to a negative bound, and `Iterator.slice` clamps a negative bound to 0 and returns an empty iterator. The File Scan operator therefore emitted **zero rows, silently, with the workflow reporting success**. Both `FileScan` and `FileScanOp` delegate to this helper, so both were affected. **Before the fix (current `main`)**——**Offset = 1, Limit left empty — zero rows ("Empty result set") while the run reports success:** <img width="1344" height="869" alt="Screenshot 2026-08-05 at 2 19 40 PM" src="https://github.com/user-attachments/assets/28c628b5-2a71-47a4-bf78-f790c9b113c0"; /> **Control: Offset = 0, Limit left empty, same file — all five rows:** <img width="1340" height="869" alt="Screenshot 2026-08-05 at 2 19 47 PM" src="https://github.com/user-attachments/assets/cd747eee-b931-4e31-8399-ad666881e1cf"; /> The fix replaces the slice arithmetic with `drop(offset)` plus an optional `take(limit)` — the shape `CSVScanSourceOpExec` and `ArrowSourceOpExec` already use — so "no limit" is expressed by not bounding the iterator rather than by a sentinel value that arithmetic can overflow. Offset and limit still apply per extracted zip entry (unchanged behavior, now pinned by a test). ### Any related issues, documentation, discussions? Closes #7345 Same bug class as #7245 (JSONL scan, fixed by #7247). ### How was this PR tested? TDD: the regression tests were written first and confirmed to fail on the unfixed code — the four offset-without-limit cases all produced empty output (e.g. `List() did not equal List("l2", "l3", "l4", "l5")`) — then the fix was applied and all tests pass. Ten new cases were added across the three File Scan specs: eight in `FileScanUtilsSpec` (offset without limit — the regression, offset 0, offset with limit, limit only, offset past EOF, offset with an `Int.MaxValue` limit, per-zip-entry offset with `extract = true`, and `isSingle` types ignoring offset/limit — documented behavior, pinned), plus one operator-level offset-without-limit case each in `FileScanSourceOpDescSpec` (source operator) and `FileScanOpDescSpec` (input-port operator), since both operators delegate to the same helper. TDD: the regression tests were written first and confirmed to fail on the unfixed code — the four offset-without-limit cases all produced empty output (e.g. `List() did not equal List("l2", "l3", "l4", "l5")`) — then the fix was applied and all tests pass. Ten new cases were added across the three File Scan specs: eight in `FileScanUtilsSpec` (offset without limit — the regression, offset 0, offset with limit, limit only, offset past EOF, offset with an `Int.MaxValue` limit, per-zip-entry offset with `extract = true`, and `isSingle` types ignoring offset/limit — documented behavior, pinned), plus one operator-level offset-without-limit case each in `FileScanSourceOpDescSpec` (source operator) and `FileScanOpDescSpec` (input-port operator), since both operators delegate to the same helper. ```bash sbt "WorkflowOperator/testOnly org.apache.texera.amber.operator.source.scan.file.FileScanUtilsSpec org.apache.texera.amber.operator.source.scan.file.FileScanSourceOpDescSpec org.apache.texera.amber.operator.source.scan.file.FileScanOpDescSpec" # 29 tests, all passed (19 pre-existing + 10 new) sbt "WorkflowOperator/scalafixAll --check" # passed, no lint issues sbt "WorkflowOperator/scalafmtCheck" "WorkflowOperator/Test/scalafmtCheck" # passed, no mis-formatted files sbt WorkflowOperator/test # full module: 2050 tests in 283 suites, all passed ``` Also verified manually in the UI with the same two-operator workflow shown in the screenshots above: <img width="1133" height="762" alt="Screenshot 2026-08-05 at 5 20 41 PM" src="https://github.com/user-attachments/assets/fe2e3098-ddf4-4787-a611-fcbb3b61e19d"; /> ### Was this PR authored or co-authored using generative AI tooling? Co-authored by: Claude Code (Claude Fable 5) --- .../operator/source/scan/file/FileScanUtils.scala | 13 +- .../source/scan/file/FileScanOpDescSpec.scala | 26 ++++ .../scan/file/FileScanSourceOpDescSpec.scala | 21 +++ .../source/scan/file/FileScanUtilsSpec.scala | 141 ++++++++++++++++++++- 4 files changed, 191 insertions(+), 10 deletions(-) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanUtils.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanUtils.scala index a7f81b4869..2c52fa9e8e 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanUtils.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanUtils.scala @@ -110,22 +110,21 @@ private[file] object FileScanUtils { TupleLike(fields.toSeq: _*) } } else { - fileEntries.flatMap(entry => - new BufferedReader(new InputStreamReader(entry, fileEncoding.getCharset)) + fileEntries.flatMap { entry => + val lines = new BufferedReader(new InputStreamReader(entry, fileEncoding.getCharset)) .lines() .iterator() .asScala - .slice( - fileScanOffset.getOrElse(0), - fileScanOffset.getOrElse(0) + fileScanLimit.getOrElse(Int.MaxValue) - ) + .drop(fileScanOffset.getOrElse(0)) + fileScanLimit + .fold(lines)(lines.take) .map(line => TupleLike(attributeType match { case FileAttributeType.SINGLE_STRING => line case _ => parseField(line, attributeType.getType) }) ) - ) + } } new AutoClosingIterator(rawIterator, () => closeables.foreach(_.close())) diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDescSpec.scala index 06cb48fe81..c5a8ffc1b7 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDescSpec.scala @@ -78,6 +78,32 @@ class FileScanOpDescSpec extends AnyFlatSpec with BeforeAndAfter { fileScanOpExec.close() } + it should "read the lines after a 5-line offset from the input file path tuple when no limit is set" in { + fileScanOpDesc.attributeType = FileAttributeType.STRING + fileScanOpDesc.fileScanOffset = Option(5) + + val inputTuple = Tuple(inputSchema, Array[Any](TestOperators.TestTextFilePath)) + val fileScanOpExec = + new FileScanOpExec(objectMapper.writeValueAsString(fileScanOpDesc)) + + fileScanOpExec.open() + val processedTuple: Iterator[Tuple] = fileScanOpExec + .processTuple(inputTuple, 0) + .map(tupleLike => + tupleLike + .asInstanceOf[SchemaEnforceable] + .enforceSchema(fileScanOpDesc.sourceSchema()) + ) + + assert(processedTuple.next().getField("line").equals("line6")) + assert(processedTuple.next().getField("line").equals("line7")) + assert(processedTuple.next().getField("line").equals("line8")) + assert(processedTuple.next().getField("line").equals("line9")) + assert(processedTuple.next().getField("line").equals("line10")) + assertThrows[java.util.NoSuchElementException](processedTuple.next().getField("line")) + fileScanOpExec.close() + } + it should "preserve the original input filename when include filename is enabled" in { fileScanOpDesc.attributeType = FileAttributeType.SINGLE_STRING fileScanOpDesc.outputFileName = true diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpDescSpec.scala index 487a59154b..fafb696f13 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpDescSpec.scala @@ -94,6 +94,27 @@ class FileScanSourceOpDescSpec extends AnyFlatSpec with BeforeAndAfter { FileScanSourceOpExec.close() } + it should "read the lines after a 5-line offset when no limit is set" in { + fileScanSourceOpDesc.attributeType = FileAttributeType.STRING + fileScanSourceOpDesc.fileScanOffset = Option(5) + val FileScanSourceOpExec = + new FileScanSourceOpExec(objectMapper.writeValueAsString(fileScanSourceOpDesc)) + FileScanSourceOpExec.open() + val processedTuple: Iterator[Tuple] = FileScanSourceOpExec + .produceTuple() + .map(tupleLike => + tupleLike.asInstanceOf[SchemaEnforceable].enforceSchema(fileScanSourceOpDesc.sourceSchema()) + ) + + assert(processedTuple.next().getField("line").equals("line6")) + assert(processedTuple.next().getField("line").equals("line7")) + assert(processedTuple.next().getField("line").equals("line8")) + assert(processedTuple.next().getField("line").equals("line9")) + assert(processedTuple.next().getField("line").equals("line10")) + assertThrows[java.util.NoSuchElementException](processedTuple.next().getField("line")) + FileScanSourceOpExec.close() + } + it should "read first 5 lines of the input text file with CRLF separators into corresponding output tuples" in { fileScanSourceOpDesc.setResolvedFileName( FileResolver.resolve(TestOperators.TestCRLFTextFilePath) diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanUtilsSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanUtilsSpec.scala index ad2ffeb79e..6e170aa60c 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanUtilsSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanUtilsSpec.scala @@ -29,11 +29,11 @@ import java.util.zip.{ZipEntry, ZipOutputStream} class FileScanUtilsSpec extends AnyFlatSpec with BeforeAndAfterAll { - private val zips = scala.collection.mutable.ArrayBuffer.empty[Path] + private val tempFiles = scala.collection.mutable.ArrayBuffer.empty[Path] private def makeZip(entries: (String, String)*): String = { val path = Files.createTempFile("filescanutils-", ".zip") - zips += path + tempFiles += path val zipOut = new ZipOutputStream(new BufferedOutputStream(new FileOutputStream(path.toFile))) try { entries.foreach { @@ -48,8 +48,15 @@ class FileScanUtilsSpec extends AnyFlatSpec with BeforeAndAfterAll { path.toFile.toURI.toString } + private def makeTextFile(content: String): String = { + val path = Files.createTempFile("filescanutils-", ".txt") + tempFiles += path + Files.write(path, content.getBytes("UTF-8")) + path.toFile.toURI.toString + } + override def afterAll(): Unit = { - zips.foreach(Files.deleteIfExists) + tempFiles.foreach(Files.deleteIfExists) super.afterAll() } @@ -105,4 +112,132 @@ class FileScanUtilsSpec extends AnyFlatSpec with BeforeAndAfterAll { .toSeq assert(contents(tuples) == Seq("l1", "l2", "l3")) } + + it should "skip the offset lines and return all remaining lines when no limit is set" in { + val tuples = FileScanUtils + .createTuplesFromFile( + fileName = makeTextFile("l1\nl2\nl3\nl4\nl5"), + displayFileName = "d", + attributeType = FileAttributeType.STRING, + fileEncoding = FileDecodingMethod.UTF_8, + extract = false, + outputFileName = false, + fileScanOffset = Some(1), + fileScanLimit = None + ) + .toSeq + assert(contents(tuples) == Seq("l2", "l3", "l4", "l5")) + } + + it should "return every line for a zero offset with no limit" in { + val tuples = FileScanUtils + .createTuplesFromFile( + fileName = makeTextFile("l1\nl2\nl3\nl4\nl5"), + displayFileName = "d", + attributeType = FileAttributeType.STRING, + fileEncoding = FileDecodingMethod.UTF_8, + extract = false, + outputFileName = false, + fileScanOffset = Some(0), + fileScanLimit = None + ) + .toSeq + assert(contents(tuples) == Seq("l1", "l2", "l3", "l4", "l5")) + } + + it should "return limit lines starting at the offset when both are set" in { + val tuples = FileScanUtils + .createTuplesFromFile( + fileName = makeTextFile("l1\nl2\nl3\nl4\nl5"), + displayFileName = "d", + attributeType = FileAttributeType.STRING, + fileEncoding = FileDecodingMethod.UTF_8, + extract = false, + outputFileName = false, + fileScanOffset = Some(1), + fileScanLimit = Some(2) + ) + .toSeq + assert(contents(tuples) == Seq("l2", "l3")) + } + + it should "return the first limit lines when only a limit is set" in { + val tuples = FileScanUtils + .createTuplesFromFile( + fileName = makeTextFile("l1\nl2\nl3\nl4\nl5"), + displayFileName = "d", + attributeType = FileAttributeType.STRING, + fileEncoding = FileDecodingMethod.UTF_8, + extract = false, + outputFileName = false, + fileScanOffset = None, + fileScanLimit = Some(2) + ) + .toSeq + assert(contents(tuples) == Seq("l1", "l2")) + } + + it should "return no tuples when the offset is past the end of the file" in { + val tuples = FileScanUtils + .createTuplesFromFile( + fileName = makeTextFile("l1\nl2\nl3\nl4\nl5"), + displayFileName = "d", + attributeType = FileAttributeType.STRING, + fileEncoding = FileDecodingMethod.UTF_8, + extract = false, + outputFileName = false, + fileScanOffset = Some(99), + fileScanLimit = None + ) + .toSeq + assert(contents(tuples) == Seq.empty) + } + + it should "return no tuples for an Int.MaxValue offset without overflowing" in { + val tuples = FileScanUtils + .createTuplesFromFile( + fileName = makeTextFile("l1\nl2\nl3\nl4\nl5"), + displayFileName = "d", + attributeType = FileAttributeType.STRING, + fileEncoding = FileDecodingMethod.UTF_8, + extract = false, + outputFileName = false, + fileScanOffset = Some(Int.MaxValue), + fileScanLimit = None + ) + .toSeq + assert(contents(tuples) == Seq.empty) + } + + it should "apply an offset without a limit to each extracted zip entry independently" in { + val tuples = FileScanUtils + .createTuplesFromFile( + fileName = makeZip("a.txt" -> "a1\na2", "b.txt" -> "b1\nb2"), + displayFileName = "d", + attributeType = FileAttributeType.STRING, + fileEncoding = FileDecodingMethod.UTF_8, + extract = true, + outputFileName = false, + fileScanOffset = Some(1), + fileScanLimit = None + ) + .toSeq + assert(contents(tuples) == Seq("a2", "b2")) + } + + it should "ignore the offset for a single-tuple attribute type" in { + val tuples = FileScanUtils + .createTuplesFromFile( + fileName = makeTextFile("l1\nl2\nl3\nl4\nl5"), + displayFileName = "d", + attributeType = FileAttributeType.SINGLE_STRING, + fileEncoding = FileDecodingMethod.UTF_8, + extract = false, + outputFileName = false, + fileScanOffset = Some(1), + fileScanLimit = None + ) + .toSeq + assert(contents(tuples) == Seq("l1\nl2\nl3\nl4\nl5")) + } }
