Support registering catalog dynamically with RESTful API

core/trino-main/src/main/java/io/trino/metadata/CatalogResource.java

@@ -0,0 +1,82 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.trino.metadata;
+
+import io.airlift.log.Logger;
+import io.trino.server.security.ResourceSecurity;
+
+import javax.inject.Inject;
+import javax.ws.rs.Consumes;
+import javax.ws.rs.GET;
+import javax.ws.rs.POST;
+import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
+import javax.ws.rs.QueryParam;
+import javax.ws.rs.core.MediaType;
+
+import java.util.List;
+import java.util.Map;
+
+import static io.trino.server.security.ResourceSecurity.AccessType.PUBLIC;
+
+@Path("/v1/catalog")
+public class CatalogResource
+{
+    private static final Logger log = Logger.get(CatalogResource.class);
+    private final DynamicCatalogStore dynamicCatalogStore;
+
+    @Inject
+    public CatalogResource(DynamicCatalogStore dynamicCatalogStore)
+    {
+        this.dynamicCatalogStore = dynamicCatalogStore;
+    }
+
+    @ResourceSecurity(PUBLIC)
+    @Path("/catalogs")
+    @GET
+    @Produces(MediaType.APPLICATION_JSON)
+    public List<String> getCatalogs()
+    {
+        return dynamicCatalogStore.getCatalogs();
+    }
+
+    @ResourceSecurity(PUBLIC)
+    @Path("/register")
+    @POST
+    @Consumes(MediaType.APPLICATION_JSON)
+    public void registerCatalog(@QueryParam("name") String catalogName, Map<String, String> catalogProperties)
+    {
+        log.info(catalogProperties.toString());
+        dynamicCatalogStore.registerCatalog(catalogName, catalogProperties);
+    }
+
+    @ResourceSecurity(PUBLIC)
+    @Path("/remove")
+    @POST
+    @Consumes(MediaType.APPLICATION_JSON)
+    public void removeCatalog(@QueryParam("name") String catalogName)
+    {
+        dynamicCatalogStore.removeCatalog(catalogName);
+    }
+
+    @ResourceSecurity(PUBLIC)
+    @Path("/update")
+    @POST
+    @Consumes(MediaType.APPLICATION_JSON)
+    public void updateCatalog(@QueryParam("name") String catalogName, Map<String, String> catalogProperties)
+    {
+        log.info(catalogProperties.toString());
+        dynamicCatalogStore.updateCatalog(catalogName, catalogProperties);
+    }
+}

core/trino-main/src/main/java/io/trino/metadata/DynamicCatalogStore.java

