import org.apache.cassandra.db.compaction.LeveledManifest; //导入依赖的package包/类
@Override
@SuppressWarnings("resource")
public boolean realAppend(UnfilteredRowIterator partition)
{
    long posBefore = sstableWriter.currentWriter().getOnDiskFilePointer();
    RowIndexEntry rie = sstableWriter.append(partition);
    totalWrittenInLevel += sstableWriter.currentWriter().getOnDiskFilePointer() - posBefore;
    partitionsWritten++;
    if (sstableWriter.currentWriter().getOnDiskFilePointer() > maxSSTableSize)
    {
        if (totalWrittenInLevel > LeveledManifest.maxBytesForLevel(currentLevel, maxSSTableSize))
        {
            totalWrittenInLevel = 0;
            currentLevel++;
        }

        averageEstimatedKeysPerSSTable = Math.round(((double) averageEstimatedKeysPerSSTable * sstablesWritten + partitionsWritten) / (sstablesWritten + 1));
        switchCompactionLocation(getWriteDirectory(expectedWriteSize));
        partitionsWritten = 0;
        sstablesWritten++;
    }
    return rie != null;

}
 

import org.apache.cassandra.db.compaction.LeveledManifest; //导入依赖的package包/类
private static void createTestFiles() throws IOException
{
    for (String cf : CFS)
    {
        List<File> fs = new ArrayList<File>();
        files.put(cf, fs);
        File dir = cfDir(cf);
        dir.mkdirs();

        createFakeSSTable(dir, cf, 1, false, fs);
        createFakeSSTable(dir, cf, 2, true, fs);
        // leveled manifest
        new File(dir, cf + LeveledManifest.EXTENSION).createNewFile();

        File backupDir = new File(dir, Directories.BACKUPS_SUBDIR);
        backupDir.mkdir();
        createFakeSSTable(backupDir, cf, 1, false, fs);

        File snapshotDir = new File(dir, Directories.SNAPSHOT_SUBDIR + File.separator + "42");
        snapshotDir.mkdirs();
        createFakeSSTable(snapshotDir, cf, 1, false, fs);
    }
}
 

import org.apache.cassandra.db.compaction.LeveledManifest; //导入依赖的package包/类
private static void checkManifest(AbstractCompactionStrategy strategy, ColumnFamilyStore cfs, Collection<SSTableReader> sstables)
{
    WrappingCompactionStrategy wrappingStrategy = (WrappingCompactionStrategy)strategy;
    int maxSizeInMB = (int)((cfs.getCompactionStrategy().getMaxSSTableBytes()) / (1024L * 1024L));
    if (wrappingStrategy.getWrappedStrategies().size() == 2 && wrappingStrategy.getWrappedStrategies().iterator().next() instanceof LeveledCompactionStrategy)
    {
        System.out.println("Checking leveled manifest");
        Predicate<SSTableReader> repairedPredicate = new Predicate<SSTableReader>()
        {
            @Override
            public boolean apply(SSTableReader sstable)
            {
                return sstable.isRepaired();
            }
        };

        List<SSTableReader> repaired = Lists.newArrayList(Iterables.filter(sstables, repairedPredicate));
        List<SSTableReader> unRepaired = Lists.newArrayList(Iterables.filter(sstables, Predicates.not(repairedPredicate)));

        LeveledManifest repairedManifest = LeveledManifest.create(cfs, maxSizeInMB, repaired);
        for (int i = 1; i < repairedManifest.getLevelCount(); i++)
        {
            repairedManifest.repairOverlappingSSTables(i);
        }
        LeveledManifest unRepairedManifest = LeveledManifest.create(cfs, maxSizeInMB, unRepaired);
        for (int i = 1; i < unRepairedManifest.getLevelCount(); i++)
        {
            unRepairedManifest.repairOverlappingSSTables(i);
        }
    }
}
 

