import org.apache.cassandra.io.sstable.SSTableReader; //导入依赖的package包/类
private void replace(Collection<SSTableReader> oldSSTables, Iterable<SSTableReader> replacements)
{
    if (!cfstore.isValid())
    {
        removeOldSSTablesSize(replacements);
        replacements = Collections.emptyList();
    }

    View currentView, newView;
    do
    {
        currentView = view.get();
        newView = currentView.replace(oldSSTables, replacements);
    }
    while (!view.compareAndSet(currentView, newView));

    postReplace(oldSSTables, replacements);
}
 

import org.apache.cassandra.io.sstable.SSTableReader; //导入依赖的package包/类
/**
 * Removes files from compacting status: this is different from 'markObsolete'
 * because it should be run regardless of whether a compaction succeeded.
 */
public void unmarkCompacting(Iterable<SSTableReader> unmark)
{
    if (!cfstore.isValid())
    {
        // The CF has been dropped.  We don't know if the original compaction suceeded or failed,
        // which makes it difficult to know if the sstable reference has already been released.
        // A "good enough" approach is to mark the sstables involved obsolete, which if compaction succeeded
        // is harmlessly redundant, and if it failed ensures that at least the sstable will get deleted on restart.
        for (SSTableReader sstable : unmark)
            sstable.markObsolete();
    }

    View currentView, newView;
    do
    {
        currentView = view.get();
        newView = currentView.unmarkCompacting(unmark);
    }
    while (!view.compareAndSet(currentView, newView));
}
 

import org.apache.cassandra.io.sstable.SSTableReader; //导入依赖的package包/类
public Future<Pair<KeyCacheKey, RowIndexEntry>> deserialize(DataInputStream input, ColumnFamilyStore cfs) throws IOException
{
    int keyLength = input.readInt();
    if (keyLength > FBUtilities.MAX_UNSIGNED_SHORT)
    {
        throw new IOException(String.format("Corrupted key cache. Key length of %d is longer than maximum of %d",
                                            keyLength, FBUtilities.MAX_UNSIGNED_SHORT));
    }
    ByteBuffer key = ByteBufferUtil.read(input, keyLength);
    int generation = input.readInt();
    SSTableReader reader = findDesc(generation, cfs.getSSTables());
    input.readBoolean(); // backwards compatibility for "promoted indexes" boolean
    if (reader == null)
    {
        RowIndexEntry.Serializer.skipPromotedIndex(input);
        return null;
    }
    RowIndexEntry entry = reader.metadata.comparator.rowIndexEntrySerializer().deserialize(input, reader.descriptor.version);
    return Futures.immediateFuture(Pair.create(new KeyCacheKey(cfs.metadata.cfId, reader.descriptor, key), entry));
}
 

import org.apache.cassandra.io.sstable.SSTableReader; //导入依赖的package包/类
/**
 * Removes every SSTable in the directory from the DataTracker's view.
 * @param directory the unreadable directory, possibly with SSTables in it, but not necessarily.
 */
void removeUnreadableSSTables(File directory)
{
    View currentView, newView;
    List<SSTableReader> remaining = new ArrayList<>();
    do
    {
        currentView = view.get();
        for (SSTableReader r : currentView.nonCompactingSStables())
            if (!r.descriptor.directory.equals(directory))
                remaining.add(r);

        if (remaining.size() == currentView.nonCompactingSStables().size())
            return;

        newView = currentView.replace(currentView.sstables, remaining);
    }
    while (!view.compareAndSet(currentView, newView));
    notifySSTablesChanged(remaining, Collections.<SSTableReader>emptySet(), OperationType.UNKNOWN);
}
 

import org.apache.cassandra.io.sstable.SSTableReader; //导入依赖的package包/类
public SSTableIndex(ColumnIndex index, File indexFile, SSTableReader referent)
{
    this.columnIndex = index;
    this.sstable = referent;

    if (!sstable.acquireReference())
        throw new IllegalStateException("Couldn't acquire reference to the sstable: " + sstable);

    AbstractType<?> validator = columnIndex.getValidator();

    assert validator != null;
    assert indexFile.exists() : String.format("SSTable %s should have index %s.",
            sstable.getFilename(),
            columnIndex.getIndexName());

    this.index = new OnDiskIndex(indexFile, validator, new DecoratedKeyFetcher(sstable));
}
 

