import org.apache.cassandra.db.DeletionTime; //导入依赖的package包/类
public void skip(DataInputPlus in) throws IOException
{
    clusteringSerializer.skip(in);
    clusteringSerializer.skip(in);
    if (version.storeRows())
    {
        in.readUnsignedVInt();
        in.readVInt();
        if (in.readBoolean())
            DeletionTime.serializer.skip(in);
    }
    else
    {
        in.skipBytes(TypeSizes.sizeof(0L));
        in.skipBytes(TypeSizes.sizeof(0L));
    }
}
 

import org.apache.cassandra.db.DeletionTime; //导入依赖的package包/类
public IndexInfo deserialize(DataInputPlus in) throws IOException
{
    ClusteringPrefix firstName = clusteringSerializer.deserialize(in);
    ClusteringPrefix lastName = clusteringSerializer.deserialize(in);
    long offset;
    long width;
    DeletionTime endOpenMarker = null;
    if (version.storeRows())
    {
        offset = in.readUnsignedVInt();
        width = in.readVInt() + WIDTH_BASE;
        if (in.readBoolean())
            endOpenMarker = DeletionTime.serializer.deserialize(in);
    }
    else
    {
        offset = in.readLong();
        width = in.readLong();
    }
    return new IndexInfo(firstName, lastName, offset, width, endOpenMarker);
}
 

import org.apache.cassandra.db.DeletionTime; //导入依赖的package包/类
private static Row.Builder createBuilder(Clustering c, int now, ByteBuffer vVal, ByteBuffer mKey, ByteBuffer mVal)
{
    long ts = secondToTs(now);
    Row.Builder builder = BTreeRow.unsortedBuilder(now);
    builder.newRow(c);
    builder.addPrimaryKeyLivenessInfo(LivenessInfo.create(kcvm, ts, now));
    if (vVal != null)
    {
        builder.addCell(BufferCell.live(kcvm, v, ts, vVal));
    }
    if (mKey != null && mVal != null)
    {
        builder.addComplexDeletion(m, new DeletionTime(ts - 1, now));
        builder.addCell(BufferCell.live(kcvm, m, ts, mVal, CellPath.create(mKey)));
    }

    return builder;
}
 

import org.apache.cassandra.db.DeletionTime; //导入依赖的package包/类
@Test
public void mergeComplexDeletionSupersededByRowDeletion()
{
    int now1 = FBUtilities.nowInSeconds();
    Row.Builder existingBuilder = createBuilder(c1, now1, null, null, null);

    int now2 = now1 + 1;
    Row.Builder updateBuilder = createBuilder(c1, now2, null, BB1, BB1);
    int now3 = now2 + 1;
    Row.Deletion expectedDeletion = new Row.Deletion(new DeletionTime(secondToTs(now3), now3), false);
    updateBuilder.addRowDeletion(expectedDeletion);

    RowBuilder builder = new RowBuilder();
    Rows.merge(existingBuilder.build(), updateBuilder.build(), builder, now3 + 1);

    Assert.assertEquals(expectedDeletion, builder.deletionTime);
    Assert.assertEquals(Collections.emptyList(), builder.complexDeletions);
    Assert.assertEquals(Collections.emptyList(), builder.cells);
}
 

import org.apache.cassandra.db.DeletionTime; //导入依赖的package包/类
/**
 * If a row's deletion time deletes a row's liveness info, the new row should have it's
 * liveness info set to empty
 */
@Test
public void mergeRowDeletionSupercedesLiveness()
{
    int now1 = FBUtilities.nowInSeconds();
    Row.Builder existingBuilder = createBuilder(c1, now1, null, null, null);

    int now2 = now1 + 1;
    Row.Builder updateBuilder = createBuilder(c1, now2, BB1, BB1, BB1);
    int now3 = now2 + 1;
    Row.Deletion expectedDeletion = new Row.Deletion(new DeletionTime(secondToTs(now3), now3), false);
    updateBuilder.addRowDeletion(expectedDeletion);

    RowBuilder builder = new RowBuilder();
    Rows.merge(existingBuilder.build(), updateBuilder.build(), builder, now3 + 1);

    Assert.assertEquals(expectedDeletion, builder.deletionTime);
    Assert.assertEquals(LivenessInfo.EMPTY, builder.livenessInfo);
    Assert.assertEquals(Collections.emptyList(), builder.complexDeletions);
    Assert.assertEquals(Collections.emptyList(), builder.cells);
}
 

