import org.skife.jdbi.v2.IDBI; //导入依赖的package包/类
private IDBI connectToDatasource(ShardInfo shardInfo) {
    PooledDataSourceFactory urlReplacedCopy =
            copyAndReplaceURL(dropwizardDSFactory, shardInfo);

    // we cannot re-register using the same name (dropwizard metrics do not merge on name collision)
    String dbShardName = humanFriendlyShardNamer.nameShard(shardInfo);
    DBI jdbi = dropwizardDbiFactory.build(
            dropwizardEnvironment,
            urlReplacedCopy,
            dbShardName);
    if (DBIInitializer != null) {
        DBIInitializer.initialize(jdbi);
    }
    logger.info("New shard connection created: " + dbShardName);

    return jdbi;
}
 

import org.skife.jdbi.v2.IDBI; //导入依赖的package包/类
public ShardIterator(long tableId, TupleDomain<RaptorColumnHandle> effectivePredicate, IDBI dbi)
{
    ShardPredicate predicate = ShardPredicate.create(effectivePredicate);

    String sql = format(
            "SELECT shard_uuid, node_ids FROM %s WHERE %s",
            shardIndexTable(tableId),
            predicate.getPredicate());

    dao = onDemandDao(dbi, ShardManagerDao.class);
    fetchNodes();

    try {
        connection = dbi.open().getConnection();
        statement = connection.prepareStatement(sql);
        enableStreamingResults(statement);
        predicate.bind(statement);
        log.debug("Running query: %s", statement);
        resultSet = statement.executeQuery();
    }
    catch (SQLException e) {
        close();
        throw metadataError(e);
    }
}
 

import org.skife.jdbi.v2.IDBI; //导入依赖的package包/类
@Inject
public ShardCleaner(
        @ForMetadata IDBI dbi,
        NodeManager nodeManager,
        StorageService storageService,
        Optional<BackupStore> backupStore,
        ShardCleanerConfig config)
{
    this(dbi,
            nodeManager.getCurrentNode().getNodeIdentifier(),
            nodeManager.getCoordinators().contains(nodeManager.getCurrentNode()),
            storageService,
            backupStore,
            config.getMaxTransactionAge(),
            config.getLocalCleanerInterval(),
            config.getLocalCleanTime(),
            config.getLocalPurgeTime(),
            config.getBackupCleanerInterval(),
            config.getBackupCleanTime(),
            config.getBackupPurgeTime(),
            config.getBackupDeletionThreads());
}
 

import org.skife.jdbi.v2.IDBI; //导入依赖的package包/类
@BeforeMethod
public void setup()
        throws Exception
{
    temporary = createTempDir();
    File directory = new File(temporary, "data");
    File backupDirectory = new File(temporary, "backup");
    backupStore = new FileBackupStore(backupDirectory);
    backupStore.start();
    storageService = new FileStorageService(directory);
    storageService.start();

    IDBI dbi = new DBI("jdbc:h2:mem:test" + System.nanoTime());
    dummyHandle = dbi.open();
    ShardManager shardManager = createShardManager(dbi);
    recoveryManager = createShardRecoveryManager(storageService, Optional.of(backupStore), shardManager);
}
 

import org.skife.jdbi.v2.IDBI; //导入依赖的package包/类
@BeforeMethod
public void setup()
        throws Exception
{
    temporary = createTempDir();
    File directory = new File(temporary, "data");
    storageService = new FileStorageService(directory);
    storageService.start();

    File backupDirectory = new File(temporary, "backup");
    fileBackupStore = new FileBackupStore(backupDirectory);
    fileBackupStore.start();
    backupStore = Optional.of(fileBackupStore);

    IDBI dbi = new DBI("jdbc:h2:mem:test" + System.nanoTime());
    dummyHandle = dbi.open();
    ShardManager shardManager = createShardManager(dbi);
    Duration discoveryInterval = new Duration(5, TimeUnit.MINUTES);
    recoveryManager = new ShardRecoveryManager(storageService, backupStore, nodeManager, shardManager, discoveryInterval, 10);

    shardRecorder = new InMemoryShardRecorder();
}
 

import org.skife.jdbi.v2.IDBI; //导入依赖的package包/类
public ShardMetadataRecordCursor(IDBI dbi, TupleDomain<Integer> tupleDomain)
{
    requireNonNull(dbi, "dbi is null");
    this.dbi = dbi;
    this.metadataDao = onDemandDao(dbi, MetadataDao.class);
    this.tupleDomain = requireNonNull(tupleDomain, "tupleDomain is null");
    this.tableIds = getTableIds(dbi, tupleDomain);
    this.columnNames = createQualifiedColumnNames();
    this.resultSetValues = new ResultSetValues(TYPES);
    this.resultSet = getNextResultSet();
}
 

