Java 类com.facebook.presto.spi.connector.ConnectorFactory 实例源码

@Override
public <T> List<T> getServices(Class<T> type)
{
    checkState(nodeManager != null, "NodeManager has not been set");
    checkState(blockEncodingSerde != null, "BlockEncodingSerde has not been set");
    checkState(typeManager != null, "TypeManager has not been set");

    if (type == ConnectorFactory.class) {
        return ImmutableList.of(type.cast(new RaptorConnectorFactory(
                name,
                metadataModule,
                backupProviders,
                optionalConfig,
                nodeManager,
                pageSorter,
                blockEncodingSerde,
                typeManager)));
    }
    return ImmutableList.of();
}

public static LocalQueryRunner createLocalQueryRunner()
{
    Session session = testSessionBuilder()
            .setCatalog("raptor")
            .setSchema("benchmark")
            .build();
    LocalQueryRunner localQueryRunner = new LocalQueryRunner(session);

    // add tpch
    InMemoryNodeManager nodeManager = localQueryRunner.getNodeManager();
    localQueryRunner.createCatalog("tpch", new TpchConnectorFactory(nodeManager, 1), ImmutableMap.<String, String>of());

    // add raptor
    ConnectorFactory raptorConnectorFactory = createRaptorConnectorFactory(TPCH_CACHE_DIR, nodeManager);
    localQueryRunner.createCatalog("raptor", raptorConnectorFactory, ImmutableMap.of());

    if (!localQueryRunner.tableExists(session, "orders")) {
        localQueryRunner.execute("CREATE TABLE orders AS SELECT * FROM tpch.sf1.orders");
    }
    if (!localQueryRunner.tableExists(session, "lineitem")) {
        localQueryRunner.execute("CREATE TABLE lineitem AS SELECT * FROM tpch.sf1.lineitem");
    }
    return localQueryRunner;
}

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());
}

@Test
public ConnectorFactory testConnectorExists()
{
    KinesisPlugin plugin = TestUtils.createPluginInstance();

    // Create factory manually to double check everything is done right
    Iterable<ConnectorFactory> iter = plugin.getConnectorFactories();

    List<ConnectorFactory> factories = new ArrayList<>();
    for (ConnectorFactory cf : iter) {
        factories.add(cf);
    }
    assertNotNull(factories);
    assertEquals(factories.size(), 1);
    ConnectorFactory factory = factories.get(0);
    assertNotNull(factory);
    return factory;
}

