aglinxinyuan commented on code in PR #8132:
URL: https://github.com/apache/texera/pull/8132#discussion_r3896210000
##########
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:
The `try`/`finally` is refused for the same reason as the line-811 thread —
same class, same inert `close()`, same per-test instance with nothing shared.
The control flow the comment describes is real; its consequence is not, since
the `ArrayBuffer` is collected regardless and a failing `rejects` aborts the
test before the second invocation.
I did address the failure-path concern behind it, in `f27a7e16`, in the form
that actually matters for this helper. `rejects` is invoked twice from one test
case (`AttributeType.ANY` and `LARGE_BINARY`), and ScalaTest's `intercept`
failure text names neither the type nor the call site — so a regression in one
probe was indistinguishable from the other. It is now wrapped in a `withClue`
naming the type.
One subtlety: it has to be `attrType.name()`, not the interpolated enum.
`AttributeType.toString` delegates to the `@JsonValue` getter, which returns
`""` for `ANY`, so `$attrType` would have rendered the clue blank for exactly
one of the two probes. Verified by temporarily replacing the `intercept` body
so nothing was thrown.
##########
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:
Refused, with the premise checked. `StableMergeSortOpExec.close()` is
exactly `if (sortedBuckets != null) sortedBuckets.clear()` — it clears one
private in-memory `ArrayBuffer`. The `OperatorExecutor` trait it implements has
a no-op `close()`; there is no file handle, socket, thread or global registry.
The exec is a per-test local, freshly constructed in each case, and the spec
has no suite-level state and no `beforeEach`/`afterEach`. So "leak executor
resources" and "impact later tests in the same JVM" do not hold for this class
— the buffer is collected either way.
Separately: this file has **15 bare `exec.close()` sites and zero
`try`/`finally`**. A diff-scoped review sees the 4 this PR adds and not the 11
already on `main`, so wrapping just these would leave 13 identical siblings and
imply to a reader that those are bugs.
To be fair to the suggestion: applying it would not have broken anything —
coverage and the mutation kills are unaffected either way. This is a refusal on
"the stated mechanism does not exist here", not on risk. If the idiom is
wanted, the right shape is all 15 sites in a separate commit.
--
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]
