import org.apache.cassandra.utils.StreamingHistogram; //导入依赖的package包/类
public ColumnStats(int columnCount,
                   long minTimestamp,
                   long maxTimestamp,
                   int maxLocalDeletionTime,
                   StreamingHistogram tombstoneHistogram,
                   List<ByteBuffer> minColumnNames,
                   List<ByteBuffer> maxColumnNames,
                   boolean hasLegacyCounterShards)
{
    this.minTimestamp = minTimestamp;
    this.maxTimestamp = maxTimestamp;
    this.maxLocalDeletionTime = maxLocalDeletionTime;
    this.columnCount = columnCount;
    this.tombstoneHistogram = tombstoneHistogram;
    this.minColumnNames = minColumnNames;
    this.maxColumnNames = maxColumnNames;
    this.hasLegacyCounterShards = hasLegacyCounterShards;
}
 

import org.apache.cassandra.utils.StreamingHistogram; //导入依赖的package包/类
/**
 * Used to serialize to an old version - needed to be able to update sstable level without a full compaction.
 *
 * @deprecated will be removed when it is assumed that the minimum upgrade-from-version is the version that this
 * patch made it into
 *
 * @param sstableStats
 * @param legacyDesc
 * @param out
 * @throws IOException
 */
@Deprecated
public void legacySerialize(SSTableMetadata sstableStats, Set<Integer> ancestors, Descriptor legacyDesc, DataOutput out) throws IOException
{
    EstimatedHistogram.serializer.serialize(sstableStats.estimatedRowSize, out);
    EstimatedHistogram.serializer.serialize(sstableStats.estimatedColumnCount, out);
    ReplayPosition.serializer.serialize(sstableStats.replayPosition, out);
    out.writeLong(sstableStats.minTimestamp);
    out.writeLong(sstableStats.maxTimestamp);
    if (legacyDesc.version.tracksMaxLocalDeletionTime)
        out.writeInt(sstableStats.maxLocalDeletionTime);
    if (legacyDesc.version.hasBloomFilterFPChance)
        out.writeDouble(sstableStats.bloomFilterFPChance);
    out.writeDouble(sstableStats.compressionRatio);
    out.writeUTF(sstableStats.partitioner);
    out.writeInt(ancestors.size());
    for (Integer g : ancestors)
        out.writeInt(g);
    StreamingHistogram.serializer.serialize(sstableStats.estimatedTombstoneDropTime, out);
    out.writeInt(sstableStats.sstableLevel);
    if (legacyDesc.version.tracksMaxMinColumnNames)
        serializeMinMaxColumnNames(sstableStats.minColumnNames, sstableStats.maxColumnNames, out);
}
 

import org.apache.cassandra.utils.StreamingHistogram; //导入依赖的package包/类
private SSTableMetadata(EstimatedHistogram rowSizes,
                        EstimatedHistogram columnCounts,
                        ReplayPosition replayPosition,
                        long minTimestamp,
                        long maxTimestamp,
                        double cr,
                        String partitioner,
                        StreamingHistogram estimatedTombstoneDropTime)
{
    this.estimatedRowSize = rowSizes;
    this.estimatedColumnCount = columnCounts;
    this.replayPosition = replayPosition;
    this.minTimestamp = minTimestamp;
    this.maxTimestamp = maxTimestamp;
    this.compressionRatio = cr;
    this.partitioner = partitioner;
    this.estimatedTombstoneDropTime = estimatedTombstoneDropTime;
}
 

