This is an automated email from the ASF dual-hosted git repository. JingsongLi pushed a commit to branch release-2.0 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 3d7878d7763c221ffd0973a3981b0a978488a5af 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=[])
