From 1ce52ffcb83e50a5a678204364c7dcd4ecf6f46f Mon Sep 17 00:00:00 2001 From: jackylee-ch Date: Tue, 22 Sep 2026 21:30:46 +0800 Subject: [PATCH 1/2] [python] Fix $files reporting empty min/max value stats The $files system table rendered min_value_stats and max_value_stats as {} for every file, even when the manifest carries per-column value stats. _render_stats_map read the row values via getattr(row, "values", []), but on the read path the min/max rows are BinaryRow (built in manifest_file_manager), which exposes get_field/__len__ and has no values attribute -- so getattr yields [] and the map renders empty. Only GenericRow, the write-path type, has values, which is why null_value_counts and min_key/max_key render correctly. Add a _row_values helper that falls back to get_field(i) when values is absent, and use it for the two value-stats maps. min_key/max_key are GenericRow, so _render_key is left unchanged. A malformed or schema-evolved stats row degrades to {} rather than aborting the whole $files listing. --- .../pypaimon/table/system/files_table.py | 22 +++++++++- .../pypaimon/tests/system/files_table_test.py | 41 +++++++++++++++++++ 2 files changed, 61 insertions(+), 2 deletions(-) diff --git a/paimon-python/pypaimon/table/system/files_table.py b/paimon-python/pypaimon/table/system/files_table.py index 1b3bb4bb106f..571892a4efb2 100644 --- a/paimon-python/pypaimon/table/system/files_table.py +++ b/paimon-python/pypaimon/table/system/files_table.py @@ -125,6 +125,24 @@ def _render_partition(partition_row) -> Optional[str]: for field, value in zip(fields, values)) +def _row_values(row) -> List[Any]: + # ``GenericRow`` exposes ``values`` directly, but a row read back from a + # manifest is a ``BinaryRow`` that only offers ``get_field``/``__len__``. + # Fall back to those so value stats are not silently rendered as ``{}``. + if row is None: + return [] + values = getattr(row, "values", None) + if values is not None: + return values + try: + return [row.get_field(i) for i in range(len(row))] + except Exception: + # A row whose stored arity disagrees with the resolved fields (schema + # evolution) or whose bytes are malformed must degrade to {} rather than + # abort the whole $files listing, which is what the old code did. + return [] + + 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) @@ -212,10 +230,10 @@ def _build_arrow_table(self) -> pyarrow.Table: rows["null_value_counts"].append( _render_null_counts(value_stats.null_counts, stats_cols)) rows["min_value_stats"].append(_render_stats_map( - getattr(value_stats.min_values, "values", []) or [], + _row_values(value_stats.min_values), stats_cols)) rows["max_value_stats"].append(_render_stats_map( - getattr(value_stats.max_values, "values", []) or [], + _row_values(value_stats.max_values), stats_cols)) rows["min_sequence_number"].append(int(meta.min_sequence_number)) diff --git a/paimon-python/pypaimon/tests/system/files_table_test.py b/paimon-python/pypaimon/tests/system/files_table_test.py index 5d6b79447249..2ce1636182b3 100644 --- a/paimon-python/pypaimon/tests/system/files_table_test.py +++ b/paimon-python/pypaimon/tests/system/files_table_test.py @@ -148,6 +148,47 @@ 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)) + if __name__ == "__main__": unittest.main() From 1cd4c82a79cd484ae443c4a0b131c98f328c6560 Mon Sep 17 00:00:00 2001 From: jackylee-ch Date: Thu, 24 Sep 2026 15:08:06 +0800 Subject: [PATCH 2/2] [python] Evolve $files value stats to the current schema by field id Render min/max/null-count stats through SimpleStatsEvolutions, matching the Java FilesTable, so a dropped or re-added column no longer inherits stats from a column with the same name. Stats decode errors now propagate instead of being rendered as {}. --- .../pypaimon/table/system/files_table.py | 53 +++++++-------- .../pypaimon/tests/system/files_table_test.py | 65 ++++++++++++++++++- 2 files changed, 88 insertions(+), 30 deletions(-) diff --git a/paimon-python/pypaimon/table/system/files_table.py b/paimon-python/pypaimon/table/system/files_table.py index 571892a4efb2..fca9ab6d8481 100644 --- a/paimon-python/pypaimon/table/system/files_table.py +++ b/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: @@ -126,21 +120,13 @@ def _render_partition(partition_row) -> Optional[str]: def _row_values(row) -> List[Any]: - # ``GenericRow`` exposes ``values`` directly, but a row read back from a - # manifest is a ``BinaryRow`` that only offers ``get_field``/``__len__``. - # Fall back to those so value stats are not silently rendered as ``{}``. - if row is None: - return [] - values = getattr(row, "values", None) - if values is not None: - return values - try: - return [row.get_field(i) for i in range(len(row))] - except Exception: - # A row whose stored arity disagrees with the resolved fields (schema - # evolution) or whose bytes are malformed must degrade to {} rather than - # abort the whole $files listing, which is what the old code did. - return [] + 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: @@ -186,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": [], @@ -225,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( _row_values(value_stats.min_values), - stats_cols)) + stats_columns)) rows["max_value_stats"].append(_render_stats_map( _row_values(value_stats.max_values), - stats_cols)) + stats_columns)) rows["min_sequence_number"].append(int(meta.min_sequence_number)) rows["max_sequence_number"].append(int(meta.max_sequence_number)) @@ -289,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({ diff --git a/paimon-python/pypaimon/tests/system/files_table_test.py b/paimon-python/pypaimon/tests/system/files_table_test.py index 2ce1636182b3..525ab5224d4d 100644 --- a/paimon-python/pypaimon/tests/system/files_table_test.py +++ b/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): @@ -189,6 +190,66 @@ def test_value_stats_render_actual_values(self): # 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()