import org.apache.cassandra.io.sstable.ReducingKeyIterator; //导入依赖的package包/类
/**
 * Builds the index using the data in the underlying CFS
 * Blocks till it's complete
 */
protected void buildIndexBlocking()
{
    logger.info(String.format("Submitting index build of %s for data in %s",
            getIndexName(), StringUtils.join(baseCfs.getSSTables(), ", ")));

    try (Refs<SSTableReader> sstables = baseCfs.selectAndReference(ColumnFamilyStore.CANONICAL_SSTABLES).refs)
    {
        SecondaryIndexBuilder builder = new SecondaryIndexBuilder(baseCfs,
                                                                  Collections.singleton(getIndexName()),
                                                                  new ReducingKeyIterator(sstables));
        Future<?> future = CompactionManager.instance.submitIndexBuild(builder);
        FBUtilities.waitOnFuture(future);
        forceBlockingFlush();
        setIndexBuilt();
    }
    logger.info("Index build of {} complete", getIndexName());
}
 

import org.apache.cassandra.io.sstable.ReducingKeyIterator; //导入依赖的package包/类
/**
 * Does a full, blocking rebuild of the indexes specified by columns from the sstables.
 * Does nothing if columns is empty.
 *
 * Caller must acquire and release references to the sstables used here.
 *
 * @param sstables the data to build from
 * @param idxNames the list of columns to index, ordered by comparator
 */
public void maybeBuildSecondaryIndexes(Collection<SSTableReader> sstables, Set<String> idxNames)
{
    if (idxNames.isEmpty())
        return;

    logger.info(String.format("Submitting index build of %s for data in %s",
                              idxNames, StringUtils.join(sstables, ", ")));

    SecondaryIndexBuilder builder = new SecondaryIndexBuilder(baseCfs, idxNames, new ReducingKeyIterator(sstables));
    Future<?> future = CompactionManager.instance.submitIndexBuild(builder);
    FBUtilities.waitOnFuture(future);

    flushIndexesBlocking();

    logger.info("Index build of {} complete", idxNames);
}
 

import org.apache.cassandra.io.sstable.ReducingKeyIterator; //导入依赖的package包/类
/**
 * Builds the index using the data in the underlying CFS
 * Blocks till it's complete
 */
protected void buildIndexBlocking()
{
    logger.info(String.format("Submitting index build of %s for data in %s",
            getIndexName(), StringUtils.join(baseCfs.getSSTables(), ", ")));

    Collection<SSTableReader> sstables = baseCfs.markCurrentSSTablesReferenced();
    SecondaryIndexBuilder builder = new SecondaryIndexBuilder(baseCfs,
                                                              Collections.singleton(getIndexName()),
                                                              new ReducingKeyIterator(sstables));
    Future<?> future = CompactionManager.instance.submitIndexBuild(builder);
    FBUtilities.waitOnFuture(future);
    forceBlockingFlush();

    setIndexBuilt();
    logger.info("Index build of {} complete", getIndexName());
}
 

import org.apache.cassandra.io.sstable.ReducingKeyIterator; //导入依赖的package包/类
private void buildIndexesBlocking(Collection<SSTableReader> sstables, Set<Index> indexes)
{
    if (indexes.isEmpty())
        return;

    logger.info("Submitting index build of {} for data in {}",
                indexes.stream().map(i -> i.getIndexMetadata().name).collect(Collectors.joining(",")),
                sstables.stream().map(SSTableReader::toString).collect(Collectors.joining(",")));

    SecondaryIndexBuilder builder = new SecondaryIndexBuilder(baseCfs,
                                                              indexes,
                                                              new ReducingKeyIterator(sstables));
    Future<?> future = CompactionManager.instance.submitIndexBuild(builder);
    FBUtilities.waitOnFuture(future);

    flushIndexesBlocking(indexes);
    logger.info("Index build of {} complete",
                indexes.stream().map(i -> i.getIndexMetadata().name).collect(Collectors.joining(",")));
}
 

