This is an automated email from the ASF dual-hosted git repository.
junhao pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 02c54f591 [core] Introduce bucket entries to optimize Spark compact
(#4162)
02c54f591 is described below
commit 02c54f59128d0861f3d7e0b0af43ce70abeb88d4
Author: Jingsong Lee <[email protected]>
AuthorDate: Wed Sep 11 14:15:48 2024 +0800
[core] Introduce bucket entries to optimize Spark compact (#4162)
---
docs/content/maintenance/system-tables.md | 16 ++
.../org/apache/paimon/manifest/BucketEntry.java | 143 +++++++++++
.../paimon/operation/AbstractFileStoreScan.java | 18 +-
.../org/apache/paimon/operation/FileStoreScan.java | 3 +
.../table/source/snapshot/SnapshotReader.java | 3 +
.../table/source/snapshot/SnapshotReaderImpl.java | 6 +
.../apache/paimon/table/system/AuditLogTable.java | 6 +
.../apache/paimon/table/system/BucketsTable.java | 277 +++++++++------------
...{BucketsTable.java => CompactBucketsTable.java} | 16 +-
.../paimon/table/system/PartitionsTable.java | 1 -
.../paimon/table/system/SystemTableLoader.java | 2 +
.../paimon/table/system/BucketsTableTest.java | 80 ++++++
.../flink/compact/MultiAwareBucketTableScan.java | 9 +-
.../flink/sink/BucketsRowChannelComputer.java | 4 +-
.../CombinedTableCompactorSourceBuilder.java | 4 +-
.../flink/source/CompactorSourceBuilder.java | 27 +-
.../source/operator/MultiTablesReadOperator.java | 10 +-
.../paimon/spark/procedure/CompactProcedure.java | 5 +-
18 files changed, 424 insertions(+), 206 deletions(-)
diff --git a/docs/content/maintenance/system-tables.md
b/docs/content/maintenance/system-tables.md
index cc542697a..f19c75756 100644
--- a/docs/content/maintenance/system-tables.md
+++ b/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`.
diff --git
a/paimon-core/src/main/java/org/apache/paimon/manifest/BucketEntry.java
b/paimon-core/src/main/java/org/apache/paimon/manifest/BucketEntry.java
new file mode 100644
index 000000000..3e93bc8cc
--- /dev/null
+++ b/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);
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java
b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java
index f1a95aec9..824f90797 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java
@@ -22,6 +22,7 @@ import org.apache.paimon.CoreOptions;
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.List;
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 abstract class AbstractFileStoreScan implements
FileStoreScan {
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();
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreScan.java
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreScan.java
index 427e958e8..b1e0a2bd9 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreScan.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreScan.java
@@ -21,6 +21,7 @@ package org.apache.paimon.operation;
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 @@ public interface FileStoreScan {
List<PartitionEntry> readPartitionEntries();
+ List<BucketEntry> readBucketEntries();
+
default List<BinaryRow> listPartitions() {
return readPartitionEntries().stream()
.map(PartitionEntry::partition)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java
index 96605ceeb..13a458d4a 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java
@@ -21,6 +21,7 @@ package org.apache.paimon.table.source.snapshot;
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 {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java
index 47e697130..03b6144be 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java
@@ -27,6 +27,7 @@ import org.apache.paimon.data.BinaryRow;
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 class SnapshotReaderImpl implements SnapshotReader {
return scan.readPartitionEntries();
}
+ @Override
+ public List<BucketEntry> bucketEntries() {
+ return scan.readBucketEntries();
+ }
+
@Override
public Plan readChanges() {
withMode(ScanMode.DELTA);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
index 50028d890..38457f29b 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
@@ -27,6 +27,7 @@ import org.apache.paimon.data.InternalRow;
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 class AuditLogTable implements DataTable,
ReadonlyTable {
public List<PartitionEntry> partitionEntries() {
return snapshotReader.partitionEntries();
}
+
+ @Override
+ public List<BucketEntry> bucketEntries() {
+ return snapshotReader.bucketEntries();
+ }
}
private class AuditLogBatchScan implements DataTableScan {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/system/BucketsTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/system/BucketsTable.java
index ddf38b7b2..e4d0bab66 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/BucketsTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/system/BucketsTable.java
@@ -18,219 +18,150 @@
package org.apache.paimon.table.system;
-import org.apache.paimon.CoreOptions;
-import org.apache.paimon.Snapshot;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.Timestamp;
import org.apache.paimon.disk.IOManager;
-import org.apache.paimon.fs.FileIO;
-import org.apache.paimon.fs.Path;
-import org.apache.paimon.io.DataFileMeta;
-import org.apache.paimon.io.DataFileMetaSerializer;
-import org.apache.paimon.manifest.IndexManifestEntry;
-import org.apache.paimon.manifest.ManifestEntry;
-import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.manifest.BucketEntry;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.reader.RecordReader;
-import org.apache.paimon.table.DataTable;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.ReadonlyTable;
-import org.apache.paimon.table.source.DataSplit;
-import org.apache.paimon.table.source.DataTableScan;
+import org.apache.paimon.table.Table;
import org.apache.paimon.table.source.InnerTableRead;
+import org.apache.paimon.table.source.InnerTableScan;
+import org.apache.paimon.table.source.ReadOnceTableScan;
+import org.apache.paimon.table.source.SingletonSplit;
import org.apache.paimon.table.source.Split;
-import org.apache.paimon.table.source.StreamDataTableScan;
import org.apache.paimon.table.source.TableRead;
-import org.apache.paimon.table.source.snapshot.SnapshotReader;
import org.apache.paimon.types.BigIntType;
-import org.apache.paimon.types.DataType;
-import org.apache.paimon.types.IntType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
-import org.apache.paimon.types.VarCharType;
-import org.apache.paimon.utils.BranchManager;
import org.apache.paimon.utils.IteratorRecordReader;
-import org.apache.paimon.utils.SimpleFileReader;
-import org.apache.paimon.utils.SnapshotManager;
-import org.apache.paimon.utils.TagManager;
+import org.apache.paimon.utils.ProjectedRow;
+import org.apache.paimon.utils.RowDataToObjectArrayConverter;
+import org.apache.paimon.utils.SerializationUtils;
-import javax.annotation.Nullable;
+import org.apache.paimon.shade.guava30.com.google.common.collect.Iterators;
-import java.io.IOException;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
-import java.util.OptionalLong;
-import static org.apache.paimon.utils.SerializationUtils.newBytesType;
-import static org.apache.paimon.utils.SerializationUtils.serializeBinaryRow;
+import static org.apache.paimon.catalog.Catalog.SYSTEM_TABLE_SPLITTER;
-/**
- * A table to produce modified partitions and buckets for each snapshot.
- *
- * <p>Only used internally by dedicated compact job sources.
- */
-public class BucketsTable implements DataTable, ReadonlyTable {
+/** A {@link Table} for showing buckets info. */
+public class BucketsTable implements ReadonlyTable {
private static final long serialVersionUID = 1L;
- private final FileStoreTable wrapped;
- private final boolean isContinuous;
-
- @Nullable private final String databaseName;
-
- private static final RowType ROW_TYPE =
- RowType.of(
- new DataType[] {
- new BigIntType(false),
- newBytesType(false),
- new IntType(false),
- newBytesType(false),
- new VarCharType(true, Integer.MAX_VALUE),
- new VarCharType(false, Integer.MAX_VALUE)
- },
- new String[] {
- "_SNAPSHOT_ID",
- "_PARTITION",
- "_BUCKET",
- "_FILES",
- "_DATABASE_NAME",
- "_TABLE_NAME"
- });
-
- public BucketsTable(FileStoreTable wrapped, boolean isContinuous) {
- this(wrapped, isContinuous, null);
- }
-
- // if need to specify the database of a table, use this method
- public BucketsTable(FileStoreTable wrapped, boolean isContinuous, String
databaseName) {
- this.wrapped = wrapped;
- this.isContinuous = isContinuous;
- this.databaseName = databaseName;
- }
-
- @Override
- public OptionalLong latestSnapshotId() {
- return wrapped.latestSnapshotId();
- }
-
- @Override
- public Snapshot snapshot(long snapshotId) {
- return wrapped.snapshot(snapshotId);
- }
+ public static final String BUCKETS = "buckets";
- @Override
- public SimpleFileReader<ManifestFileMeta> manifestListReader() {
- return wrapped.manifestListReader();
- }
+ public static final RowType TABLE_TYPE =
+ new RowType(
+ Arrays.asList(
+ new DataField(0, "partition",
SerializationUtils.newStringType(true)),
+ new DataField(1, "bucket",
DataTypes.INT().notNull()),
+ new DataField(2, "record_count", new
BigIntType(false)),
+ new DataField(3, "file_size_in_bytes", new
BigIntType(false)),
+ new DataField(4, "file_count", new
BigIntType(false)),
+ new DataField(5, "last_update_time",
DataTypes.TIMESTAMP_MILLIS())));
- @Override
- public SimpleFileReader<ManifestEntry> manifestFileReader() {
- return wrapped.manifestFileReader();
- }
+ private final FileStoreTable storeTable;
- @Override
- public SimpleFileReader<IndexManifestEntry> indexManifestFileReader() {
- return wrapped.indexManifestFileReader();
+ public BucketsTable(FileStoreTable storeTable) {
+ this.storeTable = storeTable;
}
@Override
- public Path location() {
- return wrapped.location();
- }
-
- @Override
- public SnapshotManager snapshotManager() {
- return wrapped.snapshotManager();
- }
-
- @Override
- public TagManager tagManager() {
- return wrapped.tagManager();
+ public String name() {
+ return storeTable.name() + SYSTEM_TABLE_SPLITTER + BUCKETS;
}
@Override
- public BranchManager branchManager() {
- return wrapped.branchManager();
+ public RowType rowType() {
+ return TABLE_TYPE;
}
@Override
- public DataTable switchToBranch(String branchName) {
- return new BucketsTable(wrapped.switchToBranch(branchName),
isContinuous, databaseName);
+ public List<String> primaryKeys() {
+ return Arrays.asList("partition", "bucket");
}
@Override
- public String name() {
- return "__internal_buckets_" + wrapped.location().getName();
+ public InnerTableScan newScan() {
+ return new BucketsScan();
}
@Override
- public RowType rowType() {
- return ROW_TYPE;
+ public InnerTableRead newRead() {
+ return new BucketsRead(storeTable);
}
@Override
- public Map<String, String> options() {
- return wrapped.options();
+ public Table copy(Map<String, String> dynamicOptions) {
+ return new BucketsTable(storeTable.copy(dynamicOptions));
}
- @Override
- public List<String> primaryKeys() {
- return Collections.emptyList();
- }
+ private static class BucketsScan extends ReadOnceTableScan {
- @Override
- public SnapshotReader newSnapshotReader() {
- return wrapped.newSnapshotReader();
- }
+ @Override
+ public InnerTableScan withFilter(Predicate predicate) {
+ return this;
+ }
- @Override
- public DataTableScan newScan() {
- return wrapped.newScan();
+ @Override
+ public Plan innerPlan() {
+ return () -> Collections.singletonList(new BucketsSplit());
+ }
}
- @Override
- public StreamDataTableScan newStreamScan() {
- return wrapped.newStreamScan();
- }
+ private static class BucketsSplit extends SingletonSplit {
- @Override
- public CoreOptions coreOptions() {
- return wrapped.coreOptions();
- }
+ private static final long serialVersionUID = 1L;
- @Override
- public InnerTableRead newRead() {
- return new BucketsRead();
- }
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ return o != null && getClass() == o.getClass();
+ }
- @Override
- public BucketsTable copy(Map<String, String> dynamicOptions) {
- return new BucketsTable(wrapped.copy(dynamicOptions), isContinuous,
databaseName);
+ @Override
+ public int hashCode() {
+ return 1;
+ }
}
- @Override
- public FileIO fileIO() {
- return wrapped.fileIO();
- }
+ private static class BucketsRead implements InnerTableRead {
- public static RowType getRowType() {
- return ROW_TYPE;
- }
+ private final FileStoreTable fileStoreTable;
- private class BucketsRead implements InnerTableRead {
+ private int[][] projection;
- private final DataFileMetaSerializer dataFileMetaSerializer = new
DataFileMetaSerializer();
+ public BucketsRead(FileStoreTable table) {
+ this.fileStoreTable = table;
+ }
@Override
public InnerTableRead withFilter(Predicate predicate) {
- // filter is done by scan
+ // TODO
return this;
}
@Override
public InnerTableRead withProjection(int[][] projection) {
- throw new UnsupportedOperationException("BucketsRead does not
support projection");
+ this.projection = projection;
+ return this;
}
@Override
@@ -239,30 +170,46 @@ public class BucketsTable implements DataTable,
ReadonlyTable {
}
@Override
- public RecordReader<InternalRow> createReader(Split split) throws
IOException {
- if (!(split instanceof DataSplit)) {
+ public RecordReader<InternalRow> createReader(Split split) {
+ if (!(split instanceof BucketsSplit)) {
throw new IllegalArgumentException("Unsupported split: " +
split.getClass());
}
- DataSplit dataSplit = (DataSplit) split;
+ List<BucketEntry> buckets =
fileStoreTable.newSnapshotReader().bucketEntries();
- List<DataFileMeta> files = Collections.emptyList();
- if (isContinuous) {
- // Serialized files are only useful in streaming jobs.
- // Batch compact jobs only run once, so they only need to know
what buckets should
- // be compacted and don't need to concern incremental new
files.
- files = dataSplit.dataFiles();
+ RowDataToObjectArrayConverter converter =
+ new RowDataToObjectArrayConverter(
+ fileStoreTable.schema().logicalPartitionType());
+
+ List<InternalRow> results = new ArrayList<>(buckets.size());
+ for (BucketEntry entry : buckets) {
+ results.add(toRow(entry, converter));
}
- InternalRow row =
- GenericRow.of(
- dataSplit.snapshotId(),
- serializeBinaryRow(dataSplit.partition()),
- dataSplit.bucket(),
- dataFileMetaSerializer.serializeList(files),
- BinaryString.fromString(databaseName),
- BinaryString.fromString(wrapped.name()));
-
- return new
IteratorRecordReader<>(Collections.singletonList(row).iterator());
+
+ Iterator<InternalRow> iterator = results.iterator();
+ if (projection != null) {
+ iterator =
+ Iterators.transform(
+ iterator, row ->
ProjectedRow.from(projection).replaceRow(row));
+ }
+ return new IteratorRecordReader<>(iterator);
+ }
+
+ private GenericRow toRow(
+ BucketEntry entry, RowDataToObjectArrayConverter
partitionConverter) {
+ BinaryString partitionId =
+ BinaryString.fromString(
+
Arrays.toString(partitionConverter.convert(entry.partition())));
+ return GenericRow.of(
+ partitionId,
+ entry.bucket(),
+ entry.recordCount(),
+ entry.fileSizeInBytes(),
+ entry.fileCount(),
+ Timestamp.fromLocalDateTime(
+ LocalDateTime.ofInstant(
+
Instant.ofEpochMilli(entry.lastFileCreationTime()),
+ ZoneId.systemDefault())));
}
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/system/BucketsTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/system/CompactBucketsTable.java
similarity index 92%
copy from
paimon-core/src/main/java/org/apache/paimon/table/system/BucketsTable.java
copy to
paimon-core/src/main/java/org/apache/paimon/table/system/CompactBucketsTable.java
index ddf38b7b2..8f6476c7f 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/BucketsTable.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/system/CompactBucketsTable.java
@@ -66,11 +66,12 @@ import static
org.apache.paimon.utils.SerializationUtils.newBytesType;
import static org.apache.paimon.utils.SerializationUtils.serializeBinaryRow;
/**
- * A table to produce modified partitions and buckets for each snapshot.
+ * A table to produce modified partitions and buckets (also including files in
streaming mode) for
+ * each snapshot.
*
* <p>Only used internally by dedicated compact job sources.
*/
-public class BucketsTable implements DataTable, ReadonlyTable {
+public class CompactBucketsTable implements DataTable, ReadonlyTable {
private static final long serialVersionUID = 1L;
@@ -98,12 +99,12 @@ public class BucketsTable implements DataTable,
ReadonlyTable {
"_TABLE_NAME"
});
- public BucketsTable(FileStoreTable wrapped, boolean isContinuous) {
+ public CompactBucketsTable(FileStoreTable wrapped, boolean isContinuous) {
this(wrapped, isContinuous, null);
}
// if need to specify the database of a table, use this method
- public BucketsTable(FileStoreTable wrapped, boolean isContinuous, String
databaseName) {
+ public CompactBucketsTable(FileStoreTable wrapped, boolean isContinuous,
String databaseName) {
this.wrapped = wrapped;
this.isContinuous = isContinuous;
this.databaseName = databaseName;
@@ -156,7 +157,8 @@ public class BucketsTable implements DataTable,
ReadonlyTable {
@Override
public DataTable switchToBranch(String branchName) {
- return new BucketsTable(wrapped.switchToBranch(branchName),
isContinuous, databaseName);
+ return new CompactBucketsTable(
+ wrapped.switchToBranch(branchName), isContinuous,
databaseName);
}
@Override
@@ -205,8 +207,8 @@ public class BucketsTable implements DataTable,
ReadonlyTable {
}
@Override
- public BucketsTable copy(Map<String, String> dynamicOptions) {
- return new BucketsTable(wrapped.copy(dynamicOptions), isContinuous,
databaseName);
+ public CompactBucketsTable copy(Map<String, String> dynamicOptions) {
+ return new CompactBucketsTable(wrapped.copy(dynamicOptions),
isContinuous, databaseName);
}
@Override
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/system/PartitionsTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/system/PartitionsTable.java
index 76c0768ee..09cf25167 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/system/PartitionsTable.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/system/PartitionsTable.java
@@ -115,7 +115,6 @@ public class PartitionsTable implements ReadonlyTable {
@Override
public InnerTableScan withFilter(Predicate predicate) {
- // TODO
return this;
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/system/SystemTableLoader.java
b/paimon-core/src/main/java/org/apache/paimon/table/system/SystemTableLoader.java
index 10e0c587b..a84f41ec1 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/system/SystemTableLoader.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/system/SystemTableLoader.java
@@ -42,6 +42,7 @@ import static
org.apache.paimon.table.system.AggregationFieldsTable.AGGREGATION_
import static
org.apache.paimon.table.system.AllTableOptionsTable.ALL_TABLE_OPTIONS;
import static org.apache.paimon.table.system.AuditLogTable.AUDIT_LOG;
import static org.apache.paimon.table.system.BranchesTable.BRANCHES;
+import static org.apache.paimon.table.system.BucketsTable.BUCKETS;
import static
org.apache.paimon.table.system.CatalogOptionsTable.CATALOG_OPTIONS;
import static org.apache.paimon.table.system.ConsumersTable.CONSUMERS;
import static org.apache.paimon.table.system.FilesTable.FILES;
@@ -67,6 +68,7 @@ public class SystemTableLoader {
.put(OPTIONS, OptionsTable::new)
.put(SCHEMAS, SchemasTable::new)
.put(PARTITIONS, PartitionsTable::new)
+ .put(BUCKETS, BucketsTable::new)
.put(AUDIT_LOG, AuditLogTable::new)
.put(FILES, FilesTable::new)
.put(TAGS, TagsTable::new)
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/system/BucketsTableTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/system/BucketsTableTest.java
new file mode 100644
index 000000000..57cb6605a
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/table/system/BucketsTableTest.java
@@ -0,0 +1,80 @@
+/*
+ * 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.table.system;
+
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.schema.SchemaUtils;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.TableTestBase;
+import org.apache.paimon.types.DataTypes;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.apache.paimon.catalog.Catalog.SYSTEM_TABLE_SPLITTER;
+import static org.apache.paimon.table.system.BucketsTable.BUCKETS;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Unit tests for {@link BucketsTable}. */
+public class BucketsTableTest extends TableTestBase {
+
+ private Table bucketsTable;
+
+ @BeforeEach
+ public void before() throws Exception {
+ String tableName = "MyTable";
+ FileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(String.format("%s/%s.db/%s", warehouse,
database, tableName));
+ Schema schema =
+ Schema.newBuilder()
+ .column("pk", DataTypes.INT())
+ .column("pt", DataTypes.INT())
+ .column("col1", DataTypes.INT())
+ .partitionKeys("pt")
+ .primaryKey("pk", "pt")
+ .option("bucket", "2")
+ .build();
+ TableSchema tableSchema =
+ SchemaUtils.forceCommit(new SchemaManager(fileIO, tablePath),
schema);
+ FileStoreTable table =
+ FileStoreTableFactory.create(LocalFileIO.create(), tablePath,
tableSchema);
+
+ bucketsTable = catalog.getTable(identifier(tableName +
SYSTEM_TABLE_SPLITTER + BUCKETS));
+
+ write(table, GenericRow.of(1, 1, 1), GenericRow.of(1, 2, 5));
+ write(table, GenericRow.of(1, 1, 3), GenericRow.of(1, 2, 4));
+ }
+
+ @Test
+ public void testBucketsTable() throws Exception {
+ assertThat(read(bucketsTable, new int[][] {{0}, {1}, {2}, {4}}))
+ .containsExactlyInAnyOrder(
+ GenericRow.of(BinaryString.fromString("[1]"), 0, 2L,
2L),
+ GenericRow.of(BinaryString.fromString("[2]"), 0, 2L,
2L));
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/MultiAwareBucketTableScan.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/MultiAwareBucketTableScan.java
index 68d66ec5f..747995d20 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/MultiAwareBucketTableScan.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/MultiAwareBucketTableScan.java
@@ -24,7 +24,7 @@ import org.apache.paimon.table.BucketMode;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.table.source.StreamTableScan;
-import org.apache.paimon.table.system.BucketsTable;
+import org.apache.paimon.table.system.CompactBucketsTable;
import org.apache.flink.api.java.tuple.Tuple2;
@@ -44,7 +44,7 @@ import static
org.apache.paimon.flink.utils.MultiTablesCompactorUtil.compactOpti
*/
public class MultiAwareBucketTableScan extends
MultiTableScanBase<Tuple2<Split, String>> {
- protected transient Map<Identifier, BucketsTable> tablesMap;
+ protected transient Map<Identifier, CompactBucketsTable> tablesMap;
protected transient Map<Identifier, StreamTableScan> scansMap;
public MultiAwareBucketTableScan(
@@ -87,8 +87,9 @@ public class MultiAwareBucketTableScan extends
MultiTableScanBase<Tuple2<Split,
@Override
public void addScanTable(FileStoreTable fileStoreTable, Identifier
identifier) {
if (fileStoreTable.bucketMode() != BucketMode.BUCKET_UNAWARE) {
- BucketsTable bucketsTable =
- new BucketsTable(fileStoreTable, isStreaming,
identifier.getDatabaseName())
+ CompactBucketsTable bucketsTable =
+ new CompactBucketsTable(
+ fileStoreTable, isStreaming,
identifier.getDatabaseName())
.copy(compactOptions(isStreaming));
tablesMap.put(identifier, bucketsTable);
scansMap.put(identifier,
bucketsTable.newReadBuilder().newStreamScan());
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/BucketsRowChannelComputer.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/BucketsRowChannelComputer.java
index d93d38e00..61c9dd018 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/BucketsRowChannelComputer.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/BucketsRowChannelComputer.java
@@ -20,13 +20,13 @@ package org.apache.paimon.flink.sink;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.table.sink.ChannelComputer;
-import org.apache.paimon.table.system.BucketsTable;
+import org.apache.paimon.table.system.CompactBucketsTable;
import org.apache.flink.table.data.RowData;
import static org.apache.paimon.utils.SerializationUtils.deserializeBinaryRow;
-/** {@link ChannelComputer} to partition {@link RowData} from {@link
BucketsTable}. */
+/** {@link ChannelComputer} to partition {@link RowData} from {@link
CompactBucketsTable}. */
public class BucketsRowChannelComputer implements ChannelComputer<RowData> {
private transient int numberOfChannels;
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/CombinedTableCompactorSourceBuilder.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/CombinedTableCompactorSourceBuilder.java
index 384cdcdce..e5cbbe845 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/CombinedTableCompactorSourceBuilder.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/CombinedTableCompactorSourceBuilder.java
@@ -25,7 +25,7 @@ import
org.apache.paimon.flink.source.operator.CombinedAwareBatchSourceFunction;
import
org.apache.paimon.flink.source.operator.CombinedAwareStreamingSourceFunction;
import
org.apache.paimon.flink.source.operator.CombinedUnawareBatchSourceFunction;
import
org.apache.paimon.flink.source.operator.CombinedUnawareStreamingSourceFunction;
-import org.apache.paimon.table.system.BucketsTable;
+import org.apache.paimon.table.system.CompactBucketsTable;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.Preconditions;
@@ -85,7 +85,7 @@ public class CombinedTableCompactorSourceBuilder {
public DataStream<RowData> buildAwareBucketTableSource() {
Preconditions.checkArgument(env != null, "StreamExecutionEnvironment
should not be null.");
- RowType produceType = BucketsTable.getRowType();
+ RowType produceType = CompactBucketsTable.getRowType();
if (isContinuous) {
return CombinedAwareStreamingSourceFunction.buildSource(
env,
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/CompactorSourceBuilder.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/CompactorSourceBuilder.java
index 10fc3b558..141f45ecd 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/CompactorSourceBuilder.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/CompactorSourceBuilder.java
@@ -27,7 +27,7 @@ import org.apache.paimon.options.Options;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.source.ReadBuilder;
-import org.apache.paimon.table.system.BucketsTable;
+import org.apache.paimon.table.system.CompactBucketsTable;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.Preconditions;
@@ -85,18 +85,19 @@ public class CompactorSourceBuilder {
return this;
}
- private Source<RowData, ?, ?> buildSource(BucketsTable bucketsTable) {
+ private Source<RowData, ?, ?> buildSource(CompactBucketsTable
compactBucketsTable) {
if (isContinuous) {
- bucketsTable = bucketsTable.copy(streamingCompactOptions());
+ compactBucketsTable =
compactBucketsTable.copy(streamingCompactOptions());
return new ContinuousFileStoreSource(
-
bucketsTable.newReadBuilder().withFilter(partitionPredicate),
- bucketsTable.options(),
+
compactBucketsTable.newReadBuilder().withFilter(partitionPredicate),
+ compactBucketsTable.options(),
null);
} else {
- bucketsTable = bucketsTable.copy(batchCompactOptions());
- ReadBuilder readBuilder =
bucketsTable.newReadBuilder().withFilter(partitionPredicate);
- Options options = bucketsTable.coreOptions().toConfiguration();
+ compactBucketsTable =
compactBucketsTable.copy(batchCompactOptions());
+ ReadBuilder readBuilder =
+
compactBucketsTable.newReadBuilder().withFilter(partitionPredicate);
+ Options options =
compactBucketsTable.coreOptions().toConfiguration();
return new StaticFileStoreSource(
readBuilder,
null,
@@ -110,11 +111,11 @@ public class CompactorSourceBuilder {
throw new IllegalArgumentException("StreamExecutionEnvironment
should not be null.");
}
- BucketsTable bucketsTable = new BucketsTable(table, isContinuous);
- RowType produceType = bucketsTable.rowType();
+ CompactBucketsTable compactBucketsTable = new
CompactBucketsTable(table, isContinuous);
+ RowType produceType = compactBucketsTable.rowType();
DataStreamSource<RowData> dataStream =
env.fromSource(
- buildSource(bucketsTable),
+ buildSource(compactBucketsTable),
WatermarkStrategy.noWatermarks(),
tableIdentifier + "-compact-source",
InternalTypeInfo.of(LogicalTypeConversion.toLogicalType(produceType)));
@@ -122,7 +123,7 @@ public class CompactorSourceBuilder {
Preconditions.checkArgument(
partitionIdleTime == null, "Streaming mode does not
support partitionIdleTime");
} else if (partitionIdleTime != null) {
- Map<BinaryRow, Long> partitionInfo =
getPartitionInfo(bucketsTable);
+ Map<BinaryRow, Long> partitionInfo =
getPartitionInfo(compactBucketsTable);
long historyMilli =
LocalDateTime.now()
.minus(partitionIdleTime)
@@ -175,7 +176,7 @@ public class CompactorSourceBuilder {
return this;
}
- private Map<BinaryRow, Long> getPartitionInfo(BucketsTable table) {
+ private Map<BinaryRow, Long> getPartitionInfo(CompactBucketsTable table) {
List<PartitionEntry> partitions =
table.newSnapshotReader().partitionEntries();
return partitions.stream()
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MultiTablesReadOperator.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MultiTablesReadOperator.java
index 0b48a9bcc..73d46ae1e 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MultiTablesReadOperator.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MultiTablesReadOperator.java
@@ -29,7 +29,7 @@ import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.table.source.TableRead;
-import org.apache.paimon.table.system.BucketsTable;
+import org.apache.paimon.table.system.CompactBucketsTable;
import org.apache.paimon.utils.CloseableIterator;
import org.apache.paimon.utils.Preconditions;
@@ -80,7 +80,7 @@ public class MultiTablesReadOperator extends
AbstractStreamOperator<RowData>
private transient Catalog catalog;
private transient IOManager ioManager;
- private transient Map<Identifier, BucketsTable> tablesMap;
+ private transient Map<Identifier, CompactBucketsTable> tablesMap;
private transient Map<Identifier, TableRead> readsMap;
private transient StreamRecord<RowData> reuseRecord;
private transient FlinkRowData reuseRow;
@@ -132,7 +132,7 @@ public class MultiTablesReadOperator extends
AbstractStreamOperator<RowData>
}
private TableRead getTableRead(Identifier tableId) {
- BucketsTable table = tablesMap.get(tableId);
+ CompactBucketsTable table = tablesMap.get(tableId);
if (table == null) {
try {
Table newTable = catalog.getTable(tableId);
@@ -141,7 +141,7 @@ public class MultiTablesReadOperator extends
AbstractStreamOperator<RowData>
"Only FileStoreTable supports compact action. The
table type is '%s'.",
newTable.getClass().getName());
table =
- new BucketsTable(
+ new CompactBucketsTable(
(FileStoreTable) newTable,
isStreaming,
tableId.getDatabaseName())
@@ -156,7 +156,7 @@ public class MultiTablesReadOperator extends
AbstractStreamOperator<RowData>
return readsMap.get(tableId);
}
- private Map<BinaryRow, Long> getPartitionInfo(BucketsTable table) {
+ private Map<BinaryRow, Long> getPartitionInfo(CompactBucketsTable table) {
List<PartitionEntry> partitions =
table.newSnapshotReader().partitionEntries();
return partitions.stream()
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
index 89f84f5f7..cc78d1d68 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
@@ -264,9 +264,8 @@ public class CompactProcedure extends BaseProcedure {
Set<BinaryRow> partitionToBeCompacted =
getHistoryPartition(snapshotReader, partitionIdleTime);
List<Pair<byte[], Integer>> partitionBuckets =
- snapshotReader.read().splits().stream()
- .map(split -> (DataSplit) split)
- .map(dataSplit -> Pair.of(dataSplit.partition(),
dataSplit.bucket()))
+ snapshotReader.bucketEntries().stream()
+ .map(entry -> Pair.of(entry.partition(),
entry.bucket()))
.distinct()
.filter(pair ->
partitionToBeCompacted.contains(pair.getKey()))
.map(
