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-7018-91c04dad97b5068a3a0ae506e02289e557812ec0 in repository https://gitbox.apache.org/repos/asf/texera.git
commit 359b19dcedbb7fac7ecaed1af9abff4cf29174b1 Author: Xinyuan Lin <[email protected]> AuthorDate: Wed Jul 29 04:38:07 2026 -0700 test(workflow-operator): extend operator descriptor and predicate coverage (#7018) ### What changes were proposed in this PR? Adds 47 tests across five `common/workflow-operator` classes (33 -> 80 in these suites). No source changes; all four extended specs are **pure insertions** (0 deletions), so no existing test was modified. - **`AggregationOperation`**: the SUM/COUNT/MIN/MAX `merge` lambdas — invoked when several workers report into the global stage and never exercised by `AggregateOpSpec` — each cross-checked against the equivalent single-pass aggregation; CONCAT's empty-partial branch; the unknown-function error message; and `getFinal`'s detached-copy contract. - **`FilterPredicate`**: the two attribute-type cases nothing reached — `ANY` (routes through the string path, with numeric coercion and a lexicographic fallback) and `BINARY` (throws `unsupported attribute type`) — plus the `IS_NULL`/`IS_NOT_NULL` short-circuit ahead of the type switch, value trimming, and `NumberFormatException` propagation. - **`ECDFPlotOpDesc`** and **`TimeSeriesOpDesc`**: `operatorInfo` and output schema, the optional settings that gate generated Plotly arguments, the generated-code empty-table guards, and — per the repo convention for LogicalOp subtypes — a **JSON round-trip** through `LogicalOp` asserting every config field survives, plus a minimal payload deserializing to the documented defaults. Neither spec had a round-trip before. - **`OPVersion`** (new spec): the `"N/A"` fallback for a path with no commit history, and the name-keyed memoization contract. Some branches were deliberately **not** contrived, and are listed with reasons in the review notes — notably `FilterPredicate`'s unreachable `default:` throw and dead null-guard ternaries (`evaluate` returns before them), and `OPVersion`'s success path (needs an openable repository; inside a git worktree the `.git` entry is a file and jgit leaves the handle null, so the spec asserts only what holds in both environments). ### Any related issues, documentation, discussions? Closes #7016. ### How was this PR tested? `sbt -java-home <jbr-17> "WorkflowOperator/testOnly *AggregationOperationSpec *FilterPredicateSpec *ECDFPlotOpDescSpec *TimeSeriesOpDescSpec *OPVersionSpec"` -> 80 succeeded, 0 failed. `Test/scalafmtCheck` + `Test/scalafix --check` clean. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 4.8 [1M context]) --- .../aggregate/AggregationOperationSpec.scala | 135 ++++++++++++++ .../operator/filter/FilterPredicateSpec.scala | 65 +++++++ .../amber/operator/metadata/OPVersionSpec.scala | 156 ++++++++++++++++ .../timeSeriesPlot/TimeSeriesOpDescSpec.scala | 197 ++++++++++++++++++++ .../ecdfPlot/ECDFPlotOpDescSpec.scala | 198 +++++++++++++++++++++ 5 files changed, 751 insertions(+) diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/aggregate/AggregationOperationSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/aggregate/AggregationOperationSpec.scala index 0931e3bab7..44f0f1b3da 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/aggregate/AggregationOperationSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/aggregate/AggregationOperationSpec.scala @@ -259,4 +259,139 @@ class AggregationOperationSpec extends AnyFlatSpec { agg.iterate(agg.init(), tupleOf("v", AttributeType.TIMESTAMP, null)) assert(agg.finalAgg(allNull) == null) } + + // --- merge: the partial-combination lambda of each aggregation -------------- + + // `AggregateOpSpec` drives init/iterate/finalAgg for every aggregation but only + // ever merges AVERAGE/CONCAT partials. The `merge` lambdas of SUM, COUNT, MIN + // and MAX are what the global stage calls when several workers report in, so + // each one is exercised here directly and cross-checked against the equivalent + // single-pass aggregation. + + private def aggregateAll( + agg: DistributedAggregation[Object], + t: AttributeType, + values: Seq[AnyRef] + ): Object = + agg.finalAgg(values.map(v => tupleOf("v", t, v)).foldLeft(agg.init())(agg.iterate)) + + "SUM aggregation merge" should "add two partials, matching a single-pass SUM" in { + val agg = op(AggregationFunction.SUM).getAggFunc(AttributeType.LONG) + val left = Seq[AnyRef](Long.box(1L), Long.box(2L)) + val right = Seq[AnyRef](Long.box(10L), Long.box(20L)) + + val p1 = left.map(v => tupleOf("v", AttributeType.LONG, v)).foldLeft(agg.init())(agg.iterate) + val p2 = right.map(v => tupleOf("v", AttributeType.LONG, v)).foldLeft(agg.init())(agg.iterate) + + assert(agg.finalAgg(agg.merge(p1, p2)) == Long.box(33L)) + assert(agg.finalAgg(agg.merge(p1, p2)) == aggregateAll(agg, AttributeType.LONG, left ++ right)) + // merging with an untouched (zero) partial must be a no-op + assert(agg.finalAgg(agg.merge(p1, agg.init())) == Long.box(3L)) + } + + "COUNT aggregation merge" should "add the per-worker counts" in { + val agg = op(AggregationFunction.COUNT).getAggFunc(AttributeType.INTEGER) + val p1 = Seq[AnyRef](Int.box(1), null, Int.box(3)) + .map(v => tupleOf("v", AttributeType.INTEGER, v)) + .foldLeft(agg.init())(agg.iterate) + val p2 = Seq[AnyRef](Int.box(4)) + .map(v => tupleOf("v", AttributeType.INTEGER, v)) + .foldLeft(agg.init())(agg.iterate) + + // COUNT(v) skips the null, so 2 + 1 == 3 + assert(agg.finalAgg(agg.merge(p1, p2)) == Int.box(3)) + assert(agg.finalAgg(agg.merge(agg.init(), agg.init())) == Int.box(0)) + } + + "MIN and MAX aggregation merge" should "pick the smaller and larger partial respectively" in { + val minAgg = op(AggregationFunction.MIN).getAggFunc(AttributeType.DOUBLE) + val maxAgg = op(AggregationFunction.MAX).getAggFunc(AttributeType.DOUBLE) + val left = Seq[AnyRef](Double.box(4.0), Double.box(9.0)) + val right = Seq[AnyRef](Double.box(-1.5), Double.box(2.0)) + + def partialOf(agg: DistributedAggregation[Object], values: Seq[AnyRef]): Object = + values.map(v => tupleOf("v", AttributeType.DOUBLE, v)).foldLeft(agg.init())(agg.iterate) + + assert( + minAgg.finalAgg(minAgg.merge(partialOf(minAgg, left), partialOf(minAgg, right))) + == Double.box(-1.5) + ) + // merge must be symmetric + assert( + minAgg.finalAgg(minAgg.merge(partialOf(minAgg, right), partialOf(minAgg, left))) + == Double.box(-1.5) + ) + assert( + maxAgg.finalAgg(maxAgg.merge(partialOf(maxAgg, left), partialOf(maxAgg, right))) + == Double.box(9.0) + ) + assert( + maxAgg.finalAgg(maxAgg.merge(partialOf(maxAgg, right), partialOf(maxAgg, left))) + == Double.box(9.0) + ) + } + + "MIN aggregation merge" should "stay neutral when one side saw no values" in { + val agg = op(AggregationFunction.MIN).getAggFunc(AttributeType.INTEGER) + val seen = agg.iterate(agg.init(), tupleOf("v", AttributeType.INTEGER, Int.box(7))) + + // An empty partial is the sentinel maxValue, so it must lose the comparison. + assert(agg.finalAgg(agg.merge(seen, agg.init())) == Int.box(7)) + assert(agg.finalAgg(agg.merge(agg.init(), seen)) == Int.box(7)) + // Two empty partials still finalize to null (no rows anywhere). + assert(agg.finalAgg(agg.merge(agg.init(), agg.init())) == null) + } + + // --- CONCAT: a null first value seeds the partial with an empty string ------ + + "CONCAT aggregation" should "swallow leading nulls but keep interior ones as empty slots" in { + // AggregateOpSpec only ever concatenates a null in the middle of the stream. + // A null on the very first tuple takes the `partial == ""` side of the branch + // and leaves the partial empty, so — unlike an interior null — it does not + // occupy a slot in the comma-joined output. + val agg = op(AggregationFunction.CONCAT).getAggFunc(AttributeType.STRING) + val result = Seq[AnyRef](null, "red", null, "blue") + .map(v => tupleOf("v", AttributeType.STRING, v)) + .foldLeft(agg.init())(agg.iterate) + + assert(agg.finalAgg(result) == "red,,blue") + } + + it should "return an empty string when every value is null" in { + val agg = op(AggregationFunction.CONCAT).getAggFunc(AttributeType.STRING) + val onlyNull = agg.iterate(agg.init(), tupleOf("v", AttributeType.STRING, null)) + + assert(agg.finalAgg(onlyNull) == "") + } + + it should "stringify non-string values it is pointed at" in { + // CONCAT is schema-restricted to STRING columns in the UI, but the executor + // only ever calls `.toString`, so an INTEGER column still concatenates. + val agg = op(AggregationFunction.CONCAT).getAggFunc(AttributeType.STRING) + val result = Seq[AnyRef](Int.box(1), Int.box(2)) + .map(v => tupleOf("v", AttributeType.INTEGER, v)) + .foldLeft(agg.init())(agg.iterate) + + assert(agg.finalAgg(result) == "1,2") + } + + // --- getAggregationAttribute / getFinal: message and identity guarantees ---- + + "getAggregationAttribute" should "name the unknown aggregation function in its error" in { + val ex = intercept[RuntimeException](op(null).getAggregationAttribute(AttributeType.INTEGER)) + assert(ex.getMessage == "Unknown aggregation function: null") + } + + "getFinal" should "produce a detached copy that re-reads the result column" in { + val original = op(AggregationFunction.MAX, attribute = "src", resultAttribute = "dst") + val copy = original.getFinal + + assert(copy ne original) + assert(copy.aggFunction == AggregationFunction.MAX) + // the final stage reads and writes the same (result) column + assert(copy.attribute == "dst") + assert(copy.resultAttribute == "dst") + // the original is left untouched + assert(original.attribute == "src") + } } 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 231c2c3278..e683c7b4b5 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 @@ -145,4 +145,69 @@ class FilterPredicateSpec extends AnyFlatSpec with Matchers { p.equals(null) shouldBe false p.equals("not a predicate") shouldBe false } + + // --- attribute types that no other test reaches ---------------------------- + + it should "route ANY columns through the string comparison path" in { + // ANY shares the STRING case of the type switch: the field is stringified and + // then compared numerically when both sides parse. + val t = singleFieldTuple(AttributeType.ANY, java.lang.Integer.valueOf(42)) + new FilterPredicate("col", ComparisonType.GREATER_THAN, "9").evaluate(t) shouldBe true + new FilterPredicate("col", ComparisonType.EQUAL_TO, "42").evaluate(t) shouldBe true + new FilterPredicate("col", ComparisonType.LESS_THAN, "9").evaluate(t) shouldBe false + } + + it should "fall back to lexicographic comparison when an ANY field is not numeric" in { + val t = singleFieldTuple(AttributeType.ANY, java.lang.Boolean.TRUE) + new FilterPredicate("col", ComparisonType.EQUAL_TO, "true").evaluate(t) shouldBe true + new FilterPredicate("col", ComparisonType.NOT_EQUAL_TO, "false").evaluate(t) shouldBe true + } + + it should "reject attribute types it cannot compare" in { + val t = singleFieldTuple(AttributeType.BINARY, Array[Byte](1, 2, 3)) + val ex = intercept[RuntimeException] { + new FilterPredicate("col", ComparisonType.EQUAL_TO, "1").evaluate(t) + } + ex.getMessage shouldBe "unsupported attribute type: binary" + } + + it should "still answer the null checks on an otherwise unsupported type" in { + // IS_NULL / IS_NOT_NULL short-circuit before the type switch, so a BINARY + // column is filterable for nullness even though it cannot be compared. + val t = singleFieldTuple(AttributeType.BINARY, Array[Byte](1)) + new FilterPredicate("col", ComparisonType.IS_NOT_NULL, null).evaluate(t) shouldBe true + new FilterPredicate("col", ComparisonType.IS_NULL, null).evaluate(t) shouldBe false + } + + // --- value-side parsing ---------------------------------------------------- + + it should "compare a numeric string field lexicographically when the value is not numeric" in { + // The tuple side parses as a number but the user-supplied value does not, so + // the numeric attempt aborts and both sides are compared as text. + val t = singleFieldTuple(AttributeType.STRING, "10") + new FilterPredicate("col", ComparisonType.LESS_THAN, "abc").evaluate(t) shouldBe true + new FilterPredicate("col", ComparisonType.EQUAL_TO, "abc").evaluate(t) shouldBe false + } + + it should "propagate a parse failure when the value cannot be read as the column's type" in { + val doubleTuple = singleFieldTuple(AttributeType.DOUBLE, java.lang.Double.valueOf(1.0)) + intercept[NumberFormatException] { + new FilterPredicate("col", ComparisonType.EQUAL_TO, "not-a-number").evaluate(doubleTuple) + } + val longTuple = singleFieldTuple(AttributeType.LONG, java.lang.Long.valueOf(1L)) + intercept[NumberFormatException] { + new FilterPredicate("col", ComparisonType.EQUAL_TO, "1.5").evaluate(longTuple) + } + } + + it should "trim surrounding whitespace off the value for boolean, long and timestamp columns" in { + new FilterPredicate("col", ComparisonType.EQUAL_TO, " TrUe ") + .evaluate(singleFieldTuple(AttributeType.BOOLEAN, java.lang.Boolean.TRUE)) shouldBe true + new FilterPredicate("col", ComparisonType.EQUAL_TO, " 100 ") + .evaluate(singleFieldTuple(AttributeType.LONG, java.lang.Long.valueOf(100L))) shouldBe true + new FilterPredicate("col", ComparisonType.LESS_THAN, " 2021-01-01 00:00:00 ") + .evaluate( + singleFieldTuple(AttributeType.TIMESTAMP, Timestamp.valueOf("2020-01-01 00:00:00")) + ) shouldBe true + } } diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/metadata/OPVersionSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/metadata/OPVersionSpec.scala new file mode 100644 index 0000000000..8ca517a7de --- /dev/null +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/metadata/OPVersionSpec.scala @@ -0,0 +1,156 @@ +/* + * 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.metadata + +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +import java.util.UUID + +/** + * `OPVersion` resolves an operator's version from the git history of the file + * that defines it, memoizing the answer in a process-wide static map. + * + * Its git handle is opened once in a static initializer against `TEXERA_HOME` + * (defaulting to the working directory). Whether that succeeds depends entirely + * on how the tree was checked out — inside a `git worktree` the `.git` entry is + * a file rather than a directory and jgit raises `RepositoryNotFoundException`, + * leaving the handle null; in a plain clone the handle opens and an unknown path + * yields an empty log instead. Rather than assert whatever both happen to share, + * this spec pins the handle to a known state: every test forces the private + * static `git` field to null for its duration and restores the original value + * afterwards, so the assertions describe one deterministic code path regardless + * of how the tree was checked out (and no other suite in the JVM is affected). + * + * With the handle null, `git.log()` throws `NullPointerException` and resolution + * takes the `"N/A"` fallback. On top of that this spec pins the memoization + * contract: the answer is cached per operator name and the path is ignored on a + * cache hit. + * + * Deliberately NOT covered: the success path that returns a real commit hash and + * the `GitAPIException` catch (which, note, leaves `opMap` unpopulated and so + * returns null). Both require a specific, openable repository state that is not + * guaranteed for a test run. + */ +class OPVersionSpec extends AnyFlatSpec with Matchers { + + /** The cache is static and shared, so every test uses a name nothing else can collide with. */ + private def uniqueName(): String = s"OPVersionSpec-${UUID.randomUUID()}" + + private def uniqueMissingPath(): String = s"no/such/operator/path/${UUID.randomUUID()}" + + private def declaredField(name: String): java.lang.reflect.Field = { + val field = classOf[OPVersion].getDeclaredField(name) + field.setAccessible(true) + field + } + + /** The private static memo table, so tests can seed it and clean up after themselves. */ + private def opMap: java.util.Map[String, String] = { + declaredField("opMap").get(null).asInstanceOf[java.util.Map[String, String]] + } + + /** + * Runs `body` with `OPVersion`'s private static git handle forced to null, so the + * fallback path is exercised deterministically, then restores whatever was there. + */ + private def withNullGit[T](body: => T): T = { + val field = declaredField("git") + val original = field.get(null) + field.set(null, null) + try body + finally field.set(null, original) + } + + private def withCleanCache[T](names: String*)(body: => T): T = + try body + finally names.foreach(opMap.remove) + + "OPVersion.getVersion" should "fall back to \"N/A\" when the git handle is unavailable" in { + val name = uniqueName() + withCleanCache(name) { + withNullGit { + OPVersion.getVersion(name, uniqueMissingPath()) shouldBe "N/A" + } + } + } + + it should "never return null, whatever it resolves" in { + val name = uniqueName() + withCleanCache(name) { + withNullGit { + OPVersion.getVersion(name, "common/workflow-operator/src/main/scala") should not be null + } + } + } + + it should "memoize the resolved version under the operator name" in { + val name = uniqueName() + withCleanCache(name) { + withNullGit { + opMap.containsKey(name) shouldBe false + val resolved = OPVersion.getVersion(name, uniqueMissingPath()) + opMap.containsKey(name) shouldBe true + opMap.get(name) shouldBe resolved + } + } + } + + it should "serve the memoized value and ignore the path on subsequent calls" in { + val name = uniqueName() + withCleanCache(name) { + withNullGit { + // Seed a value no lookup could ever produce: if the cache were bypassed, + // the call below would recompute and answer "N/A" instead. + opMap.put(name, "seeded-version") + OPVersion.getVersion(name, uniqueMissingPath()) shouldBe "seeded-version" + OPVersion.getVersion(name, "a/completely/different/path") shouldBe "seeded-version" + } + } + } + + it should "key the cache by operator name rather than by path" in { + val first = uniqueName() + val second = uniqueName() + withCleanCache(first, second) { + withNullGit { + opMap.put(first, "version-of-first") + opMap.put(second, "version-of-second") + + val sharedPath = "common/workflow-operator/src/main/scala" + OPVersion.getVersion(first, sharedPath) shouldBe "version-of-first" + OPVersion.getVersion(second, sharedPath) shouldBe "version-of-second" + } + } + } + + it should "resolve unknown paths consistently across repeated calls" in { + val name = uniqueName() + withCleanCache(name) { + withNullGit { + val path = uniqueMissingPath() + val first = OPVersion.getVersion(name, path) + val second = OPVersion.getVersion(name, path) + first shouldBe "N/A" + second shouldBe first + } + } + } +} diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/timeSeriesPlot/TimeSeriesOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/timeSeriesPlot/TimeSeriesOpDescSpec.scala index ba6d6e6ccb..75478edf23 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/timeSeriesPlot/TimeSeriesOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/timeSeriesPlot/TimeSeriesOpDescSpec.scala @@ -19,9 +19,17 @@ package org.apache.texera.amber.operator.timeSeriesPlot +import org.apache.texera.amber.core.tuple.{AttributeType, Schema} +import org.apache.texera.amber.core.workflow.PortIdentity +import org.apache.texera.amber.operator.LogicalOp +import org.apache.texera.amber.operator.metadata.OperatorGroupConstants import org.apache.texera.amber.operator.visualization.timeSeriesplot.TimeSeriesOpDesc +import org.apache.texera.amber.util.JSONUtils.objectMapper import org.scalatest.funsuite.AnyFunSuite +import java.nio.charset.StandardCharsets +import java.util.Base64 + class TimeSeriesOpDescSpec extends AnyFunSuite { test("generatePythonCode returns non-empty python code") { @@ -41,4 +49,193 @@ class TimeSeriesOpDescSpec extends AnyFunSuite { assert(py.contains("class ProcessTableOperator")) assert(py.contains("def process_table")) } + + // --- helpers --------------------------------------------------------------- + + /** Column names reach the emitted Python base64-encoded inside a decode call. */ + private val decodeSite = "self.decode_python_template" + + private def b64(s: String): String = + Base64.getEncoder.encodeToString(s.getBytes(StandardCharsets.UTF_8)) + + private def decodeSiteCount(s: String): Int = + s.split(java.util.regex.Pattern.quote(decodeSite), -1).length - 1 + + private def lineContaining(code: String, marker: String): String = + code.linesIterator + .find(_.contains(marker)) + .getOrElse(fail(s"no line containing '$marker' in:\n$code")) + + /** + * The bracketed `dropna(subset=[...])` list only. The surrounding line also + * carries a `sort_values(by=...)` splice, so it cannot be counted wholesale. + * Base64 payloads never contain ']', so the first ']' closes the list. + */ + private def dropnaSubset(code: String): String = { + val line = lineContaining(code, "table.dropna(subset=[") + val start = line.indexOf("subset=[") + "subset=[".length + line.substring(start, line.indexOf(']', start)) + } + + /** Minimally configured operator: both optional columns left at "No Selection". */ + private def minimalOp(): TimeSeriesOpDesc = { + val op = new TimeSeriesOpDesc + op.timeColumn = "date" + op.valueColumn = "value" + op + } + + // --- operator metadata ------------------------------------------------------ + + test("operatorInfo advertises the plot name, basic viz group, and a 1-in/1-out shape") { + val info = minimalOp().operatorInfo + assert(info.userFriendlyName == "Time Series Plot") + assert(info.operatorDescription == "Visualize trends and patterns over time.") + assert(info.operatorGroupName == OperatorGroupConstants.VISUALIZATION_BASIC_GROUP) + assert(info.inputPorts.length == 1) + assert(info.outputPorts.length == 1) + } + + test("getOutputSchemas emits one html-content STRING column on the declared output port") { + val op = minimalOp() + val inputPort = op.operatorInfo.inputPorts.head.id + val out = op.getOutputSchemas( + Map(inputPort -> Schema().add("date", AttributeType.TIMESTAMP)) + ) + + assert(out.keySet == Set(op.operatorInfo.outputPorts.head.id)) + val schema = out(op.operatorInfo.outputPorts.head.id) + assert(schema.getAttributeNames == List("html-content")) + assert(schema.getAttribute("html-content").getType == AttributeType.STRING) + } + + test("getOutputSchemas ignores the input schema (the plot output shape is fixed)") { + val op = minimalOp() + val fromEmpty = op.getOutputSchemas(Map.empty[PortIdentity, Schema]) + val fromPopulated = + op.getOutputSchemas(Map(PortIdentity(3) -> Schema().add("x", AttributeType.INTEGER))) + assert(fromEmpty == fromPopulated) + } + + // --- optional columns ------------------------------------------------------- + + test("the default 'No Selection' optional columns add neither color, facet, nor dropna entries") { + val py = minimalOp().generatePythonCode() + + assert(!py.contains(", color=")) + assert(!py.contains(", facet_col=")) + // only timeColumn and valueColumn are required to be non-null + assert(decodeSiteCount(dropnaSubset(py)) == 2) + } + + test("a configured category column adds a color argument and a dropna entry") { + val op = minimalOp() + op.CategoryColumn = "region" + val py = op.generatePythonCode() + + assert(py.contains(", color=")) + assert(!py.contains(", facet_col=")) + assert(py.contains(b64("region"))) + assert(decodeSiteCount(dropnaSubset(py)) == 3) + } + + test("a configured facet column adds a facet_col argument and a dropna entry") { + val op = minimalOp() + op.CategoryColumn = "region" + op.facetColumn = "store" + val py = op.generatePythonCode() + + assert(py.contains(", color=")) + assert(py.contains(", facet_col=")) + assert(py.contains(b64("store"))) + assert(decodeSiteCount(dropnaSubset(py)) == 4) + } + + // --- plot type and range slider --------------------------------------------- + + test("plotType selects px.line by default and px.area when asked for an area chart") { + assert(minimalOp().generatePythonCode().contains("fig = px.line(table")) + + val areaOp = minimalOp() + areaOp.plotType = "area" + val areaPy = areaOp.generatePythonCode() + assert(areaPy.contains("fig = px.area(table")) + assert(!areaPy.contains("px.line(")) + + // any unrecognized plot type falls back to a line chart + val oddOp = minimalOp() + oddOp.plotType = "bar" + assert(oddOp.generatePythonCode().contains("fig = px.line(table")) + } + + test("showRangeSlider gates the rangeslider_visible call") { + val off = minimalOp() + off.showRangeSlider = false + val offPy = off.generatePythonCode() + assert(offPy.contains("if False:")) + assert(!offPy.contains("if True:")) + + val on = minimalOp() + on.showRangeSlider = true + val onPy = on.generatePythonCode() + assert(onPy.contains("if True:")) + assert(onPy.contains("fig.update_xaxes(rangeslider_visible=True)")) + } + + test("showRangeSlider defaults to false on a freshly constructed descriptor") { + assert(!new TimeSeriesOpDesc().showRangeSlider) + } + + // --- generated code shape ---------------------------------------------------- + + test("generated code coerces the time column, sorts by it, and guards both empty cases") { + val py = minimalOp().generatePythonCode() + + assert(py.contains("pd.to_datetime(")) + assert(py.contains("errors='coerce'")) + assert(py.contains(".sort_values(by=")) + assert(py.contains("Input table is empty.")) + assert(py.contains("Table became empty after filtering.")) + assert(py.contains("except Exception as e:")) + assert(py.contains("plotly.io.to_html(fig, include_plotlyjs='cdn', full_html=False)")) + // column names are encoded, never spliced raw + assert(py.contains(b64("date")) && py.contains(b64("value"))) + } + + // --- JSON round-trip --------------------------------------------------------- + + test("the descriptor round-trips all of its config fields through the polymorphic base") { + val op = new TimeSeriesOpDesc + op.timeColumn = "date" + op.valueColumn = "sales" + op.CategoryColumn = "region" + op.facetColumn = "store" + op.plotType = "area" + op.showRangeSlider = true + + val restored = objectMapper.readValue(objectMapper.writeValueAsString(op), classOf[LogicalOp]) + + assert(restored.isInstanceOf[TimeSeriesOpDesc]) + val d = restored.asInstanceOf[TimeSeriesOpDesc] + assert(d.timeColumn == "date") + assert(d.valueColumn == "sales") + assert(d.CategoryColumn == "region") + assert(d.facetColumn == "store") + assert(d.plotType == "area") + assert(d.showRangeSlider) + } + + test("a minimal JSON payload deserializes with the documented defaults") { + val json = + """{"operatorType":"TimeSeriesPlot","operatorID":"TimeSeriesPlot-1",""" + + """"timeColumn":"date","valueColumn":"sales"}""" + val d = objectMapper.readValue(json, classOf[LogicalOp]).asInstanceOf[TimeSeriesOpDesc] + + assert(d.timeColumn == "date") + assert(d.valueColumn == "sales") + assert(d.CategoryColumn == "No Selection") + assert(d.facetColumn == "No Selection") + assert(d.plotType == "line") + assert(!d.showRangeSlider) + } } diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/visualization/ecdfPlot/ECDFPlotOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/visualization/ecdfPlot/ECDFPlotOpDescSpec.scala index 8bded2f414..b1fe949935 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/visualization/ecdfPlot/ECDFPlotOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/visualization/ecdfPlot/ECDFPlotOpDescSpec.scala @@ -19,6 +19,11 @@ package org.apache.texera.amber.operator.visualization.ecdfPlot +import org.apache.texera.amber.core.tuple.{AttributeType, Schema} +import org.apache.texera.amber.core.workflow.PortIdentity +import org.apache.texera.amber.operator.LogicalOp +import org.apache.texera.amber.operator.metadata.OperatorGroupConstants +import org.apache.texera.amber.util.JSONUtils.objectMapper import org.scalatest.BeforeAndAfter import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers @@ -97,4 +102,197 @@ class ECDFPlotOpDescSpec extends AnyFlatSpec with BeforeAndAfter with Matchers { val code = opDesc.generatePythonCode() assert(carries(code, "ecdf_value_col")) } + + // --- helpers for the argument-shape assertions ----------------------------- + + // Every column name spliced into the emitted Python becomes a decode site, so + // counting them on one line tells us how many columns that line references. + private val decodeSite = "self.decode_python_template" + + private def decodeSiteCount(s: String): Int = + s.split(java.util.regex.Pattern.quote(decodeSite), -1).length - 1 + + private def lineContaining(code: String, marker: String): String = + code.linesIterator + .find(_.contains(marker)) + .getOrElse(fail(s"no line containing '$marker' in:\n$code")) + + // --- operator metadata ------------------------------------------------------ + + "ECDFPlotOpDesc.operatorInfo" should + "advertise the visualization name, group, and a 1-in/1-out shape" in { + val info = opDesc.operatorInfo + info.userFriendlyName shouldBe "Empirical Cumulative Distribution Plot" + info.operatorDescription shouldBe + "Visualize the empirical cumulative distribution of a numeric column." + info.operatorGroupName shouldBe OperatorGroupConstants.VISUALIZATION_STATISTICAL_GROUP + info.inputPorts should have length 1 + info.outputPorts should have length 1 + } + + "ECDFPlotOpDesc.getOutputSchemas" should + "emit a single html-content STRING column keyed by the declared output port" in { + opDesc.valueColumn = "score" + val inputPort = opDesc.operatorInfo.inputPorts.head.id + val out = opDesc.getOutputSchemas( + Map(inputPort -> Schema().add("score", AttributeType.DOUBLE)) + ) + + out.keySet shouldBe Set(opDesc.operatorInfo.outputPorts.head.id) + val schema = out(opDesc.operatorInfo.outputPorts.head.id) + schema.getAttributeNames should contain theSameElementsAs List("html-content") + schema.getAttribute("html-content").getType shouldBe AttributeType.STRING + } + + it should "ignore the input schema entirely (the plot output shape is fixed)" in { + opDesc.valueColumn = "score" + val fromEmpty = opDesc.getOutputSchemas(Map.empty[PortIdentity, Schema]) + val fromPopulated = opDesc.getOutputSchemas( + Map(PortIdentity(7) -> Schema().add("anything", AttributeType.INTEGER)) + ) + fromEmpty shouldBe fromPopulated + } + + // --- createPlotlyFigure: the argument list it assembles --------------------- + + "ECDFPlotOpDesc.createPlotlyFigure" should + "emit only the value column when every optional setting is left at its default" in { + opDesc.valueColumn = "score" + val call = lineContaining(opDesc.createPlotlyFigure().plain, "px.ecdf(") + + decodeSiteCount(call) shouldBe 1 + call should include(s"px.ecdf(table, x=$decodeSite") + call should not include "ecdfnorm" + call should not include ", y=" + call should not include "color=" + call should not include "facet_col=" + call should not include "ecdfmode=" + call should not include "orientation=" + call should not include "markers=True" + call should not include "marginal=" + } + + it should "request a cumulative sum by turning off normalization and passing y" in { + opDesc.valueColumn = "score" + opDesc.yAxisMode = "sum" + val call = lineContaining(opDesc.createPlotlyFigure().plain, "px.ecdf(") + + call should include("ecdfnorm=None") + call should include(s", y=$decodeSite") + // the value column is spliced twice: once as x, once as y + decodeSiteCount(call) shouldBe 2 + } + + it should "turn off normalization without a y argument for raw counts" in { + opDesc.valueColumn = "score" + opDesc.yAxisMode = "count" + val call = lineContaining(opDesc.createPlotlyFigure().plain, "px.ecdf(") + + call should include("ecdfnorm=None") + call should not include ", y=" + decodeSiteCount(call) shouldBe 1 + } + + it should "omit ecdfmode, orientation and marginal when they hold their default values" in { + opDesc.valueColumn = "score" + opDesc.cdfMode = "standard" + opDesc.orientation = "vertical" + opDesc.marginal = "none" + opDesc.showMarkers = false + val call = lineContaining(opDesc.createPlotlyFigure().plain, "px.ecdf(") + + call should not include "ecdfmode=" + call should not include "orientation=" + call should not include "marginal=" + call should not include "markers=" + } + + // --- manipulateTable: the dropna column list ------------------------------- + + "ECDFPlotOpDesc.manipulateTable" should + "require only the value column when no optional column is configured" in { + opDesc.valueColumn = "score" + val required = lineContaining(opDesc.manipulateTable().plain, "required_cols = [") + + decodeSiteCount(required) shouldBe 1 + assert(carries(required, "score")) + } + + it should "require the color and separate-by columns too when they are configured" in { + opDesc.valueColumn = "score" + opDesc.colorColumn = "group" + opDesc.separateBy = "category" + val required = lineContaining(opDesc.manipulateTable().plain, "required_cols = [") + + decodeSiteCount(required) shouldBe 3 + assert(carries(required, "score")) + assert(carries(required, "group")) + assert(carries(required, "category")) + } + + it should "coerce the value column to numeric and drop the rows that fail" in { + opDesc.valueColumn = "score" + val table = opDesc.manipulateTable().plain + + table should include("pd.to_numeric(") + table should include("errors='coerce'") + table should include("inplace=True") + } + + // --- generated code --------------------------------------------------------- + + "ECDFPlotOpDesc.generatePythonCode" should + "wrap the figure in a table operator that guards both empty-table cases" in { + opDesc.valueColumn = "score" + val code = opDesc.generatePythonCode() + + code should include("class ProcessTableOperator(UDFTableOperator)") + code should include("def process_table") + code should include("input table is empty.") + code should include("no valid rows left after removing missing or non-numeric values.") + code should include("plotly.io.to_html(fig, include_plotlyjs='cdn', auto_play=False)") + code should include("yield {'html-content': html}") + } + + // --- JSON round-trip -------------------------------------------------------- + + "ECDFPlotOpDesc" should "round-trip all of its config fields through the polymorphic base" in { + opDesc.valueColumn = "score" + opDesc.colorColumn = "group" + opDesc.separateBy = "category" + opDesc.yAxisMode = "sum" + opDesc.cdfMode = "complementary" + opDesc.orientation = "horizontal" + opDesc.showMarkers = true + opDesc.marginal = "rug" + + val restored = + objectMapper.readValue(objectMapper.writeValueAsString(opDesc), classOf[LogicalOp]) + + restored shouldBe a[ECDFPlotOpDesc] + val d = restored.asInstanceOf[ECDFPlotOpDesc] + d.valueColumn shouldBe "score" + d.colorColumn shouldBe "group" + d.separateBy shouldBe "category" + d.yAxisMode shouldBe "sum" + d.cdfMode shouldBe "complementary" + d.orientation shouldBe "horizontal" + d.showMarkers shouldBe true + d.marginal shouldBe "rug" + } + + it should "keep its documented defaults when deserialized from a minimal JSON payload" in { + val json = + """{"operatorType":"ECDFPlot","operatorID":"ECDFPlot-1","valueColumn":"score"}""" + val d = objectMapper.readValue(json, classOf[LogicalOp]).asInstanceOf[ECDFPlotOpDesc] + + d.valueColumn shouldBe "score" + d.colorColumn shouldBe "" + d.separateBy shouldBe "" + d.yAxisMode shouldBe "probability" + d.cdfMode shouldBe "standard" + d.orientation shouldBe "vertical" + d.showMarkers shouldBe false + d.marginal shouldBe "none" + } }
