yihua commented on code in PR #12826:
URL: https://github.com/apache/hudi/pull/12826#discussion_r1976138615
##########
hudi-common/src/main/java/org/apache/hudi/common/util/ClusteringUtils.java:
##########
@@ -190,22 +189,47 @@ private static Option<HoodieRequestedReplaceMetadata>
getRequestedReplaceMetadat
String action = factory instanceof InstantGeneratorV2 ?
HoodieTimeline.CLUSTERING_ACTION : HoodieTimeline.REPLACE_COMMIT_ACTION;
requestedInstant =
factory.createNewInstant(HoodieInstant.State.REQUESTED, action,
pendingReplaceOrClusterInstant.requestedTime());
}
- Option<byte[]> content = Option.empty();
try {
- content = timeline.getInstantDetails(requestedInstant);
+ // First assume the instant file is not empty and parse it.
+ return getHoodieRequestedReplaceMetadataOption(timeline,
pendingReplaceOrClusterInstant, factory, requestedInstant);
+ } catch (Exception ex) {
+ // If anything goes wrong, check if this is empty file.
+ if (isEmptyReplaceOrClusteringInstant(timeline,
pendingReplaceOrClusterInstant, factory, requestedInstant)) {
+ return Option.empty();
+ }
+ // If still no luck, throw the exception.
+ throw ex;
+ }
+ }
+
+ private static Option<HoodieRequestedReplaceMetadata>
getHoodieRequestedReplaceMetadataOption(
+ HoodieTimeline timeline, HoodieInstant pendingReplaceOrClusterInstant,
InstantGenerator factory, HoodieInstant requestedInstant) throws IOException {
+ try {
+ return
Option.of(timeline.loadRequestedReplaceMetadata(requestedInstant));
} catch (HoodieIOException e) {
if (e.getCause() instanceof FileNotFoundException &&
pendingReplaceOrClusterInstant.isCompleted()) {
// For clustering instants, completed instant is also a replace commit
instant. For input replace commit instant,
// it is not known whether requested instant is CLUSTER or
REPLACE_COMMIT_ACTION. So we need to query both.
requestedInstant =
factory.createNewInstant(HoodieInstant.State.REQUESTED,
HoodieTimeline.REPLACE_COMMIT_ACTION,
pendingReplaceOrClusterInstant.requestedTime());
- content = timeline.getInstantDetails(requestedInstant);
+ return
Option.of(timeline.loadRequestedReplaceMetadata(requestedInstant));
}
+ throw e;
}
- if (!content.isPresent() || content.get().length == 0) {
- // few operations create requested file without any content. Assume
these are not clustering
- return Option.empty();
+ }
+
+ private static boolean isEmptyReplaceOrClusteringInstant(
+ HoodieTimeline timeline, HoodieInstant pendingReplaceOrClusterInstant,
InstantGenerator factory, HoodieInstant requestedInstant) {
+ try {
+ return timeline.isEmpty(requestedInstant);
+ } catch (HoodieIOException e) {
+ if (e.getCause() instanceof FileNotFoundException &&
pendingReplaceOrClusterInstant.isCompleted()) {
+ // For clustering instants, completed instant is also a replace commit
instant. For input replace commit instant,
+ // it is not known whether requested instant is CLUSTER or
REPLACE_COMMIT_ACTION. So we need to query both.
+ requestedInstant =
factory.createNewInstant(HoodieInstant.State.REQUESTED,
HoodieTimeline.REPLACE_COMMIT_ACTION,
pendingReplaceOrClusterInstant.requestedTime());
Review Comment:
This follows the same logic as `getRequestedReplaceMetadataOption`. Also,
the new request instant is only constructed during exception handling after
`timeline.isEmpty(requestedInstant)` is called on the requested cluster instant
if it exists. So this is OK.
--
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]
