[HUDI-7576] Improve efficiency of getRelativePartitionPath, reduce computation of partitionPath in AbstractTableFileSystemView (apache#11001)

Author: the-other-tim-brown

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/clean/CleanPlanner.java

@@ -467,7 +467,7 @@ private boolean hasPendingFiles(String partitionPath) {
     try {
       HoodieTableFileSystemView fsView = new HoodieTableFileSystemView(hoodieTable.getMetaClient(), hoodieTable.getActiveTimeline());
       StoragePath fullPartitionPath = new StoragePath(hoodieTable.getMetaClient().getBasePathV2(), partitionPath);
-      fsView.addFilesToView(FSUtils.getAllDataFilesInPartition(
+      fsView.addFilesToView(partitionPath, FSUtils.getAllDataFilesInPartition(
           hoodieTable.getMetaClient().getStorage(), fullPartitionPath));
       // use #getAllFileGroups(partitionPath) instead of #getAllFileGroups() to exclude the replaced file groups.
       return fsView.getAllFileGroups(partitionPath).findAny().isPresent();

hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/commit/TestUpsertPartitioner.java

@@ -469,9 +469,9 @@ public void testUpsertPartitionerWithSmallFileHandlingPickingMultipleCandidates(
     assertEquals(3, partitioner.numPartitions());
     assertEquals(
         Arrays.asList(
-            new BucketInfo(BucketType.UPDATE, "fg-1", partitionPath),
+            new BucketInfo(BucketType.UPDATE, "fg-3", partitionPath),
             new BucketInfo(BucketType.UPDATE, "fg-2", partitionPath),
-            new BucketInfo(BucketType.UPDATE, "fg-3", partitionPath)
+            new BucketInfo(BucketType.UPDATE, "fg-1", partitionPath)
         ),
         partitioner.getBucketInfos());
   }

hudi-common/src/main/java/org/apache/hudi/common/fs/FSUtils.java

@@ -234,17 +234,16 @@ public static String getRelativePartitionPath(StoragePath basePath, StoragePath
     fullPartitionPath = getPathWithoutSchemeAndAuthority(fullPartitionPath);
 
     String fullPartitionPathStr = fullPartitionPath.toString();
+    String basePathString = basePath.toString();
 
-    if (!fullPartitionPathStr.startsWith(basePath.toString())) {
+    if (!fullPartitionPathStr.startsWith(basePathString)) {
       throw new IllegalArgumentException("Partition path \"" + fullPartitionPathStr
           + "\" does not belong to base-path \"" + basePath + "\"");
     }
 
-    int partitionStartIndex = fullPartitionPathStr.indexOf(basePath.getName(),
-        basePath.getParent() == null ? 0 : basePath.getParent().toString().length());
     // Partition-Path could be empty for non-partitioned tables
-    return partitionStartIndex + basePath.getName().length() == fullPartitionPathStr.length() ? ""
-        : fullPartitionPathStr.substring(partitionStartIndex + basePath.getName().length() + 1);
+    return fullPartitionPathStr.length() == basePathString.length() ? ""
+        : fullPartitionPathStr.substring(basePathString.length() + 1);
   }
 
   public static StoragePath getPathWithoutSchemeAndAuthority(StoragePath path) {

hudi-common/src/main/java/org/apache/hudi/common/table/view/AbstractTableFileSystemView.java

@@ -109,10 +109,6 @@ public abstract class AbstractTableFileSystemView implements SyncableFileSystemV
 
   private BootstrapIndex bootstrapIndex;
 
-  private String getPartitionPathFor(HoodieBaseFile baseFile) {
-    return FSUtils.getRelativePartitionPath(metaClient.getBasePathV2(), baseFile.getStoragePath().getParent());
-  }
-
   /**
    * Initialize the view.
    */
@@ -158,10 +154,21 @@ public Option<String> getCompletionTime(String instantTime) {
 
   /**
    * Adds the provided statuses into the file system view, and also caches it inside this object.
+   * If the file statuses are limited to a single partition, use {@link #addFilesToView(String, List)} instead.
    */
   public List<HoodieFileGroup> addFilesToView(List<StoragePathInfo> statuses) {
+    Map<String, List<StoragePathInfo>> statusesByPartitionPath = statuses.stream()
+        .collect(Collectors.groupingBy(fileStatus -> FSUtils.getRelativePartitionPath(metaClient.getBasePathV2(), fileStatus.getPath().getParent())));
+    return statusesByPartitionPath.entrySet().stream().map(entry -> addFilesToView(entry.getKey(), entry.getValue()))
+        .flatMap(List::stream).collect(Collectors.toList());
+  }
+
+  /**
+   * Adds the provided statuses into the file system view for a single partition, and also caches it inside this object.
+   */
+  public List<HoodieFileGroup> addFilesToView(String partitionPath, List<StoragePathInfo> statuses) {
     HoodieTimer timer = HoodieTimer.start();
-    List<HoodieFileGroup> fileGroups = buildFileGroups(statuses, visibleCommitsAndCompactionTimeline, true);
+    List<HoodieFileGroup> fileGroups = buildFileGroups(partitionPath, statuses, visibleCommitsAndCompactionTimeline, true);
     long fgBuildTimeTakenMs = timer.endTimer();
     timer.startTimer();
     // Group by partition for efficient updates for both InMemory and DiskBased structures.
@@ -191,37 +198,28 @@ public List<HoodieFileGroup> addFilesToView(List<StoragePathInfo> statuses) {
   /**
    * Build FileGroups from passed in file-status.
    */
-  protected List<HoodieFileGroup> buildFileGroups(List<StoragePathInfo> statuses, HoodieTimeline timeline,
+  protected List<HoodieFileGroup> buildFileGroups(String partition, List<StoragePathInfo> statuses, HoodieTimeline timeline,
                                                   boolean addPendingCompactionFileSlice) {
-    return buildFileGroups(convertFileStatusesToBaseFiles(statuses), convertFileStatusesToLogFiles(statuses),
+    return buildFileGroups(partition, convertFileStatusesToBaseFiles(statuses), convertFileStatusesToLogFiles(statuses),
         timeline,
         addPendingCompactionFileSlice);
   }
 
-  protected List<HoodieFileGroup> buildFileGroups(Stream<HoodieBaseFile> baseFileStream,
+  protected List<HoodieFileGroup> buildFileGroups(String partition, Stream<HoodieBaseFile> baseFileStream,
                                                   Stream<HoodieLogFile> logFileStream, HoodieTimeline timeline, boolean addPendingCompactionFileSlice) {
-    Map<Pair<String, String>, List<HoodieBaseFile>> baseFiles =
-        baseFileStream.collect(Collectors.groupingBy(baseFile -> {
-          String partitionPathStr = getPartitionPathFor(baseFile);
-          return Pair.of(partitionPathStr, baseFile.getFileId());
-        }));
-
-    Map<Pair<String, String>, List<HoodieLogFile>> logFiles = logFileStream.collect(Collectors.groupingBy((logFile) -> {
-      String partitionPathStr =
-          FSUtils.getRelativePartitionPath(metaClient.getBasePathV2(), logFile.getPath().getParent());
-      return Pair.of(partitionPathStr, logFile.getFileId());
-    }));
-
-    Set<Pair<String, String>> fileIdSet = new HashSet<>(baseFiles.keySet());
+    Map<String, List<HoodieBaseFile>> baseFiles =
+        baseFileStream.collect(Collectors.groupingBy(HoodieBaseFile::getFileId));
+
+    Map<String, List<HoodieLogFile>> logFiles = logFileStream.collect(Collectors.groupingBy(HoodieLogFile::getFileId));
+
+    Set<String> fileIdSet = new HashSet<>(baseFiles.keySet());
     fileIdSet.addAll(logFiles.keySet());
 
-    List<HoodieFileGroup> fileGroups = new ArrayList<>();
-    fileIdSet.forEach(pair -> {
-      String fileId = pair.getValue();
-      String partitionPath = pair.getKey();
-      HoodieFileGroup group = new HoodieFileGroup(partitionPath, fileId, timeline);
-      if (baseFiles.containsKey(pair)) {
-        baseFiles.get(pair).forEach(group::addBaseFile);
+    List<HoodieFileGroup> fileGroups = new ArrayList<>(fileIdSet.size());
+    fileIdSet.forEach(fileId -> {
+      HoodieFileGroup group = new HoodieFileGroup(partition, fileId, timeline);
+      if (baseFiles.containsKey(fileId)) {
+        baseFiles.get(fileId).forEach(group::addBaseFile);
       }
       if (addPendingCompactionFileSlice) {
         // pending compaction file slice must be added before log files so that
@@ -235,8 +233,8 @@ protected List<HoodieFileGroup> buildFileGroups(Stream<HoodieBaseFile> baseFileS
           group.addNewFileSliceAtInstant(pendingCompaction.get().getKey());
         }
       }
-      if (logFiles.containsKey(pair)) {
-        logFiles.get(pair).stream().sorted(HoodieLogFile.getLogFileComparator()).forEach(logFile -> group.addLogFile(completionTimeQueryView, logFile));
+      if (logFiles.containsKey(fileId)) {
+        logFiles.get(fileId).stream().sorted(HoodieLogFile.getLogFileComparator()).forEach(logFile -> group.addLogFile(completionTimeQueryView, logFile));
       }
       fileGroups.add(group);
     });
@@ -379,9 +377,9 @@ private void ensurePartitionsLoadedCorrectly(List<String> partitionList) {
           LOG.debug("Time taken to list partitions " + partitionSet + " =" + (endLsTs - beginLsTs));
           pathInfoMap.forEach((partitionPair, statuses) -> {
             String relativePartitionStr = partitionPair.getLeft();
-            List<HoodieFileGroup> groups = addFilesToView(statuses);
+            List<HoodieFileGroup> groups = addFilesToView(relativePartitionStr, statuses);
             if (groups.isEmpty()) {
-              storePartitionView(relativePartitionStr, new ArrayList<>());
+              storePartitionView(relativePartitionStr, Collections.emptyList());
             }
             LOG.debug("#files found in partition (" + relativePartitionStr + ") =" + statuses.size());
           });
@@ -469,7 +467,7 @@ protected void ensurePartitionLoadedCorrectly(String partition) {
         // Not loaded yet
         try {
           LOG.info("Building file system view for partition (" + partitionPathStr + ")");
-          List<HoodieFileGroup> groups = addFilesToView(getAllFilesInPartition(partitionPathStr));
+          List<HoodieFileGroup> groups = addFilesToView(partitionPathStr, getAllFilesInPartition(partitionPathStr));
           if (groups.isEmpty()) {
             storePartitionView(partitionPathStr, new ArrayList<>());
           }
@@ -541,11 +539,10 @@ private Stream<HoodieLogFile> convertFileStatusesToLogFiles(List<StoragePathInfo
    * With async compaction, it is possible to see partial/complete base-files due to inflight-compactions, Ignore those
    * base-files.
    *
+   * @param partitionPath partition path for the base file
    * @param baseFile base File
    */
-  protected boolean isBaseFileDueToPendingCompaction(HoodieBaseFile baseFile) {
-    final String partitionPath = getPartitionPathFor(baseFile);
-
+  protected boolean isBaseFileDueToPendingCompaction(String partitionPath, HoodieBaseFile baseFile) {
     Option<Pair<String, CompactionOperation>> compactionWithInstantTime =
         getPendingCompactionOperationWithInstant(new HoodieFileGroupId(partitionPath, baseFile.getFileId()));
     return (compactionWithInstantTime.isPresent()) && (null != compactionWithInstantTime.get().getKey())
@@ -745,7 +742,7 @@ private Stream<HoodieBaseFile> getLatestBaseFilesBeforeOrOnFromCache(String part
         .map(fileGroup -> Option.fromJavaOptional(fileGroup.getAllBaseFiles()
             .filter(baseFile -> HoodieTimeline.compareTimestamps(baseFile.getCommitTime(), HoodieTimeline.LESSER_THAN_OR_EQUALS, maxCommitTime
             ))
-            .filter(df -> !isBaseFileDueToPendingCompaction(df) && !isBaseFileDueToPendingClustering(df)).findFirst()))
+            .filter(df -> !isBaseFileDueToPendingCompaction(partitionPath, df) && !isBaseFileDueToPendingClustering(df)).findFirst()))
         .filter(Option::isPresent).map(Option::get)
         .map(df -> addBootstrapBaseFileIfPresent(new HoodieFileGroupId(partitionPath, df.getFileId()), df));
   }
@@ -761,7 +758,7 @@ public final Option<HoodieBaseFile> getBaseFileOn(String partitionStr, String in
       } else {
         return fetchHoodieFileGroup(partitionPath, fileId).map(fileGroup -> fileGroup.getAllBaseFiles()
                 .filter(baseFile -> HoodieTimeline.compareTimestamps(baseFile.getCommitTime(), HoodieTimeline.EQUALS,
-                    instantTime)).filter(df -> !isBaseFileDueToPendingCompaction(df) && !isBaseFileDueToPendingClustering(df)).findFirst().orElse(null))
+                    instantTime)).filter(df -> !isBaseFileDueToPendingCompaction(partitionPath, df) && !isBaseFileDueToPendingClustering(df)).findFirst().orElse(null))
             .map(df -> addBootstrapBaseFileIfPresent(new HoodieFileGroupId(partitionPath, fileId), df));
       }
     } finally {
@@ -797,7 +794,7 @@ public final Stream<HoodieBaseFile> getLatestBaseFilesInRange(List<String> commi
           .filter(fileGroup -> !isFileGroupReplacedBeforeAny(fileGroup.getFileGroupId(), commitsToReturn))
           .map(fileGroup -> Pair.of(fileGroup.getFileGroupId(), Option.fromJavaOptional(
               fileGroup.getAllBaseFiles().filter(baseFile -> commitsToReturn.contains(baseFile.getCommitTime())
-                  && !isBaseFileDueToPendingCompaction(baseFile) && !isBaseFileDueToPendingClustering(baseFile)).findFirst()))).filter(p -> p.getValue().isPresent())
+                  && !isBaseFileDueToPendingCompaction(fileGroup.getPartitionPath(), baseFile) && !isBaseFileDueToPendingClustering(baseFile)).findFirst()))).filter(p -> p.getValue().isPresent())
           .map(p -> addBootstrapBaseFileIfPresent(p.getKey(), p.getValue().get()));
     } finally {
       readLock.unlock();
@@ -833,7 +830,7 @@ public final Stream<HoodieBaseFile> getAllBaseFiles(String partitionStr) {
       return fetchAllBaseFiles(partitionPath)
           .filter(df -> !isFileGroupReplaced(partitionPath, df.getFileId()))
           .filter(df -> visibleCommitsAndCompactionTimeline.containsOrBeforeTimelineStarts(df.getCommitTime()))
-          .filter(df -> !isBaseFileDueToPendingCompaction(df) && !isBaseFileDueToPendingClustering(df))
+          .filter(df -> !isBaseFileDueToPendingCompaction(partitionPath, df) && !isBaseFileDueToPendingClustering(df))
           .map(df -> addBootstrapBaseFileIfPresent(new HoodieFileGroupId(partitionPath, df.getFileId()), df));
     } finally {
       readLock.unlock();
@@ -875,7 +872,7 @@ public final Stream<FileSlice> getLatestFileSlicesStateless(String partitionStr)
       return getLatestFileSlices(partition);
     } else {
       try {
-        Stream<FileSlice> fileSliceStream = buildFileGroups(getAllFilesInPartition(partition), visibleCommitsAndCompactionTimeline, true).stream()
+        Stream<FileSlice> fileSliceStream = buildFileGroups(partition, getAllFilesInPartition(partition), visibleCommitsAndCompactionTimeline, true).stream()
             .filter(fg -> !isFileGroupReplaced(fg))
             .map(HoodieFileGroup::getLatestFileSlice)
             .filter(Option::isPresent).map(Option::get)
@@ -1090,7 +1087,7 @@ public final Stream<HoodieFileGroup> getAllFileGroupsStateless(String partitionS
       return getAllFileGroups(partition);
     } else {
       try {
-        Stream<HoodieFileGroup> fileGroupStream = buildFileGroups(getAllFilesInPartition(partition), visibleCommitsAndCompactionTimeline, true).stream()
+        Stream<HoodieFileGroup> fileGroupStream = buildFileGroups(partition, getAllFilesInPartition(partition), visibleCommitsAndCompactionTimeline, true).stream()
             .filter(fg -> !isFileGroupReplaced(fg));
         if (bootstrapIndex.useIndex()) {
           final Map<HoodieFileGroupId, BootstrapBaseFileMapping> bootstrapBaseFileMappings = getBootstrapBaseFileMappings(partition);
@@ -1430,7 +1427,7 @@ public Stream<HoodieBaseFile> fetchLatestBaseFiles(final String partitionPath) {
 
   protected Option<HoodieBaseFile> getLatestBaseFile(HoodieFileGroup fileGroup) {
     return Option
-        .fromJavaOptional(fileGroup.getAllBaseFiles().filter(df -> !isBaseFileDueToPendingCompaction(df) && !isBaseFileDueToPendingClustering(df)).findFirst());
+        .fromJavaOptional(fileGroup.getAllBaseFiles().filter(df -> !isBaseFileDueToPendingCompaction(fileGroup.getPartitionPath(), df) && !isBaseFileDueToPendingClustering(df)).findFirst());
   }
 
   /**

hudi-common/src/main/java/org/apache/hudi/common/table/view/IncrementalTimelineSyncFileSystemView.java

@@ -273,7 +273,7 @@ private void updatePartitionWriteFileGroups(Map<String, List<HoodieWriteStat>> p
                 p.getFileSizeInBytes(), false, (short) 0, 0, 0))
             .collect(Collectors.toList());
         List<HoodieFileGroup> fileGroups =
-            buildFileGroups(pathInfoList, timeline.filterCompletedAndCompactionInstants(), false);
+            buildFileGroups(partition, pathInfoList, timeline.filterCompletedAndCompactionInstants(), false);
         applyDeltaFileSlicesToPartitionView(partition, fileGroups, DeltaApplyMode.ADD);
       } else {
         LOG.warn("Skipping partition (" + partition + ") when syncing instant (" + instant + ") as it is not loaded");
@@ -382,7 +382,7 @@ private void removeFileSlicesForPartition(HoodieTimeline timeline, HoodieInstant
           .map(p -> new StoragePathInfo(new StoragePath(p), 0, false, (short) 0, 0, 0))
           .collect(Collectors.toList());
       List<HoodieFileGroup> fileGroups =
-          buildFileGroups(pathInfoList, timeline.filterCompletedAndCompactionInstants(), false);
+          buildFileGroups(partition, pathInfoList, timeline.filterCompletedAndCompactionInstants(), false);
       applyDeltaFileSlicesToPartitionView(partition, fileGroups, DeltaApplyMode.REMOVE);
     } else {
       LOG.warn("Skipping partition (" + partition + ") when syncing instant (" + instant + ") as it is not loaded");
@@ -451,7 +451,7 @@ protected void applyDeltaFileSlicesToPartitionView(String partition, List<Hoodie
 
     HoodieTimeline timeline = deltaFileGroups.stream().map(df -> df.getTimeline()).findAny().get();
     List<HoodieFileGroup> fgs =
-        buildFileGroups(viewDataFiles.values().stream(), viewLogFiles.values().stream(), timeline, true);
+        buildFileGroups(partition, viewDataFiles.values().stream(), viewLogFiles.values().stream(), timeline, true);
     storePartitionView(partition, fgs);
   }
 

hudi-common/src/test/java/org/apache/hudi/common/fs/TestFSUtils.java

@@ -46,6 +46,8 @@
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
 
 import java.io.IOException;
 import java.nio.file.Files;
@@ -204,35 +206,20 @@ public void testGetRelativePartitionPath() {
     assertThrows(IllegalArgumentException.class, () -> FSUtils.getRelativePartitionPath(basePath, nonPartitionPath));
   }
 
-  @Test
-  public void testGetRelativePartitionPathWithStoragePath() {
-    StoragePath basePath = new StoragePath("/test/apache");
-    StoragePath partitionPath = new StoragePath("/test/apache/hudi/sub");
-    assertEquals("hudi/sub", FSUtils.getRelativePartitionPath(basePath, partitionPath));
-
-    StoragePath nonPartitionPath = new StoragePath("/test/something/else");
-    assertThrows(IllegalArgumentException.class, () -> FSUtils.getRelativePartitionPath(basePath, nonPartitionPath));
-  }
-
-  @Test
-  public void testGetRelativePartitionPathSameFolder() {
-    Path basePath = new Path("/test");
-    Path partitionPath = new Path("/test");
-    assertEquals("", FSUtils.getRelativePartitionPath(basePath, partitionPath));
-  }
-
-  @Test
-  public void testGetRelativePartitionPathRepeatedFolderNameBasePath() {
-    Path basePath = new Path("/test/apache/apache");
-    Path partitionPath = new Path("/test/apache/apache/hudi");
-    assertEquals("hudi", FSUtils.getRelativePartitionPath(basePath, partitionPath));
-  }
-
-  @Test
-  public void testGetRelativePartitionPathRepeatedFolderNamePartitionPath() {
-    Path basePath = new Path("/test/apache");
-    Path partitionPath = new Path("/test/apache/apache/hudi");
-    assertEquals("apache/hudi", FSUtils.getRelativePartitionPath(basePath, partitionPath));
+  @ParameterizedTest
+  @CsvSource({
+      "/test,/test,",
+      "s3://test,s3://test,",
+      "s3://test/foo,s3://test/foo,",
+      "/test/foo,/test/foo,",
+      "/test/apache/apache,/test/apache/apache/hudi,hudi",
+      "/test/apache,/test/apache/hudi,hudi",
+      "s3://test/apache,s3://test/apache/apache/hudi,apache/hudi"})
+  public void testGetRelativePartitionPath(String basePathStr, String partitionPathStr, String expected) {
+    StoragePath basePath = new StoragePath(basePathStr);
+    StoragePath partitionPath = new StoragePath(partitionPathStr);
+    String result = FSUtils.getRelativePartitionPath(basePath, partitionPath);
+    assertEquals(expected == null ? "" : expected, result);
   }
 
   @Test