Implement lazy wide table check execution for Ibis

pandera/backends/ibis/base.py

@@ -2,10 +2,13 @@
 
 import warnings
 from collections import defaultdict
+from typing import TYPE_CHECKING
 
 import ibis
 import ibis.selectors as s
+import pandas as pd
 
+from pandera.api.checks import Check
 from pandera.api.ibis.error_handler import ErrorHandler
 from pandera.api.ibis.types import CheckResult
 from pandera.backends.base import BaseSchemaBackend, CoreCheckResult
@@ -179,3 +182,112 @@ def join(left, right):
                 acc = acc & out[col]
 
         return out.filter(acc).drop(s.endswith(CHECK_OUTPUT_SUFFIX))
+
+    def _extract_check_results(
+        self,
+        original_table: ibis.Table,
+        wide_executed: pd.DataFrame,
+        checks_applied: list[tuple[int, Check]],
+        schema,
+    ) -> list[CoreCheckResult]:
+        """Extract individual CoreCheckResult from executed wide table.
+
+        :param original_table: The original Ibis table being validated.
+        :param wide_executed: The executed wide table as a pandas DataFrame.
+        :param checks_applied: List of (check_index, check) tuples for applied checks.
+        :param schema: The schema being validated against.
+        :returns: List of CoreCheckResult objects.
+        """
+        from pandera.api.pandas.types import is_table
+
+        results = []
+
+        # Get original column names (columns without check output suffix)
+        original_cols = [
+            c
+            for c in wide_executed.columns
+            if not c.endswith(CHECK_OUTPUT_SUFFIX)
+        ]
+
+        for check_index, check in checks_applied:
+            # Find check columns by prefix pattern: {check_index}_{col}{CHECK_OUTPUT_SUFFIX}
+            prefix = f"{check_index}_"
+            check_cols = [
+                c
+                for c in wide_executed.columns
+                if c.startswith(prefix) and c.endswith(CHECK_OUTPUT_SUFFIX)
+            ]
+
+            if not check_cols:
+                results.append(
+                    CoreCheckResult(
+                        passed=True,
+                        check=check,
+                        check_index=check_index,
+                        reason_code=SchemaErrorReason.DATAFRAME_CHECK,
+                    )
+                )
+                continue
+
+            # Compute passed: all check columns must be True for all rows
+            passed = wide_executed[check_cols].all(axis=None)
+
+            if passed:
+                results.append(
+                    CoreCheckResult(
+                        passed=True,
+                        check=check,
+                        check_index=check_index,
+                        reason_code=SchemaErrorReason.DATAFRAME_CHECK,
+                    )
+                )
+            else:
+                # Extract failure cases: rows where any check column is False
+                failure_mask = ~wide_executed[check_cols].all(axis=1)
+                failure_rows = wide_executed.loc[failure_mask, original_cols]
+
+                # Apply n_failure_cases limit (limiting rows before conversion)
+                if check.n_failure_cases is not None:
+                    failure_rows = failure_rows.head(check.n_failure_cases)
+
+                # Convert to dict records format, matching original run_check behavior
+                # Each row becomes a single failure case (a dict of column values)
+                if is_table(failure_rows):
+                    failure_cases = (
+                        pd.Series(failure_rows.to_dict("records"))
+                        .rename("failure_case")
+                        .to_frame()
+                    )
+                else:
+                    failure_cases = failure_rows
+
+                failure_cases = reshape_failure_cases(
+                    failure_cases, check.ignore_na
+                )
+                message = format_vectorized_error_message(
+                    schema, check, check_index, failure_cases
+                )
+
+                if check.raise_warning:
+                    warnings.warn(message, SchemaWarning)
+                    results.append(
+                        CoreCheckResult(
+                            passed=True,
+                            check=check,
+                            check_index=check_index,
+                            reason_code=SchemaErrorReason.DATAFRAME_CHECK,
+                        )
+                    )
+                else:
+                    results.append(
+                        CoreCheckResult(
+                            passed=False,
+                            check=check,
+                            check_index=check_index,
+                            reason_code=SchemaErrorReason.DATAFRAME_CHECK,
+                            message=message,
+                            failure_cases=failure_cases,
+                        )
+                    )
+
+        return results

