Skip to content

Commit 69556c3

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 1fd0971 commit 69556c3

84 files changed

Lines changed: 12139 additions & 475 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;
@@ -939,4 +942,36 @@ CompletableFuture<List<RemoteLogManifestInfo>> listRemoteLogManifests(
939942
* @return per-bucket active snapshot ids grouped by bucket
940943
*/
941944
CompletableFuture<ActiveKvSnapshots> listKvSnapshots(long tableId, @Nullable Long partitionId);
945+
946+
/** Submits cluster compaction. */
947+
CompletableFuture<String> compact(CompactionType type);
948+
949+
/** Submits database compaction. */
950+
CompletableFuture<String> compact(String databaseName, CompactionType type);
951+
952+
/** Submits table compaction. */
953+
CompletableFuture<String> compact(TablePath tablePath, CompactionType type);
954+
955+
/** Submits partition compaction. */
956+
CompletableFuture<String> compact(
957+
TablePath tablePath, String partitionName, CompactionType type);
958+
959+
/** Lists cluster compactions. */
960+
CompletableFuture<List<CompactionInfo>> listCompactions();
961+
962+
/** Lists compactions associated with a database. */
963+
CompletableFuture<List<CompactionInfo>> listCompactions(String databaseName);
964+
965+
/** Lists compactions associated with a table. */
966+
CompletableFuture<List<CompactionInfo>> listCompactions(TablePath tablePath);
967+
968+
/** Lists compactions associated with a partition. */
969+
CompletableFuture<List<CompactionInfo>> listCompactions(
970+
TablePath tablePath, String partitionName);
971+
972+
/** Returns progress of an operation. */
973+
CompletableFuture<CompactionProgress> getCompaction(String operationId);
974+
975+
/** Cancels an operation; repeated cancellation is allowed. */
976+
CompletableFuture<Void> cancelCompaction(String operationId);
942977
}

‎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.CreateTableFromLakeRequest;
@@ -79,6 +85,7 @@
7985
import org.apache.fluss.rpc.messages.DropDatabaseRequest;
8086
import org.apache.fluss.rpc.messages.DropTableRequest;
8187
import org.apache.fluss.rpc.messages.GetClusterHealthRequest;
88+
import org.apache.fluss.rpc.messages.GetCompactionRequest;
8289
import org.apache.fluss.rpc.messages.GetDatabaseInfoRequest;
8390
import org.apache.fluss.rpc.messages.GetKvSnapshotMetadataRequest;
8491
import org.apache.fluss.rpc.messages.GetLakeSnapshotRequest;
@@ -89,6 +96,7 @@
8996
import org.apache.fluss.rpc.messages.GetTableStatsRequest;
9097
import org.apache.fluss.rpc.messages.GetTableStatsResponse;
9198
import org.apache.fluss.rpc.messages.ListAclsRequest;
99+
import org.apache.fluss.rpc.messages.ListCompactionsRequest;
92100
import org.apache.fluss.rpc.messages.ListDatabasesRequest;
93101
import org.apache.fluss.rpc.messages.ListDatabasesResponse;
94102
import org.apache.fluss.rpc.messages.ListKvSnapshotsRequest;
@@ -114,6 +122,7 @@
114122
import org.apache.fluss.rpc.messages.TableExistsRequest;
115123
import org.apache.fluss.rpc.messages.TableExistsResponse;
116124
import org.apache.fluss.rpc.protocol.ApiError;
125+
import org.apache.fluss.rpc.util.CommonRpcMessageUtils;
117126
import org.apache.fluss.security.acl.AclBinding;
118127
import org.apache.fluss.security.acl.AclBindingFilter;
119128
import org.apache.fluss.utils.ExceptionUtils;
@@ -133,6 +142,7 @@
133142
import java.util.concurrent.ConcurrentHashMap;
134143
import java.util.concurrent.ExecutorService;
135144
import java.util.concurrent.Executors;
145+
import java.util.stream.Collectors;
136146

