test(workflow-operator): extend operator descriptor and predicate coverage

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")
+  }
 }

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
+  }
 }

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
+      }
+    }
+  }
+}