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/03 09:40:39 UTC
(texera) branch main updated (d861036e74 -> 48ab6ca5d6)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/texera.git
from d861036e74 chore(deps): bump transformers from 5.0.0rc3 to 5.3.0 in /amber (#6067)
new ad6ea02dce test(workflow-operator): add unit test coverage for operator support utilities (#6077)
new 48ab6ca5d6 test(workflow-operator): add unit test coverage for scan source executors (#6076)
The 2 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.
Summary of changes:
.../operator/SpecialPhysicalOpFactorySpec.scala | 100 ++++++++++++++++++
.../texera/amber/operator/TestOperatorsSpec.scala | 111 ++++++++++++++++++++
.../source/scan/arrow/ArrowSourceOpExecSpec.scala | 115 +++++++++++++++++++++
.../scan/csv/ParallelCSVScanSourceOpExecSpec.scala | 106 +++++++++++++++++++
.../scan/csvOld/CSVOldScanSourceOpExecSpec.scala | 85 +++++++++++++++
.../source/scan/file/FileScanUtilsSpec.scala | 108 +++++++++++++++++++
.../scan/json/JSONLScanSourceOpExecSpec.scala | 84 +++++++++++++++
7 files changed, 709 insertions(+)
create mode 100644 common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/SpecialPhysicalOpFactorySpec.scala
create mode 100644 common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/TestOperatorsSpec.scala
create mode 100644 common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/arrow/ArrowSourceOpExecSpec.scala
create mode 100644 common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/csv/ParallelCSVScanSourceOpExecSpec.scala
create mode 100644 common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/csvOld/CSVOldScanSourceOpExecSpec.scala
create mode 100644 common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanUtilsSpec.scala
create mode 100644 common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpExecSpec.scala
(texera) 02/02: test(workflow-operator): add unit test coverage for scan source executors (#6076)
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 main
in repository https://gitbox.apache.org/repos/asf/texera.git
commit 48ab6ca5d627e6d8c9c23e5edfa6556a3681fb07
Author: Xinyuan Lin <xi...@uci.edu>
AuthorDate: Fri Jul 3 02:24:10 2026 -0700
test(workflow-operator): add unit test coverage for scan source executors (#6076)
### What changes were proposed in this PR?
Add unit test coverage for the four file-reading scan source
**executors**, selected from the Codecov report. No production-code
changes. Every test is driven by a local temp file (`file:` URI →
`ReadonlyLocalFileDocument`); no live DB/network/S3/Iceberg.
| File | Codecov before | Missed lines targeted |
| --- | --- | --- |
| `ParallelCSVScanSourceOpExec.scala` | 0% | 43 — construct + schema
init, `open` byte-range partitioning (both `endOffset` branches +
mid-file partial-line skip + header skip), `produceTuple` happy path,
short-row null padding, all-null (blank) line discard, `close` |
| `JSONLScanSourceOpExec.scala` | 0% | 23 — construct, `produceTuple`,
`open` line counting + worker partitioning (both ternary branches), row
limit, `close` |
| `CSVOldScanSourceOpExec.scala` | 0% | 21 — construct + header-based
schema, `open` (header vs no-header start offset), `produceTuple` +
limit, `close` null-guard before open |
| `ArrowSourceOpExec.scala` | 0% | 30 — construct, `open` success +
error-wrapping path, `produceTuple` batch iteration + offset/limit,
`close` no-op before open |
New specs live in each executor's own package so their `private[...]`
constructors are reachable, mirroring the existing
`FileScanSourceOpExecSpec` temp-file/URI pattern.
### Any related issues, documentation, discussions?
Follow-up to the review feedback on #6043: prioritize tests that fill
uncovered code paths.
### How was this PR tested?
- `sbt "WorkflowOperator/testOnly *ParallelCSVScanSourceOpExecSpec
*JSONLScanSourceOpExecSpec *CSVOldScanSourceOpExecSpec
*ArrowSourceOpExecSpec"` — 17 tests, all green
- `sbt "WorkflowOperator/Test/scalafmtCheck"` and `sbt
"WorkflowOperator/scalafixAll --check"` — clean
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8 [1M context])
---
.../source/scan/arrow/ArrowSourceOpExecSpec.scala | 115 +++++++++++++++++++++
.../scan/csv/ParallelCSVScanSourceOpExecSpec.scala | 106 +++++++++++++++++++
.../scan/csvOld/CSVOldScanSourceOpExecSpec.scala | 85 +++++++++++++++
.../scan/json/JSONLScanSourceOpExecSpec.scala | 84 +++++++++++++++
4 files changed, 390 insertions(+)
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/arrow/ArrowSourceOpExecSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/arrow/ArrowSourceOpExecSpec.scala
new file mode 100644
index 0000000000..7e7d98dca2
--- /dev/null
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/arrow/ArrowSourceOpExecSpec.scala
@@ -0,0 +1,115 @@
+/*
+ * 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.arrow
+
+import org.apache.arrow.memory.RootAllocator
+import org.apache.arrow.vector.VectorSchemaRoot
+import org.apache.arrow.vector.ipc.ArrowFileWriter
+import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple}
+import org.apache.texera.amber.util.ArrowUtils
+import org.apache.texera.amber.util.JSONUtils.objectMapper
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import java.io.{File, FileOutputStream}
+import java.nio.channels.Channels
+import java.nio.file.Files
+
+class ArrowSourceOpExecSpec extends AnyFlatSpec with Matchers {
+
+ private val schema = Schema(List(new Attribute("s", AttributeType.STRING)))
+
+ private def writeArrowFile(rows: Seq[String]): File = {
+ val file = File.createTempFile("arrow-src-", ".arrow")
+ file.deleteOnExit()
+ val allocator = new RootAllocator()
+ val root = VectorSchemaRoot.create(ArrowUtils.fromTexeraSchema(schema), allocator)
+ val out = new FileOutputStream(file)
+ val writer = new ArrowFileWriter(root, null, Channels.newChannel(out))
+ try {
+ writer.start()
+ root.allocateNew()
+ rows.zipWithIndex.foreach {
+ case (value, i) =>
+ ArrowUtils.setTexeraTuple(
+ Tuple.builder(schema).addSequentially(Array[Any](value)).build(),
+ i,
+ root
+ )
+ }
+ root.setRowCount(rows.size)
+ writer.writeBatch()
+ writer.end()
+ } finally {
+ writer.close()
+ root.close()
+ allocator.close()
+ out.close()
+ }
+ file
+ }
+
+ private def descString(
+ file: File,
+ offset: Option[Int] = None,
+ limit: Option[Int] = None
+ ): String = {
+ val desc = new ArrowSourceOpDesc()
+ desc.fileName = Some(file.toURI.toString)
+ desc.offset = offset
+ desc.limit = limit
+ objectMapper.writeValueAsString(desc)
+ }
+
+ private def values(exec: ArrowSourceOpExec): List[String] =
+ exec.produceTuple().toList.map(_.asInstanceOf[Tuple].getField[String]("s"))
+
+ "ArrowSourceOpExec" should "read every row of a single-batch Arrow file" in {
+ val exec = new ArrowSourceOpExec(descString(writeArrowFile(Seq("a", "b", "c"))))
+ exec.open()
+ try assert(values(exec) == List("a", "b", "c"))
+ finally exec.close()
+ }
+
+ it should "apply offset and limit" in {
+ val exec =
+ new ArrowSourceOpExec(
+ descString(writeArrowFile(Seq("a", "b", "c", "d", "e")), Some(1), Some(2))
+ )
+ exec.open()
+ try assert(values(exec) == List("b", "c"))
+ finally exec.close()
+ }
+
+ it should "wrap failures to open an invalid Arrow file" in {
+ val bogus = File.createTempFile("not-arrow-", ".arrow")
+ bogus.deleteOnExit()
+ Files.write(bogus.toPath, "this is not arrow".getBytes)
+ val exec = new ArrowSourceOpExec(descString(bogus))
+ val ex = intercept[RuntimeException](exec.open())
+ ex.getMessage shouldBe "Failed to open Arrow source"
+ ex.getCause should not be null
+ }
+
+ it should "treat close before open as a safe no-op" in {
+ val exec = new ArrowSourceOpExec(descString(writeArrowFile(Seq("a"))))
+ noException should be thrownBy exec.close()
+ }
+}
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/csv/ParallelCSVScanSourceOpExecSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/csv/ParallelCSVScanSourceOpExecSpec.scala
new file mode 100644
index 0000000000..51f9424454
--- /dev/null
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/csv/ParallelCSVScanSourceOpExecSpec.scala
@@ -0,0 +1,106 @@
+/*
+ * 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.csv
+
+import org.apache.texera.amber.util.JSONUtils.objectMapper
+import org.scalatest.flatspec.AnyFlatSpec
+
+import java.nio.charset.StandardCharsets
+import java.nio.file.Files
+
+class ParallelCSVScanSourceOpExecSpec extends AnyFlatSpec {
+
+ private def writeCsv(content: String): String = {
+ val file = Files.createTempFile("parallel-csv-", ".csv")
+ file.toFile.deleteOnExit()
+ Files.write(file, content.getBytes(StandardCharsets.UTF_8))
+ file.toFile.toURI.toString
+ }
+
+ private def descString(uri: String, delimiter: String = ",", header: Boolean = true): String = {
+ val desc = new ParallelCSVScanSourceOpDesc()
+ desc.fileName = Some(uri)
+ desc.customDelimiter = Some(delimiter)
+ desc.hasHeader = header
+ desc.setResolvedFileName(new java.net.URI(uri))
+ objectMapper.writeValueAsString(desc)
+ }
+
+ private def drain(exec: ParallelCSVScanSourceOpExec): List[Seq[Any]] = {
+ exec.open()
+ try exec.produceTuple().map(_.getFields.toSeq).toList
+ finally exec.close()
+ }
+
+ "ParallelCSVScanSourceOpExec" should "read every data row of a headered file" in {
+ val exec =
+ new ParallelCSVScanSourceOpExec(descString(writeCsv("id,name,age\n1,Alice,30\n2,Bob,25\n")))
+ val rows = drain(exec)
+ assert(rows.size == 2)
+ assert(rows.head == Seq(1, "Alice", 30))
+ assert(rows(1) == Seq(2, "Bob", 25))
+ }
+
+ it should "not skip the first row when the file has no header" in {
+ val exec =
+ new ParallelCSVScanSourceOpExec(
+ descString(writeCsv("1,Alice,30\n2,Bob,25\n"), header = false)
+ )
+ assert(drain(exec).size == 2)
+ }
+
+ it should "pad short rows with trailing nulls to the schema width" in {
+ // the wide first data row fixes the schema at 3 columns; the short row is padded
+ val exec = new ParallelCSVScanSourceOpExec(descString(writeCsv("a,b,c\n1,2,3\n4,5\n")))
+ val rows = drain(exec)
+ assert(rows.size == 2)
+ assert(rows(1).length == 3)
+ assert(rows(1)(2) == null)
+ }
+
+ it should "discard all-null (blank) lines" in {
+ val exec =
+ new ParallelCSVScanSourceOpExec(descString(writeCsv("name,city\nAlice,NYC\n\nBob,LA\n")))
+ val rows = drain(exec)
+ assert(rows.size == 2)
+ assert(rows.map(_.head) == Seq("Alice", "Bob"))
+ }
+
+ it should "partition a headerless file across workers by byte range" in {
+ // Two workers over the same file: worker 0 (idx != workerCount-1) takes the
+ // first byte-range chunk; worker 1 seeks past its chunk start and drops the
+ // leading partial line. Rows are deliberately uneven widths so the split
+ // point (totalBytes/2 = 12) lands *inside* the second row rather than on a
+ // newline, exercising the partial-line skip. Exact boundary rows depend on
+ // the block reader reading one line past the boundary, so assert only that
+ // the union covers all rows without duplication.
+ val uri = writeCsv("1,aaaa\n2,bbb\n3,ccccc\n4,d\n")
+ val rows0 = drain(
+ new ParallelCSVScanSourceOpExec(descString(uri, header = false), idx = 0, workerCount = 2)
+ )
+ val rows1 = drain(
+ new ParallelCSVScanSourceOpExec(descString(uri, header = false), idx = 1, workerCount = 2)
+ )
+ val keys = (rows0 ++ rows1).map(_.head)
+ assert(keys.toSet == Set(1, 2, 3, 4))
+ assert(keys.size == 4) // each row read exactly once
+ assert(rows0.nonEmpty && rows1.nonEmpty)
+ }
+}
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/csvOld/CSVOldScanSourceOpExecSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/csvOld/CSVOldScanSourceOpExecSpec.scala
new file mode 100644
index 0000000000..f4a775fcba
--- /dev/null
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/csvOld/CSVOldScanSourceOpExecSpec.scala
@@ -0,0 +1,85 @@
+/*
+ * 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.csvOld
+
+import org.apache.texera.amber.operator.source.scan.FileDecodingMethod
+import org.apache.texera.amber.util.JSONUtils.objectMapper
+import org.scalatest.flatspec.AnyFlatSpec
+
+import java.nio.charset.StandardCharsets
+import java.nio.file.Files
+
+class CSVOldScanSourceOpExecSpec extends AnyFlatSpec {
+
+ private def descString(
+ content: String,
+ header: Boolean = true,
+ delimiter: String = ",",
+ limit: Option[Int] = None,
+ offset: Option[Int] = None
+ ): String = {
+ val file = Files.createTempFile("csvold-", ".csv")
+ file.toFile.deleteOnExit()
+ Files.write(file, content.getBytes(StandardCharsets.UTF_8))
+ val desc = new CSVOldScanSourceOpDesc
+ desc.setResolvedFileName(file.toFile.toURI)
+ desc.customDelimiter = Some(delimiter)
+ desc.hasHeader = header
+ desc.fileEncoding = FileDecodingMethod.UTF_8
+ desc.limit = limit
+ desc.offset = offset
+ objectMapper.writeValueAsString(desc)
+ }
+
+ private def drain(exec: CSVOldScanSourceOpExec): List[Seq[Any]] = {
+ exec.open()
+ try exec.produceTuple().map(_.getFields.toSeq).toList
+ finally exec.close()
+ }
+
+ "CSVOldScanSourceOpExec" should "infer the schema from the header at construction" in {
+ val exec = new CSVOldScanSourceOpExec(descString("a,b\n1,2\n3,4\n", header = true))
+ assert(exec.desc.hasHeader)
+ assert(exec.schema.getAttributeNames == List("a", "b"))
+ }
+
+ it should "read every data row, skipping the header" in {
+ val exec = new CSVOldScanSourceOpExec(descString("a,b\n1,2\n3,4\n", header = true))
+ val rows = drain(exec)
+ assert(rows.size == 2)
+ assert(rows.head == Seq(1, 2))
+ assert(rows(1) == Seq(3, 4))
+ }
+
+ it should "not skip the first row when the file has no header" in {
+ val exec = new CSVOldScanSourceOpExec(descString("1,2\n3,4\n", header = false))
+ assert(drain(exec).size == 2)
+ }
+
+ it should "honor the row limit" in {
+ val exec = new CSVOldScanSourceOpExec(descString("a,b\n1,2\n3,4\n5,6\n", limit = Some(1)))
+ assert(drain(exec).size == 1)
+ }
+
+ it should "treat close before open as a no-op" in {
+ val exec = new CSVOldScanSourceOpExec(descString("a\n1\n"))
+ exec.close() // reader is still null -> guarded, no exception
+ }
+}
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpExecSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpExecSpec.scala
new file mode 100644
index 0000000000..2338e48f7c
--- /dev/null
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpExecSpec.scala
@@ -0,0 +1,84 @@
+/*
+ * 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.json
+
+import org.apache.texera.amber.operator.source.scan.FileDecodingMethod
+import org.apache.texera.amber.util.JSONUtils.objectMapper
+import org.scalatest.flatspec.AnyFlatSpec
+
+import java.net.URI
+import java.nio.charset.StandardCharsets
+import java.nio.file.Files
+
+class JSONLScanSourceOpExecSpec extends AnyFlatSpec {
+
+ private def writeJsonl(lines: String*): URI = {
+ val path = Files.createTempFile("jsonl-scan-", ".jsonl")
+ path.toFile.deleteOnExit()
+ Files.write(path, lines.mkString("\n").getBytes(StandardCharsets.UTF_8))
+ path.toFile.toURI
+ }
+
+ private def descString(
+ uri: URI,
+ flatten: Boolean = false,
+ limit: Option[Int] = None,
+ offset: Option[Int] = None
+ ): String = {
+ val desc = new JSONLScanSourceOpDesc
+ desc.setResolvedFileName(uri)
+ desc.fileEncoding = FileDecodingMethod.UTF_8
+ desc.flatten = flatten
+ desc.limit = limit
+ desc.offset = offset
+ objectMapper.writeValueAsString(desc)
+ }
+
+ private def drain(exec: JSONLScanSourceOpExec): List[Seq[Any]] = {
+ exec.open()
+ try exec.produceTuple().map(_.getFields.toSeq).toList
+ finally exec.close()
+ }
+
+ "JSONLScanSourceOpExec" should "read each JSON line, ordering fields by sorted attribute name" in {
+ // keys are written name-then-id; the output must be reordered to id-then-name
+ val exec = new JSONLScanSourceOpExec(
+ descString(writeJsonl("""{"name":"a","id":1}""", """{"name":"b","id":2}"""))
+ )
+ val rows = drain(exec)
+ assert(rows.size == 2)
+ assert(rows.head == Seq(1, "a"))
+ assert(rows(1) == Seq(2, "b"))
+ }
+
+ it should "partition rows across workers" in {
+ val uri = writeJsonl("""{"v":0}""", """{"v":1}""", """{"v":2}""", """{"v":3}""")
+ val worker0 = new JSONLScanSourceOpExec(descString(uri), idx = 0, workerCount = 2)
+ val worker1 = new JSONLScanSourceOpExec(descString(uri), idx = 1, workerCount = 2)
+ assert(drain(worker0).map(_.head) == Seq(0, 1))
+ assert(drain(worker1).map(_.head) == Seq(2, 3))
+ }
+
+ it should "apply the row limit" in {
+ val uri = writeJsonl("""{"v":0}""", """{"v":1}""", """{"v":2}""", """{"v":3}""", """{"v":4}""")
+ val exec = new JSONLScanSourceOpExec(descString(uri, limit = Some(2)))
+ assert(drain(exec).map(_.head) == Seq(0, 1))
+ }
+}
(texera) 01/02: test(workflow-operator): add unit test coverage for operator support utilities (#6077)
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 main
in repository https://gitbox.apache.org/repos/asf/texera.git
commit ad6ea02dce56f29be8e61e0d0203f2a8f59374ea
Author: Xinyuan Lin <xi...@uci.edu>
AuthorDate: Fri Jul 3 02:24:06 2026 -0700
test(workflow-operator): add unit test coverage for operator support utilities (#6077)
### What changes were proposed in this PR?
Add unit test coverage for three `workflow-operator` support utilities,
selected from the Codecov report. No production-code changes.
| File | Codecov before | Missed lines targeted |
| --- | --- | --- |
| `TestOperators.scala` | 26.3% | 56 — every desc-factory helper
(CSV/JSONL scans, hash-join, keyword-search, aggregate + group-by,
AsterixDB source, python & python-source UDFs); asserts the returned
descriptors' fields |
| `SpecialPhysicalOpFactory.scala` | 0% | 17 — `newSourcePhysicalOp`:
decoded-port identity derivation, layer-name interpolation (incl.
dash→underscore), empty input ports, output port + schema propagation,
and the no-port-id failure |
| `FileScanUtils.scala` | 62.9% | 11 — the `extract == true` archive
branch of `createTuplesFromFile`: zip-entry iteration, `__MACOSX`
filtering, single-string vs per-line attribute paths |
`TestOperators` is itself a shared test-helper factory (in `src/main`)
used by many specs, but nothing asserted on its output, so its bodies
read as uncovered. All tests run in-memory or against local temp files /
committed test resources.
### Any related issues, documentation, discussions?
Follow-up to the review feedback on #6043: prioritize tests that fill
uncovered code paths.
### How was this PR tested?
- `sbt "WorkflowOperator/testOnly *TestOperatorsSpec
*SpecialPhysicalOpFactorySpec *FileScanUtilsSpec"` — 16 tests, all green
- `sbt "WorkflowOperator/Test/scalafmtCheck"` and `sbt
"WorkflowOperator/scalafixAll --check"` — clean
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8 [1M context])
---
.../operator/SpecialPhysicalOpFactorySpec.scala | 100 +++++++++++++++++++
.../texera/amber/operator/TestOperatorsSpec.scala | 111 +++++++++++++++++++++
.../source/scan/file/FileScanUtilsSpec.scala | 108 ++++++++++++++++++++
3 files changed, 319 insertions(+)
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/SpecialPhysicalOpFactorySpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/SpecialPhysicalOpFactorySpec.scala
new file mode 100644
index 0000000000..d676dfc30b
--- /dev/null
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/SpecialPhysicalOpFactorySpec.scala
@@ -0,0 +1,100 @@
+/*
+ * 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
+
+import org.apache.texera.amber.core.storage.VFSURIFactory
+import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema}
+import org.apache.texera.amber.core.virtualidentity.{
+ ExecutionIdentity,
+ OperatorIdentity,
+ PhysicalOpIdentity,
+ WorkflowIdentity
+}
+import org.apache.texera.amber.core.workflow.{GlobalPortIdentity, PortIdentity, PreferController}
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+class SpecialPhysicalOpFactorySpec extends AnyFlatSpec with Matchers {
+
+ private val workflowId = WorkflowIdentity(42L)
+ private val executionId = ExecutionIdentity(7L)
+ private val schema = Schema().add(new Attribute("col", AttributeType.STRING))
+
+ private def portUri(opName: String, layer: String, portId: Int) = {
+ val gpid = GlobalPortIdentity(
+ PhysicalOpIdentity(OperatorIdentity(opName), layer),
+ PortIdentity(portId),
+ input = false
+ )
+ VFSURIFactory.resultURI(VFSURIFactory.createPortBaseURI(workflowId, executionId, gpid))
+ }
+
+ "SpecialPhysicalOpFactory.newSourcePhysicalOp" should
+ "derive a source op that wires the decoded port identity, ports, and schema" in {
+ val uri = portUri("srcOp", "layerA", 3)
+ val downstream = PhysicalOpIdentity(OperatorIdentity("down-stream"), "main")
+ val op = SpecialPhysicalOpFactory.newSourcePhysicalOp(
+ workflowId,
+ executionId,
+ uri,
+ downstream,
+ PortIdentity(5),
+ schema
+ )
+ op.id.logicalOpId shouldBe OperatorIdentity("srcOp")
+ // layerName: "${layerName}_source_${portId.id}_${downstreamLogicalId}_${downstreamPort.id}"
+ // with '-' in the downstream logical id replaced by '_'
+ op.id.layerName shouldBe "layerA_source_3_down_stream_5"
+ op.workflowId shouldBe workflowId
+ op.executionId shouldBe executionId
+ op.isSourceOperator shouldBe true
+ op.inputPorts shouldBe empty
+ op.outputPorts.keySet shouldBe Set(PortIdentity(0))
+ op.locationPreference shouldBe Some(PreferController)
+ op.parallelizable shouldBe false
+ op.outputPorts(PortIdentity(0))._3 shouldBe Right(schema)
+ }
+
+ it should "replace every dash in the downstream operator id when building the layer name" in {
+ val op = SpecialPhysicalOpFactory.newSourcePhysicalOp(
+ workflowId,
+ executionId,
+ portUri("srcOp", "layerA", 3),
+ PhysicalOpIdentity(OperatorIdentity("a-b-c"), "main"),
+ PortIdentity(9),
+ schema
+ )
+ op.id.layerName shouldBe "layerA_source_3_a_b_c_9"
+ }
+
+ it should "fail when the URI carries no global port identity" in {
+ val noPortUri = VFSURIFactory.createRuntimeStatisticsURI(workflowId, executionId)
+ intercept[NoSuchElementException] {
+ SpecialPhysicalOpFactory.newSourcePhysicalOp(
+ workflowId,
+ executionId,
+ noPortUri,
+ PhysicalOpIdentity(OperatorIdentity("down"), "main"),
+ PortIdentity(0),
+ schema
+ )
+ }
+ }
+}
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/TestOperatorsSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/TestOperatorsSpec.scala
new file mode 100644
index 0000000000..7a92394f70
--- /dev/null
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/TestOperatorsSpec.scala
@@ -0,0 +1,111 @@
+/*
+ * 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
+
+import org.apache.texera.amber.core.tuple.AttributeType
+import org.apache.texera.amber.operator.aggregate.AggregationFunction
+import org.scalatest.flatspec.AnyFlatSpec
+
+class TestOperatorsSpec extends AnyFlatSpec {
+
+ "TestOperators.getCsvScanOpDesc" should "build a resolved CSV desc honoring the header flag" in {
+ val op = TestOperators.getCsvScanOpDesc(TestOperators.CountrySalesSmallCsvPath, header = true)
+ assert(op.hasHeader)
+ assert(op.customDelimiter.contains(","))
+ assert(op.fileName.isDefined)
+ assert(op.fileResolved())
+ val headerless =
+ TestOperators.getCsvScanOpDesc(
+ TestOperators.CountrySalesHeaderlessSmallCsvPath,
+ header = false,
+ multiLine = true
+ )
+ assert(!headerless.hasHeader)
+ }
+
+ "TestOperators CSV factories" should "resolve their backing files" in {
+ assert(!TestOperators.headerlessSmallCsvScanOpDesc().hasHeader)
+ assert(!TestOperators.headerlessSmallMultiLineDataCsvScanOpDesc().hasHeader)
+ assert(TestOperators.smallCsvScanOpDesc().hasHeader)
+ assert(TestOperators.mediumCsvScanOpDesc().hasHeader)
+ assert(TestOperators.smallCsvScanOpDesc().fileResolved())
+ }
+
+ "TestOperators.getJSONLScanOpDesc" should "build a resolved JSONL desc honoring the flatten flag" in {
+ val op = TestOperators.getJSONLScanOpDesc(TestOperators.smallJsonLPath)
+ assert(!op.flatten)
+ assert(op.fileResolved())
+ val flattened = TestOperators.getJSONLScanOpDesc(TestOperators.mediumJsonLPath, flatten = true)
+ assert(flattened.flatten)
+ }
+
+ "TestOperators JSONL factories" should "resolve their backing files" in {
+ assert(!TestOperators.smallJSONLScanOpDesc().flatten)
+ assert(TestOperators.mediumFlattenJSONLScanOpDesc().flatten)
+ assert(TestOperators.smallJSONLScanOpDesc().fileResolved())
+ }
+
+ "TestOperators.joinOpDesc" should "set the build and probe attribute names" in {
+ val op = TestOperators.joinOpDesc("build_col", "probe_col")
+ assert(op.buildAttributeName == "build_col")
+ assert(op.probeAttributeName == "probe_col")
+ }
+
+ "TestOperators.keywordSearchOpDesc" should "set the attribute and keyword" in {
+ val op = TestOperators.keywordSearchOpDesc("text", "hello")
+ assert(op.attribute == "text")
+ assert(op.keyword == "hello")
+ }
+
+ "TestOperators.aggregateAndGroupByDesc" should "wire a single aggregation with group-by keys" in {
+ val op = TestOperators.aggregateAndGroupByDesc("price", AggregationFunction.SUM, List("region"))
+ assert(op.aggregations.length == 1)
+ assert(op.aggregations.head.aggFunction == AggregationFunction.SUM)
+ assert(op.aggregations.head.attribute == "price")
+ assert(op.aggregations.head.resultAttribute == "aggregate-result")
+ assert(op.groupByKeys == List("region"))
+ }
+
+ "TestOperators.asterixDBSourceOpDesc" should "build the AsterixDB desc without contacting a server" in {
+ val op = TestOperators.asterixDBSourceOpDesc()
+ assert(op.host == "ipubmed4.ics.uci.edu")
+ assert(op.port == "default")
+ assert(op.database == "twitter")
+ assert(op.table == "ds_tweet")
+ assert(op.limit.contains(1000L))
+ }
+
+ "TestOperators.pythonOpDesc" should "build a python UDF desc" in {
+ val op = TestOperators.pythonOpDesc()
+ assert(op.workers == 1)
+ assert(op.retainInputColumns)
+ assert(op.code.contains("ProcessTupleOperator"))
+ assert(op.code.contains("process_tuple"))
+ }
+
+ "TestOperators.pythonSourceOpDesc" should "build a python source UDF desc interpolating the tuple count" in {
+ val op = TestOperators.pythonSourceOpDesc(5)
+ assert(op.workers == 1)
+ assert(op.columns.length == 1)
+ assert(op.columns.head.getName == "field_1")
+ assert(op.columns.head.getType == AttributeType.STRING)
+ assert(op.code.contains("range(5)"))
+ }
+}
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..ad2ffeb79e
--- /dev/null
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanUtilsSpec.scala
@@ -0,0 +1,108 @@
+/*
+ * 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 zips = scala.collection.mutable.ArrayBuffer.empty[Path]
+
+ private def makeZip(entries: (String, String)*): String = {
+ val path = Files.createTempFile("filescanutils-", ".zip")
+ zips += 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
+ }
+
+ override def afterAll(): Unit = {
+ zips.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"))
+ }
+}