carloea2 commented on code in PR #8359: URL: https://github.com/apache/texera/pull/8359#discussion_r4357268034
########## workflow-compiling-service/src/test/resources/python/compare.py: ########## @@ -0,0 +1,897 @@ +#!/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 _holds_pickle(frame, col) -> bool: + """Whether every value in `col` is a base64 pickle, which is how a model is + written. A column the frame lacks, or one with no value, holds none.""" + import base64 + import pickle + + if col not in frame.columns: + return False + cells = [v for v in frame[col] if isinstance(v, str)] + if not cells: + return False + try: + for v in cells: + pickle.loads(base64.b64decode(v)) + except Exception: + return False + return True + + +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. A null + # stays None so each dtype takes its own hole: float64 cannot hold + # pd.NA. + widened = any(isinstance(c, float) for c in cells) + out[column] = pd.array(cells, dtype="float64" if widened else "Int64") + 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) + out[column] = pd.array(cells, dtype="boolean" if kept else "float64") + return out + + return arrays(actual), arrays(expected) + + +_DOUBLE_TEXT = {"NaN": "nan", "Infinity": "inf", "-Infinity": "-inf", "-0.0": "-0.0"} + + +def _double_kind(cell) -> str: + """Which of the values a float64 comparison cannot tell apart this cell is. + + `read_json` reads a null and a NaN as one NaN, and -0.0 equals 0.0, so a side + that answered NaN where the other answered null, or flipped a zero's sign, + compared equal. Each side spells the four as the engine's writer does. + """ + import math + + if cell is None: + return "null" + if isinstance(cell, str): + return cell + if cell != cell: + return "NaN" + if cell in (math.inf, -math.inf): + return "Infinity" if cell > 0 else "-Infinity" + if cell == 0.0 and math.copysign(1.0, cell) < 0: + return "-0.0" + return "number" + + +def _declared_doubles(actual_path: str, expected_path: str, columns: list) -> tuple: + """Both sides' declared DOUBLE columns as numbers, plus what kind of value + each cell is. The numbers keep the tolerance; the kinds are compared exactly.""" + + def split(values): + numbers, kinds = {}, {} + for column, cells in values.items(): + kinds[column] = [_double_kind(c) for c in cells] + numbers[column] = [ + float(_DOUBLE_TEXT.get(c, "nan")) if isinstance(c, str) + else float("nan") if c is None + else float(c) + for c in cells + ] + return numbers, kinds + + return split(_raw_column(actual_path, columns)), split(_raw_column(expected_path, columns)) + + +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] + + double_cols = [ + name + for name, kind in declared.items() + if kind == "double" and name in actual.columns and name in expected.columns + ] + double_kinds_a, double_kinds_e = {}, {} + if double_cols: + (numbers_a, double_kinds_a), (numbers_e, double_kinds_e) = _declared_doubles( + actual_path, expected_path, double_cols + ) + for column in double_cols: + actual[column] = pd.array(numbers_a[column], dtype="float64") + expected[column] = pd.array(numbers_e[column], dtype="float64") + + # A BINARY column is not always a model: File Scan hands back a file's bytes. + # Bytes neither side can unpickle are plain data, compared exactly with the + # rest of the frame. A column one side lacks, or one side can unpickle, stays + # a model column, so a model on one side only still fails as one. + model_cols = [ + col + for col in model_cols + if col not in actual.columns + or col not in expected.columns + or _holds_pickle(actual, col) + or _holds_pickle(expected, col) + ] + + # 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 + ) + + # Added after the column check, which is about the operator's columns. Each + # rides with its row through the sort below and is compared as text. + for column in double_cols: + if column in actual.columns: + kind_column = f"{column} (null/NaN/±Infinity/-0.0)" Review Comment: This helper name can overwrite a real output column and hide a difference. I compared {"v":1.0,"v (null/NaN/±Infinity/-0.0)":"before"} against the same row with "after", declaring v as DOUBLE. Both CLI and worker return success because the second column becomes "number" on both sides. Please keep the kind metadata separate or choose a collision-free name. ########## workflow-compiling-service/src/test/resources/python/compare.py: ########## @@ -0,0 +1,897 @@ +#!/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 _holds_pickle(frame, col) -> bool: + """Whether every value in `col` is a base64 pickle, which is how a model is + written. A column the frame lacks, or one with no value, holds none.""" + import base64 + import pickle + + if col not in frame.columns: + return False + cells = [v for v in frame[col] if isinstance(v, str)] + if not cells: + return False + try: + for v in cells: + pickle.loads(base64.b64decode(v)) + except Exception: + return False + return True + + +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: Review Comment: These raw JSON reads still use Windows' cp1252 default. Matching UTF-8 files containing {"v":"ひ"} fail in _declared_strings, and a LONG column named é silently loses its declared type: 100000 and 100001 compare equal. Both reproduce in CLI and worker modes; both behave correctly with PYTHONUTF8=1. Please read the sidecar and raw rows explicitly as UTF-8. -- 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]