import org.apache.cassandra.io.sstable.SSTableReader; //导入依赖的package包/类
public boolean reduceScopeForLimitedSpace()
{
    if (partialCompactionsAcceptable() && sstables.size() > 1)
    {
        // Try again w/o the largest one.
        logger.warn("insufficient space to compact all requested files {}", StringUtils.join(sstables, ", "));
        // Note that we have removed files that are still marked as compacting.
        // This suboptimal but ok since the caller will unmark all the sstables at the end.
        SSTableReader removedSSTable = cfs.getMaxSizeFile(sstables);
        if (sstables.remove(removedSSTable))
        {
            cfs.getDataTracker().unmarkCompacting(Arrays.asList(removedSSTable));
            return true;
        }
    }
    return false;
}
 

import org.apache.cassandra.io.sstable.SSTableReader; //导入依赖的package包/类
/**
 * A special kind of replacement for SSTableReaders that were cloned with a new index summary sampling level (see
 * SSTableReader.cloneWithNewSummarySamplingLevel and CASSANDRA-5519).  This does not mark the old reader
 * as compacted.
 * @param oldSSTables replaced readers
 * @param newSSTables replacement readers
 */
private void replaceReaders(Collection<SSTableReader> oldSSTables, Collection<SSTableReader> newSSTables, boolean notify)
{
    View currentView, newView;
    do
    {
        currentView = view.get();
        newView = currentView.replace(oldSSTables, newSSTables);
    }
    while (!view.compareAndSet(currentView, newView));

    if (!oldSSTables.isEmpty() && notify)
        notifySSTablesChanged(oldSSTables, newSSTables, OperationType.UNKNOWN);

    for (SSTableReader sstable : newSSTables)
        sstable.setTrackedBy(this);

    Refs.release(Refs.selfRefs(oldSSTables));
}
 

import org.apache.cassandra.io.sstable.SSTableReader; //导入依赖的package包/类
/**
 * Removes every SSTable in the directory from the DataTracker's view.
 * @param directory the unreadable directory, possibly with SSTables in it, but not necessarily.
 */
void removeUnreadableSSTables(File directory)
{
    View currentView, newView;
    Set<SSTableReader> remaining = new HashSet<>();
    do
    {
        currentView = view.get();
        for (SSTableReader r : currentView.nonCompactingSStables())
            if (!r.descriptor.directory.equals(directory))
                remaining.add(r);

        if (remaining.size() == currentView.nonCompactingSStables().size())
            return;

        newView = currentView.replace(currentView.sstables, remaining);
    }
    while (!view.compareAndSet(currentView, newView));
    for (SSTableReader sstable : currentView.sstables)
        if (!remaining.contains(sstable))
            sstable.selfRef().release();
    notifySSTablesChanged(remaining, Collections.<SSTableReader>emptySet(), OperationType.UNKNOWN);
}
 

import org.apache.cassandra.io.sstable.SSTableReader; //导入依赖的package包/类
@Override
public synchronized AbstractCompactionTask getNextBackgroundTask(int gcBefore)
{
    if (!isEnabled())
        return null;

    while (true)
    {
        List<SSTableReader> latestBucket = getNextBackgroundSSTables(gcBefore);

        if (latestBucket.isEmpty())
            return null;

        if (cfs.getDataTracker().markCompacting(latestBucket))
            return new CompactionTask(cfs, latestBucket, gcBefore, false);
    }
}
 

import org.apache.cassandra.io.sstable.SSTableReader; //导入依赖的package包/类
/**
 *
 * @param gcBefore
 * @return
 */
