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 01/16] 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 02/16] 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 03/16] 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 04/16] 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 05/16] 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 06/16] 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 07/16] 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 08/16] 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 09/16] 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
-    )

From 791171ea786158bf864ac8b27ae2f4a54c95dde4 Mon Sep 17 00:00:00 2001
From: kary zheng <kary09302005@gmail.com>
Date: Fri, 18 Sep 2026 15:56:24 -0700
Subject: [PATCH 10/16] fix(verify): compare every chart the two paths drew,
 not the first

The plotly comparison read one figure from each side: the first row of the
runtime path's output, and the exported script's file as a single figure. An
operator that draws a chart per row had every chart after the first go unread,
so the exported script could draw the wrong one and the run still agreed.

Both sides are read as sequences now. The exported side accepts a lone figure,
which is what a one-chart operator writes, or an array. A length difference is
reported as one, and a value difference names which chart it was.

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

diff --git a/workflow-compiling-service/src/test/resources/python/compare.py b/workflow-compiling-service/src/test/resources/python/compare.py
index 74f31494b0b..4af8920757f 100644
--- a/workflow-compiling-service/src/test/resources/python/compare.py
+++ b/workflow-compiling-service/src/test/resources/python/compare.py
@@ -324,21 +324,28 @@ def _run_comparison(
     return None
 
 
-def _load_actual_plot(path) -> dict:
+def _load_actual_plots(path) -> list:
+    """Every chart the run drew, one per row of the runtime path's output."""
     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:
+        lines = [raw for raw in fh if raw.strip()]
+    if not lines:
         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")
+    plots = []
+    for number, line in enumerate(lines, start=1):
+        row = json.loads(line)
+        if "json-content" in row and row["json-content"]:
+            value = row["json-content"]
+            plots.append(json.loads(value) if isinstance(value, str) else value)
+        elif "html-content" in row and row["html-content"]:
+            plots.append(_plotly_payload_from_html(row["html-content"]))
+        else:
+            raise AssertionError(
+                f"{path} row {number} has neither html-content nor json-content"
+            )
+    return plots
 
 
 def _plotly_payload_from_html(html: str) -> dict:
@@ -367,12 +374,18 @@ def _plotly_payload_from_html(html: str) -> dict:
     return {"data": args[1], "layout": args[2]}
 
 
-def _load_expected_plot(path) -> dict:
+def _load_expected_plots(path) -> list:
+    """Every chart the exported script drew. An operator that draws one writes a
+    lone figure; one that draws a chart per row writes the sequence."""
     import json
 
     with open(path, "r", encoding="utf-8") as fh:
         value = json.load(fh)
-    return {"data": value.get("data", []), "layout": value.get("layout", {})}
+    figures = value if isinstance(value, list) else [value]
+    return [
+        {"data": fig.get("data", []), "layout": fig.get("layout", {})}
+        for fig in figures
+    ]
 
 
 def _strip_unstable(value):
@@ -410,19 +423,28 @@ def _run_plotly_comparison(actual_path, expected_path) -> "str | None":
     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),
-        ]
-    )
+    actual = [_strip_unstable(p) for p in _load_actual_plots(actual_path)]
+    expected = [_strip_unstable(p) for p in _load_expected_plots(expected_path)]
+
+    if len(actual) != len(expected):
+        return (
+            f"Plotly chart count mismatch: the run drew {len(actual)} and the "
+            f"exported script drew {len(expected)}"
+        )
+
+    for index, (one, other) in enumerate(zip(actual, expected)):
+        if _plots_equal(one, other):
+            continue
+        return "\n".join(
+            [
+                f"Plotly JSON mismatch on chart {index + 1} of {len(actual)}",
+                "--- actual ---",
+                json.dumps(one, indent=2, sort_keys=True),
+                "--- expected ---",
+                json.dumps(other, indent=2, sort_keys=True),
+            ]
+        )
+    return None
 
 
 def main() -> None:

From cc6a9e55071b0f63798d790c470f50951c845825 Mon Sep 17 00:00:00 2001
From: kary zheng <kary09302005@gmail.com>
Date: Fri, 18 Sep 2026 16:01:01 -0700
Subject: [PATCH 11/16] test(verify): pin that the plotly comparison reads
 every chart

Four cases run compare.py directly, since the comparison lives in the script
rather than in Scala: three charts that agree, a third pair whose second chart
differs, an exported side that drew fewer than the run, and the lone figure a
one-chart operator writes.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../translator/verify/ComparatorSpec.scala    | 93 ++++++++++++++++++-
 1 file changed, 92 insertions(+), 1 deletion(-)

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 4ec7066ef14..c001b404f14 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
@@ -24,7 +24,9 @@ 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}
+import java.nio.file.{Files, Path, StandardCopyOption}
+import java.util.concurrent.TimeUnit
+import scala.io.Source
 
 // Tagged @IntegrationTest: Comparator.assertEqual shells out to compare.py, so
 // this spec needs Python and must run in the Python-provisioned integration job.
@@ -189,4 +191,93 @@ class ComparatorSpec extends AnyFlatSpec with Matchers {
       )
     }.getMessage should include("missing from actual")
   }