import org.skife.jdbi.v2.IDBI; //导入依赖的package包/类
@VisibleForTesting
static Iterator<Long> getTableIds(IDBI dbi, TupleDomain<Integer> tupleDomain)
{
    Map<Integer, Domain> domains = tupleDomain.getDomains().get();
    Domain schemaNameDomain = domains.get(getColumnIndex(SHARD_METADATA, SCHEMA_NAME));
    Domain tableNameDomain = domains.get(getColumnIndex(SHARD_METADATA, TABLE_NAME));

    StringBuilder sql = new StringBuilder("SELECT table_id FROM tables ");
    if (schemaNameDomain != null || tableNameDomain != null) {
        sql.append("WHERE ");
        List<String> predicates = new ArrayList<>();
        if (tableNameDomain != null && tableNameDomain.isSingleValue()) {
            predicates.add(format("table_name = '%s'", getStringValue(tableNameDomain.getSingleValue())));
        }
        if (schemaNameDomain != null && schemaNameDomain.isSingleValue()) {
            predicates.add(format("schema_name = '%s'", getStringValue(schemaNameDomain.getSingleValue())));
        }
        sql.append(Joiner.on(" AND ").join(predicates));
    }

    ImmutableList.Builder<Long> tableIds = ImmutableList.builder();
    try (Connection connection = dbi.open().getConnection();
            Statement statement = connection.createStatement();
            ResultSet resultSet = statement.executeQuery(sql.toString())) {
        while (resultSet.next()) {
            tableIds.add(resultSet.getLong("table_id"));
        }
    }
    catch (SQLException e) {
        throw metadataError(e);
    }
    return tableIds.build().iterator();
}
 

import org.skife.jdbi.v2.IDBI; //导入依赖的package包/类
@ForMetadata
@Singleton
@Provides
public IDBI createDBI(@ForMetadata ConnectionFactory connectionFactory, TypeManager typeManager)
{
    DBI dbi = new DBI(connectionFactory);
    dbi.registerMapper(new TableColumn.Mapper(typeManager));
    return dbi;
}
 

import org.skife.jdbi.v2.IDBI; //导入依赖的package包/类
public RaptorMetadata(
        String connectorId,
        IDBI dbi,
        ShardManager shardManager,
        JsonCodec<ShardInfo> shardInfoCodec,
        JsonCodec<ShardDelta> shardDeltaCodec)
{
    this.connectorId = requireNonNull(connectorId, "connectorId is null");
    this.dbi = requireNonNull(dbi, "dbi is null");
    this.dao = onDemandDao(dbi, MetadataDao.class);
    this.shardManager = requireNonNull(shardManager, "shardManager is null");
    this.shardInfoCodec = requireNonNull(shardInfoCodec, "shardInfoCodec is null");
    this.shardDeltaCodec = requireNonNull(shardDeltaCodec, "shardDeltaCodec is null");
}
 

import org.skife.jdbi.v2.IDBI; //导入依赖的package包/类
public static void createTablesWithRetry(IDBI dbi)
{
    Duration delay = new Duration(2, TimeUnit.SECONDS);
    while (true) {
        try (Handle handle = dbi.open()) {
            createTables(handle.attach(SchemaDao.class));
            return;
        }
        catch (UnableToObtainConnectionException e) {
            log.warn("Failed to connect to database. Will retry again in %s. Exception: %s", delay, e.getMessage());
            sleep(delay);
        }
    }
}
 

import org.skife.jdbi.v2.IDBI; //导入依赖的package包/类
@Inject
public DatabaseShardManager(@ForMetadata IDBI dbi, NodeSupplier nodeSupplier)
{
    this.dbi = requireNonNull(dbi, "dbi is null");
    this.dao = onDemandDao(dbi, ShardManagerDao.class);
    this.nodeSupplier = requireNonNull(nodeSupplier, "nodeSupplier is null");

    createTablesWithRetry(dbi);
}
 

import org.skife.jdbi.v2.IDBI; //导入依赖的package包/类
public static <T> T runTransaction(IDBI dbi, TransactionCallback<T> callback)
{
    try {
        return dbi.inTransaction(callback);
    }
    catch (DBIException e) {
        propagateIfInstanceOf(e.getCause(), PrestoException.class);
        throw metadataError(e);
    }
}
 

