[python] Let Python write file without value stats by default

Author: JingsongLi

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

@@ -37,7 +37,7 @@ def empty_stats(cls):
         if cls._empty_stats is None:
             min_values = GenericRow([], [])
             max_values = GenericRow([], [])
-            cls._empty_stats = cls(min_values, max_values, None)
+            cls._empty_stats = cls(min_values, max_values, [])
         return cls._empty_stats
 
 

paimon-python/pypaimon/write/writer/data_blob_writer.py

@@ -196,13 +196,15 @@ def _split_data(self, data: pa.RecordBatch) -> Tuple[pa.RecordBatch, pa.RecordBa
 
         return normal_data, blob_data
 
-    def _process_normal_data(self, data: pa.RecordBatch) -> pa.Table:
+    @staticmethod
+    def _process_normal_data(data: pa.RecordBatch) -> pa.Table:
         """Process normal data (similar to base DataWriter)."""
         if data is None or data.num_rows == 0:
             return pa.Table.from_batches([])
         return pa.Table.from_batches([data])
 
-    def _merge_normal_data(self, existing_data: pa.Table, new_data: pa.Table) -> pa.Table:
+    @staticmethod
+    def _merge_normal_data(existing_data: pa.Table, new_data: pa.Table) -> pa.Table:
         return pa.concat_tables([existing_data, new_data])
 
     def _should_roll_normal(self) -> bool:
@@ -243,7 +245,7 @@ def _close_current_writers(self):
         logger.info(f"Closed both writers - normal: {normal_meta.file_name}, "
                     f"added {len(blob_metas)} blob file metadata after normal metadata")
 
-    def _write_normal_data_to_file(self, data: pa.Table) -> DataFileMeta:
+    def _write_normal_data_to_file(self, data: pa.Table) -> Optional[DataFileMeta]:
         if data.num_rows == 0:
             return None
 
@@ -270,37 +272,15 @@ def _write_normal_data_to_file(self, data: pa.Table) -> DataFileMeta:
 
     def _create_data_file_meta(self, file_name: str, file_path: str, data: pa.Table,
                                external_path: Optional[str] = None) -> DataFileMeta:
-        # Column stats (only for normal columns)
-        column_stats = {
-            field.name: self._get_column_stats(data, field.name)
-            for field in self.table.table_schema.fields
-            if field.name != self.blob_column_name
-        }
-
-        # Get normal fields only
-        normal_fields = [field for field in self.table.table_schema.fields
-                         if field.name != self.blob_column_name]
-
-        min_value_stats = [column_stats[field.name]['min_values'] for field in normal_fields]
-        max_value_stats = [column_stats[field.name]['max_values'] for field in normal_fields]
-        value_null_counts = [column_stats[field.name]['null_counts'] for field in normal_fields]
-
         self.sequence_generator.start = self.sequence_generator.current
-
         return DataFileMeta.create(
             file_name=file_name,
             file_size=self.file_io.get_file_size(file_path),
             row_count=data.num_rows,
             min_key=GenericRow([], []),
             max_key=GenericRow([], []),
-            key_stats=SimpleStats(
-                GenericRow([], []),
-                GenericRow([], []),
-                []),
-            value_stats=SimpleStats(
-                GenericRow(min_value_stats, normal_fields),
-                GenericRow(max_value_stats, normal_fields),
-                value_null_counts),
+            key_stats=SimpleStats.empty_stats(),
+            value_stats=SimpleStats.empty_stats(),
             min_sequence_number=-1,
             max_sequence_number=-1,
             schema_id=self.table.table_schema.id,
@@ -309,7 +289,7 @@ def _create_data_file_meta(self, file_name: str, file_path: str, data: pa.Table,
             creation_time=Timestamp.now(),
             delete_row_count=0,
             file_source=0,
-            value_stats_cols=self.normal_column_names,
+            value_stats_cols=[],
             external_path=external_path,
             file_path=file_path,
             write_cols=self.write_cols)

paimon-python/pypaimon/write/writer/data_writer.py

@@ -26,7 +26,6 @@
 from pypaimon.data.timestamp import Timestamp
 from pypaimon.manifest.schema.data_file_meta import DataFileMeta
 from pypaimon.manifest.schema.simple_stats import SimpleStats
-from pypaimon.schema.data_types import PyarrowFieldParser
 from pypaimon.table.bucket_mode import BucketMode
 from pypaimon.table.row.generic_row import GenericRow
 
@@ -190,21 +189,14 @@ def _write_data_to_file(self, data: pa.Table):
         min_key = [col.to_pylist()[0] for col in min_key_row_batch.columns]
         max_key = [col.to_pylist()[0] for col in max_key_row_batch.columns]
 
-        # key stats & value stats
-        data_fields = self.table.fields if self.table.is_primary_key_table \
-            else PyarrowFieldParser.to_paimon_schema(data.schema)
-        column_stats = {
+        key_column_stats = {
             field.name: self._get_column_stats(data, field.name)
-            for field in data_fields
+            for field in self.table.trimmed_primary_keys
         }
-        all_fields = data_fields
-        min_value_stats = [column_stats[field.name]['min_values'] for field in all_fields]
-        max_value_stats = [column_stats[field.name]['max_values'] for field in all_fields]
-        value_null_counts = [column_stats[field.name]['null_counts'] for field in all_fields]
         key_fields = self.trimmed_primary_keys_fields
-        min_key_stats = [column_stats[field.name]['min_values'] for field in key_fields]
-        max_key_stats = [column_stats[field.name]['max_values'] for field in key_fields]
-        key_null_counts = [column_stats[field.name]['null_counts'] for field in key_fields]
+        min_key_stats = [key_column_stats[field.name]['min_values'] for field in key_fields]
+        max_key_stats = [key_column_stats[field.name]['max_values'] for field in key_fields]
+        key_null_counts = [key_column_stats[field.name]['null_counts'] for field in key_fields]
         if not all(count == 0 for count in key_null_counts):
             raise RuntimeError("Primary key should not be null")
 
@@ -222,11 +214,7 @@ def _write_data_to_file(self, data: pa.Table):
                 GenericRow(max_key_stats, self.trimmed_primary_keys_fields),
                 key_null_counts,
             ),
-            value_stats=SimpleStats(
-                GenericRow(min_value_stats, data_fields),
-                GenericRow(max_value_stats, data_fields),
-                value_null_counts,
-            ),
+            value_stats=SimpleStats.empty_stats(),
             min_sequence_number=min_seq,
             max_sequence_number=max_seq,
             schema_id=self.table.table_schema.id,
@@ -235,7 +223,7 @@ def _write_data_to_file(self, data: pa.Table):
             creation_time=Timestamp.now(),
             delete_row_count=0,
             file_source=0,
-            value_stats_cols=None,  # None means all columns in the data have statistics
+            value_stats_cols=[],
             external_path=external_path_str,  # Set external path if using external paths
             first_row_id=None,
             write_cols=self.write_cols,