From 9111b31630308bb22b23ce1209f3285a3ecb667c Mon Sep 17 00:00:00 2001
From: kary zheng <zhengj29@uci.edu>
Date: Wed, 2 Sep 2026 16:30:44 -0700
Subject: [PATCH 1/9] test(verify): compare what the two paths produced

Two frames agree when they hold the same rows in the same order with the same
values, and the comparison says which of those failed rather than that they
differ. A float is compared within a tolerance, since the two paths reach the
same number by different arithmetic.

A visualization has no frame to compare. What it emits is a figure, and the
meaning of a figure is in the numbers behind it, so the figures are compared
number by number; a plot that renders as an image is compared as the HTML it
emits instead.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../src/test/resources/python/compare.py      | 444 ++++++++++++++++++
 .../amber/translator/verify/Comparator.scala  | 189 ++++++++
 .../translator/verify/ComparatorSpec.scala    |  97 ++++
 .../verify/VisualizationHtmlComparator.scala  |  85 ++++
 .../verify/VisualizationJsonComparator.scala  | 123 +++++
 5 files changed, 938 insertions(+)
 create mode 100644 workflow-compiling-service/src/test/resources/python/compare.py
 create mode 100644 workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/Comparator.scala
 create mode 100644 workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/ComparatorSpec.scala
 create mode 100644 workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/VisualizationHtmlComparator.scala
 create mode 100644 workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/VisualizationJsonComparator.scala

