Repository: capitalone/datacompy
Pull request #426 (2025-07-08 01:11:07)

feat: add per-column tolerance support for comparison operations
Fixes #416

## Changes

- Added support for per-column absolute and relative tolerances in:
  - Pandas (core)
  - Polars
  - Spark SQL
  - Snowflake
- Enhanced report generation to include tolerance information
- Added comprehensive test coverage for new functionality
- Updated documentation for new tolerance features
- Removed legacy spark code and unused templates



Patch:
diff --git a/README.md b/README.md
index 9a28d52..e49ddeb 100644
--- a/README.md
+++ b/README.md
@@ -46,10 +46,10 @@ pip install datacompy[snowflake]
 
 ### LegacySparkCompare and SparkPandasCompare removal
 
-With version ``v0.17.0`` the ``LegacySparkCompare`` and ``SparkPandasCompare`` have been removed.
+Starting with v0.17.0, both `LegacySparkCompare` and `SparkPandasCompare` have been removed.
 
 
-#### Supported versions and dependncies
+#### Supported versions and dependencies
 
 Different versions of Spark, Pandas, and Python interact differently. Below is a matrix of what we test with.
 With the move to Pandas on Spark API and compatability issues with Pandas 2+ we will for the mean time note support Pandas 2
diff --git a/datacompy/base.py b/datacompy/base.py
index c374338..39495c2 100644
--- a/datacompy/base.py
+++ b/datacompy/base.py
@@ -24,7 +24,7 @@ two dataframes.
 import logging
 from abc import ABC, abstractmethod
 from pathlib import Path
-from typing import Any
+from typing import Any, Dict
 
 from jinja2 import Environment, FileSystemLoader, select_autoescape
 from ordered_set import OrderedSet
@@ -348,3 +348,98 @@ def df_to_str(df: Any, sample_count: int | None = None, on_index: bool = False)
 
     # Fallback to str() if we can't determine the type
     return str(df)
+
+
+def get_column_tolerance(column: str, tol_dict: Dict[str, float]) -> float:
+    """
+    Return the tolerance value for a given column from a dictionary of tolerances.
+
+    Parameters
+    ----------
+    column : str
+        The name of the column for which to retrieve the tolerance.
+    tol_dict : dict of str to float
+        Dictionary mapping column names to their tolerance values.
+        May contain a "default" key for columns not explicitly listed.
+
+    Returns
+    -------
+    float
+        The tolerance value for the specified column, or the "default" tolerance if the column is not found.
+        Returns 0 if neither the column nor "default" is present in the dictionary.
+    """
+    return tol_dict.get(column, tol_dict.get("default", 0.0))
+
+
+def _validate_tolerance_parameter(
+    param_value: float | Dict[str, float],
+    param_name: str,
+    case_mode: str = "lower",
+) -> Dict[str, float]:
+    """Validate and normalize tolerance parameter input.
+
+    Parameters
+    ----------
+    param_value : float or dict
+        The tolerance value to validate. Can be either a float or a dictionary mapping
+        column names to float values.
+    param_name : str
+        Name of the parameter being validated ('abs_tol' or 'rel_tol')
+    case_mode : str
+        How to handle column name case. Options are:
+        - "lower": convert to lowercase
+        - "upper": convert to uppercase
+        - "preserve": keep original case
+
+    Returns
+    -------
+    dict
+        Normalized dictionary of tolerance values
+
+    Raises
+    ------
+    TypeError
+        If param_value is not a float or dict
+    ValueError
+        If any tolerance values are not numeric or negative or if case_mode is invalid
+    """
+    if case_mode not in ["lower", "upper", "preserve"]:
+        raise ValueError("case_mode must be 'lower', 'upper', or 'preserve'")
+
+    # If float, convert to dict with default value
+    if isinstance(param_value, int | float):
+        if param_value < 0:
+            raise ValueError(f"{param_name} cannot be negative")
+        return {"default": float(param_value)}
+
+    # If dict, validate values and format
+    if isinstance(param_value, dict):
+        result = {}
+
+        # Convert all values to float and validate
+        for col, value in param_value.items():
+            if not isinstance(value, int | float):
+                raise ValueError(
+                    f"Value for column '{col}' in {param_name} must be numeric"
+                )
+            if value < 0:
+                raise ValueError(
+                    f"Value for column '{col}' in {param_name} cannot be negative"
+                )
+
+            # Handle column name case according to case_mode
+            col_key = str(col)
+            if case_mode == "lower":
+                col_key = col_key.lower()
+            elif case_mode == "upper":
+                col_key = col_key.upper()
+
+            result[col_key] = float(value)
+
+        # If no default provided, add 0.0
+        if "default" not in result:
+            result["default"] = 0.0
+
+        return result
+
+    raise TypeError(f"{param_name} must be a float or dictionary")
diff --git a/datacompy/core.py b/datacompy/core.py
index d61e51a..ac5026a 100644
--- a/datacompy/core.py
+++ b/datacompy/core.py
@@ -30,7 +30,9 @@ from ordered_set import OrderedSet
 
 from datacompy.base import (
     BaseCompare,
+    _validate_tolerance_parameter,
     df_to_str,
+    get_column_tolerance,
     render,
     save_html_report,
     temp_column_name,
@@ -59,10 +61,14 @@ class Compare(BaseCompare):
         If True, the index will be used to join the two dataframes.  If both
         ``join_columns`` and ``on_index`` are provided, an exception will be
         raised.
-    abs_tol : float, optional
-        Absolute tolerance between two values.
-    rel_tol : float, optional
-        Relative tolerance between two values.
+    abs_tol : float or dict, optional
+        Absolute tolerance between two values. Can be either a float value applied to all columns,
+        or a dictionary mapping column names to specific tolerance values. The special key "default"
+        in the dictionary specifies the tolerance for columns not explicitly listed.
+    rel_tol : float or dict, optional
+        Relative tolerance between two values. Can be either a float value applied to all columns,
+        or a dictionary mapping column names to specific tolerance values. The special key "default"
+        in the dictionary specifies the tolerance for columns not explicitly listed.
     df1_name : str, optional
         A string name for the first dataframe.  This allows the reporting to
         print out an actual name instead of "df1", and allows human users to
@@ -91,8 +97,8 @@ class Compare(BaseCompare):
         df2: pd.DataFrame,
         join_columns: List[str] | str | None = None,
         on_index: bool = False,
-        abs_tol: float = 0,
-        rel_tol: float = 0,
+        abs_tol: float | Dict[str, float] = 0,
+        rel_tol: float | Dict[str, float] = 0,
         df1_name: str = "df1",
         df2_name: str = "df2",
         ignore_spaces: bool = False,
@@ -100,6 +106,15 @@ class Compare(BaseCompare):
         cast_column_names_lower: bool = True,
     ) -> None:
         self.cast_column_names_lower = cast_column_names_lower
+
+        # Validate tolerance parameters first
+        self._abs_tol_dict = _validate_tolerance_parameter(
+            abs_tol, "abs_tol", "lower" if cast_column_names_lower else "preserve"
+        )
+        self._rel_tol_dict = _validate_tolerance_parameter(
+            rel_tol, "rel_tol", "lower" if cast_column_names_lower else "preserve"
+        )
+
         if on_index and join_columns is not None:
             raise Exception("Only provide on_index or join_columns")
         elif on_index:
@@ -369,12 +384,12 @@ class Compare(BaseCompare):
                     [
                         self.intersect_rows,
                         columns_equal(
-                            self.intersect_rows[col_1],
-                            self.intersect_rows[col_2],
-                            self.rel_tol,
-                            self.abs_tol,
-                            ignore_spaces,
-                            ignore_case,
+                            col_1=self.intersect_rows[col_1],
+                            col_2=self.intersect_rows[col_2],
+                            rel_tol=get_column_tolerance(column, self._rel_tol_dict),
+                            abs_tol=get_column_tolerance(column, self._abs_tol_dict),
+                            ignore_spaces=ignore_spaces,
+                            ignore_case=ignore_case,
                         ).to_frame(name=col_match),
                     ],
                     axis=1,
@@ -414,6 +429,8 @@ class Compare(BaseCompare):
                     ),
                     "max_diff": max_diff,
                     "null_diff": null_diff,
+                    "rel_tol": get_column_tolerance(column, self._rel_tol_dict),
+                    "abs_tol": get_column_tolerance(column, self._abs_tol_dict),
                 }
             )
 
