import org.apache.cassandra.io.sstable.metadata.MetadataType; //导入依赖的package包/类
private static void writeMetadata(Descriptor desc, Map<MetadataType, MetadataComponent> components)
{
    SequentialWriter out = SequentialWriter.open(new File(desc.filenameFor(Component.STATS)));
    try
    {
        desc.getMetadataSerializer().serialize(components, out.stream);
    }
    catch (IOException e)
    {
        throw new FSWriteError(e, out.getPath());
    }
    finally
    {
        out.close();
    }
}
 

import org.apache.cassandra.io.sstable.metadata.MetadataType; //导入依赖的package包/类
private Pair<Descriptor, StatsMetadata> close(long repairedAt)
{

    // index and filter
    iwriter.close();
    // main data, close will truncate if necessary
    dataFile.close();
    dataFile.writeFullChecksum(descriptor);
    // write sstable statistics
    Map<MetadataType, MetadataComponent> metadataComponents = sstableMetadataCollector.finalizeMetadata(
                                                                                partitioner.getClass().getCanonicalName(),
                                                                                metadata.getBloomFilterFpChance(),
                                                                                repairedAt);
    writeMetadata(descriptor, metadataComponents);

    // save the table of components
    SSTable.appendTOC(descriptor, components);

    // remove the 'tmp' marker from all components
    return Pair.create(rename(descriptor, components), (StatsMetadata) metadataComponents.get(MetadataType.STATS));

}
 

import org.apache.cassandra.io.sstable.metadata.MetadataType; //导入依赖的package包/类
public Pair<Descriptor, StatsMetadata> close()
{
    // index and filter
    iwriter.close();
    // main data, close will truncate if necessary
    dataFile.close();
    // write sstable statistics
    Map<MetadataType, MetadataComponent> metadataComponents = sstableMetadataCollector.finalizeMetadata(
                                                                                partitioner.getClass().getCanonicalName(),
                                                                                metadata.getBloomFilterFpChance());
    writeMetadata(descriptor, metadataComponents);

    // save the table of components
    SSTable.appendTOC(descriptor, components);

    // remove the 'tmp' marker from all components
    return Pair.create(rename(descriptor, components), (StatsMetadata) metadataComponents.get(MetadataType.STATS));
}
 

import org.apache.cassandra.io.sstable.metadata.MetadataType; //导入依赖的package包/类
private static void writeMetadata(Descriptor desc, Map<MetadataType, MetadataComponent> components)
{
    SequentialWriter out = SequentialWriter.open(new File(desc.filenameFor(Component.STATS)), true);
    try
    {
        desc.getMetadataSerializer().serialize(components, out.stream);
    }
    catch (IOException e)
    {
        throw new FSWriteError(e, out.getPath());
    }
    finally
    {
        out.close();
    }
}
 

import org.apache.cassandra.io.sstable.metadata.MetadataType; //导入依赖的package包/类
private Pair<Descriptor, StatsMetadata> close(long repairedAt)
{
    dataFile.writeFullChecksum(descriptor);
    // index and filter
    iwriter.close();
    // main data, close will truncate if necessary
    dataFile.close();
    // write sstable statistics
    Map<MetadataType, MetadataComponent> metadataComponents = sstableMetadataCollector.finalizeMetadata(
                                                                                partitioner.getClass().getCanonicalName(),
                                                                                metadata.getBloomFilterFpChance(),
                                                                                repairedAt);
    writeMetadata(descriptor, metadataComponents);

    // save the table of components
    SSTable.appendTOC(descriptor, components);

    // remove the 'tmp' marker from all components
    return Pair.create(rename(descriptor, components), (StatsMetadata) metadataComponents.get(MetadataType.STATS));

}
 

import org.apache.cassandra.io.sstable.metadata.MetadataType; //导入依赖的package包/类
/**
 * Create a new instance of a partitioner defined in an SSTable Descriptor
 * @param desc Descriptor of an sstable
 * @return a new IPartitioner instance
 * @throws IOException
 */