diff --git a/workflow-compiling-service/src/test/resources/python/compare.py b/workflow-compiling-service/src/test/resources/python/compare.py
new file mode 100644
index 00000000000..518b5fac901
--- /dev/null
+++ b/workflow-compiling-service/src/test/resources/python/compare.py
@@ -0,0 +1,444 @@
+#!/usr/bin/env python3
+#
+# 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.
+"""
+Compare the two paths' outputs for one operator: JSONL DataFrames, or the Plotly
+figure a visualization operator renders.
+
+Usage: compare.py [--unordered] [--ignore-cols c1,c2]
+                  [--model-cols c1,c2 --probe features.jsonl]
+                  <actual.jsonl> <expected.jsonl>
+       compare.py --plotly <actual.jsonl> <expected.json>
+
+  --unordered   Sort both DataFrames lexicographically by all columns before
+                comparing, so rows match as a set/bag rather than positionally.
+                This is the norm: the engine runs operators across parallel
+                workers, so output row order is not part of the contract.
+                Without this flag the comparator matches rows positionally
+                (after reset_index(drop=True)) — used only for the sort family,
+                whose output order IS meaningful.
+
+  --ignore-cols Comma-separated column names to drop from both frames before
+                comparing. For opaque columns whose value isn't compared.
+
+  --model-cols  Comma-separated columns holding a base64(pickle) sklearn model.
+                Rather than byte-compare them (two independently-trained models
+                are functionally equal but not bit-identical), the comparator
+                unpickles both sides, has each model predict on the --probe
+                feature set, and asserts the predictions match — verifying the
+                two code paths produce behaviorally-equivalent models. The raw
+                model columns are then dropped before the frame comparison.
+
+  --probe       JSONL feature set the --model-cols models predict on. Each
+                model uses its own feature_names_in_ to select columns, so the
+                probe may include extra columns (e.g. the training target).
+
+  --plotly      Compare Plotly figures instead of DataFrames. The actual side is
+                a one-row JSONL with `html-content` or `json-content`; for
+                `html-content` the first `Plotly.newPlot(...)` payload is
+                extracted. The expected side is the standalone path's
+                `fig.write_json(...)`. Only data and layout are compared, with
+                display-only `uid` fields stripped and floats matched by
+                tolerance. Takes none of the DataFrame flags.
+
+Exit 0  - Outputs equal (and model predictions match, if --model-cols)
+Exit 1  - Outputs differ; detail on stderr
+Exit 2  - Bad invocation
+
+Persistent mode: `compare.py --serve` imports pandas once and then serves many
+comparisons over its lifetime, reading one JSON job per line on stdin and
+writing one JSON result per line on stdout. This avoids paying the ~214 ms
+pandas import on every comparison (the comparison itself is ~ms). It reuses the
+exact same functions the CLI calls, so behavior is identical.
+
+  request   {"kind": "dataframe", "actual": "<abs>", "expected": "<abs>",
+             "unordered": false, "ignoreCols": [], "modelCols": [],
+             "probe": null}\n
+            {"kind": "plotly", "actual": "<abs>", "expected": "<abs>"}\n
+  response  {"exit": 0|1, "stdout": "", "stderr": "<diff on mismatch>"}\n
+
+`kind` defaults to "dataframe". Both kinds are served by the same worker so a
+run needs one comparison pool rather than one per output shape; the Plotly side
+needs nothing pandas does not already pull in.
+
+A mismatch is exit 1 with the diff on `stderr`, mirroring the CLI's nonzero
+exit so the Scala side's ComparatorMismatchException path is unchanged. A
+comparison error never kills the server; only closing stdin (EOF) ends it.
+"""
+import sys
+
+# pandas is imported where it is used, not here: the --plotly comparison needs
+# nothing from it, and a module-level import would make that one-shot invocation
+# pay ~500 ms for an interpreter that then compares two JSON documents. `serve()`
+# imports it eagerly at startup instead, so a pooled worker still pays it once
+# rather than once per DataFrame comparison.
+
+
+def _compare_model_predictions(actual, expected, model_cols, probe_path) -> None:
+    """For each model column, unpickle both sides and assert their predictions
+    on the probe set match. Raises AssertionError on any divergence."""
+    import base64
+    import pickle
+
+    import numpy as np
+    import pandas as pd
+
+    if probe_path is None:
+        raise AssertionError("--model-cols requires --probe with a feature set")
+    probe = pd.read_json(probe_path, lines=True)
+    # The probe is the operator's own input table, so under the nulls scenario it
+    # carries the holes that scenario punched. What is under test is whether the
+    # two models agree, and an estimator that refuses a NaN at predict time would
+    # end the comparison over the probe rather than over either model. Drop those
+    # rows: both models are asked the same questions either way.
+    probe = probe.dropna()
+    if probe.empty:
+        raise AssertionError(
+            "probe has no complete row to predict on; the two models cannot be compared"
+        )
+
+    for col in model_cols:
+        if col not in actual.columns or col not in expected.columns:
+            continue
+        if len(actual) != len(expected):
+            raise AssertionError(
+                f"model column {col!r}: row count differs "
+                f"({len(actual)} vs {len(expected)})"
+            )
+        for i in range(len(actual)):
+            m_actual = pickle.loads(base64.b64decode(actual[col].iloc[i]))
+            m_expected = pickle.loads(base64.b64decode(expected[col].iloc[i]))
+
+            # A model with feature_names_in_ selects its (numeric) feature
+            # columns from the probe, naturally dropping the training target the
+            # probe may still carry. A model WITHOUT it was fitted on a 1-D input
+            # rather than a named frame — i.e. a text pipeline (e.g.
+            # CountVectorizer) trained on a single text Series — so feed the
+            # probe's first column as a Series, not the whole frame (predicting
+            # on a DataFrame would make CountVectorizer iterate column labels).
+            names = getattr(m_actual, "feature_names_in_", None)
+            x_a = probe[list(names)] if names is not None else probe.iloc[:, 0]
+            names_e = getattr(m_expected, "feature_names_in_", None)
+            x_e = probe[list(names_e)] if names_e is not None else probe.iloc[:, 0]
+
+            pred_a = np.asarray(m_actual.predict(x_a))
+            pred_e = np.asarray(m_expected.predict(x_e))
+
+            if pred_a.shape != pred_e.shape:
+                raise AssertionError(
+                    f"model column {col!r} row {i}: prediction shape differs "
+                    f"({pred_a.shape} vs {pred_e.shape})"
+                )
+            numeric = np.issubdtype(pred_a.dtype, np.number) and np.issubdtype(
+                pred_e.dtype, np.number
+            )
+            ok = (
+                np.allclose(pred_a, pred_e, rtol=1e-5, atol=1e-8)
+                if numeric
+                else np.array_equal(pred_a, pred_e)
+            )
+            if not ok:
+                raise AssertionError(
+                    f"model column {col!r} row {i}: predictions differ\n"
+                    f"  actual:   {pred_a}\n"
+                    f"  expected: {pred_e}"
+                )
+
+
+def _run_comparison(
+    actual_path: str,
+    expected_path: str,
+    unordered: bool,
+    ignore_cols: list,
+    model_cols: list,
+    probe_path,
+) -> "str | None":
+    """Compare two JSONL DataFrames. Returns None if they match, or a human
+    diff string if they differ (exit-1 condition). Unexpected errors (e.g. a
+    bad input file) propagate to the caller. This is the single source of
+    comparison truth shared by the CLI and the --serve loop."""
+    import pandas as pd
+
+    actual = pd.read_json(actual_path, lines=True)
+    expected = pd.read_json(expected_path, lines=True)
+
+    # Model columns: compare behavior (predictions) rather than bytes, then drop
+    # the raw columns so the frame comparison covers everything else exactly.
+    if model_cols:
+        try:
+            _compare_model_predictions(actual, expected, model_cols, probe_path)
+        except AssertionError as exc:
+            return str(exc)
+        actual = actual.drop(columns=model_cols, errors="ignore")
+        expected = expected.drop(columns=model_cols, errors="ignore")
+
+    if ignore_cols:
+        actual = actual.drop(columns=ignore_cols, errors="ignore")
+        expected = expected.drop(columns=ignore_cols, errors="ignore")
+
+    if unordered:
+        # Sort both sides by the same column key so set-equal frames collapse
+        # to the same row sequence. assert_frame_equal still does the actual
+        # value diff and respects rtol/check_dtype. Mergesort = stable, so
+        # rows that are tied on all columns keep their relative order — not
+        # strictly necessary for set equality (no ties → no duplicates after
+        # the op's dedup step) but cheap insurance.
+        cols = list(actual.columns)
+        if cols:
+            actual = actual.sort_values(
+                by=cols, kind="mergesort", na_position="last"
+            ).reset_index(drop=True)
+            expected = expected.sort_values(
+                by=cols, kind="mergesort", na_position="last"
+            ).reset_index(drop=True)
+
+    try:
+        pd.testing.assert_frame_equal(
+            actual,
+            expected,
+            check_like=True,
+            check_dtype=False,
+            rtol=1e-5,
+        )
+    except AssertionError as exc:
+        return str(exc)
+    return None
+
+
+def _load_actual_plot(path) -> dict:
+    import json
+
+    with open(path, "r", encoding="utf-8") as fh:
+        line = next((raw for raw in fh if raw.strip()), None)
+    if line is None:
+        raise AssertionError(f"{path} is empty")
+
+    row = json.loads(line)
+    if "json-content" in row and row["json-content"]:
+        value = row["json-content"]
+        return json.loads(value) if isinstance(value, str) else value
+    if "html-content" in row and row["html-content"]:
+        return _plotly_payload_from_html(row["html-content"])
+    raise AssertionError(f"{path} has neither html-content nor json-content")
+
+
+def _plotly_payload_from_html(html: str) -> dict:
+    """Pull the data/layout arguments out of the first Plotly.newPlot(...) call.
+
+    Scanned with a JSON decoder rather than a regex because the payload is
+    arbitrary nested JSON that no bracket-matching pattern handles reliably.
+    """
+    import json
+
+    marker = "Plotly.newPlot("
+    start = html.find(marker)
+    if start < 0:
+        raise AssertionError("html-content does not contain Plotly.newPlot(...)")
+
+    decoder = json.JSONDecoder()
+    index = start + len(marker)
+    args: list = []
+    while len(args) < 4:
+        while index < len(html) and html[index] in " \t\r\n,":
+            index += 1
+        value, consumed = decoder.raw_decode(html[index:])
+        args.append(value)
+        index += consumed
+
+    return {"data": args[1], "layout": args[2]}
+
+
+def _load_expected_plot(path) -> dict:
+    import json
+
+    with open(path, "r", encoding="utf-8") as fh:
+        value = json.load(fh)
+    return {"data": value.get("data", []), "layout": value.get("layout", {})}
+
+
+def _strip_unstable(value):
+    """Remove display-only fields that are unrelated to chart semantics."""
+    if isinstance(value, dict):
+        return {
+            key: _strip_unstable(child)
+            for key, child in value.items()
+            if key not in {"uid"}
+        }
+    if isinstance(value, list):
+        return [_strip_unstable(child) for child in value]
+    return value
+
+
+def _plots_equal(actual, expected) -> bool:
+    import math
+
+    if isinstance(actual, (int, float)) and isinstance(expected, (int, float)):
+        return math.isclose(float(actual), float(expected), rel_tol=1e-9, abs_tol=1e-12)
+    if isinstance(actual, dict) and isinstance(expected, dict):
+        return actual.keys() == expected.keys() and all(
+            _plots_equal(actual[key], expected[key]) for key in actual.keys()
+        )
+    if isinstance(actual, list) and isinstance(expected, list):
+        return len(actual) == len(expected) and all(
+            _plots_equal(left, right) for left, right in zip(actual, expected)
+        )
+    return actual == expected
+
+
+def _run_plotly_comparison(actual_path, expected_path) -> "str | None":
+    """Compare two Plotly figures. Returns None if they match, or a human diff
+    string if they differ — the same contract as `_run_comparison`, so the CLI
+    and the --serve loop treat both kinds identically."""
+    import json
+
+    actual = _strip_unstable(_load_actual_plot(actual_path))
+    expected = _strip_unstable(_load_expected_plot(expected_path))
+    if _plots_equal(actual, expected):
+        return None
+    return "\n".join(
+        [
+            "Plotly JSON mismatch",
+            "--- actual ---",
+            json.dumps(actual, indent=2, sort_keys=True),
+            "--- expected ---",
+            json.dumps(expected, indent=2, sort_keys=True),
+        ]
+    )
+
+
+def main() -> None:
+    args = sys.argv[1:]
+    unordered = False
+    ignore_cols: list = []
+    model_cols: list = []
+    probe_path = None
+
+    if args and args[0] == "--plotly":
+        if len(args) != 3:
+            print(
+                f"usage: {sys.argv[0]} --plotly <actual.jsonl> <expected.json>",
+                file=sys.stderr,
+            )
+            sys.exit(2)
+        msg = _run_plotly_comparison(args[1], args[2])
+        if msg is not None:
+            print(msg, file=sys.stderr)
+            sys.exit(1)
+        return
+
+    while args and args[0].startswith("--"):
+        if args[0] == "--unordered":
+            unordered = True
+            args = args[1:]
+        elif args[0] == "--ignore-cols":
+            if len(args) < 2:
+                print("--ignore-cols requires an argument", file=sys.stderr)
+                sys.exit(2)
+            ignore_cols = [c for c in args[1].split(",") if c]
+            args = args[2:]
+        elif args[0] == "--model-cols":
+            if len(args) < 2:
+                print("--model-cols requires an argument", file=sys.stderr)
+                sys.exit(2)
+            model_cols = [c for c in args[1].split(",") if c]
+            args = args[2:]
+        elif args[0] == "--probe":
+            if len(args) < 2:
+                print("--probe requires an argument", file=sys.stderr)
+                sys.exit(2)
+            probe_path = args[1]
+            args = args[2:]
+        else:
+            print(f"unknown flag: {args[0]}", file=sys.stderr)
+            sys.exit(2)
+    if len(args) != 2:
+        print(
+            f"usage: {sys.argv[0]} [--unordered] [--ignore-cols c1,c2] "
+            f"[--model-cols c1,c2 --probe features.jsonl] "
+            f"<actual.jsonl> <expected.jsonl>",
+            file=sys.stderr,
+        )
+        sys.exit(2)
+
+    msg = _run_comparison(
+        args[0], args[1], unordered, ignore_cols, model_cols, probe_path
+    )
+    if msg is not None:
+        print(msg, file=sys.stderr)
+        sys.exit(1)
+
+
+def serve() -> None:
+    """Persistent comparison server. See the module docstring for the protocol.
+
+    Each job runs the same function the CLI calls for its kind. A comparison
+    error is reported as exit 1 with the diff on `stderr`; only closing stdin
+    ends the loop.
+    """
+    import io
+    import json
+    import traceback
+    from contextlib import redirect_stderr, redirect_stdout
+
+    # Eagerly, before signalling ready: the point of a persistent worker is that
+    # this cost is paid once per worker instead of once per comparison, and
+    # `ready` should mean the worker is warm.
+    import pandas  # noqa: F401
+
+    sys.stdout.write(json.dumps({"ready": True}) + "\n")
+    sys.stdout.flush()
+
+    for line in sys.stdin:
+        line = line.strip()
+        if not line:
+            continue
+        out_buf, err_buf = io.StringIO(), io.StringIO()
+        try:
+            job = json.loads(line)
+            with redirect_stdout(out_buf), redirect_stderr(err_buf):
+                if job.get("kind", "dataframe") == "plotly":
+                    msg = _run_plotly_comparison(job["actual"], job["expected"])
+                else:
+                    msg = _run_comparison(
+                        job["actual"],
+                        job["expected"],
+                        job.get("unordered", False),
+                        job.get("ignoreCols", []),
+                        job.get("modelCols", []),
+                        job.get("probe"),
+                    )
+            resp = {
+                "exit": 0 if msg is None else 1,
+                "stdout": out_buf.getvalue(),
+                "stderr": err_buf.getvalue() + ("" if msg is None else msg),
+            }
+        except BaseException:  # noqa: BLE001 — a bad job must not kill the server
+            resp = {
+                "exit": 1,
+                "stdout": out_buf.getvalue(),
+                "stderr": err_buf.getvalue() + traceback.format_exc(),
+            }
+        sys.stdout.write(json.dumps(resp) + "\n")
+        sys.stdout.flush()
+
+
+if __name__ == "__main__":
+    if len(sys.argv) > 1 and sys.argv[1] == "--serve":
+        serve()
+    else:
+        main()
diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/Comparator.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/Comparator.scala
new file mode 100644
index 00000000000..17926e593b2
--- /dev/null
+++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/Comparator.scala
@@ -0,0 +1,189 @@
+/*
+ * 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.typesafe.scalalogging.LazyLogging
+import org.apache.texera.amber.util.JSONUtils.objectMapper
+import org.apache.texera.amber.util.python.PythonWorkerPool
+
+import java.nio.file.{Files, Path, StandardCopyOption}
+import scala.collection.mutable.ArrayBuffer
+import scala.sys.process._
+
+/**
+  * Runs the Python comparator (`compare.py`) on two JSONL files emitted by
+  * [[OpExecHarness]] (actual) and [[StandaloneRunner]] (expected). The
+  * comparator uses `pandas.testing.assert_frame_equal` with `check_like=True`
+  * and `check_dtype=False` so row/column-order differences and the
+  * pandas-int64/float64 coercion that happens when JSONL round-trips through
+  * `pd.read_json` don't trigger false negatives. Float tolerance: `rtol=1e-5`.
+  *
+  * Throws [[ComparatorMismatchException]] on any non-zero exit code (the
+  * pandas diff is in `stderr` on the exception). Successful comparisons
+  * return unit.
+  *
+  * Python resolution mirrors [[StandaloneRunner.resolvePython]]:
+  * `UDF_PYTHON_PATH` env var first, else `python3.12` on PATH.
+  */
+object Comparator extends LazyLogging {
+
+  // Resource path is absolute (leading slash) so getResourceAsStream resolves
+  // against the classpath root regardless of caller's package.
+  private val ScriptResourcePath = "/python/compare.py"
+
+  def assertEqual(
+      actual: Path,
+      expected: Path,
+      orderSensitive: Boolean = true,
+      ignoreColumns: Seq[String] = Seq.empty,
+      modelColumns: Seq[String] = Seq.empty,
+      probePath: Option[Path] = None,
+      pythonExe: String = resolvePython()
+  ): Unit = {
+    val (exit, stdout, stderr) =
+      compare(actual, expected, orderSensitive, ignoreColumns, modelColumns, probePath, pythonExe)
+    if (exit != 0) {
+      throw new ComparatorMismatchException(
+        actual = actual,
+        expected = expected,
+        exitCode = exit,
+        stdout = stdout,
+        stderr = stderr
+      )
+    }
+  }
+
+  // Prefer a pooled persistent worker (imports pandas once via `compare.py
+  // --serve`, so the ~214 ms import isn't repaid per comparison — the diff
+  // itself is ~ms). A rare hard worker crash falls back to the one-shot CLI so
+  // behavior is never worse than the original path. Both invoke the same
+  // `_run_comparison`, so results are identical.
+  private def compare(
+      actual: Path,
+      expected: Path,
+      orderSensitive: Boolean,
+      ignoreColumns: Seq[String],
+      modelColumns: Seq[String],
+      probePath: Option[Path],
+      pythonExe: String
+  ): (Int, String, String) = {
+    if (PythonWorkerPool.enabled) {
+      try {
+        val req = objectMapper.createObjectNode()
+        req.put("actual", actual.toString)
+        req.put("expected", expected.toString)
+        req.put("unordered", !orderSensitive)
+        val ignoreArr = req.putArray("ignoreCols")
+        ignoreColumns.foreach(ignoreArr.add)
+        val modelArr = req.putArray("modelCols")
+        modelColumns.foreach(modelArr.add)
+        // --probe only applies with --model-cols (mirrors the CLI's guard).
+        probePath.filter(_ => modelColumns.nonEmpty) match {
+          case Some(p) => req.put("probe", p.toString)
+          case None    => req.putNull("probe")
+        }
+        val o = PythonWorkerPool.run(ScriptResourcePath, Seq("--serve"), pythonExe, req)
+        return (o.exit, o.stdout, o.stderr)
+      } catch {
+        case e: PythonWorkerPool.WorkerDiedException =>
+          logger.warn(
+            s"Comparator worker unavailable; falling back to one-shot CLI: ${e.getMessage}"
+          )
+      }
+    }
+    runCli(actual, expected, orderSensitive, ignoreColumns, modelColumns, probePath, pythonExe)
+  }
+
+  // Original one-subprocess-per-comparison CLI path. Retained as the fallback
+  // and as the behavior selected by TEXERA_TEST_PYTHON_WORKER=0.
+  private def runCli(
+      actual: Path,
+      expected: Path,
+      orderSensitive: Boolean,
+      ignoreColumns: Seq[String],
+      modelColumns: Seq[String],
+      probePath: Option[Path],
+      pythonExe: String
+  ): (Int, String, String) = {
+    val scriptPath = extractScript()
+    val outBuf = ArrayBuffer.empty[String]
+    val errBuf = ArrayBuffer.empty[String]
+    val procLogger = ProcessLogger(line => outBuf += line, line => errBuf += line)
+    // --unordered tells compare.py to lex-sort both DataFrames by all columns
+    // before assert_frame_equal — needed for set-semantics ops whose JVM
+    // emission order doesn't match the pandas equivalent. Default stays
+    // positional so deterministic-order ops still catch row-order regressions.
+    // --ignore-cols drops opaque columns whose value isn't compared.
+    // --model-cols + --probe compare a model column by behavior: unpickle both
+    // sides and assert their predictions on the probe feature set match (two
+    // independently-trained models are functionally equal but not bit-equal).
+    val baseArgs = Seq(pythonExe, scriptPath.toString)
+    val flagArgs = if (!orderSensitive) Seq("--unordered") else Seq.empty
+    val ignoreArgs =
+      if (ignoreColumns.nonEmpty) Seq("--ignore-cols", ignoreColumns.mkString(",")) else Seq.empty
+    val modelArgs =
+      if (modelColumns.nonEmpty) Seq("--model-cols", modelColumns.mkString(",")) else Seq.empty
+    val probeArgs =
+      probePath
+        .filter(_ => modelColumns.nonEmpty)
+        .map(p => Seq("--probe", p.toString))
+        .getOrElse(Seq.empty)
+    val cmd =
+      baseArgs ++ flagArgs ++ ignoreArgs ++ modelArgs ++ probeArgs ++ Seq(
+        actual.toString,
+        expected.toString
+      )
+    val exit = Process(cmd).!(procLogger)
+    (exit, outBuf.mkString("\n"), errBuf.mkString("\n"))
+  }
+
+  // Resources may live inside a jar at runtime; copy to a temp file so Python
+  // can exec it. deleteOnExit so test runs don't accumulate /tmp clutter.
+  private def extractScript(): Path = {
+    val stream = getClass.getResourceAsStream(ScriptResourcePath)
+    require(
+      stream != null,
+      s"compare.py not found on classpath at $ScriptResourcePath"
+    )
+    try {
+      val tmp = Files.createTempFile("compare-", ".py")
+      Files.copy(stream, tmp, StandardCopyOption.REPLACE_EXISTING)
+      tmp.toFile.deleteOnExit()
+      tmp
+    } finally stream.close()
+  }
+
+  private def resolvePython(): String =
+    sys.env.get("UDF_PYTHON_PATH").filter(_.nonEmpty).getOrElse("python3.12")
+}
+
+final class ComparatorMismatchException(
+    val actual: Path,
+    val expected: Path,
+    val exitCode: Int,
+    val stdout: String,
+    val stderr: String
+) extends RuntimeException(
+      s"""DataFrame mismatch (compare.py exit $exitCode):
+         |  actual:   $actual
+         |  expected: $expected
+         |--- stderr ---
+         |$stderr""".stripMargin
+    )
diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/ComparatorSpec.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/ComparatorSpec.scala
new file mode 100644
index 00000000000..dc5e5fda029
--- /dev/null
+++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/ComparatorSpec.scala
@@ -0,0 +1,97 @@
+/*
+ * 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.core.tuple.{Attribute, AttributeType, Schema, Tuple}
+import org.apache.texera.amber.translator.verify.tags.IntegrationTest
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import java.nio.file.{Files, Path}
+
+// Tagged @IntegrationTest: Comparator.assertEqual shells out to compare.py, so
+// this spec needs Python and must run in the Python-provisioned integration job.
+@IntegrationTest
+class ComparatorSpec extends AnyFlatSpec with Matchers {
+
+  private val schema: Schema = Schema()
+    .add(new Attribute("id", AttributeType.INTEGER))
+    .add(new Attribute("name", AttributeType.STRING))
+
+  private val idAttr = new Attribute("id", AttributeType.INTEGER)
+  private val nameAttr = new Attribute("name", AttributeType.STRING)
+
+  private def row(id: Int, name: String): Tuple =
+    Tuple
+      .builder(schema)
+      .add(idAttr, Int.box(id))
+      .add(nameAttr, name)
+      .build()
+
+  private def writeJsonl(dir: Path, name: String, rows: Seq[Tuple]): Path = {
+    val p = dir.resolve(name)
+    TupleIO.writeTuples(p, rows.iterator, schema)
+    p
+  }
+
+  "Comparator.assertEqual" should "pass when JSONL files contain identical rows" in {
+    val dir = Files.createTempDirectory("comparator-spec-equal-")
+    val rows = Seq(row(1, "alice"), row(2, "bob"))
+    val a = writeJsonl(dir, "a.jsonl", rows)
+    val b = writeJsonl(dir, "b.jsonl", rows)
+    noException should be thrownBy Comparator.assertEqual(a, b)
+  }
+
+  it should "throw ComparatorMismatchException when JSONL files differ" in {
+    val dir = Files.createTempDirectory("comparator-spec-diff-")
+    val a = writeJsonl(dir, "a.jsonl", Seq(row(1, "alice"), row(2, "bob")))
+    val b = writeJsonl(dir, "b.jsonl", Seq(row(1, "alice"), row(2, "carol")))
+    intercept[ComparatorMismatchException] {
+      Comparator.assertEqual(a, b)
+    }
+  }
+
+  it should "treat row-reordered files as unequal under positional comparison" in {
+    val dir = Files.createTempDirectory("comparator-spec-reorder-strict-")
+    val a = writeJsonl(dir, "a.jsonl", Seq(row(1, "alice"), row(2, "bob")))
+    val b = writeJsonl(dir, "b.jsonl", Seq(row(2, "bob"), row(1, "alice")))
+    intercept[ComparatorMismatchException] {
+      Comparator.assertEqual(a, b)
+    }
+  }
+
+  it should "treat row-reordered files as equal under orderSensitive=false" in {
+    val dir = Files.createTempDirectory("comparator-spec-reorder-loose-")
+    val a = writeJsonl(dir, "a.jsonl", Seq(row(1, "alice"), row(2, "bob")))
+    val b = writeJsonl(dir, "b.jsonl", Seq(row(2, "bob"), row(1, "alice")))
+    noException should be thrownBy Comparator.assertEqual(a, b, orderSensitive = false)
+  }
+
+  it should "still report value mismatches under orderSensitive=false" in {
+    // orderSensitive=false relaxes row ORDER, not row CONTENT — a genuinely
+    // different cell must still fail.
+    val dir = Files.createTempDirectory("comparator-spec-reorder-content-diff-")
+    val a = writeJsonl(dir, "a.jsonl", Seq(row(1, "alice"), row(2, "bob")))
+    val b = writeJsonl(dir, "b.jsonl", Seq(row(2, "bob"), row(1, "carol")))
+    intercept[ComparatorMismatchException] {
+      Comparator.assertEqual(a, b, orderSensitive = false)
+    }
+  }
+}
diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/VisualizationHtmlComparator.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/VisualizationHtmlComparator.scala
new file mode 100644
index 00000000000..c9fa74cf877
--- /dev/null
+++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/VisualizationHtmlComparator.scala
@@ -0,0 +1,85 @@
+/*
+ * 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.util.JSONUtils.objectMapper
+
+import java.nio.charset.StandardCharsets
+import java.nio.file.{Files, Path}
+
+object VisualizationHtmlComparator {
+
+  /** A pandas Styler namespaces its CSS with a uuid drawn per Styler instance, so
+    * the same table rendered twice differs in every `id=` and every selector even
+    * though the markup is identical. The uuid carries no information about the
+    * table — it only keeps two tables on one page from colliding — so it is
+    * normalized away before comparing. Only the random prefix is replaced: the
+    * `_row0_col0` suffix that identifies the cell stays, so a genuine structural
+    * difference still fails.
+    */
+  private val StylerUuid = "T_[0-9a-f]+".r
+
+  private def normalize(html: String): String = StylerUuid.replaceAllIn(html, "T_uuid")
+
+  def assertEqual(actualVisualizationJsonl: Path, expectedHtmlFile: Path): Unit = {
+    val actual = readActualHtml(actualVisualizationJsonl)
+    val expected = new String(Files.readAllBytes(expectedHtmlFile), StandardCharsets.UTF_8)
+
+    if (normalize(actual) != normalize(expected)) {
+      throw new VisualizationHtmlMismatchException(
+        actual = actualVisualizationJsonl,
+        expected = expectedHtmlFile,
+        actualHtml = actual,
+        expectedHtml = expected
+      )
+    }
+  }
+
+  private def readActualHtml(path: Path): String = {
+    val line = Files
+      .readAllLines(path, StandardCharsets.UTF_8)
+      .stream()
+      .filter(_.trim.nonEmpty)
+      .findFirst()
+      .orElseThrow(() => new AssertionError(s"$path is empty"))
+
+    val node = objectMapper.readTree(line)
+    val htmlNode = node.get("html-content")
+    if (htmlNode == null || htmlNode.isNull) {
+      throw new AssertionError(s"$path has no html-content field")
+    }
+    htmlNode.asText()
+  }
+}
+
+final class VisualizationHtmlMismatchException(
+    val actual: Path,
+    val expected: Path,
+    val actualHtml: String,
+    val expectedHtml: String
+) extends RuntimeException(
+      s"""Visualization HTML mismatch:
+         |  actual:   $actual
+         |  expected: $expected
+         |--- actual html ---
+         |$actualHtml
+         |--- expected html ---
+         |$expectedHtml""".stripMargin
+    )
diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/VisualizationJsonComparator.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/VisualizationJsonComparator.scala
new file mode 100644
index 00000000000..f559b2c39e6
--- /dev/null
+++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/VisualizationJsonComparator.scala
@@ -0,0 +1,123 @@
+/*
+ * 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.typesafe.scalalogging.LazyLogging
+import org.apache.texera.amber.util.JSONUtils.objectMapper
+import org.apache.texera.amber.util.python.PythonWorkerPool
+
+import java.nio.file.{Files, Path, StandardCopyOption}
+import scala.collection.mutable.ArrayBuffer
+import scala.sys.process._
+
+/**
+  * Compares the Plotly figure the two paths render, via `compare.py`'s
+  * `--plotly` mode.
+  *
+  * Shares that script — and therefore [[Comparator]]'s pool — rather than
+  * carrying one of its own: a worker is bound to the script it was launched
+  * with, so a separate script would mean a separate pool of interpreters for
+  * what is the same job, comparing one operator's two outputs. One comparison
+  * pool serves both output shapes.
+  */
+object VisualizationJsonComparator extends LazyLogging {
+
+  private val ScriptResourcePath = "/python/compare.py"
+
+  def assertEqual(
+      actualVisualizationJsonl: Path,
+      expectedPlotlyJson: Path,
+      pythonExe: String = resolvePython()
+  ): Unit = {
+    val (exit, stdout, stderr) =
+      compare(actualVisualizationJsonl, expectedPlotlyJson, pythonExe)
+    if (exit != 0) {
+      throw new VisualizationJsonMismatchException(
+        actual = actualVisualizationJsonl,
+        expected = expectedPlotlyJson,
+        exitCode = exit,
+        stdout = stdout,
+        stderr = stderr
+      )
+    }
+  }
+
+  // Pooled worker first, one-shot CLI as the fallback and as the behavior
+  // selected by TEXERA_TEST_PYTHON_WORKER=0. Both run the same
+  // `_run_plotly_comparison`, so results are identical.
+  private def compare(actual: Path, expected: Path, pythonExe: String): (Int, String, String) = {
+    if (PythonWorkerPool.enabled) {
+      try {
+        val req = objectMapper.createObjectNode()
+        req.put("kind", "plotly")
+        req.put("actual", actual.toString)
+        req.put("expected", expected.toString)
+        val o = PythonWorkerPool.run(ScriptResourcePath, Seq("--serve"), pythonExe, req)
+        return (o.exit, o.stdout, o.stderr)
+      } catch {
+        case e: PythonWorkerPool.WorkerDiedException =>
+          logger.warn(
+            s"Comparator worker unavailable; falling back to one-shot CLI: ${e.getMessage}"
+          )
+      }
+    }
+    runCli(actual, expected, pythonExe)
+  }
+
+  private def runCli(actual: Path, expected: Path, pythonExe: String): (Int, String, String) = {
+    val scriptPath = extractScript()
+    val outBuf = ArrayBuffer.empty[String]
+    val errBuf = ArrayBuffer.empty[String]
+    val processLogger = ProcessLogger(line => outBuf += line, line => errBuf += line)
+
+    val exit = Process(
+      Seq(pythonExe, scriptPath.toString, "--plotly", actual.toString, expected.toString)
+    ).!(processLogger)
+    (exit, outBuf.mkString("\n"), errBuf.mkString("\n"))
+  }
+
+  private def extractScript(): Path = {
+    val stream = getClass.getResourceAsStream(ScriptResourcePath)
+    require(stream != null, s"compare.py not found at $ScriptResourcePath")
+    try {
+      val tmp = Files.createTempFile("compare-", ".py")
+      Files.copy(stream, tmp, StandardCopyOption.REPLACE_EXISTING)
+      tmp.toFile.deleteOnExit()
+      tmp
+    } finally stream.close()
+  }
+
+  private def resolvePython(): String =
+    sys.env.get("UDF_PYTHON_PATH").filter(_.nonEmpty).getOrElse("python3.12")
+}
+
+final class VisualizationJsonMismatchException(
+    val actual: Path,
+    val expected: Path,
+    val exitCode: Int,
+    val stdout: String,
+    val stderr: String
+) extends RuntimeException(
+      s"""Visualization JSON mismatch (compare.py --plotly exit $exitCode):
+         |  actual:   $actual
+         |  expected: $expected
+         |--- stderr ---
+         |$stderr""".stripMargin
+    )

