[HUDI-66] FSUtils.getRelativePartitionPath does not handle repeated f…

…older names

hoodie-common/src/main/java/com/uber/hoodie/common/util/FSUtils.java

@@ -153,9 +153,17 @@ public static List<String> getAllFoldersThreeLevelsDown(FileSystem fs, String ba
     return datePartitions;
   }
 
+  /**
+   * Given a base partition and a partition path, return
+   * relative path of partition path to the base path
+   */
   public static String getRelativePartitionPath(Path basePath, Path partitionPath) {
+    basePath = Path.getPathWithoutSchemeAndAuthority(basePath);
+    partitionPath = Path.getPathWithoutSchemeAndAuthority(partitionPath);
     String partitionFullPath = partitionPath.toString();
-    int partitionStartIndex = partitionFullPath.lastIndexOf(basePath.getName());
+    int partitionStartIndex = partitionFullPath.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() == partitionFullPath.length() ? "" :
         partitionFullPath.substring(partitionStartIndex + basePath.getName().length() + 1);

hoodie-common/src/test/java/com/uber/hoodie/common/util/TestFSUtils.java

@@ -24,6 +24,7 @@
 import java.util.Date;
 import java.util.UUID;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.contrib.java.lang.system.EnvironmentVariables;
@@ -92,4 +93,32 @@ public void testEnvVarVariablesPickedup() {
     assertEquals("value11", conf.get("fs.key1"));
     assertEquals("value2", conf.get("fs.key2"));
   }
+
+  @Test
+  public void testGetRelativePartitionPath() {
+    Path basePath = new Path("/test/apache");
+    Path partitionPath = new Path("/test/apache/hudi/sub");
+    assertEquals("hudi/sub",FSUtils.getRelativePartitionPath(basePath, partitionPath));
+  }
+
+  @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));
+  }
 }