public static IPartitioner newPartitioner(Descriptor desc) throws IOException
{
    EnumSet<MetadataType> types = EnumSet.of(MetadataType.VALIDATION, MetadataType.HEADER);
    Map<MetadataType, MetadataComponent> sstableMetadata = desc.getMetadataSerializer().deserialize(desc, types);
    ValidationMetadata validationMetadata = (ValidationMetadata) sstableMetadata.get(MetadataType.VALIDATION);
    SerializationHeader.Component header = (SerializationHeader.Component) sstableMetadata.get(MetadataType.HEADER);
    return newPartitioner(validationMetadata.partitioner, Optional.of(header.getKeyType()));
}
 

import org.apache.cassandra.io.sstable.metadata.MetadataType; //导入依赖的package包/类
private void writeMetadata(Descriptor desc, Map<MetadataType, MetadataComponent> components)
{
    try (SequentialWriter out = new SequentialWriter(desc.filenameFor(Component.STATS), writerOption, desc.getConfiguration()))
    {
        desc.getMetadataSerializer().serialize(components, out, desc.version);
        out.finish();
    }
    catch (IOException e)
    {
        throw new FSWriteError(e, desc.filenameFor(Component.STATS));
    }
}
 

import org.apache.cassandra.io.sstable.metadata.MetadataType; //导入依赖的package包/类
protected Map<MetadataType, MetadataComponent> finalizeMetadata()
{
    return metadataCollector.finalizeMetadata(getPartitioner().getClass().getCanonicalName(),
                                              metadata.params.bloomFilterFpChance,
                                              repairedAt,
                                              header);
}
 

import org.apache.cassandra.io.sstable.metadata.MetadataType; //导入依赖的package包/类
private Pair<Descriptor, StatsMetadata> close(FinishType type, long repairedAt)
{
    switch (type)
    {
        case EARLY: case CLOSE: case NORMAL:
        iwriter.close();
        dataFile.close();
        if (type == FinishType.CLOSE)
            iwriter.bf.close();
    }

    // write sstable statistics
    Map<MetadataType, MetadataComponent> metadataComponents ;
    metadataComponents = sstableMetadataCollector
                         .finalizeMetadata(partitioner.getClass().getCanonicalName(),
                                           metadata.getBloomFilterFpChance(),repairedAt);

    // remove the 'tmp' marker from all components
    Descriptor descriptor = this.descriptor;
    if (type.isFinal)
    {
        dataFile.writeFullChecksum(descriptor);
        writeMetadata(descriptor, metadataComponents);
        // save the table of components
        SSTable.appendTOC(descriptor, components);
        descriptor = rename(descriptor, components);
    }

    return Pair.create(descriptor, (StatsMetadata) metadataComponents.get(MetadataType.STATS));
}
 

import org.apache.cassandra.io.sstable.metadata.MetadataType; //导入依赖的package包/类
public Set<Integer> getAncestors()
{
    try
    {
        CompactionMetadata compactionMetadata = (CompactionMetadata) descriptor.getMetadataSerializer().deserialize(descriptor, MetadataType.COMPACTION);
        return compactionMetadata.ancestors;
    }
    catch (IOException e)
    {
        SSTableReader.logOpenException(descriptor, e);
        return Collections.emptySet();
    }
}
 

import org.apache.cassandra.io.sstable.metadata.MetadataType; //导入依赖的package包/类
private static void writeMetadata(Descriptor desc, Map<MetadataType, MetadataComponent> components)
{
    File file = new File(desc.filenameFor(Component.STATS));
    try (SequentialWriter out = SequentialWriter.open(file))
    {
        desc.getMetadataSerializer().serialize(components, out, desc.version);
        out.setDescriptor(desc).finish();
    }
    catch (IOException e)
    {
        throw new FSWriteError(e, file.getPath());
    }
}
 

import org.apache.cassandra.io.sstable.metadata.MetadataType; //导入依赖的package包/类
public MetadataType getType()
{
    return MetadataType.HEADER;
}
 

import org.apache.cassandra.io.sstable.metadata.MetadataType; //导入依赖的package包/类
protected StatsMetadata statsMetadata()
{
    return (StatsMetadata) finalizeMetadata().get(MetadataType.STATS);
}
 

import org.apache.cassandra.io.sstable.metadata.MetadataType; //导入依赖的package包/类
/**
 * @param args a list of sstables whose metadata we are changing
 */
