diff --git a/paimon-python/pypaimon/table/system/files_table.py b/paimon-python/pypaimon/table/system/files_table.py index 1b3bb4bb106f..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: @@ -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({ diff --git a/paimon-python/pypaimon/tests/system/files_table_test.py b/paimon-python/pypaimon/tests/system/files_table_test.py index 5d6b79447249..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): @@ -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()