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:24:31 UTC
(texera) 01/01: test(workflow-operator): add unit test coverage for scan source executors (#6076)
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-6076-ad6ea02dce56f29be8e61e0d0203f2a8f59374ea
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))
+ }
+}