feat: add max_reported_failures config to limit SchemaError output

pandera/backends/error_formatters.py

@@ -0,0 +1,45 @@
+"""Common error formatting utilities for all backends."""
+
+from typing import Callable, Optional, Tuple, TypeVar, Union
+
+T = TypeVar("T")
+
+
+def format_failure_cases_with_truncation(
+    failure_cases: T,
+    total_failures: int,
+    max_reported_failures: int,
+    format_all_cases: Callable[[T], str],
+    format_limited_cases: Callable[[T, int], Tuple[str, int]],
+) -> str:
+    """
+    Format failure cases with truncation based on max_reported_failures.
+
+    This function provides a unified way to handle failure case formatting
+    across different backends (pandas, polars, etc.) while allowing each
+    backend to maintain its specific formatting requirements.
+
+    :param failure_cases: The failure cases to format (backend-specific type)
+    :param total_failures: Total number of failures
+    :param max_reported_failures: Maximum failures to report
+        (-1 for unlimited, 0 for summary only)
+    :param format_all_cases: Function to format all cases without truncation
+    :param format_limited_cases: Function to format limited number of cases,
+        returns tuple of (formatted_string, actual_count_shown)
+    :return: Formatted failure cases string with truncation message if needed
+    """
+    if max_reported_failures == -1:
+        return format_all_cases(failure_cases)
+
+    if max_reported_failures == 0:
+        return f"... {total_failures} failure cases"
+
+    formatted_str, shown_count = format_limited_cases(
+        failure_cases, max_reported_failures
+    )
+
+    if shown_count < total_failures:
+        omitted_count = total_failures - shown_count
+        return f"{formatted_str} ... and {omitted_count} more failure cases ({total_failures} total)"
+
+    return formatted_str

pandera/backends/ibis/base.py

@@ -1,28 +1,30 @@
 """Ibis parsing, validation, and error-reporting backends."""
 
 import warnings
 from collections import defaultdict
 
 import ibis
 import ibis.selectors as s
 
 from pandera.api.ibis.error_handler import ErrorHandler
 from pandera.api.ibis.types import CheckResult
 from pandera.backends.base import BaseSchemaBackend, CoreCheckResult
 from pandera.backends.ibis.constants import POSITIONAL_JOIN_BACKENDS
 from pandera.backends.pandas.error_formatters import (
     consolidate_failure_cases,
+    reshape_failure_cases,
+)
+from pandera.backends.ibis.error_formatters import (
     format_generic_error_message,
     format_vectorized_error_message,
-    reshape_failure_cases,
 )
 from pandera.constants import CHECK_OUTPUT_KEY, CHECK_OUTPUT_SUFFIX
 from pandera.errors import (
     FailureCaseMetadata,
     SchemaError,
     SchemaErrorReason,
     SchemaWarning,
 )
 
 
 class IbisSchemaBackend(BaseSchemaBackend):

pandera/backends/ibis/error_formatters.py

@@ -0,0 +1,80 @@
+"""Make schema error messages human-friendly for Ibis."""
+
+import re
+from typing import Any, Optional
+
+import pandas as pd
+
+from pandera.backends.error_formatters import (
+    format_failure_cases_with_truncation,
+)
+from pandera.backends.pandas.error_formatters import (
+    format_generic_error_message,
+    reshape_failure_cases,
+)
+from pandera.config import get_config_context
+
+
+def format_vectorized_error_message(
+    parent_schema,
+    check,
+    check_index: int,
+    reshaped_failure_cases: Any,
+    max_reported_failures: Optional[int] = None,
+) -> str:
+    """Construct an error message when a validator fails.
+
+    :param parent_schema: class of schema being validated.
+    :param check: check that generated error.
+    :param check_index: The validator that failed.
+    :param reshaped_failure_cases: The failure cases encountered by the
+        element-wise or vectorized validator.
+    :param max_reported_failures: Maximum number of failures to report
+        in the error message. If None, use config value.
+    """
+    if max_reported_failures is None:
+        config = get_config_context()
+        max_reported_failures = config.max_reported_failures
+
+    pattern = r"<Check\s+([^:>]+):\s*([^>]+)>"
+    matches = re.findall(pattern, str(check))
+
+    check_strs = [f"{match[1]}" for match in matches]
+
+    if check_strs:
+        check_str = check_strs[0]
+    else:
+        check_str = str(check)
+
+    failure_cases = reshaped_failure_cases.failure_case
+    total_failures = len(failure_cases)
+
+    def format_all(cases):
+        return ", ".join(cases.apply(str))
+
+    def format_limited(cases, limit):
+        limited = cases.iloc[:limit]
+        formatted = ", ".join(limited.apply(str))
+        return formatted, len(limited)
+
+    failure_cases_string = format_failure_cases_with_truncation(
+        failure_cases,
+        total_failures,
+        max_reported_failures,
+        format_all,
+        format_limited,
+    )
+
+    return (
+        f"{parent_schema.__class__.__name__} '{parent_schema.name}' failed "
+        f"element-wise validator number {check_index}: "
+        f"{check_str} failure cases: {failure_cases_string}"
+    )
+
+
+# Re-export functions that don't need modification
+__all__ = [
+    "format_generic_error_message",
+    "format_vectorized_error_message",
+    "reshape_failure_cases",
+]

