fix fail tests

Author: mingfeng

paimon-core/src/main/java/org/apache/paimon/manifest/PartitionEntry.java

@@ -81,6 +81,10 @@ public long lastFileCreationTime() {
         return lastFileCreationTime;
     }
 
+    public int bucketCount() {
+        return bucketCount;
+    }
+
     public PartitionEntry merge(PartitionEntry entry) {
         return new PartitionEntry(
                 partition,

paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java

@@ -1165,7 +1165,10 @@ private List<PartitionStatistics> toPartitionStatistics(
             Map<BinaryRow, Integer> bucketCountMap = scanner.readBucketCounts(snapshot, partitions);
             for (PartitionEntry entry : deltaStatistics) {
                 Integer bucketCount = bucketCountMap.get(entry.partition());
-                statistics.add(entry.toPartitionStatistics(partitionComputer, bucketCount));
+                statistics.add(
+                        entry.toPartitionStatistics(
+                                partitionComputer,
+                                Optional.ofNullable(bucketCount).orElse(entry.bucketCount())));
             }
         } else {
             for (PartitionEntry entry : deltaStatistics) {

paimon-core/src/test/java/org/apache/paimon/utils/PartitionStatisticsReporterTest.java

@@ -135,7 +135,7 @@ public void close() {
         assertThat(partitionParams).containsKey("c1=a/");
         assertThat(partitionParams.get("c1=a/").toString())
                 .isEqualTo(
-                        "{spec={c1=a}, recordCount=2, fileSizeInBytes=705, fileCount=1, lastFileCreationTime=1729598544974}");
+                        "{spec={c1=a}, recordCount=2, fileSizeInBytes=705, fileCount=1, lastFileCreationTime=1729598544974, bucketCount=0}");
         action.close();
         assertThat(closed).isTrue();
     }

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RESTCatalogITCase.java

@@ -19,6 +19,7 @@
 package org.apache.paimon.flink;
 
 import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.partition.Partition;
 import org.apache.paimon.rest.RESTToken;
 
 import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableList;
@@ -192,11 +193,12 @@ public void testBucketCountStatistics() throws Exception {
 
     private void validateBucketCount(
             String databaseName, String tableName, Integer expectedBucketCount) throws Exception {
-        Catalog catalog = tEnv.getCatalog(tEnv.getCurrentCatalog()).get();
-        org.apache.paimon.catalog.Catalog paimonCatalog = ((FlinkCatalog) catalog).catalog();
-        List<org.apache.paimon.partition.Partition> partitions =
-                paimonCatalog.listPartitions(Identifier.create(databaseName, tableName));
-        assertThat(partitions).isNotEmpty();
-        assertThat(partitions.get(0).bucketCount()).isEqualTo(expectedBucketCount);
+        Catalog flinkCatalog = tEnv.getCatalog(tEnv.getCurrentCatalog()).get();
+        try (org.apache.paimon.catalog.Catalog catalog = ((FlinkCatalog) flinkCatalog).catalog()) {
+            List<Partition> partitions =
+                    catalog.listPartitions(Identifier.create(databaseName, tableName));
+            assertThat(partitions).isNotEmpty();
+            assertThat(partitions.get(0).bucketCount()).isEqualTo(expectedBucketCount);
+        }
     }
 }