@@ -589,12 +606,12 @@ class Compare(BaseCompare):
                 orig_col_name = col[:-6]
 
                 col_comparison = columns_equal(
-                    self.intersect_rows[orig_col_name + "_" + self.df1_name],
-                    self.intersect_rows[orig_col_name + "_" + self.df2_name],
-                    self.rel_tol,
-                    self.abs_tol,
-                    self.ignore_spaces,
-                    self.ignore_case,
+                    col_1=self.intersect_rows[orig_col_name + "_" + self.df1_name],
+                    col_2=self.intersect_rows[orig_col_name + "_" + self.df2_name],
+                    rel_tol=get_column_tolerance(orig_col_name, self._rel_tol_dict),
+                    abs_tol=get_column_tolerance(orig_col_name, self._abs_tol_dict),
+                    ignore_spaces=self.ignore_spaces,
+                    ignore_case=self.ignore_case,
                 )
 
                 if not ignore_matching_cols or (
@@ -717,6 +734,8 @@ class Compare(BaseCompare):
                         "unequal_cnt": column["unequal_cnt"],
                         "max_diff": column["max_diff"],
                         "null_diff": column["null_diff"],
+                        "rel_tol": column["rel_tol"],
+                        "abs_tol": column["abs_tol"],
                     }
                 )
                 if column["unequal_cnt"] > 0:
@@ -961,6 +980,7 @@ def columns_equal(
         col_2, ignore_spaces=ignore_spaces, ignore_case=ignore_case
     )
 
+    # Rest of comparison logic using rel_tol and abs_tol
     # short circuit if comparing mixed type columns. Check list/arrrays or just return false for everything else.
     if pd.api.types.infer_dtype(col_1).startswith("mixed") or pd.api.types.infer_dtype(
         col_2
diff --git a/datacompy/fugue.py b/datacompy/fugue.py
index 04e7223..5377f2e 100644
--- a/datacompy/fugue.py
+++ b/datacompy/fugue.py
@@ -625,6 +625,8 @@ def report(
                     "unequal_cnt": col["unequal_cnt"],
                     "max_diff": col["max_diff"],
                     "null_diff": col["null_diff"],
+                    "rel_tol": rel_tol,
+                    "abs_tol": abs_tol,
                 }
                 for col in column_stats
                 if not col["all_match"]
diff --git a/datacompy/polars.py b/datacompy/polars.py
index f59605b..504ed58 100644
--- a/datacompy/polars.py
+++ b/datacompy/polars.py
@@ -31,7 +31,9 @@ from ordered_set import OrderedSet
 
 from datacompy.base import (
     BaseCompare,
+    _validate_tolerance_parameter,
     df_to_str,
+    get_column_tolerance,
     render,
     save_html_report,
     temp_column_name,
@@ -59,10 +61,13 @@ class PolarsCompare(BaseCompare):
     join_columns : list or str
         Column(s) to join dataframes on.  If a string is passed in, that one
         column will be used.
-    abs_tol : float, optional
-        Absolute tolerance between two values.
-    rel_tol : float, optional
-        Relative tolerance between two values.
+    abs_tol : float or dict, optional
+        Absolute tolerance between two values. Can be either a float value applied to all columns,
+        or a dictionary mapping column names to specific tolerance values. The special key "default"
+        in the dictionary specifies the tolerance for columns not explicitly listed.
+    rel_tol : float or dict, optional
+        Relative tolerance between two values. Can be either a float value applied to all columns,
+        or a dictionary mapping column names to specific tolerance values. The special key "default"
     df1_name : str, optional
         A string name for the first dataframe.  This allows the reporting to
         print out an actual name instead of "df1", and allows human users to
@@ -90,8 +95,8 @@ class PolarsCompare(BaseCompare):
         df1: pl.DataFrame,
         df2: pl.DataFrame,
         join_columns: List[str] | str,
-        abs_tol: float = 0,
-        rel_tol: float = 0,
+        abs_tol: float | Dict[str, float] = 0,
+        rel_tol: float | Dict[str, float] = 0,
         df1_name: str = "df1",
         df2_name: str = "df2",
         ignore_spaces: bool = False,
@@ -100,6 +105,14 @@ class PolarsCompare(BaseCompare):
     ) -> None:
         self.cast_column_names_lower = cast_column_names_lower
 
+        # Validate tolerance parameters first
+        self._abs_tol_dict = _validate_tolerance_parameter(
+            abs_tol, "abs_tol", "lower" if cast_column_names_lower else "preserve"
+        )
+        self._rel_tol_dict = _validate_tolerance_parameter(
+            rel_tol, "rel_tol", "lower" if cast_column_names_lower else "preserve"
+        )
+
         if isinstance(join_columns, str):
             self.join_columns = [
                 str(join_columns).lower()
@@ -371,12 +384,12 @@ class PolarsCompare(BaseCompare):
                 col_match = column + "_match"
                 self.intersect_rows = self.intersect_rows.with_columns(
                     columns_equal(
-                        self.intersect_rows[col_1],
-                        self.intersect_rows[col_2],
-                        self.rel_tol,
-                        self.abs_tol,
-                        ignore_spaces,
-                        ignore_case,
+                        col_1=self.intersect_rows[col_1],
+                        col_2=self.intersect_rows[col_2],
+                        rel_tol=get_column_tolerance(column, self._rel_tol_dict),
+                        abs_tol=get_column_tolerance(column, self._abs_tol_dict),
+                        ignore_spaces=ignore_spaces,
+                        ignore_case=ignore_case,
                     ).alias(col_match)
                 )
                 match_cnt = self.intersect_rows[col_match].sum()
@@ -409,6 +422,8 @@ class PolarsCompare(BaseCompare):
                     ),
                     "max_diff": max_diff,
                     "null_diff": null_diff,
+                    "rel_tol": get_column_tolerance(column, self._rel_tol_dict),
+                    "abs_tol": get_column_tolerance(column, self._abs_tol_dict),
                 }
             )
 
@@ -588,12 +603,12 @@ class PolarsCompare(BaseCompare):
                 orig_col_name = col[:-6]
 
                 col_comparison = columns_equal(
-                    self.intersect_rows[orig_col_name + "_" + self.df1_name],
-                    self.intersect_rows[orig_col_name + "_" + self.df2_name],
-                    self.rel_tol,
-                    self.abs_tol,
-                    self.ignore_spaces,
-                    self.ignore_case,
+                    col_1=self.intersect_rows[orig_col_name + "_" + self.df1_name],
+                    col_2=self.intersect_rows[orig_col_name + "_" + self.df2_name],
+                    rel_tol=get_column_tolerance(orig_col_name, self._rel_tol_dict),
+                    abs_tol=get_column_tolerance(orig_col_name, self._abs_tol_dict),
+                    ignore_spaces=self.ignore_spaces,
+                    ignore_case=self.ignore_case,
                 )
 
                 if not ignore_matching_cols or (
@@ -717,6 +732,8 @@ class PolarsCompare(BaseCompare):
                         "unequal_cnt": column["unequal_cnt"],
                         "max_diff": column["max_diff"],
                         "null_diff": column["null_diff"],
+                        "rel_tol": column["rel_tol"],
+                        "abs_tol": column["abs_tol"],
                     }
                 )
                 if column["unequal_cnt"] > 0:
