Fix post-filter to use task.residual and preserve dictionary encoding

Two fixes for pluggable backend regressions:

1. Post-filter now uses task.residual instead of row_filter
   - The residual is the part of the filter not handled by partition pruning
   - If residual is AlwaysTrue, partition pruning handled everything (no post-filter needed)
   - The original row_filter may reference columns not in projected schema
   - Fixes test_partitioned_tables which filters by 'ts' but only selects 'number'

2. Pass dictionary_columns to PyArrow ParquetFileFormat
   - The PyArrow backend wasn't passing dictionary_columns to the parquet reader
   - Without this, dictionary encoding was silently dropped
   - Fixes test_to_arrow_batch_reader_preserves_dictionary_columns

Author: qzyu999

pyiceberg/execution/_orchestrate.py

@@ -343,29 +343,34 @@ def _execute_task(task: FileScanTask) -> list[pa.RecordBatch]:
         if puffin_positions:
             result_batches = _apply_puffin_positions(result_batches, puffin_positions)
 
-        # Post-filter guarantees correctness; read_parquet pushdown is best-effort only.
-        # Use the row_filter parameter (not task.residual) for post-filtering.
-        # This allows callers like Transaction.delete's _read_live_rows to override
-        # the task's residual (e.g., pass AlwaysTrue() to read all rows).
+        # Post-filter uses the task's residual, NOT the original row_filter.
+        # The residual is what remains after partition pruning - it only references
+        # columns that are either:
+        # 1. In the projected schema (user selected them)
+        # 2. Partition columns projected via identity transform
         #
-        # The residual from ManifestGroupPlanner may contain unbound predicates
-        # (e.g., for unpartitioned tables or predicates not involving partition columns).
-        # We must bind to the projected schema before converting to PyArrow expression.
-        # However, some residuals may already be bound (from tests or REST scan planning),
-        # so we catch the TypeError from bind() and use the residual as-is.
+        # If residual is AlwaysTrue, partition pruning handled the entire filter
+        # and no row-level filtering is needed. If the original row_filter references
+        # columns not in the projection, the residual should be AlwaysTrue (because
+        # partition pruning eliminated all non-matching files) or the filter should
+        # have been pushed down during read_parquet.
         #
         # Post-filter happens AFTER reconciliation so that:
         # - Projected partition columns are available for filtering
-        # - Renamed columns use their current names
-        if not isinstance(row_filter, AlwaysTrue):
+        # - Renamed columns use their current names (for schema evolution)
+        if not isinstance(task.residual, AlwaysTrue):
             from pyiceberg.expressions.visitors import bind
 
             try:
-                bound_filter = bind(projected_schema, row_filter, case_sensitive)
-            except TypeError:
-                # Predicate is already bound
-                bound_filter = row_filter
-            result_batches = list(backends.compute.filter(iter(result_batches), bound_filter))
+                bound_filter = bind(projected_schema, task.residual, case_sensitive)
+            except (TypeError, ValueError):
+                # Predicate is already bound, or references columns not in projected schema.
+                # The latter can happen if the filter references non-partition columns that
+                # weren't selected - in this case we skip post-filter and rely on pushdown.
+                bound_filter = None
+
+            if bound_filter is not None:
+                result_batches = list(backends.compute.filter(iter(result_batches), bound_filter))
 
         return result_batches
 

pyiceberg/execution/backends/pyarrow_backend.py

@@ -253,7 +253,14 @@ def read_parquet(
                 pa_filter = None
 
         filesystem, path = _resolve_filesystem(location, io_properties)
-        dataset = ds.dataset(path, format="parquet", filesystem=filesystem)
+
+        # Configure parquet format with dictionary columns if specified
+        format_kwargs: dict[str, Any] = {}
+        if dictionary_columns:
+            format_kwargs["dictionary_columns"] = list(dictionary_columns)
+        parquet_format = ds.ParquetFileFormat(**format_kwargs) if format_kwargs else "parquet"
+
+        dataset = ds.dataset(path, format=parquet_format, filesystem=filesystem)
 
         # Only request columns that exist in the file. Missing columns (from schema
         # evolution) will be filled with NULLs by the schema reconciliation layer in