Added name config option for ice-rest-catalog that enables multiple catalogs share the same etcd/postgresql/etc. instance

Author: shyiko

ice-rest-catalog/src/main/java/com/altinity/ice/rest/catalog/Main.java

@@ -310,12 +310,13 @@ public Integer call() throws Exception {
     // TODO: ensure all http handlers are hooked in
     JvmMetrics.builder().register();
 
+    String catalogName = config.name();
     String catalogImpl = icebergConfig.get(CatalogProperties.CATALOG_IMPL);
     Catalog catalog;
     if (EtcdCatalog.class.getName().equals(catalogImpl)) {
-      catalog = newEctdCatalog(icebergConfig);
+      catalog = newEctdCatalog(catalogName, icebergConfig);
     } else {
-      catalog = CatalogUtil.buildIcebergCatalog("default", icebergConfig, null);
+      catalog = CatalogUtil.buildIcebergCatalog(catalogName, icebergConfig, null);
     }
 
     // Initialize and start the maintenance scheduler
@@ -369,7 +370,7 @@ private void initializeMaintenanceScheduler(Catalog catalog, Config config) {
     }
   }
 
-  private static Catalog newEctdCatalog(Map<String, String> config) {
+  private static Catalog newEctdCatalog(String name, Map<String, String> config) {
     // TODO: remove; params all verified by config
     String uri = config.getOrDefault(CatalogProperties.URI, "etcd:http://localhost:2379");
     Preconditions.checkArgument(
@@ -388,8 +389,7 @@ private static Catalog newEctdCatalog(Map<String, String> config) {
         CatalogProperties.FILE_IO_IMPL);
 
     var io = CatalogUtil.loadFileIO(ioImpl, config, null);
-    return new EtcdCatalog(
-        "default", Strings.removePrefix(uri, "etcd:"), inputWarehouseLocation, io);
+    return new EtcdCatalog(name, Strings.removePrefix(uri, "etcd:"), inputWarehouseLocation, io);
   }
 
   public static void main(String[] args) throws Exception {

ice-rest-catalog/src/main/java/com/altinity/ice/rest/catalog/internal/config/Config.java

@@ -41,6 +41,7 @@ public record Config(
     @JsonPropertyDescription("host:port (0.0.0.0:5001 by default)") String debugAddr,
     @JsonPropertyDescription("host:port, e.g. localhost:5002 (disabled by default)")
         String adminAddr,
+    @JsonPropertyDescription("Catalog name (defaults to \"default\")") String name,
     @JsonPropertyDescription("Catalog storage URI: jdbc:..., etcd:...") String uri,
     @JsonPropertyDescription("Path to warehouse, e.g. s3://..., file://...") String warehouse,
     @JsonPropertyDescription(
@@ -69,6 +70,7 @@ public Config(
       String addr,
       String debugAddr,
       String adminAddr,
+      String name,
       @JsonProperty(required = true) String uri,
       @JsonProperty(required = true) String warehouse,
       String localFileIOBaseDir,
@@ -82,6 +84,7 @@ public Config(
     this.addr = Strings.orDefault(addr, DEFAULT_ADDR);
     this.debugAddr = Strings.orDefault(debugAddr, DEFAULT_DEBUG_ADDR);
     this.adminAddr = Strings.orDefault(adminAddr, System.getenv("ICE_REST_CATALOG_ADMIN_ADDR"));
+    this.name = Strings.orDefault(name, "default");
     this.uri = uri;
     this.warehouse = warehouse;
     this.localFileIOBaseDir = localFileIOBaseDir;

ice-rest-catalog/src/main/java/com/altinity/ice/rest/catalog/internal/etcd/EtcdCatalog.java

@@ -52,6 +52,7 @@
 import org.apache.iceberg.exceptions.RuntimeIOException;
 import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.util.LocationUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -71,6 +72,8 @@ public class EtcdCatalog extends BaseMetastoreCatalog implements SupportsNamespa
 
   public EtcdCatalog(String name, String uri, String warehouseLocation, FileIO io) {
     this.catalogName = name;
+    Preconditions.checkArgument(
+        name != null && !name.isBlank() && !name.contains("/"), "Invalid catalog name");
     this.warehouseLocation = LocationUtil.stripTrailingSlash(warehouseLocation);
     var etcdClient =
         Client.builder().endpoints(uri.split(",")).keepaliveWithoutCalls(false).build();
@@ -124,8 +127,15 @@ public boolean namespaceExists(Namespace namespace) {
     return unwrap(kv.get(key, GetOption.builder().withCountOnly(true).build())).getCount() > 0;
   }
 
-  private static String namespaceKey(Namespace namespace) {
-    return NAMESPACE_PREFIX + namespaceToPath(namespace);
+  private String namespaceKey(Namespace namespace) {
+    return namespacePrefix() + namespaceToPath(namespace);
+  }
+
+  private String namespacePrefix() {
+    if ("default".equals(catalogName)) { // for backward-compatibility
+      return NAMESPACE_PREFIX;
+    }
+    return catalogName + "/" + NAMESPACE_PREFIX;
   }
 
   private static <T> T unwrapCommit(java.util.concurrent.CompletableFuture<T> x) {
@@ -195,9 +205,7 @@ private void validateTableIdentifier(TableIdentifier identifier) {
   public List<Namespace> listNamespaces(Namespace namespace) throws NoSuchNamespaceException {
     validateNamespace(namespace);
     String prefix = namespaceKey(namespace);
-    if (!namespace.isEmpty()) {
-      prefix += "/";
-    }
+    prefix = prefix.endsWith("/") ? prefix : prefix + "/";
     GetResponse res = unwrap(kv.get(byteSeq(prefix), GetOption.builder().isPrefix(true).build()));
     if (res.getKvs().isEmpty() && !namespace.isEmpty()) {
       if (!namespaceExists(namespace)) {
@@ -209,7 +217,7 @@ public List<Namespace> listNamespaces(Namespace namespace) throws NoSuchNamespac
             k ->
                 Namespace.of(
                     Strings.removePrefix(
-                            k.getKey().toString(StandardCharsets.UTF_8), NAMESPACE_PREFIX)
+                            k.getKey().toString(StandardCharsets.UTF_8), namespacePrefix())
                         .split("/")))
         .toList();
   }
@@ -330,9 +338,7 @@ protected TableOperations newTableOps(TableIdentifier tableIdentifier) {
   public List<TableIdentifier> listTables(Namespace namespace) {
     validateNamespace(namespace);
     String prefix = tableKey(namespace);
-    if (!namespace.isEmpty()) {
-      prefix += "/";
-    }
+    prefix = prefix.endsWith("/") ? prefix : prefix + "/";
     GetResponse res = unwrap(kv.get(byteSeq(prefix), GetOption.builder().isPrefix(true).build()));
     if (res.getKvs().isEmpty() && !namespace.isEmpty()) {
       if (!namespaceExists(namespace)) {
@@ -343,17 +349,24 @@ public List<TableIdentifier> listTables(Namespace namespace) {
         .map(
             k ->
                 TableIdentifier.of(
-                    Strings.removePrefix(k.getKey().toString(StandardCharsets.UTF_8), TABLE_PREFIX)
+                    Strings.removePrefix(k.getKey().toString(StandardCharsets.UTF_8), tablePrefix())
                         .split("/")))
         .toList();
   }
 
-  private static String tableKey(Namespace namespace) {
-    return TABLE_PREFIX + namespaceToPath(namespace);
+  private String tableKey(Namespace namespace) {
+    return tablePrefix() + namespaceToPath(namespace);
+  }
+
+  private String tableKey(TableIdentifier identifier) {
+    return tablePrefix() + tableIdentifierToPath(identifier);
   }
 
-  private static String tableKey(TableIdentifier identifier) {
-    return TABLE_PREFIX + tableIdentifierToPath(identifier);
+  private String tablePrefix() {
+    if ("default".equals(catalogName)) { // for backward-compatibility
+      return TABLE_PREFIX;
+    }
+    return catalogName + "/" + TABLE_PREFIX;
   }
 
   private static String tableIdentifierToPath(TableIdentifier identifier) {