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/07/27 02:44:12 UTC
(texera) branch gh-readonly-queue/main/pr-6902-f5217504c3cfd8ac1d56f6617d7849895af17f59 created (now 9de13865a0)
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-6902-f5217504c3cfd8ac1d56f6617d7849895af17f59
in repository https://gitbox.apache.org/repos/asf/texera.git
at 9de13865a0 test(workflow-execution-service): extend ResultExportService unit test coverage (#6902)
This branch includes the following new commits:
new 9de13865a0 test(workflow-execution-service): extend ResultExportService unit test coverage (#6902)
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(workflow-execution-service): extend ResultExportService unit test coverage (#6902)
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-6902-f5217504c3cfd8ac1d56f6617d7849895af17f59
in repository https://gitbox.apache.org/repos/asf/texera.git
commit 9de13865a0bda4f93041662f3ff736cdb8724900
Author: Meng Wang <me...@uci.edu>
AuthorDate: Sun Jul 26 19:43:50 2026 -0700
test(workflow-execution-service): extend ResultExportService unit test coverage (#6902)
### What changes were proposed in this PR?
`ResultExportService` sits at ~6% line coverage: the existing spec only
covers `parseOperators` and `validateExportRequest`, leaving the whole
export-encoding layer untested. The `streamDocumentAs*` writers look
storage-bound but are not — each takes a `VirtualDocument[Tuple]`
parameter, so a small in-spec fake document makes them testable against
a `ByteArrayOutputStream`, with no Iceberg, MinIO or DB.
This extends `ResultExportServiceSpec` with 19 tests that reach the
private writers through `PrivateMethodTester`: `streamDocumentAsCSV`
(both early returns, inferred vs supplied headers, and a document larger
than `Constants.CHUNK_SIZE` to exercise the buffered chunk loop),
`streamCellData` (rowIndex/columnIndex validation messages, an
unretrievable row, and the happy-path bytes), `convertFieldToBytes` (all
three branches), `streamDocumentAsHTML`, `streamDocumentAsParquetZip`
(verbatim copy plus source-stream close, and error propagation),
`streamDocumentAsArrow` (round-trips the tuples back through
`ArrowFileReader`), and `NonClosingOutputStream`.
Two existing behaviors are pinned as-is rather than changed: with
supplied headers the first row is not consumed by header inference, so
it still appears in the body (the asymmetry against the `None` branch),
and `streamDocumentAsHTML` throws `NoSuchElementException` on an empty
document because `.head` is unguarded. `getOperatorDocument`,
`saveToDatasets`, `saveStreamToDataset` and `generateFileName` stay
uncovered — they need `DocumentFactory.openDocument`, a live
file-service endpoint, or embed `LocalDateTime.now()`.
### Any related issues, documentation, discussions?
Closes #6898. Extends the spec added by #6705.
### How was this PR tested?
`sbt "WorkflowExecutionService/testOnly *ResultExportServiceSpec"`
passes with 24 tests (5 existing, 19 new); `Test/scalafmtCheck` and
`Test/scalafix --check` are clean. Verified the failure path by
inverting the supplied-headers assertion and confirming the suite went
red and exited non-zero. Every test runs in-process against an in-spec
fake document, so no external service is involved.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Fable 5)
---
.../web/service/ResultExportServiceSpec.scala | 272 ++++++++++++++++++++-
1 file changed, 268 insertions(+), 4 deletions(-)
diff --git a/amber/src/test/scala/org/apache/texera/web/service/ResultExportServiceSpec.scala b/amber/src/test/scala/org/apache/texera/web/service/ResultExportServiceSpec.scala
index 7d8870d41a..eec51ed5c2 100644
--- a/amber/src/test/scala/org/apache/texera/web/service/ResultExportServiceSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/service/ResultExportServiceSpec.scala
@@ -20,16 +20,30 @@
package org.apache.texera.web.service
import com.fasterxml.jackson.core.JsonProcessingException
-import jakarta.ws.rs.core.Response
+import org.apache.arrow.memory.RootAllocator
+import org.apache.arrow.vector.ipc.ArrowFileReader
+import org.apache.arrow.vector.util.ByteArrayReadableSeekableByteChannel
+import org.apache.texera.amber.core.storage.model.VirtualDocument
+import org.apache.texera.amber.core.tuple.{AttributeType, Schema, Tuple}
import org.apache.texera.amber.core.virtualidentity.WorkflowIdentity
+import org.apache.texera.amber.util.ArrowUtils
import org.apache.texera.web.model.http.request.result.{OperatorExportInfo, ResultExportRequest}
+import org.scalatest.PrivateMethodTester
import org.scalatest.flatspec.AnyFlatSpec
import org.scalatest.matchers.should.Matchers
+import java.io.{ByteArrayInputStream, ByteArrayOutputStream, InputStream}
+import java.net.URI
+import java.nio.charset.StandardCharsets
+import javax.ws.rs.WebApplicationException
+import javax.ws.rs.core.Response
+
// Unit tests for the ResultExportService request helpers parseOperators and
-// validateExportRequest. Constructing the service is cheap (it only stores the
-// identity and computing-unit id), so no engine/DB dependency is needed here.
-class ResultExportServiceSpec extends AnyFlatSpec with Matchers {
+// validateExportRequest, plus the private export-encoding writers. Constructing
+// the service is cheap (it only stores the identity and computing-unit id), and
+// every writer takes a VirtualDocument parameter, so an in-spec fake document is
+// enough — no Iceberg, MinIO or DB is involved.
+class ResultExportServiceSpec extends AnyFlatSpec with Matchers with PrivateMethodTester {
private val service = new ResultExportService(WorkflowIdentity(1L), computingUnitId = 0)
@@ -85,4 +99,254 @@ class ResultExportServiceSpec extends AnyFlatSpec with Matchers {
service.validateExportRequest(requestWith(List(OperatorExportInfo("op-1", "csv"))))
result shouldBe None
}
+
+ // -- fakes for the export writers -------------------------------------------
+
+ private val schema: Schema = Schema()
+ .add("name", AttributeType.STRING)
+ .add("count", AttributeType.INTEGER)
+
+ private def tupleOf(name: String, count: Int): Tuple =
+ Tuple.builder(schema).addSequentially(Array[Any](name, count)).build()
+
+ // A minimal in-memory document overriding only the members the writers touch;
+ // every other VirtualDocument method keeps its default throwing body.
+ private class FakeDoc(
+ rows: Seq[Tuple],
+ countOverride: Option[Long] = None,
+ stream: Option[InputStream] = None
+ ) extends VirtualDocument[Tuple] {
+ override def getURI: URI = new URI("file:///stub/export")
+ override def clear(): Unit = ()
+ override def getCount: Long = countOverride.getOrElse(rows.length.toLong)
+ override def get(): Iterator[Tuple] = rows.iterator
+ override def getRange(from: Int, until: Int, columns: Option[Seq[String]]): Iterator[Tuple] =
+ rows.slice(from, until).iterator
+ override def asInputStream(): InputStream = stream.getOrElse(super.asInputStream())
+ }
+
+ private class TrackingInputStream(bytes: Array[Byte]) extends ByteArrayInputStream(bytes) {
+ var closed = false
+ override def close(): Unit = { closed = true; super.close() }
+ }
+
+ private class TrackingOutputStream extends ByteArrayOutputStream {
+ var closed = false
+ override def close(): Unit = { closed = true; super.close() }
+ }
+
+ private val streamDocumentAsCSV = PrivateMethod[Unit](Symbol("streamDocumentAsCSV"))
+ private val streamDocumentAsArrow = PrivateMethod[Unit](Symbol("streamDocumentAsArrow"))
+ private val streamDocumentAsHTML = PrivateMethod[Unit](Symbol("streamDocumentAsHTML"))
+ private val streamDocumentAsParquetZip =
+ PrivateMethod[Unit](Symbol("streamDocumentAsParquetZip"))
+ private val streamCellData = PrivateMethod[Unit](Symbol("streamCellData"))
+ private val convertFieldToBytes = PrivateMethod[Array[Byte]](Symbol("convertFieldToBytes"))
+
+ private def utf8(out: ByteArrayOutputStream): String =
+ new String(out.toByteArray, StandardCharsets.UTF_8)
+
+ // Assert on parsed lines rather than raw bytes so the CSV library's choice of
+ // line terminator does not make these tests brittle.
+ private def csvLines(out: ByteArrayOutputStream): List[String] = utf8(out).linesIterator.toList
+
+ // -- streamDocumentAsCSV ----------------------------------------------------
+
+ "streamDocumentAsCSV" should "write nothing when the document is empty" in {
+ val out = new ByteArrayOutputStream()
+ service invokePrivate streamDocumentAsCSV(new FakeDoc(Seq.empty), out, None)
+ out.size shouldBe 0
+ }
+
+ it should "write nothing when the iterator yields no rows despite a non-zero count" in {
+ val out = new ByteArrayOutputStream()
+ val doc = new FakeDoc(Seq.empty, countOverride = Some(2L))
+ service invokePrivate streamDocumentAsCSV(doc, out, None)
+ out.size shouldBe 0
+ }
+
+ it should "infer the header from the first row's schema and still write that row" in {
+ val out = new ByteArrayOutputStream()
+ val rows = Seq(tupleOf("a", 1), tupleOf("b", 2))
+ service invokePrivate streamDocumentAsCSV(new FakeDoc(rows), out, None)
+
+ csvLines(out) shouldBe List("name,count", "a,1", "b,2")
+ }
+
+ it should "use the supplied headers without consuming the first row" in {
+ val out = new ByteArrayOutputStream()
+ val rows = Seq(tupleOf("a", 1), tupleOf("b", 2))
+ val headers = Some(List("h1", "h2"))
+ service invokePrivate streamDocumentAsCSV(new FakeDoc(rows), out, headers)
+
+ // The supplied header replaces the inferred one, and — unlike the None
+ // branch above — the first row is not consumed by header inference, so it
+ // still appears in the body.
+ csvLines(out) shouldBe List("h1,h2", "a,1", "b,2")
+ }
+
+ it should "write every row when the document exceeds the chunk size" in {
+ val out = new ByteArrayOutputStream()
+ val rowCount = Constants.CHUNK_SIZE * 2 + 5
+ val rows = (1 to rowCount).map(i => tupleOf(s"r$i", i))
+ service invokePrivate streamDocumentAsCSV(new FakeDoc(rows), out, Some(List("h1", "h2")))
+
+ val lines = csvLines(out)
+ lines.head shouldBe "h1,h2"
+ lines.tail should have length rowCount
+ lines.tail.head shouldBe "r1,1"
+ lines.last shouldBe s"r$rowCount,$rowCount"
+ }
+
+ // -- streamCellData ---------------------------------------------------------
+
+ "streamCellData" should "reject a rowIndex beyond the document count" in {
+ val doc = new FakeDoc(Seq(tupleOf("a", 1), tupleOf("b", 2)))
+ val request = requestWith(List(OperatorExportInfo("op-1", "data"))).copy(rowIndex = 5)
+
+ val ex = intercept[WebApplicationException] {
+ service invokePrivate streamCellData(new ByteArrayOutputStream(), request, doc)
+ }
+ ex.getMessage shouldBe "Invalid rowIndex (5). Total rows: 2"
+ }
+
+ it should "reject a columnIndex beyond the selected row's field count" in {
+ val doc = new FakeDoc(Seq(tupleOf("a", 1), tupleOf("b", 2)))
+ val request = requestWith(List(OperatorExportInfo("op-1", "data"))).copy(columnIndex = 2)
+
+ val ex = intercept[WebApplicationException] {
+ service invokePrivate streamCellData(new ByteArrayOutputStream(), request, doc)
+ }
+ ex.getMessage shouldBe "Invalid columnIndex (2). Total columns: 2"
+ }
+
+ it should "fail when the row cannot be retrieved despite a non-zero count" in {
+ val doc = new FakeDoc(Seq.empty, countOverride = Some(2L))
+ val request = requestWith(List(OperatorExportInfo("op-1", "data")))
+
+ val ex = intercept[RuntimeException] {
+ service invokePrivate streamCellData(new ByteArrayOutputStream(), request, doc)
+ }
+ ex.getMessage shouldBe "Could not retrieve row at index 0"
+ }
+
+ it should "write the selected cell as UTF-8 bytes" in {
+ val out = new ByteArrayOutputStream()
+ val doc = new FakeDoc(Seq(tupleOf("café", 1), tupleOf("b", 2)))
+ val request = requestWith(List(OperatorExportInfo("op-1", "data")))
+
+ service invokePrivate streamCellData(out, request, doc)
+
+ out.toByteArray shouldBe "café".getBytes(StandardCharsets.UTF_8)
+ }
+
+ // -- convertFieldToBytes ----------------------------------------------------
+
+ "convertFieldToBytes" should "pass a byte array through unchanged" in {
+ val bytes = Array[Byte](1, 2, 3)
+ val result = service invokePrivate convertFieldToBytes(bytes)
+ result should be theSameInstanceAs bytes
+ }
+
+ it should "encode a string as UTF-8" in {
+ val result = service invokePrivate convertFieldToBytes("café")
+ result shouldBe "café".getBytes(StandardCharsets.UTF_8)
+ }
+
+ it should "fall back to toString for any other type" in {
+ val result = service invokePrivate convertFieldToBytes(42)
+ result shouldBe "42".getBytes(StandardCharsets.UTF_8)
+ }
+
+ // -- streamDocumentAsHTML ---------------------------------------------------
+
+ "streamDocumentAsHTML" should "write the first field of the first row as UTF-8" in {
+ val out = new ByteArrayOutputStream()
+ val doc = new FakeDoc(Seq(tupleOf("<p>café</p>", 1), tupleOf("<p>ignored</p>", 2)))
+
+ service invokePrivate streamDocumentAsHTML(out, doc)
+
+ out.toByteArray shouldBe "<p>café</p>".getBytes(StandardCharsets.UTF_8)
+ }
+
+ it should "throw on an empty document because the first row is unguarded" in {
+ val out = new ByteArrayOutputStream()
+ a[NoSuchElementException] should be thrownBy {
+ service invokePrivate streamDocumentAsHTML(out, new FakeDoc(Seq.empty))
+ }
+ }
+
+ // -- streamDocumentAsParquetZip ---------------------------------------------
+
+ "streamDocumentAsParquetZip" should "copy the document stream verbatim and close it" in {
+ val payload = "fake-parquet-zip-bytes".getBytes(StandardCharsets.UTF_8)
+ val source = new TrackingInputStream(payload)
+ val out = new ByteArrayOutputStream()
+
+ service invokePrivate streamDocumentAsParquetZip(
+ new FakeDoc(Seq.empty, stream = Some(source)),
+ out
+ )
+
+ out.toByteArray shouldBe payload
+ source.closed shouldBe true
+ }
+
+ it should "propagate a failure from the underlying document stream" in {
+ val failing = new VirtualDocument[Tuple] {
+ override def getURI: URI = new URI("file:///stub/export")
+ override def clear(): Unit = ()
+ override def asInputStream(): InputStream = throw new IllegalStateException("no stream")
+ }
+
+ val ex = intercept[IllegalStateException] {
+ service invokePrivate streamDocumentAsParquetZip(failing, new ByteArrayOutputStream())
+ }
+ ex.getMessage shouldBe "no stream"
+ }
+
+ // -- streamDocumentAsArrow --------------------------------------------------
+
+ "streamDocumentAsArrow" should "write nothing when the document is empty" in {
+ val out = new ByteArrayOutputStream()
+ service invokePrivate streamDocumentAsArrow(new FakeDoc(Seq.empty), out)
+ out.size shouldBe 0
+ }
+
+ it should "round-trip the tuples through an Arrow stream" in {
+ val out = new ByteArrayOutputStream()
+ val rows = Seq(tupleOf("café", 1), tupleOf("b", 2))
+ service invokePrivate streamDocumentAsArrow(new FakeDoc(rows), out)
+
+ val allocator = new RootAllocator()
+ val reader = new ArrowFileReader(
+ new ByteArrayReadableSeekableByteChannel(out.toByteArray),
+ allocator
+ )
+ try {
+ reader.loadNextBatch() shouldBe true
+ val root = reader.getVectorSchemaRoot
+ root.getRowCount shouldBe 2
+
+ val readBack = (0 until root.getRowCount).map(ArrowUtils.getTexeraTuple(_, root))
+ readBack.map(_.getField[String]("name")) shouldBe Seq("café", "b")
+ readBack.map(_.getField[Any]("count")) shouldBe Seq(1, 2)
+ } finally {
+ reader.close()
+ allocator.close()
+ }
+ }
+
+ // -- NonClosingOutputStream -------------------------------------------------
+
+ "NonClosingOutputStream" should "flush but not close the wrapped stream" in {
+ val wrapped = new TrackingOutputStream()
+ val stream = new NonClosingOutputStream(wrapped)
+
+ stream.write("payload".getBytes(StandardCharsets.UTF_8))
+ stream.close()
+
+ utf8(wrapped) shouldBe "payload"
+ wrapped.closed shouldBe false
+ }
}