Skip to content

Commit 69a434c

Browse files
platinumhamburgplatinumhamburg
committed
[kv] Add asynchronous manual compaction for primary-key tables
Support scoped manual compaction with durable at-least-once scheduling, confirmed snapshot publication, and asynchronous TabletServer execution. Use idempotent bucket starts and bounded immutable progress metadata.
1 parent 5af9dda commit 69a434c

85 files changed

Lines changed: 12144 additions & 488 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎fluss-client/src/main/java/org/apache/fluss/client/admin/Admin.java‎

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,9 @@
2424
import org.apache.fluss.client.metadata.LakeSnapshot;
2525
import org.apache.fluss.client.metadata.RemoteLogManifestInfo;
2626
import org.apache.fluss.cluster.ServerNode;
27+
import org.apache.fluss.cluster.compaction.CompactionInfo;
28+
import org.apache.fluss.cluster.compaction.CompactionProgress;
29+
import org.apache.fluss.cluster.compaction.CompactionType;
2730
import org.apache.fluss.cluster.rebalance.GoalType;
2831
import org.apache.fluss.cluster.rebalance.RebalanceProgress;
2932
import org.apache.fluss.cluster.rebalance.ServerTag;
@@ -917,4 +920,36 @@ CompletableFuture<List<RemoteLogManifestInfo>> listRemoteLogManifests(
917920
* @return per-bucket active snapshot ids grouped by bucket
918921
*/
919922
CompletableFuture<ActiveKvSnapshots> listKvSnapshots(long tableId, @Nullable Long partitionId);
923+
924+
/** Submits cluster compaction. */
925+
CompletableFuture<String> compact(CompactionType type);
926+
927+
/** Submits database compaction. */
928+
CompletableFuture<String> compact(String databaseName, CompactionType type);
929+
930+
/** Submits table compaction. */
931+
CompletableFuture<String> compact(TablePath tablePath, CompactionType type);
932+
933+
/** Submits partition compaction. */
934+
CompletableFuture<String> compact(
935+
TablePath tablePath, String partitionName, CompactionType type);
936+
937+
/** Lists cluster compactions. */
938+
CompletableFuture<List<CompactionInfo>> listCompactions();
939+
940+
/** Lists compactions associated with a database. */
941+
CompletableFuture<List<CompactionInfo>> listCompactions(String databaseName);
942+
943+
/** Lists compactions associated with a table. */
944+
CompletableFuture<List<CompactionInfo>> listCompactions(TablePath tablePath);
945+
946+
/** Lists compactions associated with a partition. */
947+
CompletableFuture<List<CompactionInfo>> listCompactions(
948+
TablePath tablePath, String partitionName);
949+
950+
/** Returns progress of an operation. */
951+
CompletableFuture<CompactionProgress> getCompaction(String operationId);
952+
953+
/** Cancels an operation; repeated cancellation is allowed. */
954+
CompletableFuture<Void> cancelCompaction(String operationId);
920955
}

‎fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java‎

