import com.facebook.presto.spi.connector.Connector; //导入依赖的package包/类
@Override
public Connector create(String connectorId, Map<String, String> config, ConnectorContext context)
{
    requireNonNull(config, "config is null");

    try {
        Bootstrap app = new Bootstrap(
                new JsonModule(),
                new HDFSModule(connectorId, context.getTypeManager())
        );

        Injector injector = app
                .strictConfig()
                .doNotInitializeLogging()
                .setRequiredConfigurationProperties(config)
                .initialize();

        return injector.getInstance(HDFSConnector.class);
    }
    catch (Exception e) {
        e.printStackTrace();
    }
    return null;
}
 

import com.facebook.presto.spi.connector.Connector; //导入依赖的package包/类
@Override
public Connector create(final String connectorId, Map<String, String> requiredConfig, ConnectorContext context)
{
    requireNonNull(requiredConfig, "config is null");

    try {
        Bootstrap app = new Bootstrap(
                binder -> binder.bind(NodeManager.class).toInstance(context.getNodeManager()),
                new KuduModule(connectorId));

        Injector injector = app
                .strictConfig()
                .doNotInitializeLogging()
                .setRequiredConfigurationProperties(requiredConfig)
                .initialize();

        return injector.getInstance(KuduConnector.class);
    }
    catch (Exception e) {
        throw Throwables.propagate(e);
    }
}
 

import com.facebook.presto.spi.connector.Connector; //导入依赖的package包/类
@PreDestroy
public void stop()
{
    if (stopped.getAndSet(true)) {
        return;
    }

    for (Map.Entry<String, Connector> entry : connectors.entrySet()) {
        Connector connector = entry.getValue();
        try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(connector.getClass().getClassLoader())) {
            connector.shutdown();
        }
        catch (Throwable t) {
            log.error(t, "Error shutting down connector: %s", entry.getKey());
        }
    }
}
 

import com.facebook.presto.spi.connector.Connector; //导入依赖的package包/类
private synchronized void addCatalogConnector(String catalogName, String connectorId, ConnectorFactory factory, Map<String, String> properties)
{
    Connector connector = createConnector(connectorId, factory, properties);

    addConnectorInternal(ConnectorType.STANDARD, catalogName, connectorId, connector);

    String informationSchemaId = makeInformationSchemaConnectorId(connectorId);
    addConnectorInternal(ConnectorType.INFORMATION_SCHEMA, catalogName, informationSchemaId, new InformationSchemaConnector(catalogName, nodeManager, metadataManager));

    String systemId = makeSystemTablesConnectorId(connectorId);
    addConnectorInternal(ConnectorType.SYSTEM, catalogName, systemId, new SystemConnector(
            systemId,
            nodeManager,
            connector.getSystemTables(),
            transactionId -> transactionManager.getConnectorTransaction(transactionId, connectorId)));

    // Register session and table properties once per catalog
    metadataManager.getSessionPropertyManager().addConnectorSessionProperties(catalogName, connector.getSessionProperties());
    metadataManager.getTablePropertyManager().addTableProperties(catalogName, connector.getTableProperties());
}
 

import com.facebook.presto.spi.connector.Connector; //导入依赖的package包/类
private static void registerBogusConnector(TransactionManager transactionManager, String connectorId)
{
    transactionManager.addConnector(connectorId, new Connector()
    {
        @Override
        public ConnectorTransactionHandle beginTransaction(IsolationLevel isolationLevel, boolean readOnly)
        {
            // Just return something
            return new ConnectorTransactionHandle() {};
        }

        @Override
        public ConnectorMetadata getMetadata(ConnectorTransactionHandle transactionHandle)
        {
            throw new UnsupportedOperationException();
        }

        @Override
        public ConnectorSplitManager getSplitManager()
        {
            throw new UnsupportedOperationException();
        }
    });
}
 

import com.facebook.presto.spi.connector.Connector; //导入依赖的package包/类
private static Connector createTestingConnector(String connectorId)
{
    return new LegacyTransactionConnector(connectorId, new com.facebook.presto.spi.Connector()
    {
        private final ConnectorMetadata metadata = new TestingMetadata();

        @Override
        public ConnectorMetadata getMetadata()
        {
            return metadata;
        }

        @Override
        public ConnectorSplitManager getSplitManager()
        {
            throw new UnsupportedOperationException();
        }
    });
}
 

