This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-8132-7ba7b9865f741c652ec9af9eae80645387a377fd in repository https://gitbox.apache.org/repos/asf/texera.git
commit fed849dc3262433a66c4878ce2ec8170d845267b Author: Xinyuan Lin <[email protected]> AuthorDate: Mon Aug 31 19:16:21 2026 +0000 test(workflow-operator): cover the sort, filter and codegen gaps (#8132) ### What changes were proposed in this PR? Three existing specs extended and one new, 66 tests to 73. | File | Codecov-missed | Newly covered | |---|---|---| | `StableMergeSortOpExec.scala` | 5 → **1** | 62, 71, 269, 275 | | `FilterPredicate.java` | 4 → **2** | 66, 80 | | `SklearnClassifierOpDesc.scala` | 2 → **0** | 28, 30 | | `TaskCodegen.scala` | 2 → **1** | 80 | **+9 fully-covered lines and +4 branch arms** — 279/307 → 288/307 across the bundle (90.9% → 93.8%). ### Two files from the original scope were dropped - **`AsterixDBSourceOpExec`** (12 missed of 104) — the nominated headline. No spec written; it stays 12 → 12. - **`OPVersion.java`** (3 missed of 18) — stays 3 → 3. Both are named rather than quietly omitted, since the bundle was pitched partly on `AsterixDBSourceOpExec`'s 11 partial arms. A known defect in `AsterixDBSourceOpExec` is deliberately **not** pinned: `close()` does not clear the cached tuple. ### Verification **16 mutations, all 16 killed** — the builder's 9 re-derived from scratch, plus 7 that two reviewers found surviving the first draft. Each was preceded by a false-kill guard (sha256 of all four production files against a scratch snapshot **and** an empty `git diff -- '*/src/main/*'`) and followed by a revert-from-snapshot that re-verified both; all 16 revert checks logged clean. Anchor text was asserted to occur exactly once before each apply. No mutation failed to compile, so none was discarded. Two reviewers returned nine findings; all repaired. The repair added discrimination, not count — the coverage figures are byte-identical to the builder's and were reproduced independently rather than adopted. Measured with two `WorkflowOperator/jacoco` runs, one fresh sbt JVM each, the jacoco directory removed between them, `AMBER_TEST_FILTER=skip-integration TEXERA_SERVICE_LOG_LEVEL=WARN` to match CI. Both sides used an identical suite-name filter in a throwaway `.sbt` that excluded `FileScanSourceOpExecSpec` — that suite aborts on Windows in its own cleanup, and because sbt-jacoco runs unforked and skips `saveRuntimeData` when the test task fails, an unfiltered run emits an **all-zero** report rather than a partial one. The throwaway file was deleted. The before-state restored each modified spec with `git show HEAD:<exact single path>` and deleted the new one — never `git checkout -- <directory>`. **A pre-existing flake found while reconciling the module totals, and reported rather than absorbed.** Module-wide branch misses moved 1150 → 1147, which is −3 where these four files contribute −4. Diffing every sourcefile between the two reports isolated the discrepancy to `IntervalJoinOpExec.scala` (21 → 22 missed arms), a file this PR never touches. Its covering spec uses an unseeded RNG, so its per-run branch totals are not stable — worth knowing before anyone quotes module-wide arm counts from a single run. No production file is touched; `git status --porcelain` is clean, with no leaked `test_large_binary.txt` and no throwaway probe spec left behind. ### Any related issues, documentation, discussions? Closes #8130 ### How was this PR tested? ``` sbt "WorkflowOperator/testOnly org.apache.texera.amber.operator.sort.StableMergeSortOpExecSpec org.apache.texera.amber.operator.filter.FilterPredicateSpec org.apache.texera.amber.operator.sklearn.SklearnClassifierOpDescCodegenSpec org.apache.texera.amber.operator.huggingFace.codegen.TaskCodegenSpec" ``` ``` [info] Total number of tests run: 73 [info] Tests: succeeded 73, failed 0, canceled 0, ignored 0, pending 0 [info] All tests passed. ``` `WorkflowOperator/Test/scalafmtCheck` and `WorkflowOperator/Test/scalafix --check` both pass. Re-run after rebasing onto current `main`. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 5) --- .../operator/filter/FilterPredicateSpec.scala | 28 ++++++ .../huggingFace/codegen/TaskCodegenSpec.scala | 62 ++++++++++++ .../SklearnClassifierOpDescCodegenSpec.scala | 33 +++++++ .../operator/sort/StableMergeSortOpExecSpec.scala | 106 ++++++++++++++++++++- 4 files changed, 228 insertions(+), 1 deletion(-) diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/FilterPredicateSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/FilterPredicateSpec.scala index 0bb9a3a475..4f801d02a0 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/FilterPredicateSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/FilterPredicateSpec.scala @@ -247,6 +247,34 @@ class FilterPredicateSpec extends AnyFlatSpec with Matchers { evaluates(ComparisonType.LESS_THAN_OR_EQUAL_TO, 30) shouldBe false } + // --- the comparison switch's fail-loud arm --------------------------------- + + "FilterPredicate's comparison switch" should "fail loudly on a comparison type it does not implement" in { + // Reachability, stated plainly: ComparisonType has 8 constants; evaluate() + // answers IS_NULL / IS_NOT_NULL itself and returns before it ever dispatches + // them, and the other 6 all have explicit cases. The default arm is therefore + // DEAD through the public API, and this test reaches it by reflecting into a + // private static. It documents the fail-loud contract a newly added + // ComparisonType would hit; it is not a regression guard on a shipped path. + val evaluateFilter = classOf[FilterPredicate].getDeclaredMethod( + "evaluateFilter", + classOf[Comparable[_]], + classOf[Comparable[_]], + classOf[ComparisonType] + ) + evaluateFilter.setAccessible(true) + val thrown = intercept[java.lang.reflect.InvocationTargetException] { + evaluateFilter.invoke(null, "a", "b", ComparisonType.IS_NULL) + } + // Exact class, not `a[RuntimeException]`: every unchecked exception is a + // RuntimeException, so the loose form is satisfied by any substituted + // exception type and would leave the "fail loudly" claim resting on the + // message alone. + thrown.getCause.getClass shouldBe classOf[RuntimeException] + thrown.getCause.getMessage shouldBe + "Unable to do comparison: unknown comparison type: IS_NULL" + } + "FilterPredicate.equals" should "distinguish predicates that differ only in attribute, or only in condition" in { val base = new FilterPredicate("age", ComparisonType.EQUAL_TO, "1") val otherAttribute = new FilterPredicate("name", ComparisonType.EQUAL_TO, "1") diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/huggingFace/codegen/TaskCodegenSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/huggingFace/codegen/TaskCodegenSpec.scala new file mode 100644 index 0000000000..7ee15f935d --- /dev/null +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/huggingFace/codegen/TaskCodegenSpec.scala @@ -0,0 +1,62 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.texera.amber.operator.huggingFace.codegen + +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +/** + * Covers the one piece of behavior the TaskCodegen trait implements itself: + * the default `tasks` set. + * + * Reachability, stated plainly: no production path executes this default today. + * The only four `.tasks` call sites in main are HuggingFaceInferenceOpDesc's + * dispatcher registrations (lines 175-178), and all four are on codegens that + * declare `override val tasks`; TextGenCodegen, the sole single-task codegen, is + * registered by its `task` string and never asks for `tasks`. This suite + * therefore documents an extension point for future single-task codegens rather + * than guarding a shipped path. + */ +class TaskCodegenSpec extends AnyFlatSpec with Matchers { + + /** + * Implements only the three abstract members, leaving `tasks` to the trait. + * + * `task` is declared `override val` to match every shipped codegen + * (AudioTaskCodegen, ImageTaskCodegen, MediaGenCodegen, QaRankingCodegen and + * TextGenCodegen all use `override val task`). The shape matters: were the + * trait's `tasks` ever turned into a `val`, it would be initialized during + * trait construction — before the implementation's `val task` is assigned — + * and would capture `Set(null)`. A stub written with `override def task` is + * immune to that initialization-order hazard and so would not notice it. + */ + private object SingleTaskStub extends TaskCodegen { + override val task: String = "probe-task-zX7q42" + override def payloadPython(ctx: CodegenContext): String = "" + override def parsePython(ctx: CodegenContext): String = "" + } + + "TaskCodegen.tasks" should "default to the singleton set of the codegen's own task" in { + // The stub's task string is deliberately not one of the real Hugging Face + // pipeline names, so a default hardcoded to some shipped task string rather + // than derived from `task` cannot pass this. + SingleTaskStub.tasks shouldBe Set("probe-task-zX7q42") + } +} diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/SklearnClassifierOpDescCodegenSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/SklearnClassifierOpDescCodegenSpec.scala index 0bdd32ec90..489a09ad1f 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/SklearnClassifierOpDescCodegenSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/SklearnClassifierOpDescCodegenSpec.scala @@ -133,4 +133,37 @@ class SklearnClassifierOpDescCodegenSpec extends AnyFlatSpec with Matchers { ) code should not include "CountVectorizer()" } + + // --- the base's own model-identity defaults -------------------------------- + + // Declared inside the spec class on purpose: PythonClassgraphScanner drops + // non-static enclosed classes, which is what keeps SklearnOpDescRegistrySpec's + // and PythonCodeRawInvalidTextSpec's classpath scans from treating this stub as + // a shipped operator. A top-level subclass here would break both suites. + private class BareClassifier extends SklearnClassifierOpDesc + + // Overrides only the model name, so the two hooks hold different values. That + // is what separates the base's own getImportStatements body from a body that + // merely forwards to the other hook — on BareClassifier alone both return "", + // which makes an exchange between them invisible. + private class NamedOnlyClassifier extends SklearnClassifierOpDesc { + override def getUserFriendlyModelName = "ProbeModel" + } + + "SklearnClassifierOpDesc" should "leave both model-identity hooks blank as base placeholders" in { + // NOT a claim that "" is the intended design. SklearnModelOpDesc declares both + // hooks abstract, and all of the shipped classifiers override them; these two + // bodies are placeholders that satisfy the abstract contract for the family. + // What is pinned is therefore what a subclass that forgets to override + // actually ships — a nameless operator whose generated pipeline stage comes + // out empty — not a default anyone should rely on. Leaving these two hooks + // abstract on SklearnClassifierOpDesc is filed as a follow-up rather than + // asserted here. + val bare = new BareClassifier + bare.getImportStatements shouldBe "" + bare.getUserFriendlyModelName shouldBe "" + + // The import hook is its own constant, not an alias for the name hook. + new NamedOnlyClassifier().getImportStatements shouldBe "" + } } diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sort/StableMergeSortOpExecSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sort/StableMergeSortOpExecSpec.scala index 46c3727cd3..1c05fe26a9 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sort/StableMergeSortOpExecSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sort/StableMergeSortOpExecSpec.scala @@ -19,7 +19,7 @@ package org.apache.texera.amber.operator.sort -import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple} +import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, LargeBinary, Schema, Tuple} import org.apache.texera.amber.util.JSONUtils.objectMapper import org.scalatest.flatspec.AnyFlatSpec @@ -779,4 +779,108 @@ 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) + // Clued because this helper runs twice within the one test case: intercept's + // own failure text ("Expected exception ... but no exception was thrown") + // names neither the attribute type nor the call site, so a regression in one + // of the two probes would be indistinguishable from the other. Built from + // name() rather than the enum itself: toString delegates to the @JsonValue + // getName, which is deliberately "" for ANY, so an interpolated $attrType + // would render this clue blank for exactly one of the two probes. + val thrown = withClue(s"attributeType=${attrType.name()}: ") { + intercept[IllegalStateException] { + exec.processTuple(tupleOf(schema, "value" -> second), 0) + } + } + exec.close() + thrown + } + + // Compared against a message built from the enum rather than a spelled-out + // literal: AttributeType.toString delegates to the @JsonValue getName, which + // is the empty string for ANY, and that rendering is workflow-core's artifact + // rather than something this operator should be made to own. Deriving it + // still pins WHICH operand the diagnostic interpolates. + assert( + rejects( + AttributeType.ANY, + java.lang.Integer.valueOf(1), + java.lang.Integer.valueOf(2) + ).getMessage == s"Unsupported attribute type ${AttributeType.ANY} in StableMergeSort" + ) + assert( + rejects( + AttributeType.LARGE_BINARY, + new LargeBinary("s3://bucket/a"), + new LargeBinary("s3://bucket/b") + ).getMessage == + s"Unsupported attribute type ${AttributeType.LARGE_BINARY} in StableMergeSort" + ) + } + }
