import org.apache.cassandra.utils.CounterId; //导入依赖的package包/类
/**
 * Finds the position of a shard with the given id within the context (via binary search).
 */
@VisibleForTesting
public int findPositionOf(ByteBuffer context, CounterId id)
{
    int headerLength = headerLength(context);
    int offset = context.position() + headerLength;

    int left = 0;
    int right = (context.remaining() - headerLength) / STEP_LENGTH - 1;

    while (right >= left)
    {
        int middle = (left + right) / 2;
        int cmp = compareId(context, offset + middle * STEP_LENGTH, id.bytes(), id.bytes().position());

        if (cmp == -1)
            left = middle + 1;
        else if (cmp == 0)
            return offset + middle * STEP_LENGTH;
        else
            right = middle - 1;
    }

    return -1; // position not found
}
 

import org.apache.cassandra.utils.CounterId; //导入依赖的package包/类
@Test
public void testTotal()
{
    ContextState mixed = ContextState.allocate(0, 1, 4);
    mixed.writeRemote(CounterId.fromInt(1), 1L, 1L);
    mixed.writeRemote(CounterId.fromInt(2), 2L, 2L);
    mixed.writeRemote(CounterId.fromInt(4), 4L, 4L);
    mixed.writeRemote(CounterId.fromInt(5), 5L, 5L);
    mixed.writeLocal(CounterId.getLocalId(), 12L, 12L);
    assertEquals(24L, cc.total(mixed.context));

    ContextState global = ContextState.allocate(3, 0, 0);
    global.writeGlobal(CounterId.fromInt(1), 1L, 1L);
    global.writeGlobal(CounterId.fromInt(2), 2L, 2L);
    global.writeGlobal(CounterId.fromInt(3), 3L, 3L);
    assertEquals(6L, cc.total(global.context));
}
 

import org.apache.cassandra.utils.CounterId; //导入依赖的package包/类
public void performCleanup(ColumnFamilyStore cfStore, final CounterId.OneShotRenewer renewer) throws InterruptedException, ExecutionException
{
    performAllSSTableOperation(cfStore, new AllSSTablesOperation()
    {
        public void perform(ColumnFamilyStore store, Collection<SSTableReader> sstables) throws IOException
        {
            // Sort the column families in order of SSTable size, so cleanup of smaller CFs
            // can free up space for larger ones
            List<SSTableReader> sortedSSTables = new ArrayList<SSTableReader>(sstables);
            Collections.sort(sortedSSTables, new Comparator<SSTableReader>()
            {
                public int compare(SSTableReader o1, SSTableReader o2)
                {
                    return Longs.compare(o1.onDiskLength(), o2.onDiskLength());
                }
            });

            doCleanupCompaction(store, sortedSSTables, renewer);
        }
    });
}
 

import org.apache.cassandra.utils.CounterId; //导入依赖的package包/类
/**
 * Read the current local node id from the system table or null if no
 * such node id is recorded.
 */
public static CounterId getCurrentLocalCounterId()
{
    Table table = Table.open(Table.SYSTEM_KS);

    // Get the last CounterId (since CounterId are timeuuid is thus ordered from the older to the newer one)
    QueryFilter filter = QueryFilter.getSliceFilter(decorate(ALL_LOCAL_NODE_ID_KEY),
                                                    new QueryPath(COUNTER_ID_CF),
                                                    ByteBufferUtil.EMPTY_BYTE_BUFFER,
                                                    ByteBufferUtil.EMPTY_BYTE_BUFFER,
                                                    true,
                                                    1);
    ColumnFamily cf = table.getColumnFamilyStore(COUNTER_ID_CF).getColumnFamily(filter);
    if (cf != null && cf.getColumnCount() != 0)
        return CounterId.wrap(cf.iterator().next().name());
    else
        return null;
}
 

import org.apache.cassandra.utils.CounterId; //导入依赖的package包/类
public static List<CounterId.CounterIdRecord> getOldLocalCounterIds()
{
    List<CounterId.CounterIdRecord> l = new ArrayList<CounterId.CounterIdRecord>();

    Table table = Table.open(Table.SYSTEM_KS);
    QueryFilter filter = QueryFilter.getIdentityFilter(decorate(ALL_LOCAL_NODE_ID_KEY), new QueryPath(COUNTER_ID_CF));
    ColumnFamily cf = table.getColumnFamilyStore(COUNTER_ID_CF).getColumnFamily(filter);

    CounterId previous = null;
    for (IColumn c : cf)
    {
        if (previous != null)
            l.add(new CounterId.CounterIdRecord(previous, c.timestamp()));

        // this will ignore the last column on purpose since it is the
        // current local node id
        previous = CounterId.wrap(c.name());
    }
    return l;
}
 

