kz930 commented on code in PR #8359: URL: https://github.com/apache/texera/pull/8359#discussion_r3983303507
########## workflow-compiling-service/src/test/resources/python/compare.py: ########## @@ -0,0 +1,481 @@ +#!/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 _string_columns(actual_path: str) -> dict: + """Which columns the engine declared as strings, read off the schema it + writes beside its output. Empty when there is no sidecar, which leaves the + inference in place rather than guessing.""" + import json + import os + + sidecar = actual_path + ".schema.json" + if not os.path.exists(sidecar): + return {} + with open(sidecar) as fh: + schema = json.load(fh) + return { + a["attributeName"]: str + for a in schema.get("attributes", []) + if a.get("attributeType") == "string" + } + + +def _run_comparison( + actual_path: str, + expected_path: str, + 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. + str_cols = _string_columns(actual_path) + actual = pd.read_json(actual_path, lines=True, dtype=str_cols or None) + expected = pd.read_json(expected_path, lines=True, dtype=str_cols or None) + + # Model columns: compare behavior (predictions) rather than bytes, then drop + # the raw columns so the frame comparison covers everything else exactly. + if model_cols: + try: + _compare_model_predictions(actual, expected, model_cols, probe_path) + except AssertionError as exc: + return str(exc) + actual = actual.drop(columns=model_cols, errors="ignore") + expected = expected.drop(columns=model_cols, errors="ignore") + + if ignore_cols: + actual = actual.drop(columns=ignore_cols, errors="ignore") + expected = expected.drop(columns=ignore_cols, errors="ignore") + + if unordered: + # Sort both sides by the same column key so set-equal frames collapse + # to the same row sequence. assert_frame_equal still does the actual + # value diff and respects rtol/check_dtype. Mergesort = stable, so + # rows that are tied on all columns keep their relative order — not + # strictly necessary for set equality (no ties → no duplicates after + # the op's dedup step) but cheap insurance. + cols = list(actual.columns) + if cols: + actual = actual.sort_values( + by=cols, kind="mergesort", na_position="last" + ).reset_index(drop=True) + expected = expected.sort_values( + by=cols, kind="mergesort", na_position="last" + ).reset_index(drop=True) + + try: + pd.testing.assert_frame_equal( + actual, + expected, + check_like=True, + check_dtype=False, + rtol=1e-5, Review Comment: Fixed: declared integer columns are compared exactly, and re-read with Python's json so a nullable LONG is not rounded on the way in. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
