[hcrosse]

Closes #4276

Rationale for this change

In the issue's example, p != 'a' gives count() == 9 versus len(to_arrow()) == 4 because count() includes NULL rows that to_arrow() drops.

ResidualVisitor now lets a NULL partition value decide only IsNull and NotNull. Other predicates stay in the residual, so count() reads those files instead of using record_count. test_is_not_nan now expects the predicate for a NULL value.

Are these changes tested?

test_null_partition_value_only_decides_null_checks covers the visitor. Two tests in tests/catalog/test_catalog_behaviors.py cover count(). They run on the memory and SQL catalogs and fail on main:

• test_count_matches_to_arrow_with_null_partition_values checks the results across partition transforms and spec evolution.
• test_count_reads_null_partition_only_when_filter_compares_it checks which files count() reads.

The unit suite passed (4624 passed, 5 skipped), and prek run -a passed.

Are there any user-facing changes?

For these scans, count() matches to_arrow() and reads some files that it previously counted from metadata. FileScanTask.residual changes for NULL partition values. There is no API change.

Please add the changelog label.

[Copilot]

🟡 Changes recommended

Omitted REST partition values can trigger an IndexError during filtered server-planned scans; missing values need a safe fallback to reading the file and regression coverage.

1 open issue

• Handle omitted REST partition values without raising IndexError · New

What changed in this PR

This PR addresses #4276 by making count() follow to_arrow() for filters involving NULL partition values, without changing residual evaluation or Arrow’s NULL semantics.

Changes:

• Reads affected files instead of relying on metadata counts, while preserving eligible metadata-only counts.
• Adds regression coverage for identity, bucket, and evolved partition specs, plus checks that unaffected files are not read.

File	Description
tests/​catalog/​test_catalog_behaviors.py	Adds count-consistency and metadata-only regression tests.
pyiceberg/​table/​__init__.py	Detects relevant NULL partition values and routes affected count tasks through Arrow.

_{Review effort: Balanced}

[Fokko]

Thanks for addressing this bug, this is something we should fix. However, I don't think this is the right angle. I've been playing around locally, and I think we should fix ResidualVisitor.

Something like:

diff --git a/pyiceberg/expressions/visitors.py b/pyiceberg/expressions/visitors.py
index a8de08866..750454850 100644
--- a/pyiceberg/expressions/visitors.py
+++ b/pyiceberg/expressions/visitors.py
@@ -50,7 +50,9 @@ from pyiceberg.expressions import (
     BoundStartsWith,
     BoundTerm,
     BoundUnaryPredicate,
+    IsNull,
     Not,
+    NotNull,
     Or,
     UnboundPredicate,
 )
@@ -1969,36 +1971,29 @@ class ResidualVisitor(BoundBooleanExpressionVisitor[BooleanExpression], ABC):
             return predicate
 
         for part in parts:
-            strict_projection = part.transform.strict_project(part.name, predicate)
-            strict_result = None
-
-            if strict_projection is not None:
-                bound = strict_projection.bind(self.partition_schema, case_sensitive=self.case_sensitive)
-                if isinstance(bound, BoundPredicate):
-                    strict_result = super().visit_bound_predicate(bound)
-                else:
-                    # if the result is not a predicate, then it must be a constant like alwaysTrue or alwaysFalse
-                    strict_result = bound
-
-            if isinstance(strict_result, AlwaysTrue):
+            if isinstance(self._eval_projection(part.transform.strict_project(part.name, predicate)), AlwaysTrue):
                 return AlwaysTrue()
 
-            inclusive_projection = part.transform.project(part.name, predicate)
-            inclusive_result = None
-            if inclusive_projection is not None:
-                bound_inclusive = inclusive_projection.bind(self.partition_schema, case_sensitive=self.case_sensitive)
-                if isinstance(bound_inclusive, BoundPredicate):
-                    # using predicate method specific to inclusive
-                    inclusive_result = super().visit_bound_predicate(bound_inclusive)
-                else:
-                    # if the result is not a predicate, then it must be a constant like alwaysTrue or
-                    # alwaysFalse
-                    inclusive_result = bound_inclusive
-            if isinstance(inclusive_result, AlwaysFalse):
+            if isinstance(self._eval_projection(part.transform.project(part.name, predicate)), AlwaysFalse):
                 return AlwaysFalse()
 
         return predicate
 
+    def _eval_projection(self, projection: UnboundPredicate | None) -> BooleanExpression | None:
+        if projection is None:
+            return None
+
+        bound = projection.bind(self.partition_schema, case_sensitive=self.case_sensitive)
+        if not isinstance(bound, BoundPredicate):
+            # if the result is not a predicate, then it must be a constant like alwaysTrue or alwaysFalse
+            return bound
+
+        # A NULL partition value only decides null checks; leave other predicates to the reader's NULL semantics
+        if bound.as_unbound not in (IsNull, NotNull) and bound.term.eval(self.struct) is None:
+            return None
+
+        return super().visit_bound_predicate(bound)
+
     def visit_unbound_predicate(self, predicate: UnboundPredicate) -> BooleanExpression:
         bound = predicate.bind(self.schema, case_sensitive=self.case_sensitive)
 