import org.apache.cassandra.utils.CounterId; //导入依赖的package包/类
/**
 * Creates a counter context with a single global, 2.1+ shard (a result of increment).
 */
public ByteBuffer createGlobal(CounterId id, long clock, long count)
{
    ContextState state = ContextState.allocate(1, 0, 0);
    state.writeGlobal(id, clock, count);
    return state.context;
}
 

import org.apache.cassandra.utils.CounterId; //导入依赖的package包/类
/**
 * Creates a counter context with a single remote shard.
 * For use by tests of compatibility with pre-2.1 counters only.
 */
public ByteBuffer createRemote(CounterId id, long clock, long count)
{
    ContextState state = ContextState.allocate(0, 0, 1);
    state.writeRemote(id, clock, count);
    return state.context;
}
 

import org.apache.cassandra.utils.CounterId; //导入依赖的package包/类
/**
 * Returns the clock and the count associated with the given counter id, or (0, 0) if no such shard is present.
 */
@VisibleForTesting
public ClockAndCount getClockAndCountOf(ByteBuffer context, CounterId id)
{
    int position = findPositionOf(context, id);
    if (position == -1)
        return ClockAndCount.BLANK;

    long clock = context.getLong(position + CounterId.LENGTH);
    long count = context.getLong(position + CounterId.LENGTH + CLOCK_LENGTH);
    return ClockAndCount.create(clock, count);
}
 

import org.apache.cassandra.utils.CounterId; //导入依赖的package包/类
private void writeElement(CounterId id, long clock, long count, boolean isGlobal, boolean isLocal)
{
    writeElementAtOffset(context, context.position() + bodyOffset, id, clock, count);

    if (isGlobal)
        context.putShort(context.position() + headerOffset, (short) (getElementIndex() + Short.MIN_VALUE));
    else if (isLocal)
        context.putShort(context.position() + headerOffset, (short) getElementIndex());

    currentIsGlobal = isGlobal;
    currentIsLocal = isLocal;
    moveToNext();
}
 

import org.apache.cassandra.utils.CounterId; //导入依赖的package包/类
private void writeElementAtOffset(ByteBuffer ctx, int offset, CounterId id, long clock, long count)
{
    ctx = ctx.duplicate();
    ctx.position(offset);
    ctx.put(id.bytes().duplicate());
    ctx.putLong(clock);
    ctx.putLong(count);
}
 

import org.apache.cassandra.utils.CounterId; //导入依赖的package包/类
@Test
public void testCounterDeletion()
{
    long timestamp = FBUtilities.timestampMicros();
    CellName name = cellname("counter1");

    BufferCounterCell counter = new BufferCounterCell(name,
                                                      CounterContext.instance().createGlobal(CounterId.fromInt(1), 1, 1),
                                                      timestamp);
    BufferDeletedCell tombstone = new BufferDeletedCell(name, (int) (System.currentTimeMillis() / 1000), 0L);

    // check that the tombstone won the reconcile despite the counter cell having a higher timestamp
    assertTrue(counter.reconcile(tombstone) == tombstone);

    // check that a range tombstone overrides the counter cell, even with a lower timestamp than the counter
    ColumnFamily cf0 = ArrayBackedSortedColumns.factory.create("Keyspace1", "Counter1");
    cf0.addColumn(counter);
    cf0.delete(new RangeTombstone(cellname("counter0"), cellname("counter2"), 0L, (int) (System.currentTimeMillis() / 1000)));
    assertTrue(cf0.deletionInfo().isDeleted(counter));
    assertTrue(cf0.deletionInfo().inOrderTester(false).isDeleted(counter));

    // check that a top-level deletion info overrides the counter cell, even with a lower timestamp than the counter
    ColumnFamily cf1 = ArrayBackedSortedColumns.factory.create("Keyspace1", "Counter1");
    cf1.addColumn(counter);
    cf1.delete(new DeletionInfo(0L, (int) (System.currentTimeMillis() / 1000)));
    assertTrue(cf1.deletionInfo().isDeleted(counter));
}
 