@@ -0,0 +1,127 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.trino.metadata;
+
+import com.google.common.base.Joiner;
+import com.google.common.base.Splitter;
+import com.google.common.collect.ImmutableMap;
+import io.airlift.discovery.client.Announcer;
+import io.airlift.discovery.client.ServiceAnnouncement;
+import io.airlift.log.Logger;
+import io.trino.connector.ConnectorManager;
+
+import javax.inject.Inject;
+
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static com.google.common.base.Preconditions.checkState;
+import static io.airlift.discovery.client.ServiceAnnouncement.serviceAnnouncement;
+
+public class DynamicCatalogStore
+{
+    private static final Logger log = Logger.get(DynamicCatalogStore.class);
+    private final ConnectorManager connectorManager;
+    private final CatalogManager catalogManager;
+    private final Announcer announcer;
+
+    @Inject
+    public DynamicCatalogStore(ConnectorManager connectorManager, CatalogManager catalogManager, Announcer announcer)
+    {
+        this.connectorManager = connectorManager;
+        this.catalogManager = catalogManager;
+        this.announcer = announcer;
+    }
+
+    public List<String> getCatalogs()
+    {
+        List<Catalog> catalogs = catalogManager.getCatalogs();
+        return catalogs.stream().map(catalog -> catalog.getCatalogName()).collect(Collectors.toList());
+    }
+
+    public void registerCatalog(String catalogName, Map<String, String> properties)
+    {
+        log.info("-- Loading catalog %s --", catalogName);
+        String connectorName = properties.remove("connector.name");
+        checkState(connectorName != null, "Catalog configuration does not contain connector.name");
+        connectorManager.createCatalog(catalogName, connectorName, ImmutableMap.copyOf(properties));
+        updateConnectorIds(announcer, catalogName, CatalogAction.ADD);
+        log.info("-- Added catalog %s using connector %s --", catalogName, connectorName);
+    }
+
+    public void removeCatalog(String catalogName)
+    {
+        log.info("-- Removing catalog %s --", catalogName);
+        connectorManager.dropConnection(catalogName);
+        updateConnectorIds(announcer, catalogName, CatalogAction.DELETE);
+        log.info("-- Removed catalog %s --", catalogName);
+    }
+
+    public void updateCatalog(String catalogName, Map<String, String> properties)
+    {
+        log.info("-- Updating catalog %s --", catalogName);
+        removeCatalog(catalogName);
+        registerCatalog(catalogName, properties);
+        log.info("-- Updated catalog %s --", catalogName);
+    }
+
+    private void updateConnectorIds(Announcer announcer, String catalogName, CatalogAction action)
+    {
+        // get existing announcement
+        ServiceAnnouncement announcement = getTrinoAnnouncement(announcer.getServiceAnnouncements());
+        String property = announcement.getProperties().get("connectorIds");
+
+        List<String> values = Splitter.on(',').trimResults().omitEmptyStrings().splitToList(property);
+        Set<String> connectorIds = new LinkedHashSet<>(values);
+        switch (action) {
+            case ADD:
+                connectorIds.add(catalogName);
+                break;
+            case DELETE:
+                connectorIds.remove(catalogName);
+                break;
+        }
+
+        // build announcement with updated sources
+        ServiceAnnouncement.ServiceAnnouncementBuilder builder = serviceAnnouncement(announcement.getType());
+        for (Map.Entry<String, String> entry : announcement.getProperties().entrySet()) {
+            if (!entry.getKey().equals("connectorIds")) {
+                builder.addProperty(entry.getKey(), entry.getValue());
+            }
+        }
+        builder.addProperty("connectorIds", Joiner.on(',').join(connectorIds));
+
+        // update announcement
+        announcer.removeServiceAnnouncement(announcement.getId());
+        announcer.addServiceAnnouncement(builder.build());
+    }
+
+    private ServiceAnnouncement getTrinoAnnouncement(Set<ServiceAnnouncement> announcements)
+    {
+        for (ServiceAnnouncement announcement : announcements) {
+            if (announcement.getType().equals("trino")) {
+                return announcement;
+            }
+        }
+        throw new IllegalArgumentException("Trino announcement not found: " + announcements);
+    }
+
+    public enum CatalogAction
+    {
+        ADD, DELETE
+    }
+}

core/trino-main/src/main/java/io/trino/server/ServerMainModule.java

@@ -64,9 +64,11 @@
 import io.trino.metadata.AnalyzePropertyManager;
 import io.trino.metadata.BlockEncodingManager;
 import io.trino.metadata.CatalogManager;
+import io.trino.metadata.CatalogResource;
 import io.trino.metadata.ColumnPropertyManager;
 import io.trino.metadata.DisabledSystemSecurityMetadata;
 import io.trino.metadata.DiscoveryNodeManager;
+import io.trino.metadata.DynamicCatalogStore;
 import io.trino.metadata.ForNodeManager;
 import io.trino.metadata.FunctionBundle;
 import io.trino.metadata.FunctionManager;
@@ -382,6 +384,7 @@ protected void setup(Binder binder)
 
         // metadata
         binder.bind(StaticCatalogStore.class).in(Scopes.SINGLETON);
+        binder.bind(DynamicCatalogStore.class).in(Scopes.SINGLETON);
         configBinder(binder).bindConfig(StaticCatalogStoreConfig.class);
         binder.bind(MetadataManager.class).in(Scopes.SINGLETON);
         binder.bind(Metadata.class).to(MetadataManager.class).in(Scopes.SINGLETON);
@@ -398,6 +401,7 @@ protected void setup(Binder binder)
         binder.bind(TableProceduresRegistry.class).in(Scopes.SINGLETON);
         binder.bind(TableFunctionRegistry.class).in(Scopes.SINGLETON);
         binder.bind(PlannerContext.class).in(Scopes.SINGLETON);
+        jaxrsBinder(binder).bind(CatalogResource.class);
 
         // function
         binder.bind(FunctionManager.class).in(Scopes.SINGLETON);