Copilot commented on code in PR #8132:
URL: https://github.com/apache/texera/pull/8132#discussion_r3894097081
##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sort/StableMergeSortOpExecSpec.scala:
##########
@@ -779,4 +779,99 @@ class StableMergeSortOpExecSpec extends AnyFlatSpec {
exec.close()
}
+ //
===========================================================================
+ // H. Lifecycle guards and unsupported key types
+ //
===========================================================================
+
+ // The bucket stack is allocated by open(), so both readers of the field
guard
+ // against being called on an executor the engine has not opened (or has
already
+ // torn down). Every other test in this spec goes through open() first,
which is
+ // why these paths need their own cases.
+ //
+ // Honest scope note: debugBucketSizes is itself a test hook (its only
callers
+ // anywhere in the repo are this spec's getBucketSizes helper and the cases
+ // below), so its null guard is not shipped behavior. What is pinned here is
+ // which field that guard reads and what close() actually does to the stack.
+
+ "debugBucketSizes" should "report no buckets before open and the live stack
after it" in {
+ val schema = schemaOf("value" -> AttributeType.INTEGER)
+ val desc = new StableMergeSortOpDesc(); desc.keys =
sortKeysBuffer(sortKey("value"))
+ val exec = new StableMergeSortOpExec(objectMapper.writeValueAsString(desc))
+ assert(getBucketSizes(exec) == Nil)
+
+ // open() allocates the bucket stack but leaves the sort keys uncompiled —
+ // compileSortKeys runs on the first processTuple — so pushing a bucket
+ // straight through the internal hook reaches a state where the stack is
+ // non-empty while compiledSortKeys is still null. That state is what
+ // distinguishes which of the two fields the Nil guard is reading; a single
+ // size-1 push performs no merge, so no comparison is attempted.
+ exec.open()
+ exec.pushBucketAndCombine(ArrayBuffer(tupleOf(schema, "value" -> 1)))
+ assert(getBucketSizes(exec) == List(1))
+ exec.close()
Review Comment:
If the assertion fails, `exec.close()` won’t run, which can leave executor
state/resources allocated and potentially impact later tests in the same JVM.
Consider structuring the test so `close()` is guaranteed (e.g., `try`/`finally`
around the post-`open()` section).
##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sort/StableMergeSortOpExecSpec.scala:
##########
@@ -779,4 +779,99 @@ class StableMergeSortOpExecSpec extends AnyFlatSpec {
exec.close()
}
+ //
===========================================================================
+ // H. Lifecycle guards and unsupported key types
+ //
===========================================================================
+
+ // The bucket stack is allocated by open(), so both readers of the field
guard
+ // against being called on an executor the engine has not opened (or has
already
+ // torn down). Every other test in this spec goes through open() first,
which is
+ // why these paths need their own cases.
+ //
+ // Honest scope note: debugBucketSizes is itself a test hook (its only
callers
+ // anywhere in the repo are this spec's getBucketSizes helper and the cases
+ // below), so its null guard is not shipped behavior. What is pinned here is
+ // which field that guard reads and what close() actually does to the stack.
+
+ "debugBucketSizes" should "report no buckets before open and the live stack
after it" in {
+ val schema = schemaOf("value" -> AttributeType.INTEGER)
+ val desc = new StableMergeSortOpDesc(); desc.keys =
sortKeysBuffer(sortKey("value"))
+ val exec = new StableMergeSortOpExec(objectMapper.writeValueAsString(desc))
+ assert(getBucketSizes(exec) == Nil)
+
+ // open() allocates the bucket stack but leaves the sort keys uncompiled —
+ // compileSortKeys runs on the first processTuple — so pushing a bucket
+ // straight through the internal hook reaches a state where the stack is
+ // non-empty while compiledSortKeys is still null. That state is what
+ // distinguishes which of the two fields the Nil guard is reading; a single
+ // size-1 push performs no merge, so no comparison is attempted.
+ exec.open()
+ exec.pushBucketAndCombine(ArrayBuffer(tupleOf(schema, "value" -> 1)))
+ assert(getBucketSizes(exec) == List(1))
+ exec.close()
+ }
+
+ "close" should "be a no-op when it runs before open" in {
+ val desc = new StableMergeSortOpDesc(); desc.keys =
sortKeysBuffer(sortKey("value"))
+ val exec = new StableMergeSortOpExec(objectMapper.writeValueAsString(desc))
+ // Pure "does not throw": the executor was never opened, so this case can
only
+ // catch a crash in the guard, never a change in what close() does. The
+ // companion case below is the one that pins the effect.
+ exec.close()
+ assert(getBucketSizes(exec) == Nil)
+ }
+
+ it should "drop the buffered buckets when it runs after open" in {
+ val schema = schemaOf("value" -> AttributeType.INTEGER)
+ val desc = new StableMergeSortOpDesc(); desc.keys =
sortKeysBuffer(sortKey("value"))
+ val exec = new
StableMergeSortOpExec(objectMapper.writeValueAsString(desc)); exec.open()
+ List(3, 1, 2).foreach(i => exec.processTuple(tupleOf(schema, "value" ->
i), 0))
+ assert(getBucketSizes(exec) == List(2, 1))
+ exec.close()
+ assert(getBucketSizes(exec) == Nil)
+ }
+
+ "StableMergeSortOpExec" should "reject a sort key whose attribute type it
cannot compare" in {
+ // The comparison switch implements 7 of AttributeType's 9 constants; ANY
and
+ // LARGE_BINARY both fall through to the catch-all, and both are offered as
+ // sort keys by the attribute picker (SortCriteriaUnit.attributeName
carries a
+ // bare @AutofillAttributeName with no type restriction). Probing both is
what
+ // makes the arm's breadth a contract rather than a case for one type.
+ def rejects(attrType: AttributeType, first: Any, second: Any):
IllegalStateException = {
+ val schema = schemaOf("value" -> attrType)
+ val desc = new StableMergeSortOpDesc(); desc.keys =
sortKeysBuffer(sortKey("value"))
+ val exec = new
StableMergeSortOpExec(objectMapper.writeValueAsString(desc)); exec.open()
+ // The first tuple only compiles the sort keys. The second makes the
stack's
+ // top two buckets equal-sized, and the binary carry that merges them is
the
+ // first comparison of two values.
+ exec.processTuple(tupleOf(schema, "value" -> first), 0)
+ val thrown = intercept[IllegalStateException] {
+ exec.processTuple(tupleOf(schema, "value" -> second), 0)
+ }
+ exec.close()
+ thrown
Review Comment:
`exec.close()` won’t run if `intercept` fails (e.g., no exception is thrown)
or if a preceding call/assertion throws, which can leak executor resources and
make the suite more brittle. Wrap the opened executor usage in `try`/`finally`
(closing in `finally`) so cleanup happens even on test failure.
--
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]