137147
import static org.apache.fluss.client.utils.ClientRpcMessageUtils.makeAlterDatabaseRequest;
138148
import static org.apache.fluss.client.utils.ClientRpcMessageUtils.makeAlterTableRequest;
@@ -1205,4 +1215,100 @@ private static void validateRackRequest(List<String> racks, ServerTag serverTag)
12051215
"rack element must not be null, empty, or whitespace-only");
12061216
}
12071217
}
1218+
1219+
@Override
1220+
public CompletableFuture<String> compact(CompactionType type) {
1221+
CompactRequest request = new CompactRequest();
1222+
request.setType(checkNotNull(type, "type").name());
1223+
return gateway.compact(request).thenApply(CompactResponse::getOperationId);
1224+
}
1225+
1226+
@Override
1227+
public CompletableFuture<String> compact(String databaseName, CompactionType type) {
1228+
CompactRequest request = new CompactRequest();
1229+
request.setType(checkNotNull(type, "type").name());
1230+
request.setDatabaseName(checkNotNull(databaseName, "databaseName"));
1231+
return gateway.compact(request).thenApply(CompactResponse::getOperationId);
1232+
}
1233+
1234+
@Override
1235+
public CompletableFuture<String> compact(TablePath tablePath, CompactionType type) {
1236+
CompactRequest request = new CompactRequest();
1237+
request.setType(checkNotNull(type, "type").name());
1238+
request.setDatabaseName(tablePath.getDatabaseName()).setTableName(tablePath.getTableName());
1239+
return gateway.compact(request).thenApply(CompactResponse::getOperationId);
1240+
}
1241+
1242+
@Override
1243+
public CompletableFuture<String> compact(
1244+
TablePath tablePath, String partitionName, CompactionType type) {
1245+
CompactRequest request = new CompactRequest();
1246+
request.setType(checkNotNull(type, "type").name());
1247+
request.setDatabaseName(tablePath.getDatabaseName()).setTableName(tablePath.getTableName());
1248+
request.setPartitionName(checkNotNull(partitionName, "partitionName"));
1249+
return gateway.compact(request).thenApply(CompactResponse::getOperationId);
1250+
}
1251+
1252+
@Override
1253+
public CompletableFuture<List<CompactionInfo>> listCompactions() {
1254+
ListCompactionsRequest request = new ListCompactionsRequest();
1255+
return gateway.listCompactions(request)
1256+
.thenApply(
1257+
response ->
1258+
response.getCompactionsList().stream()
1259+
.map(CommonRpcMessageUtils::toCompactionInfo)
1260+
.collect(Collectors.toList()));
1261+
}
1262+
1263+
@Override
1264+
public CompletableFuture<List<CompactionInfo>> listCompactions(String databaseName) {
1265+
ListCompactionsRequest request = new ListCompactionsRequest();
1266+
request.setDatabaseName(checkNotNull(databaseName, "databaseName"));
1267+
return gateway.listCompactions(request)
1268+
.thenApply(
1269+
response ->
1270+
response.getCompactionsList().stream()
1271+
.map(CommonRpcMessageUtils::toCompactionInfo)
1272+
.collect(Collectors.toList()));
1273+
}
1274+
1275+
@Override
1276+
public CompletableFuture<List<CompactionInfo>> listCompactions(TablePath tablePath) {
1277+
ListCompactionsRequest request = new ListCompactionsRequest();
1278+
request.setDatabaseName(tablePath.getDatabaseName()).setTableName(tablePath.getTableName());
1279+
return gateway.listCompactions(request)
1280+
.thenApply(
1281+
response ->
1282+
response.getCompactionsList().stream()
1283+
.map(CommonRpcMessageUtils::toCompactionInfo)
1284+
.collect(Collectors.toList()));
1285+
}
1286+
1287+
@Override
1288+
public CompletableFuture<List<CompactionInfo>> listCompactions(
1289+
TablePath tablePath, String partitionName) {
1290+
ListCompactionsRequest request = new ListCompactionsRequest();
1291+
request.setDatabaseName(tablePath.getDatabaseName()).setTableName(tablePath.getTableName());
1292+
request.setPartitionName(checkNotNull(partitionName, "partitionName"));
1293+
return gateway.listCompactions(request)
1294+
.thenApply(
1295+
response ->
1296+
response.getCompactionsList().stream()
1297+
.map(CommonRpcMessageUtils::toCompactionInfo)
1298+
.collect(Collectors.toList()));
1299+
}
1300+
1301+
@Override
1302+
public CompletableFuture<CompactionProgress> getCompaction(String operationId) {
1303+
return gateway.getCompaction(new GetCompactionRequest().setOperationId(operationId))
1304+
.thenApply(
1305+
response ->
1306+
CommonRpcMessageUtils.toCompactionProgress(response.getProgress()));
1307+
}
1308+
1309+
@Override
1310+
public CompletableFuture<Void> cancelCompaction(String operationId) {
1311+
return gateway.cancelCompaction(new CancelCompactionRequest().setOperationId(operationId))
1312+
.thenApply(response -> null);
1313+
}
12081314
}

0 commit comments

Comments
 (0)