We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Pandas backend: `check_nullable` is inefficient when a column schema has `nullable=True`
`check_nullable` in the pandas backend seems to compute the null values mask by calling `isna()` even when not needed.

https://github.com/unionai-oss/pandera/blob/9c484a92cc6e63ba11652444e9e6df9e587d668e/pandera/backends/pandas/array.py#L196C9-L198

```
isna = check_obj.isna()
passed = schema.nullable or not isna.any()
```

We see that even if `schema.nullable=True`, `isna` is already computed. This can lead to a performance issue in dataframe with millions of rows.





--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `pandera/backends/pandas/array.py`
Content:
```
1 """Pandera array backends."""
2 
3 from typing import cast, List, Optional
4 
5 import pandas as pd
6 from multimethod import DispatchError
7 from pandera.api.base.error_handler import ErrorHandler
8 
9 from pandera.api.pandas.types import is_field
10 from pandera.backends.base import CoreCheckResult
11 from pandera.backends.pandas.base import PandasSchemaBackend
12 from pandera.backends.pandas.error_formatters import (
13     reshape_failure_cases,
14     scalar_failure_case,
15 )
16 from pandera.backends.utils import convert_uniquesettings
17 from pandera.config import ValidationScope
18 from pandera.engines.pandas_engine import Engine
19 from pandera.errors import (
20     ParserError,
21     SchemaError,
22     SchemaErrorReason,
23     SchemaErrors,
24     SchemaDefinitionError,
25 )
26 from pandera.validation_depth import (
27     validation_type,
28     validate_scope,
29 )
30 
31 
32 class ArraySchemaBackend(PandasSchemaBackend):
33     """Backend for pandas arrays."""
34 
35     def preprocess(self, check_obj, inplace: bool = False):
36         return check_obj if inplace else check_obj.copy()
37 
38     def validate(
39         self,
40         check_obj,
41         schema,
42         *,
43         head: Optional[int] = None,
44         tail: Optional[int] = None,
45         sample: Optional[int] = None,
46         random_state: Optional[int] = None,
47         lazy: bool = False,
48         inplace: bool = False,
49     ):
50         # pylint: disable=too-many-locals
51         error_handler = ErrorHandler(lazy)
52         check_obj = self.preprocess(check_obj, inplace)
53 
54         if getattr(schema, "drop_invalid_rows", False) and not lazy:
55             raise SchemaDefinitionError(
56                 "When drop_invalid_rows is True, lazy must be set to True."
57             )
58 
59         # fill nans with `default` if it's present
60         if hasattr(schema, "default") and schema.default is not None:
61             check_obj = self.set_default(check_obj, schema)
62 
63         try:
64             if is_field(check_obj) and schema.coerce:
65                 check_obj = self.coerce_dtype(check_obj, schema=schema)
66             elif schema.coerce:
67                 check_obj[schema.name] = self.coerce_dtype(
68                     check_obj[schema.name], schema=schema
69                 )
70         except SchemaError as exc:
71             error_handler.collect_error(
72                 validation_type(exc.reason_code),
73                 exc.reason_code,
74                 exc,
75             )
76 
77         # run the core checks
78         error_handler = self.run_checks_and_handle_errors(
79             error_handler,
80             schema,
81             check_obj,
82             head=head,
83             tail=tail,
84             sample=sample,
85             random_state=random_state,
86         )
87 
88         if lazy and error_handler.collected_errors:
89             if getattr(schema, "drop_invalid_rows", False):
90                 check_obj = self.drop_invalid_rows(check_obj, error_handler)
91             else:
92                 raise SchemaErrors(
93                     schema=schema,
94                     schema_errors=error_handler.schema_errors,
95                     data=check_obj,
96                 )
97 
98         return check_obj
99 
100     def run_checks_and_handle_errors(
101         self, error_handler, schema, check_obj, **subsample_kwargs
102     ):
103         """Run checks on schema"""
104         # pylint: disable=too-many-locals
105         field_obj_subsample = self.subsample(
106             check_obj if is_field(check_obj) else check_obj[schema.name],
107             **subsample_kwargs,
108         )
109 
110         check_obj_subsample = self.subsample(check_obj, **subsample_kwargs)
111 
112         core_checks = [
113             (self.check_name, (field_obj_subsample, schema)),
114             (self.check_nullable, (field_obj_subsample, schema)),
115             (self.check_unique, (field_obj_subsample, schema)),
116             (self.check_dtype, (field_obj_subsample, schema)),
117             (self.run_checks, (check_obj_subsample, schema)),
118         ]
119 
120         for core_check, args in core_checks:
121             results = core_check(*args)
122             if isinstance(results, CoreCheckResult):
123                 results = [results]
124             results = cast(List[CoreCheckResult], results)
125             for result in results:
126                 if result.passed:
127                     continue
128 
129                 if result.schema_error is not None:
130                     error = result.schema_error
131                 else:
132                     error = SchemaError(
133                         schema=schema,
134                         data=check_obj,
135                         message=result.message,
136                         failure_cases=result.failure_cases,
137                         check=result.check,
138                         check_index=result.check_index,
139                         check_output=result.check_output,
140                         reason_code=result.reason_code,
141                     )
142                     error_handler.collect_error(
143                         validation_type(result.reason_code),
144                         result.reason_code,
145                         error,
146                         original_exc=result.original_exc,
147                     )
148 
149         return error_handler
150 
151     def coerce_dtype(
152         self,
153         check_obj,
154         schema=None,
155         # pylint: disable=unused-argument
156     ):
157         """Coerce type of a pd.Series by type specified in dtype.
158 
159         :param pd.Series series: One-dimensional ndarray with axis labels
160             (including time series).
161         :returns: ``Series`` with coerced data type
162         """
163         assert schema is not None, "The `schema` argument must be provided."
164         if schema.dtype is None or not schema.coerce:
165             return check_obj
166 
167         try:
168             return schema.dtype.try_coerce(check_obj)
169         except ParserError as exc:
170             raise SchemaError(
171                 schema=schema,
172                 data=check_obj,
173                 message=(
174                     f"Error while coercing '{schema.name}' to type "
175                     f"{schema.dtype}: {exc}:\n{exc.failure_cases}"
176                 ),
177                 failure_cases=exc.failure_cases,
178                 check=f"coerce_dtype('{schema.dtype}')",
179                 reason_code=SchemaErrorReason.DATATYPE_COERCION,
180             ) from exc
181 
182     @validate_scope(scope=ValidationScope.SCHEMA)
183     def check_name(self, check_obj: pd.Series, schema) -> CoreCheckResult:
184         return CoreCheckResult(
185             passed=schema.name is None or check_obj.name == schema.name,
186             check=f"field_name('{schema.name}')",
187             reason_code=SchemaErrorReason.WRONG_FIELD_NAME,
188             message=(
189                 f"Expected {type(check_obj)} to have name '{schema.name}', "
190                 f"found '{check_obj.name}'"
191             ),
192             failure_cases=scalar_failure_case(check_obj.name),
193         )
194 
195     @validate_scope(scope=ValidationScope.SCHEMA)
196     def check_nullable(self, check_obj: pd.Series, schema) -> CoreCheckResult:
197         isna = check_obj.isna()
198         passed = schema.nullable or not isna.any()
199         return CoreCheckResult(
200             passed=cast(bool, passed),
201             check="not_nullable",
202             reason_code=SchemaErrorReason.SERIES_CONTAINS_NULLS,
203             message=(
204                 f"non-nullable series '{check_obj.name}' contains "
205                 f"null values:\n{check_obj[isna]}"
206             ),
207             failure_cases=reshape_failure_cases(
208                 check_obj[isna], ignore_na=False
209             ),
210         )
211 
212     @validate_scope(scope=ValidationScope.DATA)
213     def check_unique(self, check_obj: pd.Series, schema) -> CoreCheckResult:
214         passed = True
215         failure_cases = None
216         message = None
217 
218         if schema.unique:
219             keep_argument = convert_uniquesettings(schema.report_duplicates)
220             if type(check_obj).__module__.startswith("pyspark.pandas"):
221                 # pylint: disable=import-outside-toplevel
222                 import pyspark.pandas as ps
223 
224                 duplicates = (
225                     check_obj.to_frame()  # type: ignore
226                     .duplicated(keep=keep_argument)  # type: ignore
227                     .reindex(check_obj.index)
228                 )
229                 with ps.option_context("compute.ops_on_diff_frames", True):
230                     failed = check_obj[duplicates]
231             else:
232                 duplicates = check_obj.duplicated(keep=keep_argument)  # type: ignore
233                 failed = check_obj[duplicates]
234 
235             if duplicates.any():
236                 passed = False
237                 failure_cases = reshape_failure_cases(failed)
238                 message = (
239                     f"series '{check_obj.name}' contains duplicate "
240                     f"values:\n{failed}"
241                 )
242 
243         return CoreCheckResult(
244             passed=passed,
245             check="field_uniqueness",
246             reason_code=SchemaErrorReason.SERIES_CONTAINS_DUPLICATES,
247             message=message,
248             failure_cases=failure_cases,
249         )
250 
251     @validate_scope(scope=ValidationScope.SCHEMA)
252     def check_dtype(self, check_obj: pd.Series, schema) -> CoreCheckResult:
253         passed = True
254         failure_cases = None
255         msg = None
256 
257         if schema.dtype is not None:
258             dtype_check_results = schema.dtype.check(
259                 Engine.dtype(check_obj.dtype),
260                 check_obj,
261             )
262             if isinstance(dtype_check_results, bool):
263                 passed = dtype_check_results
264                 failure_cases = scalar_failure_case(str(check_obj.dtype))
265                 msg = (
266                     f"expected series '{check_obj.name}' to have type "
267                     f"{schema.dtype}, got {check_obj.dtype}"
268                 )
269             else:
270                 passed = dtype_check_results.all()
271                 failure_cases = reshape_failure_cases(
272                     check_obj[~dtype_check_results.astype(bool)],
273                     ignore_na=False,
274                 )
275                 msg = (
276                     f"expected series '{check_obj.name}' to have type "
277                     f"{schema.dtype}:\nfailure cases:\n{failure_cases}"
278                 )
279 
280         return CoreCheckResult(
281             passed=passed,
282             check=f"dtype('{schema.dtype}')",
283             reason_code=SchemaErrorReason.WRONG_DATATYPE,
284             message=msg,
285             failure_cases=failure_cases,
286         )
287 
288     @validate_scope(scope=ValidationScope.DATA)
289     def run_checks(self, check_obj, schema) -> List[CoreCheckResult]:
290         check_results: List[CoreCheckResult] = []
291         for check_index, check in enumerate(schema.checks):
292             check_args = [None] if is_field(check_obj) else [schema.name]
293             try:
294                 check_results.append(
295                     self.run_check(
296                         check_obj,
297                         schema,
298                         check,
299                         check_index,
300                         *check_args,
301                     )
302                 )
303             except Exception as err:  # pylint: disable=broad-except
304                 # catch other exceptions that may occur when executing the Check
305                 if isinstance(err, DispatchError):
306                     # if the error was raised by a check registered via
307                     # multimethod, get the underlying __cause__
308                     err = err.__cause__
309                 err_msg = f'"{err.args[0]}"' if len(err.args) > 0 else ""
310                 msg = f"{err.__class__.__name__}({err_msg})"
311                 check_results.append(
312                     CoreCheckResult(
313                         passed=False,
314                         check=check,
315                         check_index=check_index,
316                         reason_code=SchemaErrorReason.CHECK_ERROR,
317                         message=msg,
318                         failure_cases=scalar_failure_case(msg),
319                         original_exc=err,
320                     )
321                 )
322         return check_results
323 
324     def set_default(self, check_obj, schema):
325         """Sets the ``schema.default`` value on the ``check_obj``"""
326         # Ignore sparse dtype as it can't assign default value directly
327         if is_field(check_obj) and not isinstance(
328             check_obj.dtype, pd.SparseDtype
329         ):
330             check_obj = check_obj.fillna(schema.default)
331         elif not is_field(check_obj) and not isinstance(
332             check_obj[schema.name].dtype, pd.SparseDtype
333         ):
334             check_obj[schema.name] = check_obj[schema.name].fillna(
335                 schema.default
336             )
337 
338         return check_obj
339 
340 
341 class SeriesSchemaBackend(ArraySchemaBackend):
342     """Backend for pandas Series objects."""
343 
344     def coerce_dtype(
345         self,
346         check_obj,
347         schema=None,
348     ):
349         if hasattr(check_obj, "pandera"):
350             check_obj = check_obj.pandera.add_schema(schema)
351 
352         check_obj = super().coerce_dtype(check_obj, schema=schema)
353 
354         if hasattr(check_obj, "pandera"):
355             check_obj = check_obj.pandera.add_schema(schema)
356         return check_obj
357
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/pandera/backends/pandas/array.py b/pandera/backends/pandas/array.py
--- a/pandera/backends/pandas/array.py
+++ b/pandera/backends/pandas/array.py
@@ -194,8 +194,16 @@
 
     @validate_scope(scope=ValidationScope.SCHEMA)
     def check_nullable(self, check_obj: pd.Series, schema) -> CoreCheckResult:
+        if schema.nullable:
+            # Avoid to compute anything for perf reasons. GH#1533
+            return CoreCheckResult(
+                passed=True,
+                check="not_nullable",
+            )
+
+        # Check actual column contents
         isna = check_obj.isna()
-        passed = schema.nullable or not isna.any()
+        passed = not isna.any()
         return CoreCheckResult(
             passed=cast(bool, passed),
             check="not_nullable",

{"golden_diff": "diff --git a/pandera/backends/pandas/array.py b/pandera/backends/pandas/array.py\n--- a/pandera/backends/pandas/array.py\n+++ b/pandera/backends/pandas/array.py\n@@ -194,8 +194,16 @@\n \n     @validate_scope(scope=ValidationScope.SCHEMA)\n     def check_nullable(self, check_obj: pd.Series, schema) -> CoreCheckResult:\n+        if schema.nullable:\n+            # Avoid to compute anything for perf reasons. GH#1533\n+            return CoreCheckResult(\n+                passed=True,\n+                check=\"not_nullable\",\n+            )\n+\n+        # Check actual column contents\n         isna = check_obj.isna()\n-        passed = schema.nullable or not isna.any()\n+        passed = not isna.any()\n         return CoreCheckResult(\n             passed=cast(bool, passed),\n             check=\"not_nullable\",\n", "issue": "Pandas backend: `check_nullable` is inefficient when a column schema has `nullable=True`\n`check_nullable` in the pandas backend seems to compute the null values mask by calling `isna()` even when not needed.\r\n\r\nhttps://github.com/unionai-oss/pandera/blob/9c484a92cc6e63ba11652444e9e6df9e587d668e/pandera/backends/pandas/array.py#L196C9-L198\r\n\r\n```\r\nisna = check_obj.isna()\r\npassed = schema.nullable or not isna.any()\r\n```\r\n\r\nWe see that even if `schema.nullable=True`, `isna` is already computed. This can lead to a performance issue in dataframe with millions of rows.\r\n\r\n\r\n\r\n\n", "before_files": [{"content": "\"\"\"Pandera array backends.\"\"\"\n\nfrom typing import cast, List, Optional\n\nimport pandas as pd\nfrom multimethod import DispatchError\nfrom pandera.api.base.error_handler import ErrorHandler\n\nfrom pandera.api.pandas.types import is_field\nfrom pandera.backends.base import CoreCheckResult\nfrom pandera.backends.pandas.base import PandasSchemaBackend\nfrom pandera.backends.pandas.error_formatters import (\n    reshape_failure_cases,\n    scalar_failure_case,\n)\nfrom pandera.backends.utils import convert_uniquesettings\nfrom pandera.config import ValidationScope\nfrom pandera.engines.pandas_engine import Engine\nfrom pandera.errors import (\n    ParserError,\n    SchemaError,\n    SchemaErrorReason,\n    SchemaErrors,\n    SchemaDefinitionError,\n)\nfrom pandera.validation_depth import (\n    validation_type,\n    validate_scope,\n)\n\n\nclass ArraySchemaBackend(PandasSchemaBackend):\n    \"\"\"Backend for pandas arrays.\"\"\"\n\n    def preprocess(self, check_obj, inplace: bool = False):\n        return check_obj if inplace else check_obj.copy()\n\n    def validate(\n        self,\n        check_obj,\n        schema,\n        *,\n        head: Optional[int] = None,\n        tail: Optional[int] = None,\n        sample: Optional[int] = None,\n        random_state: Optional[int] = None,\n        lazy: bool = False,\n        inplace: bool = False,\n    ):\n        # pylint: disable=too-many-locals\n        error_handler = ErrorHandler(lazy)\n        check_obj = self.preprocess(check_obj, inplace)\n\n        if getattr(schema, \"drop_invalid_rows\", False) and not lazy:\n            raise SchemaDefinitionError(\n                \"When drop_invalid_rows is True, lazy must be set to True.\"\n            )\n\n        # fill nans with `default` if it's present\n        if hasattr(schema, \"default\") and schema.default is not None:\n            check_obj = self.set_default(check_obj, schema)\n\n        try:\n            if is_field(check_obj) and schema.coerce:\n                check_obj = self.coerce_dtype(check_obj, schema=schema)\n            elif schema.coerce:\n                check_obj[schema.name] = self.coerce_dtype(\n                    check_obj[schema.name], schema=schema\n                )\n        except SchemaError as exc:\n            error_handler.collect_error(\n                validation_type(exc.reason_code),\n                exc.reason_code,\n                exc,\n            )\n\n        # run the core checks\n        error_handler = self.run_checks_and_handle_errors(\n            error_handler,\n            schema,\n            check_obj,\n            head=head,\n            tail=tail,\n            sample=sample,\n            random_state=random_state,\n        )\n\n        if lazy and error_handler.collected_errors:\n            if getattr(schema, \"drop_invalid_rows\", False):\n                check_obj = self.drop_invalid_rows(check_obj, error_handler)\n            else:\n                raise SchemaErrors(\n                    schema=schema,\n                    schema_errors=error_handler.schema_errors,\n                    data=check_obj,\n                )\n\n        return check_obj\n\n    def run_checks_and_handle_errors(\n        self, error_handler, schema, check_obj, **subsample_kwargs\n    ):\n        \"\"\"Run checks on schema\"\"\"\n        # pylint: disable=too-many-locals\n        field_obj_subsample = self.subsample(\n            check_obj if is_field(check_obj) else check_obj[schema.name],\n            **subsample_kwargs,\n        )\n\n        check_obj_subsample = self.subsample(check_obj, **subsample_kwargs)\n\n        core_checks = [\n            (self.check_name, (field_obj_subsample, schema)),\n            (self.check_nullable, (field_obj_subsample, schema)),\n            (self.check_unique, (field_obj_subsample, schema)),\n            (self.check_dtype, (field_obj_subsample, schema)),\n            (self.run_checks, (check_obj_subsample, schema)),\n        ]\n\n        for core_check, args in core_checks:\n            results = core_check(*args)\n            if isinstance(results, CoreCheckResult):\n                results = [results]\n            results = cast(List[CoreCheckResult], results)\n            for result in results:\n                if result.passed:\n                    continue\n\n                if result.schema_error is not None:\n                    error = result.schema_error\n                else:\n                    error = SchemaError(\n                        schema=schema,\n                        data=check_obj,\n                        message=result.message,\n                        failure_cases=result.failure_cases,\n                        check=result.check,\n                        check_index=result.check_index,\n                        check_output=result.check_output,\n                        reason_code=result.reason_code,\n                    )\n                    error_handler.collect_error(\n                        validation_type(result.reason_code),\n                        result.reason_code,\n                        error,\n                        original_exc=result.original_exc,\n                    )\n\n        return error_handler\n\n    def coerce_dtype(\n        self,\n        check_obj,\n        schema=None,\n        # pylint: disable=unused-argument\n    ):\n        \"\"\"Coerce type of a pd.Series by type specified in dtype.\n\n        :param pd.Series series: One-dimensional ndarray with axis labels\n            (including time series).\n        :returns: ``Series`` with coerced data type\n        \"\"\"\n        assert schema is not None, \"The `schema` argument must be provided.\"\n        if schema.dtype is None or not schema.coerce:\n            return check_obj\n\n        try:\n            return schema.dtype.try_coerce(check_obj)\n        except ParserError as exc:\n            raise SchemaError(\n                schema=schema,\n                data=check_obj,\n                message=(\n                    f\"Error while coercing '{schema.name}' to type \"\n                    f\"{schema.dtype}: {exc}:\\n{exc.failure_cases}\"\n                ),\n                failure_cases=exc.failure_cases,\n                check=f\"coerce_dtype('{schema.dtype}')\",\n                reason_code=SchemaErrorReason.DATATYPE_COERCION,\n            ) from exc\n\n    @validate_scope(scope=ValidationScope.SCHEMA)\n    def check_name(self, check_obj: pd.Series, schema) -> CoreCheckResult:\n        return CoreCheckResult(\n            passed=schema.name is None or check_obj.name == schema.name,\n            check=f\"field_name('{schema.name}')\",\n            reason_code=SchemaErrorReason.WRONG_FIELD_NAME,\n            message=(\n                f\"Expected {type(check_obj)} to have name '{schema.name}', \"\n                f\"found '{check_obj.name}'\"\n            ),\n            failure_cases=scalar_failure_case(check_obj.name),\n        )\n\n    @validate_scope(scope=ValidationScope.SCHEMA)\n    def check_nullable(self, check_obj: pd.Series, schema) -> CoreCheckResult:\n        isna = check_obj.isna()\n        passed = schema.nullable or not isna.any()\n        return CoreCheckResult(\n            passed=cast(bool, passed),\n            check=\"not_nullable\",\n            reason_code=SchemaErrorReason.SERIES_CONTAINS_NULLS,\n            message=(\n                f\"non-nullable series '{check_obj.name}' contains \"\n                f\"null values:\\n{check_obj[isna]}\"\n            ),\n            failure_cases=reshape_failure_cases(\n                check_obj[isna], ignore_na=False\n            ),\n        )\n\n    @validate_scope(scope=ValidationScope.DATA)\n    def check_unique(self, check_obj: pd.Series, schema) -> CoreCheckResult:\n        passed = True\n        failure_cases = None\n        message = None\n\n        if schema.unique:\n            keep_argument = convert_uniquesettings(schema.report_duplicates)\n            if type(check_obj).__module__.startswith(\"pyspark.pandas\"):\n                # pylint: disable=import-outside-toplevel\n                import pyspark.pandas as ps\n\n                duplicates = (\n                    check_obj.to_frame()  # type: ignore\n                    .duplicated(keep=keep_argument)  # type: ignore\n                    .reindex(check_obj.index)\n                )\n                with ps.option_context(\"compute.ops_on_diff_frames\", True):\n                    failed = check_obj[duplicates]\n            else:\n                duplicates = check_obj.duplicated(keep=keep_argument)  # type: ignore\n                failed = check_obj[duplicates]\n\n            if duplicates.any():\n                passed = False\n                failure_cases = reshape_failure_cases(failed)\n                message = (\n                    f\"series '{check_obj.name}' contains duplicate \"\n                    f\"values:\\n{failed}\"\n                )\n\n        return CoreCheckResult(\n            passed=passed,\n            check=\"field_uniqueness\",\n            reason_code=SchemaErrorReason.SERIES_CONTAINS_DUPLICATES,\n            message=message,\n            failure_cases=failure_cases,\n        )\n\n    @validate_scope(scope=ValidationScope.SCHEMA)\n    def check_dtype(self, check_obj: pd.Series, schema) -> CoreCheckResult:\n        passed = True\n        failure_cases = None\n        msg = None\n\n        if schema.dtype is not None:\n            dtype_check_results = schema.dtype.check(\n                Engine.dtype(check_obj.dtype),\n                check_obj,\n            )\n            if isinstance(dtype_check_results, bool):\n                passed = dtype_check_results\n                failure_cases = scalar_failure_case(str(check_obj.dtype))\n                msg = (\n                    f\"expected series '{check_obj.name}' to have type \"\n                    f\"{schema.dtype}, got {check_obj.dtype}\"\n                )\n            else:\n                passed = dtype_check_results.all()\n                failure_cases = reshape_failure_cases(\n                    check_obj[~dtype_check_results.astype(bool)],\n                    ignore_na=False,\n                )\n                msg = (\n                    f\"expected series '{check_obj.name}' to have type \"\n                    f\"{schema.dtype}:\\nfailure cases:\\n{failure_cases}\"\n                )\n\n        return CoreCheckResult(\n            passed=passed,\n            check=f\"dtype('{schema.dtype}')\",\n            reason_code=SchemaErrorReason.WRONG_DATATYPE,\n            message=msg,\n            failure_cases=failure_cases,\n        )\n\n    @validate_scope(scope=ValidationScope.DATA)\n    def run_checks(self, check_obj, schema) -> List[CoreCheckResult]:\n        check_results: List[CoreCheckResult] = []\n        for check_index, check in enumerate(schema.checks):\n            check_args = [None] if is_field(check_obj) else [schema.name]\n            try:\n                check_results.append(\n                    self.run_check(\n                        check_obj,\n                        schema,\n                        check,\n                        check_index,\n                        *check_args,\n                    )\n                )\n            except Exception as err:  # pylint: disable=broad-except\n                # catch other exceptions that may occur when executing the Check\n                if isinstance(err, DispatchError):\n                    # if the error was raised by a check registered via\n                    # multimethod, get the underlying __cause__\n                    err = err.__cause__\n                err_msg = f'\"{err.args[0]}\"' if len(err.args) > 0 else \"\"\n                msg = f\"{err.__class__.__name__}({err_msg})\"\n                check_results.append(\n                    CoreCheckResult(\n                        passed=False,\n                        check=check,\n                        check_index=check_index,\n                        reason_code=SchemaErrorReason.CHECK_ERROR,\n                        message=msg,\n                        failure_cases=scalar_failure_case(msg),\n                        original_exc=err,\n                    )\n                )\n        return check_results\n\n    def set_default(self, check_obj, schema):\n        \"\"\"Sets the ``schema.default`` value on the ``check_obj``\"\"\"\n        # Ignore sparse dtype as it can't assign default value directly\n        if is_field(check_obj) and not isinstance(\n            check_obj.dtype, pd.SparseDtype\n        ):\n            check_obj = check_obj.fillna(schema.default)\n        elif not is_field(check_obj) and not isinstance(\n            check_obj[schema.name].dtype, pd.SparseDtype\n        ):\n            check_obj[schema.name] = check_obj[schema.name].fillna(\n                schema.default\n            )\n\n        return check_obj\n\n\nclass SeriesSchemaBackend(ArraySchemaBackend):\n    \"\"\"Backend for pandas Series objects.\"\"\"\n\n    def coerce_dtype(\n        self,\n        check_obj,\n        schema=None,\n    ):\n        if hasattr(check_obj, \"pandera\"):\n            check_obj = check_obj.pandera.add_schema(schema)\n\n        check_obj = super().coerce_dtype(check_obj, schema=schema)\n\n        if hasattr(check_obj, \"pandera\"):\n            check_obj = check_obj.pandera.add_schema(schema)\n        return check_obj\n", "path": "pandera/backends/pandas/array.py"}], "after_files": [{"content": "\"\"\"Pandera array backends.\"\"\"\n\nfrom typing import cast, List, Optional\n\nimport pandas as pd\nfrom multimethod import DispatchError\nfrom pandera.api.base.error_handler import ErrorHandler\n\nfrom pandera.api.pandas.types import is_field\nfrom pandera.backends.base import CoreCheckResult\nfrom pandera.backends.pandas.base import PandasSchemaBackend\nfrom pandera.backends.pandas.error_formatters import (\n    reshape_failure_cases,\n    scalar_failure_case,\n)\nfrom pandera.backends.utils import convert_uniquesettings\nfrom pandera.config import ValidationScope\nfrom pandera.engines.pandas_engine import Engine\nfrom pandera.errors import (\n    ParserError,\n    SchemaError,\n    SchemaErrorReason,\n    SchemaErrors,\n    SchemaDefinitionError,\n)\nfrom pandera.validation_depth import (\n    validation_type,\n    validate_scope,\n)\n\n\nclass ArraySchemaBackend(PandasSchemaBackend):\n    \"\"\"Backend for pandas arrays.\"\"\"\n\n    def preprocess(self, check_obj, inplace: bool = False):\n        return check_obj if inplace else check_obj.copy()\n\n    def validate(\n        self,\n        check_obj,\n        schema,\n        *,\n        head: Optional[int] = None,\n        tail: Optional[int] = None,\n        sample: Optional[int] = None,\n        random_state: Optional[int] = None,\n        lazy: bool = False,\n        inplace: bool = False,\n    ):\n        # pylint: disable=too-many-locals\n        error_handler = ErrorHandler(lazy)\n        check_obj = self.preprocess(check_obj, inplace)\n\n        if getattr(schema, \"drop_invalid_rows\", False) and not lazy:\n            raise SchemaDefinitionError(\n                \"When drop_invalid_rows is True, lazy must be set to True.\"\n            )\n\n        # fill nans with `default` if it's present\n        if hasattr(schema, \"default\") and schema.default is not None:\n            check_obj = self.set_default(check_obj, schema)\n\n        try:\n            if is_field(check_obj) and schema.coerce:\n                check_obj = self.coerce_dtype(check_obj, schema=schema)\n            elif schema.coerce:\n                check_obj[schema.name] = self.coerce_dtype(\n                    check_obj[schema.name], schema=schema\n                )\n        except SchemaError as exc:\n            error_handler.collect_error(\n                validation_type(exc.reason_code),\n                exc.reason_code,\n                exc,\n            )\n\n        # run the core checks\n        error_handler = self.run_checks_and_handle_errors(\n            error_handler,\n            schema,\n            check_obj,\n            head=head,\n            tail=tail,\n            sample=sample,\n            random_state=random_state,\n        )\n\n        if lazy and error_handler.collected_errors:\n            if getattr(schema, \"drop_invalid_rows\", False):\n                check_obj = self.drop_invalid_rows(check_obj, error_handler)\n            else:\n                raise SchemaErrors(\n                    schema=schema,\n                    schema_errors=error_handler.schema_errors,\n                    data=check_obj,\n                )\n\n        return check_obj\n\n    def run_checks_and_handle_errors(\n        self, error_handler, schema, check_obj, **subsample_kwargs\n    ):\n        \"\"\"Run checks on schema\"\"\"\n        # pylint: disable=too-many-locals\n        field_obj_subsample = self.subsample(\n            check_obj if is_field(check_obj) else check_obj[schema.name],\n            **subsample_kwargs,\n        )\n\n        check_obj_subsample = self.subsample(check_obj, **subsample_kwargs)\n\n        core_checks = [\n            (self.check_name, (field_obj_subsample, schema)),\n            (self.check_nullable, (field_obj_subsample, schema)),\n            (self.check_unique, (field_obj_subsample, schema)),\n            (self.check_dtype, (field_obj_subsample, schema)),\n            (self.run_checks, (check_obj_subsample, schema)),\n        ]\n\n        for core_check, args in core_checks:\n            results = core_check(*args)\n            if isinstance(results, CoreCheckResult):\n                results = [results]\n            results = cast(List[CoreCheckResult], results)\n            for result in results:\n                if result.passed:\n                    continue\n\n                if result.schema_error is not None:\n                    error = result.schema_error\n                else:\n                    error = SchemaError(\n                        schema=schema,\n                        data=check_obj,\n                        message=result.message,\n                        failure_cases=result.failure_cases,\n                        check=result.check,\n                        check_index=result.check_index,\n                        check_output=result.check_output,\n                        reason_code=result.reason_code,\n                    )\n                    error_handler.collect_error(\n                        validation_type(result.reason_code),\n                        result.reason_code,\n                        error,\n                        original_exc=result.original_exc,\n                    )\n\n        return error_handler\n\n    def coerce_dtype(\n        self,\n        check_obj,\n        schema=None,\n        # pylint: disable=unused-argument\n    ):\n        \"\"\"Coerce type of a pd.Series by type specified in dtype.\n\n        :param pd.Series series: One-dimensional ndarray with axis labels\n            (including time series).\n        :returns: ``Series`` with coerced data type\n        \"\"\"\n        assert schema is not None, \"The `schema` argument must be provided.\"\n        if schema.dtype is None or not schema.coerce:\n            return check_obj\n\n        try:\n            return schema.dtype.try_coerce(check_obj)\n        except ParserError as exc:\n            raise SchemaError(\n                schema=schema,\n                data=check_obj,\n                message=(\n                    f\"Error while coercing '{schema.name}' to type \"\n                    f\"{schema.dtype}: {exc}:\\n{exc.failure_cases}\"\n                ),\n                failure_cases=exc.failure_cases,\n                check=f\"coerce_dtype('{schema.dtype}')\",\n                reason_code=SchemaErrorReason.DATATYPE_COERCION,\n            ) from exc\n\n    @validate_scope(scope=ValidationScope.SCHEMA)\n    def check_name(self, check_obj: pd.Series, schema) -> CoreCheckResult:\n        return CoreCheckResult(\n            passed=schema.name is None or check_obj.name == schema.name,\n            check=f\"field_name('{schema.name}')\",\n            reason_code=SchemaErrorReason.WRONG_FIELD_NAME,\n            message=(\n                f\"Expected {type(check_obj)} to have name '{schema.name}', \"\n                f\"found '{check_obj.name}'\"\n            ),\n            failure_cases=scalar_failure_case(check_obj.name),\n        )\n\n    @validate_scope(scope=ValidationScope.SCHEMA)\n    def check_nullable(self, check_obj: pd.Series, schema) -> CoreCheckResult:\n        if schema.nullable:\n            # Avoid to compute anything for perf reasons. GH#1533\n            return CoreCheckResult(\n                passed=True,\n                check=\"not_nullable\",\n            )\n\n        # Check actual column contents\n        isna = check_obj.isna()\n        passed = not isna.any()\n        return CoreCheckResult(\n            passed=cast(bool, passed),\n            check=\"not_nullable\",\n            reason_code=SchemaErrorReason.SERIES_CONTAINS_NULLS,\n            message=(\n                f\"non-nullable series '{check_obj.name}' contains \"\n                f\"null values:\\n{check_obj[isna]}\"\n            ),\n            failure_cases=reshape_failure_cases(\n                check_obj[isna], ignore_na=False\n            ),\n        )\n\n    @validate_scope(scope=ValidationScope.DATA)\n    def check_unique(self, check_obj: pd.Series, schema) -> CoreCheckResult:\n        passed = True\n        failure_cases = None\n        message = None\n\n        if schema.unique:\n            keep_argument = convert_uniquesettings(schema.report_duplicates)\n            if type(check_obj).__module__.startswith(\"pyspark.pandas\"):\n                # pylint: disable=import-outside-toplevel\n                import pyspark.pandas as ps\n\n                duplicates = (\n                    check_obj.to_frame()  # type: ignore\n                    .duplicated(keep=keep_argument)  # type: ignore\n                    .reindex(check_obj.index)\n                )\n                with ps.option_context(\"compute.ops_on_diff_frames\", True):\n                    failed = check_obj[duplicates]\n            else:\n                duplicates = check_obj.duplicated(keep=keep_argument)  # type: ignore\n                failed = check_obj[duplicates]\n\n            if duplicates.any():\n                passed = False\n                failure_cases = reshape_failure_cases(failed)\n                message = (\n                    f\"series '{check_obj.name}' contains duplicate \"\n                    f\"values:\\n{failed}\"\n                )\n\n        return CoreCheckResult(\n            passed=passed,\n            check=\"field_uniqueness\",\n            reason_code=SchemaErrorReason.SERIES_CONTAINS_DUPLICATES,\n            message=message,\n            failure_cases=failure_cases,\n        )\n\n    @validate_scope(scope=ValidationScope.SCHEMA)\n    def check_dtype(self, check_obj: pd.Series, schema) -> CoreCheckResult:\n        passed = True\n        failure_cases = None\n        msg = None\n\n        if schema.dtype is not None:\n            dtype_check_results = schema.dtype.check(\n                Engine.dtype(check_obj.dtype),\n                check_obj,\n            )\n            if isinstance(dtype_check_results, bool):\n                passed = dtype_check_results\n                failure_cases = scalar_failure_case(str(check_obj.dtype))\n                msg = (\n                    f\"expected series '{check_obj.name}' to have type \"\n                    f\"{schema.dtype}, got {check_obj.dtype}\"\n                )\n            else:\n                passed = dtype_check_results.all()\n                failure_cases = reshape_failure_cases(\n                    check_obj[~dtype_check_results.astype(bool)],\n                    ignore_na=False,\n                )\n                msg = (\n                    f\"expected series '{check_obj.name}' to have type \"\n                    f\"{schema.dtype}:\\nfailure cases:\\n{failure_cases}\"\n                )\n\n        return CoreCheckResult(\n            passed=passed,\n            check=f\"dtype('{schema.dtype}')\",\n            reason_code=SchemaErrorReason.WRONG_DATATYPE,\n            message=msg,\n            failure_cases=failure_cases,\n        )\n\n    @validate_scope(scope=ValidationScope.DATA)\n    def run_checks(self, check_obj, schema) -> List[CoreCheckResult]:\n        check_results: List[CoreCheckResult] = []\n        for check_index, check in enumerate(schema.checks):\n            check_args = [None] if is_field(check_obj) else [schema.name]\n            try:\n                check_results.append(\n                    self.run_check(\n                        check_obj,\n                        schema,\n                        check,\n                        check_index,\n                        *check_args,\n                    )\n                )\n            except Exception as err:  # pylint: disable=broad-except\n                # catch other exceptions that may occur when executing the Check\n                if isinstance(err, DispatchError):\n                    # if the error was raised by a check registered via\n                    # multimethod, get the underlying __cause__\n                    err = err.__cause__\n                err_msg = f'\"{err.args[0]}\"' if len(err.args) > 0 else \"\"\n                msg = f\"{err.__class__.__name__}({err_msg})\"\n                check_results.append(\n                    CoreCheckResult(\n                        passed=False,\n                        check=check,\n                        check_index=check_index,\n                        reason_code=SchemaErrorReason.CHECK_ERROR,\n                        message=msg,\n                        failure_cases=scalar_failure_case(msg),\n                        original_exc=err,\n                    )\n                )\n        return check_results\n\n    def set_default(self, check_obj, schema):\n        \"\"\"Sets the ``schema.default`` value on the ``check_obj``\"\"\"\n        # Ignore sparse dtype as it can't assign default value directly\n        if is_field(check_obj) and not isinstance(\n            check_obj.dtype, pd.SparseDtype\n        ):\n            check_obj = check_obj.fillna(schema.default)\n        elif not is_field(check_obj) and not isinstance(\n            check_obj[schema.name].dtype, pd.SparseDtype\n        ):\n            check_obj[schema.name] = check_obj[schema.name].fillna(\n                schema.default\n            )\n\n        return check_obj\n\n\nclass SeriesSchemaBackend(ArraySchemaBackend):\n    \"\"\"Backend for pandas Series objects.\"\"\"\n\n    def coerce_dtype(\n        self,\n        check_obj,\n        schema=None,\n    ):\n        if hasattr(check_obj, \"pandera\"):\n            check_obj = check_obj.pandera.add_schema(schema)\n\n        check_obj = super().coerce_dtype(check_obj, schema=schema)\n\n        if hasattr(check_obj, \"pandera\"):\n            check_obj = check_obj.pandera.add_schema(schema)\n        return check_obj\n", "path": "pandera/backends/pandas/array.py"}]}