public static void main(String[] args) throws IOException
{
    PrintStream out = System.out;
    if (args.length == 0)
    {
        out.println("This command should be run with Cassandra stopped!");
        out.println("Usage: sstablelevelreset <keyspace> <columnfamily>");
        System.exit(1);
    }

    if (!args[0].equals("--really-reset") || args.length != 3)
    {
        out.println("This command should be run with Cassandra stopped, otherwise you will get very strange behavior");
        out.println("Verify that Cassandra is not running and then execute the command like this:");
        out.println("Usage: sstablelevelreset --really-reset <keyspace> <columnfamily>");
        System.exit(1);
    }

    // TODO several daemon threads will run from here.
    // So we have to explicitly call System.exit.
    try
    {
        // load keyspace descriptions.
        DatabaseDescriptor.loadSchemas();

        String keyspaceName = args[1];
        String columnfamily = args[2];
        // validate columnfamily
        if (Schema.instance.getCFMetaData(keyspaceName, columnfamily) == null)
        {
            System.err.println("ColumnFamily not found: " + keyspaceName + "/" + columnfamily);
            System.exit(1);
        }

        Keyspace keyspace = Keyspace.openWithoutSSTables(keyspaceName);
        ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(columnfamily);
        boolean foundSSTable = false;
        for (Map.Entry<Descriptor, Set<Component>> sstable : cfs.directories.sstableLister().list().entrySet())
        {
            if (sstable.getValue().contains(Component.STATS))
            {
                foundSSTable = true;
                Descriptor descriptor = sstable.getKey();
                StatsMetadata metadata = (StatsMetadata) descriptor.getMetadataSerializer().deserialize(descriptor, MetadataType.STATS);
                if (metadata.sstableLevel > 0)
                {
                    out.println("Changing level from " + metadata.sstableLevel + " to 0 on " + descriptor.filenameFor(Component.DATA));
                    descriptor.getMetadataSerializer().mutateLevel(descriptor, 0);
                }
                else
                {
                    out.println("Skipped " + descriptor.filenameFor(Component.DATA) + " since it is already on level 0");
                }
            }
        }

        if (!foundSSTable)
        {
            out.println("Found no sstables, did you give the correct keyspace/columnfamily?");
        }
    }
    catch (Throwable t)
    {
        JVMStabilityInspector.inspectThrowable(t);
        t.printStackTrace();
        System.exit(1);
    }
    System.exit(0);
}
 

import org.apache.cassandra.io.sstable.metadata.MetadataType; //导入依赖的package包/类
/**
 * @param args a list of sstables whose metadata we are changing
 */
public static void main(String[] args)
{
    PrintStream out = System.out;
    if (args.length == 0)
    {
        out.println("This command should be run with Cassandra stopped!");
        out.println("Usage: sstablelevelreset <keyspace> <table>");
        System.exit(1);
    }

    if (!args[0].equals("--really-reset") || args.length != 3)
    {
        out.println("This command should be run with Cassandra stopped, otherwise you will get very strange behavior");
        out.println("Verify that Cassandra is not running and then execute the command like this:");
        out.println("Usage: sstablelevelreset --really-reset <keyspace> <table>");
        System.exit(1);
    }

    Util.initDatabaseDescriptor();

    // TODO several daemon threads will run from here.
    // So we have to explicitly call System.exit.
    try
    {
        // load keyspace descriptions.
        Schema.instance.loadFromDiskForTool();

        String keyspaceName = args[1];
        String columnfamily = args[2];
        // validate columnfamily
        if (Schema.instance.getCFMetaData(keyspaceName, columnfamily) == null)
        {
            System.err.println("ColumnFamily not found: " + keyspaceName + "/" + columnfamily);
            System.exit(1);
        }

        Keyspace keyspace = Keyspace.openWithoutSSTables(keyspaceName);
        ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(columnfamily);
        boolean foundSSTable = false;
        for (Map.Entry<Descriptor, Set<Component>> sstable : cfs.getDirectories().sstableLister(Directories.OnTxnErr.THROW).list().entrySet())
        {
            if (sstable.getValue().contains(Component.STATS))
            {
                foundSSTable = true;
                Descriptor descriptor = sstable.getKey();
                StatsMetadata metadata = (StatsMetadata) descriptor.getMetadataSerializer().deserialize(descriptor, MetadataType.STATS);
                if (metadata.sstableLevel > 0)
                {
                    out.println("Changing level from " + metadata.sstableLevel + " to 0 on " + descriptor.filenameFor(Component.DATA));
                    descriptor.getMetadataSerializer().mutateLevel(descriptor, 0);
                }
                else
                {
                    out.println("Skipped " + descriptor.filenameFor(Component.DATA) + " since it is already on level 0");
                }
            }
        }

        if (!foundSSTable)
        {
            out.println("Found no sstables, did you give the correct keyspace/table?");
        }
    }
    catch (Throwable t)
    {
        JVMStabilityInspector.inspectThrowable(t);
        t.printStackTrace();
        System.exit(1);
    }
    System.exit(0);
}
 