import org.apache.cassandra.db.DeletionTime; //导入依赖的package包/类
/**
 * Reconciles/merges two cells, one being an update to an existing cell,
 * yielding index updates if appropriate.
 * <p>
 * Note that this method assumes that the provided cells can meaningfully
 * be reconciled together, that is that those cells are for the same row and same
 * column (and same cell path if the column is complex).
 * <p>
 * Also note that which cell is provided as {@code existing} and which is
 * provided as {@code update} matters for index updates.
 *
 * @param existing the pre-existing cell, the one that is updated. This can be
 * {@code null} if this reconciliation correspond to an insertion.
 * @param update the newly added cell, the update. This can be {@code null} out
 * of convenience, in which case this function simply copy {@code existing} to
 * {@code writer}.
 * @param deletion the deletion time that applies to the cells being considered.
 * This deletion time may delete both {@code existing} or {@code update}.
 * @param builder the row builder to which the result of the reconciliation is written.
 * @param nowInSec the current time in seconds (which plays a role during reconciliation
 * because deleted cells always have precedence on timestamp equality and deciding if a
 * cell is a live or not depends on the current time due to expiring cells).
 *
 * @return the timestamp delta between existing and update, or {@code Long.MAX_VALUE} if one
 * of them is {@code null} or deleted by {@code deletion}).
 */
public static long reconcile(Cell existing,
                             Cell update,
                             DeletionTime deletion,
                             Row.Builder builder,
                             int nowInSec)
{
    existing = existing == null || deletion.deletes(existing) ? null : existing;
    update = update == null || deletion.deletes(update) ? null : update;
    if (existing == null || update == null)
    {
        if (update != null)
        {
            builder.addCell(update);
        }
        else if (existing != null)
        {
            builder.addCell(existing);
        }
        return Long.MAX_VALUE;
    }

    Cell reconciled = reconcile(existing, update, nowInSec);
    builder.addCell(reconciled);

    return Math.abs(existing.timestamp() - update.timestamp());
}
 

import org.apache.cassandra.db.DeletionTime; //导入依赖的package包/类
/**
 * Computes the reconciliation of a complex column given its pre-existing
 * cells and the ones it is updated with, and generating index update if
 * appropriate.
 * <p>
 * Note that this method assumes that the provided cells can meaningfully
 * be reconciled together, that is that the cells are for the same row and same
 * complex column.
 * <p>
 * Also note that which cells is provided as {@code existing} and which are
 * provided as {@code update} matters for index updates.
 *
 * @param column the complex column the cells are for.
 * @param existing the pre-existing cells, the ones that are updated. This can be
 * {@code null} if this reconciliation correspond to an insertion.
 * @param update the newly added cells, the update. This can be {@code null} out
 * of convenience, in which case this function simply copy the cells from
 * {@code existing} to {@code writer}.
 * @param deletion the deletion time that applies to the cells being considered.
 * This deletion time may delete cells in both {@code existing} and {@code update}.
 * @param builder the row build to which the result of the reconciliation is written.
 * @param nowInSec the current time in seconds (which plays a role during reconciliation
 * because deleted cells always have precedence on timestamp equality and deciding if a
 * cell is a live or not depends on the current time due to expiring cells).
 *
 * @return the smallest timestamp delta between corresponding cells from existing and update. A
 * timestamp delta being computed as the difference between a cell from {@code update} and the
 * cell in {@code existing} having the same cell path (if such cell exists). If the intersection
 * of cells from {@code existing} and {@code update} having the same cell path is empty, this
 * returns {@code Long.MAX_VALUE}.
 */
