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:07:57 UTC
(texera) branch gh-readonly-queue/release/v1.2/pr-7530-9ead9e9b457a63787f52b81b608baffbc1853582 created (now 80de40f1b3)
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-7530-9ead9e9b457a63787f52b81b608baffbc1853582
in repository https://gitbox.apache.org/repos/asf/texera.git
at 80de40f1b3 fix(workflow-operator, v1.2): File Scan operator using offset with an empty limit emits no rows (#7530)
This branch includes the following new commits:
new 80de40f1b3 fix(workflow-operator, v1.2): File Scan operator using offset with an empty limit emits no rows (#7530)
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): File Scan operator using offset with an empty limit emits no rows (#7530)
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-7530-9ead9e9b457a63787f52b81b608baffbc1853582
in repository https://gitbox.apache.org/repos/asf/texera.git
commit 80de40f1b3b4900cd2a1cb49642dc9174dfdb0c3
Author: Eugene Gu <eu...@outlook.com>
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 <16...@users.noreply.github.com>
---
.../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"))
+ }
+}