@@ -768,7 +785,9 @@ class PolarsCompare(BaseCompare):
             "df1_unique_rows": {
                 "has_rows": min_sample_count_df1 > 0,
                 "rows": df_to_str(
-                    self.df1_unq_rows[:, :min_column_count_df1],
+                    self.df1_unq_rows.select(
+                        self.df1_unq_rows.columns[:min_column_count_df1]
+                    ),
                     sample_count=min_sample_count_df1,
                 )
                 if self.df1_unq_rows.shape[0] > 0
@@ -780,7 +799,9 @@ class PolarsCompare(BaseCompare):
             "df2_unique_rows": {
                 "has_rows": min_sample_count_df2 > 0,
                 "rows": df_to_str(
-                    self.df2_unq_rows[:, :min_column_count_df2],
+                    self.df2_unq_rows.select(
+                        self.df2_unq_rows.columns[:min_column_count_df2]
+                    ),
                     sample_count=min_sample_count_df2,
                 )
                 if self.df2_unq_rows.shape[0] > 0
diff --git a/datacompy/snowflake.py b/datacompy/snowflake.py
index 41ed0c8..0b40bac 100644
--- a/datacompy/snowflake.py
+++ b/datacompy/snowflake.py
@@ -28,7 +28,14 @@ from typing import Any, Dict, List, Union, cast
 
 from ordered_set import OrderedSet
 
-from datacompy.base import BaseCompare, df_to_str, render, save_html_report
+from datacompy.base import (
+    BaseCompare,
+    _validate_tolerance_parameter,
+    df_to_str,
+    get_column_tolerance,
+    render,
+    save_html_report,
+)
 
 LOG = logging.getLogger(__name__)
 
@@ -98,10 +105,14 @@ class SnowflakeCompare(BaseCompare):
     join_columns : list or str, optional
         Column(s) to join dataframes on.  If a string is passed in, that one
         column will be used.
-    abs_tol : float, optional
-        Absolute tolerance between two values.
-    rel_tol : float, optional
-        Relative tolerance between two values.
+    abs_tol : float or dict, optional
+        Absolute tolerance between two values. Can be either a float value applied to all columns,
+        or a dictionary mapping column names to specific tolerance values. The special key "default"
+        in the dictionary specifies the tolerance for columns not explicitly listed.
+    rel_tol : float or dict, optional
+        Relative tolerance between two values. Can be either a float value applied to all columns,
+        or a dictionary mapping column names to specific tolerance values. The special key "default"
+        in the dictionary specifies the tolerance for columns not explicitly listed.
     df1_name : str, optional
         A string name for the first dataframe. If used alongside a snowflake table,
         overrides the default convention of naming the dataframe after the table.
@@ -125,12 +136,20 @@ class SnowflakeCompare(BaseCompare):
         df1: Union[str, "sp.DataFrame"],
         df2: Union[str, "sp.DataFrame"],
         join_columns: List[str] | str | None,
-        abs_tol: float = 0,
-        rel_tol: float = 0,
+        abs_tol: float | Dict[str, float] = 0,
+        rel_tol: float | Dict[str, float] = 0,
         df1_name: str | None = None,
         df2_name: str | None = None,
         ignore_spaces: bool = False,
     ) -> None:
+        # Validate tolerance parameters first
+        self._abs_tol_dict = _validate_tolerance_parameter(
+            abs_tol, "abs_tol", case_mode="upper"
+        )
+        self._rel_tol_dict = _validate_tolerance_parameter(
+            rel_tol, "rel_tol", case_mode="upper"
+        )
+
         if join_columns is None:
             errmsg = "join_columns cannot be None"
             raise ValueError(errmsg)
@@ -426,13 +445,13 @@ class SnowflakeCompare(BaseCompare):
                 col_2 = col + "_" + self.df2_name
                 col_match = col + "_MATCH"
                 self.intersect_rows = columns_equal(
-                    self.intersect_rows,
-                    col_1,
-                    col_2,
-                    col_match,
-                    self.rel_tol,
-                    self.abs_tol,
-                    ignore_spaces,
+                    dataframe=self.intersect_rows,
+                    col_1=col_1,
+                    col_2=col_2,
+                    col_match=col_match,
+                    rel_tol=get_column_tolerance(col, self._rel_tol_dict),
+                    abs_tol=get_column_tolerance(col, self._abs_tol_dict),
+                    ignore_spaces=ignore_spaces,
                 )
 
         with ThreadPoolExecutor() as executor:
@@ -519,6 +538,8 @@ class SnowflakeCompare(BaseCompare):
                 ),
                 "max_diff": max_diff,
                 "null_diff": null_diff,
+                "rel_tol": get_column_tolerance(column, self._rel_tol_dict),
+                "abs_tol": get_column_tolerance(column, self._abs_tol_dict),
             }
         )
 
@@ -721,13 +742,13 @@ class SnowflakeCompare(BaseCompare):
                 orig_col_name = c[:-6]
 
                 col_comparison = columns_equal(
-                    self.intersect_rows,
-                    orig_col_name + "_" + self.df1_name,
-                    orig_col_name + "_" + self.df2_name,
-                    c,
-                    self.rel_tol,
-                    self.abs_tol,
-                    self.ignore_spaces,
+                    dataframe=self.intersect_rows,
+                    col_1=orig_col_name + "_" + self.df1_name,
+                    col_2=orig_col_name + "_" + self.df2_name,
+                    col_match=c,
+                    rel_tol=get_column_tolerance(orig_col_name, self._rel_tol_dict),
+                    abs_tol=get_column_tolerance(orig_col_name, self._abs_tol_dict),
+                    ignore_spaces=self.ignore_spaces,
                 )
 
                 if not ignore_matching_cols or (
@@ -863,6 +884,8 @@ class SnowflakeCompare(BaseCompare):
                         "unequal_cnt": column["unequal_cnt"],
                         "max_diff": column["max_diff"],
                         "null_diff": column["null_diff"],
+                        "rel_tol": column["rel_tol"],
+                        "abs_tol": column["abs_tol"],
                     }
                 )
                 if column["unequal_cnt"] > 0:
diff --git a/datacompy/spark/sql.py b/datacompy/spark/sql.py
index 9c2b02f..6d66681 100644
--- a/datacompy/spark/sql.py
+++ b/datacompy/spark/sql.py
@@ -23,14 +23,16 @@ two dataframes.
 
 import logging
 from copy import deepcopy
-from typing import Any, List, Tuple
+from typing import Any, Dict, List, Tuple
 
 import pandas as pd
 from ordered_set import OrderedSet
 
 from datacompy.base import (
     BaseCompare,
+    _validate_tolerance_parameter,
     df_to_str,
+    get_column_tolerance,
     render,
     save_html_report,
     temp_column_name,
@@ -104,10 +106,14 @@ class SparkSQLCompare(BaseCompare):
     join_columns : list or str, optional
         Column(s) to join dataframes on.  If a string is passed in, that one
         column will be used.
-    abs_tol : float, optional
-        Absolute tolerance between two values.
-    rel_tol : float, optional
-        Relative tolerance between two values.
+    abs_tol : float or dict, optional
+        Absolute tolerance between two values. Can be either a float value applied to all columns,
+        or a dictionary mapping column names to specific tolerance values. The special key "default"
+        in the dictionary specifies the tolerance for columns not explicitly listed.
+    rel_tol : float or dict, optional
+        Relative tolerance between two values. Can be either a float value applied to all columns,
+        or a dictionary mapping column names to specific tolerance values. The special key "default"
+        in the dictionary specifies the tolerance for columns not explicitly listed.
     df1_name : str, optional
         A string name for the first dataframe.  This allows the reporting to
         print out an actual name instead of "df1", and allows human users to
@@ -138,8 +144,8 @@ class SparkSQLCompare(BaseCompare):
         df1: "pyspark.sql.DataFrame",
         df2: "pyspark.sql.DataFrame",
         join_columns: List[str] | str,
-        abs_tol: float = 0,
-        rel_tol: float = 0,
+        abs_tol: float | Dict[str, float] = 0,
+        rel_tol: float | Dict[str, float] = 0,
         df1_name: str = "df1",
         df2_name: str = "df2",
         ignore_spaces: bool = False,
@@ -147,6 +153,15 @@ class SparkSQLCompare(BaseCompare):
         cast_column_names_lower: bool = True,
     ) -> None:
         self.cast_column_names_lower = cast_column_names_lower
+
+        # Validate tolerance parameters first
+        self._abs_tol_dict = _validate_tolerance_parameter(
+            abs_tol, "abs_tol", "lower" if cast_column_names_lower else "preserve"
+        )
+        self._rel_tol_dict = _validate_tolerance_parameter(
+            rel_tol, "rel_tol", "lower" if cast_column_names_lower else "preserve"
+        )
+
         if isinstance(join_columns, str | int | float):
             self.join_columns = [
                 (
@@ -484,14 +499,14 @@ class SparkSQLCompare(BaseCompare):
                 col_2 = column + "_" + self.df2_name
                 col_match = column + "_match"
                 self.intersect_rows = columns_equal(
-                    self.intersect_rows,
-                    col_1,
-                    col_2,
-                    col_match,
-                    self.rel_tol,
-                    self.abs_tol,
-                    ignore_spaces,
-                    ignore_case,
+                    dataframe=self.intersect_rows,
+                    col_1=col_1,
+                    col_2=col_2,
+                    col_match=col_match,
+                    rel_tol=get_column_tolerance(column, self._rel_tol_dict),
+                    abs_tol=get_column_tolerance(column, self._abs_tol_dict),
+                    ignore_spaces=ignore_spaces,
+                    ignore_case=ignore_case,
                 )
                 match_cnt = (
                     self.intersect_rows.select(col_match)
@@ -526,6 +541,8 @@ class SparkSQLCompare(BaseCompare):
                     ),
                     "max_diff": max_diff,
                     "null_diff": null_diff,
+                    "rel_tol": get_column_tolerance(column, self._rel_tol_dict),
+                    "abs_tol": get_column_tolerance(column, self._abs_tol_dict),
                 }
             )
 