import org.apache.cassandra.utils.CounterId; //导入依赖的package包/类
@Test
public void testFindPositionOf()
{
    ContextState state = ContextState.allocate(3, 3, 3);

    state.writeGlobal(CounterId.fromInt(1), 1L, 1L);
    state.writeRemote(CounterId.fromInt(2), 2L, 2L);
    state.writeLocal( CounterId.fromInt(3), 3L, 3L);
    state.writeGlobal(CounterId.fromInt(4), 4L, 4L);
    state.writeRemote(CounterId.fromInt(5), 5L, 5L);
    state.writeLocal( CounterId.fromInt(6), 6L, 6L);
    state.writeGlobal(CounterId.fromInt(7), 7L, 7L);
    state.writeRemote(CounterId.fromInt(8), 8L, 8L);
    state.writeLocal(CounterId.fromInt(9), 9L, 9L);

    int headerLength = headerSizeLength + 6 * headerEltLength;
    assertEquals(headerLength, cc.findPositionOf(state.context, CounterId.fromInt(1)));
    assertEquals(headerLength + stepLength, cc.findPositionOf(state.context, CounterId.fromInt(2)));
    assertEquals(headerLength + 2 * stepLength, cc.findPositionOf(state.context, CounterId.fromInt(3)));
    assertEquals(headerLength + 3 * stepLength, cc.findPositionOf(state.context, CounterId.fromInt(4)));
    assertEquals(headerLength + 4 * stepLength, cc.findPositionOf(state.context, CounterId.fromInt(5)));
    assertEquals(headerLength + 5 * stepLength, cc.findPositionOf(state.context, CounterId.fromInt(6)));
    assertEquals(headerLength + 6 * stepLength, cc.findPositionOf(state.context, CounterId.fromInt(7)));
    assertEquals(headerLength + 7 * stepLength, cc.findPositionOf(state.context, CounterId.fromInt(8)));
    assertEquals(headerLength + 8 * stepLength, cc.findPositionOf(state.context, CounterId.fromInt(9)));

    assertEquals(-1, cc.findPositionOf(state.context, CounterId.fromInt(0)));
    assertEquals(-1, cc.findPositionOf(state.context, CounterId.fromInt(10)));
    assertEquals(-1, cc.findPositionOf(state.context, CounterId.fromInt(15)));
    assertEquals(-1, cc.findPositionOf(state.context, CounterId.fromInt(20)));
}
 

import org.apache.cassandra.utils.CounterId; //导入依赖的package包/类
@Test
public void testGetGlockAndCountOf()
{
    ContextState state = ContextState.allocate(3, 3, 3);

    state.writeGlobal(CounterId.fromInt(1), 1L, 1L);
    state.writeRemote(CounterId.fromInt(2), 2L, 2L);
    state.writeLocal( CounterId.fromInt(3), 3L, 3L);
    state.writeGlobal(CounterId.fromInt(4), 4L, 4L);
    state.writeRemote(CounterId.fromInt(5), 5L, 5L);
    state.writeLocal( CounterId.fromInt(6), 6L, 6L);
    state.writeGlobal(CounterId.fromInt(7), 7L, 7L);
    state.writeRemote(CounterId.fromInt(8), 8L, 8L);
    state.writeLocal(CounterId.fromInt(9), 9L, 9L);

    assertEquals(ClockAndCount.create(1L, 1L), cc.getClockAndCountOf(state.context, CounterId.fromInt(1)));
    assertEquals(ClockAndCount.create(2L, 2L), cc.getClockAndCountOf(state.context, CounterId.fromInt(2)));
    assertEquals(ClockAndCount.create(3L, 3L), cc.getClockAndCountOf(state.context, CounterId.fromInt(3)));
    assertEquals(ClockAndCount.create(4L, 4L), cc.getClockAndCountOf(state.context, CounterId.fromInt(4)));
    assertEquals(ClockAndCount.create(5L, 5L), cc.getClockAndCountOf(state.context, CounterId.fromInt(5)));
    assertEquals(ClockAndCount.create(6L, 6L), cc.getClockAndCountOf(state.context, CounterId.fromInt(6)));
    assertEquals(ClockAndCount.create(7L, 7L), cc.getClockAndCountOf(state.context, CounterId.fromInt(7)));
    assertEquals(ClockAndCount.create(8L, 8L), cc.getClockAndCountOf(state.context, CounterId.fromInt(8)));
    assertEquals(ClockAndCount.create(9L, 9L), cc.getClockAndCountOf(state.context, CounterId.fromInt(9)));

    assertEquals(ClockAndCount.create(0L, 0L), cc.getClockAndCountOf(state.context, CounterId.fromInt(0)));
    assertEquals(ClockAndCount.create(0L, 0L), cc.getClockAndCountOf(state.context, CounterId.fromInt(10)));
    assertEquals(ClockAndCount.create(0L, 0L), cc.getClockAndCountOf(state.context, CounterId.fromInt(15)));
    assertEquals(ClockAndCount.create(0L, 0L), cc.getClockAndCountOf(state.context, CounterId.fromInt(20)));
}
 

