[core] Introduce bucket entries to optimize Spark compact (#4162)

Author: JingsongLi

docs/content/maintenance/system-tables.md

@@ -309,6 +309,22 @@ SELECT * FROM my_table$partitions;
 */
 ```
 
+### Buckets Table
+
+You can query the bucket files of the table.
+
+```sql
+SELECT * FROM my_table$buckets;
+
+/*
++---------------+--------+----------------+--------------------+--------------------+------------------------+
+|  partition    | bucket |   record_count |  file_size_in_bytes|          file_count|        last_update_time|
++---------------+--------+----------------+--------------------+--------------------+------------------------+
+|  [1]          |   0    |           1    |             645    |                1   | 2024-06-24 10:25:57.400|
++---------------+--------+----------------+--------------------+--------------------+------------------------+
+*/
+```
+
 ## Global System Table
 
 Global system tables contain the statistical information of all the tables exists in paimon. For convenient of searching, we create a reference system database called `sys`.

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

@@ -0,0 +1,143 @@
+/*
+ * 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.manifest;
+
+import org.apache.paimon.annotation.Public;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.utils.Pair;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+/** Entry representing a bucket. */
+@Public
+public class BucketEntry {
+
+    private final BinaryRow partition;
+    private final int bucket;
+    private final long recordCount;
+    private final long fileSizeInBytes;
+    private final long fileCount;
+    private final long lastFileCreationTime;
+
+    public BucketEntry(
+            BinaryRow partition,
+            int bucket,
+            long recordCount,
+            long fileSizeInBytes,
+            long fileCount,
+            long lastFileCreationTime) {
+        this.partition = partition;
+        this.bucket = bucket;
+        this.recordCount = recordCount;
+        this.fileSizeInBytes = fileSizeInBytes;
+        this.fileCount = fileCount;
+        this.lastFileCreationTime = lastFileCreationTime;
+    }
+
+    public BinaryRow partition() {
+        return partition;
+    }
+
+    public int bucket() {
+        return bucket;
+    }
+
+    public long recordCount() {
+        return recordCount;
+    }
+
+    public long fileSizeInBytes() {
+        return fileSizeInBytes;
+    }
+
+    public long fileCount() {
+        return fileCount;
+    }
+
+    public long lastFileCreationTime() {
+        return lastFileCreationTime;
+    }
+
+    public BucketEntry merge(BucketEntry entry) {
+        return new BucketEntry(
+                partition,
+                bucket,
+                recordCount + entry.recordCount,
+                fileSizeInBytes + entry.fileSizeInBytes,
+                fileCount + entry.fileCount,
+                Math.max(lastFileCreationTime, entry.lastFileCreationTime));
+    }
+
+    public static BucketEntry fromManifestEntry(ManifestEntry entry) {
+        PartitionEntry partitionEntry = PartitionEntry.fromManifestEntry(entry);
+        return new BucketEntry(
+                partitionEntry.partition(),
+                entry.bucket(),
+                partitionEntry.recordCount(),
+                partitionEntry.fileSizeInBytes(),
+                partitionEntry.fileCount(),
+                partitionEntry.lastFileCreationTime());
+    }
+
+    public static Collection<BucketEntry> merge(Collection<ManifestEntry> fileEntries) {
+        Map<Pair<BinaryRow, Integer>, BucketEntry> buckets = new HashMap<>();
+        for (ManifestEntry entry : fileEntries) {
+            BucketEntry bucketEntry = fromManifestEntry(entry);
+            buckets.compute(
+                    Pair.of(entry.partition(), entry.bucket()),
+                    (part, old) -> old == null ? bucketEntry : old.merge(bucketEntry));
+        }
+        return buckets.values();
+    }
+
+    public static void merge(
+            Collection<BucketEntry> from, Map<Pair<BinaryRow, Integer>, BucketEntry> to) {
+        for (BucketEntry entry : from) {
+            to.compute(
+                    Pair.of(entry.partition(), entry.bucket),
+                    (part, old) -> old == null ? entry : old.merge(entry));
+        }
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        BucketEntry that = (BucketEntry) o;
+        return recordCount == that.recordCount
+                && fileSizeInBytes == that.fileSizeInBytes
+                && fileCount == that.fileCount
+                && lastFileCreationTime == that.lastFileCreationTime
+                && bucket == that.bucket
+                && Objects.equals(partition, that.partition);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(
+                partition, bucket, recordCount, fileSizeInBytes, fileCount, lastFileCreationTime);
+    }
+}

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

@@ -22,6 +22,7 @@
 import org.apache.paimon.Snapshot;
 import org.apache.paimon.data.BinaryRow;
 import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.manifest.BucketEntry;
 import org.apache.paimon.manifest.FileEntry;
 import org.apache.paimon.manifest.ManifestCacheFilter;
 import org.apache.paimon.manifest.ManifestEntry;
@@ -55,7 +56,6 @@
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.ThreadPoolExecutor;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.stream.Collectors;
@@ -259,18 +259,28 @@ public List<SimpleFileEntry> readSimpleEntries() {
     public List<PartitionEntry> readPartitionEntries() {
         List<ManifestFileMeta> manifests = readManifests().getRight();
         Map<BinaryRow, PartitionEntry> partitions = new ConcurrentHashMap<>();
-        // Can be executed in disorder
-        ThreadPoolExecutor executor = getExecutorService(scanManifestParallelism);
         Consumer<ManifestFileMeta> processor =
                 m ->
                         PartitionEntry.merge(
                                 PartitionEntry.merge(readManifestFileMeta(m)), partitions);
-        randomlyOnlyExecute(executor, processor, manifests);
+        randomlyOnlyExecute(getExecutorService(scanManifestParallelism), processor, manifests);
         return partitions.values().stream()
                 .filter(p -> p.fileCount() > 0)
                 .collect(Collectors.toList());
     }
 
+    @Override
+    public List<BucketEntry> readBucketEntries() {
+        List<ManifestFileMeta> manifests = readManifests().getRight();
+        Map<Pair<BinaryRow, Integer>, BucketEntry> buckets = new ConcurrentHashMap<>();
+        Consumer<ManifestFileMeta> processor =
+                m -> BucketEntry.merge(BucketEntry.merge(readManifestFileMeta(m)), buckets);
+        randomlyOnlyExecute(getExecutorService(scanManifestParallelism), processor, manifests);
+        return buckets.values().stream()
+                .filter(p -> p.fileCount() > 0)
+                .collect(Collectors.toList());
+    }
+
     private Pair<Snapshot, List<ManifestEntry>> doPlan() {
         long started = System.nanoTime();
         Pair<Snapshot, List<ManifestFileMeta>> snapshotListPair = readManifests();

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

@@ -21,6 +21,7 @@
 import org.apache.paimon.Snapshot;
 import org.apache.paimon.data.BinaryRow;
 import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.manifest.BucketEntry;
 import org.apache.paimon.manifest.FileKind;
 import org.apache.paimon.manifest.ManifestCacheFilter;
 import org.apache.paimon.manifest.ManifestEntry;
@@ -98,6 +99,8 @@ default Long totalRecordCount(Snapshot snapshot) {
 
     List<PartitionEntry> readPartitionEntries();
 
+    List<BucketEntry> readBucketEntries();
+
     default List<BinaryRow> listPartitions() {
         return readPartitionEntries().stream()
                 .map(PartitionEntry::partition)

paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java

@@ -21,6 +21,7 @@
 import org.apache.paimon.Snapshot;
 import org.apache.paimon.consumer.ConsumerManager;
 import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.manifest.BucketEntry;
 import org.apache.paimon.manifest.ManifestEntry;
 import org.apache.paimon.manifest.PartitionEntry;
 import org.apache.paimon.metrics.MetricRegistry;
@@ -88,6 +89,8 @@ public interface SnapshotReader {
 
     List<PartitionEntry> partitionEntries();
 
+    List<BucketEntry> bucketEntries();
+
     /** Result plan of this scan. */
     interface Plan extends TableScan.Plan {
 

paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java

@@ -27,6 +27,7 @@
 import org.apache.paimon.index.IndexFileHandler;
 import org.apache.paimon.index.IndexFileMeta;
 import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.manifest.BucketEntry;
 import org.apache.paimon.manifest.FileKind;
 import org.apache.paimon.manifest.ManifestEntry;
 import org.apache.paimon.manifest.PartitionEntry;
@@ -335,6 +336,11 @@ public List<PartitionEntry> partitionEntries() {
         return scan.readPartitionEntries();
     }
 
+    @Override
+    public List<BucketEntry> bucketEntries() {
+        return scan.readBucketEntries();
+    }
+
     @Override
     public Plan readChanges() {
         withMode(ScanMode.DELTA);

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

@@ -27,6 +27,7 @@
 import org.apache.paimon.disk.IOManager;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.Path;
+import org.apache.paimon.manifest.BucketEntry;
 import org.apache.paimon.manifest.IndexManifestEntry;
 import org.apache.paimon.manifest.ManifestEntry;
 import org.apache.paimon.manifest.ManifestFileMeta;
@@ -359,6 +360,11 @@ public List<BinaryRow> partitions() {
         public List<PartitionEntry> partitionEntries() {
             return snapshotReader.partitionEntries();
         }
+
+        @Override
+        public List<BucketEntry> bucketEntries() {
+            return snapshotReader.bucketEntries();
+        }
     }
 
     private class AuditLogBatchScan implements DataTableScan {