@@ -718,14 +735,14 @@ class SparkSQLCompare(BaseCompare):
                 orig_col_name = c[:-6]
 
                 col_comparison = columns_equal(
-                    self.intersect_rows,
-                    orig_col_name + "_" + self.df1_name,
-                    orig_col_name + "_" + self.df2_name,
-                    c,
-                    self.rel_tol,
-                    self.abs_tol,
-                    self.ignore_spaces,
-                    self.ignore_case,
+                    dataframe=self.intersect_rows,
+                    col_1=orig_col_name + "_" + self.df1_name,
+                    col_2=orig_col_name + "_" + self.df2_name,
+                    col_match=c,
+                    rel_tol=get_column_tolerance(orig_col_name, self._rel_tol_dict),
+                    abs_tol=get_column_tolerance(orig_col_name, self._abs_tol_dict),
+                    ignore_spaces=self.ignore_spaces,
+                    ignore_case=self.ignore_case,
                 )
 
                 if not ignore_matching_cols or (
@@ -860,6 +877,8 @@ class SparkSQLCompare(BaseCompare):
                         "unequal_cnt": column["unequal_cnt"],
                         "max_diff": column["max_diff"],
                         "null_diff": column["null_diff"],
+                        "rel_tol": column["rel_tol"],
+                        "abs_tol": column["abs_tol"],
                     }
                 )
                 if column["unequal_cnt"] > 0:
@@ -992,7 +1011,7 @@ class SparkSQLCompare(BaseCompare):
         df2_count = self.df2.count()
 
         # Prepare template data
-        template_data: dict[str, Any] = {
+        template_data: Dict[str, Any] = {
             **self._get_column_summary(),
             **self._get_row_summary(),
             **self._get_column_comparison(),
diff --git a/datacompy/templates/report_template.j2 b/datacompy/templates/report_template.j2
index 58913bc..f002098 100644
--- a/datacompy/templates/report_template.j2
+++ b/datacompy/templates/report_template.j2
@@ -22,8 +22,8 @@ Row Summary
 
 Matched on: {{ row_summary.match_columns }}
 Any duplicates on match values: {{ row_summary.has_duplicates }}
-Absolute Tolerance: {{ row_summary.abs_tol }}
-Relative Tolerance: {{ row_summary.rel_tol }}
+Default Absolute Tolerance: {{ row_summary.abs_tol }}
+Default Relative Tolerance: {{ row_summary.rel_tol }}
 Number of rows in common: {{ "{:,}".format(row_summary.common_rows) }}
 Number of rows in {{ row_summary.df1_name }} but not in {{ row_summary.df2_name }}: {{ "{:,}".format(row_summary.df1_unique) }}
 Number of rows in {{ row_summary.df2_name }} but not in {{ row_summary.df1_name }}: {{ "{:,}".format(row_summary.df2_unique) }}
@@ -47,17 +47,21 @@ Columns with Unequal Values or Types
                  mismatch_stats.df2_name + ' dtype',
                  '# Unequal',
                  'Max Diff',
-                 '# Null Diff'] %}
-{{ '%-20s %-15s %-15s %10s %10s %12s'|format(headers[0], headers[1], headers[2], headers[3], headers[4], headers[5]) }}
-{{ '-' * 20 }} {{ '-' * 15 }} {{ '-' * 15 }} {{ '-' * 10 }} {{ '-' * 10 }} {{ '-' * 12 }}
+                 '# Null Diff',
+                 'Rel Tol',
+                 'Abs Tol'] %}
+{{ '%-20s %-15s %-15s %10s %10s %12s %10s %10s'|format(headers[0], headers[1], headers[2], headers[3], headers[4], headers[5], headers[6], headers[7]) }}
+{{ '-' * 20 }} {{ '-' * 15 }} {{ '-' * 15 }} {{ '-' * 10 }} {{ '-' * 10 }} {{ '-' * 12 }} {{ '-' * 10 }} {{ '-' * 10 }}
 {% for stat in mismatch_stats.stats %}
-{{ '%-20s %-15s %-15s %10d %10.4f %12d'|format(
+{{ '%-20s %-15s %-15s %10d %10.4f %12d %10.4f %10.4f'|format(
     stat.column[:18] + ('...' if stat.column|length > 18 else ''),
     stat.dtype1,
     stat.dtype2,
     stat.unequal_cnt,
     stat.max_diff,
-    stat.null_diff) }}
+    stat.null_diff,
+    stat.rel_tol,
+    stat.abs_tol) }}
 {% endfor %}
 
 {% if mismatch_stats.has_samples %}
diff --git a/pyproject.toml b/pyproject.toml
index 621b517..5f939aa 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -46,7 +46,7 @@ zip-safe = false
 include-package-data = true
 
 [tool.setuptools.package-data]
-"*" = ["templates/*.txt"]
+"*" = ["templates/*.j2", "templates/*.txt"]
 
 [tool.setuptools.dynamic]
 version = {attr = "datacompy.__version__"}


Test patch:
diff --git a/tests/test_base.py b/tests/test_base.py
index a3ab3a7..b47e645 100644
--- a/tests/test_base.py
+++ b/tests/test_base.py
@@ -11,10 +11,14 @@ import pytest
 # Add parent directory to path to allow importing from datacompy
 sys.path.insert(0, str(Path(__file__).parent.parent))
 
+from typing import Dict
+
 from datacompy.base import (
     BaseCompare,
     _resolve_template_path,
+    _validate_tolerance_parameter,
     df_to_str,
+    get_column_tolerance,
     render,
     save_html_report,
     temp_column_name,
@@ -351,3 +355,77 @@ def test_resolve_with_subdirectories(tmp_path, monkeypatch):
 
     assert template_dir == str(templates_dir)
     assert resolved_name == "subdir/test_template.j2"
+
+
+def test_validate_tolerance_float() -> None:
+    """Test validation of float tolerance values."""
+    assert _validate_tolerance_parameter(0.1, "abs_tol") == {"default": 0.1}
+    assert _validate_tolerance_parameter(0, "abs_tol") == {"default": 0.0}
+    with pytest.raises(ValueError, match="abs_tol cannot be negative"):
+        _validate_tolerance_parameter(-0.1, "abs_tol")
+
+
+def test_validate_tolerance_dict() -> None:
+    """Test validation of dictionary tolerance values."""
+    tol_dict: Dict[str, float] = {"col1": 0.1, "col2": 0.2, "default": 0.05}
+    result = _validate_tolerance_parameter(tol_dict, "abs_tol")
+    assert result == {"col1": 0.1, "col2": 0.2, "default": 0.05}
+
+    # Test dictionary without default value
+    tol_dict = {"col1": 0.1, "col2": 0.2}
+    result = _validate_tolerance_parameter(tol_dict, "abs_tol")
+    assert result == {"col1": 0.1, "col2": 0.2, "default": 0.0}
+
+    # Test invalid values
+    with pytest.raises(ValueError, match="must be numeric"):
+        _validate_tolerance_parameter({"col1": "invalid"}, "abs_tol")  # type: ignore
+    with pytest.raises(ValueError, match="cannot be negative"):
+        _validate_tolerance_parameter({"col1": -0.1}, "abs_tol")
+
+
+def test_case_sensitivity() -> None:
+    """Test case sensitivity handling."""
+    tol_dict = {"COL1": 0.1, "Col2": 0.2}
+
+    # Test with case sensitivity lower
+    result = _validate_tolerance_parameter(tol_dict, "abs_tol", case_mode="lower")
+    assert result == {"col1": 0.1, "col2": 0.2, "default": 0.0}
+
+    # Test with case sensitivity disabled
+    result = _validate_tolerance_parameter(tol_dict, "abs_tol", case_mode="preserve")
+    assert result == {"COL1": 0.1, "Col2": 0.2, "default": 0.0}
+
+    # Test with case sensitivity upper
+    result = _validate_tolerance_parameter(tol_dict, "abs_tol", case_mode="upper")
+    assert result == {"COL1": 0.1, "COL2": 0.2, "default": 0.0}
+
+
+def test_get_column_tolerance_exact_match():
+    """Test get_column_tolerance returns the value for an exact column match."""
+    tol_dict = {"col1": 0.1, "col2": 0.2, "default": 0.05}
+    assert get_column_tolerance("col1", tol_dict) == 0.1
+    assert get_column_tolerance("col2", tol_dict) == 0.2
+
+
+def test_get_column_tolerance_default():
+    """Test get_column_tolerance returns the default value if column not found."""
+    tol_dict = {"col1": 0.1, "default": 0.05}
+    assert get_column_tolerance("colX", tol_dict) == 0.05
+
+
+def test_get_column_tolerance_no_default():
+    """Test get_column_tolerance returns 0.0 if column and default not found."""
+    tol_dict = {"col1": 0.1}
+    assert get_column_tolerance("colX", tol_dict) == 0.0
+
+
+def test_get_column_tolerance_empty_dict():
+    """Test get_column_tolerance returns 0.0 if tol_dict is empty."""
+    tol_dict = {}
+    assert get_column_tolerance("col1", tol_dict) == 0.0
+
+
+def test_get_column_tolerance_column_is_default():
+    """Test get_column_tolerance returns the value for 'default' if column is literally 'default'."""
+    tol_dict = {"default": 0.07}
+    assert get_column_tolerance("default", tol_dict) == 0.07
diff --git a/tests/test_core.py b/tests/test_core.py
index 44f15cb..77a72e8 100644
--- a/tests/test_core.py
+++ b/tests/test_core.py
@@ -1973,3 +1973,87 @@ def test_normalize_string_column(data, ignore_spaces, ignore_case, expected):
         data, ignore_spaces=ignore_spaces, ignore_case=ignore_case
     )
     assert_series_equal(result, expected, check_names=False)
