test(workflow-operator): fill uncovered branches in source/scan and projection descriptors

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

@@ -20,7 +20,13 @@
 package org.apache.texera.amber.operator.projection
 
 import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema}
-import org.apache.texera.amber.core.workflow.PortIdentity
+import org.apache.texera.amber.core.workflow.{
+  HashPartition,
+  PortIdentity,
+  RangePartition,
+  SinglePartition,
+  UnknownPartition
+}
 import org.scalatest.BeforeAndAfter
 import org.scalatest.flatspec.AnyFlatSpec
 class ProjectionOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
@@ -110,4 +116,31 @@ class ProjectionOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
 
   }
 
+  it should "preserve a HashPartition when its attributes are non-empty" in {
+    val out = projectionOpDesc.derivePartition()(List(HashPartition(List("field1"))))
+    assert(out == HashPartition(List("field1")))
+  }
+
+  it should "downgrade an empty HashPartition to UnknownPartition" in {
+    val out = projectionOpDesc.derivePartition()(List(HashPartition(List.empty)))
+    assert(out == UnknownPartition())
+  }
+
+  it should "preserve a RangePartition when its attributes are non-empty" in {
+    val out = projectionOpDesc.derivePartition()(List(RangePartition(List("field2"), 0L, 100L)))
+    assert(out == RangePartition(List("field2"), 0L, 100L))
+  }
+
+  it should "downgrade an empty RangePartition to UnknownPartition" in {
+    // RangePartition's companion apply already rewrites empty attributes to UnknownPartition,
+    // so an empty range never reaches the range arm; either way the result is UnknownPartition.
+    val out = projectionOpDesc.derivePartition()(List(RangePartition(List.empty, 0L, 100L)))
+    assert(out == UnknownPartition())
+  }
+
+  it should "pass through partitions that are neither hash nor range" in {
+    val out = projectionOpDesc.derivePartition()(List(SinglePartition()))
+    assert(out == SinglePartition())
+  }
+
 }

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

@@ -26,6 +26,8 @@ import org.apache.texera.amber.core.tuple.{
   SchemaEnforceable,
   Tuple
 }
+import org.apache.texera.amber.core.executor.OpExecWithClassName
+import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import org.apache.texera.amber.operator.TestOperators
 import org.apache.texera.amber.operator.source.scan.{FileAttributeType, FileDecodingMethod}
 import org.apache.texera.amber.util.JSONUtils.objectMapper
@@ -96,4 +98,24 @@ class FileScanOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
     assert(processedTuple.getField[String]("filename") == inputFilePath)
     fileScanOpExec.close()
   }
+
+  "FileScanOpDesc.getPhysicalOp" should
+    "wire the FileScanOpExec class with one input port and one output port" in {
+    val physical = fileScanOpDesc.getPhysicalOp(WorkflowIdentity(1L), ExecutionIdentity(1L))
+    physical.opExecInitInfo match {
+      case OpExecWithClassName(className, payload) =>
+        assert(className == classOf[FileScanOpExec].getName)
+        assert(payload.nonEmpty)
+      case other => fail(s"expected OpExecWithClassName, got $other")
+    }
+    assert(physical.inputPorts.size == 1)
+    assert(physical.outputPorts.size == 1)
+  }
+
+  it should "propagate sourceSchema to its single output port" in {
+    val physical = fileScanOpDesc.getPhysicalOp(WorkflowIdentity(1L), ExecutionIdentity(1L))
+    val outPortId = fileScanOpDesc.operatorInfo.outputPorts.head.id
+    val out = physical.propagateSchema.func(Map.empty)
+    assert(out(outPortId) == fileScanOpDesc.sourceSchema())
+  }
 }

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

@@ -19,8 +19,11 @@
 
 package org.apache.texera.amber.operator.source.scan.file
 
+import com.fasterxml.jackson.databind.node.ObjectNode
+import org.apache.texera.amber.core.executor.OpExecWithClassName
 import org.apache.texera.amber.core.storage.FileResolver
 import org.apache.texera.amber.core.tuple.{AttributeType, Schema, SchemaEnforceable, Tuple}