Lines changed: 106 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,9 @@
2828
import org.apache.fluss.client.utils.ClientUtils;
2929
import org.apache.fluss.cluster.Cluster;
3030
import org.apache.fluss.cluster.ServerNode;
31+
import org.apache.fluss.cluster.compaction.CompactionInfo;
32+
import org.apache.fluss.cluster.compaction.CompactionProgress;
33+
import org.apache.fluss.cluster.compaction.CompactionType;
3134
import org.apache.fluss.cluster.rebalance.GoalType;
3235
import org.apache.fluss.cluster.rebalance.RebalanceProgress;
3336
import org.apache.fluss.cluster.rebalance.ServerTag;
@@ -64,7 +67,10 @@
6467
import org.apache.fluss.rpc.messages.AlterClusterConfigsRequest;
6568
import org.apache.fluss.rpc.messages.AlterDatabaseRequest;
6669
import org.apache.fluss.rpc.messages.AlterTableRequest;
70+
import org.apache.fluss.rpc.messages.CancelCompactionRequest;
6771
import org.apache.fluss.rpc.messages.CancelRebalanceRequest;
72+
import org.apache.fluss.rpc.messages.CompactRequest;
73+
import org.apache.fluss.rpc.messages.CompactResponse;
6874
import org.apache.fluss.rpc.messages.CreateAclsRequest;
6975
import org.apache.fluss.rpc.messages.CreateDatabaseRequest;
7076
import org.apache.fluss.rpc.messages.CreateTableRequest;
@@ -77,6 +83,7 @@
7783
import org.apache.fluss.rpc.messages.DropDatabaseRequest;
7884
import org.apache.fluss.rpc.messages.DropTableRequest;
7985
import org.apache.fluss.rpc.messages.GetClusterHealthRequest;
86+
import org.apache.fluss.rpc.messages.GetCompactionRequest;
8087
import org.apache.fluss.rpc.messages.GetDatabaseInfoRequest;
8188
import org.apache.fluss.rpc.messages.GetKvSnapshotMetadataRequest;
8289
import org.apache.fluss.rpc.messages.GetLakeSnapshotRequest;
@@ -87,6 +94,7 @@
8794
import org.apache.fluss.rpc.messages.GetTableStatsRequest;
8895
import org.apache.fluss.rpc.messages.GetTableStatsResponse;
8996
import org.apache.fluss.rpc.messages.ListAclsRequest;
97+
import org.apache.fluss.rpc.messages.ListCompactionsRequest;
9098
import org.apache.fluss.rpc.messages.ListDatabasesRequest;
9199
import org.apache.fluss.rpc.messages.ListDatabasesResponse;
92100
import org.apache.fluss.rpc.messages.ListKvSnapshotsRequest;
@@ -110,6 +118,7 @@
110118
import org.apache.fluss.rpc.messages.TableExistsRequest;
111119
import org.apache.fluss.rpc.messages.TableExistsResponse;
112120
import org.apache.fluss.rpc.protocol.ApiError;
121+
import org.apache.fluss.rpc.util.CommonRpcMessageUtils;
113122
import org.apache.fluss.security.acl.AclBinding;
114123
import org.apache.fluss.security.acl.AclBindingFilter;
115124
import org.apache.fluss.utils.ExceptionUtils;
@@ -129,6 +138,7 @@
129138
import java.util.concurrent.ConcurrentHashMap;
130139
import java.util.concurrent.ExecutorService;
131140
import java.util.concurrent.Executors;
141+
import java.util.stream.Collectors;
132142

