public static QueryRunner createKuduQueryRunner(String schema)
        throws Exception
{
    QueryRunner runner = null;
    String kuduSchema = schema;
    try {
        runner = DistributedQueryRunner.builder(createSession(kuduSchema)).setNodeCount(3).build();

        installKuduConnector(runner, kuduSchema);

        return runner;
    }
    catch (Throwable e) {
        closeAllSuppress(e, runner);
        throw e;
    }
}
 

public static QueryRunner createKuduQueryRunnerTpch(Iterable<TpchTable<?>> tables)
        throws Exception
{
    DistributedQueryRunner runner = null;
    String kuduSchema = "tpch";
    try {
        runner = DistributedQueryRunner.builder(createSession(kuduSchema)).setNodeCount(3).build();

        runner.installPlugin(new TpchPlugin());
        runner.createCatalog("tpch", "tpch");

        installKuduConnector(runner, kuduSchema);

        copyTpchTables(runner, "tpch", TINY_SCHEMA_NAME, createSession(kuduSchema), tables);

        return runner;
    }
    catch (Throwable e) {
        closeAllSuppress(e, runner);
        throw e;
    }
}
 

/**
 * Install the plugin into the given query runner, using the mock client and the given table descriptions.
 *
 * The plug in is returned so that the injector can be accessed and other setup items tested.
 *
 * @param queryRunner
 * @param streamDescriptions
 * @return
 */
public static KinesisPlugin installKinesisPlugin(QueryRunner queryRunner, Map<SchemaTableName, KinesisStreamDescription> streamDescriptions)
{
    KinesisPlugin kinesisPlugin = createPluginInstance();
    // Note: function literal with provided descriptions instead of KinesisTableDescriptionSupplier:
    kinesisPlugin.setTableDescriptionSupplier(() -> streamDescriptions);
    kinesisPlugin.setAltProviderClass(KinesisTestClientManager.class);

    queryRunner.installPlugin(kinesisPlugin);

    Map<String, String> kinesisConfig = ImmutableMap.of(
            "kinesis.default-schema", "default",
            "kinesis.access-key", "",
            "kinesis.secret-key", "");
    queryRunner.createCatalog("kinesis", "kinesis", kinesisConfig);

    return kinesisPlugin;
}
 

private static void installKuduConnector(QueryRunner runner, String schema)
{
    String masterAddresses = System.getProperty("kudu.client.master-addresses", "localhost:7051");
    Map<String, String> properties = ImmutableMap.of(
                "kudu.client.master-addresses", masterAddresses);

    runner.installPlugin(new KuduPlugin());
    runner.createCatalog("kudu", "kudu", properties);

    runner.execute("DROP SCHEMA IF EXISTS " + schema);
    runner.execute("CREATE SCHEMA " + schema);
}
 

/**
 * Install the plug in into the given query runner, using normal setup but with the given table descriptions.
 *
 * Note that this uses the actual client and will incur charges from AWS when run.  Mainly for full
 * integration tests.
 *
 * @param queryRunner
 * @param streamDescriptions
 * @param accessKey
 * @param secretKey
 */
public static void installKinesisPlugin(QueryRunner queryRunner, Map<SchemaTableName, KinesisStreamDescription> streamDescriptions, String accessKey, String secretKey)
{
    KinesisPlugin kinesisPlugin = createPluginInstance();
    // Note: function literal with provided descriptions instead of KinesisTableDescriptionSupplier:
    kinesisPlugin.setTableDescriptionSupplier(() -> streamDescriptions);
    queryRunner.installPlugin(kinesisPlugin);

    Map<String, String> kinesisConfig = ImmutableMap.of(
                "kinesis.default-schema", "default",
                "kinesis.access-key", accessKey,
                "kinesis.secret-key", secretKey);
    queryRunner.createCatalog("kinesis", "kinesis", kinesisConfig);
}
 

public static QueryRunner createKuduQueryRunnerTpch(TpchTable<?>... tables)
        throws Exception
{
    return createKuduQueryRunnerTpch(ImmutableList.copyOf(tables));
}
 

public QueryRunner get()
{
    return queryRunner;
}