+
+
+def test_per_column_tolerances() -> None:
+    """Test comparison with per-column tolerances."""
+    df1 = pd.DataFrame(
+        {"id": [1, 2, 3], "col1": [1.0, 2.0, 3.0], "col2": [1.0, 2.0, 3.0]}
+    )
+    df2 = pd.DataFrame(
+        {
+            "id": [1, 2, 3],
+            "col1": [1.1, 2.2, 3.3],  # Larger differences
+            "col2": [1.01, 2.01, 3.01],  # Smaller differences
+        }
+    )
+
+    compare = datacompy.core.Compare(
+        df1,
+        df2,
+        join_columns=["id"],
+        abs_tol={"col1": 0.5, "col2": 0.00001},  # col1 should match, col2 should not
+    )
+
+    col1_stats = next(stat for stat in compare.column_stats if stat["column"] == "col1")
+    col2_stats = next(stat for stat in compare.column_stats if stat["column"] == "col2")
+    assert col1_stats["unequal_cnt"] == 0
+    assert col2_stats["unequal_cnt"] > 0
+    assert compare._rel_tol_dict == {"default": 0.0}
+
+
+def test_default_tolerance() -> None:
+    """Test default tolerance behavior."""
+    df1 = pd.DataFrame({"id": [1, 2], "col1": [1.0, 2.0], "col2": [1.0, 2.0]})
+    df2 = pd.DataFrame({"id": [1, 2], "col1": [1.1, 2.1], "col2": [1.1, 2.1]})
+
+    compare = datacompy.core.Compare(
+        df1, df2, join_columns=["id"], abs_tol={"col1": 0.05, "default": 0.2}
+    )
+
+    col1_stats = next(stat for stat in compare.column_stats if stat["column"] == "col1")
+    col2_stats = next(stat for stat in compare.column_stats if stat["column"] == "col2")
+    assert col1_stats["unequal_cnt"] > 0  # col1 should not match (tolerance 0.05)
+    assert col2_stats["unequal_cnt"] == 0  # col2 should match (default tolerance 0.2)
+    assert compare._rel_tol_dict == {"default": 0.0}
+
+
+def test_mixed_tolerances() -> None:
+    """Test mixing absolute and relative tolerances."""
+    df1 = pd.DataFrame(
+        {
+            "id": [1, 2],
+            "small_vals": [
+                1.0,
+                2.0,
+            ],  # Small values where absolute tolerance matters more
+            "large_vals": [
+                1000.0,
+                2000.0,
+            ],  # Large values where relative tolerance matters more
+        }
+    )
+    df2 = pd.DataFrame(
+        {"id": [1, 2], "small_vals": [1.1, 2.1], "large_vals": [1001.0, 2002.0]}
+    )
+
+    compare = datacompy.core.Compare(
+        df1,
+        df2,
+        join_columns=["id"],
+        abs_tol={"small_vals": 0.2, "default": 0.0},
+        rel_tol={"large_vals": 0.001, "default": 0.0},
+    )
+
+    small_vals_stats = next(
+        stat for stat in compare.column_stats if stat["column"] == "small_vals"
+    )
+    large_vals_stats = next(
+        stat for stat in compare.column_stats if stat["column"] == "large_vals"
+    )
+    assert small_vals_stats["unequal_cnt"] == 0  # small_vals should match (abs_tol 0.2)
+    assert (
+        large_vals_stats["unequal_cnt"] == 0
+    )  # large_vals should match (rel_tol 0.001 = 0.1%)
+    assert compare._rel_tol_dict == {"large_vals": 0.001, "default": 0.0}
+    assert compare._abs_tol_dict == {"small_vals": 0.2, "default": 0.0}
diff --git a/tests/test_polars.py b/tests/test_polars.py
index d988896..9f068c1 100644
--- a/tests/test_polars.py
+++ b/tests/test_polars.py
@@ -1886,3 +1886,87 @@ def test_html_report_generation():
         # The result should be the same as the rendered HTML content
         assert isinstance(result, str)
         assert len(result) > 0
+
+
+def test_per_column_tolerances() -> None:
+    """Test comparison with per-column tolerances."""
+    df1 = pl.DataFrame(
+        {"id": [1, 2, 3], "col1": [1.0, 2.0, 3.0], "col2": [1.0, 2.0, 3.0]}
+    )
+    df2 = pl.DataFrame(
+        {
+            "id": [1, 2, 3],
+            "col1": [1.1, 2.2, 3.3],  # Larger differences
+            "col2": [1.01, 2.01, 3.01],  # Smaller differences
+        }
+    )
+
+    compare = PolarsCompare(
+        df1,
+        df2,
+        join_columns=["id"],
+        abs_tol={"col1": 0.5, "col2": 0.00001},  # col1 should match, col2 should not
+    )
+
+    col1_stats = next(stat for stat in compare.column_stats if stat["column"] == "col1")
+    col2_stats = next(stat for stat in compare.column_stats if stat["column"] == "col2")
+    assert col1_stats["unequal_cnt"] == 0
+    assert col2_stats["unequal_cnt"] > 0
+    assert compare._rel_tol_dict == {"default": 0.0}
+
+
+def test_default_tolerance() -> None:
+    """Test default tolerance behavior."""
+    df1 = pl.DataFrame({"id": [1, 2], "col1": [1.0, 2.0], "col2": [1.0, 2.0]})
+    df2 = pl.DataFrame({"id": [1, 2], "col1": [1.1, 2.1], "col2": [1.1, 2.1]})
+
+    compare = PolarsCompare(
+        df1, df2, join_columns=["id"], abs_tol={"col1": 0.05, "default": 0.2}
+    )
+
+    col1_stats = next(stat for stat in compare.column_stats if stat["column"] == "col1")
+    col2_stats = next(stat for stat in compare.column_stats if stat["column"] == "col2")
+    assert col1_stats["unequal_cnt"] > 0  # col1 should not match (tolerance 0.05)
+    assert col2_stats["unequal_cnt"] == 0  # col2 should match (default tolerance 0.2)
+    assert compare._rel_tol_dict == {"default": 0.0}
+
+
+def test_mixed_tolerances() -> None:
+    """Test mixing absolute and relative tolerances."""
+    df1 = pl.DataFrame(
+        {
+            "id": [1, 2],
+            "small_vals": [
+                1.0,
+                2.0,
+            ],  # Small values where absolute tolerance matters more
+            "large_vals": [
+                1000.0,
+                2000.0,
+            ],  # Large values where relative tolerance matters more
+        }
+    )
+    df2 = pl.DataFrame(
+        {"id": [1, 2], "small_vals": [1.1, 2.1], "large_vals": [1001.0, 2002.0]}
+    )
+
+    compare = PolarsCompare(
+        df1,
+        df2,
+        join_columns=["id"],
+        abs_tol={"small_vals": 0.2, "default": 0.0},
+        rel_tol={"large_vals": 0.001, "default": 0.0},
+    )
+
+    small_vals_stats = next(
+        stat for stat in compare.column_stats if stat["column"] == "small_vals"
+    )
+    large_vals_stats = next(
+        stat for stat in compare.column_stats if stat["column"] == "large_vals"
+    )
+    assert small_vals_stats["unequal_cnt"] == 0  # small_vals should match (abs_tol 0.2)
+    assert (
+        large_vals_stats["unequal_cnt"] == 0
+    )  # large_vals should match (rel_tol 0.001 = 0.1%)
+    assert compare._rel_tol_dict == {"large_vals": 0.001, "default": 0.0}
+    assert compare._abs_tol_dict == {"small_vals": 0.2, "default": 0.0}
diff --git a/tests/test_snowflake.py b/tests/test_snowflake.py
index d38c389..b9ef241 100644
--- a/tests/test_snowflake.py
+++ b/tests/test_snowflake.py
@@ -572,6 +572,92 @@ def test_10k_rows(snowpark_session):
     assert not compare_no_tol.intersect_rows_match()
 
 
