import org.apache.cassandra.io.sstable.SSTableUtils; //导入依赖的package包/类
/**
 * Generates two sstables to be used to test migrating from a .json manifest to keeping the level in the sstable
 * metadata.
 *
 * Do this:
 * 1. remove @Ignore
 * 2. comment out the @Before and @After methods above
 * 3. run this method
 * 4. checkout trunk
 * 5. copy the .json file from the previous version to the current one
 *    (ie; test/data/migration-sstables/ic/Keyspace1/legacyleveled/legacyleveled.json)
 * 6. update LegacyLeveledManifestTest to use the new version.
 */
@Test
public void generateSSTable() throws IOException
{
    File legacySSTableDir = getLegacySSTableDir(Descriptor.Version.current_version);
    FileUtils.createDirectory(legacySSTableDir);
    Set<String> keys = new HashSet<String>();
    for(int i = 0; i < 10; i++)
    {
        keys.add("key"+i);
    }
    for(int i = 0; i < 3; i++)
    {
        SSTableReader ssTable = SSTableUtils.prepare().ks(KS).cf(CF).dest(new Descriptor(legacySSTableDir, KS, CF, i, false)).write(keys);
        System.out.println(ssTable);
    }
}
 

import org.apache.cassandra.io.sstable.SSTableUtils; //导入依赖的package包/类
protected void testCompaction(int sstableCount, int rowsPerSSTable, int colsPerRow) throws Exception
{
    CompactionManager.instance.disableAutoCompaction();

    Keyspace keyspace = Keyspace.open(KEYSPACE1);
    ColumnFamilyStore store = keyspace.getColumnFamilyStore("Standard1");

    ArrayList<SSTableReader> sstables = new ArrayList<SSTableReader>();
    for (int k = 0; k < sstableCount; k++)
    {
        SortedMap<String,ColumnFamily> rows = new TreeMap<String,ColumnFamily>();
        for (int j = 0; j < rowsPerSSTable; j++)
        {
            String key = String.valueOf(j);
            Column[] cols = new Column[colsPerRow];
            for (int i = 0; i < colsPerRow; i++)
            {
                // last sstable has highest timestamps
                cols[i] = Util.column(String.valueOf(i), String.valueOf(i), k);
            }
            rows.put(key, SSTableUtils.createCF(Long.MIN_VALUE, Integer.MIN_VALUE, cols));
        }
        SSTableReader sstable = SSTableUtils.prepare().write(rows);
        sstables.add(sstable);
        store.addSSTable(sstable);
    }

    // give garbage collection a bit of time to catch up
    Thread.sleep(1000);

    long start = System.nanoTime();
    final int gcBefore = (int) (System.currentTimeMillis() / 1000) - Schema.instance.getCFMetaData(KEYSPACE1, "Standard1").getGcGraceSeconds();
    new CompactionTask(store, sstables, gcBefore).execute(null);
    System.out.println(String.format("%s: sstables=%d rowsper=%d colsper=%d: %d ms",
                                     this.getClass().getName(),
                                     sstableCount,
                                     rowsPerSSTable,
                                     colsPerRow,
                                     TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start)));
}
 

import org.apache.cassandra.io.sstable.SSTableUtils; //导入依赖的package包/类
protected void testCompaction(int sstableCount, int rowsPerSSTable, int colsPerRow) throws Exception
{
    CompactionManager.instance.disableAutoCompaction();

    Keyspace keyspace = Keyspace.open(KEYSPACE1);
    ColumnFamilyStore store = keyspace.getColumnFamilyStore("Standard1");

    ArrayList<SSTableReader> sstables = new ArrayList<SSTableReader>();
    for (int k = 0; k < sstableCount; k++)
    {
        SortedMap<String,ColumnFamily> rows = new TreeMap<String,ColumnFamily>();
        for (int j = 0; j < rowsPerSSTable; j++)
        {
            String key = String.valueOf(j);
            Cell[] cols = new Cell[colsPerRow];
            for (int i = 0; i < colsPerRow; i++)
            {
                // last sstable has highest timestamps
                cols[i] = Util.column(String.valueOf(i), String.valueOf(i), k);
            }
            rows.put(key, SSTableUtils.createCF(KEYSPACE1, CF_STANDARD, Long.MIN_VALUE, Integer.MIN_VALUE, cols));
        }
        SSTableReader sstable = SSTableUtils.prepare().write(rows);
        sstables.add(sstable);
        store.addSSTable(sstable);
    }

    // give garbage collection a bit of time to catch up
    Thread.sleep(1000);

    long start = System.nanoTime();
    final int gcBefore = (int) (System.currentTimeMillis() / 1000) - Schema.instance.getCFMetaData(KEYSPACE1, "Standard1").getGcGraceSeconds();
    new CompactionTask(store, sstables, gcBefore, false).execute(null);
    System.out.println(String.format("%s: sstables=%d rowsper=%d colsper=%d: %d ms",
                                     this.getClass().getName(),
                                     sstableCount,
                                     rowsPerSSTable,
                                     colsPerRow,
                                     TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start)));
}
 