import org.apache.cassandra.utils.CounterId; //导入依赖的package包/类
@Test
public void testCounterTable() throws Throwable
{
    /*
     * We can't use CQL to insert counters as both the timestamp and counter ID are automatically assigned and unpredictable.
     * So we need to built it ourselves in a way that is totally equivalent between 2.2 and 3.0 which makes the test a little
     * bit less readable. In any case, the code to generate the equivalent mutation on 2.2 is:
     * ColumnFamily cf = ArrayBackedSortedColumns.factory.create(getCurrentColumnFamilyStore().metadata);
     * ByteBuffer value = CounterContext.instance().createGlobal(CounterId.fromInt(1), 1L, 42L);
     * cf.addColumn(new BufferCounterCell(CellNames.simpleSparse(new ColumnIdentifier("c", true)) , value, 0L, Long.MIN_VALUE));
     * new Mutation(KEYSPACE, ByteBufferUtil.bytes(key), cf).applyUnsafe();
     *
     * Also note that we use COMPACT STORAGE only because it has no bearing on the test and was slightly easier in 2.2 to create
     * the mutation.
     */

    createTable("CREATE TABLE %s (k text PRIMARY KEY, c counter) WITH COMPACT STORAGE");

    String key = "someKey";

    CFMetaData metadata = getCurrentColumnFamilyStore().metadata;
    ColumnDefinition column = metadata.getColumnDefinition(ByteBufferUtil.bytes("c"));
    ByteBuffer value = CounterContext.instance().createGlobal(CounterId.fromInt(1), 1L, 42L);
    Row row = BTreeRow.singleCellRow(Clustering.STATIC_CLUSTERING, BufferCell.live(metadata, column, 0L, value));

    new Mutation(PartitionUpdate.singleRowUpdate(metadata, Util.dk(key), row)).applyUnsafe();

    assertDigest("3a5f7b48c320538b4cd2f829e05c6db3", readAndDigest(key));
}
 

import org.apache.cassandra.utils.CounterId; //导入依赖的package包/类
@Test
public void testCounterDeletion()
{
    long timestamp = FBUtilities.timestampMicros();
    CellName name = cellname("counter1");

    BufferCounterCell counter = new BufferCounterCell(name,
                                                      CounterContext.instance().createGlobal(CounterId.fromInt(1), 1, 1),
                                                      timestamp);
    BufferDeletedCell tombstone = new BufferDeletedCell(name, (int) (System.currentTimeMillis() / 1000), 0L);

    // check that the tombstone won the reconcile despite the counter cell having a higher timestamp
    assertTrue(counter.reconcile(tombstone) == tombstone);

    // check that a range tombstone overrides the counter cell, even with a lower timestamp than the counter
    ColumnFamily cf0 = ArrayBackedSortedColumns.factory.create(KEYSPACE1, CF_COUNTER1);
    cf0.addColumn(counter);
    cf0.delete(new RangeTombstone(cellname("counter0"), cellname("counter2"), 0L, (int) (System.currentTimeMillis() / 1000)));
    assertTrue(cf0.deletionInfo().isDeleted(counter));
    assertTrue(cf0.deletionInfo().inOrderTester(false).isDeleted(counter));

    // check that a top-level deletion info overrides the counter cell, even with a lower timestamp than the counter
    ColumnFamily cf1 = ArrayBackedSortedColumns.factory.create(KEYSPACE1, CF_COUNTER1);
    cf1.addColumn(counter);
    cf1.delete(new DeletionInfo(0L, (int) (System.currentTimeMillis() / 1000)));
    assertTrue(cf1.deletionInfo().isDeleted(counter));
}
 