+import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import org.apache.texera.amber.operator.TestOperators
 import org.apache.texera.amber.operator.source.scan.{FileAttributeType, FileDecodingMethod}
 import org.apache.texera.amber.util.JSONUtils.objectMapper
@@ -185,4 +188,37 @@ class FileScanSourceOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
     FileScanSourceOpExec.close()
   }
 
+  "FileScanSourceOpDesc.getPhysicalOp" should
+    "wire the FileScanSourceOpExec class as a source op and propagate its schema" in {
+    val physical =
+      fileScanSourceOpDesc.getPhysicalOp(WorkflowIdentity(1L), ExecutionIdentity(1L))
+    physical.opExecInitInfo match {
+      case OpExecWithClassName(className, descString) =>
+        assert(className == classOf[FileScanSourceOpExec].getName)
+        assert(descString.nonEmpty)
+      case other => fail(s"expected OpExecWithClassName, got $other")
+    }
+    assert(physical.inputPorts.isEmpty)
+    val outPortId = fileScanSourceOpDesc.operatorInfo.outputPorts.head.id
+    assert(physical.outputPorts.keySet == Set(outPortId))
+    val propagated = physical.propagateSchema.func(Map.empty)
+    assert(propagated(outPortId) == fileScanSourceOpDesc.sourceSchema())
+  }
+
+  "FileScanSourceOpDesc.sourceSchema" should
+    "prepend a filename column when outputFileName is enabled" in {
+    // outputFileName is a val; round-trip through JSON (which carries the operatorType
+    // discriminator) with the flag flipped on, since it can't be set on an instance.
+    val node =
+      objectMapper
+        .readTree(objectMapper.writeValueAsString(fileScanSourceOpDesc))
+        .asInstanceOf[ObjectNode]
+    node.put("outputFileName", true)
+    val withFlag = objectMapper.treeToValue(node, classOf[FileScanSourceOpDesc])
+    val schema: Schema = withFlag.sourceSchema()
+    assert(schema.getAttributes.length == 2)
+    assert(schema.getAttribute("filename").getType == AttributeType.STRING)
+    assert(schema.getAttribute("line").getType == AttributeType.STRING)
+  }
+
 }

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

@@ -19,7 +19,9 @@
 
 package org.apache.texera.amber.operator.source.scan.text
 
+import org.apache.texera.amber.core.executor.OpExecWithClassName
 import org.apache.texera.amber.core.tuple.{AttributeType, Schema, SchemaEnforceable, Tuple}
+import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import org.apache.texera.amber.operator.TestOperators
 import org.apache.texera.amber.operator.source.scan.FileAttributeType
 import org.apache.texera.amber.util.JSONUtils.objectMapper
@@ -180,4 +182,27 @@ class TextInputSourceOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
     val path: Path = Paths.get(filePath)
     new String(Files.readAllBytes(path), StandardCharsets.UTF_8)
   }
+
+  "TextInputSourceOpDesc.getPhysicalOp" should
+    "wire the TextInputSourceOpExec class as a source op with one output port" in {
+    val physical =
+      textInputSourceOpDesc.getPhysicalOp(WorkflowIdentity(1L), ExecutionIdentity(1L))
+    physical.opExecInitInfo match {
+      case OpExecWithClassName(className, _) =>
+        assert(className == classOf[TextInputSourceOpExec].getName)
+      case other => fail(s"expected OpExecWithClassName, got $other")
+    }
+    assert(physical.inputPorts.isEmpty)
+    assert(physical.outputPorts.size == 1)
+  }
+
+  it should "propagate sourceSchema to its single output port" in {
+    textInputSourceOpDesc.attributeType = FileAttributeType.STRING
+    val physical =
+      textInputSourceOpDesc.getPhysicalOp(WorkflowIdentity(1L), ExecutionIdentity(1L))
+    val outPortId = textInputSourceOpDesc.operatorInfo.outputPorts.head.id
+    val out = physical.propagateSchema.func(Map.empty)
+    assert(out.keySet == Set(outPortId))
+    assert(out(outPortId) == textInputSourceOpDesc.sourceSchema())
+  }
 }