import com.facebook.presto.spi.connector.Connector; //导入依赖的package包/类
@Override
public Connector create(String connectorId, Map<String, String> config, ConnectorContext context) {
  Path basePath = new Path(config.get(BASEPATH));
  String spreadsheetSubDir = config.get(SUBDIR);
  String useFileCacheStr = config.get(USE_FILE_CACHE);
  String proxyUserStr = config.get(PROXY_USER);
  boolean proxyUser = false;
  if (proxyUserStr != null) {
    proxyUser = Boolean.parseBoolean(proxyUserStr);
  }
  boolean useFileCache = true;
  if (useFileCacheStr != null) {
    useFileCache = Boolean.parseBoolean(useFileCacheStr);
  }
  try {
    return new SpreadsheetConnector(UserGroupInformation.getCurrentUser(), _configuration, basePath,
        spreadsheetSubDir, useFileCache, proxyUser);
  } catch (IOException e) {
    throw new RuntimeException(e);
  }
}
 

import com.facebook.presto.spi.connector.Connector; //导入依赖的package包/类
/**
 * Build a connector instance from the plug in, supplying the given properties.
 *
 * This can build a connector with the mock client which is normally done in testing.
 * The plug in is created first with createPluginInstance.
 *
 * @param plugin
 * @param properties
 * @param withMockClient
 * @return
 */
public static KinesisConnector createConnector(KinesisPlugin plugin, Map<String, String> properties, boolean withMockClient)
{
    requireNonNull(plugin, "Plugin instance should not be null");
    requireNonNull(properties, "Properties map should not be null (can be empty)");

    if (withMockClient) {
        plugin.setAltProviderClass(KinesisTestClientManager.class);
    }

    ConnectorFactory factory = plugin.getConnectorFactories().iterator().next();
    assertNotNull(factory);

    Connector connector = factory.create("kinesis", properties, new TestingConnectorContext() {});
    assertTrue(connector instanceof KinesisConnector);
    return (KinesisConnector) connector;
}
 

import com.facebook.presto.spi.connector.Connector; //导入依赖的package包/类
@Override
public Connector create(String s, Map<String, String> config, ConnectorContext context)
{
    NodeManager nodeManager = context.getNodeManager();

    return new RestConnector(nodeManager, restFactory.create(config));
}
 

import com.facebook.presto.spi.connector.Connector; //导入依赖的package包/类
@Override
    public Connector create(String connectorId, Map<String, String> config, ConnectorContext context) {
        requireNonNull(connectorId, "connectorId is null");
        requireNonNull(config, "config is null");

        try {
            Bootstrap app = new Bootstrap(
//                    new JsonModule(),
                    new EthereumConnectorModule(),
                    binder -> {
                        binder.bind(EthereumConnectorId.class).toInstance(new EthereumConnectorId(connectorId));
                        binder.bind(TypeManager.class).toInstance(context.getTypeManager());
                        binder.bind(NodeManager.class).toInstance(context.getNodeManager());
                    }
            );

            Injector injector = app.strictConfig()
                    .doNotInitializeLogging()
                    .setRequiredConfigurationProperties(config)
                    .initialize();

            return injector.getInstance(EthereumConnector.class);
        }
        catch (Exception e) {
            throw Throwables.propagate(e);
        }
    }
 

import com.facebook.presto.spi.connector.Connector; //导入依赖的package包/类
public Connector create(final String connectorId, Map<String, String> requiredConfig,
                        ConnectorContext context) {
  requireNonNull(requiredConfig, "requiredConfig is null");

  final String
      locator_host =
      requiredConfig
          .getOrDefault(MonarchProperties.LOCATOR_HOST, MonarchProperties.LOCATOR_HOST_DEFAULT);
  final int
      locator_port =
      Integer.parseInt(requiredConfig
          .getOrDefault(MonarchProperties.LOCATOR_PORT, MonarchProperties.LOCATOR_PORT_DEFAULT));

  // Create a client that connects to the Ampool cluster via a locator (that is already running!).
  final Properties props = new Properties();
  props.setProperty(Constants.MClientCacheconfig.MONARCH_CLIENT_LOG, requiredConfig
      .getOrDefault(MonarchProperties.MONARCH_CLIENT_LOG, MonarchProperties.MONARCH_CLIENT_LOG_DEFAULT_LOCATION));
  final AmpoolClient aClient = new AmpoolClient(locator_host, locator_port, props);
  log.info("INFORMATION: AmpoolClient created successfully.");

  try {
    Bootstrap
        app =
        new Bootstrap(new AmpoolModule(connectorId, aClient, context.getTypeManager()));

    Injector injector = app
        .doNotInitializeLogging()
        .setRequiredConfigurationProperties(requiredConfig)
        .initialize();

    log.info("INFORMATION: Injector initialized successfully.");
    return injector.getInstance(AmpoolConnector.class);
  } catch (Exception e) {
    throw Throwables.propagate(e);
  }
}
 