import org.apache.cassandra.utils.StreamingHistogram; //导入依赖的package包/类
public StatsMetadata(EstimatedHistogram estimatedRowSize,
                     EstimatedHistogram estimatedColumnCount,
                     ReplayPosition replayPosition,
                     long minTimestamp,
                     long maxTimestamp,
                     int maxLocalDeletionTime,
                     double compressionRatio,
                     StreamingHistogram estimatedTombstoneDropTime,
                     int sstableLevel,
                     List<ByteBuffer> minColumnNames,
                     List<ByteBuffer> maxColumnNames)
{
    this.estimatedRowSize = estimatedRowSize;
    this.estimatedColumnCount = estimatedColumnCount;
    this.replayPosition = replayPosition;
    this.minTimestamp = minTimestamp;
    this.maxTimestamp = maxTimestamp;
    this.maxLocalDeletionTime = maxLocalDeletionTime;
    this.compressionRatio = compressionRatio;
    this.estimatedTombstoneDropTime = estimatedTombstoneDropTime;
    this.sstableLevel = sstableLevel;
    this.minColumnNames = minColumnNames;
    this.maxColumnNames = maxColumnNames;
}
 

import org.apache.cassandra.utils.StreamingHistogram; //导入依赖的package包/类
public StatsMetadata(EstimatedHistogram estimatedPartitionSize,
                     EstimatedHistogram estimatedColumnCount,
                     IntervalSet<CommitLogPosition> commitLogIntervals,
                     long minTimestamp,
                     long maxTimestamp,
                     int minLocalDeletionTime,
                     int maxLocalDeletionTime,
                     int minTTL,
                     int maxTTL,
                     double compressionRatio,
                     StreamingHistogram estimatedTombstoneDropTime,
                     int sstableLevel,
                     List<ByteBuffer> minClusteringValues,
                     List<ByteBuffer> maxClusteringValues,
                     boolean hasLegacyCounterShards,
                     long repairedAt,
                     long totalColumnsSet,
                     long totalRows)
{
    this.estimatedPartitionSize = estimatedPartitionSize;
    this.estimatedColumnCount = estimatedColumnCount;
    this.commitLogIntervals = commitLogIntervals;
    this.minTimestamp = minTimestamp;
    this.maxTimestamp = maxTimestamp;
    this.minLocalDeletionTime = minLocalDeletionTime;
    this.maxLocalDeletionTime = maxLocalDeletionTime;
    this.minTTL = minTTL;
    this.maxTTL = maxTTL;
    this.compressionRatio = compressionRatio;
    this.estimatedTombstoneDropTime = estimatedTombstoneDropTime;
    this.sstableLevel = sstableLevel;
    this.minClusteringValues = minClusteringValues;
    this.maxClusteringValues = maxClusteringValues;
    this.hasLegacyCounterShards = hasLegacyCounterShards;
    this.repairedAt = repairedAt;
    this.totalColumnsSet = totalColumnsSet;
    this.totalRows = totalRows;
}
 

import org.apache.cassandra.utils.StreamingHistogram; //导入依赖的package包/类
public StatsMetadata(EstimatedHistogram estimatedRowSize,
                     EstimatedHistogram estimatedColumnCount,
                     ReplayPosition replayPosition,
                     long minTimestamp,
                     long maxTimestamp,
                     int maxLocalDeletionTime,
                     double compressionRatio,
                     StreamingHistogram estimatedTombstoneDropTime,
                     int sstableLevel,
                     List<ByteBuffer> minColumnNames,
                     List<ByteBuffer> maxColumnNames,
                     boolean hasLegacyCounterShards,
                     long repairedAt)
{
    this.estimatedRowSize = estimatedRowSize;
    this.estimatedColumnCount = estimatedColumnCount;
    this.replayPosition = replayPosition;
    this.minTimestamp = minTimestamp;
    this.maxTimestamp = maxTimestamp;
    this.maxLocalDeletionTime = maxLocalDeletionTime;
    this.compressionRatio = compressionRatio;
    this.estimatedTombstoneDropTime = estimatedTombstoneDropTime;
    this.sstableLevel = sstableLevel;
    this.minColumnNames = minColumnNames;
    this.maxColumnNames = maxColumnNames;
    this.hasLegacyCounterShards = hasLegacyCounterShards;
    this.repairedAt = repairedAt;
}
 