import org.apache.cassandra.utils.CounterId; //导入依赖的package包/类
/**
 * Write a new current local node id to the system table.
 *
 * @param oldCounterId the previous local node id (that {@code newCounterId}
 * replace) or null if no such node id exists (new node or removed system
 * table)
 * @param newCounterId the new current local node id to record
 * @param now microsecond time stamp.
 */
public static void writeCurrentLocalCounterId(CounterId oldCounterId, CounterId newCounterId, long now)
{
    ByteBuffer ip = ByteBuffer.wrap(FBUtilities.getBroadcastAddress().getAddress());

    ColumnFamily cf = ColumnFamily.create(Table.SYSTEM_KS, COUNTER_ID_CF);
    cf.addColumn(new Column(newCounterId.bytes(), ip, now));
    RowMutation rm = new RowMutation(Table.SYSTEM_KS, ALL_LOCAL_NODE_ID_KEY);
    rm.add(cf);
    rm.apply();
    forceBlockingFlush(COUNTER_ID_CF);
}
 

import org.apache.cassandra.utils.CounterId; //导入依赖的package包/类
@Before
public void setUp() throws Exception {
    conf = new Configuration();
    attemptId = new TaskAttemptID();
    Path inputPath = new Path(TABLE_PATH_STR);
    inputSplit = new FileSplit(inputPath, 0, 1, null);
    Descriptor desc = new Descriptor(new File(TABLE_PATH_STR), "keyspace", "columnFamily", 1,
                                     Type.FINAL);

    doReturn(desc).when(ssTableColumnRecordReader).getDescriptor();

    doNothing().when(ssTableColumnRecordReader).copyTablesToLocal(any(FileSystem.class),
                                                                  any(FileSystem.class),
                                                                  any(Path.class),
                                                                  any(TaskAttemptContext.class));

    doReturn(ssTableReader).when(ssTableColumnRecordReader)
        .openSSTableReader(any(IPartitioner.class), any(CFMetaData.class));

    when(ssTableReader.estimatedKeys()).thenReturn(2L);
    when(ssTableReader.getScanner()).thenReturn(tableScanner);

    when(tableScanner.hasNext()).thenReturn(true, true, false);

    key = new BufferDecoratedKey(new StringToken("a"), ByteBuffer.wrap("b".getBytes()));
    CellNameType simpleDenseCellType = new SimpleDenseCellNameType(BytesType.instance);
    CellName cellName = simpleDenseCellType.cellFromByteBuffer(ByteBuffer.wrap("n".getBytes()));
    ByteBuffer counterBB = CounterContext.instance()
        .createGlobal(CounterId.fromInt(0), System.currentTimeMillis(), 123L);
    value = BufferCounterCell.create(cellName, counterBB, System.currentTimeMillis(), 0L,
                                     Flag.PRESERVE_SIZE);

    SSTableIdentityIterator row1 = getNewRow();
    SSTableIdentityIterator row2 = getNewRow();

    when(tableScanner.next()).thenReturn(row1, row2);
}
 

import org.apache.cassandra.utils.CounterId; //导入依赖的package包/类
private static int compareId(ByteBuffer bb1, int pos1, ByteBuffer bb2, int pos2)
{
    return ByteBufferUtil.compareSubArrays(bb1, pos1, bb2, pos2, CounterId.LENGTH);
}
 

import org.apache.cassandra.utils.CounterId; //导入依赖的package包/类
/**
 * Returns the clock and the count associated with the local counter id, or (0, 0) if no such shard is present.
 */
public ClockAndCount getLocalClockAndCount(ByteBuffer context)
{
    return getClockAndCountOf(context, CounterId.getLocalId());
}
 

import org.apache.cassandra.utils.CounterId; //导入依赖的package包/类
public CounterId getCounterId()
{
    return CounterId.wrap(context, context.position() + bodyOffset);
}