pandera/backends/pandas/error_formatters.py

@@ -1,10 +1,14 @@
 """Make schema error messages human-friendly."""
 
 import re
-from typing import Any, Union
+from typing import Any, Optional, Union
 
 import pandas as pd
 
+from pandera.backends.error_formatters import (
+    format_failure_cases_with_truncation,
+)
+from pandera.config import get_config_context
 from pandera.errors import SchemaError
 
 
@@ -26,11 +30,41 @@
     )
 
 
+def _format_failure_cases_string(
+    failure_cases,
+    total_failures: int,
+    max_reported_failures: int,
+    is_pyspark: bool = False,
+) -> str:
+    """Format failure cases into a string with appropriate truncation."""
+
+    def format_all(cases):
+        return ", ".join(cases.astype(str) if is_pyspark else cases.apply(str))
+
+    def format_limited(cases, limit):
+        if is_pyspark:
+            limited = cases[:limit]
+            formatted = ", ".join(limited.astype(str))
+        else:
+            limited = cases.iloc[:limit]
+            formatted = ", ".join(limited.apply(str))
+        return formatted, len(limited)
+
+    return format_failure_cases_with_truncation(
+        failure_cases,
+        total_failures,
+        max_reported_failures,
+        format_all,
+        format_limited,
+    )
+
+
 def format_vectorized_error_message(
     parent_schema,
     check,
     check_index: int,
     reshaped_failure_cases: Any,
+    max_reported_failures: Optional[int] = None,
 ) -> str:
     """Construct an error message when a validator fails.
 
@@ -39,8 +73,13 @@
     :param check_index: The validator that failed.
     :param reshaped_failure_cases: The failure cases encountered by the
         element-wise or vectorized validator.
+    :param max_reported_failures: Maximum number of failures to report
+        in the error message. If None, use config value.
 
     """
+    if max_reported_failures is None:
+        config = get_config_context()
+        max_reported_failures = config.max_reported_failures
 
     pattern = r"<Check\s+([^:>]+):\s*([^>]+)>"
     matches = re.findall(pattern, str(check))
@@ -52,14 +91,19 @@
     else:
         check_str = str(check)
 
-    if type(reshaped_failure_cases.failure_case).__module__.startswith(
-        "pyspark.pandas"
-    ):
+    is_pyspark = type(
+        reshaped_failure_cases.failure_case
+    ).__module__.startswith("pyspark.pandas")
+
+    if is_pyspark:
         failure_cases = reshaped_failure_cases.failure_case.to_numpy()
-        failure_cases_string = ", ".join(failure_cases.astype(str))
     else:
         failure_cases = reshaped_failure_cases.failure_case
-        failure_cases_string = ", ".join(failure_cases.apply(str))
+
+    total_failures = len(failure_cases)
+    failure_cases_string = _format_failure_cases_string(
+        failure_cases, total_failures, max_reported_failures, is_pyspark
+    )
 
     return (
         f"{parent_schema.__class__.__name__} '{parent_schema.name}' failed "
@@ -149,14 +193,23 @@
     return df.index.to_frame().drop_duplicates()
 
 
-def consolidate_failure_cases(schema_errors: list[SchemaError]):
+def consolidate_failure_cases(
+    schema_errors: list[SchemaError],
+    max_reported_failures: Optional[int] = None,
+):
     """Consolidate schema error dicts to produce data for error message."""
     from pandera.api.pandas.types import is_table
 
     assert schema_errors, (
         "schema_errors input cannot be empty. Check how the backend "
         "validation logic is handling/raising SchemaError(s)."
     )
+
+    # Get max_reported_failures from config if not provided
+    if max_reported_failures is None:
+        config = get_config_context()
+        max_reported_failures = config.max_reported_failures
+
     check_failure_cases = []
     scalar_check_failure_cases = []
 

pandera/backends/polars/base.py

@@ -13,6 +13,9 @@
     get_lazyframe_schema,
 )
 from pandera.backends.base import BaseSchemaBackend, CoreCheckResult
