[python] Add totalRecordCount and deltaRecordCount in Snapshot (apache#6141)

Author: JingsongLi

paimon-python/pypaimon/tests/writer_test.py

@@ -73,3 +73,22 @@ def test_writer(self):
         self.assertTrue(os.path.exists(self.warehouse + "/test_db.db/test_table/bucket-0"))
         self.assertEqual(len(glob.glob(self.warehouse + "/test_db.db/test_table/manifest/*.avro")), 2)
         self.assertEqual(len(glob.glob(self.warehouse + "/test_db.db/test_table/bucket-0/*.parquet")), 1)
+
+        with open(self.warehouse + '/test_db.db/test_table/snapshot/snapshot-1', 'r', encoding='utf-8') as file:
+            content = ''.join(file.readlines())
+            self.assertTrue(content.__contains__('\"totalRecordCount\": 3'))
+            self.assertTrue(content.__contains__('\"deltaRecordCount\": 3'))
+
+        write_builder = table.new_batch_write_builder()
+        table_write = write_builder.new_write()
+        table_commit = write_builder.new_commit()
+        table_write.write_arrow(expect)
+        commit_messages = table_write.prepare_commit()
+        table_commit.commit(commit_messages)
+        table_write.close()
+        table_commit.close()
+
+        with open(self.warehouse + '/test_db.db/test_table/snapshot/snapshot-2', 'r', encoding='utf-8') as file:
+            content = ''.join(file.readlines())
+            self.assertTrue(content.__contains__('\"totalRecordCount\": 6'))
+            self.assertTrue(content.__contains__('\"deltaRecordCount\": 3'))

paimon-python/pypaimon/write/file_store_commit.py

@@ -58,10 +58,19 @@ def commit(self, commit_messages: List[CommitMessage], commit_identifier: int):
         new_manifest_files = self.manifest_file_manager.write(commit_messages)
         if not new_manifest_files:
             return
+
         latest_snapshot = self.snapshot_manager.get_latest_snapshot()
+
         existing_manifest_files = []
+        record_count_add = self._generate_record_count_add(commit_messages)
+        total_record_count = record_count_add
+
         if latest_snapshot:
             existing_manifest_files = self.manifest_list_manager.read_all_manifest_files(latest_snapshot)
+            previous_record_count = latest_snapshot.total_record_count
+            if previous_record_count:
+                total_record_count += previous_record_count
+
         new_manifest_files.extend(existing_manifest_files)
         manifest_list = self.manifest_list_manager.write(new_manifest_files)
 
@@ -72,6 +81,8 @@ def commit(self, commit_messages: List[CommitMessage], commit_identifier: int):
             schema_id=0,
             base_manifest_list=manifest_list,
             delta_manifest_list=manifest_list,
+            total_record_count=total_record_count,
+            delta_record_count=record_count_add,
             commit_user=self.commit_user,
             commit_identifier=commit_identifier,
             commit_kind="APPEND",
@@ -100,13 +111,17 @@ def overwrite(self, partition, commit_messages: List[CommitMessage], commit_iden
         # In overwrite mode, we don't merge with existing manifests
         manifest_list = self.manifest_list_manager.write(new_manifest_files)
 
+        record_count_add = self._generate_record_count_add(commit_messages)
+
         new_snapshot_id = self._generate_snapshot_id()
         snapshot_data = Snapshot(
             version=3,
             id=new_snapshot_id,
             schema_id=0,
             base_manifest_list=manifest_list,
             delta_manifest_list=manifest_list,
+            total_record_count=record_count_add,
+            delta_record_count=record_count_add,
             commit_user=self.commit_user,
             commit_identifier=commit_identifier,
             commit_kind="OVERWRITE",
@@ -234,3 +249,25 @@ def _generate_partition_statistics(self, commit_messages: List[CommitMessage]) -
             )
             for stats in partition_stats.values()
         ]
+
+    def _generate_record_count_add(self, commit_messages: List[CommitMessage]) -> int:
+        """
+        Generate record count add from commit messages.
+
+        This method follows the Java implementation pattern from
+        org.apache.paimon.manifest.ManifestEntry.recordCountAdd().
+
+        Args:
+            commit_messages: List of commit messages to analyze
+
+        Returns:
+            Count of add record
+        """
+        record_count = 0
+
+        for message in commit_messages:
+            new_files = message.new_files()
+            for file_meta in new_files:
+                record_count += file_meta.row_count
+
+        return record_count