Fix duplicate filtering path in Arrow task batches

[GayathriSrividya]

Closes #3272

What this changes

This PR updates the Arrow scan path in _task_to_record_batches to avoid redundant filtering when there are no positional deletes.

• Keeps predicate pushdown in Scanner.from_fragment as the only filter path when positional_deletes is absent.
• Applies current_batch.filter(pyarrow_filter) only in the positional-delete path, after deletes are applied.
• Preserves empty-batch handling after both delete application and conditional filtering.

Why

The previous flow could perform an extra table-level refilter even when the scanner already applied the predicate. This change removes that stale workaround path while keeping correct behavior for positional delete scenarios.

Tests

Added regression coverage in tests/io/test_pyarrow.py:

• test_task_to_record_batches_filter_without_positional_deletes_avoids_table_refilter
• test_task_to_record_batches_filter_with_positional_deletes_handles_empty_batch

Validated locally:

• python -m pytest tests/io/test_pyarrow.py -q -k "task_to_record_batches_nanos or filter_without_positional_deletes_avoids_table_refilter or filter_with_positional_deletes_handles_empty_batch"
• make lint

[Fokko]

This is a great catch @GayathriSrividya This logic must have been messed up somewhere throughout refactoring.

I left one comment to avoid having two blocks for positional_deletes, let me know what you think. This will reduce the number of lines quite a bit.

[Fokko]

Amazing, thanks @GayathriSrividya 🙌

[ndrluis]

This change replaces the Table-based filtering workaround.

PyArrow 17-20 raise IndexError when RecordBatch.filter(Expression) produces zero rows. PyArrow 21 fixed this behavior, but we still declares support for pyarrow>=17.0.0.

On PyArrow 17-20 this fails with IndexError: list index out of range

We should either keep that workaround for supported PyArrow versions or raise the minimum supported version to pyarrow>=21.0.0.

A regression test should cover the positional-delete path where the post-delete row filter produces an empty result, for example:

def test_task_to_record_batches_filter_after_positional_deletes_empty_result(tmpdir: str) -> None:
    from pyiceberg.expressions.visitors import bind

    arrow_schema = pa.schema(
        (
            pa.field(
                "id",
                pa.int32(),
                nullable=True,
                metadata={PYARROW_PARQUET_FIELD_ID_KEY: "1"},
            ),
        )
    )
    arrow_table = pa.table([pa.array([1, 2, 3], type=pa.int32())], schema=arrow_schema)
    data_file = _write_table_to_data_file(
        f"{tmpdir}/test_filter_after_positional_deletes_empty_result.parquet",
        arrow_schema,
        arrow_table,
    )

    table_schema = Schema(NestedField(1, "id", IntegerType(), required=False))
    positional_deletes = [pa.chunked_array([pa.array([], type=pa.int64())])]

    result_batches = list(
        _task_to_record_batches(
            PyArrowFileIO(),
            FileScanTask(data_file),
            bound_row_filter=bind(table_schema, GreaterThan("id", 10), case_sensitive=True),
            projected_schema=table_schema,
            table_schema=table_schema,
            projected_field_ids={1},
            positional_deletes=positional_deletes,
            case_sensitive=True,
        )
    )

    assert result_batches == []

[Fokko]

How about?

Suggested change

	if pyarrow_filter is not None:
	try:
	current_batch = current_batch.filter(pyarrow_filter)
	except IndexError:
	# PyArrow < 21 raises IndexError when filter produces zero rows
	# (fixed in https://github.com/apache/arrow/pull/46057)
	current_batch = current_batch.slice(0, 0)
	if pyarrow_filter is not None and len(current_batch) > 0:
	current_batch = current_batch.filter(pyarrow_filter)