You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@texera.apache.org by gi...@apache.org on 2026/07/29 06:34:33 UTC
(texera) 01/01: test(workflow-operator): extend operator descriptor and predicate coverage (#7018)
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-a3d12b02201e32fff70270fd48ed25abb86ad8af
in repository https://gitbox.apache.org/repos/asf/texera.git
commit 123b9f9b8ba6f571810543653be8af267e659938
Author: Xinyuan Lin <xi...@uci.edu>
AuthorDate: Tue Jul 28 23:33:53 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"
+ }
}