otterc commented on code in PR #37922:
URL: https://github.com/apache/spark/pull/37922#discussion_r1068388955
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -396,6 +403,59 @@ public void applicationRemoved(String appId, boolean
cleanupLocalDirs) {
}
}
+ @Override
+ public void removeShuffleMerge(RemoveShuffleMerge msg) {
+ AppShuffleInfo appShuffleInfo = validateAndGetAppShuffleInfo(msg.appId);
+ if (appShuffleInfo.attemptId != msg.appAttemptId) {
+ throw new IllegalArgumentException(
+ String.format("The attempt id %s in this RemoveShuffleMerge message
does not match "
+ + "with the current attempt id %s stored in shuffle service
for application %s",
+ msg.appAttemptId, appShuffleInfo.attemptId, msg.appId));
+ }
+ appShuffleInfo.shuffles.compute(msg.shuffleId, (shuffleId,
mergePartitionsInfo) -> {
+ if (mergePartitionsInfo == null) {
+ if (msg.shuffleMergeId == DELETE_ALL_MERGED_SHUFFLE) {
+ return null;
+ } else {
+ writeAppAttemptShuffleMergeInfoToDB(new AppAttemptShuffleMergeId(
+ msg.appId, msg.appAttemptId, msg.shuffleId, msg.shuffleMergeId));
+ return new AppShuffleMergePartitionsInfo(msg.shuffleMergeId, true);
+ }
+ }
+ boolean deleteCurrentMergedShuffle =
+ msg.shuffleMergeId == DELETE_ALL_MERGED_SHUFFLE ||
+ msg.shuffleMergeId == mergePartitionsInfo.shuffleMergeId;
+ int shuffleMergeId = msg.shuffleMergeId != DELETE_ALL_MERGED_SHUFFLE ?
+ msg.shuffleMergeId : mergePartitionsInfo.shuffleMergeId;
+ AppAttemptShuffleMergeId currentAppAttemptShuffleMergeId =
+ new AppAttemptShuffleMergeId(
+ msg.appId, msg.appAttemptId, msg.shuffleId,
mergePartitionsInfo.shuffleMergeId);
+ AppAttemptShuffleMergeId appAttemptShuffleMergeId = new
AppAttemptShuffleMergeId(
+ msg.appId, msg.appAttemptId, msg.shuffleId, shuffleMergeId);
+ if(deleteCurrentMergedShuffle || shuffleMergeId >
mergePartitionsInfo.shuffleMergeId) {
+ if (!mergePartitionsInfo.isFinalized()) {
+ // Clean up shuffle data before the shuffle was finalized. Close and
delete all the open
+ // files.
+ submitCleanupTask(() ->
+ closeAndDeleteOutdatedPartitions(
+ currentAppAttemptShuffleMergeId,
mergePartitionsInfo.shuffleMergePartitions));
+ } else {
+ // Current shuffle was finalized, delete all the merged files
through reduceIds set
+ // in finalizeShuffleMerge method.
+ submitCleanupTask(() ->
+ deleteMergedFiles(currentAppAttemptShuffleMergeId,
appShuffleInfo,
+ mergePartitionsInfo.getReduceIds(), false));
+ }
+ } else if(shuffleMergeId < mergePartitionsInfo.shuffleMergeId) {
Review Comment:
nit: space missing after `if`. Also the condition here is not necessary. It
will always be true (IDE shows that as well).
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -396,6 +403,59 @@ public void applicationRemoved(String appId, boolean
cleanupLocalDirs) {
}
}
+ @Override
+ public void removeShuffleMerge(RemoveShuffleMerge msg) {
+ AppShuffleInfo appShuffleInfo = validateAndGetAppShuffleInfo(msg.appId);
+ if (appShuffleInfo.attemptId != msg.appAttemptId) {
+ throw new IllegalArgumentException(
+ String.format("The attempt id %s in this RemoveShuffleMerge message
does not match "
+ + "with the current attempt id %s stored in shuffle service
for application %s",
+ msg.appAttemptId, appShuffleInfo.attemptId, msg.appId));
+ }
+ appShuffleInfo.shuffles.compute(msg.shuffleId, (shuffleId,
mergePartitionsInfo) -> {
+ if (mergePartitionsInfo == null) {
+ if (msg.shuffleMergeId == DELETE_ALL_MERGED_SHUFFLE) {
+ return null;
+ } else {
+ writeAppAttemptShuffleMergeInfoToDB(new AppAttemptShuffleMergeId(
+ msg.appId, msg.appAttemptId, msg.shuffleId, msg.shuffleMergeId));
+ return new AppShuffleMergePartitionsInfo(msg.shuffleMergeId, true);
+ }
+ }
+ boolean deleteCurrentMergedShuffle =
+ msg.shuffleMergeId == DELETE_ALL_MERGED_SHUFFLE ||
+ msg.shuffleMergeId == mergePartitionsInfo.shuffleMergeId;
+ int shuffleMergeId = msg.shuffleMergeId != DELETE_ALL_MERGED_SHUFFLE ?
+ msg.shuffleMergeId : mergePartitionsInfo.shuffleMergeId;
+ AppAttemptShuffleMergeId currentAppAttemptShuffleMergeId =
+ new AppAttemptShuffleMergeId(
+ msg.appId, msg.appAttemptId, msg.shuffleId,
mergePartitionsInfo.shuffleMergeId);
+ AppAttemptShuffleMergeId appAttemptShuffleMergeId = new
AppAttemptShuffleMergeId(
+ msg.appId, msg.appAttemptId, msg.shuffleId, shuffleMergeId);
+ if(deleteCurrentMergedShuffle || shuffleMergeId >
mergePartitionsInfo.shuffleMergeId) {
Review Comment:
Nit: missing a space after `if`
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -470,6 +530,40 @@ void closeAndDeleteOutdatedPartitions(
});
}
+ void deleteMergedFiles(
+ AppAttemptShuffleMergeId appAttemptShuffleMergeId,
+ AppShuffleInfo appShuffleInfo,
+ int[] reduceIds,
+ boolean deleteFromDB) {
+ if(deleteFromDB) {
Review Comment:
When is `deleteFromDb` here true?
Nit: no space after `if`.
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -396,6 +403,59 @@ public void applicationRemoved(String appId, boolean
cleanupLocalDirs) {
}
}
+ @Override
+ public void removeShuffleMerge(RemoveShuffleMerge msg) {
+ AppShuffleInfo appShuffleInfo = validateAndGetAppShuffleInfo(msg.appId);
+ if (appShuffleInfo.attemptId != msg.appAttemptId) {
+ throw new IllegalArgumentException(
+ String.format("The attempt id %s in this RemoveShuffleMerge message
does not match "
+ + "with the current attempt id %s stored in shuffle service
for application %s",
+ msg.appAttemptId, appShuffleInfo.attemptId, msg.appId));
+ }
+ appShuffleInfo.shuffles.compute(msg.shuffleId, (shuffleId,
mergePartitionsInfo) -> {
+ if (mergePartitionsInfo == null) {
+ if (msg.shuffleMergeId == DELETE_ALL_MERGED_SHUFFLE) {
+ return null;
+ } else {
+ writeAppAttemptShuffleMergeInfoToDB(new AppAttemptShuffleMergeId(
+ msg.appId, msg.appAttemptId, msg.shuffleId, msg.shuffleMergeId));
+ return new AppShuffleMergePartitionsInfo(msg.shuffleMergeId, true);
+ }
+ }
+ boolean deleteCurrentMergedShuffle =
+ msg.shuffleMergeId == DELETE_ALL_MERGED_SHUFFLE ||
+ msg.shuffleMergeId == mergePartitionsInfo.shuffleMergeId;
+ int shuffleMergeId = msg.shuffleMergeId != DELETE_ALL_MERGED_SHUFFLE ?
+ msg.shuffleMergeId : mergePartitionsInfo.shuffleMergeId;
+ AppAttemptShuffleMergeId currentAppAttemptShuffleMergeId =
+ new AppAttemptShuffleMergeId(
+ msg.appId, msg.appAttemptId, msg.shuffleId,
mergePartitionsInfo.shuffleMergeId);
+ AppAttemptShuffleMergeId appAttemptShuffleMergeId = new
AppAttemptShuffleMergeId(
+ msg.appId, msg.appAttemptId, msg.shuffleId, shuffleMergeId);
+ if(deleteCurrentMergedShuffle || shuffleMergeId >
mergePartitionsInfo.shuffleMergeId) {
+ if (!mergePartitionsInfo.isFinalized()) {
+ // Clean up shuffle data before the shuffle was finalized. Close and
delete all the open
+ // files.
+ submitCleanupTask(() ->
+ closeAndDeleteOutdatedPartitions(
+ currentAppAttemptShuffleMergeId,
mergePartitionsInfo.shuffleMergePartitions));
+ } else {
+ // Current shuffle was finalized, delete all the merged files
through reduceIds set
+ // in finalizeShuffleMerge method.
+ submitCleanupTask(() ->
+ deleteMergedFiles(currentAppAttemptShuffleMergeId,
appShuffleInfo,
+ mergePartitionsInfo.getReduceIds(), false));
+ }
+ } else if(shuffleMergeId < mergePartitionsInfo.shuffleMergeId) {
+ throw new RuntimeException(String.format("Asked to remove old shuffle
merged data for " +
+ "application %s shuffleId %s shuffleMergeId %s, but current
shuffleMergeId %s ",
+ msg.appId, msg.shuffleId, shuffleMergeId,
mergePartitionsInfo.shuffleMergeId));
+ }
+ writeAppAttemptShuffleMergeInfoToDB(appAttemptShuffleMergeId);
+ return new AppShuffleMergePartitionsInfo(shuffleMergeId, true);
+ });
Review Comment:
```suggestion
int shuffleMergeIdToDelete = msg.shuffleMergeId !=
DELETE_ALL_MERGED_SHUFFLE ?
msg.shuffleMergeId : mergePartitionsInfo.shuffleMergeId;
if (deleteCurrentMergedShuffle ||
shuffleMergeIdToDelete > mergePartitionsInfo.shuffleMergeId) {
AppAttemptShuffleMergeId currentAppAttemptShuffleMergeId =
new AppAttemptShuffleMergeId(
msg.appId, msg.appAttemptId, msg.shuffleId,
mergePartitionsInfo.shuffleMergeId);
if (!mergePartitionsInfo.isFinalized()) {
// Clean up shuffle data before the shuffle was finalized. Close
and delete all the open
// files.
submitCleanupTask(() ->
closeAndDeleteOutdatedPartitions(
currentAppAttemptShuffleMergeId,
mergePartitionsInfo.shuffleMergePartitions));
} else {
// Current shuffle was finalized, delete all the merged files
through reduceIds set
// in finalizeShuffleMerge method.
submitCleanupTask(() ->
deleteMergedFiles(currentAppAttemptShuffleMergeId,
appShuffleInfo,
mergePartitionsInfo.getReduceIds(), false));
}
} else {
throw new RuntimeException(String.format("Asked to remove old
shuffle merged data for " +
"application %s shuffleId %s shuffleMergeId %s, but current
shuffleMergeId %s ",
msg.appId, msg.shuffleId, shuffleMergeIdToDelete,
mergePartitionsInfo.shuffleMergeId));
}
writeAppAttemptShuffleMergeInfoToDB(new AppAttemptShuffleMergeId(
msg.appId, msg.appAttemptId, msg.shuffleId,
shuffleMergeIdToDelete));
return new AppShuffleMergePartitionsInfo(shuffleMergeIdToDelete, true);
});
```
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -470,6 +530,40 @@ void closeAndDeleteOutdatedPartitions(
});
}
+ void deleteMergedFiles(
+ AppAttemptShuffleMergeId appAttemptShuffleMergeId,
+ AppShuffleInfo appShuffleInfo,
+ int[] reduceIds,
+ boolean deleteFromDB) {
+ if(deleteFromDB) {
+ removeAppShufflePartitionInfoFromDB(appAttemptShuffleMergeId);
+ }
+ int shuffleId = appAttemptShuffleMergeId.shuffleId;
+ int shuffleMergeId = appAttemptShuffleMergeId.shuffleMergeId;
+ int dataFilesDeleteCnt = 0;
+ int indexFilesDeleteCnt = 0;
+ int metaFilesDeleteCnt = 0;
+ for (int reduceId : reduceIds) {
+ File dataFile =
+ appShuffleInfo.getMergedShuffleDataFile(shuffleId, shuffleMergeId,
reduceId);
+ if (dataFile.delete()) {
Review Comment:
If this throws exception, then it will not try to delete index and meta
file. I think we should try to delete other files even when there is an
exception deleting one.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
