This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/texera.git
The following commit(s) were added to refs/heads/main by this push:
new 8982803165 fix(workflow-operator): File Scan operator using offset
with an empty limit emits no rows (#7348)
8982803165 is described below
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"))
+ }
}