import org.apache.cassandra.utils.StreamingHistogram; //导入依赖的package包/类
private SSTableMetadata(EstimatedHistogram rowSizes,
                        EstimatedHistogram columnCounts,
                        ReplayPosition replayPosition,
                        long minTimestamp,
                        long maxTimestamp,
                        int maxLocalDeletionTime,
                        double bloomFilterFPChance,
                        double compressionRatio,
                        String partitioner,
                        StreamingHistogram estimatedTombstoneDropTime,
                        int sstableLevel,
                        List<ByteBuffer> minColumnNames,
                        List<ByteBuffer> maxColumnNames)
{
    this.estimatedRowSize = rowSizes;
    this.estimatedColumnCount = columnCounts;
    this.replayPosition = replayPosition;
    this.minTimestamp = minTimestamp;
    this.maxTimestamp = maxTimestamp;
    this.maxLocalDeletionTime = maxLocalDeletionTime;
    this.bloomFilterFPChance = bloomFilterFPChance;
    this.compressionRatio = compressionRatio;
    this.partitioner = partitioner;
    this.estimatedTombstoneDropTime = estimatedTombstoneDropTime;
    this.sstableLevel = sstableLevel;
    this.minColumnNames = minColumnNames;
    this.maxColumnNames = maxColumnNames;
}
 

import org.apache.cassandra.utils.StreamingHistogram; //导入依赖的package包/类
public ColumnStats(int columnCount, long minTimestamp, long maxTimestamp, int maxLocalDeletionTime, StreamingHistogram tombstoneHistogram, List<ByteBuffer> minColumnNames, List<ByteBuffer> maxColumnNames)
{
    this.minTimestamp = minTimestamp;
    this.maxTimestamp = maxTimestamp;
    this.maxLocalDeletionTime = maxLocalDeletionTime;
    this.columnCount = columnCount;
    this.tombstoneHistogram = tombstoneHistogram;
    this.minColumnNames = minColumnNames;
    this.maxColumnNames = maxColumnNames;
}
 

import org.apache.cassandra.utils.StreamingHistogram; //导入依赖的package包/类
public StatsMetadata(EstimatedHistogram estimatedPartitionSize,
                     EstimatedHistogram estimatedColumnCount,
                     IntervalSet<ReplayPosition> commitLogIntervals,
                     long minTimestamp,
                     long maxTimestamp,
                     int minLocalDeletionTime,
                     int maxLocalDeletionTime,
                     int minTTL,
                     int maxTTL,
                     double compressionRatio,
                     StreamingHistogram estimatedTombstoneDropTime,
                     int sstableLevel,
                     List<ByteBuffer> minClusteringValues,
                     List<ByteBuffer> maxClusteringValues,
                     boolean hasLegacyCounterShards,
                     long repairedAt,
                     long totalColumnsSet,
                     long totalRows)
{
    this.estimatedPartitionSize = estimatedPartitionSize;
    this.estimatedColumnCount = estimatedColumnCount;
    this.commitLogIntervals = commitLogIntervals;
    this.minTimestamp = minTimestamp;
    this.maxTimestamp = maxTimestamp;
    this.minLocalDeletionTime = minLocalDeletionTime;
    this.maxLocalDeletionTime = maxLocalDeletionTime;
    this.minTTL = minTTL;
    this.maxTTL = maxTTL;
    this.compressionRatio = compressionRatio;
    this.estimatedTombstoneDropTime = estimatedTombstoneDropTime;
    this.sstableLevel = sstableLevel;
    this.minClusteringValues = minClusteringValues;
    this.maxClusteringValues = maxClusteringValues;
    this.hasLegacyCounterShards = hasLegacyCounterShards;
    this.repairedAt = repairedAt;
    this.totalColumnsSet = totalColumnsSet;
    this.totalRows = totalRows;
}
 

