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..74f31494b0b
--- /dev/null
+++ b/workflow-compiling-service/src/test/resources/python/compare.py
@@ -0,0 +1,548 @@
+#!/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:
+        # 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)
+            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 _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 _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,
+    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)
+
+    # 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:
+        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)
+
+    # 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[loose_cols],
+            expected[loose_cols],
+            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..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..4ec7066ef14
--- /dev/null
+++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/ComparatorSpec.scala
@@ -0,0 +1,192 @@
+/*
+ * 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
+  }
+
+  // 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 "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,
+    // 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")
+  }
+}