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 🙌

[kevinjqliu]

LGTM, the refactor makes sense.

pyarrow_filter is already applied for the none positional delete path:

iceberg-python/pyiceberg/io/pyarrow.py

Lines 1663 to 1697 in fb96df1

	fragment_scanner = ds.Scanner.from_fragment(
	fragment=fragment,
	schema=physical_schema,
	# This will push down the query to Arrow.
	# But in case there are positional deletes, we have to apply them first
	filter=pyarrow_filter if not positional_deletes else None,
	columns=[col.name for col in file_project_schema.columns],
	)
	next_index = 0
	batches = fragment_scanner.to_batches()
	for batch in batches:
	next_index = next_index + len(batch)
	current_index = next_index - len(batch)
	current_batch = batch
	if positional_deletes:
	# Create the mask of indices that we're interested in
	indices = _combine_positional_deletes(positional_deletes, current_index, current_index + len(batch))
	current_batch = current_batch.take(indices)
	# skip empty batches
	if current_batch.num_rows == 0:
	continue
	# Apply the user filter
	if pyarrow_filter is not None:
	# Temporary fix until PyArrow 21 is released ( https://github.com/apache/arrow/pull/46057 )
	table = pa.Table.from_batches([current_batch])
	table = table.filter(pyarrow_filter)
	# skip empty batches
	if table.num_rows == 0:
	continue
	current_batch = table.combine_chunks().to_batches()[0]

Looks like it was accidentally added to the non positional delete path in https://github.com/apache/iceberg-python/pull/1621/changes#diff-8d5e63f2a87ead8cebe2fd8ac5dcf2198d229f01e16bb9e06e21f7277c328abdR1401-R1423

[kevinjqliu]

Thanks for the PR @GayathriSrividya and thank you everyone for the reviews!