+def test_10k_rows_abs_tol_per_column(snowpark_session):
+    rng = np.random.default_rng()
+    pdf = pd.DataFrame(rng.integers(0, 100, size=(10000, 2)), columns=["B", "C"])
+    pdf.reset_index(inplace=True)
+    pdf.columns = ["A", "B", "C"]
+    pdf2 = pdf.copy()
+    pdf2["B"] = pdf2["B"] + 0.1
+    df1 = snowpark_session.createDataFrame(pdf)
+    df2 = snowpark_session.createDataFrame(pdf2)
+    compare_tol = SnowflakeCompare(
+        snowpark_session, df1, df2, ["A"], abs_tol={"B": 0.2}
+    )
+    assert compare_tol.matches()
+    assert compare_tol.df1_unq_rows.count() == 0
+    assert compare_tol.df2_unq_rows.count() == 0
+    assert compare_tol.intersect_columns() == {"A", "B", "C"}
+    assert compare_tol.all_columns_match()
+    assert compare_tol.all_rows_overlap()
+    assert compare_tol.intersect_rows_match()
+
+
+def test_10k_rows_abs_tol_per_column_default(snowpark_session):
+    rng = np.random.default_rng()
+    pdf = pd.DataFrame(rng.integers(0, 100, size=(10000, 2)), columns=["B", "C"])
+    pdf.reset_index(inplace=True)
+    pdf.columns = ["A", "B", "C"]
+    pdf2 = pdf.copy()
+    pdf2["B"] = pdf2["B"] + 0.1
+    pdf2["C"] = pdf2["C"] + 0.3
+    df1 = snowpark_session.createDataFrame(pdf)
+    df2 = snowpark_session.createDataFrame(pdf2)
+    compare_tol = SnowflakeCompare(
+        snowpark_session, df1, df2, ["A"], abs_tol={"c": 0.0, "default": 0.2}
+    )
+    assert not compare_tol.matches()
+    assert compare_tol.df1_unq_rows.count() == 0
+    assert compare_tol.df2_unq_rows.count() == 0
+    assert compare_tol.intersect_columns() == {"A", "B", "C"}
+    assert compare_tol.all_columns_match()
+    assert compare_tol.all_rows_overlap()
+    assert not compare_tol.intersect_rows_match()
+
+
+def test_10k_rows_rel_tol_per_column(snowpark_session):
+    rng = np.random.default_rng()
+    pdf = pd.DataFrame(rng.integers(0, 100, size=(10000, 2)), columns=["B", "C"])
+    pdf.reset_index(inplace=True)
+    pdf.columns = ["A", "B", "C"]
+    pdf2 = pdf.copy()
+    pdf2["B"] = pdf2["B"] + 0.1
+    df1 = snowpark_session.createDataFrame(pdf)
+    df2 = snowpark_session.createDataFrame(pdf2)
+    compare_tol = SnowflakeCompare(
+        snowpark_session, df1, df2, ["A"], rel_tol={"B": 1.0}
+    )
+    assert compare_tol.matches()
+    assert compare_tol.df1_unq_rows.count() == 0
+    assert compare_tol.df2_unq_rows.count() == 0
+    assert compare_tol.intersect_columns() == {"A", "B", "C"}
+    assert compare_tol.all_columns_match()
+    assert compare_tol.all_rows_overlap()
+    assert compare_tol.intersect_rows_match()
+
+
+def test_10k_rows_rel_tol_per_column_default(snowpark_session):
+    rng = np.random.default_rng()
+    pdf = pd.DataFrame(rng.integers(0, 100, size=(10000, 2)), columns=["B", "C"])
+    pdf.reset_index(inplace=True)
+    pdf.columns = ["A", "B", "C"]
+    pdf2 = pdf.copy()
+    pdf2["B"] = pdf2["B"] + 0.1
+    pdf2["C"] = pdf2["C"] + 0.1
+    df1 = snowpark_session.createDataFrame(pdf)
+    df2 = snowpark_session.createDataFrame(pdf2)
+    compare_tol = SnowflakeCompare(
+        snowpark_session, df1, df2, ["A"], rel_tol={"c": 0.0, "default": 1}
+    )
+    assert not compare_tol.matches()
+    assert compare_tol.df1_unq_rows.count() == 0
+    assert compare_tol.df2_unq_rows.count() == 0
+    assert compare_tol.intersect_columns() == {"A", "B", "C"}
+    assert compare_tol.all_columns_match()
+    assert compare_tol.all_rows_overlap()
+    assert not compare_tol.intersect_rows_match()
+
+
 def test_subset(snowpark_session, caplog):
     caplog.set_level(logging.DEBUG)
     df1 = snowpark_session.createDataFrame(
@@ -1231,6 +1317,181 @@ def test_all_mismatch_ignore_matching_cols_no_cols_matching(snowpark_session):
     assert (~(output.DATE_FLD_DF1 != output.DATE_FLD_DF2)).values.sum() == 0
 
 
+def test_all_mismatch_ignore_matching_cols_no_cols_matching_abs_tol_float(
+    snowpark_session,
+):
+    data1 = """ACCT_ID,DOLLAR_AMT,NAME,FLOAT_FLD,DATE_FLD
+        10000001234,123.45,George Maharis,14530.1555,2017-01-01
+        10000001235,0.45,Michael Bluth,1,2017-01-01
+        10000001236,1345,George Bluth,,2017-01-01
+        10000001237,123456,Bob Loblaw,345.12,2017-01-01
+        10000001239,1.05,Lucille Bluth,,2017-01-01
+        10000001240,123.45,George Maharis,14530.1555,2017-01-02
+        """
+
+    data2 = """ACCT_ID,DOLLAR_AMT,NAME,FLOAT_FLD,DATE_FLD
+        10000001234,123.4,George Maharis,14530.1555,2017-01-01
+        10000001235,0.45,Michael Bluth,1.04,2017-01-01
+        10000001236,1345,George Bluth,1,
+        10000001237,123456,Robert Loblaw,345.12,
+        10000001238,1.05,Loose Seal Bluth,111,
+        10000001240,123.45,George Maharis,14530.1555,2017-01-02
+        """
+    df1 = snowpark_session.createDataFrame(pd.read_csv(StringIO(data1), sep=","))
+    df2 = snowpark_session.createDataFrame(pd.read_csv(StringIO(data2), sep=","))
+    compare = SnowflakeCompare(snowpark_session, df1, df2, "ACCT_ID", 0.05)
+
+    output = compare.all_mismatch().toPandas()
+    assert output.shape[0] == 2
+    assert output.shape[1] == 9
+
+    assert (output.NAME_DF1 != output.NAME_DF2).values.sum() == 1
+    assert (~(output.NAME_DF1 != output.NAME_DF2)).values.sum() == 1
+
+    assert (output.DOLLAR_AMT_DF1 != output.DOLLAR_AMT_DF2).values.sum() == 0
+    assert (~(output.DOLLAR_AMT_DF1 != output.DOLLAR_AMT_DF2)).values.sum() == 2
+
+    assert (output.FLOAT_FLD_DF1 != output.FLOAT_FLD_DF2).values.sum() == 1
+    assert (~(output.FLOAT_FLD_DF1 != output.FLOAT_FLD_DF2)).values.sum() == 1
+
+    assert (output.DATE_FLD_DF1 != output.DATE_FLD_DF2).values.sum() == 2
+    assert (~(output.DATE_FLD_DF1 != output.DATE_FLD_DF2)).values.sum() == 0
+
+
+def test_all_mismatch_ignore_matching_cols_no_cols_matching_abs_tol_dict(
+    snowpark_session,
+):
+    data1 = """ACCT_ID,DOLLAR_AMT,NAME,FLOAT_FLD,DATE_FLD
+        10000001234,123.45,George Maharis,14530.1555,2017-01-01
+        10000001235,0.45,Michael Bluth,1,2017-01-01
+        10000001236,1345,George Bluth,,2017-01-01
+        10000001237,123456,Bob Loblaw,345.12,2017-01-01
+        10000001239,1.05,Lucille Bluth,,2017-01-01
+        10000001240,123.45,George Maharis,14530.1555,2017-01-02
+        """
+
+    data2 = """ACCT_ID,DOLLAR_AMT,NAME,FLOAT_FLD,DATE_FLD
+        10000001234,123.4,George Maharis,14530.1555,2017-01-01
+        10000001235,0.45,Michael Bluth,1.05,2017-01-01
+        10000001236,1345,George Bluth,1,
+        10000001237,123456,Robert Loblaw,345.12,
+        10000001238,1.05,Loose Seal Bluth,111,
+        10000001240,123.45,George Maharis,14530.1555,2017-01-02
+        """
+    df1 = snowpark_session.createDataFrame(pd.read_csv(StringIO(data1), sep=","))
+    df2 = snowpark_session.createDataFrame(pd.read_csv(StringIO(data2), sep=","))
+    compare = SnowflakeCompare(
+        snowpark_session, df1, df2, "ACCT_ID", {"DOLLAR_AMT": 0.05}
+    )
+
+    output = compare.all_mismatch().toPandas()
+    assert output.shape[0] == 3
+    assert output.shape[1] == 9
+
+    assert (output.NAME_DF1 != output.NAME_DF2).values.sum() == 1
+    assert (~(output.NAME_DF1 != output.NAME_DF2)).values.sum() == 2
+
+    assert (output.DOLLAR_AMT_DF1 != output.DOLLAR_AMT_DF2).values.sum() == 0
+    assert (~(output.DOLLAR_AMT_DF1 != output.DOLLAR_AMT_DF2)).values.sum() == 3
+
+    assert (output.FLOAT_FLD_DF1 != output.FLOAT_FLD_DF2).values.sum() == 2
+    assert (~(output.FLOAT_FLD_DF1 != output.FLOAT_FLD_DF2)).values.sum() == 1
+
+    assert (output.DATE_FLD_DF1 != output.DATE_FLD_DF2).values.sum() == 2
+    assert (~(output.DATE_FLD_DF1 != output.DATE_FLD_DF2)).values.sum() == 1
+
+
+def test_all_mismatch_ignore_matching_cols_no_cols_matching_rel_tol_float(
+    snowpark_session,
+):
+    data1 = """ACCT_ID,DOLLAR_AMT,NAME,FLOAT_FLD,DATE_FLD
+        10000001234,123.45,George Maharis,14530.1555,2017-01-01
+        10000001235,0.45,Michael Bluth,1,2017-01-01
+        10000001236,1345,George Bluth,,2017-01-01
+        10000001237,123456,Bob Loblaw,345.12,2017-01-01
+        10000001239,1.05,Lucille Bluth,,2017-01-01
+        10000001240,123.45,George Maharis,14530.1555,2017-01-02
+        """
+
+    data2 = """ACCT_ID,DOLLAR_AMT,NAME,FLOAT_FLD,DATE_FLD
+        10000001234,123.4,George Maharis,14530.1555,2017-01-01
+        10000001235,0.45,Michael Bluth,1.04,2017-01-01
+        10000001236,1345,George Bluth,1,
+        10000001237,123456,Robert Loblaw,345.12,
+        10000001238,1.05,Loose Seal Bluth,111,
+        10000001240,123.45,George Maharis,14530.1555,2017-01-02
+        """
+    df1 = snowpark_session.createDataFrame(pd.read_csv(StringIO(data1), sep=","))
+    df2 = snowpark_session.createDataFrame(pd.read_csv(StringIO(data2), sep=","))
+    compare = SnowflakeCompare(snowpark_session, df1, df2, "ACCT_ID", rel_tol=0.1)
+
+    output = compare.all_mismatch().toPandas()
+    assert output.shape[0] == 2
+    assert output.shape[1] == 9
+
+    assert (output.NAME_DF1 != output.NAME_DF2).values.sum() == 1
+    assert (~(output.NAME_DF1 != output.NAME_DF2)).values.sum() == 1
+
+    assert (output.DOLLAR_AMT_DF1 != output.DOLLAR_AMT_DF2).values.sum() == 0
+    assert (~(output.DOLLAR_AMT_DF1 != output.DOLLAR_AMT_DF2)).values.sum() == 2
+
+    assert (output.FLOAT_FLD_DF1 != output.FLOAT_FLD_DF2).values.sum() == 1
+    assert (~(output.FLOAT_FLD_DF1 != output.FLOAT_FLD_DF2)).values.sum() == 1
+
+    assert (output.DATE_FLD_DF1 != output.DATE_FLD_DF2).values.sum() == 2
+    assert (~(output.DATE_FLD_DF1 != output.DATE_FLD_DF2)).values.sum() == 0
+
+
+def test_all_mismatch_ignore_matching_cols_no_cols_matching_rel_tol_dict(
+    snowpark_session,
+):
+    data1 = """ACCT_ID,DOLLAR_AMT,NAME,FLOAT_FLD,DATE_FLD
+        10000001234,123.45,George Maharis,14530.1555,2017-01-01
+        10000001235,0.45,Michael Bluth,1,2017-01-01
+        10000001236,1345,George Bluth,,2017-01-01
+        10000001237,123456,Bob Loblaw,345.12,2017-01-01
+        10000001239,1.05,Lucille Bluth,,2017-01-01
+        10000001240,123.45,George Maharis,14530.1555,2017-01-02
+        """
+
+    data2 = """ACCT_ID,DOLLAR_AMT,NAME,FLOAT_FLD,DATE_FLD
+        10000001234,123.4,George Maharis,14530.1555,2017-01-01
+        10000001235,0.45,Michael Bluth,1.05,2017-01-01
+        10000001236,1345,George Bluth,1,
+        10000001237,123456,Robert Loblaw,345.12,
+        10000001238,1.05,Loose Seal Bluth,111,
+        10000001240,123.45,George Maharis,14530.1555,2017-01-02
+        """
+    df1 = snowpark_session.createDataFrame(pd.read_csv(StringIO(data1), sep=","))
+    df2 = snowpark_session.createDataFrame(pd.read_csv(StringIO(data2), sep=","))
+    compare = SnowflakeCompare(
+        snowpark_session,
+        df1,
+        df2,
+        "ACCT_ID",
+        abs_tol={"FLOAT_FLD": 0.01},
+        rel_tol={"DOLLAR_AMT": 0.10, "FLOAT_FLD": 0.02},
+    )
+
+    output = compare.all_mismatch().toPandas()
+    r = compare.report()
+
+    assert output.shape[0] == 3
+    assert output.shape[1] == 9
+
+    assert (output.NAME_DF1 != output.NAME_DF2).values.sum() == 1
+    assert (~(output.NAME_DF1 != output.NAME_DF2)).values.sum() == 2
+
+    assert (output.DOLLAR_AMT_DF1 != output.DOLLAR_AMT_DF2).values.sum() == 0
+    assert (~(output.DOLLAR_AMT_DF1 != output.DOLLAR_AMT_DF2)).values.sum() == 3
+
+    assert (output.FLOAT_FLD_DF1 != output.FLOAT_FLD_DF2).values.sum() == 2
+    assert (~(output.FLOAT_FLD_DF1 != output.FLOAT_FLD_DF2)).values.sum() == 1
+
+    assert (output.DATE_FLD_DF1 != output.DATE_FLD_DF2).values.sum() == 2
+    assert (~(output.DATE_FLD_DF1 != output.DATE_FLD_DF2)).values.sum() == 1
+
+
 @pytest.mark.parametrize(
     "column, expected",
     [
diff --git a/tests/test_spark/test_sql_spark.py b/tests/test_spark/test_sql_spark.py
index becbeeb..6ca1a54 100644
--- a/tests/test_spark/test_sql_spark.py
+++ b/tests/test_spark/test_sql_spark.py
@@ -1691,7 +1691,7 @@ def test_nonexistent_template(spark_session):
     df2 = spark_session.createDataFrame([("a", 1), ("b", 3)], ["id", "value"])
     compare = SparkSQLCompare(spark_session, df1, df2, ["id"])
 
-    with pytest.raises(Exception) as exc_info:
+    with pytest.raises(FileNotFoundError) as exc_info:
         compare.report(template_path="nonexistent_template.j2")
     # Check that the error message is helpful
     assert "Template not found" in str(
@@ -1758,3 +1758,145 @@ def test_html_report_generation(mock_render, mock_save_html, spark_session):
         # Clean up the temporary file
         if os.path.exists(html_file):
             os.unlink(html_file)
+
+
+def test_10k_rows_default_tol(spark_session):
+    rng = np.random.default_rng()
+    pdf = pd.DataFrame(rng.integers(0, 100, size=(10000, 2)), columns=["B", "C"])
+    pdf.reset_index(inplace=True)
+    pdf.columns = ["a", "B", "C"]
+    pdf2 = pdf.copy()
+    pdf2["B"] = pdf2["B"] + 0.1
+    df1 = spark_session.createDataFrame(pdf)
+    df2 = spark_session.createDataFrame(pdf2)
+    compare_tol = SparkSQLCompare(spark_session, df1, df2, ["A"], abs_tol=0.2)
+    assert compare_tol.matches()
+    assert compare_tol.df1_unq_rows.count() == 0
+    assert compare_tol.df2_unq_rows.count() == 0
+    assert compare_tol.intersect_columns() == {
+        "a",
+        "b",
+        "c",
+    }  # cast_column_names_lower is True by default
+    assert compare_tol.all_columns_match()
+    assert compare_tol.all_rows_overlap()
+    assert compare_tol.intersect_rows_match()
+
+    compare_no_tol = SparkSQLCompare(spark_session, df1, df2, ["A"])
+    assert not compare_no_tol.matches()
+    assert compare_no_tol.df1_unq_rows.count() == 0
+    assert compare_no_tol.df2_unq_rows.count() == 0
+    assert compare_no_tol.intersect_columns() == {"a", "b", "c"}
+    assert compare_no_tol.all_columns_match()
+    assert compare_no_tol.all_rows_overlap()
+    assert not compare_no_tol.intersect_rows_match()
+
+
+def test_10k_rows_abs_tol_per_column(spark_session):
+    rng = np.random.default_rng()
+    pdf = pd.DataFrame(rng.integers(0, 100, size=(10000, 2)), columns=["B", "C"])
+    pdf.reset_index(inplace=True)
+    pdf.columns = ["A", "B", "C"]
+    pdf2 = pdf.copy()
+    pdf2["B"] = pdf2["B"] + 0.1
+    df1 = spark_session.createDataFrame(pdf)
+    df2 = spark_session.createDataFrame(pdf2)
+    compare_tol = SparkSQLCompare(spark_session, df1, df2, ["A"], abs_tol={"b": 0.2})
+    assert compare_tol.matches()
+    assert compare_tol.df1_unq_rows.count() == 0
+    assert compare_tol.df2_unq_rows.count() == 0
+    assert compare_tol.intersect_columns() == {
+        "a",
+        "b",
+        "c",
+    }  # cast_column_names_lower is True by default
+    assert compare_tol.all_columns_match()
+    assert compare_tol.all_rows_overlap()
+    assert compare_tol.intersect_rows_match()
+
+    compare_tol = SparkSQLCompare(
+        spark_session,
+        df1,
+        df2,
+        ["A"],
+        abs_tol={"b": 0.2},
+        cast_column_names_lower=False,  # b != B
+    )
+    assert not compare_tol.matches()
+    assert compare_tol.df1_unq_rows.count() == 0
+    assert compare_tol.df2_unq_rows.count() == 0
+    assert compare_tol.intersect_columns() == {
+        "A",
+        "B",
+        "C",
+    }  # cast_column_names_lower is False
+    assert compare_tol.all_columns_match()
+    assert compare_tol.all_rows_overlap()
+    assert not compare_tol.intersect_rows_match()
+
+
+def test_10k_rows_abs_tol_per_column_default(spark_session):
+    rng = np.random.default_rng()
+    pdf = pd.DataFrame(rng.integers(0, 100, size=(10000, 2)), columns=["B", "C"])
+    pdf.reset_index(inplace=True)
+    pdf.columns = ["A", "B", "C"]
+    pdf2 = pdf.copy()
+    pdf2["B"] = pdf2["B"] + 0.1
+    pdf2["C"] = pdf2["C"] + 0.3
+    df1 = spark_session.createDataFrame(pdf)
+    df2 = spark_session.createDataFrame(pdf2)
+    compare_tol = SparkSQLCompare(
+        spark_session, df1, df2, ["A"], abs_tol={"c": 0.0, "default": 0.2}
+    )
+    assert not compare_tol.matches()
+    assert compare_tol.df1_unq_rows.count() == 0
+    assert compare_tol.df2_unq_rows.count() == 0
+    assert compare_tol.intersect_columns() == {
+        "a",
+        "b",
+        "c",
+    }  # cast_column_names_lower is True by default
+    assert compare_tol.all_columns_match()
+    assert compare_tol.all_rows_overlap()
+    assert not compare_tol.intersect_rows_match()
+
+
+def test_10k_rows_rel_tol_per_column(spark_session):
+    rng = np.random.default_rng()
+    pdf = pd.DataFrame(rng.integers(0, 100, size=(10000, 2)), columns=["B", "C"])
+    pdf.reset_index(inplace=True)
+    pdf.columns = ["A", "B", "C"]
+    pdf2 = pdf.copy()
+    pdf2["B"] = pdf2["B"] + 0.1
+    df1 = spark_session.createDataFrame(pdf)
+    df2 = spark_session.createDataFrame(pdf2)
+    compare_tol = SparkSQLCompare(spark_session, df1, df2, ["A"], rel_tol={"B": 1.0})
+    assert compare_tol.matches()
+    assert compare_tol.df1_unq_rows.count() == 0
+    assert compare_tol.df2_unq_rows.count() == 0
+    assert compare_tol.intersect_columns() == {"a", "b", "c"}
+    assert compare_tol.all_columns_match()
+    assert compare_tol.all_rows_overlap()
+    assert compare_tol.intersect_rows_match()
+
+
+def test_10k_rows_rel_tol_per_column_default(spark_session):
+    rng = np.random.default_rng()
+    pdf = pd.DataFrame(rng.integers(0, 100, size=(10000, 2)), columns=["B", "C"])
+    pdf.reset_index(inplace=True)
+    pdf.columns = ["A", "B", "C"]
+    pdf2 = pdf.copy()
+    pdf2["B"] = pdf2["B"] + 0.1
+    pdf2["C"] = pdf2["C"] + 0.1
+    df1 = spark_session.createDataFrame(pdf)
+    df2 = spark_session.createDataFrame(pdf2)
+    compare_tol = SparkSQLCompare(
+        spark_session, df1, df2, ["A"], rel_tol={"c": 0.0, "default": 1}
+    )
+    assert not compare_tol.matches()
+    assert compare_tol.df1_unq_rows.count() == 0
+    assert compare_tol.df2_unq_rows.count() == 0
+    assert compare_tol.intersect_columns() == {"a", "b", "c"}
+    assert compare_tol.all_columns_match()
+    assert compare_tol.all_rows_overlap()
+    assert not compare_tol.intersect_rows_match()