From 6850be42deeb45605c64f71191b6c767112324ed Mon Sep 17 00:00:00 2001 From: Faisal Dosani Date: Tue, 4 Aug 2026 21:58:32 -0300 Subject: [PATCH 1/6] fix: support Spark Connect sessions in SparkSQLCompare --- .github/workflows/test-package.yml | 12 ++ .gitignore | 4 +- CLAUDE.md | 28 ++- README.md | 2 +- datacompy/base.py | 17 +- datacompy/comparator/array.py | 8 +- datacompy/comparator/boolean.py | 20 ++- datacompy/comparator/numeric.py | 8 +- datacompy/comparator/string.py | 9 +- datacompy/comparator/utility.py | 90 ++++++++++ datacompy/spark.py | 58 +++++-- docs/source/install.rst | 13 ++ docs/source/spark_usage.rst | 56 ++++-- pytest-ansi.ini | 4 + pytest-connect.ini | 25 +++ pytest.ini | 4 + tests/comparator/test_utility_spark.py | 68 +++++++- tests/test_spark_connect.py | 226 +++++++++++++++++++++++++ 18 files changed, 594 insertions(+), 58 deletions(-) create mode 100644 pytest-connect.ini create mode 100644 tests/test_spark_connect.py diff --git a/.github/workflows/test-package.yml b/.github/workflows/test-package.yml index f34c8e10..1db87813 100644 --- a/.github/workflows/test-package.yml +++ b/.github/workflows/test-package.yml @@ -117,6 +117,18 @@ jobs: - name: Test with pytest (ANSI mode) run: | python -m pytest -c pytest-ansi.ini --cov=datacompy --cov-report=xml --cov-report=term-missing + # Spark Connect runs in two separate pytest processes: a classic and a + # Connect session cannot share one, because starting a local Connect + # server sets SPARK_LOCAL_REMOTE, after which every later + # SparkSession.builder.getOrCreate() returns the Connect session. + # Only on Spark 4.x: the Connect server jar is bundled in the 4.x wheel, + # whereas Spark 3.5 requires resolving it via --packages. + - name: Test with pytest (Spark Connect - existing suite) + run: | + python -m pytest -c pytest-connect.ini tests/test_spark.py tests/comparator/ + - name: Test with pytest (Spark Connect - regression suite) + run: | + python -m pytest -m spark_connect tests/test_spark_connect.py test-basic-install: runs-on: ubuntu-latest diff --git a/.gitignore b/.gitignore index beab08bc..514c4979 100644 --- a/.gitignore +++ b/.gitignore @@ -26,8 +26,10 @@ docs/source/api/ test.html -.coverage +.coverage* coverage.xml # benchmark datasets benchmarks/data/ + +.tox/ diff --git a/CLAUDE.md b/CLAUDE.md index 03b9b12b..4b0d8f70 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -18,12 +18,13 @@ pre-commit install ### Testing ```bash -pytest # all tests -pytest tests/test_pandas.py # single backend -pytest tests/test_pandas.py::test_numeric_columns_equal_abs # single test -pytest -k "tolerance and not spark" # by expression -pytest --cov=datacompy --cov-report=term-missing # with coverage -pytest -c pytest-ansi.ini # Spark ANSI mode +pytest # all tests +pytest tests/test_pandas.py # single backend +pytest tests/test_pandas.py::TestPandasCompare::test_method # single test +pytest --cov=datacompy --cov-report=term-missing # with coverage + +pytest -c pytest-connect.ini tests/test_spark.py tests/comparator/ # existing Spark suite, against Spark Connect +pytest -m spark_connect tests/test_spark_connect.py # Spark Connect regression suite ``` CI runs the suite twice, once with the default `pytest.ini` and once with `-c pytest-ansi.ini`, which only differs by `spark.sql.ansi.enabled`. A change touching Spark casting or null handling needs both. @@ -37,6 +38,8 @@ export JAVA_HOME=$CONDA_PREFIX/lib/jvm **Snowflake** tests need a live session, or `--snowflake-session local` for Snowpark's local testing mode. Local mode is an emulator, not Snowflake: `eqNullSafe` returns `True` for every row and high-precision decimals are truncated on DataFrame creation. Tests that depend on either must request the `requires_live_snowflake_session` fixture (`tests/conftest.py`), which skips them in local mode. +The two Spark Connect commands must each run in their own pytest process, and are excluded from the default run via `addopts`. Starting a local Spark Connect server sets `SPARK_LOCAL_REMOTE`, after which every later `SparkSession.builder.getOrCreate()` in that process returns the Connect session — so a classic and a Connect session cannot coexist in one run. + ### Linting & Formatting ```bash ruff check # lint @@ -81,9 +84,22 @@ Beyond the report, each backend exposes `df1_unq_rows`, `df2_unq_rows`, and `int - `numeric.py` → Numeric comparators per backend (handles tolerances) - `string.py` → String comparators per backend - `array.py` → Array-like comparators per backend +- `utility.py` → Shared Spark/Snowflake helpers, including `get_spark_functions` / `get_spark_window` Each type has backend-specific implementations: `Pandas*Comparator`, `Polars*Comparator`, `Spark*Comparator`, `Snowflake*Comparator`. +### Spark Connect + +Never import `pyspark.sql.functions` or `pyspark.sql.Window` at module scope in Spark code paths. Those dispatch to the Spark Connect implementations only when the process-global `SPARK_CONNECT_MODE_ENABLED` environment variable is set, which a Connect session from a notebook or serverless runtime does not necessarily set. Instead resolve them from the DataFrame or Column being operated on: + +```python +F = get_spark_functions(dataframe) # datacompy/spark.py +psf = get_spark_functions(dataframe) # datacompy/comparator/*.py +Window = get_spark_window(dataframe) +``` + +Because there is no module-level binding, ruff's `F821` flags any call site that forgets the local. For the same reason, never import `pyspark.sql.connect.*` at module scope — that package requires the optional `grpcio` dependency, and `__init__.py`'s `except ImportError` would silently drop `SparkSQLCompare` from the package. Use `is_spark_connect_object()` instead. + ### Reporting Reports use Jinja2 templates from `datacompy/templates/report_template.j2`. The `render()` function in `base.py` handles template resolution. Custom templates can be passed via `report(template_path=...)`. diff --git a/README.md b/README.md index 3b6ba1ed..69afdf3a 100644 --- a/README.md +++ b/README.md @@ -15,7 +15,7 @@ and lets you tweak how accurate matches have to be). Supported types include: - Pandas - Polars -- Spark +- Spark (classic and Spark Connect) - Snowflake > [!IMPORTANT] diff --git a/datacompy/base.py b/datacompy/base.py index aeddf12a..1fa7020f 100644 --- a/datacompy/base.py +++ b/datacompy/base.py @@ -710,6 +710,17 @@ def df_to_str(df: Any, sample_count: int | None = None, on_index: bool = False) str String representation of the DataFrame """ + # Handle Spark DataFrame and Snowflake DataFrame. + # This must come *before* the ``to_string`` check below: a Spark Connect + # DataFrame synthesizes a Column for any unknown attribute, so + # ``hasattr(df, "to_string")`` is True for it and it would otherwise take + # the pandas branch. Nothing else is caught here -- pandas has no + # ``toPandas`` and Polars exposes ``to_pandas``, not ``toPandas``. + if hasattr(df, "toPandas"): + if sample_count is not None: + df = df.limit(sample_count) + return df.toPandas().to_string() + # Handle pandas DataFrame if hasattr(df, "to_string"): if sample_count is not None and len(df) > sample_count: @@ -718,12 +729,6 @@ def df_to_str(df: Any, sample_count: int | None = None, on_index: bool = False) df = df.reset_index(drop=True) return df.to_string() - # Handle Spark DataFrame and Snowflake DataFrame - if hasattr(df, "toPandas"): - if sample_count is not None: - df = df.limit(sample_count) - return df.toPandas().to_string() - # Handle Polars DataFrame if hasattr(df, "to_pandas"): if sample_count is not None and len(df) > sample_count: diff --git a/datacompy/comparator/array.py b/datacompy/comparator/array.py index b94c937c..14b80ebe 100644 --- a/datacompy/comparator/array.py +++ b/datacompy/comparator/array.py @@ -29,12 +29,13 @@ try: import pyspark as ps - import pyspark.sql.functions as psf - from datacompy.comparator.utility import get_spark_column_dtypes + from datacompy.comparator.utility import ( + get_spark_column_dtypes, + get_spark_functions, + ) except ImportError: ps = None - psf = None try: import snowflake.snowpark as sp @@ -152,6 +153,7 @@ def compare( None if the columns are not comparable. """ + psf = get_spark_functions(dataframe) base_dtype, compare_dtype = get_spark_column_dtypes(dataframe, col1, col2) if base_dtype.startswith("array") and compare_dtype.startswith("array"): when_clause = psf.col(col1).eqNullSafe(psf.col(col2)) diff --git a/datacompy/comparator/boolean.py b/datacompy/comparator/boolean.py index f6fe8cbc..aad923b1 100644 --- a/datacompy/comparator/boolean.py +++ b/datacompy/comparator/boolean.py @@ -15,6 +15,7 @@ """Boolean comparator classes.""" +from types import ModuleType from typing import Any import pandas as pd @@ -32,13 +33,14 @@ # Optional Spark dependencies try: import pyspark as ps - import pyspark.sql.functions as psf from datacompy.comparator.numeric import NUMERIC_PYSPARK_TYPES - from datacompy.comparator.utility import get_spark_column_dtypes + from datacompy.comparator.utility import ( + get_spark_column_dtypes, + get_spark_functions, + ) except ImportError: ps = None - psf = None NUMERIC_PYSPARK_TYPES = None # Optional Snowflake dependencies @@ -171,7 +173,9 @@ class SparkBooleanComparator(BaseComparator): """Comparator for Boolean columns in PySpark.""" @staticmethod - def _boolean_equals_numeric(boolean_col: str, numeric_col: str) -> "ps.sql.Column": + def _boolean_equals_numeric( + boolean_col: str, numeric_col: str, psf: ModuleType + ) -> "ps.sql.Column": """Compare a Boolean column against a numeric column's 1/0 equivalents. The numeric side is compared against integer literals rather than both @@ -186,6 +190,9 @@ def _boolean_equals_numeric(boolean_col: str, numeric_col: str) -> "ps.sql.Colum The name of the Boolean column. numeric_col : str The name of the numeric column. + psf : ModuleType + The ``functions`` module matching the DataFrame being compared, + classic or Spark Connect. Returns ------- @@ -254,6 +261,7 @@ def compare( behaves identically under both settings and preserves the numeric column's precision. """ + psf = get_spark_functions(dataframe) base_dtype, compare_dtype = get_spark_column_dtypes(dataframe, col1, col2) base_boolean_type = base_dtype == PYSPARK_BOOLEAN_TYPE compare_boolean_type = compare_dtype == PYSPARK_BOOLEAN_TYPE @@ -265,9 +273,9 @@ def compare( if base_boolean_type and compare_boolean_type: when_clause = psf.col(col1).eqNullSafe(psf.col(col2)) elif base_boolean_type and compare_numeric_type: - when_clause = self._boolean_equals_numeric(col1, col2) + when_clause = self._boolean_equals_numeric(col1, col2, psf) elif compare_boolean_type and base_numeric_type: - when_clause = self._boolean_equals_numeric(col2, col1) + when_clause = self._boolean_equals_numeric(col2, col1, psf) else: return None diff --git a/datacompy/comparator/numeric.py b/datacompy/comparator/numeric.py index c32f08df..7d8d2e97 100644 --- a/datacompy/comparator/numeric.py +++ b/datacompy/comparator/numeric.py @@ -48,9 +48,11 @@ def __eq__(self, other): # Optional Spark dependencies try: import pyspark as ps - import pyspark.sql.functions as psf - from datacompy.comparator.utility import get_spark_column_dtypes + from datacompy.comparator.utility import ( + get_spark_column_dtypes, + get_spark_functions, + ) NUMERIC_PYSPARK_TYPES = [ "tinyint", @@ -64,7 +66,6 @@ def __eq__(self, other): SPARK_INTEGER_TYPES = {"tinyint", "smallint", "int", "bigint"} except ImportError: ps = None - psf = None NUMERIC_PYSPARK_TYPES = None SPARK_INTEGER_TYPES = None @@ -284,6 +285,7 @@ def compare( - If either column contains NaN values, they are handled explicitly to avoid incorrect comparisons. """ + psf = get_spark_functions(dataframe) base_dtype, compare_dtype = get_spark_column_dtypes(dataframe, col1, col2) base_numeric_type = any(base_dtype.startswith(t) for t in NUMERIC_PYSPARK_TYPES) compare_numeric_type = any( diff --git a/datacompy/comparator/string.py b/datacompy/comparator/string.py index 23ab2aec..b7d2216c 100644 --- a/datacompy/comparator/string.py +++ b/datacompy/comparator/string.py @@ -27,15 +27,16 @@ # Initialize optional dependencies ps = None -psf = None sp = None spf = None try: import pyspark as ps - import pyspark.sql.functions as psf - from datacompy.comparator.utility import get_spark_column_dtypes + from datacompy.comparator.utility import ( + get_spark_column_dtypes, + get_spark_functions, + ) PYSPARK_STRING_TYPE = {"string", "char", "varchar"} PYSPARK_DATE_TYPE = {"date", "timestamp"} @@ -307,6 +308,7 @@ def compare( ) or ((base_date_type) and (compare_date_type)) # date/date compare. ): + psf = get_spark_functions(dataframe) try: if base_date_type and compare_date_type: # Both are date/timestamp: compare directly, no string conversion needed. @@ -504,6 +506,7 @@ def spark_normalize_string_column( pyspark.sql.Column The normalized column """ + psf = get_spark_functions(column) if ignore_spaces: column = psf.trim(column) if ignore_case: diff --git a/datacompy/comparator/utility.py b/datacompy/comparator/utility.py index fd70eb25..6fbe763a 100644 --- a/datacompy/comparator/utility.py +++ b/datacompy/comparator/utility.py @@ -15,6 +15,9 @@ """Utility and helper functions for data comparison.""" +from types import ModuleType +from typing import Any + # Optional dependencies initialization ps = None sp = None @@ -30,6 +33,93 @@ pass +_CONNECT_MODULE_PREFIX = "pyspark.sql.connect." + + +def is_spark_connect_object(spark_object: Any) -> bool: + """Check whether an object belongs to the Spark Connect API. + + Walks the MRO comparing module names instead of using ``isinstance`` against + the Spark Connect classes: importing ``pyspark.sql.connect`` runs + ``check_dependencies()``, which raises ``ImportError`` when the optional + ``grpcio`` dependency is missing. A classic-only PySpark installation must + never pay that cost. Walking the MRO rather than only looking at + ``type(obj).__module__`` also covers subclasses layered on top by other + runtimes. + + Parameters + ---------- + spark_object : Any + Any object, typically a ``DataFrame`` or a ``Column``. + + Returns + ------- + bool + True if the object comes from the Spark Connect API, False otherwise. + """ + return any( + klass.__module__.startswith(_CONNECT_MODULE_PREFIX) + for klass in type(spark_object).__mro__ + ) + + +def get_spark_functions(spark_object: Any) -> ModuleType: + """Get the ``functions`` module matching a classic or Spark Connect object. + + ``pyspark.sql.functions`` only forwards to the Spark Connect implementations + when the process-global ``SPARK_CONNECT_MODE_ENABLED`` environment variable + is set, which is not the case for a Connect session handed over by a + notebook runtime or another framework. Selecting the module from the object + being operated on is correct however the session was created. + + Parameters + ---------- + spark_object : Any + The ``DataFrame`` or ``Column`` the expression is being built for. + + Returns + ------- + ModuleType + ``pyspark.sql.connect.functions`` for a Spark Connect object, + ``pyspark.sql.functions`` otherwise. + """ + if is_spark_connect_object(spark_object): + from pyspark.sql.connect import functions as connect_functions + + return connect_functions + + from pyspark.sql import functions as classic_functions + + return classic_functions + + +def get_spark_window(spark_object: Any) -> Any: + """Get the ``Window`` class matching a classic or Spark Connect object. + + ``pyspark.sql.Window`` dispatches through ``dispatch_window_method`` on the + same process-global flag that :func:`get_spark_functions` works around. + + Parameters + ---------- + spark_object : Any + The ``DataFrame`` or ``Column`` the window is being built for. + + Returns + ------- + Any + ``pyspark.sql.connect.window.Window`` for a Spark Connect object, + ``pyspark.sql.Window`` otherwise. + """ + if is_spark_connect_object(spark_object): + from pyspark.sql.connect.window import Window as ConnectWindow + + return ConnectWindow + + from pyspark.sql import Window as ClassicWindow + + return ClassicWindow + + def get_spark_column_dtypes( dataframe: "ps.sql.DataFrame", col_1: str, col_2: str ) -> tuple[str, str]: diff --git a/datacompy/spark.py b/datacompy/spark.py index 78c0f0f7..12539523 100644 --- a/datacompy/spark.py +++ b/datacompy/spark.py @@ -24,13 +24,11 @@ import logging from copy import deepcopy from typing import Any, Dict, List, Tuple +from uuid import uuid4 import pandas as pd import pyspark.sql -import pyspark.sql.functions as F from ordered_set import OrderedSet -from pyspark.sql import Window -from pyspark.sql.connect.dataframe import DataFrame from datacompy.base import ( BaseCompare, @@ -46,7 +44,12 @@ SparkStringComparator, ) from datacompy.comparator.base import BaseComparator -from datacompy.comparator.utility import get_spark_column_dtypes +from datacompy.comparator.utility import ( + get_spark_column_dtypes, + get_spark_functions, + get_spark_window, + is_spark_connect_object, +) LOG = logging.getLogger(__name__) @@ -223,6 +226,7 @@ def df2(self, df2: "pyspark.sql.DataFrame") -> None: def hide_sensitive_columns(self, sensitive_columns: List[str]) -> None: """Hides sensitive columns of df1 or df2 if applicable in the compare.""" + F = get_spark_functions(self.df1) # Don't allow hiding columns again before first revealing if self.sensitive_columns: raise ValueError( @@ -287,9 +291,15 @@ def _validate_dataframe( None """ dataframe = getattr(self, index) - instances = (pyspark.sql.DataFrame, DataFrame) - if not isinstance(dataframe, instances): + # On PySpark 4.x the Spark Connect DataFrame subclasses the classic one, + # so the isinstance check alone would be enough. On 3.5 it does not, + # hence the second check -- which deliberately avoids importing + # pyspark.sql.connect, since that package requires the optional grpcio + # dependency and raises ImportError without it. + if not isinstance(dataframe, pyspark.sql.DataFrame) and not ( + is_spark_connect_object(dataframe) + ): raise TypeError( f"{index} must be a pyspark.sql.DataFrame or pyspark.sql.connect.dataframe.DataFrame (Spark 3.4.0 and above)" ) @@ -383,6 +393,7 @@ def _dataframe_merge(self, ignore_spaces: bool) -> None: df1 = self.df1 df2 = self.df2 + F = get_spark_functions(df1) temp_join_columns = deepcopy(self.join_columns) if self._any_dupes: @@ -447,19 +458,28 @@ def _dataframe_merge(self, ignore_spaces: bool) -> None: {c: f"{c}_{self.df2_name}" for c in temp_join_columns} ) - # NULL SAFE Outer join using ON - df1.createOrReplaceTempView("df1") - df2.createOrReplaceTempView("df2") + # NULL SAFE Outer join using ON. + # The view names are unique per merge. Fixed names would let a later + # comparison on the same session replace the views this plan refers to, + # and would clobber any user view called "df1" or "df2". Spark Connect + # resolves views lazily, so that surfaces as an unresolved column when + # the plan is finally executed rather than silently comparing the wrong + # data. + suffix = uuid4().hex + df1_view = f"datacompy_df1_{suffix}" + df2_view = f"datacompy_df2_{suffix}" + df1.createOrReplaceTempView(df1_view) + df2.createOrReplaceTempView(df2_view) on = " and ".join( [ - f"df1.`{c}_{self.df1_name}` <=> df2.`{c}_{self.df2_name}`" + f"{df1_view}.`{c}_{self.df1_name}` <=> {df2_view}.`{c}_{self.df2_name}`" for c in params["on"] ] ) outer_join = self.spark_session.sql( - """ + f""" SELECT * FROM - df1 FULL OUTER JOIN df2 + {df1_view} FULL OUTER JOIN {df2_view} ON """ + on @@ -544,6 +564,7 @@ def _intersect_compare(self, ignore_spaces: bool, ignore_case: bool) -> None: otherwise. """ LOG.debug("Comparing intersection") + F = get_spark_functions(self.intersect_rows) max_diff: float null_diff: int exprs = {} @@ -567,7 +588,11 @@ def _intersect_compare(self, ignore_spaces: bool, ignore_case: bool) -> None: comparators=self._get_comparators(), ) - self.intersect_rows = self.intersect_rows.withColumns(exprs) + if exprs: + # Guarded because Spark Connect asserts on an empty column mapping, + # where classic Spark treats it as a no-op. This is reached when the + # dataframes share nothing but the join columns. + self.intersect_rows = self.intersect_rows.withColumns(exprs) all_rows_count = self.intersect_rows.count() if exprs: @@ -742,6 +767,7 @@ def sample_mismatch( "pertinent" columns, for rows that don't match on the provided column. """ + F = get_spark_functions(self.intersect_rows) if not self.only_join_columns() and column not in self.join_columns: row_cnt = self.intersect_rows.count() col_match = self.intersect_rows.select(f"{column}_match") @@ -808,6 +834,7 @@ def all_mismatch( pyspark.sql.DataFrame All rows of the intersection dataframe, containing any columns, that don't match. """ + F = get_spark_functions(self.intersect_rows) match_list = [] return_list = [] if self.only_join_columns(): @@ -986,6 +1013,7 @@ def columns_equal( ) return compare + F = get_spark_functions(dataframe) compare = F.lit(False) return compare @@ -1042,6 +1070,7 @@ def calculate_max_diff( float max diff """ + F = get_spark_functions(dataframe) dtype1, dtype2 = get_spark_column_dtypes(dataframe, col_1, col_2) if dtype1.startswith("array") and dtype2.startswith("array"): LOG.warning( @@ -1086,6 +1115,7 @@ def calculate_null_diff( int null diff """ + F = get_spark_functions(dataframe) nulls_df = dataframe.withColumn( "col_1_null", F.when(F.col(col_1).isNull() == True, F.lit(True)).otherwise( # noqa: E712 @@ -1133,6 +1163,8 @@ def _generate_id_within_group( pyspark.sql.DataFrame Original dataframe with the ID column that's unique in each group """ + F = get_spark_functions(dataframe) + Window = get_spark_window(dataframe) default_value = "DATACOMPY_NULL" null_cols = [f"any(isnull({c}))" for c in join_columns] default_cols = [ diff --git a/docs/source/install.rst b/docs/source/install.rst index 73f3c251..b2a530cd 100644 --- a/docs/source/install.rst +++ b/docs/source/install.rst @@ -29,6 +29,19 @@ Installing the package also provides the ``datacompy`` command line tool. See :doc:`cli`. +Spark +----- + +:: + + pip install datacompy[spark] + +This installs ``pyspark[connect]``, which covers both classic Spark and Spark +Connect. If you manage PySpark yourself, a plain ``pip install pyspark`` is +enough for classic Spark; Spark Connect additionally needs the ``connect`` +extra (``pip install "pyspark[connect]"``), which pulls in ``grpcio``. + + A Conda environment or virtual environment is highly recommended: conda (installs dependencies from Conda Forge) diff --git a/docs/source/spark_usage.rst b/docs/source/spark_usage.rst index b67e6779..22b984b9 100644 --- a/docs/source/spark_usage.rst +++ b/docs/source/spark_usage.rst @@ -3,7 +3,9 @@ Spark Usage - ``on_index`` is not supported. - Joining is done using ``<=>`` which is the equality test that is safe for null values. -- ``SparkSQLCompare`` compares ``pyspark.sql.DataFrame``'s +- ``SparkSQLCompare`` compares ``pyspark.sql.DataFrame``'s, including Spark + Connect DataFrames (``pyspark.sql.connect.dataframe.DataFrame``). See + `Spark Connect`_ below. SparkSQLCompare @@ -58,25 +60,49 @@ join column(s). print(compare.report()) -Caching -------- +Spark Connect +------------- -``SparkSQLCompare`` caches intermediate DataFrames by default, which avoids -recomputing the joined data for every part of the report. Some environments, -Databricks Serverless among them, do not support caching. Pass -``cache_intermediates=False`` there: +``SparkSQLCompare`` works with Spark Connect sessions as well as classic +JVM-backed ones. Nothing changes in how you call it -- pass the Connect session +and its DataFrames exactly as above: .. code-block:: python - compare = SparkSQLCompare( - spark, - df1, - df2, - join_columns='acct_id', - cache_intermediates=False, - ) + from datacompy import SparkSQLCompare + from pyspark.sql import SparkSession + + spark = SparkSession.builder.remote("sc://localhost:15002").getOrCreate() + + df1 = spark.createDataFrame(pd.read_csv(StringIO(data1))) + df2 = spark.createDataFrame(pd.read_csv(StringIO(data2))) + + compare = SparkSQLCompare(spark, df1, df2, join_columns='acct_id') + print(compare.report()) + +This also covers a Connect session you did not create yourself -- one provided +by a notebook or serverless runtime, or handed over by another framework. +datacompy selects the classic or Spark Connect expression API from the +DataFrame you pass in, rather than from PySpark's process-global +``SPARK_CONNECT_MODE_ENABLED`` environment variable, which such a session does +not necessarily set. + +.. note:: + + ``pip install datacompy[spark]`` installs ``pyspark[connect]``, which + covers both flavours. If you manage PySpark yourself, Spark Connect + requires the ``connect`` extra (``pip install "pyspark[connect]"``), which + pulls in ``grpcio``. + +.. note:: + + On runtimes that do not permit caching, such as Databricks Serverless, pass + ``cache_intermediates=False``. + +.. note:: -The command line equivalent is ``--no-cache-intermediates``. See :doc:`cli`. + ``spark.sql.execution.arrow.pyspark.enabled`` has no effect under Spark + Connect -- Arrow is always used on the wire. Reports diff --git a/pytest-ansi.ini b/pytest-ansi.ini index 54c331bb..b5eaca7f 100644 --- a/pytest-ansi.ini +++ b/pytest-ansi.ini @@ -1,4 +1,8 @@ [pytest] +addopts = -m "not spark_connect" +markers = + pyspark: needs pyspark installed + spark_connect: needs a Spark Connect session and its own pytest process, because starting a local Connect server sets SPARK_LOCAL_REMOTE, after which every later SparkSession.builder.getOrCreate() in the process returns the Connect session. Run with: pytest -m spark_connect tests/test_spark_connect.py spark_options = spark.master: local[*] spark.sql.catalogImplementation: in-memory diff --git a/pytest-connect.ini b/pytest-connect.ini new file mode 100644 index 00000000..42e18084 --- /dev/null +++ b/pytest-connect.ini @@ -0,0 +1,25 @@ +; Runs the existing Spark suite against a Spark Connect session. +; +; pytest-spark turns `spark_connect_url` into `SparkSession.builder.remote(...)`, +; and PySpark 4.x starts a local Connect server from the jar bundled in the +; wheel -- no external server needed. +; +; `spark.master` must NOT be set here: PySpark rejects a session configured with +; both a master and a Connect URL (CANNOT_CONFIGURE_SPARK_CONNECT_MASTER). The +; other classic-only options (executor cores, parallelism, catalog +; implementation) are meaningless server-side under Connect, and +; `spark.sql.execution.arrow.pyspark.enabled` is a no-op because Connect always +; uses Arrow on the wire. +; +; The tests marked `spark_connect` are excluded: they build their own session +; and must run in a separate pytest process. +[pytest] +addopts = -m "not spark_connect" +markers = + pyspark: needs pyspark installed + spark_connect: needs a Spark Connect session and its own pytest process, because starting a local Connect server sets SPARK_LOCAL_REMOTE, after which every later SparkSession.builder.getOrCreate() in the process returns the Connect session. Run with: pytest -m spark_connect tests/test_spark_connect.py +spark_connect_url = local[2] +spark_options = + spark.sql.shuffle.partitions: 4 + spark.sql.adaptive.enabled: false + spark.sql.ansi.enabled: false diff --git a/pytest.ini b/pytest.ini index 93500106..0eb21bb7 100644 --- a/pytest.ini +++ b/pytest.ini @@ -1,4 +1,8 @@ [pytest] +addopts = -m "not spark_connect" +markers = + pyspark: needs pyspark installed + spark_connect: needs a Spark Connect session and its own pytest process, because starting a local Connect server sets SPARK_LOCAL_REMOTE, after which every later SparkSession.builder.getOrCreate() in the process returns the Connect session. Run with: pytest -m spark_connect tests/test_spark_connect.py spark_options = spark.master: local[*] spark.sql.catalogImplementation: in-memory diff --git a/tests/comparator/test_utility_spark.py b/tests/comparator/test_utility_spark.py index 302f5aa5..9995006f 100644 --- a/tests/comparator/test_utility_spark.py +++ b/tests/comparator/test_utility_spark.py @@ -20,7 +20,12 @@ pytest.importorskip("pyspark") -from datacompy.comparator.utility import get_spark_column_dtypes +from datacompy.comparator.utility import ( + get_spark_column_dtypes, + get_spark_functions, + get_spark_window, + is_spark_connect_object, +) from pyspark.sql.types import ( DateType, DecimalType, @@ -85,3 +90,64 @@ def test_get_spark_column_dtypes_case_insensitive(spark_session): dtype1, dtype2 = get_spark_column_dtypes(df, "num", "str") assert dtype1 == "bigint" assert dtype2 == "string" + + +@pytest.mark.pyspark +def test_is_spark_connect_object_connect_branch(): + """A Spark Connect Column is built without any session or SparkContext.""" + import pandas as pd + from pyspark.sql.connect import functions as connect_functions + + assert is_spark_connect_object(connect_functions.col("value")) + assert not is_spark_connect_object(pd.DataFrame({"a": [1]})) + assert not is_spark_connect_object(object()) + + +@pytest.mark.pyspark +def test_get_spark_helpers_connect_branch(): + """Spark Connect objects resolve to the Spark Connect implementations.""" + from pyspark.sql.connect import functions as connect_functions + from pyspark.sql.connect.window import Window as ConnectWindow + + column = connect_functions.col("value") + + assert get_spark_functions(column) is connect_functions + assert get_spark_window(column) is ConnectWindow + + +@pytest.mark.pyspark +def test_get_spark_helpers_match_the_session(spark_session): + """The helpers must agree with the flavour of the session under test. + + This runs under both lanes -- the default classic run and the Spark Connect + run driven by ``pytest-connect.ini`` -- so it asserts against whichever + flavour ``spark_session`` actually is. The negative assertions matter: + without them an implementation that always returned one flavour would pass. + """ + import pyspark.sql.functions as classic_functions + from pyspark.sql import Window as ClassicWindow + from pyspark.sql.connect import functions as connect_functions + from pyspark.sql.connect.window import Window as ConnectWindow + + df = spark_session.range(1) + + if is_spark_connect_object(df): + assert get_spark_functions(df) is connect_functions + assert get_spark_functions(df) is not classic_functions + assert get_spark_window(df) is ConnectWindow + assert get_spark_window(df) is not ClassicWindow + else: + assert get_spark_functions(df) is classic_functions + assert get_spark_functions(df) is not connect_functions + assert get_spark_window(df) is ClassicWindow + assert get_spark_window(df) is not ConnectWindow + + +@pytest.mark.pyspark +def test_get_spark_helpers_default_to_classic(): + """Anything that is not a Spark Connect object falls back to classic.""" + import pyspark.sql.functions as classic_functions + from pyspark.sql import Window as ClassicWindow + + assert get_spark_functions(object()) is classic_functions + assert get_spark_window(object()) is ClassicWindow diff --git a/tests/test_spark_connect.py b/tests/test_spark_connect.py new file mode 100644 index 00000000..14890306 --- /dev/null +++ b/tests/test_spark_connect.py @@ -0,0 +1,226 @@ +# +# Copyright 2026 Capital One Services, LLC +# +# Licensed 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. + +"""End-to-end coverage of SparkSQLCompare against a Spark Connect session. + +These tests must run in their own pytest process: starting a local Spark +Connect server sets ``SPARK_LOCAL_REMOTE``, after which every later +``SparkSession.builder.getOrCreate()`` in the process returns the Connect +session. They are therefore marked ``spark_connect`` and deselected from the +default run; see ``pytest-connect.ini`` and the CI workflow. + +Nothing here may import ``pyspark.sql.functions``. Those helpers dispatch on +the same global flag this module deliberately clears, so using them to build +fixtures would break in the test rather than exercise the library. +""" + +import os + +import pandas as pd +import pytest + +pytest.importorskip("pyspark") +pytest.importorskip("grpc") + +from datacompy.spark import SparkSQLCompare + +pytestmark = pytest.mark.spark_connect + + +@pytest.fixture(scope="module") +def connect_session(): + """A Spark Connect session with PySpark's global Connect flag cleared. + + ``SparkSession.builder.remote(...)`` sets ``SPARK_CONNECT_MODE_ENABLED``, + and while it is set ``pyspark.sql.functions`` transparently forwards to the + Spark Connect implementations. Real Connect users -- a notebook runtime, a + serverless runtime, a session handed over by another framework -- do not + have that variable set, which is the situation that broke datacompy in + issue #535. Clearing it is what makes these tests able to fail; leaving it + set would make them pass against the unfixed library. + """ + from pyspark.sql import SparkSession + + try: + session = ( + SparkSession.builder.remote("local[2]") + .config("spark.sql.shuffle.partitions", "4") + .config("spark.sql.adaptive.enabled", "false") + .getOrCreate() + ) + except Exception as exc: # pragma: no cover - depends on the environment + pytest.skip(f"could not start a local Spark Connect server: {exc}") + + os.environ.pop("SPARK_CONNECT_MODE_ENABLED", None) + yield session + # Restored before teardown: PySpark's own shutdown path consults is_remote(). + os.environ["SPARK_CONNECT_MODE_ENABLED"] = "1" + session.stop() + + +def test_session_is_connect_without_the_global_flag(connect_session): + """Guard: without this holding, every other test in the file is vacuous.""" + from pyspark.sql.utils import is_remote + + assert type(connect_session).__module__.startswith("pyspark.sql.connect") + assert "SPARK_CONNECT_MODE_ENABLED" not in os.environ + assert not is_remote() + + +def test_compare_and_report(connect_session): + """The core path from issue #535, plus report() rendering.""" + df1 = connect_session.createDataFrame( + pd.DataFrame( + { + "acct_id": [1, 2, 3], + "name": ["george", "mike", "bob"], + "amount": [100.0, 200.0, 300.0], + "active": [True, False, True], + } + ) + ) + df2 = connect_session.createDataFrame( + pd.DataFrame( + { + "acct_id": [1, 2, 4], + "name": ["george", "MIKE", "sue"], + "amount": [100.0, 200.5, 400.0], + "active": [True, False, False], + } + ) + ) + compare = SparkSQLCompare(connect_session, df1, df2, join_columns="acct_id") + + assert not compare.matches() + assert compare.count_matching_rows() == 1 + assert compare.intersect_rows.count() == 2 + + report = compare.report() + assert "DataComPy Comparison" in report + assert "acct_id" in report + + +def test_compare_ignore_case_and_spaces(connect_session): + """Reaches spark_normalize_string_column, which resolves from a Column.""" + df1 = connect_session.createDataFrame( + pd.DataFrame({"acct_id": [1, 2], "name": [" george ", "mike"]}) + ) + df2 = connect_session.createDataFrame( + pd.DataFrame({"acct_id": [1, 2], "name": ["GEORGE", "MIKE"]}) + ) + normalized = SparkSQLCompare( + connect_session, + df1, + df2, + join_columns="acct_id", + ignore_spaces=True, + ignore_case=True, + ) + exact = SparkSQLCompare(connect_session, df1, df2, join_columns="acct_id") + + assert normalized.matches() + assert not exact.matches() + + +def test_duplicate_join_keys(connect_session): + """Exercises _generate_id_within_group, the only Window user.""" + df1 = connect_session.createDataFrame( + pd.DataFrame({"acct_id": [1, 1, 2], "name": ["a", "a", "b"]}) + ) + df2 = connect_session.createDataFrame( + pd.DataFrame({"acct_id": [1, 1, 2], "name": ["a", "z", "b"]}) + ) + compare = SparkSQLCompare(connect_session, df1, df2, join_columns="acct_id") + + assert not compare.matches() + assert compare.count_matching_rows() == 2 + assert compare.report() + + +def test_array_column(connect_session): + """Exercises SparkArrayLikeComparator.""" + df1 = connect_session.createDataFrame( + [(1, [1, 2]), (2, [3])], "acct_id int, tags array" + ) + df2 = connect_session.createDataFrame( + [(1, [1, 2]), (2, [9])], "acct_id int, tags array" + ) + compare = SparkSQLCompare(connect_session, df1, df2, join_columns="acct_id") + + assert not compare.matches() + assert compare.count_matching_rows() == 1 + assert compare.report() + + +def test_numeric_tolerance(connect_session): + """Exercises SparkNumericComparator, including the tolerance branch.""" + df1 = connect_session.createDataFrame( + pd.DataFrame({"acct_id": [1, 2], "amount": [100.0, 200.0]}) + ) + df2 = connect_session.createDataFrame( + pd.DataFrame({"acct_id": [1, 2], "amount": [100.01, 250.0]}) + ) + exact = SparkSQLCompare(connect_session, df1, df2, join_columns="acct_id") + assert not exact.matches() + assert exact.count_matching_rows() == 0 + + # 0.01 is inside the tolerance, 50.0 is not. + toleranced = SparkSQLCompare( + connect_session, df1, df2, join_columns="acct_id", abs_tol=0.1 + ) + assert not toleranced.matches() + assert toleranced.count_matching_rows() == 1 + + +def test_mismatch_helpers(connect_session): + """Exercises all_mismatch and sample_mismatch.""" + df1 = connect_session.createDataFrame( + pd.DataFrame({"acct_id": [1, 2, 3], "amount": [100.0, 200.0, 300.0]}) + ) + df2 = connect_session.createDataFrame( + pd.DataFrame({"acct_id": [1, 2, 3], "amount": [100.0, 999.0, 300.0]}) + ) + compare = SparkSQLCompare(connect_session, df1, df2, join_columns="acct_id") + + assert compare.all_mismatch().count() == 1 + assert compare.sample_mismatch("amount").count() == 1 + + +def test_hide_sensitive_columns(connect_session): + """Exercises hide_sensitive_columns.""" + df1 = connect_session.createDataFrame( + pd.DataFrame({"acct_id": [1, 2], "ssn": ["111", "222"]}) + ) + df2 = connect_session.createDataFrame( + pd.DataFrame({"acct_id": [1, 2], "ssn": ["111", "333"]}) + ) + compare = SparkSQLCompare(connect_session, df1, df2, join_columns="acct_id") + compare.hide_sensitive_columns(["ssn"]) + + report = compare.report() + assert "*******" in report + assert "222" not in report + assert "333" not in report + + +def test_validate_dataframe_accepts_connect_dataframe(connect_session): + """A Connect DataFrame must not be rejected by the type check.""" + df = connect_session.createDataFrame(pd.DataFrame({"acct_id": [1]})) + + # Would raise TypeError if the Connect DataFrame were not recognised. + SparkSQLCompare(connect_session, df, df, join_columns="acct_id") + + with pytest.raises(TypeError): + SparkSQLCompare(connect_session, "not a dataframe", df, join_columns="acct_id") From f215815585898838490d7b57ab581bdb5474df99 Mon Sep 17 00:00:00 2001 From: Faisal Dosani Date: Tue, 4 Aug 2026 22:29:44 -0300 Subject: [PATCH 2/6] test: bind Spark driver and Connect server to 127.0.0.1 --- pytest-ansi.ini | 3 +++ pytest-connect.ini | 12 ++++++++++++ pytest.ini | 3 +++ 3 files changed, 18 insertions(+) diff --git a/pytest-ansi.ini b/pytest-ansi.ini index b5eaca7f..9dd4bed4 100644 --- a/pytest-ansi.ini +++ b/pytest-ansi.ini @@ -5,6 +5,9 @@ markers = spark_connect: needs a Spark Connect session and its own pytest process, because starting a local Connect server sets SPARK_LOCAL_REMOTE, after which every later SparkSession.builder.getOrCreate() in the process returns the Connect session. Run with: pytest -m spark_connect tests/test_spark_connect.py spark_options = spark.master: local[*] + spark.driver.host: localhost + spark.driver.bindAddress: 127.0.0.1 + spark.connect.grpc.binding.address: 127.0.0.1 spark.sql.catalogImplementation: in-memory spark.sql.shuffle.partitions: 4 spark.default.parallelism: 4 diff --git a/pytest-connect.ini b/pytest-connect.ini index 42e18084..d61faa9c 100644 --- a/pytest-connect.ini +++ b/pytest-connect.ini @@ -11,6 +11,15 @@ ; `spark.sql.execution.arrow.pyspark.enabled` is a no-op because Connect always ; uses Arrow on the wire. ; +; `spark.connect.grpc.binding.address` keeps the Connect server on loopback -- +; it listens on *:15002 otherwise. The driver side is pinned by SPARK_LOCAL_IP, +; which tests/conftest.py sets, because an ini can only set Spark configs and +; the driver bind address is not settable that way once the JVM is up. +; +; Behind a corporate proxy you also need `no_proxy` to contain the literal host +; from the connection string -- gRPC honours `http_proxy` even for loopback and +; matches `no_proxy` on the exact host. tests/conftest.py handles that too. +; ; The tests marked `spark_connect` are excluded: they build their own session ; and must run in a separate pytest process. [pytest] @@ -20,6 +29,9 @@ markers = spark_connect: needs a Spark Connect session and its own pytest process, because starting a local Connect server sets SPARK_LOCAL_REMOTE, after which every later SparkSession.builder.getOrCreate() in the process returns the Connect session. Run with: pytest -m spark_connect tests/test_spark_connect.py spark_connect_url = local[2] spark_options = + spark.connect.grpc.binding.address: 127.0.0.1 + spark.driver.host: localhost + spark.driver.bindAddress: 127.0.0.1 spark.sql.shuffle.partitions: 4 spark.sql.adaptive.enabled: false spark.sql.ansi.enabled: false diff --git a/pytest.ini b/pytest.ini index 0eb21bb7..0eb88bfd 100644 --- a/pytest.ini +++ b/pytest.ini @@ -5,6 +5,9 @@ markers = spark_connect: needs a Spark Connect session and its own pytest process, because starting a local Connect server sets SPARK_LOCAL_REMOTE, after which every later SparkSession.builder.getOrCreate() in the process returns the Connect session. Run with: pytest -m spark_connect tests/test_spark_connect.py spark_options = spark.master: local[*] + spark.driver.host: localhost + spark.driver.bindAddress: 127.0.0.1 + spark.connect.grpc.binding.address: 127.0.0.1 spark.sql.catalogImplementation: in-memory spark.sql.shuffle.partitions: 4 spark.default.parallelism: 4 From 157e19045b0f19368261fe3d2484bcba14990eb5 Mon Sep 17 00:00:00 2001 From: Faisal Dosani Date: Wed, 5 Aug 2026 12:34:40 -0300 Subject: [PATCH 3/6] fix: include testpaths in the pytest ini files --- pytest-ansi.ini | 2 ++ pytest-connect.ini | 2 ++ pytest.ini | 5 +++++ 3 files changed, 9 insertions(+) diff --git a/pytest-ansi.ini b/pytest-ansi.ini index 9dd4bed4..352bfb2a 100644 --- a/pytest-ansi.ini +++ b/pytest-ansi.ini @@ -1,4 +1,6 @@ [pytest] +; See pytest.ini -- keeps default collection out of sibling worktrees. +testpaths = tests addopts = -m "not spark_connect" markers = pyspark: needs pyspark installed diff --git a/pytest-connect.ini b/pytest-connect.ini index d61faa9c..4446d4e8 100644 --- a/pytest-connect.ini +++ b/pytest-connect.ini @@ -23,6 +23,8 @@ ; The tests marked `spark_connect` are excluded: they build their own session ; and must run in a separate pytest process. [pytest] +; See pytest.ini -- keeps default collection out of sibling worktrees. +testpaths = tests addopts = -m "not spark_connect" markers = pyspark: needs pyspark installed diff --git a/pytest.ini b/pytest.ini index 0eb88bfd..ce131952 100644 --- a/pytest.ini +++ b/pytest.ini @@ -1,4 +1,9 @@ [pytest] +; Confine default collection to the test suite. Without this, a bare `pytest` +; walks the whole repo, and a git worktree checked out under the repo root +; (e.g. a PR review checkout) contributes a second tests/conftest.py that +; collides with this one -- pytest aborts with ImportPathMismatchError. +testpaths = tests addopts = -m "not spark_connect" markers = pyspark: needs pyspark installed From e3b2c23af7e98245b8af42086c91903e32ec62d5 Mon Sep 17 00:00:00 2001 From: Faisal Dosani Date: Wed, 5 Aug 2026 13:01:54 -0300 Subject: [PATCH 4/6] WIP: updating tox and make files --- .gitignore | 5 +- Makefile | 50 +++++++++++++- tox.ini | 195 +++++++++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 247 insertions(+), 3 deletions(-) create mode 100644 tox.ini diff --git a/.gitignore b/.gitignore index 514c4979..dd3929a8 100644 --- a/.gitignore +++ b/.gitignore @@ -26,10 +26,11 @@ docs/source/api/ test.html -.coverage* +.coverage +.coverage.* coverage.xml +.tox/ # benchmark datasets benchmarks/data/ -.tox/ diff --git a/Makefile b/Makefile index bfd3f47e..a7e61ed9 100644 --- a/Makefile +++ b/Makefile @@ -1,3 +1,51 @@ +PYTEST ?= python -m pytest +TESTS ?= tests +# Extra pytest args injected into every test target. `test-*-no-snowflake` +# targets set this to deselect the Snowflake suites, which need a live +# Snowflake session. GNU make applies a target-specific variable to the +# target's prerequisites too, which is what lets the aggregate targets reuse +# the individual ones. +PYTEST_ARGS ?= + +.PHONY: sphinx ghpages \ + test test-ansi test-connect test-connect-regression \ + test-cov test-all test-all-no-snowflake test-no-snowflake + +# Default suite: classic Spark session, ANSI mode off. +test: + $(PYTEST) $(PYTEST_ARGS) $(TESTS) + +# Same suite with spark.sql.ansi.enabled=true. +test-ansi: + $(PYTEST) -c pytest-ansi.ini $(PYTEST_ARGS) $(TESTS) + +# The existing Spark suite run against a Spark Connect session. Only the Spark +# tests: the rest are backend-agnostic and gain nothing from a second run. +test-connect: + $(PYTEST) -c pytest-connect.ini $(PYTEST_ARGS) $(TESTS)/test_spark.py $(TESTS)/comparator + +# Spark Connect regression suite. Excluded from every other target by +# `addopts`, and must be its own pytest process: starting a local Connect +# server sets SPARK_LOCAL_REMOTE, after which every later +# SparkSession.builder.getOrCreate() in the process returns the Connect session. +test-connect-regression: + $(PYTEST) -m spark_connect $(PYTEST_ARGS) $(TESTS)/test_spark_connect.py + +test-cov: + $(PYTEST) $(PYTEST_ARGS) --cov=datacompy --cov-report=term-missing $(TESTS) + +# Everything. Needs Java 17, pyspark[connect], and a live Snowflake session. +test-all: test test-ansi test-connect test-connect-regression + +# Everything except the Snowflake suites, which error without a live session. +# `--snowflake-session local` is not an alternative here: Snowpark's local +# testing mode is an emulator and most of the Snowflake suite fails against it. +test-no-snowflake: PYTEST_ARGS += -k "not snowflake" +test-no-snowflake: test + +test-all-no-snowflake: PYTEST_ARGS += -k "not snowflake" +test-all-no-snowflake: test-all + sphinx: cd docs && \ make -f Makefile clean && \ @@ -9,4 +57,4 @@ ghpages: cp -r docs/build/html/* . && \ git add -u && \ git add -A && \ - PRE_COMMIT_ALLOW_NO_CONFIG=1 git commit -m "Updated generated Sphinx documentation" \ No newline at end of file + PRE_COMMIT_ALLOW_NO_CONFIG=1 git commit -m "Updated generated Sphinx documentation" diff --git a/tox.ini b/tox.ini new file mode 100644 index 00000000..f2321273 --- /dev/null +++ b/tox.ini @@ -0,0 +1,195 @@ +; Tox configuration for datacompy's local test matrix, using conda-managed +; environments (tox-conda) so Java and PySpark can be pulled from conda-forge +; rather than requiring a preinstalled JDK. +; +; The matrix mirrors ``.github/workflows/test-package.yml``: +; +; CI job tox envs +; -------------------------- -------------------------------------------- +; lint-and-format lint +; test-basic-install py{310,311,312,313}-nospark +; test-with-spark-3-install py310-spark35-pandas2 +; py311-spark35-pandas{2,3} +; test-with-spark-4-install py310-spark4-pandas2 +; py{311,312,313}-spark4-pandas{2,3} +; +; CI excludes python 3.10 + pandas 3.0.3, so this file does too. +; +; ``py311-spark35-connect`` has no CI counterpart -- see its comment below. +; +; Snowflake tests are skipped automatically via +; ``pytest.importorskip("snowflake.snowpark")`` when the snowflake extra is not +; installed, and CI does not install it either -- so this file omits it. +; +; tox-conda 0.10.x only supports tox 3 (``envlist`` not ``env_list``, +; ``usedevelop`` not ``package = editable``). +; +; Local usage: +; pip install "tox<4" tox-conda +; tox # the full matrix -- slow, and ~20GB of conda +; # envs and pyspark wheels. Prefer -e. +; tox -e lint # ruff check + ruff format --check +; tox -e py312-spark4-pandas3 # one environment +; tox -e py311-spark35-pandas2 # Spark 3.5 against Python 3.11 / pandas 2 +; tox -e typecheck # mypy (not in envlist; not run by CI) +; tox -e py312-spark4-pandas3 -- tests/test_pandas.py # forward to pytest +; +; NOTE: ``{posargs}`` replaces the default test paths in *every* command of an +; env, including the Spark Connect ones, so passing a path runs that path under +; each pytest config rather than appending a fifth invocation. + +[tox] +requires = + tox<4 + tox-conda +envlist = + lint + py{310,311,312,313}-nospark + py310-spark35-pandas2 + py311-spark35-pandas{2,3} + py311-spark35-connect + py310-spark4-pandas2 + py{311,312,313}-spark4-pandas{2,3} + +[testenv] +description = Run the datacompy test suite for {envname} +usedevelop = true +; SPARK_LOCAL_REMOTE is deliberately NOT passed through. If it is set in the +; calling shell -- a stray export, or a leftover from a Connect run in the same +; terminal -- every ``SparkSession.builder.getOrCreate()`` returns a Connect +; session, and the classic-session envs silently stop testing classic sessions. +passenv = + JAVA_HOME + HADOOP_HOME + SPARK_HOME + SPARK_LOCAL_IP + http_proxy + https_proxy + no_proxy + HTTP_PROXY + HTTPS_PROXY + NO_PROXY +; COVERAGE_FILE: every env runs pytest from {toxinidir}, so without this they +; all write the same ``.coverage`` and each env erases the previous env's data. +; Per-env files can be merged afterwards with ``coverage combine``. +setenv = + PYTHONDONTWRITEBYTECODE = 1 + COVERAGE_FILE = {toxinidir}/.coverage.{envname} +conda_channels = + conda-forge +; ``--override-channels`` keeps conda off any base channels configured in the +; user's ``.condarc``, so the solve is reproducible across machines. +; +; That alone does NOT keep GraalPy out. conda-forge itself ships GraalVM builds +; of openjdk 17 (``17.0.5.8=0_graalvm223b08`` and ``17.0.7.4=0_graalvm230b10``), +; and picking one drags in ``graalpy-graalvm``, which replaces CPython with +; ``python-3.10.8-*_graalpy``. pip then dies with a Truffle +; ``Unable to load native posix support library`` error. It only bites the +; python 3.10 envs, because GraalPy implements python 3.10 and so is a legal +; solution for ``python=3.10`` but not for 3.11+. Both GraalVM builds are below +; 17.0.8, so the ``openjdk`` floor in each spark env's ``conda_deps`` excludes +; them by construction -- do not relax it back to a bare ``openjdk=17``. +conda_create_args = + --override-channels +conda_install_args = + --override-channels + +; Mirrors the ``lint-and-format`` CI job, which pins python 3.12. +[testenv:lint] +description = ruff lint + format check +basepython = python3.12 +extras = + qa +commands = + {envbindir}/python -m ruff check {posargs} + {envbindir}/python -m ruff format --check {posargs} + +; Not part of envlist and not run by CI, but ``mypy --strict`` is a documented +; project convention (see CLAUDE.md), so make it available as one command. +[testenv:typecheck] +description = mypy type check +basepython = python3.12 +extras = + qa +commands = + {envbindir}/python -m mypy {posargs:.} + +; Basic install: no pyspark, no snowflake. pandas/polars/base/report/comparator +; tests run; spark and snowflake tests are skipped via ``importorskip``. +; Mirrors ``test-basic-install``, which likewise pins no pandas version. +[testenv:py{310,311,312,313}-nospark] +extras = + qa + tests +commands = + {envbindir}/python -m pytest --cov=datacompy --cov-report=term-missing {posargs} + +; Spark 3.5 track, mirroring ``test-with-spark-3-install``. pyspark is pinned to +; 3.5.8 and pandas to the two versions CI exercises. openjdk 17 comes from +; conda-forge. Spark Connect is NOT exercised here -- see +; ``py311-spark35-connect`` below for that. +[testenv:py{310,311}-spark35-pandas{2,3}] +extras = + qa + tests + tests-spark +conda_deps = + openjdk>=17.0.8,<18 +deps = + pyspark[connect]==3.5.8 + pandas2: pandas==2.3.3 + pandas3: pandas==3.0.3 +commands = + {envbindir}/python -m pytest --cov=datacompy --cov-report=term-missing {posargs} + {envbindir}/python -m pytest -c pytest-ansi.ini --cov=datacompy --cov-report=term-missing --cov-append {posargs} + +; Spark 3.5 + Spark Connect. Experimental: Spark 3.5 does not bundle the +; Connect server jar in the pyspark wheel (4.x does), so we ask the JVM to +; resolve it at gateway-launch time via Ivy by setting ``PYSPARK_SUBMIT_ARGS`` +; -- PySpark's ``launch_gateway`` uses that env var to build the ``spark-submit`` +; command, and ``--packages`` there triggers Ivy resolution before the Connect +; plugin class is loaded. The first run downloads the jar from Maven Central +; and caches it under ``~/.ivy2``; subsequent runs are offline. CI has never +; exercised this combination, so treat failures here as informative rather +; than blocking -- if a red ``tox`` run is a problem, drop this from envlist. +[testenv:py311-spark35-connect] +extras = + qa + tests + tests-spark +conda_deps = + openjdk>=17.0.8,<18 +deps = + pyspark[connect]==3.5.8 +setenv = + {[testenv]setenv} + PYSPARK_SUBMIT_ARGS = --packages org.apache.spark:spark-connect_2.12:3.5.8 pyspark-shell +commands = + {envbindir}/python -m pytest -c pytest-connect.ini {posargs:tests/test_spark.py tests/comparator/} + {envbindir}/python -m pytest -m spark_connect {posargs:tests/test_spark_connect.py} + +; Spark 4 track, mirroring ``test-with-spark-4-install``. Includes classic +; sessions (default + ANSI mode) and the two Spark Connect suites. Connect must +; run in its own pytest process because starting a local Connect server sets +; SPARK_LOCAL_REMOTE, after which every later +; ``SparkSession.builder.getOrCreate()`` returns the Connect session -- so a +; classic and a Connect session cannot coexist in one run. +; +; The Connect commands carry no ``--cov``: they re-run tests already measured by +; the two classic commands, and CI does not measure them either. +[testenv:py{310,311,312,313}-spark4-pandas{2,3}] +extras = + qa + tests + tests-spark +conda_deps = + openjdk>=17.0.8,<18 +deps = + pyspark[connect]==4.1.2 + pandas2: pandas==2.3.3 + pandas3: pandas==3.0.3 +commands = + {envbindir}/python -m pytest --cov=datacompy --cov-report=term-missing {posargs} + {envbindir}/python -m pytest -c pytest-ansi.ini --cov=datacompy --cov-report=term-missing --cov-append {posargs} + {envbindir}/python -m pytest -c pytest-connect.ini {posargs:tests/test_spark.py tests/comparator/} + {envbindir}/python -m pytest -m spark_connect {posargs:tests/test_spark_connect.py} From bfc76843e87dd3c10cf9eeb8820cdc8350bb76dd Mon Sep 17 00:00:00 2001 From: Faisal Dosani Date: Wed, 5 Aug 2026 14:07:43 -0300 Subject: [PATCH 5/6] ci: cover the test axes independently and mirror the matrix in tox --- .github/workflows/test-package.yml | 76 +++++++++++++------------- CLAUDE.md | 67 ++++++++++++++++------- pyproject.toml | 3 +- tox.ini | 87 +++++++++++++++++++++--------- 4 files changed, 150 insertions(+), 83 deletions(-) diff --git a/.github/workflows/test-package.yml b/.github/workflows/test-package.yml index 1db87813..f70dffea 100644 --- a/.github/workflows/test-package.yml +++ b/.github/workflows/test-package.yml @@ -1,5 +1,23 @@ # This workflow will install Python dependencies, run tests and lint with a variety of Python versions # For more information see: https://help.github.com/actions/language-and-framework-guides/using-python-with-github-actions +# +# Matrix design. The Spark jobs are the entire cost of this workflow (~20 min +# each); lint and the four basic-install jobs finish in under a minute. So the +# test axes are covered independently rather than as a cross product: +# +# * python 3.10-3.13 breadth comes from the four cheap basic-install jobs, +# which exercise all the pandas/polars/base/report/comparator code. +# * pandas 2 vs 3 is a real API split, so both appear -- pandas 2.3.3 on the +# python 3.10 Spark job (3.10 cannot take pandas 3), pandas 3.0.3 on 3.12. +# * spark 3.5 vs 4 is a real API split, so both appear. +# * ANSI mode and Spark Connect are spark-side semantics, orthogonal to the +# python and pandas versions, so they run once on the 3.12 baseline rather +# than on every Spark job. +# +# Known gap: no Spark job runs on python 3.11 or 3.13, so a Spark break +# specific to those runtimes would not be caught. Widening is one entry in the +# test-with-spark-4-install include list (and one line in tox.ini's envlist, +# which mirrors this file job for job). name: Test package @@ -29,29 +47,18 @@ jobs: - name: Formatting by ruff run: ruff format --check + # Legacy Spark path. One job is enough to prove 3.5 still works; the wider + # combinations remain available locally via `tox -e py310-spark35-pandas2` + # and friends. test-with-spark-3-install: + name: spark 3.5 (py3.11, pandas 2.3.3) runs-on: ubuntu-latest - strategy: - fail-fast: false - matrix: - python-version: ["3.10", "3.11"] - spark-version: [3.5.8] - pandas-version: ["2.3.3", "3.0.3"] - exclude: - - python-version: "3.10" - pandas-version: "3.0.3" - - env: - PYTHON_VERSION: ${{ matrix.python-version }} - SPARK_VERSION: ${{ matrix.spark-version }} - PANDAS_VERSION: ${{ matrix.pandas-version }} - steps: - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7 - - name: Set up Python ${{ matrix.python-version }} + - name: Set up Python 3.11 uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6.3.0 with: - python-version: ${{ matrix.python-version }} + python-version: "3.11" - name: Setup Java JDK uses: actions/setup-java@ad2b38190b15e4d6bdf0c97fb4fca8412226d287 # v5.3.0 with: @@ -59,37 +66,32 @@ jobs: distribution: "adopt" - name: Install pyspark run: | - python -m pip install pyspark[connect]==${{ matrix.spark-version }} + python -m pip install pyspark[connect]==3.5.8 - name: Install pandas run: | - python -m pip install pandas==${{ matrix.pandas-version }} + python -m pip install pandas==2.3.3 - name: Install datacompy run: | python -m pip install ."[spark, qa, tests, tests-spark]" - name: Test with pytest run: | python -m pytest --cov=datacompy --cov-report=xml --cov-report=term-missing - - name: Test with pytest (ANSI mode) - run: | - python -m pytest -c pytest-ansi.ini --cov=datacompy --cov-report=xml --cov-report=term-missing - test-with-spark-4-install: + name: spark 4 (py${{ matrix.python-version }}, pandas ${{ matrix.pandas-version }}) runs-on: ubuntu-latest strategy: fail-fast: false matrix: - python-version: ["3.10", "3.11", "3.12", "3.13"] - spark-version: [4.1.2] - pandas-version: ["2.3.3", "3.0.3"] - exclude: + include: + # python 3.10 cannot take pandas 3, so it carries the pandas 2 slot. - python-version: "3.10" + pandas-version: "2.3.3" + extended: false + # The baseline. Also the only job running ANSI mode and Spark Connect. + - python-version: "3.12" pandas-version: "3.0.3" - - env: - PYTHON_VERSION: ${{ matrix.python-version }} - SPARK_VERSION: ${{ matrix.spark-version }} - PANDAS_VERSION: ${{ matrix.pandas-version }} + extended: true steps: - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7 @@ -104,7 +106,7 @@ jobs: distribution: "adopt" - name: Install pyspark run: | - python -m pip install pyspark[connect]==${{ matrix.spark-version }} + python -m pip install pyspark[connect]==4.1.2 - name: Install pandas run: | python -m pip install pandas==${{ matrix.pandas-version }} @@ -115,6 +117,7 @@ jobs: run: | python -m pytest --cov=datacompy --cov-report=xml --cov-report=term-missing - name: Test with pytest (ANSI mode) + if: matrix.extended run: | python -m pytest -c pytest-ansi.ini --cov=datacompy --cov-report=xml --cov-report=term-missing # Spark Connect runs in two separate pytest processes: a classic and a @@ -124,12 +127,16 @@ jobs: # Only on Spark 4.x: the Connect server jar is bundled in the 4.x wheel, # whereas Spark 3.5 requires resolving it via --packages. - name: Test with pytest (Spark Connect - existing suite) + if: matrix.extended run: | python -m pytest -c pytest-connect.ini tests/test_spark.py tests/comparator/ - name: Test with pytest (Spark Connect - regression suite) + if: matrix.extended run: | python -m pytest -m spark_connect tests/test_spark_connect.py + # Cheap (<1 min each) and the only job that proves datacompy imports and + # works with no pyspark installed, so it keeps the full python range. test-basic-install: runs-on: ubuntu-latest strategy: @@ -137,9 +144,6 @@ jobs: matrix: python-version: ["3.10", "3.11", "3.12", "3.13"] - env: - PYTHON_VERSION: ${{ matrix.python-version }} - steps: - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7 - name: Set up Python ${{ matrix.python-version }} diff --git a/CLAUDE.md b/CLAUDE.md index 4b0d8f70..27664de4 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -4,9 +4,7 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co ## Project Overview -DataComPy is a Python library for comparing two DataFrames/tables across multiple backends: Pandas, Polars, Spark, and Snowflake. It originated as a replacement for SAS's `PROC COMPARE`. v1 is GA; the version lives in `datacompy/__version__` and `pyproject.toml` derives the distribution version from it. - -This file is the single AI agent guide for the repository. `.github/copilot-instructions.md` used to duplicate it and was removed; put new guidance here rather than starting a second copy. +DataComPy is a Python library for comparing two DataFrames/tables across multiple backends: Pandas, Polars, Spark, and Snowflake. It originated as a replacement for SAS's `PROC COMPARE`. v1 is released; `datacompy/__init__.py`'s `__version__` is the source of truth for the current version. ## Common Commands @@ -17,6 +15,7 @@ pre-commit install ``` ### Testing + ```bash pytest # all tests pytest tests/test_pandas.py # single backend @@ -27,18 +26,39 @@ pytest -c pytest-connect.ini tests/test_spark.py tests/comparator/ # existing S pytest -m spark_connect tests/test_spark_connect.py # Spark Connect regression suite ``` -CI runs the suite twice, once with the default `pytest.ini` and once with `-c pytest-ansi.ini`, which only differs by `spark.sql.ansi.enabled`. A change touching Spark casting or null handling needs both. +The Makefile wraps each of these: `make test`, `test-ansi`, `test-connect`, `test-connect-regression`, `test-cov`, `test-all`, plus `test-no-snowflake` / `test-all-no-snowflake`. + +There are three pytest configs, and `-c` **replaces** the config file rather than layering on it — anything a run needs (markers, `testpaths`, `spark_options`) must be present in the file it names: + +- `pytest.ini` — classic Spark session, ANSI off +- `pytest-ansi.ini` — same, `spark.sql.ansi.enabled=true` +- `pytest-connect.ini` — Spark Connect session (`spark_connect_url`); must not set `spark.master`, since PySpark rejects a session with both + +All three set `testpaths = tests`. Without it a bare `pytest` walks the whole repo, and a git worktree checked out under the repo root contributes a second `tests/conftest.py` that collides with the real one — pytest then aborts the entire run with `ImportPathMismatchError`. Keep PR-review worktrees outside the repo. + +Spark tests require Java 17 and `pyspark` installed. Snowflake tests require a live Snowflake session; `--snowflake-session local` is *not* a substitute, as Snowpark's local testing mode is an emulator that most of the Snowflake suite fails against. + +`benchmarks/` holds pytest-benchmark suites (`benchmark.py`, `benchmark_spark.py`). They sit outside `testpaths`, so no ordinary run collects them, and they read parquet fixtures that `python benchmarks/generate_data.py` has to write first. Results are written up in `docs/source/benchmark.rst`. + +The two Spark Connect commands must each run in their own pytest process, and are excluded from the default run via `addopts`. Starting a local Spark Connect server sets `SPARK_LOCAL_REMOTE`, after which every later `SparkSession.builder.getOrCreate()` in that process returns the Connect session — so a classic and a Connect session cannot coexist in one run. + +### Local CI matrix (tox) -**Coverage:** always target the top-level package (`--cov=datacompy`). Passing a dotted submodule such as `--cov=datacompy.cli` triggers a numpy double-import in some environments and fails ~100 otherwise-passing tests with a confusing `_NoValueType` `TypeError`. +`tox.ini` mirrors `.github/workflows/test-package.yml` job for job, using `tox-conda` so Java and PySpark come from conda-forge rather than a system JDK. It needs `pip install "tox<4" tox-conda` — tox-conda 0.10.x only supports tox 3. -**Spark** needs `pyspark` and **Java 17** (newer JDKs fail with `py4j.protocol` errors). If the JDK came from conda (`conda install openjdk=17`, as `[edgetest.envs.core]` does), it is at `$CONDA_PREFIX/lib/jvm` and `JAVA_HOME` must point there. Activating the env normally sets this; a non-interactive shell will not inherit it: ```bash -export JAVA_HOME=$CONDA_PREFIX/lib/jvm +tox # the whole envlist +tox -e lint # ruff check + ruff format --check +tox -e py312-spark4-pandas3 # the one env covering ANSI + Spark Connect +tox -e typecheck # mypy (not in envlist; not run by CI) ``` -**Snowflake** tests need a live session, or `--snowflake-session local` for Snowpark's local testing mode. Local mode is an emulator, not Snowflake: `eqNullSafe` returns `True` for every row and high-precision decimals are truncated on DataFrame creation. Tests that depend on either must request the `requires_live_snowflake_session` fixture (`tests/conftest.py`), which skips them in local mode. +Both files deliberately cover the test axes independently rather than as a cross product, because the Spark jobs are the entire cost of a run (~20 min each) while the four no-Spark jobs finish in under a minute. Python breadth comes from the cheap no-Spark envs; ANSI mode and Spark Connect are Spark-side semantics and run once on the 3.12 baseline. Read `tox.ini`'s header before widening the matrix — it records the known gap and what closing it costs. -The two Spark Connect commands must each run in their own pytest process, and are excluded from the default run via `addopts`. Starting a local Spark Connect server sets `SPARK_LOCAL_REMOTE`, after which every later `SparkSession.builder.getOrCreate()` in that process returns the Connect session — so a classic and a Connect session cannot coexist in one run. +Two traps that cost real time to rediscover, both documented in `tox.ini`: + +- **Never loosen the `openjdk>=17.0.8,<18` floor** in the Spark envs. conda-forge ships GraalVM builds of openjdk 17 below that floor; selecting one pulls in `graalpy-graalvm`, which silently replaces CPython with GraalPy, and pip then dies with a Truffle `Unable to load native posix support library` error. It only affects Python 3.10 envs, since GraalPy implements 3.10. +- **`{envpython}` substitutes to the string `None` in tox 3.28** — use `{envbindir}/python`. The absolute path also avoids tox 3's silent fallback to a `$PATH` interpreter when a command is missing from a freshly-created env, which otherwise runs the suite against the wrong Python and only warns. ### Linting & Formatting ```bash @@ -49,9 +69,9 @@ ruff format # apply formatting mypy . # type-check (strict mode) ``` -**Use the `ruff` version pinned in `.pre-commit-config.yaml`.** The config uses recent selectors, and an older ruff fails to parse `pyproject.toml` at all rather than degrading gracefully. `pre-commit run --all-files` fetches the right version itself. +`pyproject.toml` uses selectors (`noqa-comments`, `rule-codes-in-selectors`) that only exist in recent ruff, hence the `ruff>=0.16` floor on the `qa` extra. An older ruff fails while *parsing* the config with `Unknown rule selector`, before linting anything. Note that ruff formats Python code blocks inside Markdown, so `CLAUDE.md` and other docs are in scope for `ruff format --check`. -**`mypy .` has a large pre-existing error baseline** (~185, mostly in `snowflake.py`, `polars.py`, and `pandas.py`) and is enforced by neither CI nor pre-commit; CI lint runs only `ruff check` and `ruff format --check`. New code is still expected to be clean, so check that your diff introduces no *new* errors rather than that the run is empty, and do not refactor unrelated modules to chase the baseline. Missing-stub errors for `pyspark` and `snowflake.snowpark` mean those extras are not installed in the current environment, not a code defect. +`mypy` is **not** a pre-commit hook and is not run by CI — the hooks are ruff, ruff-format, trailing-whitespace, debug-statements, end-of-file-fixer, and pyproject-fmt. Run it explicitly or via `tox -e typecheck`. ### Documentation ```bash @@ -80,21 +100,30 @@ Beyond the report, each backend exposes `df1_unq_rows`, `df2_unq_rows`, and `int `datacompy/comparator/` provides type-specific column comparison logic, also using a strategy pattern: -- `base.py` → `BaseComparator` ABC with `compare(col1, col2)` method +- `base.py` → `BaseComparator` ABC with `compare(col1, col2, **kwargs)` method - `numeric.py` → Numeric comparators per backend (handles tolerances) - `string.py` → String comparators per backend +- `boolean.py` → Boolean comparators per backend - `array.py` → Array-like comparators per backend - `utility.py` → Shared Spark/Snowflake helpers, including `get_spark_functions` / `get_spark_window` Each type has backend-specific implementations: `Pandas*Comparator`, `Polars*Comparator`, `Spark*Comparator`, `Snowflake*Comparator`. +**Dispatch protocol** — the column-compare helper in each backend module (`columns_equal` and friends) walks a comparator list in order and takes the first result that isn't `None`: + +- **`compare()` returning `None` means "not my type, try the next one."** This is how type dispatch happens — there is no `can_compare()`. A comparator that raises instead of returning `None` breaks the chain. +- Order matters. Each backend defines `__DEFAULT_COMPARATORS` (array-like → boolean → numeric → string) at module scope. +- All four backends accept `custom_comparators=[...]` on the constructor; `_get_comparators()` puts them **before** the defaults, so a custom comparator can pre-empt a built-in for a column type. +- The dispatch loop passes the built-ins their type-specific kwargs via `isinstance` branches (tolerances to numeric, `ignore_spaces`/`ignore_case` to string), and passes custom comparators **everything** as `**kwargs`. Custom comparators must therefore accept and ignore kwargs they don't use. +- If every comparator returns `None`, the column compares as all-`False` rather than erroring. + ### Spark Connect Never import `pyspark.sql.functions` or `pyspark.sql.Window` at module scope in Spark code paths. Those dispatch to the Spark Connect implementations only when the process-global `SPARK_CONNECT_MODE_ENABLED` environment variable is set, which a Connect session from a notebook or serverless runtime does not necessarily set. Instead resolve them from the DataFrame or Column being operated on: ```python -F = get_spark_functions(dataframe) # datacompy/spark.py -psf = get_spark_functions(dataframe) # datacompy/comparator/*.py +F = get_spark_functions(dataframe) # datacompy/spark.py +psf = get_spark_functions(dataframe) # datacompy/comparator/*.py Window = get_spark_window(dataframe) ``` @@ -104,6 +133,8 @@ Because there is no module-level binding, ruff's `F821` flags any call site that Reports use Jinja2 templates from `datacompy/templates/report_template.j2`. The `render()` function in `base.py` handles template resolution. Custom templates can be passed via `report(template_path=...)`. +`build_report_data()` (defined once on `BaseCompare`) is the structured counterpart to `report()`, returning a typed `ReportData` object for dashboards or JSON export without parsing the string report. `ReportData` and its member dataclasses (`RowSummary`, `ColumnSummary`, `ColumnComparison`, `MismatchStat`, `MismatchStats`, `UniqueRowsData`) live in `datacompy/report.py` and are re-exported from `datacompy/__init__.py` — unlike the backends, they are unconditional top-level exports. It is public API — changes to its shape are breaking. Snapshot tests in `tests/test_report_snapshots.py` compare rendered output against fixtures in `tests/snapshots/`, which are excluded from the trailing-whitespace and end-of-file pre-commit hooks because their exact bytes are the assertion. + ### Tolerance Handling Tolerances (`abs_tol`, `rel_tol`) can be a single float (applied globally) or a dict mapping column names to per-column values. Validated by `validate_tolerance_parameter()` in `base.py`. @@ -138,8 +169,6 @@ Tolerances (`abs_tol`, `rel_tol`) can be a single float (applied globally) or a ## Branching -- `main` is the release branch and currently the most advanced one. Recent release commits land here, so branch from `main` unless told otherwise. -- `develop` predates the v1 GA and lags `main`. Do not assume it is the integration branch without checking `git log origin/main origin/develop`. -- `support/0.19.x` is maintained for v0 users (bug fixes only). - -CI (`.github/workflows/test-package.yml`) runs on `develop`, `main`, `release/*`, `release-*`, and `support/*`. +- `main` is the default branch and the target for all active development (this changed with the v1 release — `develop` still exists and CI still builds it, but it is no longer where work lands) +- `support/0.19.x` is archived: critical security fixes only, best-effort, no features or regular maintenance +- CI runs on pushes and PRs to `develop`, `main`, `release/*`, `release-*`, and `support/*` diff --git a/pyproject.toml b/pyproject.toml index 095f3319..02b3efe3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -28,7 +28,6 @@ classifiers = [ "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", "Programming Language :: Python :: 3.13", - "Programming Language :: Python :: 3.14", ] dynamic = [ "version" ] dependencies = [ @@ -50,7 +49,7 @@ optional-dependencies.dev = [ ] optional-dependencies.docs = [ "furo", "myst-parser", "sphinx" ] optional-dependencies.edgetest = [ "edgetest", "edgetest-conda" ] -optional-dependencies.qa = [ "mypy", "pandas-stubs", "pre-commit", "ruff" ] +optional-dependencies.qa = [ "mypy", "pandas-stubs", "pre-commit", "ruff>=0.16" ] optional-dependencies.snowflake = [ "snowflake-snowpark-python>=1.37,<1.55" ] optional-dependencies.spark = [ "pyspark[connect]>=3.5,!=4,<=4.2; python_version<='3.11'", diff --git a/tox.ini b/tox.ini index f2321273..81618d13 100644 --- a/tox.ini +++ b/tox.ini @@ -2,20 +2,34 @@ ; environments (tox-conda) so Java and PySpark can be pulled from conda-forge ; rather than requiring a preinstalled JDK. ; -; The matrix mirrors ``.github/workflows/test-package.yml``: +; envlist mirrors ``.github/workflows/test-package.yml`` job for job: ; -; CI job tox envs +; CI job tox env ; -------------------------- -------------------------------------------- ; lint-and-format lint ; test-basic-install py{310,311,312,313}-nospark -; test-with-spark-3-install py310-spark35-pandas2 -; py311-spark35-pandas{2,3} +; test-with-spark-3-install py311-spark35-pandas2 ; test-with-spark-4-install py310-spark4-pandas2 -; py{311,312,313}-spark4-pandas{2,3} +; py312-spark4-pandas3 <- also ANSI + Connect ; -; CI excludes python 3.10 + pandas 3.0.3, so this file does too. +; WHY THE SPARK MATRIX IS SMALL. Spark jobs are the entire cost of a test run +; (~20 min each; the four nospark jobs finish in under a minute). The axes are +; therefore covered independently rather than as a cross product: ; -; ``py311-spark35-connect`` has no CI counterpart -- see its comment below. +; * python 3.10-3.13 breadth comes from the four cheap nospark envs, which +; exercise all the pandas/polars/base/report/comparator code. +; * pandas 2 vs 3 is a real API split, so both appear -- pandas2 on the +; python 3.10 env (CI cannot pair 3.10 with pandas 3), pandas3 on 3.12. +; * spark 3.5 vs 4 is a real API split, so both appear. +; * ANSI mode and Spark Connect are spark-side semantics, orthogonal to the +; python and pandas versions, so they run once on the 3.12 baseline rather +; than on every spark env. +; +; KNOWN GAP: no spark env runs on python 3.11 or 3.13, so a Spark break +; specific to those runtimes would not be caught here. Widening is one line -- +; add e.g. ``py313-spark4-pandas3`` to envlist and to the CI matrix. +; +; ``py311-spark35-connect`` is deliberately NOT in envlist -- see its comment. ; ; Snowflake tests are skipped automatically via ; ``pytest.importorskip("snowflake.snowpark")`` when the snowflake extra is not @@ -26,14 +40,15 @@ ; ; Local usage: ; pip install "tox<4" tox-conda -; tox # the full matrix -- slow, and ~20GB of conda -; # envs and pyspark wheels. Prefer -e. +; tox # the whole envlist ; tox -e lint # ruff check + ruff format --check -; tox -e py312-spark4-pandas3 # one environment -; tox -e py311-spark35-pandas2 # Spark 3.5 against Python 3.11 / pandas 2 +; tox -e py312-spark4-pandas3 # the one env that covers ANSI + Connect ; tox -e typecheck # mypy (not in envlist; not run by CI) ; tox -e py312-spark4-pandas3 -- tests/test_pandas.py # forward to pytest ; +; Envs outside envlist still work: any py{310,311,312,313}-spark4-pandas{2,3} +; or py{310,311}-spark35-pandas{2,3} combination can be run with -e. +; ; NOTE: ``{posargs}`` replaces the default test paths in *every* command of an ; env, including the Spark Connect ones, so passing a path runs that path under ; each pytest config rather than appending a fifth invocation. @@ -45,11 +60,9 @@ requires = envlist = lint py{310,311,312,313}-nospark - py310-spark35-pandas2 - py311-spark35-pandas{2,3} - py311-spark35-connect + py311-spark35-pandas2 py310-spark4-pandas2 - py{311,312,313}-spark4-pandas{2,3} + py312-spark4-pandas3 [testenv] description = Run the datacompy test suite for {envname} @@ -125,9 +138,13 @@ commands = {envbindir}/python -m pytest --cov=datacompy --cov-report=term-missing {posargs} ; Spark 3.5 track, mirroring ``test-with-spark-3-install``. pyspark is pinned to -; 3.5.8 and pandas to the two versions CI exercises. openjdk 17 comes from -; conda-forge. Spark Connect is NOT exercised here -- see -; ``py311-spark35-connect`` below for that. +; 3.5.8; openjdk 17 comes from conda-forge. Only ``py311-spark35-pandas2`` is in +; envlist -- Spark 3.5 is the legacy path and one job is enough to prove it +; still works; the other combinations remain available via -e. +; +; ANSI mode is not run here: it is a spark-side behaviour and is covered once, +; on py312-spark4-pandas3. Spark Connect is not run here either -- see +; ``py311-spark35-connect`` below. [testenv:py{310,311}-spark35-pandas{2,3}] extras = qa @@ -141,7 +158,6 @@ deps = pandas3: pandas==3.0.3 commands = {envbindir}/python -m pytest --cov=datacompy --cov-report=term-missing {posargs} - {envbindir}/python -m pytest -c pytest-ansi.ini --cov=datacompy --cov-report=term-missing --cov-append {posargs} ; Spark 3.5 + Spark Connect. Experimental: Spark 3.5 does not bundle the ; Connect server jar in the pyspark wheel (4.x does), so we ask the JVM to @@ -168,16 +184,36 @@ commands = {envbindir}/python -m pytest -c pytest-connect.ini {posargs:tests/test_spark.py tests/comparator/} {envbindir}/python -m pytest -m spark_connect {posargs:tests/test_spark_connect.py} -; Spark 4 track, mirroring ``test-with-spark-4-install``. Includes classic -; sessions (default + ANSI mode) and the two Spark Connect suites. Connect must -; run in its own pytest process because starting a local Connect server sets -; SPARK_LOCAL_REMOTE, after which every later +; Spark 4 track, mirroring ``test-with-spark-4-install``. A classic session +; only -- ANSI mode and Spark Connect are covered once, by the dedicated +; ``py312-spark4-pandas3`` section below. Any combination here can be run with +; -e; ``py310-spark4-pandas2`` is the one in envlist. +[testenv:py{310,311,312,313}-spark4-pandas{2,3}] +extras = + qa + tests + tests-spark +conda_deps = + openjdk>=17.0.8,<18 +deps = + pyspark[connect]==4.1.2 + pandas2: pandas==2.3.3 + pandas3: pandas==3.0.3 +commands = + {envbindir}/python -m pytest --cov=datacompy --cov-report=term-missing {posargs} + +; The one env that runs everything: classic session, ANSI mode, and both Spark +; Connect suites. An exact section name wins over the generative section above, +; so this replaces -- rather than extends -- that env's commands. +; +; Connect must run in its own pytest process because starting a local Connect +; server sets SPARK_LOCAL_REMOTE, after which every later ; ``SparkSession.builder.getOrCreate()`` returns the Connect session -- so a ; classic and a Connect session cannot coexist in one run. ; ; The Connect commands carry no ``--cov``: they re-run tests already measured by ; the two classic commands, and CI does not measure them either. -[testenv:py{310,311,312,313}-spark4-pandas{2,3}] +[testenv:py312-spark4-pandas3] extras = qa tests @@ -186,8 +222,7 @@ conda_deps = openjdk>=17.0.8,<18 deps = pyspark[connect]==4.1.2 - pandas2: pandas==2.3.3 - pandas3: pandas==3.0.3 + pandas==3.0.3 commands = {envbindir}/python -m pytest --cov=datacompy --cov-report=term-missing {posargs} {envbindir}/python -m pytest -c pytest-ansi.ini --cov=datacompy --cov-report=term-missing --cov-append {posargs} From e65b2be6359dc66450b22ddc63007da10e5bcee0 Mon Sep 17 00:00:00 2001 From: Faisal Dosani Date: Fri, 14 Aug 2026 12:40:51 -0300 Subject: [PATCH 6/6] fix: address review findings in Spark merge, type guards, tests, and CI --- .github/workflows/test-package.yml | 13 +++++- .gitignore | 1 - CLAUDE.md | 36 ++++++++-------- Makefile | 8 +++- datacompy/base.py | 12 ++++-- datacompy/comparator/utility.py | 30 ++++++++++++++ datacompy/spark.py | 57 ++++++++++++-------------- pytest-connect.ini | 10 +++-- tests/comparator/test_utility_spark.py | 29 +++++++++++++ tests/test_base.py | 12 ++++++ tests/test_spark.py | 28 +++++++++++++ tests/test_spark_connect.py | 38 +++++++++++++---- tox.ini | 32 ++++++++++----- 13 files changed, 225 insertions(+), 81 deletions(-) diff --git a/.github/workflows/test-package.yml b/.github/workflows/test-package.yml index f70dffea..c094e68a 100644 --- a/.github/workflows/test-package.yml +++ b/.github/workflows/test-package.yml @@ -11,8 +11,12 @@ # python 3.10 Spark job (3.10 cannot take pandas 3), pandas 3.0.3 on 3.12. # * spark 3.5 vs 4 is a real API split, so both appear. # * ANSI mode and Spark Connect are spark-side semantics, orthogonal to the -# python and pandas versions, so they run once on the 3.12 baseline rather -# than on every Spark job. +# python and pandas versions, so they run once per Spark job rather than on +# every python/pandas combination. ANSI is NOT orthogonal to the Spark +# version -- `spark.sql.ansi.enabled` defaults to false on 3.5 and true on +# 4.x, and the TRY_CAST / integer-cast code in datacompy/comparator/ exists +# to bridge exactly that -- so both Spark jobs run it. Spark Connect stays +# on Spark 4 only, because 3.5 does not bundle the Connect server jar. # # Known gap: no Spark job runs on python 3.11 or 3.13, so a Spark break # specific to those runtimes would not be caught. Widening is one entry in the @@ -76,6 +80,11 @@ jobs: - name: Test with pytest run: | python -m pytest --cov=datacompy --cov-report=xml --cov-report=term-missing + # ANSI defaults to false on Spark 3.5 and true on 4.x, so an ANSI-mode + # break that only reproduces on 3.5 needs this job to catch it. + - name: Test with pytest (ANSI mode) + run: | + python -m pytest -c pytest-ansi.ini --cov=datacompy --cov-report=xml --cov-report=term-missing test-with-spark-4-install: name: spark 4 (py${{ matrix.python-version }}, pandas ${{ matrix.pandas-version }}) diff --git a/.gitignore b/.gitignore index dd3929a8..4b072077 100644 --- a/.gitignore +++ b/.gitignore @@ -33,4 +33,3 @@ coverage.xml # benchmark datasets benchmarks/data/ - diff --git a/CLAUDE.md b/CLAUDE.md index 27664de4..f6be994b 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -19,7 +19,7 @@ pre-commit install ```bash pytest # all tests pytest tests/test_pandas.py # single backend -pytest tests/test_pandas.py::TestPandasCompare::test_method # single test +pytest tests/test_pandas.py::test_numeric_columns_equal_abs # single test pytest --cov=datacompy --cov-report=term-missing # with coverage pytest -c pytest-connect.ini tests/test_spark.py tests/comparator/ # existing Spark suite, against Spark Connect @@ -28,37 +28,39 @@ pytest -m spark_connect tests/test_spark_connect.py # Spark Conne The Makefile wraps each of these: `make test`, `test-ansi`, `test-connect`, `test-connect-regression`, `test-cov`, `test-all`, plus `test-no-snowflake` / `test-all-no-snowflake`. -There are three pytest configs, and `-c` **replaces** the config file rather than layering on it — anything a run needs (markers, `testpaths`, `spark_options`) must be present in the file it names: +There are three pytest configs, and `-c` **replaces** the config file rather than layering on it. Anything a run needs (markers, `testpaths`, `spark_options`) must be present in the file it names: -- `pytest.ini` — classic Spark session, ANSI off -- `pytest-ansi.ini` — same, `spark.sql.ansi.enabled=true` -- `pytest-connect.ini` — Spark Connect session (`spark_connect_url`); must not set `spark.master`, since PySpark rejects a session with both +- `pytest.ini`: classic Spark session, ANSI off +- `pytest-ansi.ini`: same, `spark.sql.ansi.enabled=true` +- `pytest-connect.ini`: Spark Connect session (`spark_connect_url`); must not set `spark.master`, since PySpark rejects a session with both -All three set `testpaths = tests`. Without it a bare `pytest` walks the whole repo, and a git worktree checked out under the repo root contributes a second `tests/conftest.py` that collides with the real one — pytest then aborts the entire run with `ImportPathMismatchError`. Keep PR-review worktrees outside the repo. +All three set `testpaths = tests`. Without it a bare `pytest` walks the whole repo, and a git worktree checked out under the repo root contributes a second `tests/conftest.py` that collides with the real one, so pytest aborts the entire run with `ImportPathMismatchError`. Keep PR-review worktrees outside the repo. + +**Coverage:** always target the top-level package (`--cov=datacompy`). Passing a dotted submodule such as `--cov=datacompy.cli` triggers a numpy double-import in some environments and fails ~100 otherwise-passing tests with a confusing `_NoValueType` `TypeError`. Spark tests require Java 17 and `pyspark` installed. Snowflake tests require a live Snowflake session; `--snowflake-session local` is *not* a substitute, as Snowpark's local testing mode is an emulator that most of the Snowflake suite fails against. `benchmarks/` holds pytest-benchmark suites (`benchmark.py`, `benchmark_spark.py`). They sit outside `testpaths`, so no ordinary run collects them, and they read parquet fixtures that `python benchmarks/generate_data.py` has to write first. Results are written up in `docs/source/benchmark.rst`. -The two Spark Connect commands must each run in their own pytest process, and are excluded from the default run via `addopts`. Starting a local Spark Connect server sets `SPARK_LOCAL_REMOTE`, after which every later `SparkSession.builder.getOrCreate()` in that process returns the Connect session — so a classic and a Connect session cannot coexist in one run. +The two Spark Connect commands must each run in their own pytest process, and are excluded from the default run via `addopts`. Starting a local Spark Connect server sets `SPARK_LOCAL_REMOTE`, after which every later `SparkSession.builder.getOrCreate()` in that process returns the Connect session, so a classic and a Connect session cannot coexist in one run. ### Local CI matrix (tox) -`tox.ini` mirrors `.github/workflows/test-package.yml` job for job, using `tox-conda` so Java and PySpark come from conda-forge rather than a system JDK. It needs `pip install "tox<4" tox-conda` — tox-conda 0.10.x only supports tox 3. +`tox.ini` mirrors `.github/workflows/test-package.yml` job for job, using `tox-conda` so Java and PySpark come from conda-forge rather than a system JDK. It needs `pip install "tox<4" tox-conda` (tox-conda 0.10.x only supports tox 3). ```bash tox # the whole envlist tox -e lint # ruff check + ruff format --check -tox -e py312-spark4-pandas3 # the one env covering ANSI + Spark Connect +tox -e py312-spark4-pandas3 # the one env covering Spark Connect tox -e typecheck # mypy (not in envlist; not run by CI) ``` -Both files deliberately cover the test axes independently rather than as a cross product, because the Spark jobs are the entire cost of a run (~20 min each) while the four no-Spark jobs finish in under a minute. Python breadth comes from the cheap no-Spark envs; ANSI mode and Spark Connect are Spark-side semantics and run once on the 3.12 baseline. Read `tox.ini`'s header before widening the matrix — it records the known gap and what closing it costs. +Both files deliberately cover the test axes independently rather than as a cross product, because the Spark jobs are the entire cost of a run (~20 min each) while the four no-Spark jobs finish in under a minute. Python breadth comes from the cheap no-Spark envs. ANSI mode runs once per Spark track (its default flips between 3.5 and 4.x, so the tracks are not interchangeable); Spark Connect runs on the Spark 4 baseline only, since 3.5 does not bundle the Connect server jar. Read `tox.ini`'s header before widening the matrix: it records the known gap and what closing it costs. Two traps that cost real time to rediscover, both documented in `tox.ini`: - **Never loosen the `openjdk>=17.0.8,<18` floor** in the Spark envs. conda-forge ships GraalVM builds of openjdk 17 below that floor; selecting one pulls in `graalpy-graalvm`, which silently replaces CPython with GraalPy, and pip then dies with a Truffle `Unable to load native posix support library` error. It only affects Python 3.10 envs, since GraalPy implements 3.10. -- **`{envpython}` substitutes to the string `None` in tox 3.28** — use `{envbindir}/python`. The absolute path also avoids tox 3's silent fallback to a `$PATH` interpreter when a command is missing from a freshly-created env, which otherwise runs the suite against the wrong Python and only warns. +- **`{envpython}` substitutes to the string `None` in tox 3.28**, so use `{envbindir}/python`. The absolute path also avoids tox 3's silent fallback to a `$PATH` interpreter when a command is missing from a freshly-created env, which otherwise runs the suite against the wrong Python and only warns. ### Linting & Formatting ```bash @@ -71,7 +73,7 @@ mypy . # type-check (strict mode) `pyproject.toml` uses selectors (`noqa-comments`, `rule-codes-in-selectors`) that only exist in recent ruff, hence the `ruff>=0.16` floor on the `qa` extra. An older ruff fails while *parsing* the config with `Unknown rule selector`, before linting anything. Note that ruff formats Python code blocks inside Markdown, so `CLAUDE.md` and other docs are in scope for `ruff format --check`. -`mypy` is **not** a pre-commit hook and is not run by CI — the hooks are ruff, ruff-format, trailing-whitespace, debug-statements, end-of-file-fixer, and pyproject-fmt. Run it explicitly or via `tox -e typecheck`. +`mypy` is **not** a pre-commit hook and is not run by CI. The hooks are ruff, ruff-format, trailing-whitespace, debug-statements, end-of-file-fixer, and pyproject-fmt. Run it explicitly or via `tox -e typecheck`. ### Documentation ```bash @@ -109,9 +111,9 @@ Beyond the report, each backend exposes `df1_unq_rows`, `df2_unq_rows`, and `int Each type has backend-specific implementations: `Pandas*Comparator`, `Polars*Comparator`, `Spark*Comparator`, `Snowflake*Comparator`. -**Dispatch protocol** — the column-compare helper in each backend module (`columns_equal` and friends) walks a comparator list in order and takes the first result that isn't `None`: +**Dispatch protocol**: the column-compare helper in each backend module (`columns_equal` and friends) walks a comparator list in order and takes the first result that isn't `None`: -- **`compare()` returning `None` means "not my type, try the next one."** This is how type dispatch happens — there is no `can_compare()`. A comparator that raises instead of returning `None` breaks the chain. +- **`compare()` returning `None` means "not my type, try the next one."** This is how type dispatch happens; there is no `can_compare()`. A comparator that raises instead of returning `None` breaks the chain. - Order matters. Each backend defines `__DEFAULT_COMPARATORS` (array-like → boolean → numeric → string) at module scope. - All four backends accept `custom_comparators=[...]` on the constructor; `_get_comparators()` puts them **before** the defaults, so a custom comparator can pre-empt a built-in for a column type. - The dispatch loop passes the built-ins their type-specific kwargs via `isinstance` branches (tolerances to numeric, `ignore_spaces`/`ignore_case` to string), and passes custom comparators **everything** as `**kwargs`. Custom comparators must therefore accept and ignore kwargs they don't use. @@ -127,13 +129,13 @@ psf = get_spark_functions(dataframe) # datacompy/comparator/*.py Window = get_spark_window(dataframe) ``` -Because there is no module-level binding, ruff's `F821` flags any call site that forgets the local. For the same reason, never import `pyspark.sql.connect.*` at module scope — that package requires the optional `grpcio` dependency, and `__init__.py`'s `except ImportError` would silently drop `SparkSQLCompare` from the package. Use `is_spark_connect_object()` instead. +Because there is no module-level binding, ruff's `F821` flags any call site that forgets the local. For the same reason, never import `pyspark.sql.connect.*` at module scope: that package requires the optional `grpcio` dependency, and `__init__.py`'s `except ImportError` would silently drop `SparkSQLCompare` from the package. Use `is_spark_connect_object()` instead, or `is_spark_connect_dataframe()` when the question is specifically "is this a Connect DataFrame" (type validation), since the broader helper also answers True for a `Column`, a `GroupedData` or a `SparkSession`. ### Reporting Reports use Jinja2 templates from `datacompy/templates/report_template.j2`. The `render()` function in `base.py` handles template resolution. Custom templates can be passed via `report(template_path=...)`. -`build_report_data()` (defined once on `BaseCompare`) is the structured counterpart to `report()`, returning a typed `ReportData` object for dashboards or JSON export without parsing the string report. `ReportData` and its member dataclasses (`RowSummary`, `ColumnSummary`, `ColumnComparison`, `MismatchStat`, `MismatchStats`, `UniqueRowsData`) live in `datacompy/report.py` and are re-exported from `datacompy/__init__.py` — unlike the backends, they are unconditional top-level exports. It is public API — changes to its shape are breaking. Snapshot tests in `tests/test_report_snapshots.py` compare rendered output against fixtures in `tests/snapshots/`, which are excluded from the trailing-whitespace and end-of-file pre-commit hooks because their exact bytes are the assertion. +`build_report_data()` (defined once on `BaseCompare`) is the structured counterpart to `report()`, returning a typed `ReportData` object for dashboards or JSON export without parsing the string report. `ReportData` and its member dataclasses (`RowSummary`, `ColumnSummary`, `ColumnComparison`, `MismatchStat`, `MismatchStats`, `UniqueRowsData`) live in `datacompy/report.py` and are re-exported from `datacompy/__init__.py`; unlike the backends, they are unconditional top-level exports. It is public API, so changes to its shape are breaking. Snapshot tests in `tests/test_report_snapshots.py` compare rendered output against fixtures in `tests/snapshots/`, which are excluded from the trailing-whitespace and end-of-file pre-commit hooks because their exact bytes are the assertion. ### Tolerance Handling @@ -169,6 +171,6 @@ Tolerances (`abs_tol`, `rel_tol`) can be a single float (applied globally) or a ## Branching -- `main` is the default branch and the target for all active development (this changed with the v1 release — `develop` still exists and CI still builds it, but it is no longer where work lands) +- `main` is the default branch and the target for all active development (this changed with the v1 release; `develop` still exists and CI still builds it, but it is no longer where work lands) - `support/0.19.x` is archived: critical security fixes only, best-effort, no features or regular maintenance - CI runs on pushes and PRs to `develop`, `main`, `release/*`, `release-*`, and `support/*` diff --git a/Makefile b/Makefile index a7e61ed9..8fe81f52 100644 --- a/Makefile +++ b/Makefile @@ -40,10 +40,14 @@ test-all: test test-ansi test-connect test-connect-regression # Everything except the Snowflake suites, which error without a live session. # `--snowflake-session local` is not an alternative here: Snowpark's local # testing mode is an emulator and most of the Snowflake suite fails against it. -test-no-snowflake: PYTEST_ARGS += -k "not snowflake" +# +# `override` is required: a command-line assignment (`make PYTEST_ARGS=-x +# test-no-snowflake`) beats a plain target-specific one, which would silently +# drop the deselection and run the Snowflake suites anyway. +test-no-snowflake: override PYTEST_ARGS += -k "not snowflake" test-no-snowflake: test -test-all-no-snowflake: PYTEST_ARGS += -k "not snowflake" +test-all-no-snowflake: override PYTEST_ARGS += -k "not snowflake" test-all-no-snowflake: test-all sphinx: diff --git a/datacompy/base.py b/datacompy/base.py index 1fa7020f..b97bd536 100644 --- a/datacompy/base.py +++ b/datacompy/base.py @@ -716,21 +716,25 @@ def df_to_str(df: Any, sample_count: int | None = None, on_index: bool = False) # ``hasattr(df, "to_string")`` is True for it and it would otherwise take # the pandas branch. Nothing else is caught here -- pandas has no # ``toPandas`` and Polars exposes ``to_pandas``, not ``toPandas``. - if hasattr(df, "toPandas"): + # + # Every branch tests ``callable`` rather than mere existence, because the + # same attribute-synthesizing behaviour cuts both ways: ``df.toPandas`` on a + # pandas frame with a column of that name is a Series, not a method. + if callable(getattr(df, "toPandas", None)): if sample_count is not None: df = df.limit(sample_count) return df.toPandas().to_string() # Handle pandas DataFrame - if hasattr(df, "to_string"): + if callable(getattr(df, "to_string", None)): if sample_count is not None and len(df) > sample_count: df = df.head(sample_count) - if not on_index and hasattr(df, "reset_index"): + if not on_index and callable(getattr(df, "reset_index", None)): df = df.reset_index(drop=True) return df.to_string() # Handle Polars DataFrame - if hasattr(df, "to_pandas"): + if callable(getattr(df, "to_pandas", None)): if sample_count is not None and len(df) > sample_count: df = df.head(sample_count) return str(df) diff --git a/datacompy/comparator/utility.py b/datacompy/comparator/utility.py index 6fbe763a..03fb8968 100644 --- a/datacompy/comparator/utility.py +++ b/datacompy/comparator/utility.py @@ -34,6 +34,36 @@ _CONNECT_MODULE_PREFIX = "pyspark.sql.connect." +_CONNECT_DATAFRAME_MODULE = "pyspark.sql.connect.dataframe" + + +def is_spark_connect_dataframe(spark_object: Any) -> bool: + """Check whether an object is a Spark Connect ``DataFrame``. + + Narrower than :func:`is_spark_connect_object`, which answers "does this come + from the Spark Connect API at all" and is therefore True for a ``Column``, a + ``GroupedData`` or a ``SparkSession`` as well. Type validation needs the + narrow question, so that passing a ``Column`` where a ``DataFrame`` is + expected still raises a useful ``TypeError``. + + Like :func:`is_spark_connect_object` this compares module names instead of + using ``isinstance``, to avoid importing ``pyspark.sql.connect`` and its + optional ``grpcio`` dependency. + + Parameters + ---------- + spark_object : Any + Any object. + + Returns + ------- + bool + True if the object is a Spark Connect ``DataFrame``, False otherwise. + """ + return any( + klass.__module__ == _CONNECT_DATAFRAME_MODULE and klass.__name__ == "DataFrame" + for klass in type(spark_object).__mro__ + ) def is_spark_connect_object(spark_object: Any) -> bool: diff --git a/datacompy/spark.py b/datacompy/spark.py index 12539523..3e3deae3 100644 --- a/datacompy/spark.py +++ b/datacompy/spark.py @@ -23,8 +23,9 @@ import logging from copy import deepcopy +from functools import reduce +from operator import and_ from typing import Any, Dict, List, Tuple -from uuid import uuid4 import pandas as pd import pyspark.sql @@ -48,7 +49,7 @@ get_spark_column_dtypes, get_spark_functions, get_spark_window, - is_spark_connect_object, + is_spark_connect_dataframe, ) LOG = logging.getLogger(__name__) @@ -296,9 +297,12 @@ def _validate_dataframe( # so the isinstance check alone would be enough. On 3.5 it does not, # hence the second check -- which deliberately avoids importing # pyspark.sql.connect, since that package requires the optional grpcio - # dependency and raises ImportError without it. + # dependency and raises ImportError without it. It must stay specific to + # the DataFrame class: a check for "any Spark Connect object" would let a + # Column or a GroupedData through, and the failure would surface later as + # an unrelated error instead of the TypeError below. if not isinstance(dataframe, pyspark.sql.DataFrame) and not ( - is_spark_connect_object(dataframe) + is_spark_connect_dataframe(dataframe) ): raise TypeError( f"{index} must be a pyspark.sql.DataFrame or pyspark.sql.connect.dataframe.DataFrame (Spark 3.4.0 and above)" @@ -422,8 +426,6 @@ def _dataframe_merge(self, ignore_spaces: bool) -> None: df1 = df1.drop("__index") df2 = df2.drop("__index") - params = {"on": temp_join_columns} - if ignore_spaces: for column in self.join_columns: if ( @@ -458,32 +460,22 @@ def _dataframe_merge(self, ignore_spaces: bool) -> None: {c: f"{c}_{self.df2_name}" for c in temp_join_columns} ) - # NULL SAFE Outer join using ON. - # The view names are unique per merge. Fixed names would let a later - # comparison on the same session replace the views this plan refers to, - # and would clobber any user view called "df1" or "df2". Spark Connect - # resolves views lazily, so that surfaces as an unresolved column when - # the plan is finally executed rather than silently comparing the wrong - # data. - suffix = uuid4().hex - df1_view = f"datacompy_df1_{suffix}" - df2_view = f"datacompy_df2_{suffix}" - df1.createOrReplaceTempView(df1_view) - df2.createOrReplaceTempView(df2_view) - on = " and ".join( + # NULL SAFE Outer join, built with the DataFrame API rather than SQL so + # that no temp views have to be registered at all. Registering them is a + # trap either way: fixed names ("df1"/"df2") are replaced by the next + # comparison on the same session and clobber any user view of that name, + # while unique names accumulate for the life of the session, because a + # Spark Connect plan holds the SQL text and re-resolves the views on + # every action -- so they can never safely be dropped afterwards. + # eqNullSafe is the DataFrame-API spelling of <=>. + join_condition = reduce( + and_, [ - f"{df1_view}.`{c}_{self.df1_name}` <=> {df2_view}.`{c}_{self.df2_name}`" - for c in params["on"] - ] - ) - outer_join = self.spark_session.sql( - f""" - SELECT * FROM - {df1_view} FULL OUTER JOIN {df2_view} - ON - """ - + on + df1[f"{c}_{self.df1_name}"].eqNullSafe(df2[f"{c}_{self.df2_name}"]) + for c in temp_join_columns + ], ) + outer_join = df1.join(df2, on=join_condition, how="full_outer") outer_join = outer_join.withColumn("_merge", F.lit(None)) # initialize col @@ -860,8 +852,11 @@ def all_mismatch( if c.endswith("_match"): orig_col_name = c[:-6] + # SUM over zero rows is NULL, so default to 0 the way + # _intersect_compare does: with an empty intersection every + # count comes back None and the comparison would raise. if not ignore_matching_cols or ( - ignore_matching_cols and mismatch_counts[f"{c}_count"] > 0 + ignore_matching_cols and (mismatch_counts[f"{c}_count"] or 0) > 0 ): LOG.debug(f"Adding column {orig_col_name} to the result.") match_list.append(c) diff --git a/pytest-connect.ini b/pytest-connect.ini index 4446d4e8..30e0d3ee 100644 --- a/pytest-connect.ini +++ b/pytest-connect.ini @@ -12,13 +12,15 @@ ; uses Arrow on the wire. ; ; `spark.connect.grpc.binding.address` keeps the Connect server on loopback -- -; it listens on *:15002 otherwise. The driver side is pinned by SPARK_LOCAL_IP, -; which tests/conftest.py sets, because an ini can only set Spark configs and -; the driver bind address is not settable that way once the JVM is up. +; it listens on *:15002 otherwise. `spark.driver.bindAddress` pins the driver +; side. tests/test_spark_connect.py builds its own session and so has to repeat +; both settings; keep the two in step. ; ; Behind a corporate proxy you also need `no_proxy` to contain the literal host ; from the connection string -- gRPC honours `http_proxy` even for loopback and -; matches `no_proxy` on the exact host. tests/conftest.py handles that too. +; matches `no_proxy` on the exact host. Nothing in the test suite sets that for +; you: export it in the shell (`export no_proxy=$no_proxy,localhost,127.0.0.1`) +; or the run hangs and then fails with `failed to connect to all addresses`. ; ; The tests marked `spark_connect` are excluded: they build their own session ; and must run in a separate pytest process. diff --git a/tests/comparator/test_utility_spark.py b/tests/comparator/test_utility_spark.py index 9995006f..8511f224 100644 --- a/tests/comparator/test_utility_spark.py +++ b/tests/comparator/test_utility_spark.py @@ -24,6 +24,7 @@ get_spark_column_dtypes, get_spark_functions, get_spark_window, + is_spark_connect_dataframe, is_spark_connect_object, ) from pyspark.sql.types import ( @@ -95,6 +96,8 @@ def test_get_spark_column_dtypes_case_insensitive(spark_session): @pytest.mark.pyspark def test_is_spark_connect_object_connect_branch(): """A Spark Connect Column is built without any session or SparkContext.""" + pytest.importorskip("grpc") + import pandas as pd from pyspark.sql.connect import functions as connect_functions @@ -103,9 +106,29 @@ def test_is_spark_connect_object_connect_branch(): assert not is_spark_connect_object(object()) +@pytest.mark.pyspark +def test_is_spark_connect_dataframe_is_narrower_than_is_connect_object(): + """Only a Connect DataFrame passes, so type validation stays meaningful.""" + pytest.importorskip("grpc") + + import pandas as pd + from pyspark.sql.connect import functions as connect_functions + + column = connect_functions.col("value") + + # A Connect Column is a Spark Connect object, but it is not a DataFrame. + assert is_spark_connect_object(column) + assert not is_spark_connect_dataframe(column) + + assert not is_spark_connect_dataframe(pd.DataFrame({"a": [1]})) + assert not is_spark_connect_dataframe(object()) + + @pytest.mark.pyspark def test_get_spark_helpers_connect_branch(): """Spark Connect objects resolve to the Spark Connect implementations.""" + pytest.importorskip("grpc") + from pyspark.sql.connect import functions as connect_functions from pyspark.sql.connect.window import Window as ConnectWindow @@ -124,6 +147,8 @@ def test_get_spark_helpers_match_the_session(spark_session): flavour ``spark_session`` actually is. The negative assertions matter: without them an implementation that always returned one flavour would pass. """ + pytest.importorskip("grpc") + import pyspark.sql.functions as classic_functions from pyspark.sql import Window as ClassicWindow from pyspark.sql.connect import functions as connect_functions @@ -131,6 +156,10 @@ def test_get_spark_helpers_match_the_session(spark_session): df = spark_session.range(1) + # For a DataFrame the broad and the narrow check must agree, whichever + # lane this is running in. + assert is_spark_connect_dataframe(df) == is_spark_connect_object(df) + if is_spark_connect_object(df): assert get_spark_functions(df) is connect_functions assert get_spark_functions(df) is not classic_functions diff --git a/tests/test_base.py b/tests/test_base.py index cd314c35..fddf4676 100644 --- a/tests/test_base.py +++ b/tests/test_base.py @@ -94,6 +94,18 @@ def test_df_to_str_pandas_dataframe_without_index(): assert result == expected +def test_df_to_str_pandas_dataframe_with_spark_named_columns(): + """A pandas column named after a Spark method must not pick the Spark branch. + + ``df.toPandas`` on such a frame is a Series, not a method, so the branch is + selected on ``callable`` rather than on the attribute merely existing. + """ + df = pd.DataFrame({"toPandas": [1, 2], "reset_index": [3, 4]}) + result = df_to_str(df) + expected = df.to_string() + assert result == expected + + def test_df_to_str_other_type(): """Test with a non-DataFrame type.""" result = df_to_str([1, 2, 3]) diff --git a/tests/test_spark.py b/tests/test_spark.py index eb620361..3059258d 100644 --- a/tests/test_spark.py +++ b/tests/test_spark.py @@ -1277,6 +1277,34 @@ def test_all_mismatch_ignore_matching_cols_some_cols_matching_diff_rows(spark_se assert not ("dollar_amt_df1" in output and "dollar_amt_df1" in output) +def test_all_mismatch_ignore_matching_cols_empty_intersection(spark_session): + # No join keys in common, so intersect_rows carries the _match columns but + # no rows. SUM over zero rows is NULL, which used to raise a TypeError when + # the counts were compared against 0. + df1 = spark_session.createDataFrame([(1, "a"), (2, "b")], ["acct_id", "name"]) + df2 = spark_session.createDataFrame([(3, "a"), (4, "b")], ["acct_id", "name"]) + compare = SparkSQLCompare(spark_session, df1, df2, "acct_id") + + assert compare.intersect_rows.count() == 0 + # Falls back to the unique rows, as it does with ignore_matching_cols=False. + assert compare.all_mismatch(ignore_matching_cols=True).count() == 4 + + +def test_compare_registers_no_temp_views(spark_session): + # The merge used to register a temp view per dataframe per comparison and + # never drop them, so a long-lived session accumulated them without bound. + before = {t.name for t in spark_session.catalog.listTables()} + + df1 = spark_session.createDataFrame([(1, "a"), (2, "b")], ["acct_id", "name"]) + df2 = spark_session.createDataFrame([(1, "a"), (2, "z")], ["acct_id", "name"]) + for _ in range(3): + compare = SparkSQLCompare(spark_session, df1, df2, "acct_id") + assert not compare.matches() + + after = {t.name for t in spark_session.catalog.listTables()} + assert after == before + + def test_all_mismatch_ignore_matching_cols_some_cols_matching(spark_session): # Columns dollar_amt and name are matching data1 = """acct_id,dollar_amt,name,float_fld,date_fld diff --git a/tests/test_spark_connect.py b/tests/test_spark_connect.py index 14890306..1caf1ed0 100644 --- a/tests/test_spark_connect.py +++ b/tests/test_spark_connect.py @@ -53,15 +53,30 @@ def connect_session(): """ from pyspark.sql import SparkSession - try: - session = ( - SparkSession.builder.remote("local[2]") - .config("spark.sql.shuffle.partitions", "4") - .config("spark.sql.adaptive.enabled", "false") - .getOrCreate() - ) - except Exception as exc: # pragma: no cover - depends on the environment - pytest.skip(f"could not start a local Spark Connect server: {exc}") + # Deliberately not wrapped in try/except+skip. pyspark and grpc being absent + # is the only environment difference that may legitimately skip this file, + # and the importorskips above cover it. A server that fails to start is a + # failure: swallowing it would let the whole regression suite disappear + # while CI still reported the step green, which is precisely how the issue + # #535 class of bug got in. + # + # The binding addresses mirror pytest-connect.ini. Without them PySpark + # starts the bundled Connect server on *:15002, i.e. an unauthenticated + # endpoint on every interface of the machine running the tests. + # `spark.driver.host` has to be pinned alongside `spark.driver.bindAddress`: + # with only the latter, the driver still advertises the machine hostname, and + # the executor class loader fails to fetch isolated artifacts from it + # (RemoteClassLoaderError) once the Connect plugin turns artifact isolation + # on. + session = ( + SparkSession.builder.remote("local[2]") + .config("spark.connect.grpc.binding.address", "127.0.0.1") + .config("spark.driver.host", "localhost") + .config("spark.driver.bindAddress", "127.0.0.1") + .config("spark.sql.shuffle.partitions", "4") + .config("spark.sql.adaptive.enabled", "false") + .getOrCreate() + ) os.environ.pop("SPARK_CONNECT_MODE_ENABLED", None) yield session @@ -224,3 +239,8 @@ def test_validate_dataframe_accepts_connect_dataframe(connect_session): with pytest.raises(TypeError): SparkSQLCompare(connect_session, "not a dataframe", df, join_columns="acct_id") + + # A Column is a Spark Connect object but not a DataFrame, so the type check + # has to reject it rather than fail later on with something unrelated. + with pytest.raises(TypeError): + SparkSQLCompare(connect_session, df["acct_id"], df, join_columns="acct_id") diff --git a/tox.ini b/tox.ini index 81618d13..72ea828b 100644 --- a/tox.ini +++ b/tox.ini @@ -8,7 +8,7 @@ ; -------------------------- -------------------------------------------- ; lint-and-format lint ; test-basic-install py{310,311,312,313}-nospark -; test-with-spark-3-install py311-spark35-pandas2 +; test-with-spark-3-install py311-spark35-pandas2 <- also ANSI ; test-with-spark-4-install py310-spark4-pandas2 ; py312-spark4-pandas3 <- also ANSI + Connect ; @@ -22,8 +22,12 @@ ; python 3.10 env (CI cannot pair 3.10 with pandas 3), pandas3 on 3.12. ; * spark 3.5 vs 4 is a real API split, so both appear. ; * ANSI mode and Spark Connect are spark-side semantics, orthogonal to the -; python and pandas versions, so they run once on the 3.12 baseline rather -; than on every spark env. +; python and pandas versions, so they run once per spark track rather than +; on every python/pandas combination. ANSI is NOT orthogonal to the spark +; version -- it defaults to false on 3.5 and true on 4.x, which is what the +; TRY_CAST / integer-cast code in datacompy/comparator/ bridges -- so both +; tracks run it. Spark Connect stays on the spark 4 env, since 3.5 does not +; bundle the Connect server jar. ; ; KNOWN GAP: no spark env runs on python 3.11 or 3.13, so a Spark break ; specific to those runtimes would not be caught here. Widening is one line -- @@ -42,16 +46,20 @@ ; pip install "tox<4" tox-conda ; tox # the whole envlist ; tox -e lint # ruff check + ruff format --check -; tox -e py312-spark4-pandas3 # the one env that covers ANSI + Connect +; tox -e py312-spark4-pandas3 # the one env that covers Spark Connect ; tox -e typecheck # mypy (not in envlist; not run by CI) ; tox -e py312-spark4-pandas3 -- tests/test_pandas.py # forward to pytest ; ; Envs outside envlist still work: any py{310,311,312,313}-spark4-pandas{2,3} ; or py{310,311}-spark35-pandas{2,3} combination can be run with -e. ; -; NOTE: ``{posargs}`` replaces the default test paths in *every* command of an -; env, including the Spark Connect ones, so passing a path runs that path under -; each pytest config rather than appending a fifth invocation. +; NOTE: ``{posargs}`` replaces the default test paths in every command that +; takes them, so passing a path runs that path under each pytest config rather +; than appending another invocation. The ``-m spark_connect`` commands +; deliberately do NOT take ``{posargs}``: they are mark-filtered, so any other +; path collects zero tests and pytest exits 5, which tox 3 treats as a failed +; command -- the env would go red even though every test the user asked for +; passed. [tox] requires = @@ -142,8 +150,9 @@ commands = ; envlist -- Spark 3.5 is the legacy path and one job is enough to prove it ; still works; the other combinations remain available via -e. ; -; ANSI mode is not run here: it is a spark-side behaviour and is covered once, -; on py312-spark4-pandas3. Spark Connect is not run here either -- see +; ANSI mode runs here as well as on spark 4: it defaults to false on 3.5 and +; true on 4.x, so the two tracks exercise different code and one run does not +; stand in for the other. Spark Connect is not run here -- see ; ``py311-spark35-connect`` below. [testenv:py{310,311}-spark35-pandas{2,3}] extras = @@ -158,6 +167,7 @@ deps = pandas3: pandas==3.0.3 commands = {envbindir}/python -m pytest --cov=datacompy --cov-report=term-missing {posargs} + {envbindir}/python -m pytest -c pytest-ansi.ini --cov=datacompy --cov-report=term-missing --cov-append {posargs} ; Spark 3.5 + Spark Connect. Experimental: Spark 3.5 does not bundle the ; Connect server jar in the pyspark wheel (4.x does), so we ask the JVM to @@ -182,7 +192,7 @@ setenv = PYSPARK_SUBMIT_ARGS = --packages org.apache.spark:spark-connect_2.12:3.5.8 pyspark-shell commands = {envbindir}/python -m pytest -c pytest-connect.ini {posargs:tests/test_spark.py tests/comparator/} - {envbindir}/python -m pytest -m spark_connect {posargs:tests/test_spark_connect.py} + {envbindir}/python -m pytest -m spark_connect tests/test_spark_connect.py ; Spark 4 track, mirroring ``test-with-spark-4-install``. A classic session ; only -- ANSI mode and Spark Connect are covered once, by the dedicated @@ -227,4 +237,4 @@ commands = {envbindir}/python -m pytest --cov=datacompy --cov-report=term-missing {posargs} {envbindir}/python -m pytest -c pytest-ansi.ini --cov=datacompy --cov-report=term-missing --cov-append {posargs} {envbindir}/python -m pytest -c pytest-connect.ini {posargs:tests/test_spark.py tests/comparator/} - {envbindir}/python -m pytest -m spark_connect {posargs:tests/test_spark_connect.py} + {envbindir}/python -m pytest -m spark_connect tests/test_spark_connect.py