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/19 10:49:01 UTC
(texera) branch gh-readonly-queue/main/pr-7774-20099db3a38b320e5d24055958997e27d0908464 created (now 4326f0ad97)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-7774-20099db3a38b320e5d24055958997e27d0908464
in repository https://gitbox.apache.org/repos/asf/texera.git
at 4326f0ad97 test(common): cover the java-source executor factory, aggregate close, file-scan helpers and operator versioning (#7774)
This branch includes the following new commits:
new 4326f0ad97 test(common): cover the java-source executor factory, aggregate close, file-scan helpers and operator versioning (#7774)
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: test(common): cover the java-source executor factory, aggregate close, file-scan helpers and operator versioning (#7774)
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/main/pr-7774-20099db3a38b320e5d24055958997e27d0908464
in repository https://gitbox.apache.org/repos/asf/texera.git
commit 4326f0ad971d60e3049a74d186eb9c6d30f2cd4c
Author: Xinyuan Lin <xi...@uci.edu>
AuthorDate: Wed Aug 19 09:05:17 2026 +0000
test(common): cover the java-source executor factory, aggregate close, file-scan helpers and operator versioning (#7774)
### What changes were proposed in this PR?
Four small files across `common/workflow-core` and
`common/workflow-operator`, each with one narrow untested region. 14
tests added (65 -> 79 across the four specs).
Measured with `jacoco` under a `Test/testOptions` filter, the same
filter for the before and after runs:
| File | Lines before | Lines after | Branches |
|---|---|---|---|
| `ExecFactory.scala` | 17/19 (89.5%) | **19/19 (100%)** | 4/4 |
| `AggregateOpExec.scala` | 14/16 (87.5%) | **16/16 (100%)** | 2/2 |
| `FileScanUtils.scala` | 39/50 (78.0%) | **50/50 (100%)** | 12/20 ->
**19/20** |
| `OPVersion.java` | 9/17 (52.9%) | **14/17 (82.4%)** | 2/2 |
One number needs a caveat rather than a headline: the `FileScanUtils`
**78.0%** baseline is measured with only `FileScanUtilsSpec` running.
The reported module-wide figure is 90.0%, because
`FileScanSourceOpExecSpec` also touches that file — it is excluded here
for the reason below, from both runs, so the before/after comparison is
apples-to-apples even though the baseline differs from Codecov's.
Beyond the four target regions, `FileScanUtils`' 7-argument
`createTuplesFromFile` overload turned out to be uncovered as well and
is now covered, which accounts for most of that file's jump.
### Two things that look untestable and are not
**`newExecFromJavaCode` can be driven for real, and a stale comment said
otherwise.** `CoreExecutorReflectionSpec` claimed a success-path test
was impossible because system javac would see only its own classpath.
`common/workflow-core/build.sbt` sets `Test / fork := true`, so the
forked JVM's `java.class.path` is the full test classpath and
`compileCode`'s null options let javac resolve `OperatorExecutor`. A
Java class implementing the trait now compiles, instantiates and
dispatches through it. The comment is replaced with the mechanism.
**`OPVersion.getVersion` needs no real repository history** — throwaway
jgit repos in temp dirs are swapped into the private static `git` field
by reflection and restored in a `finally`, following
`JGitVersionControlSpec`.
### Verification
13 mutations, **12 killed, 1 equivalent**. Anchor uniqueness was
asserted for all 13 before any edit; each was applied alone, reverted,
with `git diff -- '*/src/main/*'` confirmed empty after every revert.
Every run executed a full suite (26/26/18/9 tests, never 0), so no
result is a compile-only error.
| Mutation | Killed by |
|---|---|
| `ExecFactory`: **exchange** sibling factories, body ->
`newExecFromJavaClassName(code)` | compiles java source into a live
OperatorExecutor (+2) |
| `ExecFactory`: swallow a failed compile, return `null` | surfaces the
compiler diagnostics when the java source does not compile |
| `ExecFactory`: memoize the first executor and return it forever |
hands back an independent instance on every call (+1) |
| `AggregateOpExec`: remove `keyedPartialAggregates.clear()` | `close()`
discards the accumulated groups |
| `AggregateOpExec`: `close()` becomes a no-op | same test |
| `FileScanUtils`: **exchange** the BINARY and SINGLE_STRING hint
literals | translates an out-of-memory read into advice to switch to
large binary (+1) |
| `FileScanUtils`: **exchange** the `case _` hint with the BINARY hint |
falls back to a generic hint for any other attribute type |
| `FileScanUtils`: flip ONE catch leg, `IllegalArgumentException` ->
`IOException` | treats an IllegalArgumentException as an over-size
failure too (+1) |
| `FileScanUtils`: **exchange** `fileScanOffset` and `fileScanLimit` in
the overload | forwards the offset and limit through the seven-argument
overload |
| `FileScanUtils`: **exchange** `displayFileName = fileName` ->
`attributeType.getName` | reads a file through the seven-argument
overload under its own name |
| `OPVersion`: flip the memo guard `!containsKey` -> `containsKey` | 8
of 9 tests |
| `OPVersion`: **exchange** the two in-scope strings in `put(...)` |
resolves the newest commit that touched the operator's own path (+1) |
**The equivalent mutant:** removing `distributedAggregations = null`
from `close()` survives, and that is correct. `close()` is terminal;
`onFinish` reads the field once per accumulated group and line 42 leaves
none; a reused executor's `open()` re-nulls it anyway. The reset is
observable only via `processTuple` after `close()` with no intervening
`open()` — an order the executor lifecycle never produces. Recorded in
the spec as deliberately not asserted, rather than cemented with an
illegal-sequence test.
### Deliberately not included
`FileScanUtils` line 99 stays branch-partial: an unreachable `Tuple2`
`MatchError` leg from `Iterator.duplicate`. `OPVersion` lines 31/36/39
are the static initializer, which runs before any test can observe it —
and which pair is hit depends on checkout shape, since a `git worktree`
makes `.git` a file.
Three observations are reported rather than pinned:
- **`OPVersion.getVersion` returns `null` on `GitAPIException`.** That
catch calls `printStackTrace()` and never populates `opMap`, so the
trailing `opMap.get(operatorName)` hands back `null` and every later
call retries the failing `git log`. Its `NullPointerException` sibling
stores `"N/A"`. The test asserts only the swallow, not the `null`, so a
fix is not blocked.
- **`FileScanSourceOpExecSpec` leaks a file handle on Windows.** Its
`afterAll` fails with `FileSystemException … test_large_binary.txt: The
process cannot access the file because it is being used by another
process`, which aborts the suite and takes the whole
`WorkflowOperator/jacoco` task with it. `AutoClosingIterator` closes
only on exhaustion, so a test stopping early leaks the handle; POSIX
`unlink` masks it on CI. Pre-existing, and the reason that spec is
filtered out of the measurements above.
- **`AggregateOpExec.close()` NPEs if `open()` was never called**
(`keyedPartialAggregates` is null). Lifecycle-guarded in practice.
No production file is touched.
### Any related issues, documentation, discussions?
Closes #7773
### How was this PR tested?
```
sbt "WorkflowCore/testOnly org.apache.texera.amber.core.executor.CoreExecutorReflectionSpec" "WorkflowOperator/testOnly org.apache.texera.amber.operator.aggregate.AggregateOpSpec org.apache.texera.amber.operator.source.scan.file.FileScanUtilsSpec org.apache.texera.amber.operator.metadata.OPVersionSpec"
```
```
[info] Tests: succeeded 26, failed 0, canceled 0, ignored 0, pending 0
[info] Tests: succeeded 53, failed 0, canceled 0, ignored 0, pending 0
```
0 suites aborted. `WorkflowCore/Test/scalafmtCheck`,
`WorkflowOperator/Test/scalafmtCheck` and both modules' `Test/scalafix
--check` all pass.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5)
---
.../core/executor/CoreExecutorReflectionSpec.scala | 62 +++++++--
.../amber/operator/aggregate/AggregateOpSpec.scala | 55 ++++++++
.../amber/operator/metadata/OPVersionSpec.scala | 155 ++++++++++++++++++---
.../source/scan/file/FileScanUtilsSpec.scala | 119 ++++++++++++++--
4 files changed, 353 insertions(+), 38 deletions(-)
diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/executor/CoreExecutorReflectionSpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/executor/CoreExecutorReflectionSpec.scala
index 3cec8b0e97..94aa189d7c 100644
--- a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/executor/CoreExecutorReflectionSpec.scala
+++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/executor/CoreExecutorReflectionSpec.scala
@@ -242,16 +242,60 @@ class CoreExecutorReflectionSpec extends AnyFlatSpec {
}
// ---------------------------------------------------------------------------
- // JavaRuntimeCompilation.compileCode
+ // ExecFactory.newExecFromJavaCode
//
- // A success-path test that compiles a real OperatorExecutor subclass from a
- // string is intentionally omitted: `compiler.getTask(...)` is invoked with
- // null compilation options, which means the system javac picks up its own
- // (test) classpath rather than the project classpath. Under sbt test that
- // does not include workflow-core itself, so the compile fails with
- // "package org.apache.texera... does not exist" — a deployment-environment
- // artifact rather than a contract violation. We exercise just the diagnostic
- // path here.
+ // `compileCode` passes null compilation options, so the system javac resolves
+ // types against its own default classpath — `java.class.path`. This module sets
+ // `Test / fork := true`, so the forked test JVM is launched with the full test
+ // classpath on `-cp` and javac therefore does see workflow-core's own classes;
+ // a source string that names `OperatorExecutor` compiles here.
+ // ---------------------------------------------------------------------------
+
+ /** Java source for a UDF class implementing the trait; `single` echoes the input tuple. */
+ private val echoUDFSource: String =
+ """public class JavaUDFOpExec
+ | implements org.apache.texera.amber.core.executor.OperatorExecutor {
+ | public scala.collection.Iterator<org.apache.texera.amber.core.tuple.TupleLike>
+ | processTuple(org.apache.texera.amber.core.tuple.Tuple tuple, int port) {
+ | return scala.collection.Iterator$.MODULE$.single(tuple);
+ | }
+ |}""".stripMargin
+
+ "ExecFactory.newExecFromJavaCode" should "compile java source into a live OperatorExecutor" in {
+ val exec = ExecFactory.newExecFromJavaCode(echoUDFSource)
+ assert(exec.getClass.getName == "org.apache.texera.amber.operators.udf.java.JavaUDFOpExec")
+ // Reachable through the trait, not merely castable to it: dispatching
+ // processTuple must run the freshly compiled override...
+ assert(exec.processTuple(tuple(7), 0).toList == List(tuple(7)))
+ // ...and the class must inherit the trait's defaults for what it left out.
+ assert(exec.onFinish(0).isEmpty)
+ assert(exec.produceStateOnStart(0).isEmpty)
+ }
+
+ it should "hand back an independent instance on every call" in {
+ // The factory constructs per call; two executors compiled from the same
+ // source must not be the same object (nor share a class-level cache).
+ val first = ExecFactory.newExecFromJavaCode(echoUDFSource)
+ val second = ExecFactory.newExecFromJavaCode(echoUDFSource)
+ assert(first ne second)
+ assert(first.getClass.getName == second.getClass.getName)
+ }
+
+ it should "surface the compiler diagnostics when the java source does not compile" in {
+ // A class that claims the trait but never implements `processTuple` is
+ // rejected by javac, so no instance is ever constructed.
+ val ex = intercept[RuntimeException] {
+ ExecFactory.newExecFromJavaCode(
+ """public class JavaUDFOpExec
+ | implements org.apache.texera.amber.core.executor.OperatorExecutor {
+ |}""".stripMargin
+ )
+ }
+ assert(ex.getMessage.contains("Error at line"))
+ }
+
+ // ---------------------------------------------------------------------------
+ // JavaRuntimeCompilation.compileCode
// ---------------------------------------------------------------------------
"JavaRuntimeCompilation.compileCode" should "compile a self-contained Java class with no external deps" in {
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/aggregate/AggregateOpSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/aggregate/AggregateOpSpec.scala
index 3730181d9d..1cbec4535f 100644
--- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/aggregate/AggregateOpSpec.scala
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/aggregate/AggregateOpSpec.scala
@@ -592,4 +592,59 @@ class AggregateOpSpec extends AnyFunSuite {
assert(results.size == 1)
assert(results.head.getFields(0).asInstanceOf[Number].intValue() == 3)
}
+
+ // `close()` also resets `distributedAggregations`, which is deliberately not
+ // asserted: `onFinish` reads that list once per accumulated group and `close()`
+ // leaves none, while a reopened executor resets it again — so the reset is only
+ // observable by calling `processTuple` after `close()` without an intervening
+ // `open()`, which the executor lifecycle never does. Pinning it would cement an
+ // illegal call order.
+ test("AggregateOpExec close() discards the accumulated groups") {
+ val schema = makeSchema(
+ "city" -> AttributeType.STRING,
+ "sales" -> AttributeType.INTEGER
+ )
+
+ val desc = new AggregateOpDesc()
+ desc.aggregations = List(makeAggregationOp(AggregationFunction.SUM, "sales", "total_sales"))
+ desc.groupByKeys = List("city")
+
+ val exec = new AggregateOpExec(objectMapper.writeValueAsString(desc))
+ exec.open()
+ exec.processTuple(makeTuple(schema, "NY", 10), 0)
+ exec.processTuple(makeTuple(schema, "SF", 20), 0)
+
+ // Guard the fixture: the two groups really are accumulated, so the empty
+ // result asserted after close() cannot pass vacuously.
+ assert(exec.onFinish(0).toList.size == 2)
+
+ exec.close()
+ assert(exec.onFinish(0).toList.isEmpty)
+ }
+
+ test("AggregateOpExec open() after close() accumulates from an empty state") {
+ // The executor object outlives a single run, so a reopened executor must not
+ // fold the previous run's partial aggregates into the new result.
+ val schema = makeSchema(
+ "city" -> AttributeType.STRING,
+ "sales" -> AttributeType.INTEGER
+ )
+
+ val desc = new AggregateOpDesc()
+ desc.aggregations = List(makeAggregationOp(AggregationFunction.SUM, "sales", "total_sales"))
+ desc.groupByKeys = List("city")
+
+ val exec = new AggregateOpExec(objectMapper.writeValueAsString(desc))
+ exec.open()
+ exec.processTuple(makeTuple(schema, "NY", 10), 0)
+ exec.close()
+
+ exec.open()
+ exec.processTuple(makeTuple(schema, "NY", 3), 0)
+
+ val results = exec.onFinish(0).toList
+ assert(results.size == 1)
+ // 3, not 13: the first run's partial sum for "NY" must have been dropped.
+ assert(results.head.getFields(1).asInstanceOf[Number].intValue() == 3)
+ }
}
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/metadata/OPVersionSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/metadata/OPVersionSpec.scala
index 8ca517a7de..a61014569a 100644
--- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/metadata/OPVersionSpec.scala
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/metadata/OPVersionSpec.scala
@@ -19,10 +19,17 @@
package org.apache.texera.amber.operator.metadata
+import org.eclipse.jgit.api.Git
+import org.eclipse.jgit.api.errors.GitAPIException
+import org.eclipse.jgit.revwalk.RevCommit
import org.scalatest.flatspec.AnyFlatSpec
import org.scalatest.matchers.should.Matchers
+import java.nio.charset.StandardCharsets
+import java.nio.file.{Files, Path}
import java.util.UUID
+import scala.jdk.CollectionConverters._
+import scala.util.Using
/**
* `OPVersion` resolves an operator's version from the git history of the file
@@ -34,20 +41,33 @@ import java.util.UUID
* a file rather than a directory and jgit raises `RepositoryNotFoundException`,
* leaving the handle null; in a plain clone the handle opens and an unknown path
* yields an empty log instead. Rather than assert whatever both happen to share,
- * this spec pins the handle to a known state: every test forces the private
- * static `git` field to null for its duration and restores the original value
- * afterwards, so the assertions describe one deterministic code path regardless
- * of how the tree was checked out (and no other suite in the JVM is affected).
+ * this spec pins the handle to a known state: every test swaps the private static
+ * `git` field to a handle it controls (or to null) for its duration and restores
+ * the original value afterwards, so the assertions describe one deterministic code
+ * path regardless of how the tree was checked out (and no other suite in the JVM
+ * is affected).
*
- * With the handle null, `git.log()` throws `NullPointerException` and resolution
- * takes the `"N/A"` fallback. On top of that this spec pins the memoization
- * contract: the answer is cached per operator name and the path is ignored on a
- * cache hit.
+ * The three resolution outcomes are pinned against throwaway repositories built
+ * in a temp directory, so nothing here depends on the checkout the suite runs in:
*
- * Deliberately NOT covered: the success path that returns a real commit hash and
- * the `GitAPIException` catch (which, note, leaves `opMap` unpopulated and so
- * returns null). Both require a specific, openable repository state that is not
- * guaranteed for a test run.
+ * - a repository with history: the newest commit touching the operator's path,
+ * memoized under the operator name;
+ * - a repository with an unborn HEAD: `LogCommand.call()` raises `NoHeadException`
+ * (a `GitAPIException`) and resolution must not propagate it;
+ * - no handle at all: `git.log()` raises `NullPointerException` and resolution
+ * takes the `"N/A"` fallback.
+ *
+ * On top of that this spec pins the memoization contract: the answer is cached per
+ * operator name and the path is ignored on a cache hit.
+ *
+ * Deliberately NOT covered: the static initializer itself. It runs once, before any
+ * test can observe it, and which of its two branches executes is fixed by how the
+ * tree was checked out — no test can flip it without mutating the JVM's environment.
+ *
+ * Deliberately NOT asserted: the value the `GitAPIException` path returns. That catch
+ * leaves `opMap` unpopulated, so the trailing `opMap.get(operatorName)` hands the caller
+ * a null version — a latent defect (the `NullPointerException` sibling stores `"N/A"`).
+ * Pinning the null would cement it, so only the swallow itself is asserted.
*/
class OPVersionSpec extends AnyFlatSpec with Matchers {
@@ -68,22 +88,121 @@ class OPVersionSpec extends AnyFlatSpec with Matchers {
}
/**
- * Runs `body` with `OPVersion`'s private static git handle forced to null, so the
- * fallback path is exercised deterministically, then restores whatever was there.
+ * Runs `body` with `OPVersion`'s private static git handle forced to `handle`, so a
+ * chosen resolution path is exercised deterministically, then restores whatever was
+ * there.
*/
- private def withNullGit[T](body: => T): T = {
+ private def withGit[T](handle: Git)(body: => T): T = {
val field = declaredField("git")
val original = field.get(null)
- field.set(null, null)
+ field.set(null, handle)
try body
finally field.set(null, original)
}
+ private def withNullGit[T](body: => T): T = withGit(null)(body)
+
private def withCleanCache[T](names: String*)(body: => T): T =
try body
finally names.foreach(opMap.remove)
- "OPVersion.getVersion" should "fall back to \"N/A\" when the git handle is unavailable" in {
+ private def deleteRecursively(path: Path): Unit =
+ if (Files.exists(path)) {
+ Using.resource(Files.walk(path)) { stream =>
+ stream.iterator().asScala.toSeq.reverse.foreach { p =>
+ // best-effort: jgit can keep Windows handles on .git objects; leftover temp
+ // files are harmless and get reaped by the OS.
+ try Files.deleteIfExists(p)
+ catch { case _: java.io.IOException => () }
+ }
+ }
+ }
+
+ /** Runs `body` against a throwaway repository, closing the handle and reaping the directory. */
+ private def withTempRepo[T](prefix: String)(body: (Git, Path) => T): T = {
+ val dir = Files.createTempDirectory(prefix)
+ val handle = Git.init().setDirectory(dir.toFile).call()
+ try body(handle, dir)
+ finally {
+ handle.close()
+ deleteRecursively(dir)
+ }
+ }
+
+ /** Writes `name` and commits just that path; jgit needs an explicit identity here. */
+ private def commitFile(handle: Git, dir: Path, name: String, content: String): RevCommit = {
+ Files.write(dir.resolve(name), content.getBytes(StandardCharsets.UTF_8))
+ handle.add().addFilepattern(name).call()
+ handle
+ .commit()
+ .setMessage(s"touch $name")
+ .setAuthor("texera-test", "texera-test@example.com")
+ .setCommitter("texera-test", "texera-test@example.com")
+ .call()
+ }
+
+ // -- resolution against a repository with history ----------------------------
+
+ "OPVersion.getVersion" should "resolve the newest commit that touched the operator's own path" in {
+ val alphaOp = uniqueName()
+ val betaOp = uniqueName()
+ withCleanCache(alphaOp, betaOp) {
+ withTempRepo("opversion-history") { (handle, dir) =>
+ // Three commits, interleaved so every wrong answer is a different hash from
+ // the right one: alpha's answer is neither the tip of the log (that is beta's)
+ // nor the oldest commit that touched it.
+ val alphaFirst = commitFile(handle, dir, "alpha.txt", "alpha v1")
+ val alphaHead = commitFile(handle, dir, "alpha.txt", "alpha v2")
+ val betaHead = commitFile(handle, dir, "beta.txt", "beta v1")
+
+ // Guard the fixture: three distinct hashes, so none of the assertions below
+ // can pass by coincidence.
+ Set(alphaFirst.getName, alphaHead.getName, betaHead.getName) should have size 3
+
+ withGit(handle) {
+ OPVersion.getVersion(alphaOp, "alpha.txt") shouldBe alphaHead.getName
+ OPVersion.getVersion(betaOp, "beta.txt") shouldBe betaHead.getName
+ }
+ }
+ }
+ }
+
+ it should "memoize the resolved commit hash under the operator name" in {
+ val name = uniqueName()
+ withCleanCache(name) {
+ withTempRepo("opversion-memo") { (handle, dir) =>
+ val head = commitFile(handle, dir, "alpha.txt", "alpha v1")
+
+ withGit(handle) {
+ opMap.containsKey(name) shouldBe false
+ OPVersion.getVersion(name, "alpha.txt") shouldBe head.getName
+ opMap.get(name) shouldBe head.getName
+ }
+ }
+ }
+ }
+
+ it should "swallow a git failure instead of propagating it to the caller" in {
+ val name = uniqueName()
+ withCleanCache(name) {
+ withTempRepo("opversion-unborn") { (handle, _) =>
+ // A freshly initialised repository has an unborn HEAD, so LogCommand.call()
+ // raises NoHeadException -- a checked GitAPIException, a different catch from
+ // the NullPointerException that a missing handle produces. Asserted first so
+ // this test cannot pass vacuously if jgit ever starts returning an empty log.
+ a[GitAPIException] should be thrownBy
+ handle.log().addPath("alpha.txt").setMaxCount(1).call()
+
+ withGit(handle) {
+ noException should be thrownBy OPVersion.getVersion(name, "alpha.txt")
+ }
+ }
+ }
+ }
+
+ // -- resolution with no usable handle ----------------------------------------
+
+ it should "fall back to \"N/A\" when the git handle is unavailable" in {
val name = uniqueName()
withCleanCache(name) {
withNullGit {
@@ -113,6 +232,8 @@ class OPVersionSpec extends AnyFlatSpec with Matchers {
}
}
+ // -- memoization contract ----------------------------------------------------
+
it should "serve the memoized value and ignore the path on subsequent calls" in {
val name = uniqueName()
withCleanCache(name) {
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 2de0eacfac..a560ad0600 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
@@ -23,7 +23,7 @@ import org.apache.texera.amber.operator.source.scan.{FileAttributeType, FileDeco
import org.scalatest.BeforeAndAfterAll
import org.scalatest.flatspec.AnyFlatSpec
-import java.io.{BufferedOutputStream, FileOutputStream}
+import java.io.{BufferedOutputStream, FileOutputStream, IOException, InputStream}
import java.nio.file.{Files, Path}
import java.util.zip.{ZipEntry, ZipOutputStream}
@@ -241,20 +241,115 @@ class FileScanUtilsSpec extends AnyFlatSpec with BeforeAndAfterAll {
assert(contents(tuples) == Seq("l1\nl2\nl3\nl4\nl5"))
}
- it should "throw RuntimeException when binary file exceeds natural memory limit" in {
- val mockLargeInputStream = new java.io.InputStream {
- override def read(): Int = {
- throw new OutOfMemoryError("Requested array size exceeds VM limit")
- }
- override def read(b: Array[Byte], off: Int, len: Int): Int = {
- throw new OutOfMemoryError("Requested array size exceeds VM limit")
- }
+ it should "read a file through the seven-argument overload under its own name" in {
+ // The short overload exists only to default displayFileName to fileName, so the
+ // emitted file-name column is the observation that pins the delegation.
+ val uri = makeTextFile("only line")
+ val tuples = FileScanUtils
+ .createTuplesFromFile(
+ fileName = uri,
+ attributeType = FileAttributeType.SINGLE_STRING,
+ fileEncoding = FileDecodingMethod.UTF_8,
+ extract = false,
+ outputFileName = true,
+ fileScanOffset = None,
+ fileScanLimit = None
+ )
+ .toSeq
+ assert(tuples.size == 1)
+ assert(tuples.head.getFields.toSeq == Seq(uri, "only line"))
+ }
+
+ it should "forward the offset and limit through the seven-argument overload" in {
+ val tuples = FileScanUtils
+ .createTuplesFromFile(
+ fileName = makeTextFile("l1\nl2\nl3\nl4\nl5"),
+ attributeType = FileAttributeType.STRING,
+ fileEncoding = FileDecodingMethod.UTF_8,
+ extract = false,
+ outputFileName = false,
+ fileScanOffset = Some(1),
+ fileScanLimit = Some(2)
+ )
+ .toSeq
+ assert(contents(tuples) == Seq("l2", "l3"))
+ }
+
+ // -- safeToByteArray: the out-of-memory guard ---------------------------------
+
+ /** A stream whose every read fails with `failure`, to drive safeToByteArray's catch. */
+ private def failingStream(failure: Throwable): InputStream =
+ new InputStream {
+ override def read(): Int = throw failure
+ override def read(b: Array[Byte], off: Int, len: Int): Int = throw failure
+ }
+
+ "FileScanUtils.safeToByteArray" should
+ "translate an out-of-memory read into advice to switch to large binary" in {
+ val exception = intercept[RuntimeException] {
+ FileScanUtils.safeToByteArray(
+ failingStream(new OutOfMemoryError("Requested array size exceeds VM limit")),
+ FileAttributeType.BINARY
+ )
+ }
+ assert(
+ exception.getMessage ==
+ "File exceeds maximum safe memory size for 'binary' type. " +
+ "Please use 'large binary' attribute type instead."
+ )
+ }
+
+ it should "advise splitting or chunking the file for a single-string attribute" in {
+ // 'large binary' is meaningless for a text column, so this arm has to give
+ // different advice from the binary one.
+ val exception = intercept[RuntimeException] {
+ FileScanUtils.safeToByteArray(
+ failingStream(new OutOfMemoryError("Requested array size exceeds VM limit")),
+ FileAttributeType.SINGLE_STRING
+ )
+ }
+ assert(
+ exception.getMessage ==
+ "File exceeds maximum safe memory size for 'single string' type. " +
+ "Please split the file or use a chunked reading method."
+ )
+ }
+
+ it should "fall back to a generic hint for any other attribute type" in {
+ val exception = intercept[RuntimeException] {
+ FileScanUtils.safeToByteArray(
+ failingStream(new OutOfMemoryError("Requested array size exceeds VM limit")),
+ FileAttributeType.STRING
+ )
}
+ assert(
+ exception.getMessage ==
+ "File exceeds maximum safe memory size for 'string' type. " +
+ "File is too large to fit in memory."
+ )
+ }
+ it should "treat an IllegalArgumentException from the stream as an over-size failure too" in {
+ // An over-large allocation surfaces as IllegalArgumentException rather than
+ // OutOfMemoryError on some streams, so the guard catches both.
val exception = intercept[RuntimeException] {
- FileScanUtils.safeToByteArray(mockLargeInputStream, FileAttributeType.BINARY)
+ FileScanUtils.safeToByteArray(
+ failingStream(new IllegalArgumentException("negative capacity")),
+ FileAttributeType.BINARY
+ )
+ }
+ assert(exception.getMessage.startsWith("File exceeds maximum safe memory size"))
+ }
+
+ it should "let an unrelated read failure propagate untouched" in {
+ // Only the two over-size signals are translated; a genuine I/O failure must
+ // reach the caller as itself rather than as a misleading memory diagnosis.
+ val exception = intercept[IOException] {
+ FileScanUtils.safeToByteArray(
+ failingStream(new IOException("disk went away")),
+ FileAttributeType.BINARY
+ )
}
- assert(exception.getMessage.contains("exceeds maximum safe memory size"))
- assert(exception.getMessage.contains("Please use 'large binary'"))
+ assert(exception.getMessage == "disk went away")
}
}