public static long reconcileComplex(ColumnDefinition column,
                                    Iterator<Cell> existing,
                                    Iterator<Cell> update,
                                    DeletionTime deletion,
                                    Row.Builder builder,
                                    int nowInSec)
{
    Comparator<CellPath> comparator = column.cellPathComparator();
    Cell nextExisting = getNext(existing);
    Cell nextUpdate = getNext(update);
    long timeDelta = Long.MAX_VALUE;
    while (nextExisting != null || nextUpdate != null)
    {
        int cmp = nextExisting == null ? 1
                 : (nextUpdate == null ? -1
                 : comparator.compare(nextExisting.path(), nextUpdate.path()));
        if (cmp < 0)
        {
            reconcile(nextExisting, null, deletion, builder, nowInSec);
            nextExisting = getNext(existing);
        }
        else if (cmp > 0)
        {
            reconcile(null, nextUpdate, deletion, builder, nowInSec);
            nextUpdate = getNext(update);
        }
        else
        {
            timeDelta = Math.min(timeDelta, reconcile(nextExisting, nextUpdate, deletion, builder, nowInSec));
            nextExisting = getNext(existing);
            nextUpdate = getNext(update);
        }
    }
    return timeDelta;
}
 

import org.apache.cassandra.db.DeletionTime; //导入依赖的package包/类
/**
 * Adds to the builder a representation of the given existing cell that, when merged/reconciled with the given
 * update cell, produces the same result as merging the original with the update.
 * <p>
 * For simple cells that is either the original cell (if still live), or nothing (if shadowed).
 *
 * @param column the complex column the cells are for.
 * @param existing the pre-existing cells, the ones that are updated.
 * @param update the newly added cells, the update. This can be {@code null} out
 * of convenience, in which case this function simply copy the cells from
 * {@code existing} to {@code writer}.
 * @param deletion the deletion time that applies to the cells being considered.
 * This deletion time may delete both {@code existing} or {@code update}.
 * @param builder the row builder to which the result of the filtering is written.
 * @param nowInSec the current time in seconds (which plays a role during reconciliation
 * because deleted cells always have precedence on timestamp equality and deciding if a
 * cell is a live or not depends on the current time due to expiring cells).
 */
public static void addNonShadowedComplex(ColumnDefinition column,
                                         Iterator<Cell> existing,
                                         Iterator<Cell> update,
                                         DeletionTime deletion,
                                         Row.Builder builder,
                                         int nowInSec)
{
    Comparator<CellPath> comparator = column.cellPathComparator();
    Cell nextExisting = getNext(existing);
    Cell nextUpdate = getNext(update);
    while (nextExisting != null)
    {
        int cmp = nextUpdate == null ? -1 : comparator.compare(nextExisting.path(), nextUpdate.path());
        if (cmp < 0)
        {
            addNonShadowed(nextExisting, null, deletion, builder, nowInSec);
            nextExisting = getNext(existing);
        }
        else if (cmp == 0)
        {
            addNonShadowed(nextExisting, nextUpdate, deletion, builder, nowInSec);
            nextExisting = getNext(existing);
            nextUpdate = getNext(update);
        }
        else
        {
            nextUpdate = getNext(update);
        }
    }
}
 

import org.apache.cassandra.db.DeletionTime; //导入依赖的package包/类
ComplexColumnData(ColumnDefinition column, Object[] cells, DeletionTime complexDeletion)
{
    super(column);
    assert column.isComplex();
    assert cells.length > 0 || !complexDeletion.isLive();
    this.cells = cells;
    this.complexDeletion = complexDeletion;
}
 

import org.apache.cassandra.db.DeletionTime; //导入依赖的package包/类
private ComplexColumnData transformAndFilter(DeletionTime newDeletion, Function<? super Cell, ? extends Cell> function)
{
    Object[] transformed = BTree.transformAndFilter(cells, function);

    if (cells == transformed && newDeletion == complexDeletion)
        return this;

    if (newDeletion == DeletionTime.LIVE && BTree.isEmpty(transformed))
        return null;

    return new ComplexColumnData(column, transformed, newDeletion);
}
 

import org.apache.cassandra.db.DeletionTime; //导入依赖的package包/类
public IndexInfo(ClusteringPrefix firstName,
                 ClusteringPrefix lastName,
                 long offset,
                 long width,
                 DeletionTime endOpenMarker)
{
    this.firstName = firstName;
    this.lastName = lastName;
    this.offset = offset;
    this.width = width;
    this.endOpenMarker = endOpenMarker;
}
 

