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-4c26b3c5747e27aaef458a45f0f2bf6695a56cc8 in repository https://gitbox.apache.org/repos/asf/texera.git
commit 6b3cfb9bae35060f0ab1566c969da4c8c87e7bb7 Author: Xinyuan Lin <[email protected]> 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", "[email protected]") + .setCommitter("texera-test", "[email protected]") + .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") } }
