[PAUSED - DO NOT REVIEW] Allow for quotes in path for blob storage

polaris-core/src/main/java/org/apache/polaris/core/storage/StorageLocation.java

@@ -20,12 +20,15 @@
 
 import jakarta.annotation.Nonnull;
 import java.net.URI;
+import java.util.Locale;
+import java.util.regex.Pattern;
 import org.apache.polaris.core.storage.azure.AzureLocation;
 
 /** An abstraction over a storage location */
 public class StorageLocation {
   private final String location;
   public static final String LOCAL_PATH_PREFIX = "file:///";
+  public static final Pattern SLASHES = Pattern.compile("/+");
 
   /** Create a StorageLocation from a String path */
   public static StorageLocation of(String location) {
@@ -40,12 +43,21 @@ public static StorageLocation of(String location) {
   protected StorageLocation(@Nonnull String location) {
     if (location == null) {
       this.location = null;
-    } else if (location.startsWith("file:/") && !location.startsWith(LOCAL_PATH_PREFIX)) {
-      this.location = URI.create(location.replaceFirst("file:/+", LOCAL_PATH_PREFIX)).toString();
-    } else if (location.startsWith("/")) {
-      this.location = URI.create(location.replaceFirst("/+", LOCAL_PATH_PREFIX)).toString();
+      return;
+    }
+    String scheme = scheme(location);
+    String path;
+    if (scheme == null) {
+      path = location;
+      scheme = "file";
+    } else {
+      path = location.substring(scheme.length() + 1);
+    }
+    path = normalizePath(path);
+    if (scheme.equals("file")) {
+      this.location = URI.create(LOCAL_PATH_PREFIX + path).toString();
     } else {
-      this.location = URI.create(location).toString();
+      this.location = scheme + "://" + path;
     }
   }
 
@@ -99,4 +111,23 @@ public boolean isChildOf(StorageLocation potentialParent) {
       return slashTerminatedLocation.startsWith(slashTerminatedParentLocation);
     }
   }
+
+  private static String scheme(String location) {
+    int schemePos = location.indexOf(':');
+    if (schemePos > 0) {
+      return location.substring(0, schemePos).toLowerCase(Locale.ROOT);
+    }
+    return null;
+  }
+
+  protected String normalizePath(String path) {
+    path = SLASHES.matcher(path).replaceAll("/");
+    if (path.endsWith("/") || path.startsWith("/")) {
+      path =
+          path.substring(
+              path.startsWith("/") && path.length() > 1 ? 1 : 0,
+              path.endsWith("/") ? path.length() - 1 : path.length());
+    }
+    return path;
+  }
 }

polaris-core/src/test/java/org/apache/polaris/service/storage/StorageLocationTest.java

@@ -19,17 +19,41 @@
 package org.apache.polaris.service.storage;
 
 import org.apache.polaris.core.storage.StorageLocation;
-import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
 public class StorageLocationTest {
 
   @Test
-  public void testOfDifferentPrefixes() {
-    StorageLocation StandardLocation = StorageLocation.of("file:///path/to/file");
+  public void testAwsLocations() {
+    StorageLocation baseStorageLocation = StorageLocation.of("s3://path/to/file");
+    StorageLocation testStorageLocation = StorageLocation.of("s3://path/to/////file");
+    Assertions.assertEquals(testStorageLocation, baseStorageLocation);
+    testStorageLocation = StorageLocation.of("s3://////path/////to///file");
+    Assertions.assertEquals(testStorageLocation, baseStorageLocation);
+
+    testStorageLocation = StorageLocation.of("s3://////path'/////*to===///\"file");
+    Assertions.assertEquals("s3://path'/*to===/\"file", testStorageLocation.toString());
+  }
+
+  @Test
+  public void testGcpLocations() {
+    StorageLocation baseStorageLocation = StorageLocation.of("gcs://path/to/file");
+    StorageLocation testStorageLocation = StorageLocation.of("gcs://path/to/////file");
+    Assertions.assertEquals(testStorageLocation, baseStorageLocation);
+    testStorageLocation = StorageLocation.of("gcs://////path/////to///file");
+    Assertions.assertEquals(testStorageLocation, baseStorageLocation);
+
+    testStorageLocation = StorageLocation.of("gcs://////path'/////*to===///\"file");
+    Assertions.assertEquals("gcs://path'/*to===/\"file", testStorageLocation.toString());
+  }
+
+  @Test
+  public void testLocalFileLocations() {
+    StorageLocation storageLocation = StorageLocation.of("file:///path/to/file");
     StorageLocation slashLeadingLocation = StorageLocation.of("/path/to/file");
     StorageLocation fileSingleSlashLocation = StorageLocation.of("file:/path/to/file");
-    Assertions.assertThat(slashLeadingLocation.equals(StandardLocation)).isTrue();
-    Assertions.assertThat(fileSingleSlashLocation.equals(StandardLocation)).isTrue();
+    Assertions.assertEquals(slashLeadingLocation, storageLocation);
+    Assertions.assertEquals(fileSingleSlashLocation, storageLocation);
   }
 }