import org.apache.cassandra.utils.StreamingHistogram; //导入依赖的package包/类
public LazilyCompactedRow(CompactionController controller, List<? extends ICountableColumnIterator> rows)
{
    super(rows.get(0).getKey());
    this.rows = rows;
    this.controller = controller;
    indexer = controller.cfs.indexManager.updaterFor(key);

    long maxDelTimestamp = Long.MIN_VALUE;
    for (OnDiskAtomIterator row : rows)
    {
        ColumnFamily cf = row.getColumnFamily();
        maxDelTimestamp = Math.max(maxDelTimestamp, cf.deletionInfo().maxTimestamp());

        if (emptyColumnFamily == null)
            emptyColumnFamily = cf;
        else
            emptyColumnFamily.delete(cf);
    }
    this.shouldPurge = controller.shouldPurge(key, maxDelTimestamp);

    try
    {
        indexAndWrite(null);
    }
    catch (IOException e)
    {
        throw new RuntimeException(e);
    }
    // reach into the reducer used during iteration to get column count, size, max column timestamp
    // (however, if there are zero columns, iterator() will not be called by ColumnIndexer and reducer will be null)
    columnStats = new ColumnStats(reducer == null ? 0 : reducer.columns, 
                                  reducer == null ? Long.MAX_VALUE : reducer.minTimestampSeen, 
                                  reducer == null ? maxDelTimestamp : Math.max(maxDelTimestamp, reducer.maxTimestampSeen),
                                  reducer == null ? new StreamingHistogram(SSTable.TOMBSTONE_HISTOGRAM_BIN_SIZE) : reducer.tombstones
    );
    columnSerializedSize = reducer == null ? 0 : reducer.serializedSize;
    reducer = null;
}
 

import org.apache.cassandra.utils.StreamingHistogram; //导入依赖的package包/类
public Pair<SSTableMetadata, Set<Integer>> deserialize(DataInputStream dis, Descriptor desc) throws IOException
{
    EstimatedHistogram rowSizes = EstimatedHistogram.serializer.deserialize(dis);
    EstimatedHistogram columnCounts = EstimatedHistogram.serializer.deserialize(dis);
    ReplayPosition replayPosition = desc.version.metadataIncludesReplayPosition
                                  ? ReplayPosition.serializer.deserialize(dis)
                                  : ReplayPosition.NONE;
    if (!desc.version.metadataIncludesModernReplayPosition)
    {
        // replay position may be "from the future" thanks to older versions generating them with nanotime.
        // make sure we don't omit replaying something that we should.  see CASSANDRA-4782
        replayPosition = ReplayPosition.NONE;
    }
    long minTimestamp = desc.version.tracksMinTimestamp ? dis.readLong() : Long.MIN_VALUE;
    long maxTimestamp = desc.version.containsTimestamp() ? dis.readLong() : Long.MAX_VALUE;
    if (!desc.version.tracksMaxTimestamp) // see javadoc to Descriptor.containsTimestamp
        maxTimestamp = Long.MAX_VALUE;
    double compressionRatio = desc.version.hasCompressionRatio
                            ? dis.readDouble()
                            : NO_COMPRESSION_RATIO;
    String partitioner = desc.version.hasPartitioner ? dis.readUTF() : null;
    int nbAncestors = desc.version.hasAncestors ? dis.readInt() : 0;
    Set<Integer> ancestors = new HashSet<Integer>(nbAncestors);
    for (int i = 0; i < nbAncestors; i++)
        ancestors.add(dis.readInt());
    StreamingHistogram tombstoneHistogram = desc.version.tracksTombstones
                                           ? StreamingHistogram.serializer.deserialize(dis)
                                           : defaultTombstoneDropTimeHistogram();
    return Pair.create(new SSTableMetadata(rowSizes,
                                           columnCounts,
                                           replayPosition,
                                           minTimestamp,
                                           maxTimestamp,
                                           compressionRatio,
                                           partitioner,
                                           tombstoneHistogram),
                                           ancestors);
}
 