diff --git a/pyiceberg/table/__init__.py b/pyiceberg/table/__init__.py
index a6d882314..8552aea0f 100644
--- a/pyiceberg/table/__init__.py
+++ b/pyiceberg/table/__init__.py
@@ -36,29 +36,15 @@ from pydantic import Field
 
 from pyiceberg.encryption.kms import KeyManagementClient
 from pyiceberg.exceptions import CommitFailedException, CommitStateUnknownException, ValidationException
-from pyiceberg.expressions import (
-    AlwaysFalse,
-    AlwaysTrue,
-    And,
-    BooleanExpression,
-    BoundPredicate,
-    EqualTo,
-    IsNaN,
-    IsNull,
-    NotNull,
-    Or,
-    Reference,
-)
+from pyiceberg.expressions import AlwaysFalse, AlwaysTrue, And, BooleanExpression, EqualTo, IsNaN, IsNull, Or, Reference
 from pyiceberg.expressions.visitors import (
     ResidualEvaluator,
-    _ExpressionFieldIDs,
     _InclusiveMetricsEvaluator,
     bind,
     expression_evaluator,
     extract_field_ids,
     inclusive_projection,
     manifest_evaluator,
-    visit,
 )
 from pyiceberg.io import FileIO, load_file_io
 from pyiceberg.manifest import (
@@ -2517,14 +2503,6 @@ def _to_arrow_batch_reader_via_file_scan_tasks(
     return pa.RecordBatchReader.from_batches(target_schema, batches).cast(target_schema)
 
 
-class _FieldIDsOutsideNullChecks(_ExpressionFieldIDs):
-    # Projected null checks preserve the source's NULL status, and void does not project predicates
-    def visit_bound_predicate(self, predicate: BoundPredicate) -> set[int]:
-        if predicate.as_unbound in (IsNull, NotNull):
-            return set()
-        return super().visit_bound_predicate(predicate)
-
-
 class DataScan(TableScan):
     @cached_property
     def _manifest_planner(self) -> ManifestGroupPlanner:
@@ -2653,31 +2631,12 @@ class DataScan(TableScan):
         res = 0
         # every task is a FileScanTask
         tasks = self.plan_files()
-        bound_filter = bind(self.table_metadata.schema(), self.row_filter, case_sensitive=self.case_sensitive)
-        field_ids_outside_null_checks = visit(bound_filter, _FieldIDsOutsideNullChecks())
-        specs = self.table_metadata.specs() if field_ids_outside_null_checks else {}
-        referenced_positions = {
-            spec_id: [pos for pos, field in enumerate(spec.fields) if field.source_id in field_ids_outside_null_checks]
-            for spec_id, spec in specs.items()
-        }
 
         for task in tasks:
             # task.residual is a Boolean Expression if the filter condition is fully satisfied by the
             # partition value and task.delete_files represents that positional delete haven't been merged yet
             # hence those files have to read as a pyarrow table applying the filter and deletes
-            if (
-                task.residual == AlwaysTrue()
-                and len(task.delete_files) == 0
-                # A NULL partition value can leave an AlwaysTrue residual although the filter rejects its rows
-                and (
-                    not field_ids_outside_null_checks
-                    or (
-                        task.file.spec_id in specs
-                        and len(task.file.partition) == len(specs[task.file.spec_id].fields)
-                        and all(task.file.partition[pos] is not None for pos in referenced_positions[task.file.spec_id])
-                    )
-                )
-            ):
+            if task.residual == AlwaysTrue() and len(task.delete_files) == 0:
                 # Every File has a metadata stat that stores the file record count
                 res += task.file.record_count
             else:
diff --git a/tests/catalog/test_scan_planning_models.py b/tests/catalog/test_scan_planning_models.py
index 60bd700f4..f88e24e92 100644
--- a/tests/catalog/test_scan_planning_models.py
+++ b/tests/catalog/test_scan_planning_models.py
@@ -14,11 +14,8 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-from pathlib import Path
 from typing import Any
 
-import pyarrow as pa
-import pyarrow.parquet as pq
 import pytest
 from pydantic import TypeAdapter, ValidationError
 from requests_mock import Mocker
@@ -42,7 +39,6 @@ from pyiceberg.catalog.rest.scan_planning import (
     ValueMap,
 )
 from pyiceberg.expressions import AlwaysTrue, EqualTo, Reference
-from pyiceberg.io.pyarrow import schema_to_pyarrow
 from pyiceberg.manifest import FileFormat
 from pyiceberg.table import FileScanTask
 
@@ -836,36 +832,6 @@ def test_scan_uses_server_side_planning_from_load_table_config(
     assert plan_mock.call_count == 1
 
 
-def test_count_reads_server_planned_file_without_partition_values(
-    rest_scan_catalog: RestCatalog, requests_mock: Mocker, example_table_metadata_v2: dict[str, Any], tmp_path: Path
-) -> None:
-    _mock_load_table(requests_mock, example_table_metadata_v2, {})
-    table = rest_scan_catalog.load_table(("db", "tbl"))
-    data_file = tmp_path / "file1.parquet"
-    pq.write_table(
-        pa.table({"x": [1, 1, 1], "y": [4, 5, 6], "z": [7, 8, 9]}, schema=schema_to_pyarrow(table.schema())), data_file
-    )
-    requests_mock.post(
-        f"{TEST_URI}v1/namespaces/db/tables/tbl/plan",
-        json={
-            "status": "completed",
-            "plan-id": "plan-123",
-            "delete-files": [],
-            "file-scan-tasks": [
-                {
-                    "data-file": _rest_data_file(
-                        file_path=str(data_file), file_size_in_bytes=data_file.stat().st_size, record_count=3
-                    )
-                }
-            ],
-            "plan-tasks": [],
-        },
-        status_code=200,
-    )
-
-    assert table.scan(row_filter="x != 1").count() == 0
-
-
 def test_scan_keeps_client_side_planning_without_load_table_config(
     rest_client_planning_catalog: RestCatalog, requests_mock: Mocker, example_table_metadata_v2: dict[str, Any]
 ) -> None:
diff --git a/tests/expressions/test_residual_evaluator.py b/tests/expressions/test_residual_evaluator.py
index 4c6295e9e..abcf1fb8c 100644
--- a/tests/expressions/test_residual_evaluator.py
+++ b/tests/expressions/test_residual_evaluator.py
@@ -29,6 +29,8 @@ from pyiceberg.expressions import (
     IsNaN,
     IsNull,
     LessThan,
+    Not,
+    NotEqualTo,
     NotIn,
     NotNaN,
     NotNull,
@@ -40,7 +42,7 @@ from pyiceberg.expressions.literals import literal
 from pyiceberg.expressions.visitors import ResidualVisitor, residual_evaluator_of
 from pyiceberg.partitioning import PartitionField, PartitionSpec
 from pyiceberg.schema import Schema
-from pyiceberg.transforms import DayTransform, IdentityTransform
+from pyiceberg.transforms import BucketTransform, DayTransform, IdentityTransform
 from pyiceberg.typedef import Record
 from pyiceberg.types import DoubleType, FloatType, IntegerType, NestedField, StringType, TimestampType
 
@@ -233,7 +235,7 @@ def test_is_not_nan() -> None:
     res_eval = residual_evaluator_of(spec=spec, expr=predicate, case_sensitive=True, schema=schema)
 
     residual = res_eval.residual_for(Record(None))
-    assert residual == AlwaysTrue()
+    assert residual == predicate
 
     residual = res_eval.residual_for(Record(float("nan")))
     assert residual == AlwaysFalse()
@@ -248,7 +250,7 @@ def test_is_not_nan() -> None:
     res_eval = residual_evaluator_of(spec=spec, expr=predicate, case_sensitive=True, schema=schema)
 
     residual = res_eval.residual_for(Record(None))
-    assert residual == AlwaysTrue()
+    assert residual == predicate
 
     residual = res_eval.residual_for(Record(float("nan")))
     assert residual == AlwaysFalse()
@@ -303,3 +305,23 @@ def test_not_starts_with() -> None:
     assert res_eval.residual_for(Record("abc")) == AlwaysFalse()
     assert res_eval.residual_for(Record("a")) == AlwaysFalse()
     assert res_eval.residual_for(Record("zoo")) == AlwaysTrue()
+
+
+def test_null_partition_value_only_decides_null_checks() -> None:
+    schema = Schema(NestedField(1, "x", StringType(), required=False))
+    identity = PartitionSpec(PartitionField(1, 1001, IdentityTransform(), "x_part"))
+    bucket = PartitionSpec(PartitionField(1, 1001, BucketTransform(4), "x_bucket"))
+
+    for spec in [identity, bucket]:
+        for predicate in [NotEqualTo("x", "a"), Not(EqualTo("x", "a")), NotIn("x", ["a", "c"]), NotStartsWith("x", "a")]:
+            res_eval = residual_evaluator_of(spec=spec, expr=predicate, case_sensitive=True, schema=schema)
+            assert res_eval.residual_for(Record(None)) != AlwaysTrue(), (spec, predicate)
+
+        assert (
+            residual_evaluator_of(spec=spec, expr=IsNull("x"), case_sensitive=True, schema=schema).residual_for(Record(None))
+            == AlwaysTrue()
+        )
+        assert (
+            residual_evaluator_of(spec=spec, expr=NotNull("x"), case_sensitive=True, schema=schema).residual_for(Record(None))
+            == AlwaysFalse()
+        )

[hcrosse]

@Fokko, thanks for the patch. I applied it verbatim.

Java's ResidualEvaluator returns alwaysTrue for notEq, notIn and notNaN on NULL. TestResiduals asserts that "null not in" is true. Your patch also reverses the NotNaN expectation from #3689. Are you fine diverging from Java here?

The REST test that you removed covered a different case. Without residual-filter, PyIceberg uses AlwaysTrue, so count() trusts record_count. In that test, count() returns 3 on main and with your patch, while to_arrow() returns 0. The REST spec requires the client to compute the residual or use the original filter. Would you like me to fix that here or in a separate PR?