[python] Fix $files reporting empty min/max value stats

paimon-python/pypaimon/table/system/files_table.py

@@ -24,6 +24,7 @@
 
 from pypaimon.manifest.manifest_file_manager import ManifestFileManager
 from pypaimon.manifest.manifest_list_manager import ManifestListManager
+from pypaimon.manifest.simple_stats_evolutions import SimpleStatsEvolutions
 from pypaimon.schema.data_types import (ArrayType, AtomicType, DataField,
                                         RowType)
 from pypaimon.table.system.system_table import SystemTable
@@ -79,13 +80,6 @@ def _stringify_path(value: Any) -> str:
     return str(value)
 
 
-def _stats_columns(file_meta, table_field_names: List[str]) -> List[str]:
-    cols = getattr(file_meta, "value_stats_cols", None)
-    if cols:
-        return list(cols)
-    return list(table_field_names)
-
-
 def _to_python(value: Any) -> Any:
     """Render an internal-row cell value into a JSON-safe primitive."""
     if value is None:
@@ -125,6 +119,16 @@ def _render_partition(partition_row) -> Optional[str]:
                     for field, value in zip(fields, values))
 
 
+def _row_values(row) -> List[Any]:
+    return [row.get_field(i) for i in range(len(row))]
+
+
+def _stats_fields(stats_row) -> List[str]:
+    # The manifest decodes a stats row with the fields its file stored,
+    # resolved against the file's own schema, so they name the row's positions.
+    return [field.name for field in stats_row.fields]
+
+
 def _render_stats_map(values: List[Any], columns: List[str]) -> str:
     pairs = {}
     n = min(len(columns), len(values) if values is not None else 0)
@@ -168,7 +172,9 @@ def _build_arrow_table(self) -> pyarrow.Table:
             manifest_files, drop_stats=False)
 
         file_format = self.base_table.options.file_format()
-        table_field_names = list(self.base_table.field_names)
+        stats_evolutions = SimpleStatsEvolutions(
+            self._schema_fields, self.base_table.table_schema.id)
+        stats_columns = [field.name for field in stats_evolutions.table_fields]
 
         rows = {
             "partition": [],
@@ -207,16 +213,19 @@ def _build_arrow_table(self) -> pyarrow.Table:
             rows["min_key"].append(_render_key(meta.min_key))
             rows["max_key"].append(_render_key(meta.max_key))
 
-            stats_cols = _stats_columns(meta, table_field_names)
-            value_stats = meta.value_stats
+            # Evolve stats to the current schema by field id, so a dropped or
+            # re-added column never inherits another column's stats.
+            value_stats = stats_evolutions.get_or_create(meta.schema_id).evolution(
+                meta.value_stats, meta.row_count,
+                _stats_fields(meta.value_stats.min_values))
             rows["null_value_counts"].append(
-                _render_null_counts(value_stats.null_counts, stats_cols))
+                _render_null_counts(value_stats.null_counts, stats_columns))
             rows["min_value_stats"].append(_render_stats_map(
-                getattr(value_stats.min_values, "values", []) or [],
-                stats_cols))
+                _row_values(value_stats.min_values),
+                stats_columns))
             rows["max_value_stats"].append(_render_stats_map(
-                getattr(value_stats.max_values, "values", []) or [],
-                stats_cols))
+                _row_values(value_stats.max_values),
+                stats_columns))
 
             rows["min_sequence_number"].append(int(meta.min_sequence_number))
             rows["max_sequence_number"].append(int(meta.max_sequence_number))
@@ -271,6 +280,12 @@ def _build_arrow_table(self) -> pyarrow.Table:
                 rows["write_cols"], type=_WRITE_COLS_TYPE),
         })
 