import org.apache.cassandra.io.sstable.SSTableUtils; //导入依赖的package包/类
protected void testCompaction(int sstableCount, int rowsPerSSTable, int colsPerRow) throws Exception
{
    CompactionManager.instance.disableAutoCompaction();

    Table table = Table.open(TABLE1);
    ColumnFamilyStore store = table.getColumnFamilyStore("Standard1");

    ArrayList<SSTableReader> sstables = new ArrayList<SSTableReader>();
    for (int k = 0; k < sstableCount; k++)
    {
        SortedMap<String,ColumnFamily> rows = new TreeMap<String,ColumnFamily>();
        for (int j = 0; j < rowsPerSSTable; j++)
        {
            String key = String.valueOf(j);
            IColumn[] cols = new IColumn[colsPerRow];
            for (int i = 0; i < colsPerRow; i++)
            {
                // last sstable has highest timestamps
                cols[i] = Util.column(String.valueOf(i), String.valueOf(i), k);
            }
            rows.put(key, SSTableUtils.createCF(Long.MIN_VALUE, Integer.MIN_VALUE, cols));
        }
        SSTableReader sstable = SSTableUtils.prepare().write(rows);
        sstables.add(sstable);
        store.addSSTable(sstable);
    }

    // give garbage collection a bit of time to catch up
    Thread.sleep(1000);

    long start = System.currentTimeMillis();
    final int gcBefore = (int) (System.currentTimeMillis() / 1000) - Schema.instance.getCFMetaData(TABLE1, "Standard1").getGcGraceSeconds();
    new CompactionTask(store, sstables, gcBefore).execute(null);
    System.out.println(String.format("%s: sstables=%d rowsper=%d colsper=%d: %d ms",
                                     this.getClass().getName(),
                                     sstableCount,
                                     rowsPerSSTable,
                                     colsPerRow,
                                     System.currentTimeMillis() - start));
}
 

import org.apache.cassandra.io.sstable.SSTableUtils; //导入依赖的package包/类
protected void testCompaction(int sstableCount, int rowsPerSSTable, int colsPerRow) throws Exception
{
    CompactionManager.instance.disableAutoCompaction();

    Keyspace keyspace = Keyspace.open(KEYSPACE1);
    ColumnFamilyStore store = keyspace.getColumnFamilyStore("Standard1");

    ArrayList<SSTableReader> sstables = new ArrayList<SSTableReader>();
    for (int k = 0; k < sstableCount; k++)
    {
        SortedMap<String,ColumnFamily> rows = new TreeMap<String,ColumnFamily>();
        for (int j = 0; j < rowsPerSSTable; j++)
        {
            String key = String.valueOf(j);
            Cell[] cols = new Cell[colsPerRow];
            for (int i = 0; i < colsPerRow; i++)
            {
                // last sstable has highest timestamps
                cols[i] = Util.column(String.valueOf(i), String.valueOf(i), k);
            }
            rows.put(key, SSTableUtils.createCF(Long.MIN_VALUE, Integer.MIN_VALUE, cols));
        }
        SSTableReader sstable = SSTableUtils.prepare().write(rows);
        sstables.add(sstable);
        store.addSSTable(sstable);
    }

    // give garbage collection a bit of time to catch up
    Thread.sleep(1000);

    long start = System.nanoTime();
    final int gcBefore = (int) (System.currentTimeMillis() / 1000) - Schema.instance.getCFMetaData(KEYSPACE1, "Standard1").getGcGraceSeconds();
    new CompactionTask(store, sstables, gcBefore).execute(null);
    System.out.println(String.format("%s: sstables=%d rowsper=%d colsper=%d: %d ms",
                                     this.getClass().getName(),
                                     sstableCount,
                                     rowsPerSSTable,
                                     colsPerRow,
                                     TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start)));
}
 

import org.apache.cassandra.io.sstable.SSTableUtils; //导入依赖的package包/类
protected void testCompaction(int sstableCount, int rowsPerSSTable, int colsPerRow) throws Exception
{
    CompactionManager.instance.disableAutoCompaction();

    Table table = Table.open(TABLE1);
    ColumnFamilyStore store = table.getColumnFamilyStore("Standard1");

    ArrayList<SSTableReader> sstables = new ArrayList<SSTableReader>();
    for (int k = 0; k < sstableCount; k++)
    {
        SortedMap<String,ColumnFamily> rows = new TreeMap<String,ColumnFamily>();
        for (int j = 0; j < rowsPerSSTable; j++)
        {
            String key = String.valueOf(j);
            IColumn[] cols = new IColumn[colsPerRow];
            for (int i = 0; i < colsPerRow; i++)
            {
                // last sstable has highest timestamps
                cols[i] = Util.column(String.valueOf(i), String.valueOf(i), k);
            }
            rows.put(key, SSTableUtils.createCF(Long.MIN_VALUE, Integer.MIN_VALUE, cols));
        }
        SSTableReader sstable = SSTableUtils.prepare().write(rows);
        sstables.add(sstable);
        store.addSSTable(sstable);
    }

    // give garbage collection a bit of time to catch up
    Thread.sleep(1000);

    long start = System.currentTimeMillis();
    CompactionManager.instance.doCompaction(store, sstables, (int) (System.currentTimeMillis() / 1000) - DatabaseDescriptor.getCFMetaData(TABLE1, "Standard1").getGcGraceSeconds());
    System.out.println(String.format("%s: sstables=%d rowsper=%d colsper=%d: %d ms",
                                     this.getClass().getName(),
                                     sstableCount,
                                     rowsPerSSTable,
                                     colsPerRow,
                                     System.currentTimeMillis() - start));
}
 

import org.apache.cassandra.io.sstable.SSTableUtils; //导入依赖的package包/类
@Test
public void testTransferTableCounter() throws Exception
{
    final Keyspace keyspace = Keyspace.open("Keyspace1");
    final ColumnFamilyStore cfs = keyspace.getColumnFamilyStore("Counter1");
    final CounterContext cc = new CounterContext();

    final Map<String, ColumnFamily> cleanedEntries = new HashMap<>();

    List<String> keys = createAndTransfer(cfs, new Mutator()
    {
        /** Creates a new SSTable per key: all will be merged before streaming. */
        public void mutate(String key, String col, long timestamp) throws Exception
        {
            Map<String, ColumnFamily> entries = new HashMap<>();
            ColumnFamily cf = ArrayBackedSortedColumns.factory.create(cfs.metadata);
            ColumnFamily cfCleaned = ArrayBackedSortedColumns.factory.create(cfs.metadata);
            CounterContext.ContextState state = CounterContext.ContextState.allocate(0, 1, 3);
            state.writeLocal(CounterId.fromInt(2), 9L, 3L);
            state.writeRemote(CounterId.fromInt(4), 4L, 2L);
            state.writeRemote(CounterId.fromInt(6), 3L, 3L);
            state.writeRemote(CounterId.fromInt(8), 2L, 4L);
            cf.addColumn(new BufferCounterCell(cellname(col), state.context, timestamp));
            cfCleaned.addColumn(new BufferCounterCell(cellname(col), cc.clearAllLocal(state.context), timestamp));

            entries.put(key, cf);
            cleanedEntries.put(key, cfCleaned);
            cfs.addSSTable(SSTableUtils.prepare()
                .ks(keyspace.getName())
                .cf(cfs.name)
                .generation(0)
                .write(entries));
        }
    }, true);

    // filter pre-cleaned entries locally, and ensure that the end result is equal
    cleanedEntries.keySet().retainAll(keys);
    SSTableReader cleaned = SSTableUtils.prepare()
        .ks(keyspace.getName())
        .cf(cfs.name)
        .generation(0)
        .write(cleanedEntries);
    SSTableReader streamed = cfs.getSSTables().iterator().next();
    SSTableUtils.assertContentEquals(cleaned, streamed);

    // Retransfer the file, making sure it is now idempotent (see CASSANDRA-3481)
    cfs.clearUnsafe();
    transferSSTables(streamed);
    SSTableReader restreamed = cfs.getSSTables().iterator().next();
    SSTableUtils.assertContentEquals(streamed, restreamed);
}
 

import org.apache.cassandra.io.sstable.SSTableUtils; //导入依赖的package包/类
@Test
public void testTransferOfMultipleColumnFamilies() throws Exception
{
    String keyspace = "KeyCacheSpace";
    IPartitioner p = StorageService.getPartitioner();
    String[] columnFamilies = new String[] { "Standard1", "Standard2", "Standard3" };
    List<SSTableReader> ssTableReaders = new ArrayList<>();

    NavigableMap<DecoratedKey,String> keys = new TreeMap<>();
    for (String cf : columnFamilies)
    {
        Set<String> content = new HashSet<>();
        content.add("data-" + cf + "-1");
        content.add("data-" + cf + "-2");
        content.add("data-" + cf + "-3");
        SSTableUtils.Context context = SSTableUtils.prepare().ks(keyspace).cf(cf);
        ssTableReaders.add(context.write(content));

        // collect dks for each string key
        for (String str : content)
            keys.put(Util.dk(str), cf);
    }

    // transfer the first and last keys
    Map.Entry<DecoratedKey,String> first = keys.firstEntry();
    Map.Entry<DecoratedKey,String> last = keys.lastEntry();
    Map.Entry<DecoratedKey,String> secondtolast = keys.lowerEntry(last.getKey());
    List<Range<Token>> ranges = new ArrayList<>();
    ranges.add(new Range<>(p.getMinimumToken(), first.getKey().getToken()));
    // the left hand side of the range is exclusive, so we transfer from the second-to-last token
    ranges.add(new Range<>(secondtolast.getKey().getToken(), p.getMinimumToken()));

    // Acquiring references, transferSSTables needs it
    Refs<SSTableReader> refs = Refs.tryRef(ssTableReaders);
    if (refs == null)
        throw new AssertionError();

    new StreamPlan("StreamingTransferTest").transferFiles(LOCAL, makeStreamingDetails(ranges, refs)).execute().get();

    // check that only two keys were transferred
    for (Map.Entry<DecoratedKey,String> entry : Arrays.asList(first, last))
    {
        ColumnFamilyStore store = Keyspace.open(keyspace).getColumnFamilyStore(entry.getValue());
        List<Row> rows = Util.getRangeSlice(store);
        assertEquals(rows.toString(), 1, rows.size());
        assertEquals(entry.getKey(), rows.get(0).key);
    }
}
 

import org.apache.cassandra.io.sstable.SSTableUtils; //导入依赖的package包/类
@Test
public void testTransferTableCounter() throws Exception
{
    final Keyspace keyspace = Keyspace.open("Keyspace1");
    final ColumnFamilyStore cfs = keyspace.getColumnFamilyStore("Counter1");
    final CounterContext cc = new CounterContext();

    final Map<String, ColumnFamily> cleanedEntries = new HashMap<>();

    List<String> keys = createAndTransfer(cfs, new Mutator()
    {
        /** Creates a new SSTable per key: all will be merged before streaming. */
        public void mutate(String key, String col, long timestamp) throws Exception
        {
            Map<String, ColumnFamily> entries = new HashMap<>();
            ColumnFamily cf = TreeMapBackedSortedColumns.factory.create(cfs.metadata);
            ColumnFamily cfCleaned = TreeMapBackedSortedColumns.factory.create(cfs.metadata);
            CounterContext.ContextState state = CounterContext.ContextState.allocate(4, 1);
            state.writeElement(CounterId.fromInt(2), 9L, 3L, true);
            state.writeElement(CounterId.fromInt(4), 4L, 2L);
            state.writeElement(CounterId.fromInt(6), 3L, 3L);
            state.writeElement(CounterId.fromInt(8), 2L, 4L);
            cf.addColumn(new CounterColumn(ByteBufferUtil.bytes(col),
                    state.context,
                    timestamp));
            cfCleaned.addColumn(new CounterColumn(ByteBufferUtil.bytes(col),
                    cc.clearAllDelta(state.context),
                    timestamp));

            entries.put(key, cf);
            cleanedEntries.put(key, cfCleaned);
            cfs.addSSTable(SSTableUtils.prepare()
                .ks(keyspace.getName())
                .cf(cfs.name)
                .generation(0)
                .write(entries));
        }
    }, true);

    // filter pre-cleaned entries locally, and ensure that the end result is equal
    cleanedEntries.keySet().retainAll(keys);
    SSTableReader cleaned = SSTableUtils.prepare()
        .ks(keyspace.getName())
        .cf(cfs.name)
        .generation(0)
        .write(cleanedEntries);
    SSTableReader streamed = cfs.getSSTables().iterator().next();
    SSTableUtils.assertContentEquals(cleaned, streamed);

    // Retransfer the file, making sure it is now idempotent (see CASSANDRA-3481)
    cfs.clearUnsafe();
    transferSSTables(streamed);
    SSTableReader restreamed = cfs.getSSTables().iterator().next();
    SSTableUtils.assertContentEquals(streamed, restreamed);
}
 

import org.apache.cassandra.io.sstable.SSTableUtils; //导入依赖的package包/类
@Test
public void testTransferOfMultipleColumnFamilies() throws Exception
{
    String keyspace = "KeyCacheSpace";
    IPartitioner p = StorageService.getPartitioner();
    String[] columnFamilies = new String[] { "Standard1", "Standard2", "Standard3" };
    List<SSTableReader> ssTableReaders = new ArrayList<>();

    NavigableMap<DecoratedKey,String> keys = new TreeMap<>();
    for (String cf : columnFamilies)
    {
        Set<String> content = new HashSet<>();
        content.add("data-" + cf + "-1");
        content.add("data-" + cf + "-2");
        content.add("data-" + cf + "-3");
        SSTableUtils.Context context = SSTableUtils.prepare().ks(keyspace).cf(cf);
        ssTableReaders.add(context.write(content));

        // collect dks for each string key
        for (String str : content)
            keys.put(Util.dk(str), cf);
    }

    // transfer the first and last keys
    Map.Entry<DecoratedKey,String> first = keys.firstEntry();
    Map.Entry<DecoratedKey,String> last = keys.lastEntry();
    Map.Entry<DecoratedKey,String> secondtolast = keys.lowerEntry(last.getKey());
    List<Range<Token>> ranges = new ArrayList<>();
    ranges.add(new Range<>(p.getMinimumToken(), first.getKey().token));
    // the left hand side of the range is exclusive, so we transfer from the second-to-last token
    ranges.add(new Range<>(secondtolast.getKey().token, p.getMinimumToken()));

    // Acquiring references, transferSSTables needs it
    if (!SSTableReader.acquireReferences(ssTableReaders))
        throw new AssertionError();

    new StreamPlan("StreamingTransferTest").transferFiles(LOCAL, makeStreamingDetails(ranges, ssTableReaders)).execute().get();

    // check that only two keys were transferred
    for (Map.Entry<DecoratedKey,String> entry : Arrays.asList(first, last))
    {
        ColumnFamilyStore store = Keyspace.open(keyspace).getColumnFamilyStore(entry.getValue());
        List<Row> rows = Util.getRangeSlice(store);
        assertEquals(rows.toString(), 1, rows.size());
        assertEquals(entry.getKey(), rows.get(0).key);
    }
}
 

import org.apache.cassandra.io.sstable.SSTableUtils; //导入依赖的package包/类
@Test
public void testTransferTableCounter() throws Exception
{
    final Keyspace keyspace = Keyspace.open(KEYSPACE1);
    final ColumnFamilyStore cfs = keyspace.getColumnFamilyStore("Counter1");
    final CounterContext cc = new CounterContext();

    final Map<String, ColumnFamily> cleanedEntries = new HashMap<>();

    List<String> keys = createAndTransfer(cfs, new Mutator()
    {
        /** Creates a new SSTable per key: all will be merged before streaming. */
        public void mutate(String key, String col, long timestamp) throws Exception
        {
            Map<String, ColumnFamily> entries = new HashMap<>();
            ColumnFamily cf = ArrayBackedSortedColumns.factory.create(cfs.metadata);
            ColumnFamily cfCleaned = ArrayBackedSortedColumns.factory.create(cfs.metadata);
            CounterContext.ContextState state = CounterContext.ContextState.allocate(0, 1, 3);
            state.writeLocal(CounterId.fromInt(2), 9L, 3L);
            state.writeRemote(CounterId.fromInt(4), 4L, 2L);
            state.writeRemote(CounterId.fromInt(6), 3L, 3L);
            state.writeRemote(CounterId.fromInt(8), 2L, 4L);
            cf.addColumn(new BufferCounterCell(cellname(col), state.context, timestamp));
            cfCleaned.addColumn(new BufferCounterCell(cellname(col), cc.clearAllLocal(state.context), timestamp));

            entries.put(key, cf);
            cleanedEntries.put(key, cfCleaned);
            cfs.addSSTable(SSTableUtils.prepare()
                .ks(keyspace.getName())
                .cf(cfs.name)
                .generation(0)
                .write(entries));
        }
    }, true);

    // filter pre-cleaned entries locally, and ensure that the end result is equal
    cleanedEntries.keySet().retainAll(keys);
    SSTableReader cleaned = SSTableUtils.prepare()
        .ks(keyspace.getName())
        .cf(cfs.name)
        .generation(0)
        .write(cleanedEntries);
    SSTableReader streamed = cfs.getSSTables().iterator().next();
    SSTableUtils.assertContentEquals(cleaned, streamed);

    // Retransfer the file, making sure it is now idempotent (see CASSANDRA-3481)
    cfs.clearUnsafe();
    transferSSTables(streamed);
    SSTableReader restreamed = cfs.getSSTables().iterator().next();
    SSTableUtils.assertContentEquals(streamed, restreamed);
}
 

import org.apache.cassandra.io.sstable.SSTableUtils; //导入依赖的package包/类
@Test
public void testTransferOfMultipleColumnFamilies() throws Exception
{
    String keyspace = KEYSPACE_CACHEKEY;
    IPartitioner p = StorageService.getPartitioner();
    String[] columnFamilies = new String[] { "Standard1", "Standard2", "Standard3" };
    List<SSTableReader> ssTableReaders = new ArrayList<>();

    NavigableMap<DecoratedKey,String> keys = new TreeMap<>();
    for (String cf : columnFamilies)
    {
        Set<String> content = new HashSet<>();
        content.add("data-" + cf + "-1");
        content.add("data-" + cf + "-2");
        content.add("data-" + cf + "-3");
        SSTableUtils.Context context = SSTableUtils.prepare().ks(keyspace).cf(cf);
        ssTableReaders.add(context.write(content));

        // collect dks for each string key
        for (String str : content)
            keys.put(Util.dk(str), cf);
    }

    // transfer the first and last keys
    Map.Entry<DecoratedKey,String> first = keys.firstEntry();
    Map.Entry<DecoratedKey,String> last = keys.lastEntry();
    Map.Entry<DecoratedKey,String> secondtolast = keys.lowerEntry(last.getKey());
    List<Range<Token>> ranges = new ArrayList<>();
    ranges.add(new Range<>(p.getMinimumToken(), first.getKey().getToken()));
    // the left hand side of the range is exclusive, so we transfer from the second-to-last token
    ranges.add(new Range<>(secondtolast.getKey().getToken(), p.getMinimumToken()));

    // Acquiring references, transferSSTables needs it
    if (!SSTableReader.acquireReferences(ssTableReaders))
        throw new AssertionError();

    new StreamPlan("StreamingTransferTest").transferFiles(LOCAL, makeStreamingDetails(ranges, ssTableReaders)).execute().get();

    // check that only two keys were transferred
    for (Map.Entry<DecoratedKey,String> entry : Arrays.asList(first, last))
    {
        ColumnFamilyStore store = Keyspace.open(keyspace).getColumnFamilyStore(entry.getValue());
        List<Row> rows = Util.getRangeSlice(store);
        assertEquals(rows.toString(), 1, rows.size());
        assertEquals(entry.getKey(), rows.get(0).key);
    }
}
 

import org.apache.cassandra.io.sstable.SSTableUtils; //导入依赖的package包/类
@Test
public void testTransferTableCounter() throws Exception
{
    final Table table = Table.open("Keyspace1");
    final ColumnFamilyStore cfs = table.getColumnFamilyStore("Counter1");
    final CounterContext cc = new CounterContext();

    final Map<String, ColumnFamily> cleanedEntries = new HashMap<String, ColumnFamily>();

    List<String> keys = createAndTransfer(table, cfs, new Mutator()
    {
        /** Creates a new SSTable per key: all will be merged before streaming. */
        public void mutate(String key, String col, long timestamp) throws Exception
        {
            Map<String, ColumnFamily> entries = new HashMap<String, ColumnFamily>();
            ColumnFamily cf = ColumnFamily.create(cfs.metadata);
            ColumnFamily cfCleaned = ColumnFamily.create(cfs.metadata);
            CounterContext.ContextState state = CounterContext.ContextState.allocate(4, 1);
            state.writeElement(CounterId.fromInt(2), 9L, 3L, true);
            state.writeElement(CounterId.fromInt(4), 4L, 2L);
            state.writeElement(CounterId.fromInt(6), 3L, 3L);
            state.writeElement(CounterId.fromInt(8), 2L, 4L);
            cf.addColumn(new CounterColumn(ByteBufferUtil.bytes(col),
                                           state.context,
                                           timestamp));
            cfCleaned.addColumn(new CounterColumn(ByteBufferUtil.bytes(col),
                                                  cc.clearAllDelta(state.context),
                                                  timestamp));

            entries.put(key, cf);
            cleanedEntries.put(key, cfCleaned);
            cfs.addSSTable(SSTableUtils.prepare()
                .ks(table.name)
                .cf(cfs.columnFamily)
                .generation(0)
                .write(entries));
        }
    });

    // filter pre-cleaned entries locally, and ensure that the end result is equal
    cleanedEntries.keySet().retainAll(keys);
    SSTableReader cleaned = SSTableUtils.prepare()
        .ks(table.name)
        .cf(cfs.columnFamily)
        .generation(0)
        .write(cleanedEntries);
    SSTableReader streamed = cfs.getSSTables().iterator().next();
    SSTableUtils.assertContentEquals(cleaned, streamed);

    // Retransfer the file, making sure it is now idempotent (see CASSANDRA-3481)
    cfs.clearUnsafe();
    transfer(table, streamed);
    SSTableReader restreamed = cfs.getSSTables().iterator().next();
    SSTableUtils.assertContentEquals(streamed, restreamed);
}
 

import org.apache.cassandra.io.sstable.SSTableUtils; //导入依赖的package包/类
@Test
public void testTransferOfMultipleColumnFamilies() throws Exception
{
    String keyspace = "KeyCacheSpace";
    IPartitioner p = StorageService.getPartitioner();
    String[] columnFamilies = new String[] { "Standard1", "Standard2", "Standard3" };
    List<SSTableReader> ssTableReaders = new ArrayList<SSTableReader>();

    NavigableMap<DecoratedKey,String> keys = new TreeMap<DecoratedKey,String>();
    for (String cf : columnFamilies)
    {
        Set<String> content = new HashSet<String>();
        content.add("data-" + cf + "-1");
        content.add("data-" + cf + "-2");
        content.add("data-" + cf + "-3");
        SSTableUtils.Context context = SSTableUtils.prepare().ks(keyspace).cf(cf);
        ssTableReaders.add(context.write(content));

        // collect dks for each string key
        for (String str : content)
            keys.put(Util.dk(str), cf);
    }

    // transfer the first and last keys
    Map.Entry<DecoratedKey,String> first = keys.firstEntry();
    Map.Entry<DecoratedKey,String> last = keys.lastEntry();
    Map.Entry<DecoratedKey,String> secondtolast = keys.lowerEntry(last.getKey());
    List<Range<Token>> ranges = new ArrayList<Range<Token>>();
    ranges.add(new Range<Token>(p.getMinimumToken(), first.getKey().token));
    // the left hand side of the range is exclusive, so we transfer from the second-to-last token
    ranges.add(new Range<Token>(secondtolast.getKey().token, p.getMinimumToken()));

    // Acquiring references, transferSSTables needs it
    if (!SSTableReader.acquireReferences(ssTableReaders))
        throw new AssertionError();

    StreamOutSession session = StreamOutSession.create(keyspace, LOCAL, (IStreamCallback)null);
    StreamOut.transferSSTables(session, ssTableReaders, ranges, OperationType.BOOTSTRAP);

    session.await();

    // check that only two keys were transferred
    for (Map.Entry<DecoratedKey,String> entry : Arrays.asList(first, last))
    {
        ColumnFamilyStore store = Table.open(keyspace).getColumnFamilyStore(entry.getValue());
        List<Row> rows = Util.getRangeSlice(store);
        assertEquals(rows.toString(), 1, rows.size());
        assertEquals(entry.getKey(), rows.get(0).key);
    }
}
 

import org.apache.cassandra.io.sstable.SSTableUtils; //导入依赖的package包/类
@Test
public void testTransferTableCounter() throws Exception
{
    final Keyspace keyspace = Keyspace.open("Keyspace1");
    final ColumnFamilyStore cfs = keyspace.getColumnFamilyStore("Counter1");
    final CounterContext cc = new CounterContext();

    final Map<String, ColumnFamily> cleanedEntries = new HashMap<>();

    List<String> keys = createAndTransfer(cfs, new Mutator()
    {
        /** Creates a new SSTable per key: all will be merged before streaming. */
        public void mutate(String key, String col, long timestamp) throws Exception
        {
            Map<String, ColumnFamily> entries = new HashMap<>();
            ColumnFamily cf = TreeMapBackedSortedColumns.factory.create(cfs.metadata);
            ColumnFamily cfCleaned = TreeMapBackedSortedColumns.factory.create(cfs.metadata);
            CounterContext.ContextState state = CounterContext.ContextState.allocate(0, 1, 3, HeapAllocator.instance);
            state.writeLocal(CounterId.fromInt(2), 9L, 3L);
            state.writeRemote(CounterId.fromInt(4), 4L, 2L);
            state.writeRemote(CounterId.fromInt(6), 3L, 3L);
            state.writeRemote(CounterId.fromInt(8), 2L, 4L);
            cf.addColumn(new CounterCell(cellname(col), state.context, timestamp));
            cfCleaned.addColumn(new CounterCell(cellname(col), cc.clearAllLocal(state.context), timestamp));

            entries.put(key, cf);
            cleanedEntries.put(key, cfCleaned);
            cfs.addSSTable(SSTableUtils.prepare()
                .ks(keyspace.getName())
                .cf(cfs.name)
                .generation(0)
                .write(entries));
        }
    }, true);

    // filter pre-cleaned entries locally, and ensure that the end result is equal
    cleanedEntries.keySet().retainAll(keys);
    SSTableReader cleaned = SSTableUtils.prepare()
        .ks(keyspace.getName())
        .cf(cfs.name)
        .generation(0)
        .write(cleanedEntries);
    SSTableReader streamed = cfs.getSSTables().iterator().next();
    SSTableUtils.assertContentEquals(cleaned, streamed);

    // Retransfer the file, making sure it is now idempotent (see CASSANDRA-3481)
    cfs.clearUnsafe();
    transferSSTables(streamed);
    SSTableReader restreamed = cfs.getSSTables().iterator().next();
    SSTableUtils.assertContentEquals(streamed, restreamed);
}
 

import org.apache.cassandra.io.sstable.SSTableUtils; //导入依赖的package包/类
@Test
public void testTransferTableCounter() throws Exception
{
    final Table table = Table.open("Keyspace1");
    final ColumnFamilyStore cfs = table.getColumnFamilyStore("Counter1");
    final CounterContext cc = new CounterContext();

    final Map<String, ColumnFamily> cleanedEntries = new HashMap<String, ColumnFamily>();

    List<String> keys = createAndTransfer(table, cfs, new Mutator()
    {
        /** Creates a new SSTable per key: all will be merged before streaming. */
        public void mutate(String key, String col, long timestamp) throws Exception
        {
            Map<String, ColumnFamily> entries = new HashMap<String, ColumnFamily>();
            ColumnFamily cf = ColumnFamily.create(cfs.metadata);
            ColumnFamily cfCleaned = ColumnFamily.create(cfs.metadata);
            CounterContext.ContextState state = CounterContext.ContextState.allocate(4, 1);
            state.writeElement(CounterId.fromInt(2), 9L, 3L, true);
            state.writeElement(CounterId.fromInt(4), 4L, 2L);
            state.writeElement(CounterId.fromInt(6), 3L, 3L);
            state.writeElement(CounterId.fromInt(8), 2L, 4L);
            cf.addColumn(new CounterColumn(ByteBufferUtil.bytes(col),
                    state.context,
                    timestamp));
            cfCleaned.addColumn(new CounterColumn(ByteBufferUtil.bytes(col),
                    cc.clearAllDelta(state.context),
                    timestamp));

            entries.put(key, cf);
            cleanedEntries.put(key, cfCleaned);
            cfs.addSSTable(SSTableUtils.prepare()
                    .ks(table.name)
                    .cf(cfs.columnFamily)
                    .generation(0)
                    .write(entries));
        }
    }, true);

    // filter pre-cleaned entries locally, and ensure that the end result is equal
    cleanedEntries.keySet().retainAll(keys);
    SSTableReader cleaned = SSTableUtils.prepare()
            .ks(table.name)
            .cf(cfs.columnFamily)
            .generation(0)
            .write(cleanedEntries);
    SSTableReader streamed = cfs.getSSTables().iterator().next();
    SSTableUtils.assertContentEquals(cleaned, streamed);

    // Retransfer the file, making sure it is now idempotent (see CASSANDRA-3481)
    cfs.clearUnsafe();
    transferSSTables(table, streamed);
    SSTableReader restreamed = cfs.getSSTables().iterator().next();
    SSTableUtils.assertContentEquals(streamed, restreamed);
}
 

import org.apache.cassandra.io.sstable.SSTableUtils; //导入依赖的package包/类
@Test
public void testTransferTableCounter() throws Exception
{
    final Keyspace keyspace = Keyspace.open("Keyspace1");
    final ColumnFamilyStore cfs = keyspace.getColumnFamilyStore("Counter1");
    final CounterContext cc = new CounterContext();

    final Map<String, ColumnFamily> cleanedEntries = new HashMap<>();

    List<String> keys = createAndTransfer(cfs, new Mutator()
    {
        /** Creates a new SSTable per key: all will be merged before streaming. */
        public void mutate(String key, String col, long timestamp) throws Exception
        {
            Map<String, ColumnFamily> entries = new HashMap<>();
            ColumnFamily cf = ArrayBackedSortedColumns.factory.create(cfs.metadata);
            ColumnFamily cfCleaned = ArrayBackedSortedColumns.factory.create(cfs.metadata);
            CounterContext.ContextState state = CounterContext.ContextState.allocate(0, 1, 3);
            state.writeLocal(CounterId.fromInt(2), 9L, 3L);
            state.writeRemote(CounterId.fromInt(4), 4L, 2L);
            state.writeRemote(CounterId.fromInt(6), 3L, 3L);
            state.writeRemote(CounterId.fromInt(8), 2L, 4L);
            cf.addColumn(new CounterCell(cellname(col), state.context, timestamp));
            cfCleaned.addColumn(new CounterCell(cellname(col), cc.clearAllLocal(state.context), timestamp));

            entries.put(key, cf);
            cleanedEntries.put(key, cfCleaned);
            cfs.addSSTable(SSTableUtils.prepare()
                .ks(keyspace.getName())
                .cf(cfs.name)
                .generation(0)
                .write(entries));
        }
    }, true);

    // filter pre-cleaned entries locally, and ensure that the end result is equal
    cleanedEntries.keySet().retainAll(keys);
    SSTableReader cleaned = SSTableUtils.prepare()
        .ks(keyspace.getName())
        .cf(cfs.name)
        .generation(0)
        .write(cleanedEntries);
    SSTableReader streamed = cfs.getSSTables().iterator().next();
    SSTableUtils.assertContentEquals(cleaned, streamed);

    // Retransfer the file, making sure it is now idempotent (see CASSANDRA-3481)
    cfs.clearUnsafe();
    transferSSTables(streamed);
    SSTableReader restreamed = cfs.getSSTables().iterator().next();
    SSTableUtils.assertContentEquals(streamed, restreamed);
}
 

import org.apache.cassandra.io.sstable.SSTableUtils; //导入依赖的package包/类
@Test
public void testTransferOfMultipleColumnFamilies() throws Exception
{
    String keyspace = "KeyCacheSpace";
    IPartitioner p = StorageService.getPartitioner();
    String[] columnFamilies = new String[] { "Standard1", "Standard2", "Standard3" };
    List<SSTableReader> ssTableReaders = new ArrayList<SSTableReader>();

    NavigableMap<DecoratedKey,String> keys = new TreeMap<DecoratedKey,String>();
    for (String cf : columnFamilies)
    {
        Set<String> content = new HashSet<String>();
        content.add("data-"