Fix: correct regex pattern in _extract_filter_columns for partition column safeguard

The regex pattern was looking for FieldRef.Name(x) format but PyArrow expressions
actually stringify as (column_name == value). This caused the safeguard to return
an empty set, making it always pass (empty set is subset of any set), which then
pushed partition column filters to the scanner where they crashed.

Changes:
- Fixed regex to match actual PyArrow expression format: (column == value)
- Added patterns for function-style expressions (is_in, is_null, etc.)
- Added 7 unit tests for _extract_filter_columns

Fixes test_identity_transform_column_projection and test_migrate_table failures.

Author: qzyu999

pyiceberg/execution/backends/pyarrow_backend.py

@@ -294,8 +294,12 @@ def read_parquet(
 def _extract_filter_columns(pa_filter: pc.Expression) -> set[str]:
     """Extract column names referenced by a PyArrow compute expression.
 
-    Uses string representation parsing since PyArrow doesn't expose a clean API
-    for expression introspection.
+    Parses the string representation of the expression to find column names.
+    PyArrow expressions have string formats like:
+    - (column_name == value) for comparisons
+    - ((a == 1) and (b == 2)) for compound expressions
+    - is_in(column_name, ...) for isin expressions
+    - is_null(column_name, ...) for null checks
 
     Args:
         pa_filter: A PyArrow compute expression.
@@ -307,9 +311,17 @@ def _extract_filter_columns(pa_filter: pc.Expression) -> set[str]:
 
     columns: set[str] = set()
     expr_str = str(pa_filter)
-    # Find all field_ref('name') patterns in the expression string
-    for match in re.finditer(r"field_ref\('([^']+)'", expr_str):
+
+    # Pattern 1: column before comparison operators (==, !=, <, >, <=, >=)
+    # Matches: "column_name ==" or "column_name <" etc.
+    for match in re.finditer(r"([a-zA-Z_][a-zA-Z0-9_]*)\s*(?:==|!=|<=|>=|<|>)", expr_str):
+        columns.add(match.group(1))
+
+    # Pattern 2: function calls with column as first argument
+    # Matches: "is_in(column_name," or "is_null(column_name," etc.
+    for match in re.finditer(r"(?:is_in|is_null|is_nan|is_valid)\(([a-zA-Z_][a-zA-Z0-9_]*)", expr_str):
         columns.add(match.group(1))
+
     return columns
 
 

tests/execution/test_pyarrow_backend.py

@@ -21,6 +21,7 @@
 - _anti_join_tables struct-array O(n+m) multi-column join correctness
 - NULL handling with IS NOT DISTINCT FROM semantics
 - Full InMemoryCatalog round-trip through pluggable backend
+- _extract_filter_columns regex correctness for filter column safeguard
 """
 
 from __future__ import annotations
@@ -38,6 +39,75 @@
 from pyiceberg.schema import Schema
 from pyiceberg.types import IntegerType, LongType, NestedField, StringType
 
+# =============================================================================
+# _extract_filter_columns correctness (for partition column safeguard)
+# =============================================================================
+
+
+class TestExtractFilterColumns:
+    """_extract_filter_columns must correctly extract column names from PyArrow expressions.
+
+    This is critical for the safeguard in read_parquet() that prevents pushing filters
+    to the scanner when they reference columns not in the file (e.g., partition columns).
+    """
+
+    def test_simple_equality(self) -> None:
+        """Extract column from simple equality expression."""
+        from pyiceberg.execution.backends.pyarrow_backend import _extract_filter_columns
+
+        expr = pc.field("partition_id") == 1
+        columns = _extract_filter_columns(expr)
+        assert columns == {"partition_id"}
+
+    def test_simple_comparison(self) -> None:
+        """Extract column from comparison expression."""
+        from pyiceberg.execution.backends.pyarrow_backend import _extract_filter_columns
+
+        expr = pc.field("id") > 5
+        columns = _extract_filter_columns(expr)
+        assert columns == {"id"}
+
+    def test_multiple_columns_and(self) -> None:
+        """Extract multiple columns from AND expression."""
+        from pyiceberg.execution.backends.pyarrow_backend import _extract_filter_columns
+
+        expr = (pc.field("a") == 1) & (pc.field("b") == 2)
+        columns = _extract_filter_columns(expr)
+        assert columns == {"a", "b"}
+
+    def test_multiple_columns_or(self) -> None:
+        """Extract multiple columns from OR expression."""
+        from pyiceberg.execution.backends.pyarrow_backend import _extract_filter_columns
+
+        expr = (pc.field("x") < 10) | (pc.field("y") > 20)
+        columns = _extract_filter_columns(expr)
+        assert columns == {"x", "y"}
+
+    def test_complex_nested(self) -> None:
+        """Extract columns from complex nested expression."""
+        from pyiceberg.execution.backends.pyarrow_backend import _extract_filter_columns
+
+        expr = ((pc.field("a") == 1) & (pc.field("b") == 2)) | (pc.field("c") > 3)
+        columns = _extract_filter_columns(expr)
+        assert columns == {"a", "b", "c"}
+
+    def test_same_column_multiple_times(self) -> None:
+        """Same column referenced multiple times returns single entry."""
+        from pyiceberg.execution.backends.pyarrow_backend import _extract_filter_columns
+
+        expr = (pc.field("id") > 5) & (pc.field("id") < 10)
+        columns = _extract_filter_columns(expr)
+        assert columns == {"id"}
+
+    def test_isin_expression(self) -> None:
+        """Extract column from is_in expression."""
+        from pyiceberg.execution.backends.pyarrow_backend import _extract_filter_columns
+
+        expr = pc.field("category").isin(["a", "b", "c"])
+        columns = _extract_filter_columns(expr)
+        assert columns == {"category"}
+
+
 # =============================================================================
 # Multi-column anti-join correctness (O(n+m) struct-array approach)
 # =============================================================================