import org.apache.cassandra.io.sstable.metadata.MetadataType; //导入依赖的package包/类
/**
 * @param args a list of sstables whose metadata we are changing
 */
public static void main(String[] args)
{
    PrintStream out = System.out;
    if (args.length == 0)
    {
        out.println("This command should be run with Cassandra stopped!");
        out.println("Usage: sstablelevelreset <keyspace> <table>");
        System.exit(1);
    }

    if (!args[0].equals("--really-reset") || args.length != 3)
    {
        out.println("This command should be run with Cassandra stopped, otherwise you will get very strange behavior");
        out.println("Verify that Cassandra is not running and then execute the command like this:");
        out.println("Usage: sstablelevelreset --really-reset <keyspace> <table>");
        System.exit(1);
    }

    // TODO several daemon threads will run from here.
    // So we have to explicitly call System.exit.
    try
    {
        // load keyspace descriptions.
        DatabaseDescriptor.loadSchemas();

        String keyspaceName = args[1];
        String columnfamily = args[2];
        // validate columnfamily
        if (Schema.instance.getCFMetaData(keyspaceName, columnfamily) == null)
        {
            System.err.println("ColumnFamily not found: " + keyspaceName + "/" + columnfamily);
            System.exit(1);
        }

        Keyspace keyspace = Keyspace.openWithoutSSTables(keyspaceName);
        ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(columnfamily);
        boolean foundSSTable = false;
        for (Map.Entry<Descriptor, Set<Component>> sstable : cfs.directories.sstableLister().list().entrySet())
        {
            if (sstable.getValue().contains(Component.STATS))
            {
                foundSSTable = true;
                Descriptor descriptor = sstable.getKey();
                StatsMetadata metadata = (StatsMetadata) descriptor.getMetadataSerializer().deserialize(descriptor, MetadataType.STATS);
                if (metadata.sstableLevel > 0)
                {
                    out.println("Changing level from " + metadata.sstableLevel + " to 0 on " + descriptor.filenameFor(Component.DATA));
                    descriptor.getMetadataSerializer().mutateLevel(descriptor, 0);
                }
                else
                {
                    out.println("Skipped " + descriptor.filenameFor(Component.DATA) + " since it is already on level 0");
                }
            }
        }

        if (!foundSSTable)
        {
            out.println("Found no sstables, did you give the correct keyspace/table?");
        }
    }
    catch (Throwable e)
    {
        e.printStackTrace();
        System.exit(1);
    }
    System.exit(0);
}
 

