@@ -3267,48 +3267,99 @@ def _expected_batch(unit: str) -> pa.RecordBatch:
32673267 assert _expected_batch ("ns" if format_version > 2 else "us" ).equals (actual_result )
32683268
32693269
3270- def test_task_to_record_batches_filter_without_positional_deletes_avoids_table_refilter (tmpdir : str ) -> None :
3270+ def test_task_to_record_batches_does_not_use_table_filter_without_positional_deletes (tmpdir : str ) -> None :
3271+ """Regression test for https://github.com/apache/iceberg-python/issues/3272.
3272+
3273+ The old code always created a ``pa.Table`` via ``pa.Table.from_batches`` and called
3274+ ``filter`` / ``combine_chunks`` / ``to_batches()[0]`` — even when no positional deletes
3275+ existed and the scanner had already applied the predicate push-down. On Apple Silicon
3276+ with affected PyArrow versions this triggered a SIGSEGV. The fix removes that
3277+ Table-based code path for the no-positional-delete case.
3278+
3279+ We detect a regression by replacing ``pyarrow.Table`` with a sentinel class whose
3280+ ``from_batches`` raises immediately. ``isinstance`` checks against the original
3281+ ``pa.Table`` are preserved via a custom metaclass so the rest of the code-path is
3282+ unaffected. With the fix the sentinel is never reached; without the fix it is called
3283+ and the test fails.
3284+ """
3285+ from pyiceberg .expressions .visitors import bind
3286+
32713287 arrow_schema = pa .schema ((pa .field ("id" , pa .int32 (), nullable = True , metadata = {PYARROW_PARQUET_FIELD_ID_KEY : "1" }),))
32723288 arrow_table = pa .table ([pa .array ([1 , 2 , 3 ], type = pa .int32 ())], schema = arrow_schema )
32733289 data_file = _write_table_to_data_file (
3274- f"{ tmpdir } /test_task_to_record_batches_filter_no_positional .parquet" , arrow_schema , arrow_table
3290+ f"{ tmpdir } /test_task_to_record_batches_no_table_refilter .parquet" , arrow_schema , arrow_table
32753291 )
32763292
32773293 table_schema = Schema (NestedField (1 , "id" , IntegerType (), required = False ))
3278- from pyiceberg .expressions .visitors import bind
32793294
3280- result_batches = list (
3281- _task_to_record_batches (
3282- PyArrowFileIO (),
3283- FileScanTask (data_file ),
3284- bound_row_filter = bind (table_schema , GreaterThan ("id" , 1 ), case_sensitive = True ),
3285- projected_schema = table_schema ,
3286- table_schema = table_schema ,
3287- projected_field_ids = {1 },
3288- positional_deletes = None ,
3289- case_sensitive = True ,
3295+ _real_Table = pa .Table
3296+
3297+ class _TableFilterSentinelMeta (type ):
3298+ """Metaclass that delegates isinstance/issubclass to the real pa.Table."""
3299+
3300+ def __instancecheck__ (cls , instance : object ) -> bool :
3301+ return isinstance (instance , _real_Table )
3302+
3303+ def __subclasscheck__ (cls , subclass : type ) -> bool :
3304+ return issubclass (subclass , _real_Table )
3305+
3306+ class _TableFilterSentinel (metaclass = _TableFilterSentinelMeta ):
3307+ @staticmethod
3308+ def from_batches (* _ : object , ** __ : object ) -> None :
3309+ raise AssertionError (
3310+ "pa.Table.from_batches must not be called when positional_deletes is None: "
3311+ "the SIGSEGV-prone workaround path (from_batches/filter/combine_chunks/to_batches) was taken"
3312+ )
3313+
3314+ with patch ("pyarrow.Table" , new = _TableFilterSentinel ):
3315+ result_batches = list (
3316+ _task_to_record_batches (
3317+ PyArrowFileIO (),
3318+ FileScanTask (data_file ),
3319+ bound_row_filter = bind (table_schema , GreaterThan ("id" , 1 ), case_sensitive = True ),
3320+ projected_schema = table_schema ,
3321+ table_schema = table_schema ,
3322+ projected_field_ids = {1 },
3323+ positional_deletes = None ,
3324+ case_sensitive = True ,
3325+ )
32903326 )
3291- )
32923327 assert len (result_batches ) == 1
32933328 assert result_batches [0 ].column (0 ).to_pylist () == [2 , 3 ]
32943329
32953330
3296- def test_task_to_record_batches_filter_with_positional_deletes_handles_empty_batch (tmpdir : str ) -> None :
3331+ def test_task_to_record_batches_filter_applied_after_positional_deletes (tmpdir : str ) -> None :
3332+ """Regression test: the row filter must be applied *after* positional deletes are removed.
3333+
3334+ When positional deletes are present the scanner does not push down the predicate, so
3335+ ``_task_to_record_batches`` must apply ``pyarrow_filter`` explicitly after ``take``.
3336+ This test uses data where the expected result differs from both
3337+ "filter only" and "deletes only" projections, ensuring that skipping either step
3338+ would produce the wrong answer.
3339+ """
3340+ from pyiceberg .expressions .visitors import bind
3341+
32973342 arrow_schema = pa .schema ((pa .field ("id" , pa .int32 (), nullable = True , metadata = {PYARROW_PARQUET_FIELD_ID_KEY : "1" }),))
3298- arrow_table = pa .table ([pa .array ([1 , 2 , 3 ], type = pa .int32 ())], schema = arrow_schema )
3343+ # File rows (0-indexed positions): 0→1, 1→2, 2→3, 3→4, 4→5
3344+ arrow_table = pa .table ([pa .array ([1 , 2 , 3 , 4 , 5 ], type = pa .int32 ())], schema = arrow_schema )
32993345 data_file = _write_table_to_data_file (
33003346 f"{ tmpdir } /test_task_to_record_batches_filter_with_positional.parquet" , arrow_schema , arrow_table
33013347 )
33023348
33033349 table_schema = Schema (NestedField (1 , "id" , IntegerType (), required = False ))
3304- from pyiceberg .expressions .visitors import bind
33053350
3306- positional_deletes = [pa .chunked_array ([pa .array ([], type = pa .int64 ())])]
3351+ # Delete file-positions 1 and 3 (values 2 and 4); survivors: [1, 3, 5]
3352+ # Then apply filter id >= 3; expected result: [3, 5]
3353+ #
3354+ # Wrong results that would indicate a bug:
3355+ # [1, 3, 5] — filter not applied after deletes
3356+ # [3, 4, 5] — positional deletes not applied (scanner skips filter push-down)
3357+ positional_deletes = [pa .chunked_array ([pa .array ([1 , 3 ], type = pa .int64 ())])]
33073358 result_batches = list (
33083359 _task_to_record_batches (
33093360 PyArrowFileIO (),
33103361 FileScanTask (data_file ),
3311- bound_row_filter = bind (table_schema , GreaterThan ("id" , 100 ), case_sensitive = True ),
3362+ bound_row_filter = bind (table_schema , GreaterThan ("id" , 2 ), case_sensitive = True ),
33123363 projected_schema = table_schema ,
33133364 table_schema = table_schema ,
33143365 projected_field_ids = {1 },
@@ -3317,7 +3368,8 @@ def test_task_to_record_batches_filter_with_positional_deletes_handles_empty_bat
33173368 )
33183369 )
33193370
3320- assert result_batches == []
3371+ assert len (result_batches ) == 1
3372+ assert result_batches [0 ].column (0 ).to_pylist () == [3 , 5 ]
33213373
33223374
33233375def test_parse_location_defaults () -> None :
0 commit comments