You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "GayathriSrividya (via GitHub)" <gi...@apache.org> on 2026/06/01 06:13:59 UTC
[PR] Fix duplicate filtering path in Arrow task batches [iceberg-python]
GayathriSrividya opened a new pull request, #3448:
URL: https://github.com/apache/iceberg-python/pull/3448
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`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix duplicate filtering path in Arrow task batches [iceberg-python]
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko commented on code in PR #3448:
URL: https://github.com/apache/iceberg-python/pull/3448#discussion_r3366886708
##########
pyiceberg/io/pyarrow.py:
##########
@@ -1676,22 +1676,18 @@ def _task_to_record_batches(
# 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)
+ 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)
Review Comment:
How about?
```suggestion
if pyarrow_filter is not None and len(current_batch) > 0:
current_batch = current_batch.filter(pyarrow_filter)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix duplicate filtering path in Arrow task batches [iceberg-python]
Posted by "kevinjqliu (via GitHub)" <gi...@apache.org>.
kevinjqliu commented on PR #3448:
URL: https://github.com/apache/iceberg-python/pull/3448#issuecomment-4763786743
Thanks for the PR @GayathriSrividya and thank you everyone for the reviews!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix duplicate filtering path in Arrow task batches [iceberg-python]
Posted by "ndrluis (via GitHub)" <gi...@apache.org>.
ndrluis commented on code in PR #3448:
URL: https://github.com/apache/iceberg-python/pull/3448#discussion_r3359711150
##########
pyiceberg/io/pyarrow.py:
##########
@@ -1676,22 +1676,13 @@ def _task_to_record_batches(
# 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)
+ if pyarrow_filter is not None:
+ current_batch = current_batch.filter(pyarrow_filter)
Review Comment:
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:
```python
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 == []
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix duplicate filtering path in Arrow task batches [iceberg-python]
Posted by "ebyhr (via GitHub)" <gi...@apache.org>.
ebyhr commented on code in PR #3448:
URL: https://github.com/apache/iceberg-python/pull/3448#discussion_r3333367731
##########
tests/io/test_pyarrow.py:
##########
@@ -3267,6 +3267,59 @@ def _expected_batch(unit: str) -> pa.RecordBatch:
assert _expected_batch("ns" if format_version > 2 else "us").equals(actual_result)
+def test_task_to_record_batches_filter_without_positional_deletes_avoids_table_refilter(tmpdir: str) -> None:
+ 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_task_to_record_batches_filter_no_positional.parquet", arrow_schema, arrow_table
+ )
+
+ table_schema = Schema(NestedField(1, "id", IntegerType(), required=False))
+ from pyiceberg.expressions.visitors import bind
+
+ result_batches = list(
+ _task_to_record_batches(
+ PyArrowFileIO(),
+ FileScanTask(data_file),
+ bound_row_filter=bind(table_schema, GreaterThan("id", 1), case_sensitive=True),
+ projected_schema=table_schema,
+ table_schema=table_schema,
+ projected_field_ids={1},
+ positional_deletes=None,
+ case_sensitive=True,
+ )
+ )
+ assert len(result_batches) == 1
+ assert result_batches[0].column(0).to_pylist() == [2, 3]
+
+
+def test_task_to_record_batches_filter_with_positional_deletes_handles_empty_batch(tmpdir: str) -> None:
Review Comment:
These tests pass even after reverting changes in `pyarrow.py`. Could you please write a valid regression test?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix duplicate filtering path in Arrow task batches [iceberg-python]
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko commented on code in PR #3448:
URL: https://github.com/apache/iceberg-python/pull/3448#discussion_r3341674219
##########
pyiceberg/io/pyarrow.py:
##########
Review Comment:
For simplicity, I wonder if we just should combine this in the if block above.
```suggestion
current_batch = current_batch.take(indices).filter(pyarrow_filter)
```
A filter on an empty batch is free anyway
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix duplicate filtering path in Arrow task batches [iceberg-python]
Posted by "GayathriSrividya (via GitHub)" <gi...@apache.org>.
GayathriSrividya commented on code in PR #3448:
URL: https://github.com/apache/iceberg-python/pull/3448#discussion_r3371840170
##########
pyiceberg/io/pyarrow.py:
##########
@@ -1676,22 +1676,13 @@ def _task_to_record_batches(
# 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)
+ if pyarrow_filter is not None:
+ current_batch = current_batch.filter(pyarrow_filter)
Review Comment:
Thanks for the suggestion! We actually already have that exact regression test in the PR (test_task_to_record_batches_filter_after_positional_deletes_empty_result). You're right that the len(current_batch) > 0 guard wasn't sufficient — it only skips filtering when the batch is already empty before filtering, but doesn't protect against a non-empty batch filtering down to zero rows (which is the actual PyArrow <21 bug). I've switched back to the pa.Table.from_batches workaround as you suggested.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix duplicate filtering path in Arrow task batches [iceberg-python]
Posted by "kevinjqliu (via GitHub)" <gi...@apache.org>.
kevinjqliu merged PR #3448:
URL: https://github.com/apache/iceberg-python/pull/3448
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix duplicate filtering path in Arrow task batches [iceberg-python]
Posted by "GayathriSrividya (via GitHub)" <gi...@apache.org>.
GayathriSrividya commented on code in PR #3448:
URL: https://github.com/apache/iceberg-python/pull/3448#discussion_r3336422892
##########
tests/io/test_pyarrow.py:
##########
@@ -3267,6 +3267,59 @@ def _expected_batch(unit: str) -> pa.RecordBatch:
assert _expected_batch("ns" if format_version > 2 else "us").equals(actual_result)
+def test_task_to_record_batches_filter_without_positional_deletes_avoids_table_refilter(tmpdir: str) -> None:
+ 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_task_to_record_batches_filter_no_positional.parquet", arrow_schema, arrow_table
+ )
+
+ table_schema = Schema(NestedField(1, "id", IntegerType(), required=False))
+ from pyiceberg.expressions.visitors import bind
+
+ result_batches = list(
+ _task_to_record_batches(
+ PyArrowFileIO(),
+ FileScanTask(data_file),
+ bound_row_filter=bind(table_schema, GreaterThan("id", 1), case_sensitive=True),
+ projected_schema=table_schema,
+ table_schema=table_schema,
+ projected_field_ids={1},
+ positional_deletes=None,
+ case_sensitive=True,
+ )
+ )
+ assert len(result_batches) == 1
+ assert result_batches[0].column(0).to_pylist() == [2, 3]
+
+
+def test_task_to_record_batches_filter_with_positional_deletes_handles_empty_batch(tmpdir: str) -> None:
Review Comment:
Thanks for the feedback, @ebyhr! You're right — the original tests were passing on the unfixed code because re-filtering is idempotent (the scanner returns already-filtered rows, so applying the same predicate again gives the same result).
I've replaced both tests with proper regression coverage:
[test_task_to_record_batches_does_not_use_table_filter_without_positional_deletes](vscode-file://vscode-app/Applications/Visual%20Studio%20Code.app/Contents/Resources/app/out/vs/code/electron-browser/workbench/workbench.html)
Detects the actual regression: replaces [pyarrow.Table](vscode-file://vscode-app/Applications/Visual%20Studio%20Code.app/Contents/Resources/app/out/vs/code/electron-browser/workbench/workbench.html) with a sentinel class whose [from_batches](vscode-file://vscode-app/Applications/Visual%20Studio%20Code.app/Contents/Resources/app/out/vs/code/electron-browser/workbench/workbench.html) raises [AssertionError](vscode-file://vscode-app/Applications/Visual%20Studio%20Code.app/Contents/Resources/app/out/vs/code/electron-browser/workbench/workbench.html). A custom metaclass preserves [isinstance(x, pa.Table)](vscode-file://vscode-app/Applications/Visual%20Studio%20Code.app/Contents/Resources/app/out/vs/code/electron-browser/workbench/workbench.html) checks so the rest of the code-path is unaffected. With the fix the sentinel is never reached; without the fix it fires immediately — confirmed by running the test against the original [pyarrow.py](vscode-file://vscode-app/Applications/Visual
%20Studio%20Code.app/Contents/Resources/app/out/vs/code/electron-browser/workbench/workbench.html).
[test_task_to_record_batches_filter_applied_after_positional_deletes](vscode-file://vscode-app/Applications/Visual%20Studio%20Code.app/Contents/Resources/app/out/vs/code/electron-browser/workbench/workbench.html)
Functional test with actual positional deletes: data [1,2,3,4,5], delete positions 1 and 3 (values 2 and 4), then filter [id > 2](vscode-file://vscode-app/Applications/Visual%20Studio%20Code.app/Contents/Resources/app/out/vs/code/electron-browser/workbench/workbench.html). Expected [3, 5] — a wrong result if either positional deletes or the subsequent filter were skipped.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix duplicate filtering path in Arrow task batches [iceberg-python]
Posted by "GayathriSrividya (via GitHub)" <gi...@apache.org>.
GayathriSrividya commented on code in PR #3448:
URL: https://github.com/apache/iceberg-python/pull/3448#discussion_r3342373979
##########
pyiceberg/io/pyarrow.py:
##########
Review Comment:
Thanks for the suggestion! [pyarrow_filter](vscode-file://vscode-app/Applications/Visual%20Studio%20Code.app/Contents/Resources/app/out/vs/code/electron-browser/workbench/workbench.html) can be None when no row filter is specified (pure positional-delete scan), so we can't chain [.filter(pyarrow_filter)](vscode-file://vscode-app/Applications/Visual%20Studio%20Code.app/Contents/Resources/app/out/vs/code/electron-browser/workbench/workbench.html) directly without a guard. The current two-step form handles both cases. Happy to collapse it to a one-liner [current_batch.take(indices).filter(pyarrow_filter) if pyarrow_filter is not None else current_batch.take(indices)](vscode-file://vscode-app/Applications/Visual%20Studio%20Code.app/Contents/Resources/app/out/vs/code/electron-browser/workbench/workbench.html) if you'd prefer, though that ends up longer than the current form.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org