From 4ea66b377b90408b0628428552731fe8d60ee6e9 Mon Sep 17 00:00:00 2001
From: kary zheng <zhengj29@uci.edu>
Date: Thu, 3 Sep 2026 13:06:02 -0700
Subject: [PATCH 2/9] fix(verify): read a string column as one on both sides

`read_json` infers a type per file, so a column one side wrote as "6" and the
other wrote as "6.0" both arrive as the number 6, and a null beside the text
"nan" both arrive as NaN. Two different answers compared as one, on every
string column of every operator.

The engine writes a schema beside its output naming which columns are strings,
and both sides are now read that way.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../src/test/resources/python/compare.py      | 30 +++++++++++++++++--
 1 file changed, 28 insertions(+), 2 deletions(-)

diff --git a/workflow-compiling-service/src/test/resources/python/compare.py b/workflow-compiling-service/src/test/resources/python/compare.py
index 518b5fac901..fb1cf8bc8db 100644
--- a/workflow-compiling-service/src/test/resources/python/compare.py
+++ b/workflow-compiling-service/src/test/resources/python/compare.py
@@ -160,6 +160,25 @@ def _compare_model_predictions(actual, expected, model_cols, probe_path) -> None
                 )
 
 
+def _string_columns(actual_path: str) -> dict:
+    """Which columns the engine declared as strings, read off the schema it
+    writes beside its output. Empty when there is no sidecar, which leaves the
+    inference in place rather than guessing."""
+    import json
+    import os
+
+    sidecar = actual_path + ".schema.json"
+    if not os.path.exists(sidecar):
+        return {}
+    with open(sidecar) as fh:
+        schema = json.load(fh)
+    return {
+        a["attributeName"]: str
+        for a in schema.get("attributes", [])
+        if a.get("attributeType") == "string"
+    }
+
+
 def _run_comparison(
     actual_path: str,
     expected_path: str,
@@ -174,8 +193,15 @@ def _run_comparison(
     comparison truth shared by the CLI and the --serve loop."""
     import pandas as pd
 
-    actual = pd.read_json(actual_path, lines=True)
-    expected = pd.read_json(expected_path, lines=True)
+    # A string column has to be READ as one on both sides. Left to itself,
+    # `read_json` infers a type per file, so a column the engine wrote as "6"
+    # and the script wrote as "6.0" both arrive as the number 6, and a null
+    # beside the text "nan" both arrive as NaN -- two genuinely different
+    # answers compared as one. The engine writes a schema next to its output;
+    # it names which columns are strings, and both sides are read that way.
+    str_cols = _string_columns(actual_path)
+    actual = pd.read_json(actual_path, lines=True, dtype=str_cols or None)
+    expected = pd.read_json(expected_path, lines=True, dtype=str_cols or None)
 
     # Model columns: compare behavior (predictions) rather than bytes, then drop
     # the raw columns so the frame comparison covers everything else exactly.

From 5cffbd2c87ea60bed75a538dacb4ac579ecf9635 Mon Sep 17 00:00:00 2001
From: kary zheng <kary09302005@gmail.com>
Date: Thu, 3 Sep 2026 15:46:01 -0700
Subject: [PATCH 3/9] test(verify): say it once, in the shape the code does not
 already give

Three kinds of comment came out. A drawing of the string the code below
assembles. A restatement of a branch the reader can see. And the word MVP,
which dated the scope to a moment rather than stating it.

What replaces them says the same thing shorter, or says what the code cannot:
which cases the harness does not drive and why none of them has an operator
asking for it.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../apache/texera/amber/translator/verify/Comparator.scala   | 5 ++---
 1 file changed, 2 insertions(+), 3 deletions(-)

diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/Comparator.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/Comparator.scala
index 17926e593b2..d1be981c005 100644
--- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/Comparator.scala
+++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/Comparator.scala
@@ -35,9 +35,8 @@ import scala.sys.process._
   * pandas-int64/float64 coercion that happens when JSONL round-trips through
   * `pd.read_json` don't trigger false negatives. Float tolerance: `rtol=1e-5`.
   *
-  * Throws [[ComparatorMismatchException]] on any non-zero exit code (the
-  * pandas diff is in `stderr` on the exception). Successful comparisons
-  * return unit.
+  * Throws [[ComparatorMismatchException]] on any non-zero exit code, carrying
+  * the pandas diff from `stderr`.
   *
   * Python resolution mirrors [[StandaloneRunner.resolvePython]]:
   * `UDF_PYTHON_PATH` env var first, else `python3.12` on PATH.

From 318120e02dc0d588c179dd9b348147102e17605a Mon Sep 17 00:00:00 2001
From: kary zheng <kary09302005@gmail.com>
Date: Tue, 8 Sep 2026 23:40:36 -0700
Subject: [PATCH 4/9] test(verify): say which of the two orders check_like
 tolerates

It ignores column order. Row order is settled by --unordered, which lex-sorts
both frames unless the operator declares itself order-sensitive.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../texera/amber/translator/verify/Comparator.scala  | 12 +++++++-----
 1 file changed, 7 insertions(+), 5 deletions(-)

diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/Comparator.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/Comparator.scala
index d1be981c005..e7447f0dbf7 100644
--- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/Comparator.scala
+++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/Comparator.scala
@@ -29,11 +29,13 @@ import scala.sys.process._
 
 /**
   * Runs the Python comparator (`compare.py`) on two JSONL files emitted by
-  * [[OpExecHarness]] (actual) and [[StandaloneRunner]] (expected). The
-  * comparator uses `pandas.testing.assert_frame_equal` with `check_like=True`
-  * and `check_dtype=False` so row/column-order differences and the
-  * pandas-int64/float64 coercion that happens when JSONL round-trips through
-  * `pd.read_json` don't trigger false negatives. Float tolerance: `rtol=1e-5`.
+  * [[OpExecHarness]] (actual) and [[StandaloneRunner]] (expected). It uses
+  * `pandas.testing.assert_frame_equal` with `check_like=True`, so column order
+  * does not matter, and `check_dtype=False`, so the int64/float64 coercion a
+  * JSONL round trip through `pd.read_json` performs is not a difference. Float
+  * tolerance is `rtol=1e-5`. ROW order is a separate question, settled by
+  * `--unordered`, which lex-sorts both frames unless the operator declares
+  * itself order-sensitive.
   *
   * Throws [[ComparatorMismatchException]] on any non-zero exit code, carrying
   * the pandas diff from `stderr`.

From 70969a10f228b9bfb6d66956acb740bcc17fdc81 Mon Sep 17 00:00:00 2001
From: kary zheng <kary09302005@gmail.com>
Date: Wed, 9 Sep 2026 16:17:39 -0700
Subject: [PATCH 5/9] test(verify): require every model column on both sides

A requested column is one the engine declared as a model, so a side that
never emitted it is the divergence. Skipping it hid that: the column was
dropped from both frames afterwards and a path that produced no model at
all compared equal.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../src/test/resources/python/compare.py      | 15 ++++++--
 .../translator/verify/ComparatorSpec.scala    | 36 +++++++++++++++++++
 2 files changed, 49 insertions(+), 2 deletions(-)

diff --git a/workflow-compiling-service/src/test/resources/python/compare.py b/workflow-compiling-service/src/test/resources/python/compare.py
index fb1cf8bc8db..5528ab1a2ed 100644
--- a/workflow-compiling-service/src/test/resources/python/compare.py
+++ b/workflow-compiling-service/src/test/resources/python/compare.py
@@ -113,8 +113,19 @@ def _compare_model_predictions(actual, expected, model_cols, probe_path) -> None
         )
 
     for col in model_cols:
-        if col not in actual.columns or col not in expected.columns:
-            continue
+        # A requested column is one the engine declared as a model, so a side
+        # that never emitted it IS the divergence. Skipping it here would hide
+        # that: the column is dropped from both frames afterwards, and a path
+        # that produced no model at all would compare equal.
+        missing = [
+            side
+            for side, frame in (("actual", actual), ("expected", expected))
+            if col not in frame.columns
+        ]
+        if missing:
+            raise AssertionError(
+                f"model column {col!r} missing from {' and '.join(missing)}"
+            )
         if len(actual) != len(expected):
             raise AssertionError(
                 f"model column {col!r}: row count differs "
diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/ComparatorSpec.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/ComparatorSpec.scala
index dc5e5fda029..3a2038d1983 100644
--- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/ComparatorSpec.scala
+++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/ComparatorSpec.scala
@@ -51,6 +51,14 @@ class ComparatorSpec extends AnyFlatSpec with Matchers {
     p
   }
 
+  // Written by hand rather than through TupleIO: these cases need a column set
+  // the fixed schema above does not carry.
+  private def writeLines(dir: Path, name: String, lines: Seq[String]): Path = {
+    val p = dir.resolve(name)
+    Files.writeString(p, lines.map(_ + "\n").mkString)
+    p
+  }
+
   "Comparator.assertEqual" should "pass when JSONL files contain identical rows" in {
     val dir = Files.createTempDirectory("comparator-spec-equal-")
     val rows = Seq(row(1, "alice"), row(2, "bob"))
@@ -94,4 +102,32 @@ class ComparatorSpec extends AnyFlatSpec with Matchers {
       Comparator.assertEqual(a, b, orderSensitive = false)
     }
   }
+
+  it should "report a model column only one side produced" in {
+    // Model columns are compared by behavior and then dropped from both frames,
+    // so a side that never wrote one has to fail here: once the column is gone,
+    // the frame diff sees two identical column sets and passes.
+    val dir = Files.createTempDirectory("comparator-spec-model-missing-")
+    val withModel = writeLines(dir, "with-model.jsonl", Seq("""{"model":"eA==","score":1.0}"""))
+    val withoutModel = writeLines(dir, "without-model.jsonl", Seq("""{"score":1.0}"""))
+    val probe = writeLines(dir, "probe.jsonl", Seq("""{"petal_length":1.0,"label":0}"""))
+
+    intercept[ComparatorMismatchException] {
+      Comparator.assertEqual(
+        withModel,
+        withoutModel,
+        modelColumns = Seq("model"),
+        probePath = Some(probe)
+      )
+    }.getMessage should include("missing from expected")
+
+    intercept[ComparatorMismatchException] {
+      Comparator.assertEqual(
+        withoutModel,
+        withModel,
+        modelColumns = Seq("model"),
+        probePath = Some(probe)
+      )
+    }.getMessage should include("missing from actual")
+  }
 }

From f412a0ef321fdaacc14635b275725fa27d5a0e45 Mon Sep 17 00:00:00 2001
From: kary zheng <kary09302005@gmail.com>
Date: Wed, 9 Sep 2026 17:03:42 -0700
Subject: [PATCH 6/9] test(verify): compare visualization HTML without its line
 endings

The standalone script writes its page through Python's text mode, so on Windows
the file holds CRLF while the runtime path's JSONL carries the same markup with
LF. Five operators reported a mismatch that was only that. The line ending is
the platform writing the file rather than anything the operator chose, so it is
normalized away like the Styler uuid already is.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../verify/VisualizationHtmlComparator.scala  | 10 ++-
 .../VisualizationHtmlComparatorSpec.scala     | 72 +++++++++++++++++++
 2 files changed, 81 insertions(+), 1 deletion(-)
 create mode 100644 workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/VisualizationHtmlComparatorSpec.scala

diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/VisualizationHtmlComparator.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/VisualizationHtmlComparator.scala
index c9fa74cf877..2455833577b 100644
--- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/VisualizationHtmlComparator.scala
+++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/VisualizationHtmlComparator.scala
@@ -36,7 +36,15 @@ object VisualizationHtmlComparator {
     */
   private val StylerUuid = "T_[0-9a-f]+".r
 
-  private def normalize(html: String): String = StylerUuid.replaceAllIn(html, "T_uuid")
+  /** The standalone script writes its page with Python's text mode, which on Windows
+    * turns every newline into CRLF, while the runtime path carries the same markup
+    * through JSONL untouched and so keeps LF. The line ending is the platform writing
+    * the file rather than anything the operator chose, so it is normalized away too.
+    */
+  private val LineEnding = "\r\n|\r".r
+
+  private def normalize(html: String): String =
+    StylerUuid.replaceAllIn(LineEnding.replaceAllIn(html, "\n"), "T_uuid")
 
   def assertEqual(actualVisualizationJsonl: Path, expectedHtmlFile: Path): Unit = {
     val actual = readActualHtml(actualVisualizationJsonl)
diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/VisualizationHtmlComparatorSpec.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/VisualizationHtmlComparatorSpec.scala
new file mode 100644
index 00000000000..32ccbee7ac6
--- /dev/null
+++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/VisualizationHtmlComparatorSpec.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.util.JSONUtils.objectMapper
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import java.nio.charset.StandardCharsets
+import java.nio.file.{Files, Path}
+
+// Untagged: the comparator reads two files and compares strings, with no Python
+// to shell out to, so this belongs in the plain unit job.
+class VisualizationHtmlComparatorSpec extends AnyFlatSpec with Matchers {
+
+  private val dir: Path = Files.createTempDirectory("visualization-html-comparator-spec-")
+
+  /** The runtime path's side: the markup as one `html-content` field of a JSONL row. */
+  private def writeActual(name: String, html: String): Path = {
+    val node = objectMapper.createObjectNode()
+    node.put("html-content", html)
+    val p = dir.resolve(name)
+    Files.write(p, objectMapper.writeValueAsBytes(node))
+    p
+  }
+
+  /** The standalone path's side: the markup as the script wrote it to a file. */
+  private def writeExpected(name: String, html: String): Path = {
+    val p = dir.resolve(name)
+    Files.write(p, html.getBytes(StandardCharsets.UTF_8))
+    p
+  }
+
+  private val page = "<html>\n<body>\n<p>chart</p>\n</body>\n</html>\n"
+
+  "VisualizationHtmlComparator" should "accept two sides that differ only in line endings" in {
+    val actual = writeActual("crlf-actual.jsonl", page)
+    val expected = writeExpected("crlf-expected.html", page.replace("\n", "\r\n"))
+    noException should be thrownBy VisualizationHtmlComparator.assertEqual(actual, expected)
+  }
+
+  it should "accept two sides that differ only in a Styler uuid" in {
+    val actual = writeActual("styler-actual.jsonl", "<td id=\"T_a1b2c3_row0_col0\">1</td>")
+    val expected = writeExpected("styler-expected.html", "<td id=\"T_9f8e7d_row0_col0\">1</td>")
+    noException should be thrownBy VisualizationHtmlComparator.assertEqual(actual, expected)
+  }
+
+  it should "still reject markup that differs in more than its line endings" in {
+    val actual = writeActual("differ-actual.jsonl", page)
+    val expected =
+      writeExpected("differ-expected.html", page.replace("chart", "table").replace("\n", "\r\n"))
+    a[VisualizationHtmlMismatchException] should be thrownBy VisualizationHtmlComparator
+      .assertEqual(actual, expected)
+  }
+}

From 080f100e3b0af1bb041fa44bd5a322d728d1490b Mon Sep 17 00:00:00 2001
From: kary zheng <kary09302005@gmail.com>
Date: Thu, 10 Sep 2026 13:01:02 -0700
Subject: [PATCH 7/9] test(verify): compare an integer column exactly

The comparison gave every numeric column the tolerance a double needs,
so two whole numbers a workflow would never call the same passed it: at
rtol=1e-5, LONG 100000 and 100001 compare equal. The declared integer
columns are split out and compared exactly; the rest keep the tolerance.

Reading them again is the other half. `read_json` parses a column
holding a null through float64, so 9007199254740993 is already
9007199254740992 before anything compares it, and pinning the dtype does
not help: the rounding happens on the way in. Python's json reads the
integer exactly. A value that is not whole in a column the engine
declared integral is itself the divergence, so it is reported.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../src/test/resources/python/compare.py      | 83 +++++++++++++++++--
 .../translator/verify/ComparatorSpec.scala    | 59 +++++++++++++
 2 files changed, 134 insertions(+), 8 deletions(-)

diff --git a/workflow-compiling-service/src/test/resources/python/compare.py b/workflow-compiling-service/src/test/resources/python/compare.py
index 5528ab1a2ed..74f31494b0b 100644
--- a/workflow-compiling-service/src/test/resources/python/compare.py
+++ b/workflow-compiling-service/src/test/resources/python/compare.py
@@ -171,8 +171,8 @@ def _compare_model_predictions(actual, expected, model_cols, probe_path) -> None
                 )
 
 
-def _string_columns(actual_path: str) -> dict:
-    """Which columns the engine declared as strings, read off the schema it
+def _declared_types(actual_path: str) -> dict:
+    """What the engine declared each output column as, read off the schema it
     writes beside its output. Empty when there is no sidecar, which leaves the
     inference in place rather than guessing."""
     import json
@@ -183,13 +183,52 @@ def _string_columns(actual_path: str) -> dict:
         return {}
     with open(sidecar) as fh:
         schema = json.load(fh)
+    return {a["attributeName"]: a.get("attributeType") for a in schema.get("attributes", [])}
+
+
+def _string_columns(actual_path: str) -> dict:
+    """The declared STRING columns, as a `read_json` dtype map."""
     return {
-        a["attributeName"]: str
-        for a in schema.get("attributes", [])
-        if a.get("attributeType") == "string"
+        name: str for name, kind in _declared_types(actual_path).items() if kind == "string"
     }
 
 
+def _exact_integers(path: str, columns: list) -> dict:
+    """The named columns re-read with Python's json, whose integers are exact.
+
+    `read_json` parses a column holding a null through float64, so a LONG of
+    9007199254740993 is already 9007199254740992 by the time anything compares
+    it. Pinning the dtype does not help: the rounding happens on the way in.
+    A value that is not whole is itself the divergence, so it is reported.
+    """
+    import json
+
+    import pandas as pd
+
+    values = {column: [] for column in columns}
+    with open(path) as fh:
+        for line in fh:
+            line = line.strip()
+            if not line:
+                continue
+            row = json.loads(line)
+            for column in columns:
+                cell = row.get(column)
+                if cell is None:
+                    values[column].append(pd.NA)
+                elif isinstance(cell, bool) or not isinstance(cell, (int, float)):
+                    raise AssertionError(
+                        f"column '{column}' is declared integral but holds {cell!r} in {path}"
+                    )
+                elif cell != int(cell):
+                    raise AssertionError(
+                        f"column '{column}' is declared integral but holds {cell!r} in {path}"
+                    )
+                else:
+                    values[column].append(int(cell))
+    return {column: pd.array(cells, dtype="Int64") for column, cells in values.items()}
+
+
 def _run_comparison(
     actual_path: str,
     expected_path: str,
@@ -210,10 +249,26 @@ def _run_comparison(
     # beside the text "nan" both arrive as NaN -- two genuinely different
     # answers compared as one. The engine writes a schema next to its output;
     # it names which columns are strings, and both sides are read that way.
-    str_cols = _string_columns(actual_path)
+    declared = _declared_types(actual_path)
+    str_cols = {name: str for name, kind in declared.items() if kind == "string"}
     actual = pd.read_json(actual_path, lines=True, dtype=str_cols or None)
     expected = pd.read_json(expected_path, lines=True, dtype=str_cols or None)
 
+    # Both sides take the ENGINE's declared type, which also settles a column a
+    # hole widened to float on one path and not the other.
+    int_cols = [
+        name
+        for name, kind in declared.items()
+        if kind in ("integer", "long") and name in actual.columns and name in expected.columns
+    ]
+    if int_cols:
+        try:
+            for frame, path in ((actual, actual_path), (expected, expected_path)):
+                for column, values in _exact_integers(path, int_cols).items():
+                    frame[column] = values
+        except AssertionError as exc:
+            return str(exc)
+
     # Model columns: compare behavior (predictions) rather than bytes, then drop
     # the raw columns so the frame comparison covers everything else exactly.
     if model_cols:
@@ -244,10 +299,22 @@ def _run_comparison(
                 by=cols, kind="mergesort", na_position="last"
             ).reset_index(drop=True)
 
+    # The tolerance was letting integers through with it: at rtol=1e-5, LONG
+    # 100000 and 100001 compare equal. Integer columns are compared exactly.
+    exact_cols = [c for c in int_cols if c in actual.columns]
+    loose_cols = [c for c in actual.columns if c not in exact_cols]
     try:
+        if exact_cols:
+            pd.testing.assert_frame_equal(
+                actual[exact_cols],
+                expected[exact_cols],
+                check_like=True,
+                check_dtype=False,
+                check_exact=True,
+            )
         pd.testing.assert_frame_equal(
-            actual,
-            expected,
+            actual[loose_cols],
+            expected[loose_cols],
             check_like=True,
             check_dtype=False,
             rtol=1e-5,
diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/ComparatorSpec.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/ComparatorSpec.scala
index 3a2038d1983..4ec7066ef14 100644
--- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/ComparatorSpec.scala
+++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/ComparatorSpec.scala
@@ -103,6 +103,65 @@ class ComparatorSpec extends AnyFlatSpec with Matchers {
     }
   }
 
+  // The tolerance a double needs was being applied to integers as well, so two
+  // whole numbers a workflow would never call the same compared equal. The
+  // declared type is what separates them: a double keeps the tolerance.
+  private val longSchema: Schema = Schema().add(new Attribute("n", AttributeType.LONG))
+
+  private def writeLongs(dir: Path, name: String, values: Seq[java.lang.Long]): Path = {
+    val p = dir.resolve(name)
+    val rows =
+      values.map(v => Tuple.builder(longSchema).add(longSchema.getAttribute("n"), v).build())
+    TupleIO.writeTuples(p, rows.iterator, longSchema)
+    p
+  }
+
+  it should "reject two integers the float tolerance would have accepted" in {
+    val dir = Files.createTempDirectory("comparator-spec-long-tolerance-")
+    val a = writeLongs(dir, "a.jsonl", Seq(100000L))
+    val b = writeLongs(dir, "b.jsonl", Seq(100001L))
+    intercept[ComparatorMismatchException] {
+      Comparator.assertEqual(a, b)
+    }
+  }
+
+  it should "reject a nullable integer that only a rounding read made equal" in {
+    // A null widens the column to float on the way into pandas, and 9007199254740993
+    // is already 9007199254740992 before anything compares it.
+    val dir = Files.createTempDirectory("comparator-spec-long-precision-")
+    val a = writeLongs(dir, "a.jsonl", Seq(9007199254740993L, null))
+    val b = writeLongs(dir, "b.jsonl", Seq(9007199254740992L, null))
+    intercept[ComparatorMismatchException] {
+      Comparator.assertEqual(a, b)
+    }
+  }
+
+  it should "keep the tolerance a double needs" in {
+    val dir = Files.createTempDirectory("comparator-spec-double-tolerance-")
+    val doubleSchema = Schema().add(new Attribute("x", AttributeType.DOUBLE))
+    def write(name: String, v: Double): Path = {
+      val p = dir.resolve(name)
+      val row =
+        Tuple.builder(doubleSchema).add(doubleSchema.getAttribute("x"), Double.box(v)).build()
+      TupleIO.writeTuples(p, Iterator(row), doubleSchema)
+      p
+    }
+    noException should be thrownBy Comparator.assertEqual(
+      write("a.jsonl", 1.000001),
+      write("b.jsonl", 1.0000011)
+    )
+  }
+
+  it should "read an integer column by its declared type on both sides" in {
+    // The script widens a holed integer column to float and writes 6.0 where the
+    // engine wrote 6. Both are the integer the schema declares, so this is the one
+    // difference in spelling that is not a difference in answer.
+    val dir = Files.createTempDirectory("comparator-spec-int-spelling-")
+    val a = writeLongs(dir, "a.jsonl", Seq(6L))
+    val b = writeLines(dir, "b.jsonl", Seq("""{"n":6.0}"""))
+    noException should be thrownBy Comparator.assertEqual(a, b)
+  }
+
   it should "report a model column only one side produced" in {
     // Model columns are compared by behavior and then dropped from both frames,
     // so a side that never wrote one has to fail here: once the column is gone,

From 8fd71bcb269993ba6a63e660ae1a45cab6b806c8 Mon Sep 17 00:00:00 2001
From: kary zheng <kary09302005@gmail.com>
Date: Thu, 10 Sep 2026 16:45:39 -0700
Subject: [PATCH 8/9] test(verify): cut the doc to what the code cannot say

It restated its own signature and listed a codec table that lives in
TupleIO. The reasons a reader cannot derive stay.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../amber/translator/verify/Comparator.scala  | 20 +++++++------------
 1 file changed, 7 insertions(+), 13 deletions(-)

diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/Comparator.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/Comparator.scala
index e7447f0dbf7..ae67d5c478a 100644
--- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/Comparator.scala
+++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/Comparator.scala
@@ -28,20 +28,14 @@ import scala.collection.mutable.ArrayBuffer
 import scala.sys.process._
 
 /**
-  * Runs the Python comparator (`compare.py`) on two JSONL files emitted by
-  * [[OpExecHarness]] (actual) and [[StandaloneRunner]] (expected). It uses
-  * `pandas.testing.assert_frame_equal` with `check_like=True`, so column order
-  * does not matter, and `check_dtype=False`, so the int64/float64 coercion a
-  * JSONL round trip through `pd.read_json` performs is not a difference. Float
-  * tolerance is `rtol=1e-5`. ROW order is a separate question, settled by
-  * `--unordered`, which lex-sorts both frames unless the operator declares
-  * itself order-sensitive.
+  * Runs `compare.py` on the two JSONL files [[OpExecHarness]] and
+  * [[StandaloneRunner]] wrote, and throws [[ComparatorMismatchException]]
+  * carrying the pandas diff when they differ. What counts as equal is
+  * compare.py's to say.
   *
-  * Throws [[ComparatorMismatchException]] on any non-zero exit code, carrying
-  * the pandas diff from `stderr`.
-  *
-  * Python resolution mirrors [[StandaloneRunner.resolvePython]]:
-  * `UDF_PYTHON_PATH` env var first, else `python3.12` on PATH.
+  * The one thing decided here is row order: an operator that does not declare
+  * itself order-sensitive is compared with `--unordered`, which lex-sorts both
+  * frames first.
   */
 object Comparator extends LazyLogging {
 

From a1fec7c33f1b743ad427052f2dabf906f78f08cc Mon Sep 17 00:00:00 2001
From: kary zheng <kary09302005@gmail.com>
Date: Thu, 10 Sep 2026 21:00:31 -0700
Subject: [PATCH 9/9] refactor: leave this change the comparison of two tables

The two chart comparators go to a change of their own. They answer a
different question and share no code with this one.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../verify/VisualizationHtmlComparator.scala  |  93 -------------
 .../VisualizationHtmlComparatorSpec.scala     |  72 ----------
 .../verify/VisualizationJsonComparator.scala  | 123 ------------------
 3 files changed, 288 deletions(-)
 delete mode 100644 workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/VisualizationHtmlComparator.scala
 delete mode 100644 workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/VisualizationHtmlComparatorSpec.scala
 delete mode 100644 workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/VisualizationJsonComparator.scala

diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/VisualizationHtmlComparator.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/VisualizationHtmlComparator.scala
deleted file mode 100644
index 2455833577b..00000000000
--- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/VisualizationHtmlComparator.scala
+++ /dev/null
@@ -1,93 +0,0 @@
-/*
- * 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.util.JSONUtils.objectMapper
-
-import java.nio.charset.StandardCharsets
-import java.nio.file.{Files, Path}
-
-object VisualizationHtmlComparator {
-
-  /** A pandas Styler namespaces its CSS with a uuid drawn per Styler instance, so
-    * the same table rendered twice differs in every `id=` and every selector even
-    * though the markup is identical. The uuid carries no information about the
-    * table — it only keeps two tables on one page from colliding — so it is
-    * normalized away before comparing. Only the random prefix is replaced: the
-    * `_row0_col0` suffix that identifies the cell stays, so a genuine structural
-    * difference still fails.
-    */
-  private val StylerUuid = "T_[0-9a-f]+".r
-
-  /** The standalone script writes its page with Python's text mode, which on Windows
-    * turns every newline into CRLF, while the runtime path carries the same markup
-    * through JSONL untouched and so keeps LF. The line ending is the platform writing
-    * the file rather than anything the operator chose, so it is normalized away too.
-    */
-  private val LineEnding = "\r\n|\r".r
-
-  private def normalize(html: String): String =
-    StylerUuid.replaceAllIn(LineEnding.replaceAllIn(html, "\n"), "T_uuid")
-
-  def assertEqual(actualVisualizationJsonl: Path, expectedHtmlFile: Path): Unit = {
-    val actual = readActualHtml(actualVisualizationJsonl)
-    val expected = new String(Files.readAllBytes(expectedHtmlFile), StandardCharsets.UTF_8)
-
-    if (normalize(actual) != normalize(expected)) {
-      throw new VisualizationHtmlMismatchException(
-        actual = actualVisualizationJsonl,
-        expected = expectedHtmlFile,
-        actualHtml = actual,
-        expectedHtml = expected
-      )
-    }
-  }
-
-  private def readActualHtml(path: Path): String = {
-    val line = Files
-      .readAllLines(path, StandardCharsets.UTF_8)
-      .stream()
-      .filter(_.trim.nonEmpty)
-      .findFirst()
-      .orElseThrow(() => new AssertionError(s"$path is empty"))
-
-    val node = objectMapper.readTree(line)
-    val htmlNode = node.get("html-content")
-    if (htmlNode == null || htmlNode.isNull) {
-      throw new AssertionError(s"$path has no html-content field")
-    }
-    htmlNode.asText()
-  }
-}
-
-final class VisualizationHtmlMismatchException(
-    val actual: Path,
-    val expected: Path,
-    val actualHtml: String,
-    val expectedHtml: String
-) extends RuntimeException(
-      s"""Visualization HTML mismatch:
-         |  actual:   $actual
-         |  expected: $expected
-         |--- actual html ---
-         |$actualHtml
-         |--- expected html ---
-         |$expectedHtml""".stripMargin
-    )
diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/VisualizationHtmlComparatorSpec.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/VisualizationHtmlComparatorSpec.scala
deleted file mode 100644
index 32ccbee7ac6..00000000000
--- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/VisualizationHtmlComparatorSpec.scala
+++ /dev/null
@@ -1,72 +0,0 @@
-/*
- * 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.util.JSONUtils.objectMapper
-import org.scalatest.flatspec.AnyFlatSpec
-import org.scalatest.matchers.should.Matchers
-
-import java.nio.charset.StandardCharsets
-import java.nio.file.{Files, Path}
-
-// Untagged: the comparator reads two files and compares strings, with no Python
-// to shell out to, so this belongs in the plain unit job.
-class VisualizationHtmlComparatorSpec extends AnyFlatSpec with Matchers {
-
-  private val dir: Path = Files.createTempDirectory("visualization-html-comparator-spec-")
-
-  /** The runtime path's side: the markup as one `html-content` field of a JSONL row. */
-  private def writeActual(name: String, html: String): Path = {
-    val node = objectMapper.createObjectNode()
-    node.put("html-content", html)
-    val p = dir.resolve(name)
-    Files.write(p, objectMapper.writeValueAsBytes(node))
-    p
-  }
-
-  /** The standalone path's side: the markup as the script wrote it to a file. */
-  private def writeExpected(name: String, html: String): Path = {
-    val p = dir.resolve(name)
-    Files.write(p, html.getBytes(StandardCharsets.UTF_8))
-    p
-  }
-
-  private val page = "<html>\n<body>\n<p>chart</p>\n</body>\n</html>\n"
-
-  "VisualizationHtmlComparator" should "accept two sides that differ only in line endings" in {
-    val actual = writeActual("crlf-actual.jsonl", page)
-    val expected = writeExpected("crlf-expected.html", page.replace("\n", "\r\n"))
-    noException should be thrownBy VisualizationHtmlComparator.assertEqual(actual, expected)
-  }
-
-  it should "accept two sides that differ only in a Styler uuid" in {
-    val actual = writeActual("styler-actual.jsonl", "<td id=\"T_a1b2c3_row0_col0\">1</td>")
-    val expected = writeExpected("styler-expected.html", "<td id=\"T_9f8e7d_row0_col0\">1</td>")
-    noException should be thrownBy VisualizationHtmlComparator.assertEqual(actual, expected)
-  }
-
-  it should "still reject markup that differs in more than its line endings" in {
-    val actual = writeActual("differ-actual.jsonl", page)
-    val expected =
-      writeExpected("differ-expected.html", page.replace("chart", "table").replace("\n", "\r\n"))
-    a[VisualizationHtmlMismatchException] should be thrownBy VisualizationHtmlComparator
-      .assertEqual(actual, expected)
-  }
-}
diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/VisualizationJsonComparator.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/VisualizationJsonComparator.scala
deleted file mode 100644
index f559b2c39e6..00000000000
--- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/VisualizationJsonComparator.scala
+++ /dev/null
@@ -1,123 +0,0 @@
-/*
- * 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.typesafe.scalalogging.LazyLogging
-import org.apache.texera.amber.util.JSONUtils.objectMapper
-import org.apache.texera.amber.util.python.PythonWorkerPool
-
-import java.nio.file.{Files, Path, StandardCopyOption}
-import scala.collection.mutable.ArrayBuffer
-import scala.sys.process._
-
-/**
-  * Compares the Plotly figure the two paths render, via `compare.py`'s
-  * `--plotly` mode.
-  *
-  * Shares that script — and therefore [[Comparator]]'s pool — rather than
-  * carrying one of its own: a worker is bound to the script it was launched
-  * with, so a separate script would mean a separate pool of interpreters for
-  * what is the same job, comparing one operator's two outputs. One comparison
-  * pool serves both output shapes.
-  */
-object VisualizationJsonComparator extends LazyLogging {
-
-  private val ScriptResourcePath = "/python/compare.py"
-
-  def assertEqual(
-      actualVisualizationJsonl: Path,
-      expectedPlotlyJson: Path,
-      pythonExe: String = resolvePython()
-  ): Unit = {
-    val (exit, stdout, stderr) =
-      compare(actualVisualizationJsonl, expectedPlotlyJson, pythonExe)
-    if (exit != 0) {
-      throw new VisualizationJsonMismatchException(
-        actual = actualVisualizationJsonl,
-        expected = expectedPlotlyJson,
-        exitCode = exit,
-        stdout = stdout,
-        stderr = stderr
-      )
-    }
-  }
-
-  // Pooled worker first, one-shot CLI as the fallback and as the behavior
-  // selected by TEXERA_TEST_PYTHON_WORKER=0. Both run the same
-  // `_run_plotly_comparison`, so results are identical.
-  private def compare(actual: Path, expected: Path, pythonExe: String): (Int, String, String) = {
-    if (PythonWorkerPool.enabled) {
-      try {
-        val req = objectMapper.createObjectNode()
-        req.put("kind", "plotly")
-        req.put("actual", actual.toString)
-        req.put("expected", expected.toString)
-        val o = PythonWorkerPool.run(ScriptResourcePath, Seq("--serve"), pythonExe, req)
-        return (o.exit, o.stdout, o.stderr)
-      } catch {
-        case e: PythonWorkerPool.WorkerDiedException =>
-          logger.warn(
-            s"Comparator worker unavailable; falling back to one-shot CLI: ${e.getMessage}"
-          )
-      }
-    }
-    runCli(actual, expected, pythonExe)
-  }
-
-  private def runCli(actual: Path, expected: Path, pythonExe: String): (Int, String, String) = {
-    val scriptPath = extractScript()
-    val outBuf = ArrayBuffer.empty[String]
-    val errBuf = ArrayBuffer.empty[String]
-    val processLogger = ProcessLogger(line => outBuf += line, line => errBuf += line)
-
-    val exit = Process(
-      Seq(pythonExe, scriptPath.toString, "--plotly", actual.toString, expected.toString)
-    ).!(processLogger)
-    (exit, outBuf.mkString("\n"), errBuf.mkString("\n"))
-  }
-
-  private def extractScript(): Path = {
-    val stream = getClass.getResourceAsStream(ScriptResourcePath)
-    require(stream != null, s"compare.py not found at $ScriptResourcePath")
-    try {
-      val tmp = Files.createTempFile("compare-", ".py")
-      Files.copy(stream, tmp, StandardCopyOption.REPLACE_EXISTING)
-      tmp.toFile.deleteOnExit()
-      tmp
-    } finally stream.close()
-  }
-
-  private def resolvePython(): String =
-    sys.env.get("UDF_PYTHON_PATH").filter(_.nonEmpty).getOrElse("python3.12")
-}
-
-final class VisualizationJsonMismatchException(
-    val actual: Path,
-    val expected: Path,
-    val exitCode: Int,
-    val stdout: String,
-    val stderr: String
-) extends RuntimeException(
-      s"""Visualization JSON mismatch (compare.py --plotly exit $exitCode):
-         |  actual:   $actual
-         |  expected: $expected
-         |--- stderr ---
-         |$stderr""".stripMargin
-    )