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-7530-9ead9e9b457a63787f52b81b608baffbc1853582 in repository https://gitbox.apache.org/repos/asf/texera.git
commit 80de40f1b3b4900cd2a1cb49642dc9174dfdb0c3 Author: Eugene Gu <[email protected]> AuthorDate: Thu Aug 13 05:07:34 2026 +0000 fix(workflow-operator, v1.2): File Scan operator using offset with an empty limit emits no rows (#7530) ### What changes were proposed in this PR? Backport of #7348 to `release/v1.2`, cherry-picked from main commit 898280316. `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. 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. `FileScanUtils.scala` on `release/v1.2` is byte-identical to main's pre-fix version, so the fix applies unchanged. One adaptation was needed: `FileScanUtilsSpec.scala` was created on main (#6077) after v1.2 branched, so the cherry-pick hit a modify/delete conflict; this PR adds the file with main's full post-fix content. As a side effect, v1.2 also gains the spec's 3 pre-existing main-only tests (zip extraction, `__MACOSX` filtering, per-line flat-map) in addition to the 8 fix-related ones. They target `FileScanUtils` behavior that is identical on v1.2 and all pass. ### Any related issues, documentation, discussions? Backport of #7348 (originally closed #7345). ### How was this PR tested? The regression tests from #7348 come along with the cherry-pick. On this branch: ```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" # 3 suites, 25 tests, all passed # (main has 29: four getPhysicalOp/propagateSchema coverage tests were added # to these specs 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: Xuan Gu <[email protected]> --- .../operator/source/scan/file/FileScanUtils.scala | 13 +- .../source/scan/file/FileScanOpDescSpec.scala | 26 +++ .../scan/file/FileScanSourceOpDescSpec.scala | 21 ++ .../source/scan/file/FileScanUtilsSpec.scala | 243 +++++++++++++++++++++ 4 files changed, 296 insertions(+), 7 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 e1749b98d3..cd93ac5f69 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 @@ -76,6 +76,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 4437c018bd..dee4cfc037 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 @@ -91,6 +91,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 new file mode 100644 index 0000000000..6e170aa60c --- /dev/null +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanUtilsSpec.scala @@ -0,0 +1,243 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.texera.amber.operator.source.scan.file + +import org.apache.texera.amber.operator.source.scan.{FileAttributeType, FileDecodingMethod} +import org.scalatest.BeforeAndAfterAll +import org.scalatest.flatspec.AnyFlatSpec + +import java.io.{BufferedOutputStream, FileOutputStream} +import java.nio.file.{Files, Path} +import java.util.zip.{ZipEntry, ZipOutputStream} + +class FileScanUtilsSpec extends AnyFlatSpec with BeforeAndAfterAll { + + private val tempFiles = scala.collection.mutable.ArrayBuffer.empty[Path] + + private def makeZip(entries: (String, String)*): String = { + val path = Files.createTempFile("filescanutils-", ".zip") + tempFiles += path + val zipOut = new ZipOutputStream(new BufferedOutputStream(new FileOutputStream(path.toFile))) + try { + entries.foreach { + case (name, content) => + zipOut.putNextEntry(new ZipEntry(name)) + zipOut.write(content.getBytes("UTF-8")) + zipOut.closeEntry() + } + } finally { + zipOut.close() + } + 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 = { + tempFiles.foreach(Files.deleteIfExists) + super.afterAll() + } + + private def contents(tuples: Seq[org.apache.texera.amber.core.tuple.TupleLike]): Seq[Any] = + tuples.map(_.getFields.head) + + "FileScanUtils.createTuplesFromFile" should + "extract every zip entry as a single-string tuple" in { + val tuples = FileScanUtils + .createTuplesFromFile( + fileName = makeZip("a.txt" -> "Content A", "b.txt" -> "Content B"), + displayFileName = "ignored-when-extracting", + attributeType = FileAttributeType.SINGLE_STRING, + fileEncoding = FileDecodingMethod.UTF_8, + extract = true, + outputFileName = false, + fileScanOffset = None, + fileScanLimit = None + ) + .toSeq + assert(tuples.size == 2) + assert(contents(tuples).toSet == Set("Content A", "Content B")) + } + + it should "drop __MACOSX metadata entries when extracting" in { + val tuples = FileScanUtils + .createTuplesFromFile( + fileName = makeZip("real.txt" -> "keep me", "__MACOSX/._real.txt" -> "junk"), + displayFileName = "d", + attributeType = FileAttributeType.SINGLE_STRING, + fileEncoding = FileDecodingMethod.UTF_8, + extract = true, + outputFileName = false, + fileScanOffset = None, + fileScanLimit = None + ) + .toSeq + assert(contents(tuples) == Seq("keep me")) + } + + it should "flat-map each line of an extracted entry for a per-line attribute type" in { + val tuples = FileScanUtils + .createTuplesFromFile( + fileName = makeZip("lines.txt" -> "l1\nl2\nl3"), + displayFileName = "d", + attributeType = FileAttributeType.STRING, + fileEncoding = FileDecodingMethod.UTF_8, + extract = true, + outputFileName = false, + fileScanOffset = None, + fileScanLimit = None + ) + .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")) + } +}