+
+  // An operator that draws a chart per row writes one JSONL row per chart on the
+  // runtime side and an array on the exported one. Reading a single figure from
+  // each, as this did, left every chart after the first unread.
+  private def plotlyFigure(value: Int): String =
+    s"""{"data": [{"type": "indicator", "value": $value}], "layout": {}}"""
+
+  /** compare.py --plotly, as its own process: the exit code and what it said. */
+  private def runPlotly(actual: Path, expected: Path): (Int, String) = {
+    val script = Files.createTempFile("compare-", ".py")
+    script.toFile.deleteOnExit()
+    val stream = getClass.getResourceAsStream("/python/compare.py")
+    require(stream != null, "compare.py not found at /python/compare.py")
+    try Files.copy(stream, script, StandardCopyOption.REPLACE_EXISTING)
+    finally stream.close()
+
+    val python = sys.env.get("UDF_PYTHON_PATH").filter(_.nonEmpty).getOrElse("python3")
+    val process = new ProcessBuilder(
+      python,
+      script.toString,
+      "--plotly",
+      actual.toString,
+      expected.toString
+    ).redirectErrorStream(true).start()
+    val said = Source.fromInputStream(process.getInputStream).mkString
+    process.waitFor(120, TimeUnit.SECONDS)
+    (process.exitValue(), said)
+  }
+
+  "Comparator's plotly comparison" should "read every chart on both sides" in {
+    val dir = Files.createTempDirectory("comparator-spec-plotly-all-")
+    val actual = writeLines(
+      dir,
+      "actual.jsonl",
+      Seq(1, 2, 3).map(v => s"""{"json-content": ${plotlyFigure(v)}}""")
+    )
+    val expected = writeLines(
+      dir,
+      "expected.json",
+      Seq(Seq(1, 2, 3).map(plotlyFigure).mkString("[", ",", "]"))
+    )
+    val (exit, said) = runPlotly(actual, expected)
+    withClue(s"compare.py said:\n$said") { exit shouldBe 0 }
+  }
+
+  it should "catch a chart that differs after the first" in {
+    val dir = Files.createTempDirectory("comparator-spec-plotly-second-")
+    val actual = writeLines(
+      dir,
+      "actual.jsonl",
+      Seq(1, 2, 3).map(v => s"""{"json-content": ${plotlyFigure(v)}}""")
+    )
+    val expected = writeLines(
+      dir,
+      "expected.json",
+      Seq(Seq(1, 99, 3).map(plotlyFigure).mkString("[", ",", "]"))
+    )
+    val (exit, said) = runPlotly(actual, expected)
+    withClue(s"compare.py said:\n$said") {
+      exit should not be 0
+      said should include("chart 2 of 3")
+    }
+  }
+
+  it should "catch an exported script that drew fewer charts than the run" in {
+    val dir = Files.createTempDirectory("comparator-spec-plotly-count-")
+    val actual = writeLines(
+      dir,
+      "actual.jsonl",
+      Seq(1, 2, 3).map(v => s"""{"json-content": ${plotlyFigure(v)}}""")
+    )
+    val expected = writeLines(dir, "expected.json", Seq(plotlyFigure(1)))
+    val (exit, said) = runPlotly(actual, expected)
+    withClue(s"compare.py said:\n$said") {
+      exit should not be 0
+      said should include("drew 3")
+      said should include("drew 1")
+    }
+  }
+
+  // The operators that draw one chart write a lone figure, and there are far
+  // more of those than there are of the other kind.
+  it should "still read a lone figure as the one chart it is" in {
+    val dir = Files.createTempDirectory("comparator-spec-plotly-one-")
+    val actual = writeLines(dir, "actual.jsonl", Seq(s"""{"json-content": ${plotlyFigure(7)}}"""))
+    val expected = writeLines(dir, "expected.json", Seq(plotlyFigure(7)))
+    val (exit, said) = runPlotly(actual, expected)
+    withClue(s"compare.py said:\n$said") { exit shouldBe 0 }
+  }
 }

From b17651befe6714594010667e9c6fd928127b06a4 Mon Sep 17 00:00:00 2001
From: kary zheng <kary09302005@gmail.com>
Date: Fri, 18 Sep 2026 16:13:41 -0700
Subject: [PATCH 12/16] fix(verify): compare what the two sides actually wrote

Five ways the comparison agreed with itself rather than with the outputs.

The column list was never checked. The comparison took its columns from the
actual side and selected those out of the expected one, so a column only the
expected side carried was never looked at, and `check_like` sorted the columns
before comparing, so a reordering passed. Column order decides what a positional
UDF reads, what a file export writes, and what code asking for the first column
gets, so both halves of the list are compared now.

A boolean setting in a Plotly figure fell into the numeric branch, `bool` being
a subclass of `int` in Python, and True compared equal to 1 through isclose. A
column whose declared type changed from boolean to integer reaches a figure as
exactly that difference.

A model was asked about its own fitted feature names. Two models fitted on
different columns still answer alike on a probe carrying both, so the
predictions agreed while the models did not. The names are compared first.

A column the schema declares as text was read with `dtype=str`, which settles
the two sides on one spelling but turns the JSON number 6 into "6" as readily as
it leaves "6" alone. Those columns are re-read from the raw JSON, where a number
is still a number.

An integer column one side widened to a float passed as the integer the schema
asked for. 6 and 6.0 do not survive the same: a later cast to text writes "6"
from one and "6.0" from the other. Both sides widening alike is left alone,
because where the engine's own path is a Python operator its table goes through
pandas too, and the two still agree. A declared boolean column is read the same
way, numpy having counted True as 1 under a comparison told not to check dtypes.

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

diff --git a/workflow-compiling-service/src/test/resources/python/compare.py b/workflow-compiling-service/src/test/resources/python/compare.py
index 4af8920757f..3c424e1452a 100644
--- a/workflow-compiling-service/src/test/resources/python/compare.py
+++ b/workflow-compiling-service/src/test/resources/python/compare.py
@@ -143,9 +143,25 @@ def _compare_model_predictions(actual, expected, model_cols, probe_path) -> None
             # 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]
+
+            # What a model was fitted on is part of the model, and each side was
+            # being asked about its own features. Two models fitted on different
+            # columns still answer alike on a probe that carries both, so the
+            # predictions agreed while the models did not. One side having names
+            # where the other has none is the same divergence: it says the two
+            # were fitted on differently shaped input.
+            listed = None if names is None else list(names)
+            listed_e = None if names_e is None else list(names_e)
+            if listed != listed_e:
+                raise AssertionError(
+                    f"model column {col!r} row {i}: fitted feature names differ\n"
+                    f"  actual:   {listed}\n"
+                    f"  expected: {listed_e}"
+                )
+
+            x_a = probe[listed] if listed is not None else probe.iloc[:, 0]
+            x_e = probe[listed_e] if listed_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))
@@ -193,13 +209,15 @@ def _string_columns(actual_path: str) -> dict:
     }
 
 
