Skip to content

test(workflow-operator): deduplicate operator test suites #9237

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
aglinxinyuan wants to merge 1 commit into
base: main
Choose a base branch
from
Open

test(workflow-operator): deduplicate operator test suites #9237

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
234 changes: 89 additions & 145 deletions ...ow-operator/src/test/scala/org/apache/texera/amber/operator/hashJoin/HashJoinOpSpec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,25 +23,24 @@ import org.apache.texera.amber.core.tuple._
import org.apache.texera.amber.core.workflow.PortIdentity
import org.apache.texera.amber.operator.hashJoin.HashJoinOpDesc.HASH_JOIN_INTERNAL_KEY_NAME
import org.apache.texera.amber.util.JSONUtils.objectMapper
import org.scalatest.BeforeAndAfter
import org.scalatest.flatspec.AnyFlatSpec
class HashJoinOpSpec extends AnyFlatSpec with BeforeAndAfter {

class HashJoinOpSpec extends AnyFlatSpec {
val build: Int = 0
val probe: Int = 1

var buildOpExec: HashJoinBuildOpExec[String] = _
var probeOpExec: HashJoinProbeOpExec[String] = _
var opDesc: HashJoinOpDesc[String] = _
case class JoinResult(
tuples: List[TupleLike],
onFinishTuples: List[TupleLike]
)

def getInternalHashTableSchema(buildInputSchema: Schema): Schema = {
Schema()
.add(HASH_JOIN_INTERNAL_KEY_NAME, AttributeType.ANY)
.add(buildInputSchema)

}

def tuple(name: String, n: Int = 1, i: Option[Int]): Tuple = {

Tuple
.builder(schema(name, n))
.addSequentially(Array[Any](i.map(_.toString).orNull, i.map(_.toString).orNull))
Expand All @@ -52,174 +51,119 @@ class HashJoinOpSpec extends AnyFlatSpec with BeforeAndAfter {
Schema()
.add(new Attribute(name, AttributeType.STRING))
.add(new Attribute(name + "_" + n, AttributeType.STRING))

}

it should "work with basic two input streams with different buildAttributeName and probeAttributeName" in {

opDesc = new HashJoinOpDesc[String]()
opDesc.buildAttributeName = "build_1"
opDesc.probeAttributeName = "probe_1"
opDesc.joinType = JoinType.INNER
val inputSchemas = Map(PortIdentity() -> schema("build"), PortIdentity(1) -> schema("probe"))
private def runJoin(
opDesc: HashJoinOpDesc[String],
inputSchemas: Map[PortIdentity, Schema],
buildTuples: Seq[Tuple],
probeTuples: Seq[Tuple]
): JoinResult = {
val outputSchema = opDesc.getExternalOutputSchemas(inputSchemas).values.head
val descJson = objectMapper.writeValueAsString(opDesc)

buildOpExec = new HashJoinBuildOpExec[String](objectMapper.writeValueAsString(opDesc))
val buildOpExec = new HashJoinBuildOpExec[String](descJson)
buildOpExec.open()

(0 to 7).map(i => {
assert(
buildOpExec.processTuple(tuple("build", 1, Some(i)), build).isEmpty
)
})

val buildOpOutputIterator =
buildOpExec.onFinish(build)
assert(buildOpOutputIterator.hasNext)

probeOpExec = new HashJoinProbeOpExec[String](objectMapper.writeValueAsString(opDesc))

probeOpExec.open()
while (buildOpOutputIterator.hasNext) {
assert(
probeOpExec
.processTuple(
buildOpOutputIterator
.next()
.asInstanceOf[SchemaEnforceable]
.enforceSchema(getInternalHashTableSchema(inputSchemas.head._2)),
build
try {
buildTuples.foreach { t =>
assert(buildOpExec.processTuple(t, build).isEmpty)
}
val buildOpOutputIterator = buildOpExec.onFinish(build)
assert(buildOpOutputIterator.hasNext)

val probeOpExec = new HashJoinProbeOpExec[String](descJson)
probeOpExec.open()
try {
while (buildOpOutputIterator.hasNext) {
assert(
probeOpExec
.processTuple(
buildOpOutputIterator
.next()
.asInstanceOf[SchemaEnforceable]
.enforceSchema(getInternalHashTableSchema(inputSchemas.head._2)),
build
)
.isEmpty
)
.isEmpty
)
}
assert(probeOpExec.onFinish(build).isEmpty)

val outputTuples = probeTuples
.map(t => probeOpExec.processTuple(t, probe))
.foldLeft(Iterator[TupleLike]())(_ ++ _)
.map(_.asInstanceOf[SchemaEnforceable].enforceSchema(outputSchema))
.toList

val onFinishTuples = probeOpExec
.onFinish(probe)
.map(_.asInstanceOf[SchemaEnforceable].enforceSchema(outputSchema))
.toList

JoinResult(outputTuples, onFinishTuples)
} finally {
probeOpExec.close()
}
} finally {
buildOpExec.close()
}
assert(probeOpExec.onFinish(build).isEmpty)

buildOpExec.close()

val outputTuples = (5 to 9)
.map(i => probeOpExec.processTuple(tuple("probe", 1, Some(i)), probe))
.foldLeft(Iterator[TupleLike]())(_ ++ _)
.map(tupleLike => tupleLike.asInstanceOf[SchemaEnforceable].enforceSchema(outputSchema))
.toList

assert(probeOpExec.onFinish(probe).isEmpty)
}

assert(outputTuples.size == 3)
assert(outputTuples.head.getFields.length == 3)
it should "work with basic two input streams with different buildAttributeName and probeAttributeName" in {
val opDesc = new HashJoinOpDesc[String]()
opDesc.buildAttributeName = "build_1"
opDesc.probeAttributeName = "probe_1"
opDesc.joinType = JoinType.INNER
val inputSchemas = Map(PortIdentity() -> schema("build"), PortIdentity(1) -> schema("probe"))

probeOpExec.close()
val result = runJoin(
opDesc,
inputSchemas,
buildTuples = (0 to 7).map(i => tuple("build", 1, Some(i))),
probeTuples = (5 to 9).map(i => tuple("probe", 1, Some(i)))
)

assert(result.onFinishTuples.isEmpty)
assert(result.tuples.size == 3)
assert(result.tuples.head.getFields.length == 3)
}

it should "work with basic two input streams with the same buildAttributeName and probeAttributeName" in {
opDesc = new HashJoinOpDesc[String]()
val opDesc = new HashJoinOpDesc[String]()
opDesc.buildAttributeName = "same"
opDesc.probeAttributeName = "same"
opDesc.joinType = JoinType.INNER
val inputSchemas =
Map(PortIdentity() -> schema("same", 1), PortIdentity(1) -> schema("same", 2))
val outputSchema = opDesc.getExternalOutputSchemas(inputSchemas).values.head

buildOpExec = new HashJoinBuildOpExec[String](objectMapper.writeValueAsString(opDesc))
buildOpExec.open()

(0 to 7).map(i => {
assert(
buildOpExec.processTuple(tuple("same", 1, Some(i)), build).isEmpty
)
})
val buildOpOutputIterator =
buildOpExec.onFinish(build)
assert(buildOpOutputIterator.hasNext)

probeOpExec = new HashJoinProbeOpExec[String](objectMapper.writeValueAsString(opDesc))
probeOpExec.open()

while (buildOpOutputIterator.hasNext) {
assert(
probeOpExec
.processTuple(
buildOpOutputIterator
.next()
.asInstanceOf[SchemaEnforceable]
.enforceSchema(getInternalHashTableSchema(inputSchemas.head._2)),
build
)
.isEmpty
)
}
assert(probeOpExec.onFinish(build).isEmpty)

buildOpExec.close()

val outputTuples = (5 to 9)
.map(i => probeOpExec.processTuple(tuple("same", n = 2, Some(i)), probe))
.foldLeft(Iterator[TupleLike]())(_ ++ _)
.map(tupleLike => tupleLike.asInstanceOf[SchemaEnforceable].enforceSchema(outputSchema))
.toList

assert(probeOpExec.onFinish(probe).isEmpty)

assert(outputTuples.size == 3)
assert(outputTuples.head.getFields.length == 3)
val result = runJoin(
opDesc,
inputSchemas,
buildTuples = (0 to 7).map(i => tuple("same", 1, Some(i))),
probeTuples = (5 to 9).map(i => tuple("same", n = 2, Some(i)))
)

probeOpExec.close()
assert(result.onFinishTuples.isEmpty)
assert(result.tuples.size == 3)
assert(result.tuples.head.getFields.length == 3)
}

it should "work with basic two input streams with the same buildAttributeName and probeAttributeName with Full Outer Join" in {
opDesc = new HashJoinOpDesc[String]()
val opDesc = new HashJoinOpDesc[String]()
opDesc.buildAttributeName = "same"
opDesc.probeAttributeName = "same"
opDesc.joinType = JoinType.FULL_OUTER
val inputSchemas =
Map(PortIdentity() -> schema("same", 1), PortIdentity(1) -> schema("same", 2))
val outputSchema = opDesc.getExternalOutputSchemas(inputSchemas).values.head

buildOpExec = new HashJoinBuildOpExec[String](objectMapper.writeValueAsString(opDesc))
buildOpExec.open()

(0 to 7).map(i => {
assert(
buildOpExec.processTuple(tuple("same", 1, Some(i)), build).isEmpty
)
})
val buildOpOutputIterator =
buildOpExec.onFinish(build)
assert(buildOpOutputIterator.hasNext)

probeOpExec = new HashJoinProbeOpExec[String](objectMapper.writeValueAsString(opDesc))
probeOpExec.open()

while (buildOpOutputIterator.hasNext) {
assert(
probeOpExec
.processTuple(
buildOpOutputIterator
.next()
.asInstanceOf[SchemaEnforceable]
.enforceSchema(getInternalHashTableSchema(inputSchemas.head._2)),
build
)
.isEmpty
)
}
assert(probeOpExec.onFinish(build).isEmpty)

buildOpExec.close()

assert(
(5 to 9)
.map(_ => {
probeOpExec.processTuple(tuple("same", n = 2, None), probe)
})
.foldLeft(Iterator[TupleLike]())(_ ++ _)
.map(tupleLike => tupleLike.asInstanceOf[SchemaEnforceable].enforceSchema(outputSchema))
.size == 5
val result = runJoin(
opDesc,
inputSchemas,
buildTuples = (0 to 7).map(i => tuple("same", 1, Some(i))),
probeTuples = (5 to 9).map(_ => tuple("same", n = 2, None))
)

assert(probeOpExec.onFinish(probe).size == 8)

probeOpExec.close()
assert(result.tuples.size == 5)
assert(result.onFinishTuples.size == 8)
}
}
Loading
Loading