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 48ad4e2311 [python] Fix $files reporting empty min/max value stats
(#10111)
48ad4e2311 is described below
commit 48ad4e2311bd41bcc88fe1780c3b3a00be408e0d
Author: jackylee <[email protected]>
AuthorDate: Thu Sep 24 17:39:03 2026 +0800
[python] Fix $files reporting empty min/max value stats (#10111)
---
paimon-python/pypaimon/table/system/files_table.py | 45 ++++++---
.../pypaimon/tests/system/files_table_test.py | 106 ++++++++++++++++++++-
2 files changed, 134 insertions(+), 17 deletions(-)
diff --git a/paimon-python/pypaimon/table/system/files_table.py
b/paimon-python/pypaimon/table/system/files_table.py
index 1b3bb4bb10..fca9ab6d84 100644
--- a/paimon-python/pypaimon/table/system/files_table.py
+++ b/paimon-python/pypaimon/table/system/files_table.py
@@ -24,6 +24,7 @@ import pyarrow
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 @@ class FilesTable(SystemTable):
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 @@ class FilesTable(SystemTable):
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 @@ class FilesTable(SystemTable):
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 5d6b794472..525ab5224d 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 unittest
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 @@ class FilesTableTest(unittest.TestCase):
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()