133143
import static org.apache.fluss.client.utils.ClientRpcMessageUtils.makeAlterDatabaseRequest;
134144
import static org.apache.fluss.client.utils.ClientRpcMessageUtils.makeAlterTableRequest;
@@ -1142,4 +1152,100 @@ private static void validateRackRequest(List<String> racks, ServerTag serverTag)
11421152
"rack element must not be null, empty, or whitespace-only");
11431153
}
11441154
}
1155+
1156+
@Override
1157+
public CompletableFuture<String> compact(CompactionType type) {
1158+
CompactRequest request = new CompactRequest();
1159+
request.setType(checkNotNull(type, "type").name());
1160+
return gateway.compact(request).thenApply(CompactResponse::getOperationId);
1161+
}
1162+
1163+
@Override
1164+
public CompletableFuture<String> compact(String databaseName, CompactionType type) {
1165+
CompactRequest request = new CompactRequest();
1166+
request.setType(checkNotNull(type, "type").name());
1167+
request.setDatabaseName(checkNotNull(databaseName, "databaseName"));
1168+
return gateway.compact(request).thenApply(CompactResponse::getOperationId);
1169+
}
1170+
1171+
@Override
1172+
public CompletableFuture<String> compact(TablePath tablePath, CompactionType type) {
1173+
CompactRequest request = new CompactRequest();
1174+
request.setType(checkNotNull(type, "type").name());
1175+
request.setDatabaseName(tablePath.getDatabaseName()).setTableName(tablePath.getTableName());
1176+
return gateway.compact(request).thenApply(CompactResponse::getOperationId);
1177+
}
1178+
1179+
@Override
1180+
public CompletableFuture<String> compact(
1181+
TablePath tablePath, String partitionName, CompactionType type) {
1182+
CompactRequest request = new CompactRequest();
1183+
request.setType(checkNotNull(type, "type").name());
1184+
request.setDatabaseName(tablePath.getDatabaseName()).setTableName(tablePath.getTableName());
1185+
request.setPartitionName(checkNotNull(partitionName, "partitionName"));
1186+
return gateway.compact(request).thenApply(CompactResponse::getOperationId);
1187+
}
1188+
1189+
@Override
1190+
public CompletableFuture<List<CompactionInfo>> listCompactions() {
1191+
ListCompactionsRequest request = new ListCompactionsRequest();
1192+
return gateway.listCompactions(request)
1193+
.thenApply(
1194+
response ->
1195+
response.getCompactionsList().stream()
1196+
.map(CommonRpcMessageUtils::toCompactionInfo)
1197+
.collect(Collectors.toList()));
1198+
}
1199+
1200+
@Override
1201+
public CompletableFuture<List<CompactionInfo>> listCompactions(String databaseName) {
1202+
ListCompactionsRequest request = new ListCompactionsRequest();
1203+
request.setDatabaseName(checkNotNull(databaseName, "databaseName"));
1204+
return gateway.listCompactions(request)
1205+
.thenApply(
1206+
response ->
1207+
response.getCompactionsList().stream()
1208+
.map(CommonRpcMessageUtils::toCompactionInfo)
1209+
.collect(Collectors.toList()));
1210+
}
1211+
1212+
@Override
1213+
public CompletableFuture<List<CompactionInfo>> listCompactions(TablePath tablePath) {
1214+
ListCompactionsRequest request = new ListCompactionsRequest();
1215+
request.setDatabaseName(tablePath.getDatabaseName()).setTableName(tablePath.getTableName());
1216+
return gateway.listCompactions(request)
1217+
.thenApply(
1218+
response ->
1219+
response.getCompactionsList().stream()
1220+
.map(CommonRpcMessageUtils::toCompactionInfo)
1221+
.collect(Collectors.toList()));
1222+
}
1223+
1224+
@Override
1225+
public CompletableFuture<List<CompactionInfo>> listCompactions(
1226+
TablePath tablePath, String partitionName) {
1227+
ListCompactionsRequest request = new ListCompactionsRequest();
1228+
request.setDatabaseName(tablePath.getDatabaseName()).setTableName(tablePath.getTableName());
1229+
request.setPartitionName(checkNotNull(partitionName, "partitionName"));
1230+
return gateway.listCompactions(request)
1231+
.thenApply(
1232+
response ->
1233+
response.getCompactionsList().stream()
1234+
.map(CommonRpcMessageUtils::toCompactionInfo)
1235+
.collect(Collectors.toList()));
1236+
}
1237+
1238+
@Override
1239+
public CompletableFuture<CompactionProgress> getCompaction(String operationId) {
1240+
return gateway.getCompaction(new GetCompactionRequest().setOperationId(operationId))
1241+
.thenApply(
1242+
response ->
1243+
CommonRpcMessageUtils.toCompactionProgress(response.getProgress()));
1244+
}
1245+
1246+
@Override
1247+
public CompletableFuture<Void> cancelCompaction(String operationId) {
1248+
return gateway.cancelCompaction(new CancelCompactionRequest().setOperationId(operationId))
1249+
.thenApply(response -> null);
1250+
}
11451251
}

0 commit comments

Comments
 (0)