From d1d33f725ecbef51cec05be805094b6bb1733c3e Mon Sep 17 00:00:00 2001
From: Jing Chen <milantracy@gmail.com>
Date: Tue, 2 Apr 2019 03:29:25 -0700
Subject: [PATCH] [HUDI-66] FSUtils.getRelativePartitionPath does not handle
 repeated folder names

---
 .../com/uber/hoodie/common/util/FSUtils.java  | 10 ++++++-
 .../uber/hoodie/common/util/TestFSUtils.java  | 29 +++++++++++++++++++
 2 files changed, 38 insertions(+), 1 deletion(-)

diff --git a/hoodie-common/src/main/java/com/uber/hoodie/common/util/FSUtils.java b/hoodie-common/src/main/java/com/uber/hoodie/common/util/FSUtils.java
index 4b28adee1ed2..b014710298cf 100644
--- a/hoodie-common/src/main/java/com/uber/hoodie/common/util/FSUtils.java
+++ b/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);
diff --git a/hoodie-common/src/test/java/com/uber/hoodie/common/util/TestFSUtils.java b/hoodie-common/src/test/java/com/uber/hoodie/common/util/TestFSUtils.java
index fff22c7e1af6..8afcb902d1c8 100644
--- a/hoodie-common/src/test/java/com/uber/hoodie/common/util/TestFSUtils.java
+++ b/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));
+  }
 }