import org.apache.cassandra.io.sstable.metadata.MetadataType; //导入依赖的package包/类
public SSTableReader openEarly(long maxDataAge)
{
    StatsMetadata sstableMetadata = (StatsMetadata) sstableMetadataCollector.finalizeMetadata(partitioner.getClass().getCanonicalName(),
                                              metadata.getBloomFilterFpChance(),
                                              repairedAt).get(MetadataType.STATS);

    // find the max (exclusive) readable key
    DecoratedKey exclusiveUpperBoundOfReadableIndex = iwriter.getMaxReadableKey(0);
    if (exclusiveUpperBoundOfReadableIndex == null)
        return null;

    // create temp links if they don't already exist
    Descriptor link = descriptor.asType(Descriptor.Type.TEMPLINK);
    if (!new File(link.filenameFor(Component.PRIMARY_INDEX)).exists())
    {
        FileUtils.createHardLink(new File(descriptor.filenameFor(Component.PRIMARY_INDEX)), new File(link.filenameFor(Component.PRIMARY_INDEX)));
        FileUtils.createHardLink(new File(descriptor.filenameFor(Component.DATA)), new File(link.filenameFor(Component.DATA)));
    }

    // open the reader early, giving it a FINAL descriptor type so that it is indistinguishable for other consumers
    SegmentedFile ifile = iwriter.builder.openEarly(link.filenameFor(Component.PRIMARY_INDEX));
    SegmentedFile dfile = dbuilder.openEarly(link.filenameFor(Component.DATA));
    SSTableReader sstable = SSTableReader.internalOpen(descriptor.asType(Descriptor.Type.FINAL),
                                                       components, metadata,
                                                       partitioner, ifile,
                                                       dfile, iwriter.summary.build(partitioner, exclusiveUpperBoundOfReadableIndex),
                                                       iwriter.bf, maxDataAge, sstableMetadata, true);

    // now it's open, find the ACTUAL last readable key (i.e. for which the data file has also been flushed)
    sstable.first = getMinimalKey(first);
    sstable.last = getMinimalKey(exclusiveUpperBoundOfReadableIndex);
    DecoratedKey inclusiveUpperBoundOfReadableData = iwriter.getMaxReadableKey(1);
    if (inclusiveUpperBoundOfReadableData == null)
        return null;
    int offset = 2;
    while (true)
    {
        RowIndexEntry indexEntry = sstable.getPosition(inclusiveUpperBoundOfReadableData, SSTableReader.Operator.GT);
        if (indexEntry != null && indexEntry.position <= dataFile.getLastFlushOffset())
            break;
        inclusiveUpperBoundOfReadableData = iwriter.getMaxReadableKey(offset++);
        if (inclusiveUpperBoundOfReadableData == null)
            return null;
    }
    sstable.last = getMinimalKey(inclusiveUpperBoundOfReadableData);
    return sstable;
}
 

import org.apache.cassandra.io.sstable.metadata.MetadataType; //导入依赖的package包/类
/**
 * @param args a list of sstables whose metadata we are changing
 */
public static void main(String[] args) throws IOException
{
    PrintStream out = System.out;
    if (args.length == 0)
    {
        out.println("This command should be run with Cassandra stopped!");
        out.println("Usage: sstablelevelreset <keyspace> <columnfamily>");
        System.exit(1);
    }

    if (!args[0].equals("--really-reset") || args.length != 3)
    {
        out.println("This command should be run with Cassandra stopped, otherwise you will get very strange behavior");
        out.println("Verify that Cassandra is not running and then execute the command like this:");
        out.println("Usage: sstablelevelreset --really-reset <keyspace> <columnfamily>");
        System.exit(1);
    }

    // load keyspace descriptions.
    DatabaseDescriptor.loadSchemas();

    String keyspaceName = args[1];
    String columnfamily = args[2];
    Keyspace keyspace = Keyspace.openWithoutSSTables(keyspaceName);
    ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(columnfamily);
    boolean foundSSTable = false;
    for (Map.Entry<Descriptor, Set<Component>> sstable : cfs.directories.sstableLister().list().entrySet())
    {
        if (sstable.getValue().contains(Component.STATS))
        {
            foundSSTable = true;
            Descriptor descriptor = sstable.getKey();
            StatsMetadata metadata = (StatsMetadata) descriptor.getMetadataSerializer().deserialize(descriptor, MetadataType.STATS);
            out.println("Changing level from " + metadata.sstableLevel + " to 0 on " + descriptor.filenameFor(Component.DATA));
            descriptor.getMetadataSerializer().mutateLevel(descriptor, 0);
        }
    }

    if (!foundSSTable)
    {
        out.println("Found no sstables, did you give the correct keyspace/columnfamily?");
    }
}
 

import org.apache.cassandra.io.sstable.metadata.MetadataType; //导入依赖的package包/类
/**
 * Reloads the sstable metadata from disk.
 *
 * Called after level is changed on sstable, for example if the sstable is dropped to L0
 *
 * Might be possible to remove in future versions
 *
 * @throws IOException
 */
public void reloadSSTableMetadata() throws IOException
{
    this.sstableMetadata = (StatsMetadata) descriptor.getMetadataSerializer().deserialize(descriptor, MetadataType.STATS);
}