-def _exact_integers(path: str, columns: list) -> dict:
-    """The named columns re-read with Python's json, whose integers are exact.
+def _declared_strings(path: str, columns: list) -> dict:
+    """The named columns re-read with Python's json, which keeps a string a
+    string and leaves a number a number.
 
-    `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.
+    Reading them with `dtype=str` settles the two sides on one spelling, which
+    is what a text column needs, but it settles too much: it turns the JSON
+    number 6 into "6" as readily as it leaves "6" alone, so a side that wrote a
+    number where the engine declared text compared equal to the text. The
+    engine's own output is always text here, so a number is the divergence.
     """
     import json
 
@@ -216,17 +234,128 @@ def _exact_integers(path: str, columns: list) -> dict:
                 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):
+                elif isinstance(cell, str):
+                    values[column].append(cell)
+                else:
                     raise AssertionError(
-                        f"column '{column}' is declared integral but holds {cell!r} in {path}"
+                        f"column '{column}' is declared text 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()}
+    return {column: pd.array(cells, dtype="string") for column, cells in values.items()}
+
+
+def _raw_column(path: str, columns: list) -> dict:
+    """The named columns as Python's json read them, where an integer is exact
+    and a float is still a float."""
+    import json
+
+    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:
+                values[column].append(row.get(column))
+    return values
+
+
+def _json_kind(cell) -> str:
+    if isinstance(cell, bool):
+        return "boolean"
+    if isinstance(cell, int):
+        return "integer"
+    if isinstance(cell, float):
+        return "float"
+    return type(cell).__name__
+
+
+def _assert_sides_agree(actual: dict, expected: dict, columns: list, declared: str) -> None:
+    """Fail where the two sides wrote a column as different JSON kinds.
+
+    What the comparison is asked is whether the paths match, not whether pandas
+    kept the declared type: where the engine's own path is a Python operator its
+    table goes through pandas too, so a hole widens the column on both paths
+    alike and the two still agree. One side widening on its own is the
+    divergence, and it is the one the digits hide.
+    """
+    for column in columns:
+        kinds_a = {_json_kind(c) for c in actual[column] if c is not None}
+        kinds_e = {_json_kind(c) for c in expected[column] if c is not None}
+        if kinds_a != kinds_e:
+            raise AssertionError(
+                f"column '{column}' is declared {declared} and the two sides wrote it "
+                f"differently\n"
+                f"  actual:   {sorted(kinds_a) or ['all null']}\n"
+                f"  expected: {sorted(kinds_e) or ['all null']}"
+            )
+
+
+def _exact_integers(actual_path: str, expected_path: str, columns: list) -> tuple:
+    """Both sides' named columns re-read with Python's json, whose integers are
+    exact, having first agreed on what each column holds.
+
+    `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.
+
+    One side writing a float where the other wrote an integer is the divergence
+    the digits hide. 6.0 was read as the 6 the schema asked for, on the grounds
+    that both spell the same integer, but they do not survive the same: a later
+    cast to text writes "6" from one and "6.0" from the other.
+
+    Both sides writing a float is not a divergence; see [[_assert_sides_agree]].
+    """
+    import pandas as pd
+
+    actual = _raw_column(actual_path, columns)
+    expected = _raw_column(expected_path, columns)
+    _assert_sides_agree(actual, expected, columns, "integral")
+
+    def arrays(values):
+        out = {}
+        for column, cells in values.items():
+            # A column either side widened is compared as the float it now is,
+            # exactly rather than by tolerance; one both sides kept integral is
+            # carried in the nullable integer, where a long stays exact.
+            widened = any(isinstance(c, float) for c in cells)
+            dtype = "float64" if widened else "Int64"
+            out[column] = pd.array(
+                [pd.NA if c is None else c for c in cells], dtype=dtype
+            )
+        return out
+
+    return arrays(actual), arrays(expected)
+
+
+def _declared_booleans(actual_path: str, expected_path: str, columns: list) -> tuple:
+    """Both sides' declared BOOLEAN columns, having first agreed on what each
+    one holds.
+
+    numpy counts True as 1, and `assert_frame_equal` is asked not to check
+    dtypes, so a column of booleans compared equal to a column of ones. The
+    engine says of a BOOLEAN column that it holds true and false, and a side
+    that wrote a number there has changed the type of the answer.
+
+    As with the integers, both sides widening alike is not a divergence: a hole
+    costs a pandas boolean column its type on whichever path went through
+    pandas, and if both did they still agree.
+    """
+    import pandas as pd
+
+    actual = _raw_column(actual_path, columns)
+    expected = _raw_column(expected_path, columns)
+    _assert_sides_agree(actual, expected, columns, "boolean")
+
+    def arrays(values):
+        out = {}
+        for column, cells in values.items():
+            kept = all(c is None or isinstance(c, bool) for c in cells)
+            cells = [pd.NA if c is None else c for c in cells]
+            out[column] = pd.array(cells, dtype="boolean" if kept else "float64")
+        return out
+
+    return arrays(actual), arrays(expected)
 
 
 def _run_comparison(
@@ -254,6 +383,21 @@ def _run_comparison(
     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)
 
+    # `dtype=str` above settles the spelling, which a text column needs, but it
+    # also turns a JSON number into text, so a side that wrote 6 where the engine
+    # declared text compared equal to "6". Re-read those columns from the raw
+    # JSON, where a number is still a number and so still a divergence.
+    raw_str_cols = [
+        name for name in str_cols if name in actual.columns and name in expected.columns
+    ]
+    if raw_str_cols:
+        try:
+            for frame, path in ((actual, actual_path), (expected, expected_path)):
+                for column, values in _declared_strings(path, raw_str_cols).items():
+                    frame[column] = values
+        except AssertionError as exc:
+            return str(exc)
+
     # 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 = [
@@ -263,11 +407,29 @@ def _run_comparison(
     ]
     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
+            exact_a, exact_e = _exact_integers(actual_path, expected_path, int_cols)
+        except AssertionError as exc:
+            return str(exc)
+        for column in int_cols:
+            actual[column] = exact_a[column]
+            expected[column] = exact_e[column]
+
+    # numpy counts True as 1 and the dtype check is off, so a column of booleans
+    # compared equal to a column of ones. The engine says a BOOLEAN column holds
+    # true and false; a side that wrote a number there changed the answer's type.
+    bool_cols = [
+        name
+        for name, kind in declared.items()
+        if kind == "boolean" and name in actual.columns and name in expected.columns
+    ]
+    if bool_cols:
+        try:
+            bool_a, bool_e = _declared_booleans(actual_path, expected_path, bool_cols)
         except AssertionError as exc:
             return str(exc)
+        for column in bool_cols:
+            actual[column] = bool_a[column]
+            expected[column] = bool_e[column]
 
     # Model columns: compare behavior (predictions) rather than bytes, then drop
     # the raw columns so the frame comparison covers everything else exactly.
@@ -283,6 +445,31 @@ def _run_comparison(
         actual = actual.drop(columns=ignore_cols, errors="ignore")
         expected = expected.drop(columns=ignore_cols, errors="ignore")
 
+    # The column list is part of the answer, and neither half of it was being
+    # checked. The comparison took its columns from `actual` and selected those
+    # out of `expected`, so a column only `expected` carried was never looked
+    # at. Order went unchecked too, under `check_like`, and order is what a
+    # positional UDF reads by, what a file export writes out, and what code
+    # asking for the first column gets.
+    if list(actual.columns) != list(expected.columns):
+        only_actual = [c for c in actual.columns if c not in set(expected.columns)]
+        only_expected = [c for c in expected.columns if c not in set(actual.columns)]
+        detail = []
+        if only_actual:
+            detail.append(f"  only in actual:   {only_actual}")
+        if only_expected:
+            detail.append(f"  only in expected: {only_expected}")
+        if not detail:
+            detail.append("  the same columns in a different order")
+        return "\n".join(
+            [
+                "column mismatch",
+                f"  actual:   {list(actual.columns)}",
+                f"  expected: {list(expected.columns)}",
+            ]
+            + detail
+        )
+
     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
@@ -303,19 +490,20 @@ def _run_comparison(
     # 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]
+    # No `check_like`: it sorts both frames' columns before comparing, which is
+    # what let a reordering through. The lists are known equal by here, so each
+    # of these two selections holds the same columns in the same order.
     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[loose_cols],
             expected[loose_cols],
-            check_like=True,
             check_dtype=False,
             rtol=1e-5,
         )
@@ -404,6 +592,13 @@ def _strip_unstable(value):
 def _plots_equal(actual, expected) -> bool:
     import math
 
+    # Before the numeric branch, because `bool` is a subclass of `int` and a
+    # boolean fell into it: True and 1 went through math.isclose and compared
+    # equal. A trace says things with booleans that a number does not, and a
+    # column whose declared type changed from boolean to integer reaches the
+    # figure as exactly this difference.
+    if isinstance(actual, bool) or isinstance(expected, bool):
+        return isinstance(actual, bool) and isinstance(expected, bool) and actual == expected
     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):
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 c001b404f14..66ddcf21fa1 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
@@ -154,16 +154,101 @@ class ComparatorSpec extends AnyFlatSpec with Matchers {
     )
   }
 
-  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.
+  it should "reject an integer column only one side widened to a float" in {
+    // This passed on the grounds that 6.0 and 6 spell the same integer. They do
+    // not survive the same: a later cast to text writes "6" from one and "6.0"
+    // from the other.
     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}"""))
