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,25 @@ public class PartitionStatistics implements Serializable {
     @JsonProperty(FIELD_LAST_FILE_CREATION_TIME)
     protected final long lastFileCreationTime;
 
+    // defaults to 0 if this field is absent in the serialized data (e.g., from an older Paimon
+    // version)
+    @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 +107,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 +125,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 +148,8 @@ public String toString() {
                 + fileCount
                 + ", lastFileCreationTime="
                 + lastFileCreationTime
+                + ", totalBuckets="
+                + totalBuckets
                 + '}';
     }
 }

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

@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.partition;
+
+import org.apache.paimon.utils.JsonSerdeUtil;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link PartitionStatistics}. */
+public class PartitionStatisticsTest {
+
+    @Test
+    void testLegacyPartitionStatisticsDeserialization() {
+        String legacyPartitionStatisticsJson =
+                "{\"spec\":{\"pt\":\"1\"},\"recordCount\":100,\"fileSizeInBytes\":1024,\"fileCount\":2,\"lastFileCreationTime\":123456789}";
+        PartitionStatistics stats =
+                JsonSerdeUtil.fromJson(legacyPartitionStatisticsJson, PartitionStatistics.class);
+
+        assertThat(stats.spec()).containsEntry("pt", "1");
+        assertThat(stats.recordCount()).isEqualTo(100);
+        assertThat(stats.fileSizeInBytes()).isEqualTo(1024);
+        assertThat(stats.fileCount()).isEqualTo(2);
+        assertThat(stats.lastFileCreationTime()).isEqualTo(123456789L);
+        assertThat(stats.totalBuckets()).isEqualTo(0);
+    }
+}

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));
         }