nbalajee commented on code in PR #18279:
URL: https://github.com/apache/hudi/pull/18279#discussion_r3054267863
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/RollbackHelperV1.java:
##########
@@ -80,6 +81,92 @@ public RollbackHelperV1(HoodieTable table, HoodieWriteConfig
config) {
super(table, config);
}
+ /**
+ * Builds the lookup key for pre-computed log versions.
+ */
+ static String logVersionLookupKey(String partitionPath, String fileId,
String commitTime) {
+ return partitionPath + "|" + fileId + "|" + commitTime;
+ }
+
+ /**
+ * Pre-compute the latest log version for each (partition, fileId,
deltaCommitTime) tuple
+ * by listing each unique partition directory once. This replaces N
per-request listing
+ * calls (one per rollback request) with P per-partition listings (where P
is much less than N).
+ *
+ * <p>For file groups with no existing log files in a successfully listed
partition, a sentinel
+ * of (LOGFILE_BASE_VERSION, UNKNOWN_WRITE_TOKEN) is inserted so the caller
avoids a redundant
+ * per-request listing. If a partition listing fails (IOException), no
sentinels are inserted
+ * and the caller falls back to per-request listing naturally.
+ */
+ Map<String, Pair<Integer, String>> preComputeLogVersions(
+ List<SerializableHoodieRollbackRequest> rollbackRequests) {
+ List<SerializableHoodieRollbackRequest> logBlockRequests =
rollbackRequests.stream()
+ .filter(req -> !req.getLogBlocksToBeDeleted().isEmpty())
+ .collect(Collectors.toList());
+
+ if (logBlockRequests.isEmpty()) {
+ return Collections.emptyMap();
+ }
+
+ Map<String, Set<String>> expectedKeysByPartition = new HashMap<>();
+ for (SerializableHoodieRollbackRequest req : logBlockRequests) {
+ String key = logVersionLookupKey(req.getPartitionPath(),
req.getFileId(), req.getLatestBaseInstant());
+ expectedKeysByPartition.computeIfAbsent(req.getPartitionPath(), k -> new
HashSet<>()).add(key);
+ }
+
+ log.info("Pre-computing log versions for {} partition(s) to avoid
per-request listStatus calls",
+ expectedKeysByPartition.size());
+
+ Map<String, Pair<Integer, String>> logVersionMap = new HashMap<>();
+
+ for (Map.Entry<String, Set<String>> entry :
expectedKeysByPartition.entrySet()) {
Review Comment:
Replaced with FSUtils.isLogFile() for filtering , to avoid unnecessary files
getting picked up.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/RollbackHelperV1.java:
##########
@@ -80,6 +81,92 @@ public RollbackHelperV1(HoodieTable table, HoodieWriteConfig
config) {
super(table, config);
}
+ /**
+ * Builds the lookup key for pre-computed log versions.
+ */
+ static String logVersionLookupKey(String partitionPath, String fileId,
String commitTime) {
+ return partitionPath + "|" + fileId + "|" + commitTime;
+ }
+
+ /**
+ * Pre-compute the latest log version for each (partition, fileId,
deltaCommitTime) tuple
+ * by listing each unique partition directory once. This replaces N
per-request listing
+ * calls (one per rollback request) with P per-partition listings (where P
is much less than N).
+ *
+ * <p>For file groups with no existing log files in a successfully listed
partition, a sentinel
+ * of (LOGFILE_BASE_VERSION, UNKNOWN_WRITE_TOKEN) is inserted so the caller
avoids a redundant
+ * per-request listing. If a partition listing fails (IOException), no
sentinels are inserted
+ * and the caller falls back to per-request listing naturally.
+ */
+ Map<String, Pair<Integer, String>> preComputeLogVersions(
+ List<SerializableHoodieRollbackRequest> rollbackRequests) {
+ List<SerializableHoodieRollbackRequest> logBlockRequests =
rollbackRequests.stream()
+ .filter(req -> !req.getLogBlocksToBeDeleted().isEmpty())
+ .collect(Collectors.toList());
+
+ if (logBlockRequests.isEmpty()) {
+ return Collections.emptyMap();
+ }
+
+ Map<String, Set<String>> expectedKeysByPartition = new HashMap<>();
+ for (SerializableHoodieRollbackRequest req : logBlockRequests) {
+ String key = logVersionLookupKey(req.getPartitionPath(),
req.getFileId(), req.getLatestBaseInstant());
+ expectedKeysByPartition.computeIfAbsent(req.getPartitionPath(), k -> new
HashSet<>()).add(key);
+ }
+
+ log.info("Pre-computing log versions for {} partition(s) to avoid
per-request listStatus calls",
+ expectedKeysByPartition.size());
+
+ Map<String, Pair<Integer, String>> logVersionMap = new HashMap<>();
+
+ for (Map.Entry<String, Set<String>> entry :
expectedKeysByPartition.entrySet()) {
+ String relPartPath = entry.getKey();
+ Set<String> expectedKeys = entry.getValue();
+ StoragePath absolutePartPath =
FSUtils.constructAbsolutePath(metaClient.getBasePath(), relPartPath);
+ try {
+ List<StoragePathInfo> statuses =
metaClient.getStorage().listDirectEntries(absolutePartPath,
+ path -> path.getName().contains(HoodieLogFile.DELTA_EXTENSION));
+
+ for (StoragePathInfo status : statuses) {
+ try {
Review Comment:
Replaced with FSUtils.isLogFile() for filtering , to avoid unnecessary files
getting picked up.
--
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]