private List<SSTableReader> getNextBackgroundSSTables(final int gcBefore)
{
    if (!isEnabled() || cfs.getSSTables().isEmpty())
        return Collections.emptyList();

    Set<SSTableReader> uncompacting = Sets.intersection(sstables, cfs.getUncompactingSSTables());

    // Find fully expired SSTables. Those will be included no matter what.
    Set<SSTableReader> expired = CompactionController.getFullyExpiredSSTables(cfs, uncompacting, cfs.getOverlappingSSTables(uncompacting), gcBefore);
    Set<SSTableReader> candidates = Sets.newHashSet(filterSuspectSSTables(uncompacting));

    List<SSTableReader> compactionCandidates = new ArrayList<>(getNextNonExpiredSSTables(Sets.difference(candidates, expired), gcBefore));
    if (!expired.isEmpty())
    {
        logger.debug("Including expired sstables: {}", expired);
        compactionCandidates.addAll(expired);
    }
    return compactionCandidates;
}
 

import org.apache.cassandra.io.sstable.SSTableReader; //导入依赖的package包/类
public synchronized AbstractCompactionTask getNextBackgroundTask(int gcBefore)
{
    if (!isEnabled())
        return null;

    while (true)
    {
        List<SSTableReader> smallestBucket = getNextBackgroundSSTables(gcBefore);

        if (smallestBucket.isEmpty())
            return null;

        if (cfs.getDataTracker().markCompacting(smallestBucket))
            return new CompactionTask(cfs, smallestBucket, gcBefore);
    }
}
 

import org.apache.cassandra.io.sstable.SSTableReader; //导入依赖的package包/类
/**
 * Returns a (bucket, hotness) pair or null if there were not enough sstables in the bucket to meet minThreshold.
 * If there are more than maxThreshold sstables, the coldest sstables will be trimmed to meet the threshold.
 **/
@VisibleForTesting
static Pair<List<SSTableReader>, Double> trimToThresholdWithHotness(List<SSTableReader> bucket, int maxThreshold)
{
    // Sort by sstable hotness (descending). We first build a map because the hotness may change during the sort.
    final Map<SSTableReader, Double> hotnessSnapshot = getHotnessMap(bucket);
    Collections.sort(bucket, new Comparator<SSTableReader>()
    {
        public int compare(SSTableReader o1, SSTableReader o2)
        {
            return -1 * Double.compare(hotnessSnapshot.get(o1), hotnessSnapshot.get(o2));
        }
    });

    // and then trim the coldest sstables off the end to meet the maxThreshold
    List<SSTableReader> prunedBucket = bucket.subList(0, Math.min(bucket.size(), maxThreshold));

    // bucket hotness is the sum of the hotness of all sstable members
    double bucketHotness = 0.0;
    for (SSTableReader sstr : prunedBucket)
        bucketHotness += hotness(sstr);

    return Pair.create(prunedBucket, bucketHotness);
}
 

import org.apache.cassandra.io.sstable.SSTableReader; //导入依赖的package包/类
/**
 * removes all sstables that are not busy compacting.
 */
public void unreferenceSSTables()
{
    Set<SSTableReader> notCompacting;

    View currentView, newView;
    do
    {
        currentView = view.get();
        notCompacting = currentView.nonCompactingSStables();
        newView = currentView.replace(notCompacting, Collections.<SSTableReader>emptySet());
    }
    while (!view.compareAndSet(currentView, newView));

    if (notCompacting.isEmpty())
    {
        // notifySSTablesChanged -> LeveledManifest.promote doesn't like a no-op "promotion"
        return;
    }
    notifySSTablesChanged(notCompacting, Collections.<SSTableReader>emptySet(), OperationType.UNKNOWN);
    postReplace(notCompacting, Collections.<SSTableReader>emptySet());
}
 