+    intercept[ComparatorMismatchException] {
+      Comparator.assertEqual(a, b)
+    }.getMessage should include("the two sides wrote it differently")
+  }
+
+  it should "accept an integer column both sides widened alike" in {
+    // Where the engine's own path is a Python operator its table goes through
+    // pandas too, so a hole widens the column on both paths and the two still
+    // agree. What is asked here is whether the paths match, not whether pandas
+    // kept the type.
+    val dir = Files.createTempDirectory("comparator-spec-int-both-widened-")
+    // writeLongs for the schema sidecar it leaves; the rows are then rewritten
+    // as the widened form both paths would have produced.
+    val a = writeLongs(dir, "a.jsonl", Seq(6L))
+    Files.writeString(a, "{\"n\":6.0}\n")
+    val b = writeLines(dir, "b.jsonl", Seq("""{"n":6.0}"""))
     noException should be thrownBy Comparator.assertEqual(a, b)
   }
 
+  // The comparison took its columns from the actual side and selected those out
+  // of the expected one, so a column only the expected side carried was never
+  // looked at.
+  it should "report a column only one side carries" in {
+    val dir = Files.createTempDirectory("comparator-spec-extra-column-")
+    val a = writeLines(dir, "a.jsonl", Seq("""{"id":1,"name":"alice"}"""))
+    val b = writeLines(dir, "b.jsonl", Seq("""{"id":1,"name":"alice","extra":9}"""))
+
+    intercept[ComparatorMismatchException] {
+      Comparator.assertEqual(a, b)
+    }.getMessage should include("only in expected")
+
+    intercept[ComparatorMismatchException] {
+      Comparator.assertEqual(b, a)
+    }.getMessage should include("only in actual")
+  }
+
+  // Column order is part of the schema: it decides what a positional UDF reads,
+  // what a file export writes, and what code asking for the first column gets.
+  it should "treat column-reordered files as unequal" in {
+    val dir = Files.createTempDirectory("comparator-spec-column-order-")
+    val a = writeLines(dir, "a.jsonl", Seq("""{"a":1,"b":2}"""))
+    val b = writeLines(dir, "b.jsonl", Seq("""{"b":2,"a":1}"""))
+    intercept[ComparatorMismatchException] {
+      Comparator.assertEqual(a, b)
+    }.getMessage should include("different order")
+  }
+
+  // Reading both sides with `dtype=str` settles them on one spelling, which a
+  // text column needs, but it reads the JSON number 6 as "6" just as readily, so
+  // a side that wrote a number where the engine declared text compared equal to
+  // the text. What the file actually holds is what separates them.
+  it should "reject a number written where the schema declares text" in {
+    val dir = Files.createTempDirectory("comparator-spec-text-number-")
+    val textSchema = Schema().add(new Attribute("code", AttributeType.STRING))
+    val a = dir.resolve("a.jsonl")
+    TupleIO.writeTuples(
+      a,
+      Iterator(Tuple.builder(textSchema).add(textSchema.getAttribute("code"), "6").build()),
+      textSchema
+    )
+    val b = writeLines(dir, "b.jsonl", Seq("""{"code":6}"""))
+
+    intercept[ComparatorMismatchException] {
+      Comparator.assertEqual(a, b)
+    }.getMessage should include("declared text but holds 6")
+  }
+
+  // numpy counts True as 1 and the frame comparison is asked not to check
+  // dtypes, so a column of booleans compared equal to a column of ones. The
+  // engine says a BOOLEAN column holds true and false.
+  it should "reject a number written where the schema declares a boolean" in {
+    val dir = Files.createTempDirectory("comparator-spec-bool-number-")
+    val boolSchema = Schema().add(new Attribute("flag", AttributeType.BOOLEAN))
+    val a = dir.resolve("a.jsonl")
+    TupleIO.writeTuples(
+      a,
+      Iterator(
+        Tuple.builder(boolSchema).add(boolSchema.getAttribute("flag"), Boolean.box(true)).build()
+      ),
+      boolSchema
+    )
+    val b = writeLines(dir, "b.jsonl", Seq("""{"flag":1}"""))
+
+    intercept[ComparatorMismatchException] {
+      Comparator.assertEqual(a, b)
+    }.getMessage should include("the two sides wrote it differently")
+  }
+
   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,