pandera/backends/ibis/checks.py

@@ -56,8 +56,122 @@ def preprocess(
         # a Table with a single column. Table inputs are unaffected.
         return check_obj.as_table()
 
-    def apply(self, check_obj: IbisData):
-        """Apply the check function to a check object."""
+    def apply(
+        self,
+        check_obj: IbisData,
+        wide_table: ibis.Table | None = None,
+        check_index: int | None = None,
+    ) -> ibis.Table:
+        """Apply the check function to a check object.
+
+        If wide_table is provided, add check result columns to it for lazy
+        wide table execution. If not provided, return the check results
+        in the original format (current behavior for backwards compatibility).
+
+        :param check_obj: The IbisData object containing the table and key.
+        :param wide_table: Optional wide table to add check result columns to.
+        :param check_index: Index of the check for unique column naming.
+        :returns: Table with check result columns, or scalar/column for
+            backwards compatibility when wide_table is None.
+        """
+        # Wide table mode: build a wide table with all check results
+        if wide_table is not None:
+            return self._apply_wide_table(check_obj, wide_table, check_index)
+
+        # Original behavior: return check results in their native format
+        return self._apply_original(check_obj)
+
+    def _apply_wide_table(
+        self,
+        check_obj: IbisData,
+        wide_table: ibis.Table,
+        check_index: int | None,
+    ) -> ibis.Table:
+        """Apply check and add result columns to the wide table.
+
+        :param check_obj: The IbisData object containing the table and key.
+        :param wide_table: Wide table to add check result columns to.
+        :param check_index: Index of the check for unique column naming.
+        :returns: Wide table with check result columns added.
+        """
+        prefix = f"{check_index}_" if check_index is not None else ""
+
+        if self.check.element_wise:
+            selector = (
+                select_column(check_obj.key)
+                if check_obj.key is not None
+                else s.all()
+            )
+            out = wide_table.mutate(
+                s.across(
+                    selector,
+                    self.check_fn,
+                    f"{prefix}{{col}}{CHECK_OUTPUT_SUFFIX}",
+                )
+            )
+        else:
+            check_output = self.check_fn(check_obj)
+
+            if isinstance(check_output, dict):
+                out = wide_table.mutate(
+                    **{
+                        f"{prefix}{k}{CHECK_OUTPUT_SUFFIX}": v
+                        for k, v in check_output.items()
+                    }
+                )
+            elif isinstance(check_output, ibis.Table):
+                # Rename columns with prefix and suffix
+                check_output = check_output.rename(
+                    f"{prefix}{{name}}{CHECK_OUTPUT_SUFFIX}"
+                )
+                # Join check output to wide table
+                if wide_table.get_backend().name in POSITIONAL_JOIN_BACKENDS:
+                    out = wide_table.join(check_output, how="positional")
+                else:
+                    # For backends that do not support positional joins:
+                    # https://github.com/ibis-project/ibis/issues/9486
+                    index_col = "__idx__"
+                    out = (
+                        wide_table.mutate(
+                            **{index_col: ibis.row_number().over()}
+                        )
+                        .join(
+                            check_output.mutate(
+                                **{index_col: ibis.row_number().over()}
+                            ),
+                            index_col,
+                        )
+                        .drop(index_col)
+                    )
+            elif isinstance(check_output, bool):
+                col_name = f"{prefix}{self.check.name or 'check'}{CHECK_OUTPUT_SUFFIX}"
+                out = wide_table.mutate(
+                    **{col_name: ibis.literal(check_output)}
+                )
+            elif check_output.type().is_boolean():
+                col_name = f"{prefix}{self.check.name or 'check'}{CHECK_OUTPUT_SUFFIX}"
+                out = wide_table.mutate(**{col_name: check_output})
+            else:
+                raise TypeError(
+                    f"output type of check_fn not recognized: {type(check_output)}"
+                )
+
+        # Validate that check output columns are boolean
+        for col in out.columns:
+            if col.startswith(prefix) and col.endswith(CHECK_OUTPUT_SUFFIX):
+                assert out[col].type().is_boolean(), (
+                    f"column '{col}' is not boolean. If check function "
+                    "returns a table, it must contain only boolean columns."
+                )
+
+        return out
+
+    def _apply_original(self, check_obj: IbisData):
+        """Apply check using original behavior (backwards compatibility).
+
+        :param check_obj: The IbisData object containing the table and key.
+        :returns: Check result in native format (table, column, or scalar).
+        """
         if self.check.element_wise:
             selector = (
                 select_column(check_obj.key)

pandera/backends/ibis/container.py

@@ -14,6 +14,7 @@
 
 from pandera.api.base.error_handler import get_error_category
 from pandera.api.ibis.error_handler import ErrorHandler
+from pandera.api.ibis.types import IbisData
 from pandera.backends.base import ColumnInfo, CoreCheckResult
 from pandera.backends.ibis.base import IbisSchemaBackend
 from pandera.backends.utils import convert_uniquesettings
@@ -29,6 +30,7 @@
 from pandera.validation_depth import validate_scope, validation_type
 
 if TYPE_CHECKING:
+    from pandera.api.checks import Check
     from pandera.api.ibis.container import DataFrameSchema
 
 
@@ -138,25 +140,44 @@ def run_checks(
         check_obj: ibis.Table,
         schema,
     ) -> list[CoreCheckResult]:
-        """Run a list of checks on the check object."""
-        # dataframe-level checks
-        check_results: list[CoreCheckResult] = []
+        """Run a list of checks using lazy evaluation with a shared wide table.
+
+        This method builds a wide table by passing through each check's apply()
+        method, then executes the wide table once and extracts individual check
+        results.
+
+        :param check_obj: The Ibis table to validate.
+        :param schema: The schema containing the checks to run.
+        :returns: List of CoreCheckResult objects.
+        """
+        if not schema.checks:
+            return []
+
+        # Preprocess once
+        wide_table = check_obj.as_table()
+        ibis_data = IbisData(wide_table, None)
+        checks_applied: list[tuple[int, Check]] = []
+        immediate_errors: list[CoreCheckResult] = []
+
+        # Phase 1: Build wide table by passing through each check's apply()
         for check_index, check in enumerate(schema.checks):
             try:
-                check_results.append(
-                    self.run_check(check_obj, schema, check, check_index)
+                check_backend = check.get_backend(check_obj)(check)
+                wide_table = check_backend.apply(
+                    ibis_data, wide_table, check_index
                 )
+                checks_applied.append((check_index, check))
             except SchemaDefinitionError:
                 raise
             except Exception as err:
-                # catch other exceptions that may occur when executing the check
+                # catch other exceptions that may occur when building the check
                 err_msg = f'"{err.args[0]}"' if err.args else ""
                 err_str = f"{err.__class__.__name__}({err_msg})"
                 msg = (
                     f"Error while executing check function: {err_str}\n"
                     + traceback.format_exc()
                 )
-                check_results.append(
+                immediate_errors.append(
                     CoreCheckResult(
                         passed=False,
                         check=check,
@@ -167,7 +188,19 @@ def run_checks(
                         original_exc=err,
                     )
                 )
-        return check_results
+
+        if not checks_applied:
+            return immediate_errors
+
+        # Phase 2: Execute wide table once
+        wide_table_executed = wide_table.to_pandas()
+
+        # Phase 3: Extract individual check results
+        check_results = self._extract_check_results(
+            check_obj, wide_table_executed, checks_applied, schema
+        )
+
+        return immediate_errors + check_results
 
     def run_schema_component_checks(
         self,

pandera/schema_statistics/pandas.py

@@ -202,8 +202,9 @@ def parse_checks(checks) -> Union[list[dict[str, Any]], None]:
 
     incompatibile_checks_count = sum(
         map(
-            lambda check: check["options"]["check_name"]
-            in incompatibile_checks,
+            lambda check: (
+                check["options"]["check_name"] in incompatibile_checks
+            ),
             check_statistics,
         )
     )