import org.apache.cassandra.io.sstable.SSTableReader; //导入依赖的package包/类
public LeveledScanner(Collection<SSTableReader> sstables, Range<Token> range)
{
    this.range = range;

    // add only sstables that intersect our range, and estimate how much data that involves
    this.sstables = new ArrayList<SSTableReader>(sstables.size());
    long length = 0;
    for (SSTableReader sstable : sstables)
    {
        this.sstables.add(sstable);
        long estimatedKeys = sstable.estimatedKeys();
        double estKeysInRangeRatio = 1.0;

        if (estimatedKeys > 0 && range != null)
            estKeysInRangeRatio = ((double) sstable.estimatedKeysForRanges(Collections.singleton(range))) / estimatedKeys;

        length += sstable.uncompressedLength() * estKeysInRangeRatio;
    }

    totalLength = length;
    Collections.sort(this.sstables, SSTableReader.sstableComparator);
    sstableIterator = this.sstables.iterator();
    assert sstableIterator.hasNext(); // caller should check intersecting first
    currentScanner = sstableIterator.next().getScanner(range, CompactionManager.instance.getRateLimiter());
}
 

import org.apache.cassandra.io.sstable.SSTableReader; //导入依赖的package包/类
/**
 * Replaces old SSTables with new by creating new immutable tracker.
 *
 * @param oldSSTables A set of SSTables to remove.
 * @param newSSTables A set of SSTables to add to tracker.
 *
 * @return A collection of SSTables which don't have component attached for current index.
 */
public Iterable<SSTableReader> update(Collection<SSTableReader> oldSSTables, Collection<SSTableReader> newSSTables)
{
    final Set<SSTableIndex> newIndexes = getIndexes(newSSTables);
    final Set<SSTableReader> indexedSSTables = getSSTables(newIndexes);

    View currentView, newView;
    do
    {
        currentView = view.get();
        newView = new View(columnIndex, keyValidator, currentView.getIndexes(), oldSSTables, newIndexes);
    }
    while (!view.compareAndSet(currentView, newView));

    return Iterables.filter(newSSTables, new Predicate<SSTableReader>()
    {
        @Override
        public boolean apply(SSTableReader sstable)
        {
            return !indexedSSTables.contains(sstable);
        }
    });
}
 

import org.apache.cassandra.io.sstable.SSTableReader; //导入依赖的package包/类
public SSTableNamesIterator(SSTableReader sstable, DecoratedKey key, SortedSet<ByteBuffer> columns)
{
    assert columns != null;
    this.sstable = sstable;
    this.columns = columns;
    this.key = key;

    RowIndexEntry indexEntry = sstable.getPosition(key, SSTableReader.Operator.EQ);
    if (indexEntry == null)
        return;

    try
    {
        read(sstable, null, indexEntry);
    }
    catch (IOException e)
    {
        sstable.markSuspect();
        throw new CorruptSSTableException(e, sstable.getFilename());
    }
    finally
    {
        if (fileToClose != null)
            FileUtils.closeQuietly(fileToClose);
    }
}
 

import org.apache.cassandra.io.sstable.SSTableReader; //导入依赖的package包/类
@Test
public void testTrackTimesRangeTombstoneWithData() throws ExecutionException, InterruptedException
{
    Keyspace ks = Keyspace.open(KSNAME);
    ColumnFamilyStore cfs = ks.getColumnFamilyStore(CFNAME);
    cfs.truncateBlocking();
    String key = "rt_times";
    Mutation rm = new Mutation(KSNAME, ByteBufferUtil.bytes(key));
    add(rm, 5, 999);
    rm.apply();
    key = "rt_times2";
    rm = new Mutation(KSNAME, ByteBufferUtil.bytes(key));
    ColumnFamily cf = rm.addOrGet(CFNAME);
    int timestamp = (int)(System.currentTimeMillis()/1000);
    cf.delete(new DeletionInfo(b(1), b(2), cfs.getComparator(), 1000, timestamp));
    rm.apply();
    cfs.forceBlockingFlush();
    SSTableReader sstable = cfs.getSSTables().iterator().next();
    assertTimes(sstable.getSSTableMetadata(), 999, 1000, Integer.MAX_VALUE);
    cfs.forceMajorCompaction();
    sstable = cfs.getSSTables().iterator().next();
    assertTimes(sstable.getSSTableMetadata(), 999, 1000, Integer.MAX_VALUE);
}
 