@@ -280,4 +365,23 @@ class ComparatorSpec extends AnyFlatSpec with Matchers {
     val (exit, said) = runPlotly(actual, expected)
     withClue(s"compare.py said:\n$said") { exit shouldBe 0 }
   }
+
+  // `bool` is a subclass of `int` in Python, so a boolean setting fell into the
+  // numeric branch and True compared equal to 1. A column whose declared type
+  // changed from boolean to integer reaches a figure as exactly that.
+  it should "treat a boolean setting and the number 1 as different" in {
+    val dir = Files.createTempDirectory("comparator-spec-plotly-bool-")
+    val actual = writeLines(
+      dir,
+      "actual.jsonl",
+      Seq("""{"json-content": {"data": [{"visible": true}], "layout": {}}}""")
+    )
+    def expected(visible: String): Path =
+      writeLines(dir, s"expected-$visible.json", Seq(s"""{"data": [{"visible": $visible}], "layout": {}}"""))
+
+    val (asNumber, said) = runPlotly(actual, expected("1"))
+    withClue(s"compare.py said:\n$said") { asNumber should not be 0 }
+    val (asBoolean, alsoSaid) = runPlotly(actual, expected("true"))
+    withClue(s"compare.py said:\n$alsoSaid") { asBoolean shouldBe 0 }
+  }
 }

From b54fc2da5ab62c3f68e597cf6129cddeb1b901d3 Mon Sep 17 00:00:00 2001
From: kary zheng <kary09302005@gmail.com>
Date: Fri, 18 Sep 2026 16:34:32 -0700
Subject: [PATCH 13/16] test(verify): read a page of charts the run drew as one
 row

A chart per row is one of the two shapes a multi-chart operator hands over.
The other is a bullet or gauge chart, which draws its charts onto one page and
yields that page as a single row, while its script writes the whole set to its
page and only the first of them as a figure. Reading one payload per row left
every chart after the first on such a page uncompared, and the page the script
wrote could not be read at all.

Every Plotly.newPlot call on a row now counts, and the expected side may be a
page as readily as a figure or a list of them. The one reformat is scalafmt on
a line this file already carried.

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

diff --git a/workflow-compiling-service/src/test/resources/python/compare.py b/workflow-compiling-service/src/test/resources/python/compare.py
index 3c424e1452a..96c89e059d5 100644
--- a/workflow-compiling-service/src/test/resources/python/compare.py
+++ b/workflow-compiling-service/src/test/resources/python/compare.py
@@ -49,12 +49,15 @@
                 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.
