[Python] SimpleStats supports BinaryRow (#6444)

Author: discivigour

paimon-python/pypaimon/common/predicate.py

@@ -27,6 +27,7 @@
 from pyarrow import dataset as pyarrow_dataset
 
 from pypaimon.manifest.schema.simple_stats import SimpleStats
+from pypaimon.table.row.generic_row import GenericRow
 from pypaimon.table.row.internal_row import InternalRow
 
 
@@ -67,32 +68,31 @@ def test(self, record: InternalRow) -> bool:
         raise ValueError(f"Unsupported predicate method: {self.method}")
 
     def test_by_simple_stats(self, stat: SimpleStats, row_count: int) -> bool:
-        return self.test_by_stats({
-            "min_values": stat.min_values.to_dict(),
-            "max_values": stat.max_values.to_dict(),
-            "null_counts": {
-                stat.min_values.fields[i].name: stat.null_counts[i] for i in range(len(stat.min_values.fields))
-            },
-            "row_count": row_count,
-        })
-
-    def test_by_stats(self, stat: Dict) -> bool:
+        """Test predicate against BinaryRow stats with denseIndexMapping like Java implementation."""
         if self.method == 'and':
-            return all(p.test_by_stats(stat) for p in self.literals)
+            return all(p.test_by_simple_stats(stat, row_count) for p in self.literals)
         if self.method == 'or':
-            t = any(p.test_by_stats(stat) for p in self.literals)
-            return t
+            return any(p.test_by_simple_stats(stat, row_count) for p in self.literals)
 
-        null_count = stat["null_counts"][self.field]
-        row_count = stat["row_count"]
+        # Get null count using the mapped index
+        null_count = stat.null_counts[self.index] if stat.null_counts and self.index < len(
+            stat.null_counts) else 0
 
         if self.method == 'isNull':
             return null_count is not None and 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"][self.field]
-        max_value = stat["max_values"][self.field]
+        if not isinstance(stat.min_values, GenericRow):
+            # Parse field values using BinaryRow's direct field access by name
+            min_value = stat.min_values.get_field(self.index)
+            max_value = stat.max_values.get_field(self.index)
+        else:
+            # TODO transform partition to BinaryRow
+            min_values = stat.min_values.to_dict()
+            max_values = stat.max_values.to_dict()
+            min_value = min_values[self.field]
+            max_value = max_values[self.field]
 
         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
@@ -164,7 +164,6 @@ def __init__(cls, name, bases, dct):
 
 
 class Tester(ABC, metaclass=RegisterMeta):
-
     name = None
 
     @abstractmethod
@@ -187,7 +186,6 @@ def test_by_arrow(self, val, literals) -> bool:
 
 
 class Equal(Tester):
-
     name = 'equal'
 
     def test_by_value(self, val, literals) -> bool:
@@ -201,7 +199,6 @@ def test_by_arrow(self, val, literals) -> bool:
 
 
 class NotEqual(Tester):
-
     name = "notEqual"
 
     def test_by_value(self, val, literals) -> bool:
@@ -215,7 +212,6 @@ def test_by_arrow(self, val, literals) -> bool:
 
 
 class LessThan(Tester):
-
     name = "lessThan"
 
     def test_by_value(self, val, literals) -> bool:
@@ -229,7 +225,6 @@ def test_by_arrow(self, val, literals) -> bool:
 
 
 class LessOrEqual(Tester):
-
     name = "lessOrEqual"
 
     def test_by_value(self, val, literals) -> bool:
@@ -243,7 +238,6 @@ def test_by_arrow(self, val, literals) -> bool:
 
 
 class GreaterThan(Tester):
-
     name = "greaterThan"
 
     def test_by_value(self, val, literals) -> bool:
@@ -257,7 +251,6 @@ def test_by_arrow(self, val, literals) -> bool:
 
 
 class GreaterOrEqual(Tester):
-
     name = "greaterOrEqual"
 
     def test_by_value(self, val, literals) -> bool:
@@ -271,7 +264,6 @@ def test_by_arrow(self, val, literals) -> bool:
 
 
 class In(Tester):
-
     name = "in"
 
     def test_by_value(self, val, literals) -> bool:
@@ -285,7 +277,6 @@ def test_by_arrow(self, val, literals) -> bool:
 
 
 class NotIn(Tester):
-
     name = "notIn"
 
     def test_by_value(self, val, literals) -> bool:
@@ -299,7 +290,6 @@ def test_by_arrow(self, val, literals) -> bool:
 
 
 class Between(Tester):
-
     name = "between"
 
     def test_by_value(self, val, literals) -> bool:
@@ -313,7 +303,6 @@ def test_by_arrow(self, val, literals) -> bool:
 
 
 class StartsWith(Tester):
-
     name = "startsWith"
 
     def test_by_value(self, val, literals) -> bool:
@@ -329,7 +318,6 @@ def test_by_arrow(self, val, literals) -> bool:
 
 
 class EndsWith(Tester):
-
     name = "endsWith"
 
     def test_by_value(self, val, literals) -> bool:
@@ -343,7 +331,6 @@ def test_by_arrow(self, val, literals) -> bool:
 
 
 class Contains(Tester):
-
     name = "contains"
 
     def test_by_value(self, val, literals) -> bool:
@@ -357,7 +344,6 @@ def test_by_arrow(self, val, literals) -> bool:
 
 
 class IsNull(Tester):
-
     name = "isNull"
 
     def test_by_value(self, val, literals) -> bool:
@@ -371,7 +357,6 @@ def test_by_arrow(self, val, literals) -> bool:
 
 
 class IsNotNull(Tester):
-
     name = "isNotNull"
 
     def test_by_value(self, val, literals) -> bool:

paimon-python/pypaimon/manifest/manifest_file_manager.py

@@ -26,6 +26,7 @@
 from pypaimon.manifest.schema.simple_stats import SimpleStats
 from pypaimon.table.row.generic_row import (GenericRowDeserializer,
                                             GenericRowSerializer)
+from pypaimon.table.row.binary_row import BinaryRow
 
 
 class ManifestFileManager:
@@ -54,12 +55,11 @@ def read(self, manifest_file_name: str, manifest_entry_filter=None, drop_stats=T
             file_dict = dict(record['_FILE'])
             key_dict = dict(file_dict['_KEY_STATS'])
             key_stats = SimpleStats(
-                min_values=GenericRowDeserializer.from_bytes(key_dict['_MIN_VALUES'],
-                                                             self.trimmed_primary_key_fields),
-                max_values=GenericRowDeserializer.from_bytes(key_dict['_MAX_VALUES'],
-                                                             self.trimmed_primary_key_fields),
+                min_values=BinaryRow(key_dict['_MIN_VALUES'], self.trimmed_primary_key_fields),
+                max_values=BinaryRow(key_dict['_MAX_VALUES'], self.trimmed_primary_key_fields),
                 null_counts=key_dict['_NULL_COUNTS'],
             )
+
             value_dict = dict(file_dict['_VALUE_STATS'])
             if file_dict['_VALUE_STATS_COLS'] is None:
                 if file_dict['_WRITE_COLS'] is None:
@@ -72,8 +72,8 @@ def read(self, manifest_file_name: str, manifest_entry_filter=None, drop_stats=T
             else:
                 fields = [self.table.field_dict[col] for col in file_dict['_VALUE_STATS_COLS']]
             value_stats = SimpleStats(
-                min_values=GenericRowDeserializer.from_bytes(value_dict['_MIN_VALUES'], fields),
-                max_values=GenericRowDeserializer.from_bytes(value_dict['_MAX_VALUES'], fields),
+                min_values=BinaryRow(value_dict['_MIN_VALUES'], fields),
+                max_values=BinaryRow(value_dict['_MAX_VALUES'], fields),
                 null_counts=value_dict['_NULL_COUNTS'],
             )
             file_meta = DataFileMeta(

paimon-python/pypaimon/manifest/schema/simple_stats.py

@@ -21,12 +21,13 @@
 from typing import ClassVar
 
 from pypaimon.table.row.generic_row import GenericRow
+from pypaimon.table.row.internal_row import InternalRow
 
 
 @dataclass
 class SimpleStats:
-    min_values: GenericRow
-    max_values: GenericRow
+    min_values: InternalRow
+    max_values: InternalRow
     null_counts: Optional[List[int]]
 
     _empty_stats: ClassVar[object] = None

paimon-python/pypaimon/manifest/simple_stats_evolution.py

@@ -0,0 +1,124 @@
+################################################################################
+#  Licensed to the Apache Software Foundation (ASF) under one
+#  or more contributor license agreements.  See the NOTICE file
+#  distributed with this work for additional information
+#  regarding copyright ownership.  The ASF licenses this file
+#  to you under the Apache License, Version 2.0 (the
+#  "License"); you may not use this file except in compliance
+#  with the License.  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+from typing import List, Optional, Dict, Any
+import threading
+
+from pypaimon.schema.data_types import DataField
+from pypaimon.manifest.schema.simple_stats import SimpleStats
+from pypaimon.table.row.generic_row import GenericRow
+from pypaimon.table.row.projected_row import ProjectedRow
+
+
+class SimpleStatsEvolution:
+    """Converter for array of SimpleColStats."""
+
+    def __init__(self, data_fields: List[DataField], index_mapping: Optional[List[int]],
+                 cast_field_getters: Optional[List[Any]]):
+        self.field_names = [field.name for field in data_fields]
+        self.index_mapping = index_mapping
+        self.cast_field_getters = cast_field_getters
+        self.index_mappings: Dict[tuple, List[int]] = {}
+        self._lock = threading.Lock()
+
+        # 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)
+
+    def evolution(self, stats: SimpleStats, row_count: Optional[int],
+                  stats_fields: Optional[List[str]]) -> 'SimpleStats':
+        min_values = stats.min_values
+        max_values = stats.max_values
+        null_counts = stats.null_counts
+
+        if stats_fields is not None and not stats_fields:
+            # Optimize for empty dense fields
+            min_values = self.empty_values
+            max_values = self.empty_values
+            null_counts = self.empty_null_counts
+        elif stats_fields is not None:
+            # Apply dense field mapping
+            dense_index_mapping = self._get_dense_index_mapping(stats_fields)
+            min_values = self._project_row(min_values, dense_index_mapping)
+            max_values = self._project_row(max_values, dense_index_mapping)
+            null_counts = self._project_array(null_counts, dense_index_mapping)
+
+        if self.index_mapping is not None:
+            # TODO support schema evolution
+            min_values = self._project_row(min_values, self.index_mapping)
+            max_values = self._project_row(max_values, self.index_mapping)
+
+            if row_count is None:
+                raise RuntimeError("Schema Evolution for stats needs row count.")
+
+            null_counts = self._evolve_null_counts(null_counts, self.index_mapping, row_count)
+
+        return SimpleStats(min_values, max_values, null_counts)
+
+    def _get_dense_index_mapping(self, dense_fields: List[str]) -> List[int]:
+        """
+        Get dense index mapping similar to Java:
+        fieldNames.stream().mapToInt(denseFields::indexOf).toArray()
+        """
+        dense_fields_tuple = tuple(dense_fields)
+
+        if dense_fields_tuple not in self.index_mappings:
+            with self._lock:
+                # Double-check locking
+                if dense_fields_tuple not in self.index_mappings:
+                    mapping = []
+                    for field_name in self.field_names:
+                        try:
+                            index = dense_fields.index(field_name)
+                            mapping.append(index)
+                        except ValueError:
+                            mapping.append(-1)  # Field not found
+                    self.index_mappings[dense_fields_tuple] = mapping
+
+        return self.index_mappings[dense_fields_tuple]
+
+    def _project_row(self, row: Any, index_mapping: List[int]) -> Any:
+        """Project row based on index mapping using ProjectedRow."""
+        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]:
+        """Project array based on index mapping."""
+        if not array:
+            return [0] * 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
+
+        return projected
+
+    def _evolve_null_counts(self, null_counts: 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):
+                evolved.append(null_counts[mapped_index])
+            else:
+                evolved.append(not_found_value)  # Use row count for missing fields
+
+        return evolved