aglinxinyuan commented on code in PR #6100:
URL: https://github.com/apache/texera/pull/6100#discussion_r3522942421
##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/text/TextInputSourceOpDescSpec.scala:
##########
@@ -180,4 +182,29 @@ 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 ==
"org.apache.texera.amber.operator.source.scan.text.TextInputSourceOpExec"
+ )
Review Comment:
Done — asserts `classOf[TextInputSourceOpExec].getName` instead of the
hardcoded FQN string.
##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpDescSpec.scala:
##########
@@ -185,4 +188,39 @@ 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 ==
"org.apache.texera.amber.operator.source.scan.file.FileScanSourceOpExec"
+ )
+ assert(descString.nonEmpty)
Review Comment:
Done — asserts `classOf[FileScanSourceOpExec].getName`.
##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDescSpec.scala:
##########
@@ -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 ==
"org.apache.texera.amber.operator.source.scan.file.FileScanOpExec")
+ assert(payload.nonEmpty)
Review Comment:
Done — asserts `classOf[FileScanOpExec].getName`.
##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/projection/ProjectionOpDescSpec.scala:
##########
@@ -110,4 +116,24 @@ 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 "pass through partitions that are neither hash nor range" in {
+ val out = projectionOpDesc.derivePartition()(List(SinglePartition()))
+ assert(out == SinglePartition())
+ }
Review Comment:
Added an empty-RangePartition case. Note: RangePartition's companion `apply`
already rewrites empty attributes to UnknownPartition, so an empty range never
reaches the range arm's else — either way the result is UnknownPartition, which
the new test documents.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
