Core: Make namespace separator configurable

core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java

@@ -116,6 +116,7 @@ public class RESTSessionCatalog extends BaseViewSessionCatalog
   private static final String DEFAULT_FILE_IO_IMPL = "org.apache.iceberg.io.ResolvingFileIO";
   private static final String REST_METRICS_REPORTING_ENABLED = "rest-metrics-reporting-enabled";
   private static final String REST_SNAPSHOT_LOADING_MODE = "snapshot-loading-mode";
+  static final String NAMESPACE_SEPARATOR = "namespace-separator";
   public static final String REST_PAGE_SIZE = "rest-page-size";
   private static final List<String> TOKEN_PREFERENCE_ORDER =
       ImmutableList.of(
@@ -148,6 +149,7 @@ public class RESTSessionCatalog extends BaseViewSessionCatalog
   private boolean reportingViaRestEnabled;
   private Integer pageSize = null;
   private CloseableGroup closeables = null;
+  private String namespaceSeparator = null;
 
   // a lazy thread pool for token refresh
   private volatile ScheduledExecutorService refreshExecutor = null;
@@ -284,6 +286,9 @@ public void initialize(String name, Map<String, String> unresolved) {
 
     this.reportingViaRestEnabled =
         PropertyUtil.propertyAsBoolean(mergedProps, REST_METRICS_REPORTING_ENABLED, true);
+    this.namespaceSeparator =
+        PropertyUtil.propertyAsString(
+            mergedProps, NAMESPACE_SEPARATOR, RESTUtil.NAMESPACE_ESCAPED_SEPARATOR);
     super.initialize(name, mergedProps);
   }
 
@@ -547,7 +552,7 @@ public void createNamespace(
   public List<Namespace> listNamespaces(SessionContext context, Namespace namespace) {
     Map<String, String> queryParams = Maps.newHashMap();
     if (!namespace.isEmpty()) {
-      queryParams.put("parent", RESTUtil.encodeNamespace(namespace));
+      queryParams.put("parent", RESTUtil.encodeNamespace(namespace, namespaceSeparator));
     }
 
     ImmutableList.Builder<Namespace> namespaces = ImmutableList.builder();

core/src/main/java/org/apache/iceberg/rest/RESTUtil.java

@@ -28,15 +28,13 @@
 import org.apache.iceberg.relocated.com.google.common.base.Joiner;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.base.Splitter;
+import org.apache.iceberg.relocated.com.google.common.base.Strings;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 
 public class RESTUtil {
-  private static final String NAMESPACE_ESCAPED_SEPARATOR = "%1F";
-  private static final Joiner NAMESPACE_ESCAPED_JOINER = Joiner.on(NAMESPACE_ESCAPED_SEPARATOR);
-  private static final Splitter NAMESPACE_ESCAPED_SPLITTER =
-      Splitter.on(NAMESPACE_ESCAPED_SEPARATOR);
+  static final String NAMESPACE_ESCAPED_SEPARATOR = "%1F";
 
   /**
    * @deprecated since 1.7.0, will be made private in 1.8.0; use {@link
@@ -194,15 +192,34 @@ public static String decodeString(String encoded) {
    * @return UTF-8 encoded string representing the namespace, suitable for use as a URL parameter
    */
   public static String encodeNamespace(Namespace ns) {
-    Preconditions.checkArgument(ns != null, "Invalid namespace: null");
-    String[] levels = ns.levels();
+    return encodeNamespace(ns, NAMESPACE_ESCAPED_SEPARATOR);
+  }
+
+  /**
+   * Returns a String representation of a namespace that is suitable for use in a URL / URI.
+   *
+   * <p>This function needs to be called when a namespace is used as a path variable (or query
+   * parameter etc.), to format the namespace per the spec.
+   *
+   * <p>{@link RESTUtil#decodeNamespace(String, String)} should be used to parse the namespace from
+   * a URL parameter.
+   *
+   * @param namespace namespace to encode
+   * @param separator The namespace separator to use for encoding
+   * @return UTF-8 encoded string representing the namespace, suitable for use as a URL parameter
+   */
+  public static String encodeNamespace(Namespace namespace, String separator) {
+    Preconditions.checkArgument(namespace != null, "Invalid namespace: null");
+    Preconditions.checkArgument(
+        !Strings.isNullOrEmpty(separator), "Invalid separator: null or empty");
+    String[] levels = namespace.levels();
     String[] encodedLevels = new String[levels.length];
 
     for (int i = 0; i < levels.length; i++) {
       encodedLevels[i] = encodeString(levels[i]);
     }
 
-    return NAMESPACE_ESCAPED_JOINER.join(encodedLevels);
+    return Joiner.on(separator).join(encodedLevels);
   }
 
   /**
@@ -215,8 +232,32 @@ public static String encodeNamespace(Namespace ns) {
    * @return a namespace
    */
   public static Namespace decodeNamespace(String encodedNs) {
-    Preconditions.checkArgument(encodedNs != null, "Invalid namespace: null");
-    String[] levels = Iterables.toArray(NAMESPACE_ESCAPED_SPLITTER.split(encodedNs), String.class);
+    return decodeNamespace(encodedNs, NAMESPACE_ESCAPED_SEPARATOR);
+  }
+
+  /**
+   * Takes in a string representation of a namespace as used for a URL parameter and returns the
+   * corresponding namespace.
+   *
+   * <p>See also {@link #encodeNamespace} for generating correctly formatted URLs.
+   *
+   * @param encodedNamespace a namespace to decode
+   * @param separator The namespace separator to use for decoding. This should be the same separator
+   *     that was used when calling {@link RESTUtil#encodeNamespace(Namespace, String)}
+   * @return a namespace
+   */
+  public static Namespace decodeNamespace(String encodedNamespace, String separator) {
+    Preconditions.checkArgument(encodedNamespace != null, "Invalid namespace: null");
+    Preconditions.checkArgument(
+        !Strings.isNullOrEmpty(separator), "Invalid separator: null or empty");
+
+    // use legacy splitter for backwards compatibility in case an old clients encoded the namespace
+    // with %1F
+    Splitter splitter =
+        encodedNamespace.contains(NAMESPACE_ESCAPED_SEPARATOR)
+            ? NAMESPACE_SPLITTER
+            : Splitter.on(separator);
+    String[] levels = Iterables.toArray(splitter.split(encodedNamespace), String.class);
 
     // Decode levels in place
     for (int i = 0; i < levels.length; i++) {

core/src/main/java/org/apache/iceberg/rest/ResourcePaths.java

@@ -22,13 +22,19 @@
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.relocated.com.google.common.base.Joiner;
+import org.apache.iceberg.util.PropertyUtil;
 
 public class ResourcePaths {
   private static final Joiner SLASH = Joiner.on("/").skipNulls();
   private static final String PREFIX = "prefix";
 
   public static ResourcePaths forCatalogProperties(Map<String, String> properties) {
-    return new ResourcePaths(properties.get(PREFIX));
+    return new ResourcePaths(
+        properties.get(PREFIX),
+        PropertyUtil.propertyAsString(
+            properties,
+            RESTSessionCatalog.NAMESPACE_SEPARATOR,
+            RESTUtil.NAMESPACE_ESCAPED_SEPARATOR));
   }
 
   public static String config() {
@@ -40,39 +46,50 @@ public static String tokens() {
   }
 
   private final String prefix;
+  private final String namespaceSeparator;
 
+  /**
+   * @deprecated since 1.7.0, will be made private in 1.8.0; use {@link
+   *     ResourcePaths#forCatalogProperties(Map)} instead.
+   */
+  @Deprecated
   public ResourcePaths(String prefix) {
+    this(prefix, RESTUtil.NAMESPACE_ESCAPED_SEPARATOR);
+  }
+
+  private ResourcePaths(String prefix, String namespaceSeparator) {
     this.prefix = prefix;
+    this.namespaceSeparator = namespaceSeparator;
   }
 
   public String namespaces() {
     return SLASH.join("v1", prefix, "namespaces");
   }
 
   public String namespace(Namespace ns) {
-    return SLASH.join("v1", prefix, "namespaces", RESTUtil.encodeNamespace(ns));
+    return SLASH.join("v1", prefix, "namespaces", pathEncode(ns));
   }
 
   public String namespaceProperties(Namespace ns) {
-    return SLASH.join("v1", prefix, "namespaces", RESTUtil.encodeNamespace(ns), "properties");
+    return SLASH.join("v1", prefix, "namespaces", pathEncode(ns), "properties");
   }
 
   public String tables(Namespace ns) {
-    return SLASH.join("v1", prefix, "namespaces", RESTUtil.encodeNamespace(ns), "tables");
+    return SLASH.join("v1", prefix, "namespaces", pathEncode(ns), "tables");
   }
 
   public String table(TableIdentifier ident) {
     return SLASH.join(
         "v1",
         prefix,
         "namespaces",
-        RESTUtil.encodeNamespace(ident.namespace()),
+        pathEncode(ident.namespace()),
         "tables",
         RESTUtil.encodeString(ident.name()));
   }
 
   public String register(Namespace ns) {
-    return SLASH.join("v1", prefix, "namespaces", RESTUtil.encodeNamespace(ns), "register");
+    return SLASH.join("v1", prefix, "namespaces", pathEncode(ns), "register");
   }
 
   public String rename() {
@@ -84,7 +101,7 @@ public String metrics(TableIdentifier identifier) {
         "v1",
         prefix,
         "namespaces",
-        RESTUtil.encodeNamespace(identifier.namespace()),
+        pathEncode(identifier.namespace()),
         "tables",
         RESTUtil.encodeString(identifier.name()),
         "metrics");
@@ -95,20 +112,24 @@ public String commitTransaction() {
   }
 
   public String views(Namespace ns) {
-    return SLASH.join("v1", prefix, "namespaces", RESTUtil.encodeNamespace(ns), "views");
+    return SLASH.join("v1", prefix, "namespaces", pathEncode(ns), "views");
   }
 
   public String view(TableIdentifier ident) {
     return SLASH.join(
         "v1",
         prefix,
         "namespaces",
-        RESTUtil.encodeNamespace(ident.namespace()),
+        pathEncode(ident.namespace()),
         "views",
         RESTUtil.encodeString(ident.name()));
   }
 
   public String renameView() {
     return SLASH.join("v1", prefix, "views", "rename");
   }
+
+  private String pathEncode(Namespace ns) {
+    return RESTUtil.encodeNamespace(ns, namespaceSeparator);
+  }
 }

core/src/test/java/org/apache/iceberg/rest/RESTCatalogAdapter.java

@@ -73,6 +73,7 @@
 /** Adaptor class to translate REST requests into {@link Catalog} API calls. */
 public class RESTCatalogAdapter implements RESTClient {
   private static final Splitter SLASH = Splitter.on('/');
+  private static final String NAMESPACE_SEPARATOR = "%2E";
 
   private static final Map<Class<? extends Exception>, Integer> EXCEPTION_ERROR_CODES =
       ImmutableMap.<Class<? extends Exception>, Integer>builder()
@@ -282,7 +283,11 @@ public <T extends RESTResponse> T handleRequest(
         return castResponse(responseType, handleOAuthRequest(body));
 
       case CONFIG:
-        return castResponse(responseType, ConfigResponse.builder().build());
+        return castResponse(
+            responseType,
+            ConfigResponse.builder()
+                .withOverride(RESTSessionCatalog.NAMESPACE_SEPARATOR, NAMESPACE_SEPARATOR)
+                .build());
 
       case LIST_NAMESPACES:
         if (asNamespaceCatalog != null) {
@@ -665,7 +670,7 @@ public static void configureResponseFromException(
   }
 
   private static Namespace namespaceFromPathVars(Map<String, String> pathVars) {
-    return RESTUtil.decodeNamespace(pathVars.get("namespace"));
+    return RESTUtil.decodeNamespace(pathVars.get("namespace"), NAMESPACE_SEPARATOR);
   }
 
   private static TableIdentifier identFromPathVars(Map<String, String> pathVars) {

core/src/test/java/org/apache/iceberg/rest/TestRESTUtil.java

@@ -20,11 +20,14 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import java.util.Map;
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 
 public class TestRESTUtil {
 
@@ -67,18 +70,24 @@ public void testStripTrailingSlash() {
     }
   }
 
-  @Test
-  public void testRoundTripUrlEncodeDecodeNamespace() {
+  @ParameterizedTest
+  @ValueSource(strings = {"%1F", "%2D", "%2E", "#", "_"})
+  public void testRoundTripUrlEncodeDecodeNamespace(String namespaceSeparator) {
     // Namespace levels and their expected url encoded form
     Object[][] testCases =
         new Object[][] {
           new Object[] {new String[] {"dogs"}, "dogs"},
           new Object[] {new String[] {"dogs.named.hank"}, "dogs.named.hank"},
           new Object[] {new String[] {"dogs/named/hank"}, "dogs%2Fnamed%2Fhank"},
-          new Object[] {new String[] {"dogs", "named", "hank"}, "dogs%1Fnamed%1Fhank"},
+          new Object[] {
+            new String[] {"dogs", "named", "hank"},
+            String.format("dogs%snamed%shank", namespaceSeparator, namespaceSeparator)
+          },
           new Object[] {
             new String[] {"dogs.and.cats", "named", "hank.or.james-westfall"},
-            "dogs.and.cats%1Fnamed%1Fhank.or.james-westfall"
+            String.format(
+                "dogs.and.cats%snamed%shank.or.james-westfall",
+                namespaceSeparator, namespaceSeparator),
           }
         };
 
@@ -89,14 +98,25 @@ public void testRoundTripUrlEncodeDecodeNamespace() {
       Namespace namespace = Namespace.of(levels);
 
       // To be placed into a URL path as query parameter or path parameter
-      assertThat(RESTUtil.encodeNamespace(namespace)).isEqualTo(encodedNs);
+      assertThat(RESTUtil.encodeNamespace(namespace, namespaceSeparator)).isEqualTo(encodedNs);
 
       // Decoded (after pulling as String) from URL
-      Namespace asNamespace = RESTUtil.decodeNamespace(encodedNs);
-      assertThat(asNamespace).isEqualTo(namespace);
+      assertThat(RESTUtil.decodeNamespace(encodedNs, namespaceSeparator)).isEqualTo(namespace);
     }
   }
 
+  @Test
+  public void encodeAsOldClientAndDecodeAsNewServer() {
+    Namespace namespace = Namespace.of("first", "second", "third");
+    // old client would call encodeNamespace without specifying a separator
+    String encodedNamespace = RESTUtil.encodeNamespace(namespace);
+    assertThat(encodedNamespace).contains(RESTUtil.NAMESPACE_ESCAPED_SEPARATOR);
+
+    // newer server would try and decode the namespace with the separator it communicates to clients
+    Namespace decodeNamespace = RESTUtil.decodeNamespace(encodedNamespace, "%2E");
+    assertThat(decodeNamespace).isEqualTo(namespace);
+  }
+
   @Test
   public void testNamespaceUrlEncodeDecodeDoesNotAllowNull() {
     assertThatExceptionOfType(IllegalArgumentException.class)
@@ -139,4 +159,24 @@ public void testOAuth2FormDataDecoding() {
 
     assertThat(RESTUtil.decodeFormData(formString)).isEqualTo(expected);
   }
+
+  @Test
+  public void nullOrEmptyNamespaceSeparator() {
+    String errorMsg = "Invalid separator: null or empty";
+    assertThatThrownBy(() -> RESTUtil.encodeNamespace(Namespace.empty(), null))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage(errorMsg);
+
+    assertThatThrownBy(() -> RESTUtil.encodeNamespace(Namespace.empty(), ""))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage(errorMsg);
+
+    assertThatThrownBy(() -> RESTUtil.decodeNamespace("namespace", null))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage(errorMsg);
+
+    assertThatThrownBy(() -> RESTUtil.decodeNamespace("namespace", ""))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage(errorMsg);
+  }
 }