+    def _schema_fields(self, schema_id: int) -> List[DataField]:
+        table_schema = self.base_table.table_schema
+        if schema_id == table_schema.id:
+            return table_schema.fields
+        return self.base_table.schema_manager.get_schema(schema_id).fields
+
     @staticmethod
     def _empty_table() -> pyarrow.Table:
         return pyarrow.table({

paimon-python/pypaimon/tests/system/files_table_test.py

@@ -26,8 +26,9 @@
 import pyarrow as pa
 
 from pypaimon import CatalogFactory, Schema
-from pypaimon.schema.data_types import DataField
-from pypaimon.table.system.files_table import FilesTable
+from pypaimon.schema.data_types import AtomicType, DataField
+from pypaimon.schema.schema_change import SchemaChange
+from pypaimon.table.system.files_table import FilesTable, _row_values
 
 
 def _read(table):
@@ -148,6 +149,107 @@ def test_lists_files_with_partition_aggregation(self):
         for level in arrow_table.column("level").to_pylist():
             self.assertGreaterEqual(level, 0)
 
+    def test_value_stats_render_actual_values(self):
+        # A partitioned table with full stats: the value-stats maps must carry
+        # the real per-column values (including the partition column), not {}.
+        fields = [
+            DataField.from_dict({"id": 0, "name": "id", "type": "INT"}),
+            DataField.from_dict({"id": 1, "name": "v", "type": "STRING"}),
+            DataField.from_dict({"id": 2, "name": "dt", "type": "STRING"}),
+        ]
+        self.catalog.create_table(
+            "db.stats_t",
+            Schema(fields=fields, partition_keys=["dt"],
+                   options={"metadata.stats-mode": "full"}),
+            False,
+        )
+        table = self.catalog.get_table("db.stats_t")
+        write_builder = table.new_batch_write_builder()
+        writer = write_builder.new_write()
+        commit = write_builder.new_commit()
+        writer.write_arrow(pa.table({
+            "id": pa.array([1, 2, 3], type=pa.int32()),
+            "v": ["a", "b", "c"],
+            "dt": ["2024-01-01", "2024-01-01", "2024-01-02"],
+        }))
+        commit.commit(writer.prepare_commit())
+        writer.close()
+        commit.close()
+
+        arrow_table = _read(self.catalog.get_table("db.stats_t$files"))
+        self.assertGreater(arrow_table.num_rows, 0)
+        mins = [json.loads(s) for s in
+                arrow_table.column("min_value_stats").to_pylist()]
+        maxs = [json.loads(s) for s in
+                arrow_table.column("max_value_stats").to_pylist()]
+        # Real per-column values, not an empty {} (the BinaryRow-lacks-values bug).
+        self.assertTrue(any(m for m in mins),
+                        "min_value_stats all empty: {}".format(mins))
+        self.assertEqual(1, min(m["id"] for m in mins if "id" in m))
+        self.assertEqual(3, max(m["id"] for m in maxs if "id" in m))
+        # The partition column also lands in the stats map.
+        self.assertTrue(any("dt" in m for m in mins))
+
+    def _write(self, identifier, data):
+        write_builder = self.catalog.get_table(identifier).new_batch_write_builder()
+        writer = write_builder.new_write()
+        commit = write_builder.new_commit()
+        writer.write_arrow(data)
+        commit.commit(writer.prepare_commit())
+        writer.close()
+        commit.close()
+
+    def test_value_stats_follow_field_ids_after_drop_and_re_add(self):
+        fields = [
+            DataField.from_dict({"id": 0, "name": "id", "type": "INT"}),
+            DataField.from_dict({"id": 1, "name": "v", "type": "STRING"}),
+        ]
+        self.catalog.create_table(
+            "db.evolved",
+            Schema(fields=fields, options={"metadata.stats-mode": "full"}),
+            False,
+        )
+        self._write("db.evolved", pa.table({
+            "id": pa.array([1], type=pa.int32()), "v": ["old"]}))
+        self.catalog.alter_table(
+            "db.evolved", [SchemaChange.drop_column("v")], False)
+        self.catalog.alter_table(
+            "db.evolved", [SchemaChange.add_column("v", AtomicType("INT"))],
+            False)
+        self._write("db.evolved", pa.table({
+            "id": pa.array([2], type=pa.int32()),
+            "v": pa.array([20], type=pa.int32())}))
+
+        rows = sorted(_read(self.catalog.get_table("db.evolved$files"))
+                      .to_pylist(), key=lambda row: row["schema_id"])
+        self.assertEqual(2, len(rows))
+        old_file, new_file = rows
+        # The re-added ``v`` is a different column: the old file has no stats
+        # for it and all of its rows are null there.
+        self.assertEqual({"id": 1, "v": None},
+                         json.loads(old_file["min_value_stats"]))
+        self.assertEqual({"id": 1, "v": None},
+                         json.loads(old_file["max_value_stats"]))
+        self.assertEqual({"id": 0, "v": 1},
+                         json.loads(old_file["null_value_counts"]))
+        self.assertEqual({"id": 2, "v": 20},
+                         json.loads(new_file["min_value_stats"]))
+        self.assertEqual({"id": 2, "v": 20},
+                         json.loads(new_file["max_value_stats"]))
+        self.assertEqual({"id": 0, "v": 0},
+                         json.loads(new_file["null_value_counts"]))
+
+    def test_stats_decode_errors_are_not_swallowed(self):
+        class CorruptRow:
+            def __len__(self):
+                return 1
+
+            def get_field(self, pos):
+                raise ValueError("corrupt stats row")
+
+        with self.assertRaises(ValueError):
+            _row_values(CorruptRow())
+
 
 if __name__ == "__main__":
     unittest.main()