import org.apache.cassandra.io.sstable.ReducingKeyIterator; //导入依赖的package包/类
/**
 * Builds the index using the data in the underlying CFS
 * Blocks till it's complete
 */
protected void buildIndexBlocking()
{
    logger.info(String.format("Submitting index build of %s for data in %s",
            getIndexName(), StringUtils.join(baseCfs.getSSTables(), ", ")));

    Collection<SSTableReader> sstables = baseCfs.markCurrentSSTablesReferenced();
    try
    {
        SecondaryIndexBuilder builder = new SecondaryIndexBuilder(baseCfs,
                                                                  Collections.singleton(getIndexName()),
                                                                  new ReducingKeyIterator(sstables));
        Future<?> future = CompactionManager.instance.submitIndexBuild(builder);
        FBUtilities.waitOnFuture(future);
        forceBlockingFlush();
        setIndexBuilt();
    }
    finally
    {
        SSTableReader.releaseReferences(sstables);
    }
    logger.info("Index build of {} complete", getIndexName());
}
 

import org.apache.cassandra.io.sstable.ReducingKeyIterator; //导入依赖的package包/类
public SecondaryIndexBuilder(ColumnFamilyStore cfs, Set<Index> indexers, ReducingKeyIterator iter)
{
    this.cfs = cfs;
    this.indexers = indexers;
    this.iter = iter;
    this.compactionId = UUIDGen.getTimeUUID();
}
 

import org.apache.cassandra.io.sstable.ReducingKeyIterator; //导入依赖的package包/类
private void buildBlocking()
{
    baseCfs.forceBlockingFlush();

    try (ColumnFamilyStore.RefViewFragment viewFragment = baseCfs.selectAndReference(View.selectFunction(SSTableSet.CANONICAL));
         Refs<SSTableReader> sstables = viewFragment.refs)
    {
        if (sstables.isEmpty())
        {
            logger.info("No SSTable data for {}.{} to build index {} from, marking empty index as built",
                        baseCfs.metadata.ksName,
                        baseCfs.metadata.cfName,
                        metadata.name);
            baseCfs.indexManager.markIndexBuilt(metadata.name);
            return;
        }

        logger.info("Submitting index build of {} for data in {}",
                    metadata.name,
                    getSSTableNames(sstables));

        SecondaryIndexBuilder builder = new SecondaryIndexBuilder(baseCfs,
                                                                  Collections.singleton(this),
                                                                  new ReducingKeyIterator(sstables));
        Future<?> future = CompactionManager.instance.submitIndexBuild(builder);
        FBUtilities.waitOnFuture(future);
        indexCfs.forceBlockingFlush();
        baseCfs.indexManager.markIndexBuilt(metadata.name);
    }
    logger.info("Index build of {} complete", metadata.name);
}
 

import org.apache.cassandra.io.sstable.ReducingKeyIterator; //导入依赖的package包/类
public SecondaryIndexBuilder(ColumnFamilyStore cfs, Set<String> idxNames, ReducingKeyIterator iter)
{
    this.cfs = cfs;
    this.idxNames = idxNames;
    this.iter = iter;
}
 

import org.apache.cassandra.io.sstable.ReducingKeyIterator; //导入依赖的package包/类
public IndexBuilder createIndexBuilder(ColumnFamilyStore cfs, SortedSet<ByteBuffer> columns, ReducingKeyIterator iter)
{
    return new IndexBuilder(cfs, columns, iter);
}
 

import org.apache.cassandra.io.sstable.ReducingKeyIterator; //导入依赖的package包/类
public IndexBuilder(ColumnFamilyStore cfs, SortedSet<ByteBuffer> columns, ReducingKeyIterator iter)
{
    this.cfs = cfs;
    this.columns = columns;
    this.iter = iter;
}