import com.facebook.presto.spi.connector.Connector; //导入依赖的package包/类
@Override
public Connector create(String connectorId, Map<String, String> config)
{
    requireNonNull(connectorId, "connectorId is null");
    requireNonNull(config, "config is null");

    try {
        Bootstrap app = new Bootstrap(
                new JsonModule(),
                new KafkaConnectorModule(),
                binder -> {
                    binder.bind(KafkaConnectorId.class).toInstance(new KafkaConnectorId(connectorId));
                    binder.bind(TypeManager.class).toInstance(typeManager);
                    binder.bind(NodeManager.class).toInstance(nodeManager);

                    if (tableDescriptionSupplier.isPresent()) {
                        binder.bind(new TypeLiteral<Supplier<Map<SchemaTableName, KafkaTopicDescription>>>() {}).toInstance(tableDescriptionSupplier.get());
                    }
                    else {
                        binder.bind(new TypeLiteral<Supplier<Map<SchemaTableName, KafkaTopicDescription>>>() {}).to(KafkaTableDescriptionSupplier.class).in(Scopes.SINGLETON);
                    }
                }
        );

        Injector injector = app.strictConfig()
                .doNotInitializeLogging()
                .setRequiredConfigurationProperties(config)
                .setOptionalConfigurationProperties(optionalConfig)
                .initialize();

        return injector.getInstance(KafkaConnector.class);
    }
    catch (Exception e) {
        throw Throwables.propagate(e);
    }
}
 

import com.facebook.presto.spi.connector.Connector; //导入依赖的package包/类
@Test
public void testSpinup()
{
    ConnectorFactory factory = testConnectorExists();
    Connector c = factory.create("test-connector", ImmutableMap.<String, String>builder()
            .put("kafka.table-names", "test")
            .put("kafka.nodes", "localhost:9092")
            .build());
    assertNotNull(c);
}
 

import com.facebook.presto.spi.connector.Connector; //导入依赖的package包/类
@Override
public Connector create(String connectorId, Map<String, String> config)
{
    try {
        Bootstrap app = new Bootstrap(
                new JsonModule(),
                new MBeanModule(),
                binder -> {
                    CurrentNodeId currentNodeId = new CurrentNodeId(nodeManager.getCurrentNode().getNodeIdentifier());
                    MBeanServer mbeanServer = new RebindSafeMBeanServer(getPlatformMBeanServer());

                    binder.bind(MBeanServer.class).toInstance(mbeanServer);
                    binder.bind(CurrentNodeId.class).toInstance(currentNodeId);
                    binder.bind(NodeManager.class).toInstance(nodeManager);
                    binder.bind(PageSorter.class).toInstance(pageSorter);
                    binder.bind(BlockEncodingSerde.class).toInstance(blockEncodingSerde);
                    binder.bind(TypeManager.class).toInstance(typeManager);
                },
                metadataModule,
                new BackupModule(backupProviders),
                new StorageModule(connectorId),
                new RaptorModule(connectorId));

        Injector injector = app
                .strictConfig()
                .doNotInitializeLogging()
                .setRequiredConfigurationProperties(config)
                .setOptionalConfigurationProperties(optionalConfig)
                .initialize();

        return injector.getInstance(RaptorConnector.class);
    }
    catch (Exception e) {
        throw Throwables.propagate(e);
    }
}
 

import com.facebook.presto.spi.connector.Connector; //导入依赖的package包/类
@Override
public Connector create(String connectorId, Map<String, String> requiredConfig)
{
    return new BlackHoleConnector(
            new BlackHoleMetadata(),
            new BlackHoleSplitManager(),
            new BlackHolePageSourceProvider(),
            new BlackHolePageSinkProvider()
    );
}
 