import org.apache.cassandra.io.sstable.SSTableReader; //导入依赖的package包/类
public MetadataCollector(Collection<SSTableReader> sstables, CellNameType columnNameComparator, int level)
{
    this(columnNameComparator);

    replayPosition(ReplayPosition.getReplayPosition(sstables));
    sstableLevel(level);
    // Get the max timestamp of the precompacted sstables
    // and adds generation of live ancestors
    for (SSTableReader sstable : sstables)
    {
        addAncestor(sstable.descriptor.generation);
        for (Integer i : sstable.getAncestors())
            if (new File(sstable.descriptor.withGeneration(i).filenameFor(Component.DATA)).exists())
                addAncestor(i);
    }
}
 

import org.apache.cassandra.io.sstable.SSTableReader; //导入依赖的package包/类
View replaceFlushed(Memtable flushedMemtable, SSTableReader newSSTable)
{
    int index = flushingMemtables.indexOf(flushedMemtable);
    List<Memtable> newQueuedMemtables = ImmutableList.<Memtable>builder()
                                                     .addAll(flushingMemtables.subList(0, index))
                                                     .addAll(flushingMemtables.subList(index + 1, flushingMemtables.size()))
                                                     .build();
    Map<SSTableReader, SSTableReader> newSSTables = sstablesMap;
    SSTableIntervalTree intervalTree = this.intervalTree;
    if (newSSTable != null)
    {
        assert !sstables.contains(newSSTable);
        assert !shadowed.contains(newSSTable);
        newSSTables = ImmutableMap.<SSTableReader, SSTableReader>builder()
                                  .putAll(sstablesMap).put(newSSTable, newSSTable).build();
        intervalTree = buildIntervalTree(newSSTables.keySet());
    }
    return new View(liveMemtables, newQueuedMemtables, newSSTables, compacting, shadowed, intervalTree);
}
 

import org.apache.cassandra.io.sstable.SSTableReader; //导入依赖的package包/类
public List<SSTableMetadata> getSSTableMetadata(String ksName, String cfName) {
    ColumnFamilyStore cfStore = getStore(ksName, cfName);
    Collection<SSTableReader> tables = cfStore.getSSTables();
    List<SSTableMetadata> metaData = new ArrayList<>(tables.size());
    for (SSTableReader table : tables) {
        SSTableMetadata tableMetadata = new SSTableMetadata();
        File dataFile = new File(table.descriptor.filenameFor(Component.DATA));
        tableMetadata.filename = dataFile.getName();
        tableMetadata.generation = table.descriptor.generation;
        try {
            tableMetadata.fileTimestamp = Files.getLastModifiedTime(dataFile.toPath()).toMillis();
        } catch (IOException e) {
            tableMetadata.fileTimestamp = 0;
        }
        tableMetadata.minTimestamp = table.getMinTimestamp();
        tableMetadata.maxTimestamp = table.getMaxTimestamp();
        tableMetadata.diskLength = table.onDiskLength();
        tableMetadata.uncompressedLength = table.uncompressedLength();
        tableMetadata.keys = table.estimatedKeys();
        EstimatedHistogram rowSizeHistogram = table.getEstimatedRowSize();
        tableMetadata.maxRowSize = rowSizeHistogram.max();
        tableMetadata.avgRowSize = rowSizeHistogram.mean();
        EstimatedHistogram columnCountHistogram = table.getEstimatedColumnCount();
        tableMetadata.maxColumnCount = columnCountHistogram.max();
        tableMetadata.avgColumnCount = columnCountHistogram.mean();
        tableMetadata.droppableTombstones = table.getDroppableTombstonesBefore(Util.NOW_SECONDS - table.metadata.getGcGraceSeconds());
        tableMetadata.level = table.getSSTableLevel();
        tableMetadata.isRepaired = table.isRepaired();
        tableMetadata.repairedAt = table.getSSTableMetadata().repairedAt;
        metaData.add(tableMetadata);
    }
    return metaData;
}