+from pandera.backends.polars.error_formatters import (
+    format_failure_cases_message,
+)
 from pandera.constants import CHECK_OUTPUT_KEY
 from pandera.errors import (
     FailureCaseMetadata,
@@ -100,7 +103,8 @@ def run_check(
                     _failure_cases = _failure_cases.drop(CHECK_OUTPUT_KEY)
 
                 failure_cases = _failure_cases.collect()
-                failure_cases_msg = failure_cases.head().rows(named=True)
+                failure_cases_msg = format_failure_cases_message(failure_cases)
+
                 message = (
                     f"{schema.__class__.__name__} '{schema.name}' failed "
                     f"validator number {check_index}: "

pandera/backends/polars/error_formatters.py

@@ -1,0 +1,41 @@
+"""Make schema error messages human-friendly for Polars."""
+
+from typing import Any, Optional
+
+from pandera.backends.error_formatters import (
+    format_failure_cases_with_truncation,
+)
+from pandera.config import get_config_context
+
+
+def format_failure_cases_message(
+    failure_cases: Any,
+    max_reported_failures: Optional[int] = None,
+) -> str:
+    """Format failure cases for Polars error messages.
+
+    :param failure_cases: Polars DataFrame containing failure cases
+    :param max_reported_failures: Maximum number of failures to report.
+        If None, uses config value.
+    :return: Formatted failure cases string
+    """
+    if max_reported_failures is None:
+        config = get_config_context()
+        max_reported_failures = config.max_reported_failures
+
+    total_failures = failure_cases.height
+
+    def format_all(cases):
+        return cases.rows(named=True)
+
+    def format_limited(cases, limit):
+        limited = cases.head(limit)
+        return limited.rows(named=True), limited.height
+
+    return format_failure_cases_with_truncation(
+        failure_cases,
+        total_failures,
+        max_reported_failures,
+        format_all,
+        format_limited,
+    )

pandera/config.py

@@ -32,6 +32,7 @@ class PanderaConfig:
     export PANDERA_VALIDATION_DEPTH=DATA_ONLY
     export PANDERA_CACHE_DATAFRAME=True
     export PANDERA_KEEP_CACHED_DATAFRAME=True
+    export PANDERA_MAX_REPORTED_FAILURES=100
     """
 
     validation_enabled: bool = True
@@ -42,6 +43,9 @@ class PanderaConfig:
     validation_depth: ValidationDepth | None = None
     cache_dataframe: bool = False
     keep_cached_dataframe: bool = False
+    max_reported_failures: int = (
+        100  # Default to showing first 100 reported failures
+    )
 
 
 def _config_from_env_vars():
@@ -61,11 +65,20 @@ def _config_from_env_vars():
         "PANDERA_KEEP_CACHED_DATAFRAME", "False"
     ) in {"True", "1"}
 
+    max_reported_failures_str = os.environ.get(
+        "PANDERA_MAX_REPORTED_FAILURES", "100"
+    )
+    try:
+        max_reported_failures = int(max_reported_failures_str)
+    except ValueError:
+        max_reported_failures = 100  # Default to 100 on invalid input
+
     return PanderaConfig(
         validation_enabled=validation_enabled,
         validation_depth=validation_depth,
         cache_dataframe=cache_dataframe,
         keep_cached_dataframe=keep_cached_dataframe,
+        max_reported_failures=max_reported_failures,
     )
 
 
@@ -80,6 +93,7 @@ def config_context(
     validation_depth: ValidationDepth | None = None,
     cache_dataframe: bool | None = None,
     keep_cached_dataframe: bool | None = None,
+    max_reported_failures: int | None = None,
 ):
     """Temporarily set pandera config options to custom settings."""
     _outer_config_ctx = get_config_context(validation_depth_default=None)
@@ -93,6 +107,8 @@ def config_context(
             _CONTEXT_CONFIG.cache_dataframe = cache_dataframe
         if keep_cached_dataframe is not None:
             _CONTEXT_CONFIG.keep_cached_dataframe = keep_cached_dataframe
+        if max_reported_failures is not None:
+            _CONTEXT_CONFIG.max_reported_failures = max_reported_failures
 
         yield
     finally: