Add ConnectorId to enforce separation from catalog name

presto-hive/src/test/java/com/facebook/presto/hive/TestOrcPageSourceMemoryTracking.java

@@ -13,6 +13,7 @@
  */
 package com.facebook.presto.hive;
 
+import com.facebook.presto.connector.ConnectorId;
 import com.facebook.presto.hive.orc.OrcPageSourceFactory;
 import com.facebook.presto.metadata.Split;
 import com.facebook.presto.operator.DriverContext;
@@ -405,7 +406,7 @@ public SourceOperator newTableScanOperator(DriverContext driverContext)
                     columns.stream().map(columnHandle -> (ColumnHandle) columnHandle).collect(toList())
             );
             SourceOperator operator = sourceOperatorFactory.createOperator(driverContext);
-            operator.addSplit(new Split("test", TestingTransactionHandle.create("test"), TestingSplit.createLocalSplit()));
+            operator.addSplit(new Split(new ConnectorId("test"), TestingTransactionHandle.create(), TestingSplit.createLocalSplit()));
             return operator;
         }
 
@@ -428,7 +429,7 @@ public SourceOperator newScanFilterAndProjectOperator(DriverContext driverContex
                     types
             );
             SourceOperator operator = sourceOperatorFactory.createOperator(driverContext);
-            operator.addSplit(new Split("test", TestingTransactionHandle.create("test"), TestingSplit.createLocalSplit()));
+            operator.addSplit(new Split(new ConnectorId("test"), TestingTransactionHandle.create(), TestingSplit.createLocalSplit()));
             return operator;
         }
 

presto-main/src/main/java/com/facebook/presto/Session.java

@@ -14,6 +14,7 @@
 package com.facebook.presto;
 
 import com.facebook.presto.client.ClientSession;
+import com.facebook.presto.connector.ConnectorId;
 import com.facebook.presto.metadata.SessionPropertyManager;
 import com.facebook.presto.spi.ConnectorSession;
 import com.facebook.presto.spi.QueryId;
@@ -335,17 +336,17 @@ public ConnectorSession toConnectorSession()
         return new FullConnectorSession(queryId.toString(), identity, timeZoneKey, locale, startTime);
     }
 
-    public ConnectorSession toConnectorSession(String catalog)
+    public ConnectorSession toConnectorSession(ConnectorId connectorId)
     {
-        requireNonNull(catalog, "catalog is null");
+        requireNonNull(connectorId, "connectorId is null");
         return new FullConnectorSession(
                 queryId.toString(),
                 identity,
                 timeZoneKey,
                 locale,
                 startTime,
-                catalogProperties.getOrDefault(catalog, ImmutableMap.of()),
-                catalog,
+                catalogProperties.getOrDefault(connectorId.getCatalogName(), ImmutableMap.of()),
+                connectorId.getCatalogName(),
                 sessionPropertyManager);
     }
 

presto-main/src/main/java/com/facebook/presto/connector/ConnectorAwareNodeManager.java

@@ -27,9 +27,9 @@ public class ConnectorAwareNodeManager
 {
     private final InternalNodeManager nodeManager;
     private final String environment;
-    private final String connectorId;
+    private final ConnectorId connectorId;
 
-    public ConnectorAwareNodeManager(InternalNodeManager nodeManager, String environment, String connectorId)
+    public ConnectorAwareNodeManager(InternalNodeManager nodeManager, String environment, ConnectorId connectorId)
     {
         this.nodeManager = requireNonNull(nodeManager, "nodeManager is null");
         this.environment = requireNonNull(environment, "environment is null");

presto-main/src/main/java/com/facebook/presto/connector/ConnectorId.java

@@ -0,0 +1,84 @@
+/*
+ * 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 com.facebook.presto.connector;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonValue;
+
+import java.util.Objects;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static java.util.Objects.requireNonNull;
+
+public final class ConnectorId
+{
+    private static final String INFORMATION_SCHEMA_CONNECTOR_PREFIX = "$info_schema@";
+    private static final String SYSTEM_TABLES_CONNECTOR_PREFIX = "$system@";
+
+    private final String catalogName;
+
+    @JsonCreator
+    public ConnectorId(String catalogName)
+    {
+        this.catalogName = requireNonNull(catalogName, "catalogName is null");
+        checkArgument(!catalogName.isEmpty(), "catalogName is empty");
+    }
+
+    public String getCatalogName()
+    {
+        return catalogName;
+    }
+
+    @Override
+    public boolean equals(Object o)
+    {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        ConnectorId that = (ConnectorId) o;
+        return Objects.equals(catalogName, that.catalogName);
+    }
+
+    @Override
+    public int hashCode()
+    {
+        return Objects.hash(catalogName);
+    }
+
+    @JsonValue
+    @Override
+    public String toString()
+    {
+        return catalogName;
+    }
+
+    public static boolean isInternalSystemConnector(ConnectorId connectorId)
+    {
+        return connectorId.getCatalogName().startsWith(SYSTEM_TABLES_CONNECTOR_PREFIX) ||
+                connectorId.getCatalogName().startsWith(INFORMATION_SCHEMA_CONNECTOR_PREFIX);
+    }
+
+    public static ConnectorId createInformationSchemaConnectorId(ConnectorId connectorId)
+    {
+        return new ConnectorId(INFORMATION_SCHEMA_CONNECTOR_PREFIX + connectorId.getCatalogName());
+    }
+
+    public static ConnectorId createSystemTablesConnectorId(ConnectorId connectorId)
+    {
+        return new ConnectorId(SYSTEM_TABLES_CONNECTOR_PREFIX + connectorId.getCatalogName());
+    }
+}

presto-main/src/main/java/com/facebook/presto/connector/ConnectorManager.java

@@ -59,6 +59,8 @@
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import static com.facebook.presto.connector.ConnectorId.createInformationSchemaConnectorId;
+import static com.facebook.presto.connector.ConnectorId.createSystemTablesConnectorId;
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkState;
 import static com.google.common.collect.Sets.newConcurrentHashSet;
@@ -68,9 +70,6 @@
 @ThreadSafe
 public class ConnectorManager
 {
-    public static final String INFORMATION_SCHEMA_CONNECTOR_PREFIX = "$info_schema@";
-    public static final String SYSTEM_TABLES_CONNECTOR_PREFIX = "$system@";
-
     private static final Logger log = Logger.get(ConnectorManager.class);
 
     private final MetadataManager metadataManager;
@@ -95,7 +94,7 @@ public class ConnectorManager
     @GuardedBy("this")
     private final Set<String> catalogs = newConcurrentHashSet();
     @GuardedBy("this")
-    private final ConcurrentMap<String, Connector> connectors = new ConcurrentHashMap<>();
+    private final ConcurrentMap<ConnectorId, Connector> connectors = new ConcurrentHashMap<>();
 
     private final AtomicBoolean stopped = new AtomicBoolean();
 
@@ -138,7 +137,7 @@ public synchronized void stop()
             return;
         }
 
-        for (Map.Entry<String, Connector> entry : connectors.entrySet()) {
+        for (Map.Entry<ConnectorId, Connector> entry : connectors.entrySet()) {
             Connector connector = entry.getValue();
             try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(connector.getClass().getClassLoader())) {
                 connector.shutdown();
@@ -163,40 +162,42 @@ public synchronized void addConnectorFactory(ConnectorFactory connectorFactory)
         handleResolver.addConnectorName(connectorFactory.getName(), connectorFactory.getHandleResolver());
     }
 
-    public synchronized void createConnection(String catalogName, String connectorName, Map<String, String> properties)
+    public synchronized ConnectorId createConnection(String catalogName, String connectorName, Map<String, String> properties)
     {
         requireNonNull(connectorName, "connectorName is null");
         ConnectorFactory connectorFactory = connectorFactories.get(connectorName);
         checkArgument(connectorFactory != null, "No factory for connector %s", connectorName);
-        createConnection(catalogName, connectorFactory, properties);
+        return createConnection(catalogName, connectorFactory, properties);
     }
 
-    private synchronized void createConnection(String catalogName, ConnectorFactory connectorFactory, Map<String, String> properties)
+    private synchronized ConnectorId createConnection(String catalogName, ConnectorFactory connectorFactory, Map<String, String> properties)
     {
         checkState(!stopped.get(), "ConnectorManager is stopped");
         requireNonNull(catalogName, "catalogName is null");
         requireNonNull(properties, "properties is null");
         requireNonNull(connectorFactory, "connectorFactory is null");
         checkArgument(!catalogs.contains(catalogName), "A catalog already exists for %s", catalogName);
 
-        String connectorId = getConnectorId(catalogName);
+        ConnectorId connectorId = new ConnectorId(catalogName);
         checkState(!connectors.containsKey(connectorId), "A connector %s already exists", connectorId);
 
         addCatalogConnector(catalogName, connectorId, connectorFactory, properties);
 
         catalogs.add(catalogName);
+
+        return connectorId;
     }
 
-    private synchronized void addCatalogConnector(String catalogName, String connectorId, ConnectorFactory factory, Map<String, String> properties)
+    private synchronized void addCatalogConnector(String catalogName, ConnectorId connectorId, ConnectorFactory factory, Map<String, String> properties)
     {
         Connector connector = createConnector(connectorId, factory, properties);
 
         addConnectorInternal(ConnectorType.STANDARD, catalogName, connectorId, connector);
 
-        String informationSchemaId = makeInformationSchemaConnectorId(connectorId);
+        ConnectorId informationSchemaId = createInformationSchemaConnectorId(connectorId);
         addConnectorInternal(ConnectorType.INFORMATION_SCHEMA, catalogName, informationSchemaId, new InformationSchemaConnector(catalogName, nodeManager, metadataManager));
 
-        String systemId = makeSystemTablesConnectorId(connectorId);
+        ConnectorId systemId = createSystemTablesConnectorId(connectorId);
         addConnectorInternal(ConnectorType.SYSTEM, catalogName, systemId, new SystemConnector(
                 systemId,
                 nodeManager,
@@ -208,7 +209,7 @@ private synchronized void addCatalogConnector(String catalogName, String connect
         metadataManager.getTablePropertyManager().addProperties(catalogName, connector.getTableProperties());
     }
 
-    private synchronized void addConnectorInternal(ConnectorType type, String catalogName, String connectorId, Connector connector)
+    private synchronized void addConnectorInternal(ConnectorType type, String catalogName, ConnectorId connectorId, Connector connector)
     {
         checkState(!stopped.get(), "ConnectorManager is stopped");
         checkState(!connectors.containsKey(connectorId), "A connector %s already exists", connectorId);
@@ -327,7 +328,7 @@ else if (type == ConnectorType.SYSTEM) {
         }
     }
 
-    private Connector createConnector(String connectorId, ConnectorFactory factory, Map<String, String> properties)
+    private Connector createConnector(ConnectorId connectorId, ConnectorFactory factory, Map<String, String> properties)
     {
         Class<?> factoryClass = factory.getClass();
         if (factory instanceof LegacyTransactionConnectorFactory) {
@@ -341,7 +342,7 @@ private Connector createConnector(String connectorId, ConnectorFactory factory,
                 pageIndexerFactory);
 
         try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(factoryClass.getClassLoader())) {
-            return factory.create(connectorId, properties, context);
+            return factory.create(connectorId.getCatalogName(), properties, context);
         }
     }
 
@@ -351,20 +352,4 @@ private enum ConnectorType
         INFORMATION_SCHEMA,
         SYSTEM
     }
-
-    private static String makeInformationSchemaConnectorId(String connectorId)
-    {
-        return INFORMATION_SCHEMA_CONNECTOR_PREFIX + connectorId;
-    }
-
-    private static String makeSystemTablesConnectorId(String connectorId)
-    {
-        return SYSTEM_TABLES_CONNECTOR_PREFIX + connectorId;
-    }
-
-    private static String getConnectorId(String catalogName)
-    {
-        // for now connectorId == catalogName
-        return catalogName;
-    }
 }

presto-main/src/main/java/com/facebook/presto/connector/system/CatalogSystemTable.java

@@ -13,6 +13,7 @@
  */
 package com.facebook.presto.connector.system;
 
+import com.facebook.presto.connector.ConnectorId;
 import com.facebook.presto.metadata.Metadata;
 import com.facebook.presto.spi.ConnectorSession;
 import com.facebook.presto.spi.ConnectorTableMetadata;
@@ -66,8 +67,8 @@ public ConnectorTableMetadata getTableMetadata()
     public RecordCursor cursor(ConnectorTransactionHandle transactionHandle, ConnectorSession session, TupleDomain<Integer> constraint)
     {
         Builder table = InMemoryRecordSet.builder(CATALOG_TABLE);
-        for (Map.Entry<String, String> entry : metadata.getCatalogNames().entrySet()) {
-            table.addRow(entry.getKey(), entry.getValue());
+        for (Map.Entry<String, ConnectorId> entry : metadata.getCatalogNames().entrySet()) {
+            table.addRow(entry.getKey(), entry.getValue().toString());
         }
         return table.build().cursor();
     }

presto-main/src/main/java/com/facebook/presto/connector/system/SystemColumnHandle.java

@@ -13,6 +13,7 @@
  */
 package com.facebook.presto.connector.system;
 
+import com.facebook.presto.connector.ConnectorId;
 import com.facebook.presto.spi.ColumnHandle;
 import com.facebook.presto.spi.ColumnMetadata;
 import com.facebook.presto.spi.ConnectorTableMetadata;
@@ -28,20 +29,20 @@
 public class SystemColumnHandle
         implements ColumnHandle
 {
-    private final String connectorId;
+    private final ConnectorId connectorId;
     private final String columnName;
 
     @JsonCreator
     public SystemColumnHandle(
-            @JsonProperty("connectorId") String connectorId,
+            @JsonProperty("connectorId") ConnectorId connectorId,
             @JsonProperty("columnName") String columnName)
     {
         this.connectorId = requireNonNull(connectorId, "connectorId is null");
         this.columnName = requireNonNull(columnName, "columnName is null");
     }
 
     @JsonProperty
-    public String getConnectorId()
+    public ConnectorId getConnectorId()
     {
         return connectorId;
     }
@@ -78,7 +79,7 @@ public String toString()
         return connectorId + ":" + columnName;
     }
 
-    public static Map<String, ColumnHandle> toSystemColumnHandles(String connectorId, ConnectorTableMetadata tableMetadata)
+    public static Map<String, ColumnHandle> toSystemColumnHandles(ConnectorId connectorId, ConnectorTableMetadata tableMetadata)
     {
         ImmutableMap.Builder<String, ColumnHandle> columnHandles = ImmutableMap.builder();
         for (ColumnMetadata columnMetadata : tableMetadata.getColumns()) {

presto-main/src/main/java/com/facebook/presto/connector/system/SystemConnector.java

@@ -13,6 +13,7 @@
  */
 package com.facebook.presto.connector.system;
 
+import com.facebook.presto.connector.ConnectorId;
 import com.facebook.presto.metadata.InternalNodeManager;
 import com.facebook.presto.spi.SystemTable;
 import com.facebook.presto.spi.connector.ConnectorMetadata;
@@ -31,14 +32,14 @@
 public class SystemConnector
         implements InternalConnector
 {
-    private final String connectorId;
+    private final ConnectorId connectorId;
     private final ConnectorMetadata metadata;
     private final ConnectorSplitManager splitManager;
     private final ConnectorPageSourceProvider pageSourceProvider;
     private final Function<TransactionId, ConnectorTransactionHandle> transactionHandleFunction;
 
     public SystemConnector(
-            String connectorId,
+            ConnectorId connectorId,
             InternalNodeManager nodeManager,
             Set<SystemTable> tables,
             Function<TransactionId, ConnectorTransactionHandle> transactionHandleFunction)