This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new f0d15accad [python] Fix predicate pruning with incomplete statistics
(#8978)
f0d15accad is described below
commit f0d15accadd707b127520d966fc80497f0ea8460
Author: XiaoHongbo <[email protected]>
AuthorDate: Sat Aug 1 20:36:25 2026 +0800
[python] Fix predicate pruning with incomplete statistics (#8978)
---
paimon-python/pypaimon/common/predicate.py | 26 +++++++-
.../pypaimon/manifest/schema/simple_stats.py | 3 +-
.../pypaimon/manifest/simple_stats_evolution.py | 16 +++--
paimon-python/pypaimon/table/row/projected_row.py | 2 +-
.../tests/manifest/simple_stats_evolutions_test.py | 50 +++++++++++++++
paimon-python/pypaimon/tests/predicates_test.py | 75 ++++++++++++++++++++++
6 files changed, 160 insertions(+), 12 deletions(-)
diff --git a/paimon-python/pypaimon/common/predicate.py
b/paimon-python/pypaimon/common/predicate.py
index 540705aa5c..ed1f434a96 100644
--- a/paimon-python/pypaimon/common/predicate.py
+++ b/paimon-python/pypaimon/common/predicate.py
@@ -72,15 +72,35 @@ class Predicate:
if self.method == 'or':
return any(p.test_by_simple_stats(stat, row_count) for p in
self.literals)
- null_count = stat.null_counts[self.index]
+ index = self.index
+ if index is None or index < 0:
+ # Missing stats cannot prove that the file does not match.
+ return True
+
+ null_count = (
+ stat.null_counts[index]
+ if stat.null_counts is not None and index < len(stat.null_counts)
+ else None
+ )
if self.method == 'isNull':
return null_count is None or null_count > 0
if self.method == 'isNotNull':
return null_count is None or row_count is None or null_count <
row_count
- min_value = stat.min_values.get_field(self.index)
- max_value = stat.max_values.get_field(self.index)
+ try:
+ min_value = (
+ stat.min_values.get_field(index)
+ if index < len(stat.min_values)
+ else None
+ )
+ max_value = (
+ stat.max_values.get_field(index)
+ if index < len(stat.max_values)
+ else None
+ )
+ except IndexError:
+ return True
if min_value is None or max_value is None or (null_count is not None
and null_count == row_count):
# invalid stats, skip validation
diff --git a/paimon-python/pypaimon/manifest/schema/simple_stats.py
b/paimon-python/pypaimon/manifest/schema/simple_stats.py
index 0aa25abdde..059b050cd5 100644
--- a/paimon-python/pypaimon/manifest/schema/simple_stats.py
+++ b/paimon-python/pypaimon/manifest/schema/simple_stats.py
@@ -18,6 +18,7 @@
from dataclasses import dataclass
from typing import List
from typing import ClassVar
+from typing import Optional
from pypaimon.table.row.generic_row import GenericRow
from pypaimon.table.row.internal_row import InternalRow
@@ -27,7 +28,7 @@ from pypaimon.table.row.internal_row import InternalRow
class SimpleStats:
min_values: InternalRow
max_values: InternalRow
- null_counts: List[int]
+ null_counts: Optional[List[Optional[int]]]
_empty_stats: ClassVar[object] = None
diff --git a/paimon-python/pypaimon/manifest/simple_stats_evolution.py
b/paimon-python/pypaimon/manifest/simple_stats_evolution.py
index 601933238b..c76202b7e7 100644
--- a/paimon-python/pypaimon/manifest/simple_stats_evolution.py
+++ b/paimon-python/pypaimon/manifest/simple_stats_evolution.py
@@ -37,7 +37,7 @@ class SimpleStatsEvolution:
# Create empty values for optimization
self.empty_values = GenericRow([None] * len(self.field_names),
data_fields)
- self.empty_null_counts = [0] * len(self.field_names)
+ self.empty_null_counts = [None] * len(self.field_names)
def evolution(self, stats: SimpleStats, row_count: Optional[int],
stats_fields: Optional[List[str]]) -> 'SimpleStats':
@@ -95,28 +95,30 @@ class SimpleStatsEvolution:
projected_row = ProjectedRow.from_index_mapping(index_mapping)
return projected_row.replace_row(row)
- def _project_array(self, array: List[Any], index_mapping: List[int]) ->
List[Any]:
+ def _project_array(self, array: Optional[List[Any]], index_mapping:
List[int]) -> List[Any]:
"""Project array based on index mapping."""
if not array:
- return [0] * len(index_mapping)
+ return [None] * len(index_mapping)
projected = []
for mapped_index in index_mapping:
if mapped_index >= 0 and mapped_index < len(array):
projected.append(array[mapped_index])
else:
- projected.append(0) # Default value for missing fields
+ projected.append(None)
return projected
- def _evolve_null_counts(self, null_counts: List[Any], index_mapping:
List[int],
+ def _evolve_null_counts(self, null_counts: Optional[List[Any]],
index_mapping: List[int],
not_found_value: int) -> List[Any]:
"""Evolve null counts with schema evolution mapping."""
evolved = []
for mapped_index in index_mapping:
- if mapped_index >= 0 and mapped_index < len(null_counts):
+ if mapped_index < 0:
+ evolved.append(not_found_value)
+ elif null_counts is not None and mapped_index < len(null_counts):
evolved.append(null_counts[mapped_index])
else:
- evolved.append(not_found_value) # Use row count for missing
fields
+ evolved.append(None)
return evolved
diff --git a/paimon-python/pypaimon/table/row/projected_row.py
b/paimon-python/pypaimon/table/row/projected_row.py
index 5fc66b792e..ec96c0d193 100644
--- a/paimon-python/pypaimon/table/row/projected_row.py
+++ b/paimon-python/pypaimon/table/row/projected_row.py
@@ -65,7 +65,7 @@ class ProjectedRow(InternalRow):
def __len__(self) -> int:
"""Returns the number of fields in this row."""
- return len(self.row)
+ return len(self.index_mapping)
def __str__(self) -> str:
"""String representation of the projected row."""
diff --git
a/paimon-python/pypaimon/tests/manifest/simple_stats_evolutions_test.py
b/paimon-python/pypaimon/tests/manifest/simple_stats_evolutions_test.py
index 98d6433812..9f134a96cf 100644
--- a/paimon-python/pypaimon/tests/manifest/simple_stats_evolutions_test.py
+++ b/paimon-python/pypaimon/tests/manifest/simple_stats_evolutions_test.py
@@ -17,6 +17,7 @@
import unittest
+from pypaimon.common.predicate import Predicate
from pypaimon.manifest.simple_stats_evolutions import SimpleStatsEvolutions
from pypaimon.schema.data_types import DataField, AtomicType
from pypaimon.manifest.schema.simple_stats import SimpleStats
@@ -42,6 +43,22 @@ class SimpleStatsEvolutionsTest(unittest.TestCase):
evolution = evolutions.get_or_create(0)
self.assertIsNone(evolution.index_mapping)
+ def test_predicate_with_incomplete_projected_stats(self):
+ fields = self._make_fields([(0, 'a', 'INT'), (1, 'b', 'INT')])
+ evolution = SimpleStatsEvolutions(lambda _: fields, 0).get_or_create(0)
+ predicate = Predicate(method='equal', index=1, field='b',
literals=[15])
+
+ for min_values, max_values in (([1], [10, 20]), ([1, 2], [10])):
+ stats = SimpleStats(
+ GenericRow(min_values, fields[:len(min_values)]),
+ GenericRow(max_values, fields[:len(max_values)]),
+ [0, 0],
+ )
+ evolved = evolution.evolution(stats, 10, ['a', 'b'])
+
+ with self.subTest(min_values=min_values, max_values=max_values):
+ self.assertTrue(predicate.test_by_simple_stats(evolved, 10))
+
def test_added_column(self):
"""New column: mapping is -1, null_count = row_count."""
data_fields = self._make_fields([(0, 'a', 'INT'), (1, 'b', 'INT')])
@@ -59,6 +76,39 @@ class SimpleStatsEvolutionsTest(unittest.TestCase):
self.assertIsNone(evolved.max_values.get_field(2))
self.assertEqual(evolved.null_counts, [0, 5, 500])
+ def test_missing_null_counts_remain_unknown(self):
+ fields = self._make_fields([(0, 'a', 'INT'), (1, 'b', 'INT')])
+ schemas = {0: fields}
+ evolution = SimpleStatsEvolutions(
+ lambda sid: schemas[sid], 0).get_or_create(0)
+ stats = SimpleStats(GenericRow([], []), GenericRow([], []), [])
+ predicate = Predicate(method='isNull', index=0, field='a')
+
+ for stats_fields in ([], ['a', 'b']):
+ evolved = evolution.evolution(
+ stats, row_count=100, stats_fields=stats_fields)
+ with self.subTest(stats_fields=stats_fields):
+ self.assertEqual(evolved.null_counts, [None, None])
+ self.assertTrue(predicate.test_by_simple_stats(evolved, 100))
+
+ def test_schema_evolution_preserves_unknown_null_counts(self):
+ data_fields = self._make_fields([(0, 'a', 'INT')])
+ table_fields = self._make_fields([(0, 'a', 'INT'), (1, 'b', 'INT')])
+ schemas = {0: data_fields, 1: table_fields}
+ evolution = SimpleStatsEvolutions(
+ lambda sid: schemas[sid], 1).get_or_create(0)
+ stats = SimpleStats(
+ GenericRow([1], data_fields), GenericRow([10], data_fields), [])
+
+ evolved = evolution.evolution(
+ stats, row_count=100, stats_fields=None)
+
+ self.assertEqual(evolved.null_counts, [None, 100])
+ self.assertTrue(
+ Predicate(method='isNull', index=0, field='a')
+ .test_by_simple_stats(evolved, 100)
+ )
+
def test_dropped_column(self):
"""Dropped column is excluded from mapping."""
data_fields = self._make_fields([(0, 'a', 'INT'), (1, 'b', 'STRING'),
(2, 'c', 'BIGINT')])
diff --git a/paimon-python/pypaimon/tests/predicates_test.py
b/paimon-python/pypaimon/tests/predicates_test.py
index 9bcea3b39a..d6f92963bf 100644
--- a/paimon-python/pypaimon/tests/predicates_test.py
+++ b/paimon-python/pypaimon/tests/predicates_test.py
@@ -28,8 +28,10 @@ import pyarrow.dataset as ds
from pypaimon import CatalogFactory, Schema
from pypaimon.common.predicate import Predicate
from pypaimon.manifest.schema.simple_stats import SimpleStats
+from pypaimon.schema.data_types import DataField
from pypaimon.table.row.generic_row import GenericRow, GenericRowDeserializer
from pypaimon.table.row.offset_row import OffsetRow
+from pypaimon.table.row.projected_row import ProjectedRow
def _check_filtered_result(read_builder, expected_df):
@@ -443,6 +445,79 @@ class PredicateTest(unittest.TestCase):
)
self.assertTrue(pred.test_by_simple_stats(stat_positive, 10))
+ def test_by_simple_stats_with_incomplete_fields(self):
+ fields = [DataField(0, 'f0', 'INT'), DataField(1, 'f1', 'INT')]
+ predicate = Predicate(method='equal', index=1, field='f1',
literals=[15])
+ stats = [
+ SimpleStats(
+ min_values=GenericRow([1], fields[:1]),
+ max_values=GenericRow([10, 20], fields),
+ null_counts=[0, 0],
+ ),
+ SimpleStats(
+ min_values=GenericRow([1, 2], fields),
+ max_values=GenericRow([10], fields[:1]),
+ null_counts=[0, 0],
+ ),
+ SimpleStats(
+ min_values=GenericRow([1], fields[:1]),
+ max_values=GenericRow([10], fields[:1]),
+ null_counts=[0],
+ ),
+ ]
+
+ for stat in stats:
+ with self.subTest(stat=stat):
+ self.assertTrue(predicate.test_by_simple_stats(stat, 10))
+
+ def test_by_simple_stats_without_null_counts(self):
+ fields = [DataField(0, 'f0', 'INT')]
+ predicate = Predicate(method='equal', index=0, field='f0',
literals=[20])
+ for null_counts in (None, []):
+ stat = SimpleStats(
+ min_values=GenericRow([1], fields),
+ max_values=GenericRow([10], fields),
+ null_counts=null_counts,
+ )
+ with self.subTest(null_counts=null_counts):
+ self.assertFalse(predicate.test_by_simple_stats(stat, 10))
+
+ def test_by_simple_stats_with_invalid_index(self):
+ fields = [DataField(0, 'f0', 'INT')]
+ stat = SimpleStats(
+ min_values=GenericRow([1], fields),
+ max_values=GenericRow([10], fields),
+ null_counts=[0],
+ )
+ for index in (None, -1):
+ predicate = Predicate(
+ method='equal', index=index, field='_ROW_ID', literals=[5])
+ with self.subTest(index=index):
+ self.assertTrue(predicate.test_by_simple_stats(stat, 10))
+
+ def test_by_simple_stats_null_predicate_without_null_counts(self):
+ fields = [DataField(0, 'f0', 'INT')]
+ stat = SimpleStats(
+ min_values=GenericRow([1], fields),
+ max_values=GenericRow([10], fields),
+ null_counts=None,
+ )
+ for method in ('isNull', 'isNotNull'):
+ predicate = Predicate(method=method, index=0, field='f0')
+ with self.subTest(method=method):
+ self.assertTrue(predicate.test_by_simple_stats(stat, 10))
+
+ def test_by_simple_stats_with_projected_rows(self):
+ fields = [DataField(0, 'f0', 'INT'), DataField(1, 'f1', 'INT')]
+ row = GenericRow([1, 2], fields)
+ min_values = ProjectedRow.from_index_mapping([0]).replace_row(row)
+ max_values = ProjectedRow.from_index_mapping([0]).replace_row(row)
+ stat = SimpleStats(min_values, max_values, [0, 0])
+ predicate = Predicate(method='equal', index=1, field='f1',
literals=[2])
+
+ self.assertEqual(len(min_values), 1)
+ self.assertTrue(predicate.test_by_simple_stats(stat, 10))
+
def test_filter_with_null_and_or(self):
p_gt = Predicate(method='greaterThan', index=1, field='score',
literals=[10])
p_null = Predicate(method='isNull', index=1, field='score',
literals=[])
