From 6ec90cc041a027a36ed506d9fd43f852bde0ffd6 Mon Sep 17 00:00:00 2001 From: eugenegujing Date: Wed, 5 Aug 2026 16:38:07 -0700 Subject: [PATCH] fix(workflow-operator): file scan operator using offset with an empty limit emits no rows Replace 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 that arithmetic can overflow. Offset and limit still apply per extracted zip entry, and a test now pins that semantics. Ten regression tests: eight in FileScanUtilsSpec (offset without limit, offset zero, offset with limit, limit only, offset past EOF, offset Int.MaxValue, per-zip-entry offset, isSingle ignoring offset) plus one offset-without-limit case each in FileScanSourceOpDescSpec and FileScanOpDescSpec. Before the source change the four offset-without-limit cases failed on empty results; all 29 tests in the three specs pass after it. Closes #7345 --- .../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 a7f81b4869c..2c52fa9e8ea 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 06cb48fe813..c5a8ffc1b7e 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 487a59154b7..fafb696f131 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 ad2ffeb79e6..6e170aa60c4 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")) + } }