import org.apache.cassandra.db.DeletionTime; //导入依赖的package包/类
@Test
public void testKeyCacheValueWithDelInfo()
{
    RowIndexEntry entry = RowIndexEntry.create(123, new DeletionTime(123, 123), ColumnIndex.nothing());
    long size = entry.memorySize();
    long size2 = meter.measureDeep(entry);
    Assert.assertEquals(size, size2);
}
 

import org.apache.cassandra.db.DeletionTime; //导入依赖的package包/类
public void newColumn(ColumnDefinition column)
{
    this.column = column;
    this.complexDeletion = DeletionTime.LIVE; // default if writeComplexDeletion is not called
    if (builder == null) builder = BTree.builder(column.cellComparator());
    else builder.reuse(column.cellComparator());
}
 

import org.apache.cassandra.db.DeletionTime; //导入依赖的package包/类
private void serializeBound(RangeTombstone.Bound bound, DeletionTime deletionTime) throws IOException
{
    json.writeFieldName(bound.isStart() ? "start" : "end");
    json.writeStartObject();
    json.writeFieldName("type");
    json.writeString(bound.isInclusive() ? "inclusive" : "exclusive");
    serializeClustering(bound.clustering());
    serializeDeletion(deletionTime);
    json.writeEndObject();
}
 

import org.apache.cassandra.db.DeletionTime; //导入依赖的package包/类
private void serializeDeletion(DeletionTime deletion) throws IOException
{
    json.writeFieldName("deletion_info");
    objectIndenter.setCompact(true);
    json.writeStartObject();
    json.writeFieldName("marked_deleted");
    json.writeString(dateString(TimeUnit.MICROSECONDS, deletion.markedForDeleteAt()));
    json.writeFieldName("local_delete_time");
    json.writeString(dateString(TimeUnit.SECONDS, deletion.localDeletionTime()));
    json.writeEndObject();
    objectIndenter.setCompact(false);
}
 

import org.apache.cassandra.db.DeletionTime; //导入依赖的package包/类
private void process(Object callback, UnfilteredRowIterator rows) {
    CFMetaData cfMetaData = rows.metadata();
    DeletionTime deletionTime = rows.partitionLevelDeletion();
    DecoratedKey key = rows.partitionKey();

    begin(callback, key, cfMetaData);

    if (!deletionTime.isLive()) {
        deletePartition(key, deletionTime);
        return;
    }

    Row sr = rows.staticRow();
    if (sr != null) {
        process(sr);
    }
    
    while (rows.hasNext()) {
        Unfiltered f = rows.next();
        switch (f.kind()) {
        case RANGE_TOMBSTONE_MARKER:
            process((RangeTombstoneMarker) f);
            break;
        case ROW:
            process((Row)f);
            break;
        default:
            break;                
        }
    }
}
 

import org.apache.cassandra.db.DeletionTime; //导入依赖的package包/类
public void onComplexDeletion(int i, Clustering clustering, ColumnDefinition column, DeletionTime merged, DeletionTime original)
{
    updateClustering(clustering);
    if (!complexDeletions.containsKey(column)) complexDeletions.put(column, new LinkedList<>());
    complexDeletions.get(column).add(MergedPair.create(i, merged, original));
    updates++;
}
 

import org.apache.cassandra.db.DeletionTime; //导入依赖的package包/类
@Test
public void copy()
{
    int now = FBUtilities.nowInSeconds();
    long ts = secondToTs(now);
    Row.Builder originalBuilder = BTreeRow.unsortedBuilder(now);
    originalBuilder.newRow(c1);
    LivenessInfo liveness = LivenessInfo.create(kcvm, ts, now);
    originalBuilder.addPrimaryKeyLivenessInfo(liveness);
    DeletionTime complexDeletion = new DeletionTime(ts-1, now);
    originalBuilder.addComplexDeletion(m, complexDeletion);
    List<Cell> expectedCells = Lists.newArrayList(BufferCell.live(kcvm, v, secondToTs(now), BB1),
                                                  BufferCell.live(kcvm, m, secondToTs(now), BB1, CellPath.create(BB1)),
                                                  BufferCell.live(kcvm, m, secondToTs(now), BB2, CellPath.create(BB2)));
    expectedCells.forEach(originalBuilder::addCell);
    // We need to use ts-1 so the deletion doesn't shadow what we've created
    Row.Deletion rowDeletion = new Row.Deletion(new DeletionTime(ts-1, now), false);
    originalBuilder.addRowDeletion(rowDeletion);

    RowBuilder builder = new RowBuilder();
    Rows.copy(originalBuilder.build(), builder);

    Assert.assertEquals(c1, builder.clustering);
    Assert.assertEquals(liveness, builder.livenessInfo);
    Assert.assertEquals(rowDeletion, builder.deletionTime);
    Assert.assertEquals(Lists.newArrayList(Pair.create(m, complexDeletion)), builder.complexDeletions);
    Assert.assertEquals(Sets.newHashSet(expectedCells), Sets.newHashSet(builder.cells));
}
 

