Skip to content

Commit 34ade9a

Browse files
authored
test(workflow-operator): fill uncovered branches in source/scan and projection descriptors (#6100)
### What changes were proposed in this PR? Fill uncovered branches in four source/scan and projection descriptors, selected from the Codecov report. No production-code changes — extends the existing specs. | File | Codecov before | Missed branches now covered | | --- | --- | --- | | `ProjectionOpDesc.scala` | 69% | `derivePartition` — the hash (kept/empty→unknown), range, and pass-through arms | | `TextInputSourceOpDesc.scala` | 47% | `getPhysicalOp` wiring (exec class, ports) + schema propagation | | `FileScanSourceOpDesc.scala` | 37% | `getPhysicalOp` wiring + propagation, and the `outputFileName` filename-column branch of `sourceSchema` | | `FileScanOpDesc.scala` | 58% | `getPhysicalOp` wiring (input/output ports) + schema propagation | All targets exercise pure in-memory descriptor logic (`getPhysicalOp`/`derivePartition`/`sourceSchema`); the existing specs only drove the executors or `sourceSchema` defaults. ### 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 *ProjectionOpDescSpec *TextInputSourceOpDescSpec *FileScanSourceOpDescSpec *FileScanOpDescSpec"` — 37 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])
1 parent 6c86de5 commit 34ade9a

4 files changed

Lines changed: 117 additions & 1 deletion

File tree

common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/projection/ProjectionOpDescSpec.scala

Lines changed: 34 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,13 @@
2020
package org.apache.texera.amber.operator.projection
2121

2222
import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema}
23-
import org.apache.texera.amber.core.workflow.PortIdentity
23+
import org.apache.texera.amber.core.workflow.{
24+
HashPartition,
25+
PortIdentity,
26+
RangePartition,
27+
SinglePartition,
28+
UnknownPartition
29+
}
2430
import org.scalatest.BeforeAndAfter
2531
import org.scalatest.flatspec.AnyFlatSpec
2632
class ProjectionOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
@@ -110,4 +116,31 @@ class ProjectionOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
110116

111117
}
112118

119+
it should "preserve a HashPartition when its attributes are non-empty" in {
120+
val out = projectionOpDesc.derivePartition()(List(HashPartition(List("field1"))))
121+
assert(out == HashPartition(List("field1")))
122+
}
123+
124+
it should "downgrade an empty HashPartition to UnknownPartition" in {
125+
val out = projectionOpDesc.derivePartition()(List(HashPartition(List.empty)))
126+
assert(out == UnknownPartition())
127+
}
128+
129+
it should "preserve a RangePartition when its attributes are non-empty" in {
130+
val out = projectionOpDesc.derivePartition()(List(RangePartition(List("field2"), 0L, 100L)))
131+
assert(out == RangePartition(List("field2"), 0L, 100L))
132+
}
133+
134+
it should "downgrade an empty RangePartition to UnknownPartition" in {
135+
// RangePartition's companion apply already rewrites empty attributes to UnknownPartition,
136+
// so an empty range never reaches the range arm; either way the result is UnknownPartition.
137+
val out = projectionOpDesc.derivePartition()(List(RangePartition(List.empty, 0L, 100L)))
138+
assert(out == UnknownPartition())
139+
}
140+
141+
it should "pass through partitions that are neither hash nor range" in {
142+
val out = projectionOpDesc.derivePartition()(List(SinglePartition()))
143+
assert(out == SinglePartition())
144+
}
145+
113146
}

common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDescSpec.scala

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,8 @@ import org.apache.texera.amber.core.tuple.{
2626
SchemaEnforceable,
2727
Tuple
2828
}
29+
import org.apache.texera.amber.core.executor.OpExecWithClassName
30+
import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
2931
import org.apache.texera.amber.operator.TestOperators
3032
import org.apache.texera.amber.operator.source.scan.{FileAttributeType, FileDecodingMethod}
3133
import org.apache.texera.amber.util.JSONUtils.objectMapper
@@ -96,4 +98,24 @@ class FileScanOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
9698
assert(processedTuple.getField[String]("filename") == inputFilePath)
9799
fileScanOpExec.close()
98100
}
101+
102+
"FileScanOpDesc.getPhysicalOp" should
103+
"wire the FileScanOpExec class with one input port and one output port" in {
104+
val physical = fileScanOpDesc.getPhysicalOp(WorkflowIdentity(1L), ExecutionIdentity(1L))
105+
physical.opExecInitInfo match {
106+
case OpExecWithClassName(className, payload) =>
107+
assert(className == classOf[FileScanOpExec].getName)
108+
assert(payload.nonEmpty)
109+
case other => fail(s"expected OpExecWithClassName, got $other")
110+
}
111+
assert(physical.inputPorts.size == 1)
112+
assert(physical.outputPorts.size == 1)
113+
}
114+
115+
it should "propagate sourceSchema to its single output port" in {
116+
val physical = fileScanOpDesc.getPhysicalOp(WorkflowIdentity(1L), ExecutionIdentity(1L))
117+
val outPortId = fileScanOpDesc.operatorInfo.outputPorts.head.id
118+
val out = physical.propagateSchema.func(Map.empty)
119+
assert(out(outPortId) == fileScanOpDesc.sourceSchema())
120+
}
99121
}

