test(verify): run the source operators, and report what is covered

.jvmopts

@@ -26,3 +26,12 @@
 
 # Required by Apache Pekko: https://pekko.apache.org/docs/pekko/snapshot/release-notes/releases-2.0.html
 --add-opens=java.base/jdk.internal.misc=ALL-UNNAMED
+
+# Kryo again, reached only once a suite that serializes a timestamp runs in the
+# same JVM as one that has already forked a Python worker, which is what running
+# this module's tests without WCS_TEST_FILTER does. The two CI jobs each take
+# half, so CI never reaches it and a local full-module run does.
+--add-opens=java.base/sun.util.calendar=ALL-UNNAMED
+--add-opens=java.base/java.time=ALL-UNNAMED
+--add-opens=java.base/java.time.zone=ALL-UNNAMED
+--add-opens=java.base/java.util.concurrent=ALL-UNNAMED

workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CanonicalFixture.scala

@@ -138,7 +138,7 @@ object CanonicalFixture extends SharedFixture {
   // etc.), so a hand-edit that breaks one fails the build. `schema` above stays
   // authoritative for column types: JSON has no TIMESTAMP, so the timestamps
   // are stored as JDBC strings ("2024-01-01 00:00:00.0") and coerced back here.
-  private val fixtureResource = "/verify/canonical_fixture.json"
+  private[verify] val fixtureResource = "/verify/canonical_fixture.json"
 
   override val allRows: Vector[Tuple] = {
     val stream = Option(getClass.getResourceAsStream(fixtureResource))

workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/ConfigCoverageSpec.scala

@@ -0,0 +1,72 @@
+/*
+ * 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.translator.verify
+
+import org.apache.texera.amber.operator.LogicalOp
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+/**
+  * Holds the floor under what the harness can run.
+  *
+  * Every operator carrying standalone code is classified runnable or flagged, and
+  * a flag is how an operator stops being checked at all. For most that is a state
+  * someone is working through. For the ones below it would be a regression: they
+  * are the plainest operators the harness has, they take the shared table as it
+  * is, and nothing about them is hard to run. An operator arriving here means
+  * something upstream of the disposition broke rather than that this operator
+  * became difficult.
+  */
+class ConfigCoverageSpec extends AnyFlatSpec with Matchers {
+
+  private val mustRun = Set(
+    "IntersectOpDesc",
+    "DifferenceOpDesc",
+    "SymmetricDifferenceOpDesc",
+    "HashJoinOpDesc",
+    "SpecializedFilterOpDesc",
+    "SortOpDesc",
+    "LimitOpDesc"
+  )
+
+  "the harness" should "keep runnable every operator it must be able to run" in {
+    val discovered = OperatorBehaviorSpec
+      .discoverStandaloneOperators()
+      .filter(opClass => mustRun.contains(opClass.getSimpleName))
+    // Losing standalone support drops an operator from discovery, which would
+    // otherwise pass the runnable check below without it ever being asked.
+    val missing = mustRun -- discovered.map(_.getSimpleName)
+    withClue(s"must-run operators no longer discovered: $missing") {
+      missing shouldBe empty
+    }
+    val flagged = discovered.filterNot(runnable).map(_.getSimpleName)
+    withClue(s"must-run operators no longer runnable: $flagged") {
+      flagged shouldBe empty
+    }
+  }
+
+  private def runnable(opClass: Class[_ <: LogicalOp]): Boolean =
+    if (SourceCategoryRunner.runsAsSource(opClass))
+      SourceCategoryRunner.canRun(opClass)
+    else
+      TransformVerificationRunner
+        .disposition(opClass)
+        .isInstanceOf[TransformVerificationRunner.Runnable]
+}

workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/OperatorBehaviorSpec.scala

@@ -0,0 +1,140 @@
+/*
+ * 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.translator.verify
+
+import com.fasterxml.jackson.annotation.JsonSubTypes
+import org.apache.texera.amber.operator.{LogicalOp, StandaloneCodeGenerator}
+import org.apache.texera.amber.translator.verify.tags.IntegrationTest
+import org.scalatest.ParallelTestExecution
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+/**
+  * A parity test per operator: every one registered in [[LogicalOp]]'s
+  * `@JsonSubTypes` that implements [[StandaloneCodeGenerator]] runs through the
+  * Texera executor and through its generated Python, and the two outputs are
+  * compared. Adding an operator needs no edit here.
+  *
+  * [[TransformVerificationRunner]] configures each non-source transform and
+  * decides whether it can run; sources go to [[SourceCategoryRunner]]. Every
+  * test's NAME carries its verdict, the tier that ran it or the reason it could
+  * not, so the report lists every operator and says what it did not check.
+  *
+  * Requires Python 3 with pandas on the [[Comparator]] / [[StandaloneRunner]]
+  * resolution chain (`UDF_PYTHON_PATH` env var, then `python3.12`).
+  */
+// Tagged @IntegrationTest: it forks Python, so CI routes it to the
+// Python-provisioned integration job (see workflow-compiling-service/build.sbt
+// WCS_TEST_FILTER).
+@IntegrationTest
+class OperatorBehaviorSpec extends AnyFlatSpec with Matchers with ParallelTestExecution {
+
+  // The test list is built at class construction, one test per operator.
+  OperatorBehaviorSpec.discoverStandaloneOperators().foreach { opClass =>
+    val name = opClass.getSimpleName
+
+    if (!OperatorBehaviorSpec.isSelected(name)) {
+      // Only a local run sets those, and CI therefore runs the lot.
+      name should "NARROWED OUT — outside this run's VERIFY_ONLY / VERIFY_SKIP" ignore {}
+    } else if (SourceCategoryRunner.runsAsSource(opClass)) {
+      // A source has no input port, so its own runner writes the shared table
+      // into the file format each one reads.
+      if (SourceCategoryRunner.canRun(opClass)) {
+        name should "produce equivalent output in Texera and standalone Python (source)" in {
+          SourceCategoryRunner.run(opClass)
+        }
+      } else {
+        name should s"FLAGGED — ${SourceCategoryRunner.flagReason(opClass)}" ignore {}
+      }
+    } else {
+      TransformVerificationRunner.disposition(opClass) match {
+        case TransformVerificationRunner.Runnable(tier) =>
+          name should s"produce equivalent output in Texera and standalone Python ($tier)" in {
+            TransformVerificationRunner.run(opClass)
+          }
+        case TransformVerificationRunner.Flagged(reason) =>
+          // ConfigCoverageSpec aggregates these into its table.
+          name should s"FLAGGED — $reason" ignore {}
+      }
+    }
+  }
+
+  // Not one test per operator like the rest of this spec: it is one assertion
+  // over all of them, and it deliberately ignores the selection knobs above so a
+  // VERIFY_ONLY run still cannot hide a broken splice site.
+  "Generated standalone code" should "stay parseable when the column names are hostile" in {
+    StandaloneEscapingCheck.run() shouldBe empty
+  }
+
+  // Also one assertion over all of them: a workflow that draws two charts from
+  // one upstream hands both the same variable, and only a plan with a branch
+  // ever notices an operator writing to it.
+  it should "leave the frame it was handed alone" in {
+    StandaloneInputCheck.run() shouldBe empty
+  }
+}
+
+object OperatorBehaviorSpec {
+
+  // Case-sensitive substrings of the operator's simple name, comma-separated.
+  //
+  // There is deliberately no third list withholding operators by default: a
+  // name here would withdraw an operator's every variant and record nothing
+  // about what is wrong with it. Withholding lives where it can say why, in
+  // [[TransformVerificationRunner]]'s `variantsNotRun` or its `knownIssues`.
+  private def patterns(envVar: String): Seq[String] =
+    sys.env.getOrElse(envVar, "").split(",").iterator.map(_.trim).filter(_.nonEmpty).toSeq
+
+  private lazy val onlyPatterns: Seq[String] = patterns("VERIFY_ONLY")
+  private lazy val skipPatterns: Seq[String] = patterns("VERIFY_SKIP")
+
+  /** True if `name` should run: in VERIFY_ONLY when that is set, and not in
+    * VERIFY_SKIP. True for everything when neither is set.
+    */
+  def isSelected(name: String): Boolean = {
+    val included = onlyPatterns.isEmpty || onlyPatterns.exists(name.contains)
+    val excluded = skipPatterns.exists(name.contains)
+    included && !excluded
+  }
+
+  /**
+    * Enumerates every concrete subclass of [[LogicalOp]] declared in its
+    * `@JsonSubTypes` annotation, filters to those implementing
+    * [[StandaloneCodeGenerator]], and returns them sorted by simple name
+    * (stable test report order).
+    *
+    * Uses the same registry Jackson uses to deserialize operators — no
+    * separate discovery mechanism needed. Adding an operator to
+    * `LogicalOp.@JsonSubTypes` makes it visible here automatically.
+    */
+  def discoverStandaloneOperators(): Seq[Class[_ <: LogicalOp]] = {
+    val annotation = classOf[LogicalOp].getAnnotation(classOf[JsonSubTypes])
+    if (annotation == null) Seq.empty
+    else
+      annotation
+        .value()
+        .toSeq
+        .map(_.value())
+        .filter(classOf[StandaloneCodeGenerator].isAssignableFrom)
+        .map(_.asInstanceOf[Class[_ <: LogicalOp]])
+        .distinct
+        .sortBy(_.getSimpleName)
+  }
+}