import org.apache.cassandra.utils.StreamingHistogram; //导入依赖的package包/类
public ColumnStats(int columnCount, long minTimestamp, long maxTimestamp, StreamingHistogram tombstoneHistogram)
{
    this.minTimestamp = minTimestamp;
    this.maxTimestamp = maxTimestamp;
    this.columnCount = columnCount;
    this.tombstoneHistogram = tombstoneHistogram;
}
 

import org.apache.cassandra.utils.StreamingHistogram; //导入依赖的package包/类
static StreamingHistogram defaultTombstoneDropTimeHistogram()
{
    return new StreamingHistogram(SSTable.TOMBSTONE_HISTOGRAM_BIN_SIZE);
}
 

import org.apache.cassandra.utils.StreamingHistogram; //导入依赖的package包/类
public MetadataCollector mergeTombstoneHistogram(StreamingHistogram histogram)
{
    estimatedTombstoneDropTime.merge(histogram);
    return this;
}
 

import org.apache.cassandra.utils.StreamingHistogram; //导入依赖的package包/类
public StatsMetadata deserialize(Version version, DataInputPlus in) throws IOException
{
    EstimatedHistogram partitionSizes = EstimatedHistogram.serializer.deserialize(in);
    EstimatedHistogram columnCounts = EstimatedHistogram.serializer.deserialize(in);
    CommitLogPosition commitLogLowerBound = CommitLogPosition.NONE, commitLogUpperBound;
    commitLogUpperBound = CommitLogPosition.serializer.deserialize(in);
    long minTimestamp = in.readLong();
    long maxTimestamp = in.readLong();
    // We use MAX_VALUE as that's the default value for "no deletion time"
    int minLocalDeletionTime = version.storeRows() ? in.readInt() : Integer.MAX_VALUE;
    int maxLocalDeletionTime = in.readInt();
    int minTTL = version.storeRows() ? in.readInt() : 0;
    int maxTTL = version.storeRows() ? in.readInt() : Integer.MAX_VALUE;
    double compressionRatio = in.readDouble();
    StreamingHistogram tombstoneHistogram = StreamingHistogram.serializer.deserialize(in);
    int sstableLevel = in.readInt();
    long repairedAt = 0;
    if (version.hasRepairedAt())
        repairedAt = in.readLong();

    int colCount = in.readInt();
    List<ByteBuffer> minClusteringValues = new ArrayList<>(colCount);
    for (int i = 0; i < colCount; i++)
        minClusteringValues.add(ByteBufferUtil.readWithShortLength(in));

    colCount = in.readInt();
    List<ByteBuffer> maxClusteringValues = new ArrayList<>(colCount);
    for (int i = 0; i < colCount; i++)
        maxClusteringValues.add(ByteBufferUtil.readWithShortLength(in));

    boolean hasLegacyCounterShards = true;
    if (version.tracksLegacyCounterShards())
        hasLegacyCounterShards = in.readBoolean();

    long totalColumnsSet = version.storeRows() ? in.readLong() : -1L;
    long totalRows = version.storeRows() ? in.readLong() : -1L;

    if (version.hasCommitLogLowerBound())
        commitLogLowerBound = CommitLogPosition.serializer.deserialize(in);
    IntervalSet<CommitLogPosition> commitLogIntervals;
    if (version.hasCommitLogIntervals())
        commitLogIntervals = commitLogPositionSetSerializer.deserialize(in);
    else
        commitLogIntervals = new IntervalSet<CommitLogPosition>(commitLogLowerBound, commitLogUpperBound);

    return new StatsMetadata(partitionSizes,
                             columnCounts,
                             commitLogIntervals,
                             minTimestamp,
                             maxTimestamp,
                             minLocalDeletionTime,
                             maxLocalDeletionTime,
                             minTTL,
                             maxTTL,
                             compressionRatio,
                             tombstoneHistogram,
                             sstableLevel,
                             minClusteringValues,
                             maxClusteringValues,
                             hasLegacyCounterShards,
                             repairedAt,
                             totalColumnsSet,
                             totalRows);
}
 

