carloea2 commented on code in PR #8359: URL: https://github.com/apache/texera/pull/8359#discussion_r4253713518
########## 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 Review Comment: When both sides widen a declared LONG column, dtype is float64 but a null becomes pd.NA. pandas 2.2.3 cannot convert that marker to float64. Identical files containing {"v":1.0} and {"v":null} therefore fail with TypeError in both CLI and worker modes. The boolean helper has the same issue. Please use a missing-value representation compatible with the selected dtype and cover a null in the both-widened test. -- 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]
