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/30 00:24:08 UTC
(texera) branch gh-readonly-queue/main/pr-7018-88a70082948a13ba6f37dbcd2bd6e44ad6857b33 created (now 83849efea1)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-7018-88a70082948a13ba6f37dbcd2bd6e44ad6857b33
in repository https://gitbox.apache.org/repos/asf/texera.git
at 83849efea1 test(workflow-operator): extend operator descriptor and predicate coverage (#7018)
This branch includes the following new commits:
new 83849efea1 test(workflow-operator): extend operator descriptor and predicate coverage (#7018)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(texera) 01/01: test(workflow-operator): extend operator descriptor and predicate coverage (#7018)
Posted by gi...@apache.org.
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-88a70082948a13ba6f37dbcd2bd6e44ad6857b33
in repository https://gitbox.apache.org/repos/asf/texera.git
commit 83849efea189afc303ba074b75cd8c13731fdcff
Author: Xinyuan Lin <xi...@uci.edu>
AuthorDate: Wed Jul 29 17:05:50 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"
+ }
}