Repository: pandas-dev/pandas (python)

Issue:
ENH: Pandas 2.0 with pyarrow engine add the argument like 'skip_bad_lines=True'
### Feature Type

- [X] Adding new functionality to pandas

- [ ] Changing existing functionality in pandas

- [ ] Removing existing functionality in pandas


### Problem Description

When I use Pandas 2.0 with pyarrow engine to read CSV file, if the file get some bad lines, it will raise a error such as "pyarrow.lib.ArrowInvalid: CSV parse error: Expected 1 columns, got 2: ". 

Then I added the argument '', it will raise another error "TypeError: read_csv() got an unexpected keyword argument 'skip_bad_lines'"

So,can you add the argument like 'skip_bad_lines=True' with pyarrow engine.



```
df_arrow = pd.read_csv(r"VBAP.CSV", engine='pyarrow', dtype_backend='pyarrow')
df_arrow = pd.read_csv(r"VBAP.CSV", engine='pyarrow', dtype_backend='pyarrow', skip_bad_lines=True)
```

### Feature Description

df_arrow = pd.read_csv(r"VBAP.CSV", engine='pyarrow', dtype_backend='pyarrow', skip_bad_lines=True)

### Alternative Solutions

df_arrow = pd.read_csv(r"VBAP.CSV", engine='pyarrow', dtype_backend='pyarrow', skip_bad_lines=True)

### Additional Context

_No response_