common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpDescSpec.scala

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,8 +19,11 @@
1919

2020
package org.apache.texera.amber.operator.source.scan.file
2121

22+
import com.fasterxml.jackson.databind.node.ObjectNode
23+
import org.apache.texera.amber.core.executor.OpExecWithClassName
2224
import org.apache.texera.amber.core.storage.FileResolver
2325
import org.apache.texera.amber.core.tuple.{AttributeType, Schema, SchemaEnforceable, Tuple}
26+
import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
2427
import org.apache.texera.amber.operator.TestOperators
2528
import org.apache.texera.amber.operator.source.scan.{FileAttributeType, FileDecodingMethod}
2629
import org.apache.texera.amber.util.JSONUtils.objectMapper
@@ -185,4 +188,37 @@ class FileScanSourceOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
185188
FileScanSourceOpExec.close()
186189
}
187190

191+
"FileScanSourceOpDesc.getPhysicalOp" should
192+
"wire the FileScanSourceOpExec class as a source op and propagate its schema" in {
193+
val physical =
194+
fileScanSourceOpDesc.getPhysicalOp(WorkflowIdentity(1L), ExecutionIdentity(1L))
195+
physical.opExecInitInfo match {
196+
case OpExecWithClassName(className, descString) =>
197+
assert(className == classOf[FileScanSourceOpExec].getName)
198+
assert(descString.nonEmpty)
199+
case other => fail(s"expected OpExecWithClassName, got $other")
200+
}
201+
assert(physical.inputPorts.isEmpty)
202+
val outPortId = fileScanSourceOpDesc.operatorInfo.outputPorts.head.id
203+
assert(physical.outputPorts.keySet == Set(outPortId))
204+
val propagated = physical.propagateSchema.func(Map.empty)
205+
assert(propagated(outPortId) == fileScanSourceOpDesc.sourceSchema())
206+
}
207+
208+
"FileScanSourceOpDesc.sourceSchema" should
209+
"prepend a filename column when outputFileName is enabled" in {
210+
// outputFileName is a val; round-trip through JSON (which carries the operatorType
211+
// discriminator) with the flag flipped on, since it can't be set on an instance.
212+
val node =
213+
objectMapper
214+
.readTree(objectMapper.writeValueAsString(fileScanSourceOpDesc))
215+
.asInstanceOf[ObjectNode]
216+
node.put("outputFileName", true)
217+
val withFlag = objectMapper.treeToValue(node, classOf[FileScanSourceOpDesc])
218+
val schema: Schema = withFlag.sourceSchema()
219+
assert(schema.getAttributes.length == 2)
220+
assert(schema.getAttribute("filename").getType == AttributeType.STRING)
221+
assert(schema.getAttribute("line").getType == AttributeType.STRING)
222+
}
223+
188224
}

common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/text/TextInputSourceOpDescSpec.scala

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,9 @@
1919

2020
package org.apache.texera.amber.operator.source.scan.text
2121

22+
import org.apache.texera.amber.core.executor.OpExecWithClassName
2223
import org.apache.texera.amber.core.tuple.{AttributeType, Schema, SchemaEnforceable, Tuple}
24+
import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
2325
import org.apache.texera.amber.operator.TestOperators
2426
import org.apache.texera.amber.operator.source.scan.FileAttributeType
2527
import org.apache.texera.amber.util.JSONUtils.objectMapper
@@ -180,4 +182,27 @@ class TextInputSourceOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
180182
val path: Path = Paths.get(filePath)
181183
new String(Files.readAllBytes(path), StandardCharsets.UTF_8)
182184
}
185+
186+
"TextInputSourceOpDesc.getPhysicalOp" should
187+
"wire the TextInputSourceOpExec class as a source op with one output port" in {
188+
val physical =
189+
textInputSourceOpDesc.getPhysicalOp(WorkflowIdentity(1L), ExecutionIdentity(1L))
190+
physical.opExecInitInfo match {
191+
case OpExecWithClassName(className, _) =>
192+
assert(className == classOf[TextInputSourceOpExec].getName)
193+
case other => fail(s"expected OpExecWithClassName, got $other")
194+
}
195+
assert(physical.inputPorts.isEmpty)
196+
assert(physical.outputPorts.size == 1)
197+
}
198+
199+
it should "propagate sourceSchema to its single output port" in {
200+
textInputSourceOpDesc.attributeType = FileAttributeType.STRING
201+
val physical =
202+
textInputSourceOpDesc.getPhysicalOp(WorkflowIdentity(1L), ExecutionIdentity(1L))
203+
val outPortId = textInputSourceOpDesc.operatorInfo.outputPorts.head.id
204+
val out = physical.propagateSchema.func(Map.empty)
205+
assert(out.keySet == Set(outPortId))
206+
assert(out(outPortId) == textInputSourceOpDesc.sourceSchema())
207+
}
183208
}

0 commit comments

Comments
 (0)