[wip]PartitionStatistics support totalBuckets

docs/static/rest-catalog-open-api.yaml

@@ -3252,6 +3252,9 @@ components:
         lastFileCreationTime:
           type: integer
           format: int64
+        totalBuckets:
+          type: integer
+          format: int32
         done:
           type: boolean
         createdAt:
@@ -3285,6 +3288,9 @@ components:
         lastFileCreationTime:
           type: integer
           format: int64
+        totalBuckets:
+          type: integer
+          format: int32
   #######################################
   # Examples of different values #
   #######################################

paimon-api/src/main/java/org/apache/paimon/partition/Partition.java

@@ -81,13 +81,14 @@ public Partition(
             @JsonProperty(FIELD_FILE_SIZE_IN_BYTES) long fileSizeInBytes,
             @JsonProperty(FIELD_FILE_COUNT) long fileCount,
             @JsonProperty(FIELD_LAST_FILE_CREATION_TIME) long lastFileCreationTime,
+            @JsonProperty(FIELD_TOTAL_BUCKETS) int totalBuckets,
             @JsonProperty(FIELD_DONE) boolean done,
             @JsonProperty(FIELD_CREATED_AT) @Nullable Long createdAt,
             @JsonProperty(FIELD_CREATED_BY) @Nullable String createdBy,
             @JsonProperty(FIELD_UPDATED_AT) @Nullable Long updatedAt,
             @JsonProperty(FIELD_UPDATED_BY) @Nullable String updatedBy,
             @JsonProperty(FIELD_OPTIONS) @Nullable Map<String, String> options) {
-        super(spec, recordCount, fileSizeInBytes, fileCount, lastFileCreationTime);
+        super(spec, recordCount, fileSizeInBytes, fileCount, lastFileCreationTime, totalBuckets);
         this.done = done;
         this.createdAt = createdAt;
         this.createdBy = createdBy;
@@ -102,13 +103,15 @@ public Partition(
             long fileSizeInBytes,
             long fileCount,
             long lastFileCreationTime,
+            int totalBuckets,
             boolean done) {
         this(
                 spec,
                 recordCount,
                 fileSizeInBytes,
                 fileCount,
                 lastFileCreationTime,
+                totalBuckets,
                 done,
                 null,
                 null,
@@ -188,6 +191,8 @@ public String toString() {
                 + fileCount
                 + ", lastFileCreationTime="
                 + lastFileCreationTime
+                + ", totalBuckets="
+                + totalBuckets
                 + ", done="
                 + done
                 + ", createdAt="

paimon-api/src/main/java/org/apache/paimon/partition/PartitionStatistics.java

@@ -44,6 +44,7 @@ public class PartitionStatistics implements Serializable {
     public static final String FIELD_FILE_SIZE_IN_BYTES = "fileSizeInBytes";
     public static final String FIELD_FILE_COUNT = "fileCount";
     public static final String FIELD_LAST_FILE_CREATION_TIME = "lastFileCreationTime";
+    public static final String FIELD_TOTAL_BUCKETS = "totalBuckets";
 
     @JsonProperty(FIELD_SPEC)
     protected final Map<String, String> spec;
@@ -60,18 +61,23 @@ public class PartitionStatistics implements Serializable {
     @JsonProperty(FIELD_LAST_FILE_CREATION_TIME)
     protected final long lastFileCreationTime;
 
+    @JsonProperty(FIELD_TOTAL_BUCKETS)
+    protected final int totalBuckets;
+
     @JsonCreator
     public PartitionStatistics(
             @JsonProperty(FIELD_SPEC) Map<String, String> spec,
             @JsonProperty(FIELD_RECORD_COUNT) long recordCount,
             @JsonProperty(FIELD_FILE_SIZE_IN_BYTES) long fileSizeInBytes,
             @JsonProperty(FIELD_FILE_COUNT) long fileCount,
-            @JsonProperty(FIELD_LAST_FILE_CREATION_TIME) long lastFileCreationTime) {
+            @JsonProperty(FIELD_LAST_FILE_CREATION_TIME) long lastFileCreationTime,
+            @JsonProperty(FIELD_TOTAL_BUCKETS) int totalBuckets) {
         this.spec = spec;
         this.recordCount = recordCount;
         this.fileSizeInBytes = fileSizeInBytes;
         this.fileCount = fileCount;
         this.lastFileCreationTime = lastFileCreationTime;
+        this.totalBuckets = totalBuckets;
     }
 
     @JsonGetter(FIELD_SPEC)
@@ -99,6 +105,11 @@ public long lastFileCreationTime() {
         return lastFileCreationTime;
     }
 
+    @JsonGetter(FIELD_TOTAL_BUCKETS)
+    public int totalBuckets() {
+        return totalBuckets;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {
@@ -112,12 +123,14 @@ public boolean equals(Object o) {
                 && fileSizeInBytes == that.fileSizeInBytes
                 && fileCount == that.fileCount
                 && lastFileCreationTime == that.lastFileCreationTime
+                && totalBuckets == that.totalBuckets
                 && Objects.equals(spec, that.spec);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(spec, recordCount, fileSizeInBytes, fileCount, lastFileCreationTime);
+        return Objects.hash(
+                spec, recordCount, fileSizeInBytes, fileCount, lastFileCreationTime, totalBuckets);
     }
 
     @Override
@@ -133,6 +146,8 @@ public String toString() {
                 + fileCount
                 + ", lastFileCreationTime="
                 + lastFileCreationTime
+                + ", totalBuckets="
+                + totalBuckets
                 + '}';
     }
 }

paimon-api/src/test/java/org/apache/paimon/partition/PartitionTest.java

@@ -40,6 +40,7 @@ void testJsonSerializationWithNullValues() {
                         1024L, // fileSizeInBytes
                         2L, // fileCount
                         System.currentTimeMillis(), // lastFileCreationTime
+                        10, // totalBuckets
                         false, // done
                         null, // createdAt
                         null, // createdBy
@@ -57,6 +58,7 @@ void testJsonSerializationWithNullValues() {
 
         assertThat(json).contains("done");
         assertThat(json).contains("recordCount");
+        assertThat(json).contains("totalBuckets");
     }
 
     @Test
@@ -69,6 +71,7 @@ void testJsonSerializationWithNonNullValues() {
                         1024L,
                         2L,
                         System.currentTimeMillis(),
+                        10, // totalBuckets
                         true,
                         1234567890L, // createdAt
                         "user1", // createdBy
@@ -78,6 +81,7 @@ void testJsonSerializationWithNonNullValues() {
 
         String json = JsonSerdeUtil.toFlatJson(partition);
 
+        assertThat(json).contains("totalBuckets");
         assertThat(json).contains("createdAt");
         assertThat(json).contains("createdBy");
         assertThat(json).contains("updatedAt");

paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java

@@ -1073,6 +1073,7 @@ List<String> authTableQuery(Identifier identifier, @Nullable List<String> select
     String NUM_FILES_PROP = "numFiles";
     String TOTAL_SIZE_PROP = "totalSize";
     String LAST_UPDATE_TIME_PROP = "lastUpdateTime";
+    String TOTAL_BUCKETS = "totalBuckets";
 
     // ======================= Exceptions ===============================
 

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

@@ -30,6 +30,7 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Optional;
 
 import static org.apache.paimon.manifest.FileKind.ADD;
 import static org.apache.paimon.manifest.FileKind.DELETE;
@@ -43,18 +44,21 @@ public class PartitionEntry {
     private final long fileSizeInBytes;
     private final long fileCount;
     private final long lastFileCreationTime;
+    private final int totalBuckets;
 
     public PartitionEntry(
             BinaryRow partition,
             long recordCount,
             long fileSizeInBytes,
             long fileCount,
-            long lastFileCreationTime) {
+            long lastFileCreationTime,
+            int totalBuckets) {
         this.partition = partition;
         this.recordCount = recordCount;
         this.fileSizeInBytes = fileSizeInBytes;
         this.fileCount = fileCount;
         this.lastFileCreationTime = lastFileCreationTime;
+        this.totalBuckets = totalBuckets;
     }
 
     public BinaryRow partition() {
@@ -77,13 +81,18 @@ public long lastFileCreationTime() {
         return lastFileCreationTime;
     }
 
+    public int totalBuckets() {
+        return totalBuckets;
+    }
+
     public PartitionEntry merge(PartitionEntry entry) {
         return new PartitionEntry(
                 partition,
                 recordCount + entry.recordCount,
                 fileSizeInBytes + entry.fileSizeInBytes,
                 fileCount + entry.fileCount,
-                Math.max(lastFileCreationTime, entry.lastFileCreationTime));
+                Math.max(lastFileCreationTime, entry.lastFileCreationTime),
+                entry.totalBuckets);
     }
 
     public Partition toPartition(InternalRowPartitionComputer computer) {
@@ -93,6 +102,7 @@ public Partition toPartition(InternalRowPartitionComputer computer) {
                 fileSizeInBytes,
                 fileCount,
                 lastFileCreationTime,
+                totalBuckets,
                 false);
     }
 
@@ -102,15 +112,16 @@ public PartitionStatistics toPartitionStatistics(InternalRowPartitionComputer co
                 recordCount,
                 fileSizeInBytes,
                 fileCount,
-                lastFileCreationTime);
+                lastFileCreationTime,
+                totalBuckets);
     }
 
     public static PartitionEntry fromManifestEntry(ManifestEntry entry) {
-        return fromDataFile(entry.partition(), entry.kind(), entry.file());
+        return fromDataFile(entry.partition(), entry.kind(), entry.file(), entry.totalBuckets());
     }
 
     public static PartitionEntry fromDataFile(
-            BinaryRow partition, FileKind kind, DataFileMeta file) {
+            BinaryRow partition, FileKind kind, DataFileMeta file, int totalBuckets) {
         long recordCount = file.rowCount();
         long fileSizeInBytes = file.fileSize();
         long fileCount = 1;
@@ -120,7 +131,12 @@ public static PartitionEntry fromDataFile(
             fileCount = -fileCount;
         }
         return new PartitionEntry(
-                partition, recordCount, fileSizeInBytes, fileCount, file.creationTimeEpochMillis());
+                partition,
+                recordCount,
+                fileSizeInBytes,
+                fileCount,
+                file.creationTimeEpochMillis(),
+                totalBuckets);
     }
 
     public static Collection<PartitionEntry> merge(Collection<ManifestEntry> fileEntries) {
@@ -139,7 +155,12 @@ public static Collection<PartitionEntry> mergeSplits(Collection<DataSplit> split
         for (DataSplit split : splits) {
             BinaryRow partition = split.partition();
             for (DataFileMeta file : split.dataFiles()) {
-                PartitionEntry partitionEntry = fromDataFile(partition, ADD, file);
+                PartitionEntry partitionEntry =
+                        fromDataFile(
+                                partition,
+                                ADD,
+                                file,
+                                Optional.ofNullable(split.totalBuckets()).orElse(0));
                 partitions.compute(
                         partition,
                         (part, old) -> old == null ? partitionEntry : old.merge(partitionEntry));
@@ -170,12 +191,18 @@ public boolean equals(Object o) {
                 && fileSizeInBytes == that.fileSizeInBytes
                 && fileCount == that.fileCount
                 && lastFileCreationTime == that.lastFileCreationTime
+                && totalBuckets == that.totalBuckets
                 && Objects.equals(partition, that.partition);
     }
 
     @Override
     public int hashCode() {
         return Objects.hash(
-                partition, recordCount, fileSizeInBytes, fileCount, lastFileCreationTime);
+                partition,
+                recordCount,
+                fileSizeInBytes,
+                fileCount,
+                lastFileCreationTime,
+                totalBuckets);
     }
 }

paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java

@@ -107,7 +107,7 @@ public List<PartitionEntry> listPartitionEntries() {
         List<PartitionEntry> partitionEntries = new ArrayList<>();
         for (Pair<LinkedHashMap<String, String>, Path> partition2Path : partition2Paths) {
             BinaryRow row = toPartitionRow(partition2Path.getKey());
-            partitionEntries.add(new PartitionEntry(row, -1L, -1L, -1L, -1L));
+            partitionEntries.add(new PartitionEntry(row, -1L, -1L, -1L, -1L, -1));
         }
         return partitionEntries;
     }

paimon-core/src/main/java/org/apache/paimon/table/system/PartitionsTable.java

@@ -301,7 +301,8 @@ private PartitionEntry toPartitionEntry(Partition partition) {
                     partition.recordCount(),
                     partition.fileSizeInBytes(),
                     partition.fileCount(),
-                    partition.lastFileCreationTime());
+                    partition.lastFileCreationTime(),
+                    partition.totalBuckets());
         }
 
         private Timestamp toTimestamp(Long epochMillis) {

paimon-core/src/main/java/org/apache/paimon/utils/PartitionStatisticsReporter.java

@@ -37,6 +37,7 @@
 import java.util.Collections;
 import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.Optional;
 
 import static org.apache.paimon.utils.PartitionPathUtils.extractPartitionSpecFromPath;
 
@@ -73,18 +74,25 @@ public void report(String partition, long modifyTimeMillis) throws Exception {
             long rowCount = 0;
             long totalSize = 0;
             long fileCount = 0;
+            int totalBuckets = 0;
             for (DataSplit split : splits) {
                 List<DataFileMeta> fileMetas = split.dataFiles();
                 fileCount += fileMetas.size();
                 for (DataFileMeta fileMeta : fileMetas) {
                     rowCount += fileMeta.rowCount();
                     totalSize += fileMeta.fileSize();
                 }
+                totalBuckets = Optional.ofNullable(split.totalBuckets()).orElse(0);
             }
 
             PartitionStatistics partitionStats =
                     new PartitionStatistics(
-                            partitionSpec, rowCount, totalSize, fileCount, modifyTimeMillis);
+                            partitionSpec,
+                            rowCount,
+                            totalSize,
+                            fileCount,
+                            modifyTimeMillis,
+                            totalBuckets);
             LOG.info("alter partition {} with statistic {}.", partitionSpec, partitionStats);
             partitionHandler.alterPartitions(Collections.singletonList(partitionStats));
         }

paimon-core/src/test/java/org/apache/paimon/partition/PartitionExpireTableTest.java

@@ -51,7 +51,7 @@ public void testCustomExpire() throws Exception {
 
         Table table = catalog.getTable(identifier());
         String path = table.options().get("path");
-        PartitionEntry expire = new PartitionEntry(BinaryRow.singleColumn(1), 1, 1, 1, 1);
+        PartitionEntry expire = new PartitionEntry(BinaryRow.singleColumn(1), 1, 1, 1, 1, 1);
         TABLE_EXPIRE_PARTITIONS.put(path, Collections.singletonList(expire));
         write(table, GenericRow.of(1, 1));
         write(table, GenericRow.of(2, 2));

paimon-core/src/test/java/org/apache/paimon/rest/MockRESTMessage.java

@@ -150,7 +150,7 @@ public static AlterTableRequest alterTableRequest() {
     public static ListPartitionsResponse listPartitionsResponse() {
         Map<String, String> spec = new HashMap<>();
         spec.put("f0", "1");
-        Partition partition = new Partition(spec, 1, 1, 1, 1, false);
+        Partition partition = new Partition(spec, 1, 1, 1, 1, 1, false);
         return new ListPartitionsResponse(ImmutableList.of(partition));
     }
 

paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java

@@ -2405,6 +2405,7 @@ private MockResponse commitSnapshot(
                                                                                 .lastFileCreationTime(),
                                                                         stats
                                                                                 .lastFileCreationTime()),
+                                                                stats.totalBuckets(),
                                                                 oldPartition.done(),
                                                                 oldPartition.createdAt(),
                                                                 oldPartition.createdBy(),
@@ -2630,6 +2631,7 @@ private Partition toPartition(PartitionStatistics stats) {
                 stats.fileSizeInBytes(),
                 stats.fileCount(),
                 stats.lastFileCreationTime(),
+                stats.totalBuckets(),
                 false,
                 System.currentTimeMillis(),
                 "created",

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, totalBuckets=-1}");
         action.close();
         assertThat(closed).isTrue();
     }