+                a JSONL with `html-content` or `json-content`, a chart per row
+                or a page of them in one row; every `Plotly.newPlot(...)`
+                payload counts. The expected side is the standalone path's
+                `fig.write_json(...)`, one figure or a list of them, or its page
+                when that is where the whole set landed. The two are compared in
+                order, so a chart one path drew and the other did not is a
+                mismatch. 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
@@ -513,7 +516,9 @@ def _run_comparison(
 
 
 def _load_actual_plots(path) -> list:
-    """Every chart the run drew, one per row of the runtime path's output."""
+    """Every chart the run drew: a row per chart, or a row holding a page of
+    them, which is how an operator that draws a chart per input row hands them
+    over."""
     import json
 
     with open(path, "r", encoding="utf-8") as fh:
@@ -528,7 +533,7 @@ def _load_actual_plots(path) -> list:
             value = row["json-content"]
             plots.append(json.loads(value) if isinstance(value, str) else value)
         elif "html-content" in row and row["html-content"]:
-            plots.append(_plotly_payload_from_html(row["html-content"]))
+            plots.extend(_plotly_payloads_from_html(row["html-content"]))
         else:
             raise AssertionError(
                 f"{path} row {number} has neither html-content nor json-content"
@@ -536,8 +541,11 @@ def _load_actual_plots(path) -> list:
     return plots
 
 
-def _plotly_payload_from_html(html: str) -> dict:
-    """Pull the data/layout arguments out of the first Plotly.newPlot(...) call.
+def _plotly_payloads_from_html(html: str) -> list:
+    """Pull the data/layout arguments out of every Plotly.newPlot(...) call.
+
+    Every call, because a page can hold a chart per input row and reading the
+    first would leave the rest of them uncompared.
 
     Scanned with a JSON decoder rather than a regex because the payload is
     arbitrary nested JSON that no bracket-matching pattern handles reliably.
@@ -545,30 +553,39 @@ def _plotly_payload_from_html(html: str) -> dict:
     import json
 
     marker = "Plotly.newPlot("
-    start = html.find(marker)
-    if start < 0:
+    index = html.find(marker)
+    if index < 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
+    plots: list = []
+    while index >= 0:
+        index += 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
+        plots.append({"data": args[1], "layout": args[2]})
+        index = html.find(marker, index)
 
-    return {"data": args[1], "layout": args[2]}
+    return plots
 
 
 def _load_expected_plots(path) -> list:
     """Every chart the exported script drew. An operator that draws one writes a
-    lone figure; one that draws a chart per row writes the sequence."""
+    lone figure; one that draws a chart per row writes the sequence, or writes
+    them to its page and only the first as a figure, in which case the page is
+    what this is handed."""
     import json
 
     with open(path, "r", encoding="utf-8") as fh:
-        value = json.load(fh)
+        text = fh.read()
+    if str(path).endswith(".html"):
+        return _plotly_payloads_from_html(text)
+    value = json.loads(text)
     figures = value if isinstance(value, list) else [value]
     return [
         {"data": fig.get("data", []), "layout": fig.get("layout", {})}
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 66ddcf21fa1..d7d5dcc156d 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
@@ -21,6 +21,7 @@ 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.apache.texera.amber.util.JSONUtils.objectMapper
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
 
@@ -366,6 +367,44 @@ class ComparatorSpec extends AnyFlatSpec with Matchers {
     withClue(s"compare.py said:\n$said") { exit shouldBe 0 }
   }
 
+  // A bullet or gauge chart draws its charts onto one page and yields that page
+  // as a single row, and its script writes the whole set to its page and only
+  // the first of them as a figure. Both sides are pages then, and every call on
+  // them is a chart to compare.
+  private def plotlyPage(values: Int*): String =
+    values
+      .map(v =>
+        s"""<div><script>Plotly.newPlot("div$v", """ +
+          s"""[{"type": "indicator", "value": $v}], {}, {"responsive": true})</script></div>"""
+      )
+      .mkString
+
+  /** A page as the runtime path hands it over: one JSONL row, markup escaped. */
+  private def plotlyPageRow(values: Int*): String = {
+    val node = objectMapper.createObjectNode()
+    node.put("html-content", plotlyPage(values: _*))
+    objectMapper.writeValueAsString(node)
+  }
+
+  it should "read every chart on a page the run drew as one row" in {
+    val dir = Files.createTempDirectory("comparator-spec-plotly-page-")
+    val actual = writeLines(dir, "actual.jsonl", Seq(plotlyPageRow(1, 2, 3)))
+    val expected = writeLines(dir, "expected.html", Seq(plotlyPage(1, 2, 3)))
+    val (exit, said) = runPlotly(actual, expected)
+    withClue(s"compare.py said:\n$said") { exit shouldBe 0 }
+  }
+
+  it should "catch a chart after the first on a page the script wrote" in {
+    val dir = Files.createTempDirectory("comparator-spec-plotly-page-second-")
+    val actual = writeLines(dir, "actual.jsonl", Seq(plotlyPageRow(1, 2, 3)))
+    val expected = writeLines(dir, "expected.html", Seq(plotlyPage(1, 99, 3)))
+    val (exit, said) = runPlotly(actual, expected)
+    withClue(s"compare.py said:\n$said") {
+      exit should not be 0
+      said should include("chart 2 of 3")
+    }
+  }
+
   // `bool` is a subclass of `int` in Python, so a boolean setting fell into the
   // numeric branch and True compared equal to 1. A column whose declared type
   // changed from boolean to integer reaches a figure as exactly that.
@@ -377,7 +416,11 @@ class ComparatorSpec extends AnyFlatSpec with Matchers {
       Seq("""{"json-content": {"data": [{"visible": true}], "layout": {}}}""")
     )
     def expected(visible: String): Path =
-      writeLines(dir, s"expected-$visible.json", Seq(s"""{"data": [{"visible": $visible}], "layout": {}}"""))
+      writeLines(
+        dir,
+        s"expected-$visible.json",
+        Seq(s"""{"data": [{"visible": $visible}], "layout": {}}""")
+      )
 
     val (asNumber, said) = runPlotly(actual, expected("1"))
     withClue(s"compare.py said:\n$said") { asNumber should not be 0 }

From df57ec6f61330a3f873da5a68f84465e1c3dfc42 Mon Sep 17 00:00:00 2001
From: kary zheng <kary09302005@gmail.com>
Date: Mon, 21 Sep 2026 17:14:48 -0700
Subject: [PATCH 14/16] fix(verify): compare a classifier's predicted classes
 exactly

The model comparison chose its tolerance from the prediction's dtype, so a
classifier whose classes are numbers went through allclose. Two models that
predicted 100000 and 100001 on every row compared equal at rtol=1e-5, hiding
a change in every prediction. Only a regressor predicts a measurement where
the last bits are arithmetic, so ask sklearn what the estimator is and keep
the tolerance for that case alone.

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

diff --git a/workflow-compiling-service/src/test/resources/python/compare.py b/workflow-compiling-service/src/test/resources/python/compare.py
index 96c89e059d5..faa15c467f3 100644
--- a/workflow-compiling-service/src/test/resources/python/compare.py
+++ b/workflow-compiling-service/src/test/resources/python/compare.py
@@ -101,6 +101,12 @@ def _compare_model_predictions(actual, expected, model_cols, probe_path) -> None
     import numpy as np
     import pandas as pd
 
+    # A model column holds an sklearn estimator, so the unpickling below imports
+    # sklearn anyway; asking it what kind of estimator it handed back costs
+    # nothing more. Both are function-local so a run with no model column, which
+    # is most of them, never pays for either.
+    from sklearn.base import is_regressor
+
     if probe_path is None:
         raise AssertionError("--model-cols requires --probe with a feature set")
     probe = pd.read_json(probe_path, lines=True)
@@ -174,12 +180,23 @@ def _compare_model_predictions(actual, expected, model_cols, probe_path) -> None
                     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
+            # Only a regressor predicts a measurement, where the last bits are
+            # arithmetic and a tolerance belongs. Every other estimator predicts
+            # a label, a class or a cluster, and a different label is a
+            # different answer however near the two numbers sit: classes 100000
+            # and 100001 are one part in 1e5 apart, which allclose accepts. The
+            # estimator says which it is; the prediction's dtype cannot, since a
+            # numeric label is a number too. Anything sklearn does not call a
+            # regressor, including a non-sklearn pickle, is compared exactly.
+            tolerant = (
+                is_regressor(m_actual)
+                and is_regressor(m_expected)
+                and 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
+                if tolerant
                 else np.array_equal(pred_a, pred_e)
             )
             if not ok:
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 d7d5dcc156d..42eae420674 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
@@ -278,6 +278,80 @@ class ComparatorSpec extends AnyFlatSpec with Matchers {
     }.getMessage should include("missing from actual")
   }
 
+  /** A model column holds a pickled estimator, which only Python can fit. Writes
+    * the two one-row frames the comparison reads and the probe it predicts on,
+    * each side's model fitted on its own targets.
+    */
+  private def writeFittedModels(
+      dir: Path,
+      estimator: String,
+      actualTargets: String,
+      expectedTargets: String
+  ): (Path, Path, Path) = {
+    val program =
+      s"""import base64, json, pickle
+         |import pandas as pd
+         |from sklearn.tree import $estimator as Estimator
+         |
+         |x = pd.DataFrame({"petal_length": [1.0, 2.0, 3.0]})
+         |x.to_json(r"$dir/probe.jsonl", orient="records", lines=True)
+         |for name, y in (("actual.jsonl", $actualTargets), ("expected.jsonl", $expectedTargets)):
+         |    model = base64.b64encode(pickle.dumps(Estimator().fit(x, y))).decode()
+         |    with open(rf"$dir/{name}", "w") as handle:
+         |        handle.write(json.dumps({"model": model, "score": 1.0}) + "\\n")
+         |""".stripMargin
+    val script = Files.createTempFile("comparator-spec-fit-", ".py")
+    script.toFile.deleteOnExit()
+    Files.writeString(script, program)
+
+    val python = sys.env.get("UDF_PYTHON_PATH").filter(_.nonEmpty).getOrElse("python3")
+    val process =
+      new ProcessBuilder(python, script.toString).redirectErrorStream(true).start()
+    val said = Source.fromInputStream(process.getInputStream).mkString
+    process.waitFor(120, TimeUnit.SECONDS)
+    require(process.exitValue() == 0, s"fitting the models failed:\n$said")
+    (dir.resolve("actual.jsonl"), dir.resolve("expected.jsonl"), dir.resolve("probe.jsonl"))
+  }
+
+  // A classifier predicts a class, and a class that changed is a different
+  // answer however near the two labels sit. Numbered classes went through the
+  // numeric tolerance, where 100000 and 100001 are one part in 1e5 apart and
+  // compared equal, so every row's class could change and the models still pass.
+  it should "catch a classifier whose predicted classes all changed" in {
+    val dir = Files.createTempDirectory("comparator-spec-model-classes-")
+    val (actual, expected, probe) =
+      writeFittedModels(dir, "DecisionTreeClassifier", "[100000] * 3", "[100001] * 3")
+
+    intercept[ComparatorMismatchException] {
+      Comparator.assertEqual(
+        actual,
+        expected,
+        modelColumns = Seq("model"),
+        probePath = Some(probe)
+      )
+    }.getMessage should include("predictions differ")
+  }
+
+  // The tolerance still has to reach what it was for: a regressor computes its
+  // answer, so the two paths can land a float's last bits apart on the same fit.
+  it should "allow a regressor's predictions to differ within tolerance" in {
+    val dir = Files.createTempDirectory("comparator-spec-model-regressor-")
+    val (actual, expected, probe) =
+      writeFittedModels(
+        dir,
+        "DecisionTreeRegressor",
+        "[1.0, 2.0, 3.0]",
+        "[1.0, 2.0, 3.0 + 1e-12]"
+      )
+
+    noException should be thrownBy Comparator.assertEqual(
+      actual,
+      expected,
+      modelColumns = Seq("model"),
+      probePath = Some(probe)
+    )
+  }
+
   // An operator that draws a chart per row writes one JSONL row per chart on the
   // runtime side and an array on the exported one. Reading a single figure from
   // each, as this did, left every chart after the first unread.

From b6d5b4b34fd2ae196c980171972a7cd02161190e Mon Sep 17 00:00:00 2001
From: kary zheng <kary09302005@gmail.com>
Date: Thu, 24 Sep 2026 20:44:19 -0700
Subject: [PATCH 15/16] test(verify): leave the models unasked when neither
 output has a row

Two empty outputs hold no model to compare, and read back from an empty
file they hold no column either, so the model check reported the model
column missing from both. It now runs only when there is a row.

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

diff --git a/workflow-compiling-service/src/test/resources/python/compare.py b/workflow-compiling-service/src/test/resources/python/compare.py
index faa15c467f3..885ce897946 100644
--- a/workflow-compiling-service/src/test/resources/python/compare.py
+++ b/workflow-compiling-service/src/test/resources/python/compare.py
@@ -453,11 +453,14 @@ def _run_comparison(
 
     # Model columns: compare behavior (predictions) rather than bytes, then drop
     # the raw columns so the frame comparison covers everything else exactly.
+    # Two outputs without a row hold no model to ask, and read back from an
+    # empty file they hold no column either.
     if model_cols:
-        try:
-            _compare_model_predictions(actual, expected, model_cols, probe_path)
-        except AssertionError as exc:
-            return str(exc)
+        if len(actual) or len(expected):
+            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")
 

From bdfef7eabb9738a02caa53900f7c726c0d6bc51b Mon Sep 17 00:00:00 2001
From: kary zheng <kary09302005@gmail.com>
Date: Sat, 26 Sep 2026 20:09:42 -0700
Subject: [PATCH 16/16] test(verify): catch two models fitted on different
 columns

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

diff --git a/workflow-compiling-service/src/test/resources/python/compare.py b/workflow-compiling-service/src/test/resources/python/compare.py
index 885ce897946..aee9b5424e2 100644
--- a/workflow-compiling-service/src/test/resources/python/compare.py
+++ b/workflow-compiling-service/src/test/resources/python/compare.py
@@ -596,9 +596,8 @@ def _plotly_payloads_from_html(html: str) -> list:
 
 def _load_expected_plots(path) -> list:
     """Every chart the exported script drew. An operator that draws one writes a
-    lone figure; one that draws a chart per row writes the sequence, or writes
-    them to its page and only the first as a figure, in which case the page is
-    what this is handed."""
+    lone figure, and one that draws a chart per row writes the sequence. A page
+    is read for every chart on it."""
     import json
 
     with open(path, "r", encoding="utf-8") as fh:
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 42eae420674..d966bfa5b7d 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
@@ -280,23 +280,29 @@ class ComparatorSpec extends AnyFlatSpec with Matchers {
 
   /** A model column holds a pickled estimator, which only Python can fit. Writes
     * the two one-row frames the comparison reads and the probe it predicts on,
-    * each side's model fitted on its own targets.
+    * each side's model fitted on its own targets and its own feature column.
+    * The probe carries both columns, with the same values.
     */
   private def writeFittedModels(
       dir: Path,
       estimator: String,
       actualTargets: String,
-      expectedTargets: String
+      expectedTargets: String,
+      actualFeature: String = "petal_length",
+      expectedFeature: String = "petal_length"
   ): (Path, Path, Path) = {
     val program =
       s"""import base64, json, pickle
          |import pandas as pd
          |from sklearn.tree import $estimator as Estimator
          |
-         |x = pd.DataFrame({"petal_length": [1.0, 2.0, 3.0]})
+         |x = pd.DataFrame({"petal_length": [1.0, 2.0, 3.0], "petal_width": [1.0, 2.0, 3.0]})
          |x.to_json(r"$dir/probe.jsonl", orient="records", lines=True)
-         |for name, y in (("actual.jsonl", $actualTargets), ("expected.jsonl", $expectedTargets)):
-         |    model = base64.b64encode(pickle.dumps(Estimator().fit(x, y))).decode()
+         |for name, y, feature in (
+         |    ("actual.jsonl", $actualTargets, "$actualFeature"),
+         |    ("expected.jsonl", $expectedTargets, "$expectedFeature"),
+         |):
+         |    model = base64.b64encode(pickle.dumps(Estimator().fit(x[[feature]], y))).decode()
          |    with open(rf"$dir/{name}", "w") as handle:
          |        handle.write(json.dumps({"model": model, "score": 1.0}) + "\\n")
          |""".stripMargin
@@ -332,6 +338,29 @@ class ComparatorSpec extends AnyFlatSpec with Matchers {
     }.getMessage should include("predictions differ")
   }
 
+  // What a model was fitted on is part of the model. Two fitted on different
+  // columns answer alike on a probe carrying both, so only the names part them.
+  it should "catch two models fitted on different columns that predict alike" in {
+    val dir = Files.createTempDirectory("comparator-spec-model-features-")
+    val (actual, expected, probe) =
+      writeFittedModels(
+        dir,
+        "DecisionTreeClassifier",
+        "[0, 1, 1]",
+        "[0, 1, 1]",
+        expectedFeature = "petal_width"
+      )
+
+    intercept[ComparatorMismatchException] {
+      Comparator.assertEqual(
+        actual,
+        expected,
+        modelColumns = Seq("model"),
+        probePath = Some(probe)
+      )
+    }.getMessage should include("fitted feature names differ")
+  }
+
   // The tolerance still has to reach what it was for: a regressor computes its
   // answer, so the two paths can land a float's last bits apart on the same fit.
   it should "allow a regressor's predictions to differ within tolerance" in {
@@ -442,9 +471,8 @@ class ComparatorSpec extends AnyFlatSpec with Matchers {
   }
 
   // A bullet or gauge chart draws its charts onto one page and yields that page
-  // as a single row, and its script writes the whole set to its page and only
-  // the first of them as a figure. Both sides are pages then, and every call on
-  // them is a chart to compare.
+  // as a single row. Every call on a page is a chart to compare, whichever side
+  // the page is on.
   private def plotlyPage(values: Int*): String =
     values
       .map(v =>