import org.apache.cassandra.utils.StreamingHistogram; //导入依赖的package包/类
/**
 * Legacy serializer deserialize all components no matter what types are specified.
 */
@Override
public Map<MetadataType, MetadataComponent> deserialize(Descriptor descriptor, EnumSet<MetadataType> types) throws IOException
{
    Map<MetadataType, MetadataComponent> components = Maps.newHashMap();

    File statsFile = new File(descriptor.filenameFor(Component.STATS));
    if (!statsFile.exists() && types.contains(MetadataType.STATS))
    {
        components.put(MetadataType.STATS, MetadataCollector.defaultStatsMetadata());
    }
    else
    {
        try (DataInputStream in = new DataInputStream(new BufferedInputStream(new FileInputStream(statsFile))))
        {
            EstimatedHistogram rowSizes = EstimatedHistogram.serializer.deserialize(in);
            EstimatedHistogram columnCounts = EstimatedHistogram.serializer.deserialize(in);
            ReplayPosition replayPosition = ReplayPosition.serializer.deserialize(in);
            long minTimestamp = in.readLong();
            long maxTimestamp = in.readLong();
            int maxLocalDeletionTime = in.readInt();
            double bloomFilterFPChance = in.readDouble();
            double compressionRatio = in.readDouble();
            String partitioner = in.readUTF();
            int nbAncestors = in.readInt();
            Set<Integer> ancestors = new HashSet<>(nbAncestors);
            for (int i = 0; i < nbAncestors; i++)
                ancestors.add(in.readInt());
            StreamingHistogram tombstoneHistogram = StreamingHistogram.serializer.deserialize(in);
            int sstableLevel = 0;
            if (in.available() > 0)
                sstableLevel = in.readInt();

            int colCount = in.readInt();
            List<ByteBuffer> minColumnNames = new ArrayList<>(colCount);
            for (int i = 0; i < colCount; i++)
                minColumnNames.add(ByteBufferUtil.readWithShortLength(in));

            colCount = in.readInt();
            List<ByteBuffer> maxColumnNames = new ArrayList<>(colCount);
            for (int i = 0; i < colCount; i++)
                maxColumnNames.add(ByteBufferUtil.readWithShortLength(in));

            if (types.contains(MetadataType.VALIDATION))
                components.put(MetadataType.VALIDATION,
                               new ValidationMetadata(partitioner, bloomFilterFPChance));
            if (types.contains(MetadataType.STATS))
                components.put(MetadataType.STATS,
                               new StatsMetadata(rowSizes,
                                                 columnCounts,
                                                 replayPosition,
                                                 minTimestamp,
                                                 maxTimestamp,
                                                 maxLocalDeletionTime,
                                                 compressionRatio,
                                                 tombstoneHistogram,
                                                 sstableLevel,
                                                 minColumnNames,
                                                 maxColumnNames,
                                                 true,
                                                 ActiveRepairService.UNREPAIRED_SSTABLE));
            if (types.contains(MetadataType.COMPACTION))
                components.put(MetadataType.COMPACTION,
                               new CompactionMetadata(ancestors, null));
        }
    }
    return components;
}
 