/**
 * 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;
}

@Override
public Iterable<ConnectorFactory> getConnectorFactories()
{
    return ImmutableList.of(new RestConnectorFactory(
            "slack", config -> new SlackRest(config.get("token"))
    ));
}

@Override
public Iterable<ConnectorFactory> getConnectorFactories()
{
    return ImmutableList.of(new RestConnectorFactory(
            "github", config -> new GithubRest(config.get("token"))
    ));
}

@Override
public Iterable<ConnectorFactory> getConnectorFactories()
{
    return ImmutableList.of(new RestConnectorFactory(
            "twitter",
            config -> new TwitterRest(
                    config.get("customer_key"),
                    config.get("customer_secret"),
                    config.get("token"),
                    config.get("secret"))
    ));
}

@Override
public synchronized <T> List<T> getServices(Class<T> type)
{
    if (type == ConnectorFactory.class) {
        return ImmutableList.of(type.cast(new KafkaConnectorFactory(typeManager, nodeManager, tableDescriptionSupplier, optionalConfig)));
    }
    return ImmutableList.of();
}

@Test
public ConnectorFactory testConnectorExists()
{
    KafkaPlugin plugin = new KafkaPlugin();
    plugin.setTypeManager(new TestingTypeManager());
    plugin.setNodeManager(new TestingNodeManager());

    List<ConnectorFactory> factories = plugin.getServices(ConnectorFactory.class);
    assertNotNull(factories);
    assertEquals(factories.size(), 1);
    ConnectorFactory factory = factories.get(0);
    assertNotNull(factory);
    return factory;
}

@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);
}

private static ConnectorFactory createRaptorConnectorFactory(String cacheDir, NodeManager nodeManager)
{
    try {
        File dataDir = new File(cacheDir);
        File databaseDir = new File(dataDir, "db");

        Map<String, String> config = ImmutableMap.<String, String>builder()
                .put("metadata.db.type", "h2")
                .put("metadata.db.filename", databaseDir.getAbsolutePath())
                .put("storage.data-directory", dataDir.getAbsolutePath())
                .put("storage.compress", "false")
                .build();

        TypeManager typeManager = new TypeRegistry();
        BlockEncodingSerde blockEncodingSerde = new BlockEncodingManager(typeManager);

        RaptorPlugin plugin = new RaptorPlugin();

        plugin.setOptionalConfig(config);
        plugin.setNodeManager(nodeManager);
        plugin.setBlockEncodingSerde(blockEncodingSerde);
        plugin.setTypeManager(typeManager);

        return getOnlyElement(plugin.getServices(ConnectorFactory.class));
    }
    catch (Exception e) {
        throw Throwables.propagate(e);
    }
}

@Test
public void testPlugin()
        throws Exception
{
    RaptorPlugin plugin = loadPlugin(RaptorPlugin.class);

    plugin.setNodeManager(new InMemoryNodeManager());

    TypeRegistry typeRegistry = new TypeRegistry();
    plugin.setTypeManager(typeRegistry);
    plugin.setBlockEncodingSerde(new BlockEncodingManager(typeRegistry));

    plugin.setPageSorter(new PagesIndexPageSorter());

    List<ConnectorFactory> factories = plugin.getServices(ConnectorFactory.class);
    ConnectorFactory factory = getOnlyElement(factories);
    assertInstanceOf(factory, RaptorConnectorFactory.class);

    File tmpDir = Files.createTempDir();
    try {
        Map<String, String> config = ImmutableMap.<String, String>builder()
                .put("metadata.db.type", "h2")
                .put("metadata.db.filename", tmpDir.getAbsolutePath())
                .put("storage.data-directory", tmpDir.getAbsolutePath())
                .build();

        factory.create("test", config);
    }
    finally {
        FileUtils.deleteRecursively(tmpDir);
    }
}

@Override
public <T> List<T> getServices(Class<T> type)
{
    if (type == ConnectorFactory.class) {
        return ImmutableList.of(type.cast(new BlackHoleConnectorFactory()));
    }
    return ImmutableList.of();
}

@Override
public <T> List<T> getServices(Class<T> type)
{
    if (type == ConnectorFactory.class) {
        requireNonNull(nodeManager, "nodeManager is null");
        return ImmutableList.of(type.cast(new IndexedTpchConnectorFactory(nodeManager, indexSpec, 4)));
    }
    return ImmutableList.of();
}

@Override
public synchronized <T> List<T> getServices(Class<T> type)
{
    if (type == ConnectorFactory.class) {
        return ImmutableList.of(type.cast(new JmxConnectorFactory(mBeanServer, nodeManager)));
    }
    return ImmutableList.of();
}

public void addConnectorFactory(ConnectorFactory connectorFactory)
{
    checkState(!stopped.get(), "ConnectorManager is stopped");
    ConnectorFactory existingConnectorFactory = connectorFactories.putIfAbsent(connectorFactory.getName(), connectorFactory);
    checkArgument(existingConnectorFactory == null, "Connector %s is already registered", connectorFactory.getName());
    handleResolver.addConnectorName(connectorFactory.getName(), connectorFactory.getHandleResolver());
}

public void 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);
}

private synchronized void 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);
    checkState(!connectors.containsKey(connectorId), "A connector %s already exists", connectorId);

    addCatalogConnector(catalogName, connectorId, connectorFactory, properties);

    catalogs.add(catalogName);
}

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);
    }
}

@Deprecated
public void createCatalog(String catalogName, com.facebook.presto.spi.ConnectorFactory connectorFactory, Map<String, String> properties)
{
    nodeManager.addCurrentNodeDatasource(catalogName);
    connectorManager.addConnectorFactory(connectorFactory);
    connectorManager.createConnection(catalogName, connectorFactory.getName(), properties);
}

@Override
public <T> List<T> getServices(Class<T> type)
{
    if (type == ConnectorFactory.class) {
        requireNonNull(nodeManager, "nodeManager is null");
        return ImmutableList.of(type.cast(new TpchConnectorFactory(nodeManager)));
    }
    return ImmutableList.of();
}

@Override
public <T> List<T> getServices(Class<T> type)
{
    if (type == ConnectorFactory.class) {
        requireNonNull(nodeManager, "nodeManager is null");
        return ImmutableList.of(type.cast(new SampledTpchConnectorFactory(nodeManager, 1, 2)));
    }
    return ImmutableList.of();
}

@Override
public <T> List<T> getServices(Class<T> type)
{
    if (type == ConnectorFactory.class) {
        return ImmutableList.of(type.cast(new CassandraConnectorFactory("cassandra", optionalConfig)));
    }
    return ImmutableList.of();
}

@Override
public synchronized Iterable<ConnectorFactory> getConnectorFactories()
{
    if(factory == null) {
        this.factory = new KinesisConnectorFactory(getClassLoader(), tableDescriptionSupplier, optionalConfig, altProviderClass);
    }
    return ImmutableList.of(this.factory);
}

@Parameters({
        "kinesis.awsAccessKey",
        "kinesis.awsSecretKey"
})
@Test
public void testSpinUp(String awsAccessKey, String awsSecretKey)
{
    ConnectorFactory factory = testConnectorExists();
    // Important: this has to be created before we setup the injector in the factory:
    assertNotNull(factory.getHandleResolver());

    Connector c = factory.create("kinesis.test-connector", ImmutableMap.<String, String>builder()
            .put("kinesis.hide-internal-columns", "false")
            .put("kinesis.access-key", TestUtils.noneToBlank(awsAccessKey))
            .put("kinesis.secret-key", TestUtils.noneToBlank(awsSecretKey))
            .build(), new TestingConnectorContext() {});
    assertNotNull(c);

    // Verify that the key objects have been created on the connector
    assertNotNull(c.getRecordSetProvider());
    assertNotNull(c.getSplitManager());
    ConnectorMetadata md = c.getMetadata(KinesisTransactionHandle.INSTANCE);
    assertNotNull(md);

    ConnectorTransactionHandle handle = c.beginTransaction(READ_COMMITTED, true);
    assertTrue(handle != null && handle instanceof KinesisTransactionHandle);
}

@Override
public Iterable<ConnectorFactory> getConnectorFactories()
{
    return ImmutableList.of(new HDFSConnectorFactory());
}

@Override
public Iterable<ConnectorFactory> getConnectorFactories() {
    return ImmutableList.of(new EthereumConnectorFactory());
}

@Override
public Iterable<ConnectorFactory> getConnectorFactories() {
    return ImmutableList.of(new AmpoolConnectorFactory());
}

@Override
public Iterable<ConnectorFactory> getConnectorFactories()
{
    return ImmutableList.of(new KuduConnectorFactory());
}

public LegacyTransactionConnectorFactory(com.facebook.presto.spi.ConnectorFactory connectorFactory)
{
    this.connectorFactory = requireNonNull(connectorFactory, "connectorFactory is null");
}

public com.facebook.presto.spi.ConnectorFactory getConnectorFactory()
{
    return connectorFactory;
}

@Deprecated
public void addConnectorFactory(com.facebook.presto.spi.ConnectorFactory connectorFactory)
{
    addConnectorFactory(new LegacyTransactionConnectorFactory(connectorFactory));
}

public void createCatalog(String catalogName, ConnectorFactory connectorFactory, Map<String, String> properties)
{
    nodeManager.addCurrentNodeDatasource(catalogName);
    connectorManager.addConnectorFactory(connectorFactory);
    connectorManager.createConnection(catalogName, connectorFactory.getName(), properties);
}

@Override
public Iterable<ConnectorFactory> getConnectorFactories() {
  return ImmutableList.of(new SpreadsheetConnectorFactory(getClassLoader()));
}