import org.apache.cassandra.db.DeletionTime; //导入依赖的package包/类
@Test
public void collectStats()
{
    int now = FBUtilities.nowInSeconds();
    long ts = secondToTs(now);
    Row.Builder builder = BTreeRow.unsortedBuilder(now);
    builder.newRow(c1);
    LivenessInfo liveness = LivenessInfo.create(kcvm, ts, now);
    builder.addPrimaryKeyLivenessInfo(liveness);
    DeletionTime complexDeletion = new DeletionTime(ts-1, now);
    builder.addComplexDeletion(m, complexDeletion);
    List<Cell> expectedCells = Lists.newArrayList(BufferCell.live(kcvm, v, ts, BB1),
                                                  BufferCell.live(kcvm, m, ts, BB1, CellPath.create(BB1)),
                                                  BufferCell.live(kcvm, m, ts, BB2, CellPath.create(BB2)));
    expectedCells.forEach(builder::addCell);
    // We need to use ts-1 so the deletion doesn't shadow what we've created
    Row.Deletion rowDeletion = new Row.Deletion(new DeletionTime(ts-1, now), false);
    builder.addRowDeletion(rowDeletion);

    StatsCollector collector = new StatsCollector();
    Rows.collectStats(builder.build(), collector);

    Assert.assertEquals(Lists.newArrayList(liveness), collector.liveness);
    Assert.assertEquals(Sets.newHashSet(rowDeletion.time(), complexDeletion), Sets.newHashSet(collector.deletions));
    Assert.assertEquals(Sets.newHashSet(expectedCells), Sets.newHashSet(collector.cells));
    Assert.assertEquals(2, collector.columnCount);
    Assert.assertFalse(collector.hasLegacyCounterShards);
}
 

import org.apache.cassandra.db.DeletionTime; //导入依赖的package包/类
@Test
public void merge()
{
    int now1 = FBUtilities.nowInSeconds();
    Row.Builder existingBuilder = createBuilder(c1, now1, BB1, BB1, BB1);

    int now2 = now1 + 1;
    long ts2 = secondToTs(now2);

    Cell expectedVCell = BufferCell.live(kcvm, v, ts2, BB2);
    Cell expectedMCell = BufferCell.live(kcvm, m, ts2, BB2, CellPath.create(BB1));
    DeletionTime expectedComplexDeletionTime = new DeletionTime(ts2 - 1, now2);

    Row.Builder updateBuilder = createBuilder(c1, now2, null, null, null);
    updateBuilder.addCell(expectedVCell);
    updateBuilder.addComplexDeletion(m, expectedComplexDeletionTime);
    updateBuilder.addCell(expectedMCell);

    RowBuilder builder = new RowBuilder();
    long td = Rows.merge(existingBuilder.build(), updateBuilder.build(), builder, now2 + 1);

    Assert.assertEquals(c1, builder.clustering);
    Assert.assertEquals(LivenessInfo.create(kcvm, ts2, now2), builder.livenessInfo);
    Assert.assertEquals(Lists.newArrayList(Pair.create(m, new DeletionTime(ts2-1, now2))), builder.complexDeletions);

    Assert.assertEquals(2, builder.cells.size());
    Assert.assertEquals(Lists.newArrayList(expectedVCell, expectedMCell), Lists.newArrayList(builder.cells));
    Assert.assertEquals(ts2 - secondToTs(now1), td);
}