import org.apache.cassandra.utils.StreamingHistogram; //导入依赖的package包/类
public StatsMetadata deserialize(Descriptor.Version version, DataInput in) throws IOException
{
    EstimatedHistogram rowSizes = EstimatedHistogram.serializer.deserialize(in);
    EstimatedHistogram columnCounts = EstimatedHistogram.serializer.deserialize(in);
    ReplayPosition replayPosition = ReplayPosition.serializer.deserialize(in);
    long minTimestamp = in.readLong();
    long maxTimestamp = in.readLong();
    int maxLocalDeletionTime = in.readInt();
    double compressionRatio = in.readDouble();
    StreamingHistogram tombstoneHistogram = StreamingHistogram.serializer.deserialize(in);
    int sstableLevel = in.readInt();
    long repairedAt = 0;
    if (version.hasRepairedAt)
        repairedAt = in.readLong();

    int colCount = in.readInt();
    List<ByteBuffer> minColumnNames = new ArrayList<>(colCount);
    for (int i = 0; i < colCount; i++)
        minColumnNames.add(ByteBufferUtil.readWithShortLength(in));

    colCount = in.readInt();
    List<ByteBuffer> maxColumnNames = new ArrayList<>(colCount);
    for (int i = 0; i < colCount; i++)
        maxColumnNames.add(ByteBufferUtil.readWithShortLength(in));

    boolean hasLegacyCounterShards = true;
    if (version.tracksLegacyCounterShards)
        hasLegacyCounterShards = in.readBoolean();

    return new StatsMetadata(rowSizes,
                             columnCounts,
                             replayPosition,
                             minTimestamp,
                             maxTimestamp,
                             maxLocalDeletionTime,
                             compressionRatio,
                             tombstoneHistogram,
                             sstableLevel,
                             minColumnNames,
                             maxColumnNames,
                             hasLegacyCounterShards,
                             repairedAt);
}
 

import org.apache.cassandra.utils.StreamingHistogram; //导入依赖的package包/类
/**
 * @throws IOException if a read from the DataInput fails
 * @throws FSWriteError if a write to the dataFile fails
 */
public long appendFromStream(DecoratedKey key, CFMetaData metadata, DataInput in, Descriptor.Version version) throws IOException
{
    long currentPosition = beforeAppend(key);

    // deserialize each column to obtain maxTimestamp and immediately serialize it.
    long minTimestamp = Long.MAX_VALUE;
    long maxTimestamp = Long.MIN_VALUE;
    int maxLocalDeletionTime = Integer.MIN_VALUE;
    List<ByteBuffer> minColumnNames = Collections.emptyList();
    List<ByteBuffer> maxColumnNames = Collections.emptyList();
    StreamingHistogram tombstones = new StreamingHistogram(TOMBSTONE_HISTOGRAM_BIN_SIZE);
    ColumnFamily cf = ArrayBackedSortedColumns.factory.create(metadata);

    cf.delete(DeletionTime.serializer.deserialize(in));

    ColumnIndex.Builder columnIndexer = new ColumnIndex.Builder(cf, key.key, dataFile.stream);
    int columnCount = Integer.MAX_VALUE;
    if (version.hasRowSizeAndColumnCount)
    {
        // skip row size
        FileUtils.skipBytesFully(in, 8);
        columnCount = in.readInt();
    }
    Iterator<OnDiskAtom> iter = metadata.getOnDiskIterator(in, columnCount, ColumnSerializer.Flag.PRESERVE_SIZE, Integer.MIN_VALUE, version);
    try
    {
        while (iter.hasNext())
        {
            // deserialize column with PRESERVE_SIZE because we've written the dataSize based on the
            // data size received, so we must reserialize the exact same data
            OnDiskAtom atom = iter.next();
            if (atom == null)
                break;
            if (atom instanceof CounterColumn)
                atom = ((CounterColumn) atom).markDeltaToBeCleared();

            int deletionTime = atom.getLocalDeletionTime();
            if (deletionTime < Integer.MAX_VALUE)
            {
                tombstones.update(deletionTime);
            }
            minTimestamp = Math.min(minTimestamp, atom.minTimestamp());
            maxTimestamp = Math.max(maxTimestamp, atom.maxTimestamp());
            minColumnNames = ColumnNameHelper.minComponents(minColumnNames, atom.name(), metadata.comparator);
            maxColumnNames = ColumnNameHelper.maxComponents(maxColumnNames, atom.name(), metadata.comparator);
            maxLocalDeletionTime = Math.max(maxLocalDeletionTime, atom.getLocalDeletionTime());

            columnIndexer.add(atom); // This write the atom on disk too
        }

        columnIndexer.finish();
        dataFile.stream.writeShort(END_OF_ROW);
    }
    catch (IOException e)
    {
        throw new FSWriteError(e, dataFile.getPath());
    }

    sstableMetadataCollector.updateMinTimestamp(minTimestamp);
    sstableMetadataCollector.updateMaxTimestamp(maxTimestamp);
    sstableMetadataCollector.updateMaxLocalDeletionTime(maxLocalDeletionTime);
    sstableMetadataCollector.addRowSize(dataFile.getFilePointer() - currentPosition);
    sstableMetadataCollector.addColumnCount(columnIndexer.writtenAtomCount());
    sstableMetadataCollector.mergeTombstoneHistogram(tombstones);
    sstableMetadataCollector.updateMinColumnNames(minColumnNames);
    sstableMetadataCollector.updateMaxColumnNames(maxColumnNames);
    afterAppend(key, currentPosition, RowIndexEntry.create(currentPosition, cf.deletionInfo().getTopLevelDeletion(), columnIndexer.build()));
    return currentPosition;
}
 