import com.facebook.presto.spi.connector.Connector; //导入依赖的package包/类
@Override
public Connector create(String connectorId, Map<String, String> properties)
{
    return new Connector()
    {
        @Override
        public ConnectorTransactionHandle beginTransaction(IsolationLevel isolationLevel, boolean readOnly)
        {
            checkConnectorSupports(READ_COMMITTED, isolationLevel);
            return JmxTransactionHandle.INSTANCE;
        }

        @Override
        public ConnectorMetadata getMetadata(ConnectorTransactionHandle transactionHandle)
        {
            return new JmxMetadata(connectorId, mbeanServer);
        }

        @Override
        public ConnectorSplitManager getSplitManager()
        {
            return new JmxSplitManager(connectorId, nodeManager);
        }

        @Override
        public ConnectorRecordSetProvider getRecordSetProvider()
        {
            return new JmxRecordSetProvider(mbeanServer, nodeManager.getCurrentNode().getNodeIdentifier());
        }
    };
}
 

import com.facebook.presto.spi.connector.Connector; //导入依赖的package包/类
public synchronized ConnectorTransactionMetadata getConnectorTransactionMetadata(String connectorId, Connector connector)
{
    checkOpenTransaction();
    ConnectorTransactionMetadata transactionMetadata = connectorIdToMetadata.get(connectorId);
    if (transactionMetadata == null) {
        transactionMetadata = new ConnectorTransactionMetadata(connector, beginTransaction(connector));
        // Don't use computeIfAbsent b/c the beginTransaction call might be recursive
        checkState(connectorIdToMetadata.put(connectorId, transactionMetadata) == null);
    }
    return transactionMetadata;
}
 

import com.facebook.presto.spi.connector.Connector; //导入依赖的package包/类
private ConnectorTransactionHandle beginTransaction(Connector connector)
{
    if (connector instanceof InternalConnector) {
        return ((InternalConnector) connector).beginTransaction(transactionId, isolationLevel, readOnly);
    }
    else {
        return connector.beginTransaction(isolationLevel, readOnly);
    }
}
 

import com.facebook.presto.spi.connector.Connector; //导入依赖的package包/类
private static Connector createConnector(String connectorId, ConnectorFactory factory, Map<String, String> properties)
{
    Class<?> factoryClass = factory.getClass();
    if (factory instanceof LegacyTransactionConnectorFactory) {
        factoryClass = ((LegacyTransactionConnectorFactory) factory).getConnectorFactory().getClass();
    }

    try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(factoryClass.getClassLoader())) {
        return factory.create(connectorId, properties);
    }
}
 

import com.facebook.presto.spi.connector.Connector; //导入依赖的package包/类
@Test
public void testTransactionWorkflow()
        throws Exception
{
    try (IdleCheckExecutor executor = new IdleCheckExecutor()) {
        TransactionManager transactionManager = TransactionManager.create(new TransactionManagerConfig(), executor.getExecutor(), finishingExecutor);

        Connector c1 = new TpchConnectorFactory(new InMemoryNodeManager()).create("c1", ImmutableMap.of());
        transactionManager.addConnector("c1", c1);

        TransactionId transactionId = transactionManager.beginTransaction(false);

        Assert.assertEquals(transactionManager.getAllTransactionInfos().size(), 1);
        TransactionInfo transactionInfo = transactionManager.getTransactionInfo(transactionId);
        Assert.assertFalse(transactionInfo.isAutoCommitContext());
        Assert.assertTrue(transactionInfo.getConnectorIds().isEmpty());
        Assert.assertFalse(transactionInfo.getWrittenConnectorId().isPresent());

        ConnectorMetadata metadata = transactionManager.getMetadata(transactionId, "c1");
        metadata.listSchemaNames(TEST_SESSION.toConnectorSession("c1"));
        transactionInfo = transactionManager.getTransactionInfo(transactionId);
        Assert.assertEquals(transactionInfo.getConnectorIds(), ImmutableList.of("c1"));
        Assert.assertFalse(transactionInfo.getWrittenConnectorId().isPresent());

        transactionManager.asyncCommit(transactionId).join();

        Assert.assertTrue(transactionManager.getAllTransactionInfos().isEmpty());
    }
}