Core: Make namespace separator configurable

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

@@ -35,6 +35,8 @@ private RESTCatalogProperties() {}
 
   public static final String PAGE_SIZE = "rest-page-size";
 
+  public static final String NAMESPACE_SEPARATOR = "namespace-separator";
+
   public enum SnapshotMode {
     ALL,
     REFS

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

@@ -161,6 +161,7 @@ public class RESTSessionCatalog extends BaseViewSessionCatalog
   private CloseableGroup closeables = null;
   private Set<Endpoint> endpoints;
   private Supplier<Map<String, String>> mutationHeaders = Map::of;
+  private String namespaceSeparator = null;
 
   public RESTSessionCatalog() {
     this(
@@ -263,6 +264,12 @@ public void initialize(String name, Map<String, String> unresolved) {
             mergedProps,
             RESTCatalogProperties.METRICS_REPORTING_ENABLED,
             RESTCatalogProperties.METRICS_REPORTING_ENABLED_DEFAULT);
+    this.namespaceSeparator =
+        PropertyUtil.propertyAsString(
+            mergedProps,
+            RESTCatalogProperties.NAMESPACE_SEPARATOR,
+            RESTUtil.NAMESPACE_SEPARATOR_URLENCODED_UTF_8);
+
     super.initialize(name, mergedProps);
   }
 
@@ -580,7 +587,7 @@ public List<Namespace> listNamespaces(SessionContext context, Namespace namespac
 
     Map<String, String> queryParams = Maps.newHashMap();
     if (!namespace.isEmpty()) {
-      queryParams.put("parent", RESTUtil.namespaceToQueryParam(namespace));
+      queryParams.put("parent", RESTUtil.namespaceToQueryParam(namespace, namespaceSeparator));
     }
 
     ImmutableList.Builder<Namespace> namespaces = ImmutableList.builder();

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

@@ -26,29 +26,33 @@
 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.util.PropertyUtil;
 import org.apache.iceberg.util.UUIDUtil;
 
+@SuppressWarnings("UnicodeEscape")
 public class RESTUtil {
-  private static final char NAMESPACE_SEPARATOR = '\u001f';
-  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);
+  /** The namespace separator as Unicode character */
+  private static final char NAMESPACE_SEPARATOR_AS_UNICODE = '\u001f';
+
+  /** The namespace separator as url encoded UTF-8 character */
+  static final String NAMESPACE_SEPARATOR_URLENCODED_UTF_8 = "%1F";
 
   /**
    * @deprecated since 1.11.0, will be removed in 1.12.0; use {@link
    *     RESTUtil#namespaceToQueryParam(Namespace)}} instead.
    */
-  @Deprecated public static final Joiner NAMESPACE_JOINER = Joiner.on(NAMESPACE_SEPARATOR);
+  @Deprecated
+  public static final Joiner NAMESPACE_JOINER = Joiner.on(NAMESPACE_SEPARATOR_AS_UNICODE);
 
   /**
    * @deprecated since 1.11.0, will be removed in 1.12.0; use {@link
    *     RESTUtil#namespaceFromQueryParam(String)} instead.
    */
-  @Deprecated public static final Splitter NAMESPACE_SPLITTER = Splitter.on(NAMESPACE_SEPARATOR);
+  @Deprecated
+  public static final Splitter NAMESPACE_SPLITTER = Splitter.on(NAMESPACE_SEPARATOR_AS_UNICODE);
 
   public static final String IDEMPOTENCY_KEY_HEADER = "Idempotency-Key";
 
@@ -179,8 +183,31 @@ public static String decodeString(String encoded) {
    *     separated using the unicode character '\u001f'
    */
   public static String namespaceToQueryParam(Namespace namespace) {
+    return namespaceToQueryParam(namespace, String.valueOf(NAMESPACE_SEPARATOR_AS_UNICODE));
+  }
+
+  /**
+   * This converts the given namespace to a string and separates each part in a multipart namespace
+   * using the provided unicode separator. Note that this method is different from {@link
+   * RESTUtil#encodeNamespace(Namespace)}, which uses a UTF-8 escaped separator.
+   *
+   * <p>{@link #namespaceFromQueryParam(String, String)} should be used to convert the namespace
+   * string back to a {@link Namespace} instance.
+   *
+   * @param namespace The namespace to convert
+   * @param unicodeNamespaceSeparator The unicode namespace separator to use, such as '\u002e'
+   * @return The namespace converted to a string where each part in a multipart namespace is
+   *     separated using the given unicode separator
+   */
+  public static String namespaceToQueryParam(
+      Namespace namespace, String unicodeNamespaceSeparator) {
     Preconditions.checkArgument(null != namespace, "Invalid namespace: null");
-    return RESTUtil.NAMESPACE_JOINER.join(namespace.levels());
+    Preconditions.checkArgument(
+        !Strings.isNullOrEmpty(unicodeNamespaceSeparator), "Invalid separator: null or empty");
+
+    // decode in case the separator was already encoded with UTF-8
+    String separator = URLDecoder.decode(unicodeNamespaceSeparator, StandardCharsets.UTF_8);
+    return Joiner.on(separator).join(namespace.levels());
   }
 
   /**
@@ -191,13 +218,39 @@ public static String namespaceToQueryParam(Namespace namespace) {
    * a string.
    *
    * @param namespace The namespace to convert
-   * @return The namespace instance from the given namespace string, where each multipart separator
-   *     ('\u001f') is converted to a separate namespace level
+   * @return The namespace instance from the given namespace string, where each part in a multipart
+   *     namespace is converted using the unicode separator '\u001f'
    */
   public static Namespace namespaceFromQueryParam(String namespace) {
+    return namespaceFromQueryParam(namespace, String.valueOf(NAMESPACE_SEPARATOR_AS_UNICODE));
+  }
+
+  /**
+   * This converts a namespace where each part in a multipart namespace has been separated using the
+   * provided unicode separator to its original {@link Namespace} instance.
+   *
+   * <p>{@link #namespaceToQueryParam(Namespace, String)} should be used to convert the {@link
+   * Namespace} to a string.
+   *
+   * @param namespace The namespace to convert
+   * @param unicodeNamespaceSeparator The unicode namespace separator to use, such as '\u002e'
+   * @return The namespace instance from the given namespace string, where each part in a multipart
+   *     namespace is converted using the given unicode namespace separator
+   */
+  public static Namespace namespaceFromQueryParam(
+      String namespace, String unicodeNamespaceSeparator) {
     Preconditions.checkArgument(null != namespace, "Invalid namespace: null");
-    return Namespace.of(
-        RESTUtil.NAMESPACE_SPLITTER.splitToStream(namespace).toArray(String[]::new));
+    Preconditions.checkArgument(
+        !Strings.isNullOrEmpty(unicodeNamespaceSeparator), "Invalid separator: null or empty");
+
+    // decode in case the separator was already encoded with UTF-8
+    String separator = URLDecoder.decode(unicodeNamespaceSeparator, StandardCharsets.UTF_8);
+    Splitter splitter =
+        namespace.contains(String.valueOf(NAMESPACE_SEPARATOR_AS_UNICODE))
+            ? Splitter.on(NAMESPACE_SEPARATOR_AS_UNICODE)
+            : Splitter.on(separator);
+
+    return Namespace.of(splitter.splitToStream(namespace).toArray(String[]::new));
   }
 
   /**
@@ -210,17 +263,40 @@ public static Namespace namespaceFromQueryParam(String namespace) {
    *
    * @param ns namespace to encode
    * @return UTF-8 encoded string representing the namespace, suitable for use as a URL parameter
+   * @deprecated since 1.11.0, will be removed in 1.12.0; use {@link
+   *     RESTUtil#encodeNamespace(Namespace, String)} instead.
    */
+  @Deprecated
   public static String encodeNamespace(Namespace ns) {
-    Preconditions.checkArgument(ns != null, "Invalid namespace: null");
-    String[] levels = ns.levels();
+    return encodeNamespace(ns, NAMESPACE_SEPARATOR_URLENCODED_UTF_8);
+  }
+
+  /**
+   * 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 be used for encoding. The separator will be used
+   *     as-is and won't be UTF-8 encoded.
+   * @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);
   }
 
   /**
@@ -231,10 +307,38 @@ public static String encodeNamespace(Namespace ns) {
    *
    * @param encodedNs a namespace to decode
    * @return a namespace
+   * @deprecated since 1.11.0, will be removed in 1.12.0; use {@link
+   *     RESTUtil#decodeNamespace(String, String)} instead.
    */
+  @Deprecated
   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_SEPARATOR_URLENCODED_UTF_8);
+  }
+
+  /**
+   * 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 be used as-is 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 =
+        Splitter.on(
+            encodedNamespace.contains(NAMESPACE_SEPARATOR_URLENCODED_UTF_8)
+                ? NAMESPACE_SEPARATOR_URLENCODED_UTF_8
+                : 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,6 +22,7 @@
 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();
@@ -50,7 +51,12 @@ public class ResourcePaths {
   public static final String V1_VIEW_RENAME = "/v1/{prefix}/views/rename";
 
   public static ResourcePaths forCatalogProperties(Map<String, String> properties) {
-    return new ResourcePaths(properties.get(PREFIX));
+    return new ResourcePaths(
+        properties.get(PREFIX),
+        PropertyUtil.propertyAsString(
+            properties,
+            RESTCatalogProperties.NAMESPACE_SEPARATOR,
+            RESTUtil.NAMESPACE_SEPARATOR_URLENCODED_UTF_8));
   }
 
   public static String config() {
@@ -62,39 +68,50 @@ public static String tokens() {
   }
 
   private final String prefix;
+  private final String namespaceSeparator;
 
+  /**
+   * @deprecated since 1.11.0, will be made private in 1.12.0; use {@link
+   *     ResourcePaths#forCatalogProperties(Map)} instead.
+   */
+  @Deprecated
   public ResourcePaths(String prefix) {
+    this(prefix, RESTUtil.NAMESPACE_SEPARATOR_URLENCODED_UTF_8);
+  }
+
+  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() {
@@ -106,7 +123,7 @@ public String metrics(TableIdentifier identifier) {
         "v1",
         prefix,
         "namespaces",
-        RESTUtil.encodeNamespace(identifier.namespace()),
+        pathEncode(identifier.namespace()),
         "tables",
         RESTUtil.encodeString(identifier.name()),
         "metrics");
@@ -117,20 +134,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

@@ -81,6 +81,12 @@
 
 /** Adaptor class to translate REST requests into {@link Catalog} API calls. */
 public class RESTCatalogAdapter extends BaseHTTPClient {
+
+  @SuppressWarnings("AvoidEscapedUnicodeCharacters")
+  private static final String NAMESPACE_SEPARATOR_UNICODE = "\u002e";
+
+  private static final String NAMESPACE_SEPARATOR_URLENCODED_UTF_8 = "%2E";
+
   private static final Map<Class<? extends Exception>, Integer> EXCEPTION_ERROR_CODES =
       ImmutableMap.<Class<? extends Exception>, Integer>builder()
           .put(IllegalArgumentException.class, 400)
@@ -168,13 +174,15 @@ public <T extends RESTResponse> T handleRequest(
                     Arrays.stream(Route.values())
                         .map(r -> Endpoint.create(r.method().name(), r.resourcePath()))
                         .collect(Collectors.toList()))
+                .withOverride(
+                    RESTCatalogProperties.NAMESPACE_SEPARATOR, NAMESPACE_SEPARATOR_URLENCODED_UTF_8)
                 .build());
 
       case LIST_NAMESPACES:
         if (asNamespaceCatalog != null) {
           Namespace ns;
           if (vars.containsKey("parent")) {
-            ns = RESTUtil.namespaceFromQueryParam(vars.get("parent"));
+            ns = RESTUtil.namespaceFromQueryParam(vars.get("parent"), NAMESPACE_SEPARATOR_UNICODE);
           } else {
             ns = Namespace.empty();
           }
@@ -645,7 +653,8 @@ 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_URLENCODED_UTF_8);
   }
 
   private static TableIdentifier tableIdentFromPathVars(Map<String, String> pathVars) {