import org.apache.cassandra.utils.StreamingHistogram; //导入依赖的package包/类
public void mergeTombstoneHistogram(StreamingHistogram histogram)
{
    estimatedTombstoneDropTime.merge(histogram);
}
 

import org.apache.cassandra.utils.StreamingHistogram; //导入依赖的package包/类
public Pair<SSTableMetadata, Set<Integer>> deserialize(DataInputStream in, Descriptor desc, boolean loadSSTableLevel) throws IOException
{
    EstimatedHistogram rowSizes = EstimatedHistogram.serializer.deserialize(in);
    EstimatedHistogram columnCounts = EstimatedHistogram.serializer.deserialize(in);
    ReplayPosition replayPosition = ReplayPosition.serializer.deserialize(in);
    long minTimestamp = in.readLong();
    long maxTimestamp = in.readLong();
    int maxLocalDeletionTime = desc.version.tracksMaxLocalDeletionTime ? in.readInt() : Integer.MAX_VALUE;
    double bloomFilterFPChance = desc.version.hasBloomFilterFPChance ? in.readDouble() : NO_BLOOM_FLITER_FP_CHANCE;
    double compressionRatio = in.readDouble();
    String partitioner = in.readUTF();
    int nbAncestors = in.readInt();
    Set<Integer> ancestors = new HashSet<Integer>(nbAncestors);
    for (int i = 0; i < nbAncestors; i++)
        ancestors.add(in.readInt());
    StreamingHistogram tombstoneHistogram = StreamingHistogram.serializer.deserialize(in);
    int sstableLevel = 0;

    if (loadSSTableLevel && in.available() > 0)
        sstableLevel = in.readInt();

    List<ByteBuffer> minColumnNames;
    List<ByteBuffer> maxColumnNames;
    if (desc.version.tracksMaxMinColumnNames)
    {
        int colCount = in.readInt();
        minColumnNames = new ArrayList<ByteBuffer>(colCount);
        for (int i = 0; i < colCount; i++)
        {
            minColumnNames.add(ByteBufferUtil.readWithShortLength(in));
        }
        colCount = in.readInt();
        maxColumnNames = new ArrayList<ByteBuffer>(colCount);
        for (int i = 0; i < colCount; i++)
        {
            maxColumnNames.add(ByteBufferUtil.readWithShortLength(in));
        }
    }
    else
    {
        minColumnNames = Collections.emptyList();
        maxColumnNames = Collections.emptyList();
    }
    return Pair.create(new SSTableMetadata(rowSizes,
                               columnCounts,
                               replayPosition,
                               minTimestamp,
                               maxTimestamp,
                               maxLocalDeletionTime,
                               bloomFilterFPChance,
                               compressionRatio,
                               partitioner,
                               tombstoneHistogram,
                               sstableLevel,
                               minColumnNames,
                               maxColumnNames), ancestors);
}