Pull request description:
ENH: Add on_bad_lines for pyarrow
- [x] closes #54480 
- [x] All [code checks passed](https://pandas.pydata.org/pandas-docs/dev/development/contributing_codebase.html#pre-commit).
- [x] Update documentation

This adds the `on_bad_lines` argument to the `pyarrow` engine for the `from_csv` parser that closely follows the behaviour of the `python` engine. Internally utilizes [pyarrow's invalid_row_handler](https://arrow.apache.org/docs/python/generated/pyarrow.csv.ParseOptions.html#pyarrow.csv.ParseOptions.invalid_row_handler). The built-in callable implementation slightly differs for pyarrow, so the difference is appropriately documented by pointing to pyarrow's documentation.

Usage Example:

example.csv:
```
a,b,c
acol1,bcol1,ccol1
acol2,ccol2
```

example.py 
```
df_arrow = pd.read_csv(r"example.csv", engine='pyarrow', dtype_backend='pyarrow', on_bad_lines='warn')
print(df_arrow)
```

Console output
```
ParserWarning: Expected 3 columns, but found 2: acol2,ccol2
       a      b      c
0  acol1  bcol1  ccol1
```

Patch:
diff --git a/doc/source/conf.py b/doc/source/conf.py
index accbff596b..86d2494707 100644
--- a/doc/source/conf.py
+++ b/doc/source/conf.py
@@ -162,7 +162,7 @@ master_doc = "index"
 # General information about the project.
 project = "pandas"
 # We have our custom "pandas_footer.html" template, using copyright for the current year
-copyright = f"{datetime.now().year}"
+copyright = f"{datetime.now().year},"
 
 # The version info for the project you're documenting, acts as replacement for
 # |version| and |release|, also used in various other places throughout the
diff --git a/doc/source/whatsnew/v2.1.0.rst b/doc/source/whatsnew/v2.1.0.rst
index d422fb1be9..51b4c4f297 100644
--- a/doc/source/whatsnew/v2.1.0.rst
+++ b/doc/source/whatsnew/v2.1.0.rst
@@ -789,6 +789,7 @@ Plotting
 Groupby/resample/rolling
 ^^^^^^^^^^^^^^^^^^^^^^^^
 - Bug in :meth:`.DataFrameGroupBy.idxmin`, :meth:`.SeriesGroupBy.idxmin`, :meth:`.DataFrameGroupBy.idxmax`, :meth:`.SeriesGroupBy.idxmax` returns wrong dtype when used on an empty DataFrameGroupBy or SeriesGroupBy (:issue:`51423`)
+- Bug in :meth:`DataFrame.groupby.rank` on nullable datatypes when passing ``na_option="bottom"`` or ``na_option="top"`` (:issue:`54206`)
 - Bug in :meth:`DataFrame.resample` and :meth:`Series.resample` in incorrectly allowing non-fixed ``freq`` when resampling on a :class:`TimedeltaIndex` (:issue:`51896`)
 - Bug in :meth:`DataFrame.resample` and :meth:`Series.resample` losing time zone when resampling empty data (:issue:`53664`)
 - Bug in :meth:`DataFrame.resample` and :meth:`Series.resample` where ``origin`` has no effect in resample when values are outside of axis  (:issue:`53662`)
diff --git a/doc/source/whatsnew/v2.1.2.rst b/doc/source/whatsnew/v2.1.2.rst
index 2c9b10160d..97aeb56924 100644
--- a/doc/source/whatsnew/v2.1.2.rst
+++ b/doc/source/whatsnew/v2.1.2.rst
@@ -13,7 +13,7 @@ including other versions of pandas.
 
 Fixed regressions
 ~~~~~~~~~~~~~~~~~
--
+- Fixed bug where PDEP-6 warning about setting an item of an incompatible dtype was being shown when creating a new conditional column (:issue:`55025`)
 -
 
 .. ---------------------------------------------------------------------------
diff --git a/doc/source/whatsnew/v2.2.0.rst b/doc/source/whatsnew/v2.2.0.rst
index 8eab623a2b..930e03ae7d 100644
--- a/doc/source/whatsnew/v2.2.0.rst
+++ b/doc/source/whatsnew/v2.2.0.rst
@@ -73,6 +73,8 @@ enhancement2
 
 Other enhancements
 ^^^^^^^^^^^^^^^^^^
+
+- :func:`read_csv` now supports ``on_bad_lines`` parameter with ``engine="pyarrow"``. (:issue:`54480`)
 - :meth:`ExtensionArray._explode` interface method added to allow extension type implementations of the ``explode`` method (:issue:`54833`)
 - DataFrame.apply now allows the usage of numba (via ``engine="numba"``) to JIT compile the passed function, allowing for potential speedups (:issue:`54666`)
 -
@@ -238,7 +240,7 @@ Performance improvements
 - Performance improvement in :meth:`DataFrame.sort_index` and :meth:`Series.sort_index` when indexed by a :class:`MultiIndex` (:issue:`54835`)
 - Performance improvement in :meth:`Index.difference` (:issue:`55108`)
 - Performance improvement when indexing with more than 4 keys (:issue:`54550`)
--
+- Performance improvement when localizing time to UTC (:issue:`55241`)
 
 .. ---------------------------------------------------------------------------
 .. _whatsnew_220.bug_fixes:
diff --git a/pandas/_libs/tslibs/tzconversion.pyx b/pandas/_libs/tslibs/tzconversion.pyx
index e17d264333..a96779aa33 100644
--- a/pandas/_libs/tslibs/tzconversion.pyx
+++ b/pandas/_libs/tslibs/tzconversion.pyx
@@ -462,7 +462,7 @@ cdef str _render_tstamp(int64_t val, NPY_DATETIMEUNIT creso):
 
 
 cdef _get_utc_bounds(
-    ndarray vals,
+    ndarray[int64_t] vals,
     int64_t* tdata,
     Py_ssize_t ntrans,
     const int64_t[::1] deltas,
@@ -472,7 +472,7 @@ cdef _get_utc_bounds(
     # result_a) or right of the DST transition (store in result_b)
 
     cdef:
-        ndarray result_a, result_b
+        ndarray[int64_t] result_a, result_b
         Py_ssize_t i, n = vals.size
         int64_t val, v_left, v_right
         Py_ssize_t isl, isr, pos_left, pos_right
diff --git a/pandas/core/arrays/masked.py b/pandas/core/arrays/masked.py
index 2cf28c2842..fdcbe67bbc 100644
--- a/pandas/core/arrays/masked.py
+++ b/pandas/core/arrays/masked.py
@@ -1487,6 +1487,9 @@ class BaseMaskedArray(OpsMixin, ExtensionArray):
         else:
             result_mask = np.zeros(ngroups, dtype=bool)
 
+        if how == "rank" and kwargs.get("na_option") in ["top", "bottom"]:
+            result_mask[:] = False
+
         res_values = op._cython_op_ndim_compat(
             self._data,
             min_count=min_count,
diff --git a/pandas/core/dtypes/missing.py b/pandas/core/dtypes/missing.py
index 7117e34b23..8760c8eeca 100644
--- a/pandas/core/dtypes/missing.py
+++ b/pandas/core/dtypes/missing.py
@@ -624,6 +624,9 @@ def infer_fill_value(val):
             return np.array("NaT", dtype=DT64NS_DTYPE)
         elif dtype in ["timedelta", "timedelta64"]:
             return np.array("NaT", dtype=TD64NS_DTYPE)
+        return np.array(np.nan, dtype=object)
+    elif val.dtype.kind == "U":
+        return np.array(np.nan, dtype=val.dtype)
     return np.nan
 
 
diff --git a/pandas/core/generic.py b/pandas/core/generic.py
index c75c6c546a..d0fab0037d 100644
--- a/pandas/core/generic.py
+++ b/pandas/core/generic.py
@@ -12532,6 +12532,39 @@ Returns
 {examples}
 """
 
+_sum_prod_doc = """
+{desc}
+
+Parameters
+----------
+axis : {axis_descr}
+    Axis for the function to be applied on.
+    For `Series` this parameter is unused and defaults to 0.
+
+    .. warning::
+
+        The behavior of DataFrame.{name} with ``axis=None`` is deprecated,
+        in a future version this will reduce over both axes and return a scalar
+        To retain the old behavior, pass axis=0 (or do not pass axis).
+
+    .. versionadded:: 2.0.0
+
+skipna : bool, default True
+    Exclude NA/null values when computing the result.
+numeric_only : bool, default False
+    Include only float, int, boolean columns. Not implemented for Series.
+
+{min_count}\
+**kwargs
+    Additional keyword arguments to be passed to the function.
+
+Returns
+-------
+{name1} or scalar\
+{see_also}\
+{examples}
+"""
+
 _num_ddof_doc = """
 {desc}
 
@@ -12539,6 +12572,13 @@ Parameters
 ----------
 axis : {axis_descr}
     For `Series` this parameter is unused and defaults to 0.
+
+    .. warning::
+
+        The behavior of DataFrame.{name} with ``axis=None`` is deprecated,
+        in a future version this will reduce over both axes and return a scalar
+        To retain the old behavior, pass axis=0 (or do not pass axis).
+
 skipna : bool, default True
     Exclude NA/null values. If an entire row/column is NA, the result
     will be NA.
@@ -13246,7 +13286,7 @@ def make_doc(name: str, ndim: int) -> str:
         kwargs = {"min_count": ""}
 
     elif name == "sum":
-        base_doc = _num_doc
+        base_doc = _sum_prod_doc
         desc = (
             "Return the sum of the values over the requested axis.\n\n"
             "This is equivalent to the method ``numpy.sum``."
@@ -13256,7 +13296,7 @@ def make_doc(name: str, ndim: int) -> str:
         kwargs = {"min_count": _min_count_stub}
 
     elif name == "prod":
-        base_doc = _num_doc
+        base_doc = _sum_prod_doc
         desc = "Return the product of the values over the requested axis."
         see_also = _stat_func_see_also
         examples = _prod_examples
@@ -13540,6 +13580,7 @@ def make_doc(name: str, ndim: int) -> str:
 
     docstr = base_doc.format(
         desc=desc,
+        name=name,
         name1=name1,
         name2=name2,
         axis_descr=axis_descr,
diff --git a/pandas/io/json/_json.py b/pandas/io/json/_json.py
index ecab14a54b..d596891197 100644
--- a/pandas/io/json/_json.py
+++ b/pandas/io/json/_json.py
@@ -175,7 +175,7 @@ def to_json(
 
     if mode == "a" and (not lines or orient != "records"):
         msg = (
-            "mode='a' (append) is only supported when"
+            "mode='a' (append) is only supported when "
             "lines is True and orient is 'records'"
         )
         raise ValueError(msg)
diff --git a/pandas/io/parsers/arrow_parser_wrapper.py b/pandas/io/parsers/arrow_parser_wrapper.py
index bb6bcd3c4d..765a4ffcd2 100644
--- a/pandas/io/parsers/arrow_parser_wrapper.py
+++ b/pandas/io/parsers/arrow_parser_wrapper.py
@@ -1,11 +1,17 @@
 from __future__ import annotations
 
 from typing import TYPE_CHECKING
+import warnings
 
 from pandas._config import using_pyarrow_string_dtype
 
 from pandas._libs import lib
 from pandas.compat._optional import import_optional_dependency
+from pandas.errors import (
+    ParserError,
+    ParserWarning,
+)
+from pandas.util._exceptions import find_stack_level
 
 from pandas.core.dtypes.inference import is_integer
 
@@ -85,6 +91,30 @@ class ArrowParserWrapper(ParserBase):
             and option_name
             in ("delimiter", "quote_char", "escape_char", "ignore_empty_lines")
         }
+
+        on_bad_lines = self.kwds.get("on_bad_lines")
+        if on_bad_lines is not None:
+            if callable(on_bad_lines):
+                self.parse_options["invalid_row_handler"] = on_bad_lines
+            elif on_bad_lines == ParserBase.BadLineHandleMethod.ERROR:
+                self.parse_options[
+                    "invalid_row_handler"
+                ] = None  # PyArrow raises an exception by default
+            elif on_bad_lines == ParserBase.BadLineHandleMethod.WARN:
+
+                def handle_warning(invalid_row):
+                    warnings.warn(
+                        f"Expected {invalid_row.expected_columns} columns, but found "
+                        f"{invalid_row.actual_columns}: {invalid_row.text}",
+                        ParserWarning,
+                        stacklevel=find_stack_level(),
+                    )
+                    return "skip"
+
+                self.parse_options["invalid_row_handler"] = handle_warning
+            elif on_bad_lines == ParserBase.BadLineHandleMethod.SKIP:
+                self.parse_options["invalid_row_handler"] = lambda _: "skip"
+
         self.convert_options = {
             option_name: option_value
             for option_name, option_value in self.kwds.items()
@@ -190,12 +220,15 @@ class ArrowParserWrapper(ParserBase):
         pyarrow_csv = import_optional_dependency("pyarrow.csv")
         self._get_pyarrow_options()
 
-        table = pyarrow_csv.read_csv(
-            self.src,
-            read_options=pyarrow_csv.ReadOptions(**self.read_options),
-            parse_options=pyarrow_csv.ParseOptions(**self.parse_options),
-            convert_options=pyarrow_csv.ConvertOptions(**self.convert_options),
-        )
+        try:
+            table = pyarrow_csv.read_csv(
+                self.src,
+                read_options=pyarrow_csv.ReadOptions(**self.read_options),
+                parse_options=pyarrow_csv.ParseOptions(**self.parse_options),
+                convert_options=pyarrow_csv.ConvertOptions(**self.convert_options),
+            )
+        except pa.ArrowInvalid as e:
+            raise ParserError(e) from e
 
         dtype_backend = self.kwds["dtype_backend"]
 
diff --git a/pandas/io/parsers/readers.py b/pandas/io/parsers/readers.py
index acf35ebd6a..6ce6ac71b1 100644
--- a/pandas/io/parsers/readers.py
+++ b/pandas/io/parsers/readers.py
@@ -401,6 +401,13 @@ on_bad_lines : {{'error', 'warn', 'skip'}} or Callable, default 'error'
           expected, a ``ParserWarning`` will be emitted while dropping extra elements.
           Only supported when ``engine='python'``
 
+    .. versionchanged:: 2.2.0
+
+        - Callable, function with signature
+          as described in `pyarrow documentation
+          <https://arrow.apache.org/docs/python/generated/pyarrow.csv.ParseOptions.html
+          #pyarrow.csv.ParseOptions.invalid_row_handler>_` when ``engine='pyarrow'``
+
 delim_whitespace : bool, default False
     Specifies whether or not whitespace (e.g. ``' '`` or ``'\\t'``) will be
     used as the ``sep`` delimiter. Equivalent to setting ``sep='\\s+'``. If this option
@@ -494,7 +501,6 @@ _pyarrow_unsupported = {
     "thousands",
     "memory_map",
     "dialect",
-    "on_bad_lines",
     "delim_whitespace",
     "quoting",
     "lineterminator",
@@ -2142,9 +2148,10 @@ def _refine_defaults_read(
     elif on_bad_lines == "skip":
         kwds["on_bad_lines"] = ParserBase.BadLineHandleMethod.SKIP
     elif callable(on_bad_lines):
-        if engine != "python":
+        if engine not in ["python", "pyarrow"]:
             raise ValueError(
-                "on_bad_line can only be a callable function if engine='python'"
+                "on_bad_line can only be a callable function "
+                "if engine='python' or 'pyarrow'"
             )
         kwds["on_bad_lines"] = on_bad_lines
     else:


Test patch:
diff --git a/pandas/tests/frame/indexing/test_indexing.py b/pandas/tests/frame/indexing/test_indexing.py
index b324291bab..370cbf0f33 100644
--- a/pandas/tests/frame/indexing/test_indexing.py
+++ b/pandas/tests/frame/indexing/test_indexing.py
@@ -1890,6 +1890,21 @@ class TestDeprecatedIndexers:
             df.loc[key] = 1
 
 
+def test_adding_new_conditional_column() -> None:
+    # https://github.com/pandas-dev/pandas/issues/55025
+    df = DataFrame({"x": [1]})
+    df.loc[df["x"] == 1, "y"] = "1"
+    expected = DataFrame({"x": [1], "y": ["1"]})
+    tm.assert_frame_equal(df, expected)
+
+    df = DataFrame({"x": [1]})
+    # try inserting something which numpy would store as 'object'
+    value = lambda x: x
+    df.loc[df["x"] == 1, "y"] = value
+    expected = DataFrame({"x": [1], "y": [value]})
+    tm.assert_frame_equal(df, expected)
+
+
 class TestSetitemValidation:
     # This is adapted from pandas/tests/arrays/masked/test_indexing.py
     # but checks for warnings instead of errors.
diff --git a/pandas/tests/frame/indexing/test_set_value.py b/pandas/tests/frame/indexing/test_set_value.py
index cd9ffa0f12..32312868ad 100644
--- a/pandas/tests/frame/indexing/test_set_value.py
+++ b/pandas/tests/frame/indexing/test_set_value.py
@@ -26,10 +26,7 @@ class TestSetValue:
         assert float_frame._get_value("foobar", "qux") == 0
 
         res = float_frame.copy()
-        with tm.assert_produces_warning(
-            FutureWarning, match="Setting an item of incompatible dtype"
-        ):
-            res._set_value("foobar", "baz", "sam")
+        res._set_value("foobar", "baz", "sam")
         assert res["baz"].dtype == np.object_
 
         res = float_frame.copy()
diff --git a/pandas/tests/groupby/test_rank.py b/pandas/tests/groupby/test_rank.py
index 5d85a0783e..a3b7da3fa8 100644
--- a/pandas/tests/groupby/test_rank.py
+++ b/pandas/tests/groupby/test_rank.py
@@ -710,3 +710,12 @@ def test_rank_categorical():
 
     expected = df.astype(object).groupby("col1").rank()
     tm.assert_frame_equal(res, expected)
+
+
+@pytest.mark.parametrize("na_option", ["top", "bottom"])
+def test_groupby_op_with_nullables(na_option):
+    # GH 54206
+    df = DataFrame({"x": [None]}, dtype="Float64")
+    result = df.groupby("x", dropna=False)["x"].rank(method="min", na_option=na_option)
+    expected = Series([1.0], dtype="Float64", name=result.name)
+    tm.assert_series_equal(result, expected)
diff --git a/pandas/tests/io/json/test_readlines.py b/pandas/tests/io/json/test_readlines.py
index c2f915e33d..d7baba87bb 100644
--- a/pandas/tests/io/json/test_readlines.py
+++ b/pandas/tests/io/json/test_readlines.py
@@ -399,7 +399,7 @@ def test_to_json_append_orient(orient_):
     # Test ValueError when orient is not 'records'
     df = DataFrame({"col1": [1, 2], "col2": ["a", "b"]})
     msg = (
-        r"mode='a' \(append\) is only supported when"
+        r"mode='a' \(append\) is only supported when "
         "lines is True and orient is 'records'"
     )
     with pytest.raises(ValueError, match=msg):
@@ -411,7 +411,7 @@ def test_to_json_append_lines():
     # Test ValueError when lines is not True
     df = DataFrame({"col1": [1, 2], "col2": ["a", "b"]})
     msg = (
-        r"mode='a' \(append\) is only supported when"
+        r"mode='a' \(append\) is only supported when "
         "lines is True and orient is 'records'"
     )
     with pytest.raises(ValueError, match=msg):
diff --git a/pandas/tests/io/parser/common/test_read_errors.py b/pandas/tests/io/parser/common/test_read_errors.py
index 0c5a2e0d04..4e82dca83e 100644
--- a/pandas/tests/io/parser/common/test_read_errors.py
+++ b/pandas/tests/io/parser/common/test_read_errors.py
@@ -1,5 +1,5 @@
 """
-Tests that work on both the Python and C engines but do not have a
+Tests that work on the Python, C and PyArrow engines but do not have a
 specific classification into the other test modules.
 """
 import codecs
@@ -21,7 +21,8 @@ from pandas.errors import (
 from pandas import DataFrame
 import pandas._testing as tm
 
-pytestmark = pytest.mark.usefixtures("pyarrow_skip")
+xfail_pyarrow = pytest.mark.usefixtures("pyarrow_xfail")
+skip_pyarrow = pytest.mark.usefixtures("pyarrow_skip")
 
 
 def test_empty_decimal_marker(all_parsers):
@@ -33,10 +34,17 @@ def test_empty_decimal_marker(all_parsers):
     msg = "Only length-1 decimal markers supported"
     parser = all_parsers
 
+    if parser.engine == "pyarrow":
+        msg = (
+            "only single character unicode strings can be "
+            "converted to Py_UCS4, got length 0"
+        )
+
     with pytest.raises(ValueError, match=msg):
         parser.read_csv(StringIO(data), decimal="")
 
 
+@skip_pyarrow
 def test_bad_stream_exception(all_parsers, csv_dir_path):
     # see gh-13652
     #
@@ -57,6 +65,7 @@ def test_bad_stream_exception(all_parsers, csv_dir_path):
             parser.read_csv(stream)
 
 
+@skip_pyarrow
 def test_malformed(all_parsers):
     # see gh-6607
     parser = all_parsers
@@ -71,6 +80,7 @@ A,B,C
         parser.read_csv(StringIO(data), header=1, comment="#")
 
 
+@skip_pyarrow
 @pytest.mark.parametrize("nrows", [5, 3, None])
 def test_malformed_chunks(all_parsers, nrows):
     data = """ignore
@@ -90,6 +100,7 @@ skip
             reader.read(nrows)
 
 
+@skip_pyarrow
 def test_catch_too_many_names(all_parsers):
     # see gh-5156
     data = """\
@@ -109,6 +120,7 @@ def test_catch_too_many_names(all_parsers):
         parser.read_csv(StringIO(data), header=0, names=["a", "b", "c", "d"])
 
 
+@skip_pyarrow
 @pytest.mark.parametrize("nrows", [0, 1, 2, 3, 4, 5])
 def test_raise_on_no_columns(all_parsers, nrows):
     parser = all_parsers
@@ -147,6 +159,10 @@ def test_error_bad_lines(all_parsers):
     data = "a\n1\n1,2,3\n4\n5,6,7"
 
     msg = "Expected 1 fields in line 3, saw 3"
+
+    if parser.engine == "pyarrow":
+        msg = "CSV parse error: Expected 1 columns, got 3: 1,2,3"
+
     with pytest.raises(ParserError, match=msg):
         parser.read_csv(StringIO(data), on_bad_lines="error")
 
@@ -156,9 +172,13 @@ def test_warn_bad_lines(all_parsers):
     parser = all_parsers
     data = "a\n1\n1,2,3\n4\n5,6,7"
     expected = DataFrame({"a": [1, 4]})
+    match_msg = "Skipping line"
+
+    if parser.engine == "pyarrow":
+        match_msg = "Expected 1 columns, but found 3: 1,2,3"
 
     with tm.assert_produces_warning(
-        ParserWarning, match="Skipping line", check_stacklevel=False
+        ParserWarning, match=match_msg, check_stacklevel=False
     ):
         result = parser.read_csv(StringIO(data), on_bad_lines="warn")
     tm.assert_frame_equal(result, expected)
@@ -174,10 +194,14 @@ def test_read_csv_wrong_num_columns(all_parsers):
     parser = all_parsers
     msg = "Expected 6 fields in line 3, saw 7"
 
+    if parser.engine == "pyarrow":
+        msg = "Expected 6 columns, got 7: 6,7,8,9,10,11,12"
+
     with pytest.raises(ParserError, match=msg):
         parser.read_csv(StringIO(data))
 
 
+@skip_pyarrow
 def test_null_byte_char(request, all_parsers):
     # see gh-2741
     data = "\x00,foo"
@@ -200,6 +224,7 @@ def test_null_byte_char(request, all_parsers):
             parser.read_csv(StringIO(data), names=names)
 
 
+@skip_pyarrow
 @pytest.mark.filterwarnings("always::ResourceWarning")
 def test_open_file(request, all_parsers):
     # GH 39024
@@ -238,6 +263,8 @@ def test_bad_header_uniform_error(all_parsers):
             "Could not construct index. Requested to use 1 "
             "number of columns, but 3 left to parse."
         )
+    elif parser.engine == "pyarrow":
+        msg = "CSV parse error: Expected 1 columns, got 4: col1,col2,col3,col4"
 
     with pytest.raises(ParserError, match=msg):
         parser.read_csv(StringIO(data), index_col=0, on_bad_lines="error")
@@ -253,9 +280,13 @@ a,b,d
 a,b
 """
     expected = DataFrame({"1": "a", "2": ["b"] * 2})
+    match_msg = "Skipping line"
+
+    if parser.engine == "pyarrow":
+        match_msg = "Expected 2 columns, but found 3: a,b,c"
 
     with tm.assert_produces_warning(
-        ParserWarning, match="Skipping line", check_stacklevel=False
+        ParserWarning, match=match_msg, check_stacklevel=False
     ):
         result = parser.read_csv(StringIO(data), on_bad_lines="warn")
     tm.assert_frame_equal(result, expected)
diff --git a/pandas/tests/io/parser/test_unsupported.py b/pandas/tests/io/parser/test_unsupported.py
index ac17156818..b489c09e91 100644
--- a/pandas/tests/io/parser/test_unsupported.py
+++ b/pandas/tests/io/parser/test_unsupported.py
@@ -151,13 +151,17 @@ x   q   30      3    -0.6662 -0.5243 -0.3580  0.89145  2.5838"""
             with pytest.raises(ValueError, match=msg):
                 read_csv(StringIO(data), engine="pyarrow", **kwargs)
 
-    def test_on_bad_lines_callable_python_only(self, all_parsers):
+    def test_on_bad_lines_callable_python_or_pyarrow(self, all_parsers):
         # GH 5686
+        # GH 54643
         sio = StringIO("a,b\n1,2")
         bad_lines_func = lambda x: x
         parser = all_parsers
-        if all_parsers.engine != "python":
-            msg = "on_bad_line can only be a callable function if engine='python'"
+        if all_parsers.engine not in ["python", "pyarrow"]:
+            msg = (
+                "on_bad_line can only be a callable "
+                "function if engine='python' or 'pyarrow'"
+            )
             with pytest.raises(ValueError, match=msg):
                 parser.read_csv(sio, on_bad_lines=bad_lines_func)
         else: