import org.apache.cassandra.Util; //导入依赖的package包/类
@BeforeClass
public static void addData()
{
    cfs().clearUnsafe();

    int nbKeys = 10;
    int nbCols = 10;

    /*
     * Creates the following data:
     *   k1: c1 ... cn
     *   ...
     *   ki: c1 ... cn
     */
    for (int i = 0; i < nbKeys; i++)
    {
        Mutation rm = new Mutation(KS, bytes("k" + i));
        ColumnFamily cf = rm.addOrGet(CF);

        for (int j = 0; j < nbCols; j++)
            cf.addColumn(Util.column("c" + j, "", 0));

        rm.applyUnsafe();
    }
}
 

import org.apache.cassandra.Util; //导入依赖的package包/类
@Test
public void testGetColumn() throws IOException, ColumnFamilyNotDefinedException
{
    Keyspace keyspace = Keyspace.open("Keyspace1");
    RowMutation rm;
    DecoratedKey dk = Util.dk("key1");

    // add data
    rm = new RowMutation("Keyspace1", dk.key);
    rm.add("Standard1", ByteBufferUtil.bytes("Column1"), ByteBufferUtil.bytes("abcd"), 0);
    rm.apply();

    ReadCommand command = new SliceByNamesReadCommand("Keyspace1", dk.key, "Standard1", System.currentTimeMillis(), new NamesQueryFilter(ByteBufferUtil.bytes("Column1")));
    Row row = command.getRow(keyspace);
    Column col = row.cf.getColumn(ByteBufferUtil.bytes("Column1"));
    assertEquals(col.value(), ByteBuffer.wrap("abcd".getBytes()));
}
 

import org.apache.cassandra.Util; //导入依赖的package包/类
@Test
public void testCleanupWithNewToken() throws ExecutionException, InterruptedException, UnknownHostException
{
    StorageService.instance.getTokenMetadata().clearUnsafe();

    Keyspace keyspace = Keyspace.open(KEYSPACE1);
    ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF_STANDARD1);

    // insert data and verify we get it back w/ range query
    fillCF(cfs, "val", LOOPS);

    assertEquals(LOOPS, Util.getAll(Util.cmd(cfs).build()).size());
    TokenMetadata tmd = StorageService.instance.getTokenMetadata();

    byte[] tk1 = new byte[1], tk2 = new byte[1];
    tk1[0] = 2;
    tk2[0] = 1;
    tmd.updateNormalToken(new BytesToken(tk1), InetAddress.getByName("127.0.0.1"));
    tmd.updateNormalToken(new BytesToken(tk2), InetAddress.getByName("127.0.0.2"));
    CompactionManager.instance.performCleanup(cfs, 2);

    assertEquals(0, Util.getAll(Util.cmd(cfs).build()).size());
}
 

import org.apache.cassandra.Util; //导入依赖的package包/类
private void validateNameSort(Keyspace keyspace, int N) throws IOException
{
    for (int i = 0; i < N; ++i)
    {
        DecoratedKey key = Util.dk(Integer.toString(i));
        ColumnFamily cf;

        cf = Util.getColumnFamily(keyspace, key, "Standard1");
        Collection<Cell> cells = cf.getSortedColumns();
        for (Cell cell : cells)
        {
            String name = ByteBufferUtil.string(cell.name().toByteBuffer());
            int j = Integer.valueOf(name.substring(name.length() - 1));
            byte[] bytes = j % 2 == 0 ? "a".getBytes() : "b".getBytes();
            assertEquals(new String(bytes), ByteBufferUtil.string(cell.value()));
        }
    }
}
 

import org.apache.cassandra.Util; //导入依赖的package包/类
@Test
public void testGetColumn()
{
    ColumnFamilyStore cfs = Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF);

    new RowUpdateBuilder(cfs.metadata, 0, ByteBufferUtil.bytes("key1"))
            .clustering("Column1")
            .add("val", ByteBufferUtil.bytes("abcd"))
            .build()
            .apply();

    ColumnDefinition col = cfs.metadata.getColumnDefinition(ByteBufferUtil.bytes("val"));
    int found = 0;
    for (FilteredPartition partition : Util.getAll(Util.cmd(cfs).build()))
    {
        for (Row r : partition)
        {
            if (r.getCell(col).value().equals(ByteBufferUtil.bytes("abcd")))
                ++found;
        }
    }
    assertEquals(1, found);
}
 

import org.apache.cassandra.Util; //导入依赖的package包/类
@Test
public void testGetRowNoColumns() throws Throwable
{
    String tableName = createTable("CREATE TABLE %s (a text, b int, c int, PRIMARY KEY (a, b))");

    execute("INSERT INTO %s (a, b, c) VALUES (?, ?, ?)", "0", 0, 0);

    final ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(tableName);

    for (int round = 0; round < 2; round++)
    {
        // slice with limit 0
        Util.assertEmpty(Util.cmd(cfs, "0").columns("c").withLimit(0).build());

        // slice with nothing in between the bounds
        Util.assertEmpty(Util.cmd(cfs, "0").columns("c").fromIncl(1).toIncl(1).build());

        // fetch a non-existent name
        Util.assertEmpty(Util.cmd(cfs, "0").columns("c").includeRow(1).build());

        if (round == 0)
            cfs.forceBlockingFlush();
    }
}
 

import org.apache.cassandra.Util; //导入依赖的package包/类
@Test
public void testTrackTimesPartitionTombstone() throws ExecutionException, InterruptedException
{
    Keyspace ks = Keyspace.open(KSNAME);
    ColumnFamilyStore cfs = ks.getColumnFamilyStore(CFNAME);
    cfs.truncateBlocking();
    String key = "rt_times";

    int nowInSec = FBUtilities.nowInSeconds();
    new Mutation(PartitionUpdate.fullPartitionDelete(cfs.metadata, Util.dk(key), 1000, nowInSec)).apply();
    cfs.forceBlockingFlush();

    SSTableReader sstable = cfs.getLiveSSTables().iterator().next();
    assertTimes(sstable.getSSTableMetadata(), 1000, 1000, nowInSec);
    cfs.forceMajorCompaction();
    sstable = cfs.getLiveSSTables().iterator().next();
    assertTimes(sstable.getSSTableMetadata(), 1000, 1000, nowInSec);
}
 

import org.apache.cassandra.Util; //导入依赖的package包/类
private SSTableReader writeFile(ColumnFamilyStore cfs, int count)
{
    ArrayBackedSortedColumns cf = ArrayBackedSortedColumns.factory.create(cfs.metadata);
    for (int i = 0; i < count; i++)
        cf.addColumn(Util.column(String.valueOf(i), "a", 1));
    File dir = cfs.directories.getDirectoryForNewSSTables();
    String filename = cfs.getTempSSTablePath(dir);

    SSTableWriter writer = new SSTableWriter(filename,
            0,
            0,
            cfs.metadata,
            StorageService.getPartitioner(),
            new MetadataCollector(cfs.metadata.comparator));

    for (int i = 0; i < count * 5; i++)
        writer.append(StorageService.getPartitioner().decorateKey(ByteBufferUtil.bytes(i)), cf);
    return writer.closeAndOpenReader();
}
 

import org.apache.cassandra.Util; //导入依赖的package包/类
@Test
public void testFlushUnwriteableDie() throws Throwable
{
    makeTable();

    KillerForTests killerForTests = new KillerForTests();
    JVMStabilityInspector.Killer originalKiller = JVMStabilityInspector.replaceKiller(killerForTests);
    DiskFailurePolicy oldPolicy = DatabaseDescriptor.getDiskFailurePolicy();
    try (Closeable c = Util.markDirectoriesUnwriteable(getCurrentColumnFamilyStore()))
    {
        DatabaseDescriptor.setDiskFailurePolicy(DiskFailurePolicy.die);
        flushAndExpectError();
        Assert.assertTrue(killerForTests.wasKilled());
        Assert.assertFalse(killerForTests.wasKilledQuietly()); //only killed quietly on startup failure
    }
    finally
    {
        DatabaseDescriptor.setDiskFailurePolicy(oldPolicy);
        JVMStabilityInspector.replaceKiller(originalKiller);
    }
}
 

import org.apache.cassandra.Util; //导入依赖的package包/类
public void run()
{
    RateLimiter rl = rateLimit != 0 ? RateLimiter.create(rateLimit) : null;
    final Random rand = random != null ? random : ThreadLocalRandom.current();
    while (!stop)
    {
        if (rl != null)
            rl.acquire();
        ByteBuffer key = randomBytes(16, rand);

        UpdateBuilder builder = UpdateBuilder.create(Schema.instance.getCFMetaData("Keyspace1", "Standard1"), Util.dk(key));
        for (int ii = 0; ii < numCells; ii++)
        {
            int sz = randomSize ? rand.nextInt(cellSize) : cellSize;
            ByteBuffer bytes = randomBytes(sz, rand);
            builder.newRow("name" + ii).add("val", bytes);
            hash = hash(hash, bytes);
            ++cells;
            dataSize += sz;
        }

        rp = commitLog.add(new Mutation(builder.build()));
        counter.incrementAndGet();
    }
}
 

import org.apache.cassandra.Util; //导入依赖的package包/类
public SSTableWriter getDummyWriter() throws IOException
{
    File tempSS = tempSSTableFile("Keyspace1", "Standard1");
    ColumnFamily cfamily = TreeMapBackedSortedColumns.factory.create("Keyspace1", "Standard1");
    SSTableWriter writer = new SSTableWriter(tempSS.getPath(), 2);

    // Add rowA
    cfamily.addColumn(ByteBufferUtil.bytes("colA"), ByteBufferUtil.bytes("valA"), System.currentTimeMillis());
    writer.append(Util.dk("rowA"), cfamily);
    cfamily.clear();
    
    cfamily.addColumn(ByteBufferUtil.bytes("colB"), ByteBufferUtil.bytes("valB"), System.currentTimeMillis());
    writer.append(Util.dk("rowB"), cfamily);
    cfamily.clear();
    
    
    return writer;

}
 

import org.apache.cassandra.Util; //导入依赖的package包/类
@Test
public void testColumnDelete()
{
    // issue a column delete and test that the configured index instance was notified to update
    Mutation rm;
    rm = new Mutation("PerRowSecondaryIndex", ByteBufferUtil.bytes("k2"));
    rm.delete("Indexed1", Util.cellname("indexed"), 1);
    rm.apply();

    ColumnFamily indexedRow = PerRowSecondaryIndexTest.TestIndex.LAST_INDEXED_ROW;
    assertNotNull(indexedRow);

    for (Cell cell : indexedRow.getSortedColumns())
        assertFalse(cell.isLive());

    assertTrue(Arrays.equals("k2".getBytes(), PerRowSecondaryIndexTest.TestIndex.LAST_INDEXED_KEY.array()));
}
 

import org.apache.cassandra.Util; //导入依赖的package包/类
private void verify2(int key, int expectedRows, int minVal, int maxVal)
{
    ReadCommand cmd = Util.cmd(getCurrentColumnFamilyStore(), Util.dk(ByteBufferUtil.bytes(key))).build();
    int foundRows = 0;
    try (ReadOrderGroup orderGroup = cmd.startOrderGroup(); UnfilteredPartitionIterator iterator = cmd.executeLocally(orderGroup))
    {
        while (iterator.hasNext())
        {
            UnfilteredRowIterator rowIter = iterator.next();
            while (rowIter.hasNext())
            {
                AbstractRow row = (AbstractRow) rowIter.next();
                for (int i = 0; i < row.clustering().size(); i++)
                {
                    foundRows++;
                    int val = ByteBufferUtil.toInt(row.clustering().get(i));
                    assertTrue("val=" + val, val >= minVal && val < maxVal);
                }
            }
        }
    }
    assertEquals(expectedRows, foundRows);
}
 

import org.apache.cassandra.Util; //导入依赖的package包/类
@Test
public void testGetColumn()
{
    Keyspace keyspace = Keyspace.open("Keyspace1");
    CellNameType type = keyspace.getColumnFamilyStore("Standard1").getComparator();
    Mutation rm;
    DecoratedKey dk = Util.dk("key1");

    // add data
    rm = new Mutation("Keyspace1", dk.getKey());
    rm.add("Standard1", Util.cellname("Column1"), ByteBufferUtil.bytes("abcd"), 0);
    rm.apply();

    ReadCommand command = new SliceByNamesReadCommand("Keyspace1", dk.getKey(), "Standard1", System.currentTimeMillis(), new NamesQueryFilter(FBUtilities.singleton(Util.cellname("Column1"), type)));
    Row row = command.getRow(keyspace);
    Cell col = row.cf.getColumn(Util.cellname("Column1"));
    assertEquals(col.value(), ByteBuffer.wrap("abcd".getBytes()));
}
 

import org.apache.cassandra.Util; //导入依赖的package包/类
@Test
public void testFlushUnwriteableStop() throws Throwable
{
    makeTable();

    DiskFailurePolicy oldPolicy = DatabaseDescriptor.getDiskFailurePolicy();
    try (Closeable c = Util.markDirectoriesUnwriteable(getCurrentColumnFamilyStore()))
    {
        DatabaseDescriptor.setDiskFailurePolicy(DiskFailurePolicy.stop);
        flushAndExpectError();
        Assert.assertFalse(Gossiper.instance.isEnabled());
    }
    finally
    {
        DatabaseDescriptor.setDiskFailurePolicy(oldPolicy);
    }
}
 

import org.apache.cassandra.Util; //导入依赖的package包/类
private void verify(int expectedRows, int minVal, int maxVal)
{
    ReadCommand cmd = Util.cmd(getCurrentColumnFamilyStore()).build();
    int foundRows = 0;
    try (ReadOrderGroup orderGroup = cmd.startOrderGroup(); UnfilteredPartitionIterator iterator = cmd.executeLocally(orderGroup))
    {
        while (iterator.hasNext())
        {
            UnfilteredRowIterator rowIter = iterator.next();
            if (!rowIter.partitionKey().equals(Util.dk(ByteBufferUtil.bytes(999)))) // partition key 999 is 'live' and used to avoid sstables from being dropped
            {
                while (rowIter.hasNext())
                {
                    AbstractRow row = (AbstractRow) rowIter.next();
                    for (int i = 0; i < row.clustering().size(); i++)
                    {
                        foundRows++;
                        int val = ByteBufferUtil.toInt(row.clustering().get(i));
                        assertTrue("val=" + val, val >= minVal && val < maxVal);
                    }
                }
            }
        }
    }
    assertEquals(expectedRows, foundRows);
}
 

import org.apache.cassandra.Util; //导入依赖的package包/类
protected void fillCF(ColumnFamilyStore cfs, int rowsPerSSTable)
{
    CompactionManager.instance.disableAutoCompaction();

    for (int i = 0; i < rowsPerSSTable; i++)
    {
        String key = String.valueOf(i);
        // create a row and update the birthdate value, test that the index query fetches the new version
        Mutation rm;
        rm = new Mutation(KEYSPACE1, ByteBufferUtil.bytes(key));
        rm.add(cfs.name, Util.cellname(COLUMN), VALUE, System.currentTimeMillis());
        rm.applyUnsafe();
    }

    cfs.forceBlockingFlush();
}
 

import org.apache.cassandra.Util; //导入依赖的package包/类
@Test
@BMRule(name = "Greatest name in the world",
        targetClass="HintsBufferPool",
        targetMethod="switchCurrentBuffer",
        targetLocation="AT INVOKE java.util.concurrent.BlockingQueue.take",
        action="org.apache.cassandra.hints.HintsBufferPoolTest.blockedOnBackpressure = true;")
public void testBackpressure() throws Exception
{
    Queue<HintsBuffer> returnedBuffers = new ConcurrentLinkedQueue<>();
    HintsBufferPool pool = new HintsBufferPool(256, (buffer, p) -> returnedBuffers.offer(buffer));

    Thread blocked = new Thread(() -> {
        for (int ii = 0; ii < 512; ii++)
            pool.write(ImmutableList.of(UUID.randomUUID()), HintsBufferTest.createHint(ii, ii));
    });
    blocked.start();

    Util.spinAssertEquals(State.WAITING, () -> blocked.getState(), 1);

    while (blocked.isAlive())
        if (!returnedBuffers.isEmpty())
            pool.offer(returnedBuffers.poll().recycle());

    assertTrue(blockedOnBackpressure);
}
 

import org.apache.cassandra.Util; //导入依赖的package包/类
@Test
public void test7810() throws ExecutionException, InterruptedException
{
    Keyspace ks = Keyspace.open(KSNAME);
    ColumnFamilyStore cfs = ks.getColumnFamilyStore(CFNAME);
    cfs.metadata.gcGraceSeconds(2);

    String key = "7810";

    UpdateBuilder builder = UpdateBuilder.create(cfs.metadata, key).withTimestamp(0);
    for (int i = 10; i < 20; i ++)
        builder.newRow(i).add("val", i);
    builder.apply();
    cfs.forceBlockingFlush();

    new RowUpdateBuilder(cfs.metadata, 1, key).addRangeTombstone(10, 11).build().apply();
    cfs.forceBlockingFlush();

    Thread.sleep(5);
    cfs.forceMajorCompaction();
    assertEquals(8, Util.getOnlyPartitionUnfiltered(Util.cmd(cfs, key).build()).rowCount());
}
 

import org.apache.cassandra.Util; //导入依赖的package包/类
@Test
public void testGetColumnWithWrongBF() throws IOException, ExecutionException, InterruptedException
{
    Keyspace keyspace = Keyspace.open("Keyspace1");
    ColumnFamilyStore cfs = keyspace.getColumnFamilyStore("Standard1");
    cfs.truncateBlocking();

    List<IMutation> rms = new LinkedList<IMutation>();
    RowMutation rm;
    rm = new RowMutation("Keyspace1", ByteBufferUtil.bytes("key1"));
    rm.add("Standard1", ByteBufferUtil.bytes("Column1"), ByteBufferUtil.bytes("asdf"), 0);
    rm.add("Standard1", ByteBufferUtil.bytes("Column2"), ByteBufferUtil.bytes("asdf"), 0);
    rms.add(rm);
    Util.writeColumnFamily(rms);

    List<SSTableReader> ssTables = keyspace.getAllSSTables();
    assertEquals(1, ssTables.size());
    ssTables.get(0).forceFilterFailures();
    ColumnFamily cf = cfs.getColumnFamily(QueryFilter.getIdentityFilter(Util.dk("key2"), "Standard1", System.currentTimeMillis()));
    assertNull(cf);
}