import org.skife.jdbi.v2.IDBI; //导入依赖的package包/类
@Inject
public ShardCompactionManager(@ForMetadata IDBI dbi, NodeManager nodeManager, ShardManager shardManager, ShardCompactor compactor, StorageManagerConfig config)
{
    this(dbi,
            nodeManager.getCurrentNode().getNodeIdentifier(),
            shardManager,
            compactor,
            config.getCompactionInterval(),
            config.getMaxShardSize(),
            config.getMaxShardRows(),
            config.getCompactionThreads(),
            config.isCompactionEnabled());
}
 

import org.skife.jdbi.v2.IDBI; //导入依赖的package包/类
@Inject
public RaptorMetadataFactory(
        RaptorConnectorId connectorId,
        @ForMetadata IDBI dbi,
        ShardManager shardManager,
        JsonCodec<ShardInfo> shardInfoCodec,
        JsonCodec<ShardDelta> shardDeltaCodec)
{
    this.connectorId = requireNonNull(connectorId, "connectorId is null").toString();
    this.dbi = requireNonNull(dbi, "dbi is null");
    this.shardManager = requireNonNull(shardManager, "shardManager is null");
    this.shardInfoCodec = requireNonNull(shardInfoCodec, "shardInfoCodec is null");
    this.shardDeltaCodec = requireNonNull(shardDeltaCodec, "shardDeltaCodec is null");
}
 

import org.skife.jdbi.v2.IDBI; //导入依赖的package包/类
@BeforeMethod
public void setup()
        throws Exception
{
    IDBI dbi = new DBI("jdbc:h2:mem:test" + System.nanoTime());
    dummyHandle = dbi.open();
    dao = dbi.onDemand(MetadataDao.class);
    createTablesWithRetry(dbi);
}
 

import org.skife.jdbi.v2.IDBI; //导入依赖的package包/类
@BeforeMethod
public void setup()
        throws Exception
{
    temporary = createTempDir();
    IDBI dbi = new DBI("jdbc:h2:mem:test" + System.nanoTime());
    dummyHandle = dbi.open();
    storageManager = createOrcStorageManager(dbi, temporary, MAX_SHARD_ROWS);
    compactor = new ShardCompactor(storageManager, READER_ATTRIBUTES);
}
 

import org.skife.jdbi.v2.IDBI; //导入依赖的package包/类
public static OrcStorageManager createOrcStorageManager(IDBI dbi, File temporary, int maxShardRows)
        throws IOException
{
    File directory = new File(temporary, "data");
    StorageService storageService = new FileStorageService(directory);
    storageService.start();

    File backupDirectory = new File(temporary, "backup");
    FileBackupStore fileBackupStore = new FileBackupStore(backupDirectory);
    fileBackupStore.start();
    Optional<BackupStore> backupStore = Optional.of(fileBackupStore);

    ShardManager shardManager = createShardManager(dbi);
    ShardRecoveryManager recoveryManager = new ShardRecoveryManager(
            storageService,
            backupStore,
            new InMemoryNodeManager(),
            shardManager,
            MISSING_SHARD_DISCOVERY,
            10);
    return createOrcStorageManager(
            storageService,
            backupStore,
            recoveryManager,
            new InMemoryShardRecorder(),
            maxShardRows,
            MAX_FILE_SIZE);
}
 

import org.skife.jdbi.v2.IDBI; //导入依赖的package包/类
@VisibleForTesting
public IDBI getDBI() {
    final DBI dbi = new DBI(dataSource);
    dbi.registerArgumentFactory(new DateTimeArgumentFactory());
    dbi.registerArgumentFactory(new DateTimeZoneArgumentFactory());
    dbi.registerArgumentFactory(new UUIDArgumentFactory());
    return dbi;
}
 

import org.skife.jdbi.v2.IDBI; //导入依赖的package包/类
@Inject
public TimelineAggregator(final IDBI dbi, final TimelineDao timelineDao, final TimelineCoder timelineCoder,
                          final SampleCoder sampleCoder, final MeterConfig config) {
    this.dbi = dbi;
    this.timelineDao = timelineDao;
    this.timelineCoder = timelineCoder;
    this.sampleCoder = sampleCoder;
    this.config = config;
    this.aggregatorSqlDao = dbi.onDemand(TimelineAggregatorSqlDao.class);
    this.timelineChunkMapper = new TimelineChunkMapper();
}
 

import org.skife.jdbi.v2.IDBI; //导入依赖的package包/类
public static IDBI getDBI() {
    try {
        final MeterModule meterModule = new MeterModule(helper.getDataSource());
        return meterModule.getDBI();
    } catch (IOException e) {
        Assert.fail(e.toString());
        return null;
    }
}