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..aee9b5424e2
--- /dev/null
+++ b/workflow-compiling-service/src/test/resources/python/compare.py
@@ -0,0 +1,801 @@
+#!/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 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
+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
+
+    # 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)
+    # 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:
+        # 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 "
+                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)
+            names_e = getattr(m_expected, "feature_names_in_", None)
+
+            # 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))
+
+            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})"
+                )
+            # 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 tolerant
+                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 _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
+    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"]: 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 {
+        name: str for name, kind in _declared_types(actual_path).items() if kind == "string"
+    }
+
+
+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.
+
+    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
+
+    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, str):
+                    values[column].append(cell)
+                else:
+                    raise AssertionError(
+                        f"column '{column}' is declared text but holds {cell!r} in {path}"
+                    )
+    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(
+    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
+
+    # 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.
+    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)
+
+    # `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 = [
+        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:
+            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.
+    # 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:
+        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")
+
+    if ignore_cols:
+        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
+        # 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)
+
+    # 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]
+    # 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_dtype=False,
+                check_exact=True,
+            )
+        pd.testing.assert_frame_equal(
+            actual[loose_cols],
+            expected[loose_cols],
+            check_dtype=False,
+            rtol=1e-5,
+        )
+    except AssertionError as exc:
+        return str(exc)
+    return None
+
+
+def _load_actual_plots(path) -> list:
+    """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:
+        lines = [raw for raw in fh if raw.strip()]
+    if not lines:
+        raise AssertionError(f"{path} is empty")
+
+    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.extend(_plotly_payloads_from_html(row["html-content"]))
+        else:
+            raise AssertionError(
+                f"{path} row {number} has neither html-content nor json-content"
+            )
+    return plots
+
+
+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.
+    """
+    import json
+
+    marker = "Plotly.newPlot("
+    index = html.find(marker)
+    if index < 0:
+        raise AssertionError("html-content does not contain Plotly.newPlot(...)")
+
+    decoder = json.JSONDecoder()
+    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 plots
+
+
+def _load_expected_plots(path) -> list:
+    """Every chart the exported script drew. An operator that draws one writes a
+    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:
+        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", {})}
+        for fig in figures
+    ]
+
+
+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
+
+    # 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):
+        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(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:
+    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..ae67d5c478a
--- /dev/null
+++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/Comparator.scala
@@ -0,0 +1,184 @@
+/*
+ * 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 `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.
+  *
+  * 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 {
+
+  // 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..d966bfa5b7d
--- /dev/null
+++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/ComparatorSpec.scala
@@ -0,0 +1,532 @@
+/*
+ * 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.apache.texera.amber.util.JSONUtils.objectMapper
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+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.
+@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
+  }
+
+  // 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"))
+    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)
+    }
+  }
+
+  // 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 "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,
+    // 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")
+  }
+
+  /** 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 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,
+      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], "petal_width": [1.0, 2.0, 3.0]})
+         |x.to_json(r"$dir/probe.jsonl", orient="records", lines=True)
+         |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
+    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")
+  }
+
+  // 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 {
+    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.
+  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 }
+  }
+
+  // A bullet or gauge chart draws its charts onto one page and yields that page
+  // 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 =>
+        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.
+  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 }
+  }
+}