import org.apache.cassandra.db.compaction.LeveledManifest; //导入依赖的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);
    }

    String keyspace = args[1];
    String columnfamily = args[2];
    Directories directories = Directories.create(keyspace, columnfamily);
    boolean foundSSTable = false;
    for (Map.Entry<Descriptor, Set<Component>> sstable : directories.sstableLister().list().entrySet())
    {
        if (sstable.getValue().contains(Component.STATS))
        {
            foundSSTable = true;
            Descriptor descriptor = sstable.getKey();
            Pair<SSTableMetadata, Set<Integer>> metadata = SSTableMetadata.serializer.deserialize(descriptor);
            out.println("Changing level from " + metadata.left.sstableLevel + " to 0 on " + descriptor.filenameFor(Component.DATA));
            LeveledManifest.mutateLevel(metadata, descriptor, descriptor.filenameFor(Component.STATS), 0);
        }
    }

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

import org.apache.cassandra.db.compaction.LeveledManifest; //导入依赖的package包/类
public File tryGetLeveledManifest()
{
    for (File dir : sstableDirectories)
    {
        File manifestFile = new File(dir, cfname + LeveledManifest.EXTENSION);
        if (manifestFile.exists())
        {
            logger.debug("Found manifest at {}", manifestFile);
            return manifestFile;
        }
    }
    logger.debug("No level manifest found");
    return null;
}
 

import org.apache.cassandra.db.compaction.LeveledManifest; //导入依赖的package包/类
public File getOrCreateLeveledManifest()
{
    File manifestFile = tryGetLeveledManifest();
    if (manifestFile == null)
        manifestFile = new File(sstableDirectories[0], cfname + LeveledManifest.EXTENSION);
    return manifestFile;
}
 

import org.apache.cassandra.db.compaction.LeveledManifest; //导入依赖的package包/类
private static void migrateFile(File file, File ksDir, String additionalPath)
{
    if (file.isDirectory())
        return;

    try
    {
        String name = file.getName();
        boolean isManifest = name.endsWith(LeveledManifest.EXTENSION);
        int separatorIndex = name.indexOf(Component.separator);

        if (isManifest || (separatorIndex >= 0))
        {
            String cfname = isManifest
                          ? getCfNameFromManifest(name)
                          : name.substring(0, separatorIndex);

            int idx = cfname.indexOf(SECONDARY_INDEX_NAME_SEPARATOR); // idx > 0 => secondary index
            String dirname = idx > 0 ? cfname.substring(0, idx) : cfname;
            File destDir = getOrCreate(ksDir, dirname, additionalPath);

            File destFile = new File(destDir, isManifest ? name : ksDir.getName() + Component.separator + name);
            logger.debug(String.format("[upgrade to 1.1] Moving %s to %s", file, destFile));
            FileUtils.renameWithConfirm(file, destFile);
        }
        else
        {
            logger.warn("Found unrecognized file {} while migrating sstables from pre 1.1 format, ignoring.", file);
        }
    }
    catch (Exception e)
    {
        throw new RuntimeException(String.format("Failed migrating file %s from pre 1.1 format.", file.getPath()), e);
    }
}
 

import org.apache.cassandra.db.compaction.LeveledManifest; //导入依赖的package包/类
private static String getCfNameFromManifest(String name)
{
    String withoutExt = name.substring(0, name.length() - LeveledManifest.EXTENSION.length());
    return withoutExt.endsWith("-old") || withoutExt.endsWith("-tmp")
            ? withoutExt.substring(0, withoutExt.length() - 4)
            : withoutExt;
}
 

import org.apache.cassandra.db.compaction.LeveledManifest; //导入依赖的package包/类
@Test
public void testLeveledManifestPath()
{
    for (String cf : CFS)
    {
        Directories directories = Directories.create(KS, cf);
        File manifest = new File(cfDir(cf), cf + LeveledManifest.EXTENSION);
        Assert.assertEquals(manifest, directories.tryGetLeveledManifest());
    }
}
 

import org.apache.cassandra.db.compaction.LeveledManifest; //导入依赖的package包/类
private static void checkManifest(LeveledManifest manifest)
{
    System.out.println(String.format("Checking leveled manifest"));
    for (int i = 1; i <= manifest.getLevelCount(); ++i)
        manifest.repairOverlappingSSTables(i);
}