mirror of https://github.com/apache/cassandra
Adds "shadowable" row tombstones for MV timestamp issues
The patch allow row tombstones to be either regular (as before this patch) or shadowable. If a row tombstone is shadowable, it only applies to the row and its content as long as the row timestamp is less than the tombstone timestamp. As soon as the row is updated with a timestamp greater than the one of the shadowable tombstone, that tombstone is removed, even if some cells inside the row may have older timestamps. Patch by Sylvain Lebresne and tjake; reviewed by Tyler Hobbs for CASSANDRA-10261
This commit is contained in:
Sylvain Lebresne
T Jake Luciani
parent
30f39a2312
commit
665f74740c
|
|
@ -1,4 +1,5 @@
|
|||
3.0.0-rc1
|
||||
* Add "shadowable" row tombstones to deal with mv timestamp issues (CASSANDRA-10261)
|
||||
* CFS.loadNewSSTables() broken for pre-3.0 sstables
|
||||
* Cache selected index in read command to reduce lookups (CASSANDRA-10215)
|
||||
* Small optimizations of sstable index serialization (CASSANDRA-10232)
|
||||
|
|
|
|||
|
|
@ -153,7 +153,7 @@ public class UpdateParameters
|
|||
if (metadata.isCompactTable() && builder.clustering() != Clustering.STATIC_CLUSTERING)
|
||||
addTombstone(metadata.compactValueColumn());
|
||||
else
|
||||
builder.addRowDeletion(deletionTime);
|
||||
builder.addRowDeletion(Row.Deletion.regular(deletionTime));
|
||||
}
|
||||
|
||||
public void addTombstone(ColumnDefinition column) throws InvalidRequestException
|
||||
|
|
|
|||
|
|
@ -416,9 +416,9 @@ public interface ClusteringPrefix extends IMeasurableMemory, Clusterable
|
|||
this.serializationHeader = header;
|
||||
}
|
||||
|
||||
public void prepare(int flags) throws IOException
|
||||
public void prepare(int flags, int extendedFlags) throws IOException
|
||||
{
|
||||
assert !UnfilteredSerializer.isStatic(flags) : "Flags = " + flags;
|
||||
assert !UnfilteredSerializer.isStatic(extendedFlags) : "Flags = " + flags;
|
||||
this.nextIsRow = UnfilteredSerializer.kind(flags) == Unfiltered.Kind.ROW;
|
||||
this.nextKind = nextIsRow ? Kind.CLUSTERING : ClusteringPrefix.Kind.values()[in.readByte()];
|
||||
this.nextSize = nextIsRow ? comparator.size() : in.readUnsignedShort();
|
||||
|
|
|
|||
|
|
@ -682,7 +682,7 @@ public abstract class LegacyLayout
|
|||
LegacyBound start = new LegacyLayout.LegacyBound(startBound, false, null);
|
||||
LegacyBound end = new LegacyLayout.LegacyBound(endBound, false, null);
|
||||
|
||||
deletions.add(start, end, row.deletion().markedForDeleteAt(), row.deletion().localDeletionTime());
|
||||
deletions.add(start, end, row.deletion().time().markedForDeleteAt(), row.deletion().time().localDeletionTime());
|
||||
}
|
||||
|
||||
for (ColumnData cd : row)
|
||||
|
|
@ -1155,7 +1155,7 @@ public abstract class LegacyLayout
|
|||
|
||||
clustering = tombstone.start.getAsClustering(metadata);
|
||||
builder.newRow(clustering);
|
||||
builder.addRowDeletion(tombstone.deletionTime);
|
||||
builder.addRowDeletion(Row.Deletion.regular(tombstone.deletionTime));
|
||||
rowDeletion = tombstone;
|
||||
return true;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -194,7 +194,7 @@ public class RowUpdateBuilder
|
|||
builder.newRow(Clustering.STATIC_CLUSTERING);
|
||||
else
|
||||
builder.newRow(clusteringValues.length == 0 ? Clustering.EMPTY : update.metadata().comparator.make(clusteringValues));
|
||||
builder.addRowDeletion(new DeletionTime(timestamp, localDeletionTime));
|
||||
builder.addRowDeletion(Row.Deletion.regular(new DeletionTime(timestamp, localDeletionTime)));
|
||||
|
||||
update.add(builder.build());
|
||||
}
|
||||
|
|
|
|||
|
|
@ -108,6 +108,7 @@ public abstract class UnfilteredDeserializer
|
|||
private final SerializationHeader header;
|
||||
|
||||
private int nextFlags;
|
||||
private int nextExtendedFlags;
|
||||
private boolean isReady;
|
||||
private boolean isDone;
|
||||
|
||||
|
|
@ -146,7 +147,9 @@ public abstract class UnfilteredDeserializer
|
|||
return;
|
||||
}
|
||||
|
||||
clusteringDeserializer.prepare(nextFlags);
|
||||
nextExtendedFlags = UnfilteredSerializer.isExtended(nextFlags) ? in.readUnsignedByte() : 0;
|
||||
|
||||
clusteringDeserializer.prepare(nextFlags, nextExtendedFlags);
|
||||
isReady = true;
|
||||
}
|
||||
|
||||
|
|
@ -185,7 +188,7 @@ public abstract class UnfilteredDeserializer
|
|||
else
|
||||
{
|
||||
builder.newRow(clusteringDeserializer.deserializeNextClustering());
|
||||
return UnfilteredSerializer.serializer.deserializeRowBody(in, header, helper, nextFlags, builder);
|
||||
return UnfilteredSerializer.serializer.deserializeRowBody(in, header, helper, nextFlags, nextExtendedFlags, builder);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -199,7 +202,7 @@ public abstract class UnfilteredDeserializer
|
|||
}
|
||||
else
|
||||
{
|
||||
UnfilteredSerializer.serializer.skipRowBody(in, header, nextFlags);
|
||||
UnfilteredSerializer.serializer.skipRowBody(in, header, nextFlags, nextExtendedFlags);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -160,7 +160,7 @@ public abstract class AbstractBTreePartition implements Partition, Iterable<Row>
|
|||
activeDeletion = rt.deletionTime();
|
||||
|
||||
if (row == null)
|
||||
return activeDeletion.isLive() ? null : BTreeRow.emptyDeletedRow(clustering, activeDeletion);
|
||||
return activeDeletion.isLive() ? null : BTreeRow.emptyDeletedRow(clustering, Row.Deletion.regular(activeDeletion));
|
||||
|
||||
return row.filter(columns, activeDeletion, true, metadata);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -78,7 +78,7 @@ public abstract class AbstractRow extends AbstractCollection<ColumnData> impleme
|
|||
}
|
||||
|
||||
primaryKeyLivenessInfo().validate();
|
||||
if (deletion().localDeletionTime() < 0)
|
||||
if (deletion().time().localDeletionTime() < 0)
|
||||
throw new MarshalException("A local deletion time should not be negative");
|
||||
|
||||
for (ColumnData cd : this)
|
||||
|
|
|
|||
|
|
@ -48,7 +48,7 @@ public class BTreeRow extends AbstractRow
|
|||
|
||||
private final Clustering clustering;
|
||||
private final LivenessInfo primaryKeyLivenessInfo;
|
||||
private final DeletionTime deletion;
|
||||
private final Deletion deletion;
|
||||
|
||||
// The data for each columns present in this row in column sorted order.
|
||||
private final Object[] btree;
|
||||
|
|
@ -62,8 +62,9 @@ public class BTreeRow extends AbstractRow
|
|||
// no expiring cells, this will be Integer.MAX_VALUE;
|
||||
private final int minLocalDeletionTime;
|
||||
|
||||
private BTreeRow(Clustering clustering, LivenessInfo primaryKeyLivenessInfo, DeletionTime deletion, Object[] btree, int minLocalDeletionTime)
|
||||
private BTreeRow(Clustering clustering, LivenessInfo primaryKeyLivenessInfo, Deletion deletion, Object[] btree, int minLocalDeletionTime)
|
||||
{
|
||||
assert !deletion.isShadowedBy(primaryKeyLivenessInfo);
|
||||
this.clustering = clustering;
|
||||
this.primaryKeyLivenessInfo = primaryKeyLivenessInfo;
|
||||
this.deletion = deletion;
|
||||
|
|
@ -71,10 +72,15 @@ public class BTreeRow extends AbstractRow
|
|||
this.minLocalDeletionTime = minLocalDeletionTime;
|
||||
}
|
||||
|
||||
// Note that it's often easier/safer to use the sortedBuilder/unsortedBuilder or one of the static creation method below. Only directly useful in a small amount of cases.
|
||||
public static BTreeRow create(Clustering clustering, LivenessInfo primaryKeyLivenessInfo, DeletionTime deletion, Object[] btree)
|
||||
private BTreeRow(Clustering clustering, Object[] btree, int minLocalDeletionTime)
|
||||
{
|
||||
int minDeletionTime = Math.min(minDeletionTime(primaryKeyLivenessInfo), minDeletionTime(deletion));
|
||||
this(clustering, LivenessInfo.EMPTY, Deletion.LIVE, btree, minLocalDeletionTime);
|
||||
}
|
||||
|
||||
// Note that it's often easier/safer to use the sortedBuilder/unsortedBuilder or one of the static creation method below. Only directly useful in a small amount of cases.
|
||||
public static BTreeRow create(Clustering clustering, LivenessInfo primaryKeyLivenessInfo, Deletion deletion, Object[] btree)
|
||||
{
|
||||
int minDeletionTime = Math.min(minDeletionTime(primaryKeyLivenessInfo), minDeletionTime(deletion.time()));
|
||||
if (minDeletionTime != Integer.MIN_VALUE)
|
||||
{
|
||||
for (ColumnData cd : BTree.<ColumnData>iterable(btree))
|
||||
|
|
@ -86,19 +92,19 @@ public class BTreeRow extends AbstractRow
|
|||
|
||||
public static BTreeRow emptyRow(Clustering clustering)
|
||||
{
|
||||
return new BTreeRow(clustering, LivenessInfo.EMPTY, DeletionTime.LIVE, BTree.empty(), Integer.MAX_VALUE);
|
||||
return new BTreeRow(clustering, BTree.empty(), Integer.MAX_VALUE);
|
||||
}
|
||||
|
||||
public static BTreeRow singleCellRow(Clustering clustering, Cell cell)
|
||||
{
|
||||
if (cell.column().isSimple())
|
||||
return new BTreeRow(clustering, LivenessInfo.EMPTY, DeletionTime.LIVE, BTree.singleton(cell), minDeletionTime(cell));
|
||||
return new BTreeRow(clustering, BTree.singleton(cell), minDeletionTime(cell));
|
||||
|
||||
ComplexColumnData complexData = new ComplexColumnData(cell.column(), new Cell[]{ cell }, DeletionTime.LIVE);
|
||||
return new BTreeRow(clustering, LivenessInfo.EMPTY, DeletionTime.LIVE, BTree.singleton(complexData), minDeletionTime(cell));
|
||||
return new BTreeRow(clustering, BTree.singleton(complexData), minDeletionTime(cell));
|
||||
}
|
||||
|
||||
public static BTreeRow emptyDeletedRow(Clustering clustering, DeletionTime deletion)
|
||||
public static BTreeRow emptyDeletedRow(Clustering clustering, Deletion deletion)
|
||||
{
|
||||
assert !deletion.isLive();
|
||||
return new BTreeRow(clustering, LivenessInfo.EMPTY, deletion, BTree.empty(), Integer.MIN_VALUE);
|
||||
|
|
@ -107,7 +113,7 @@ public class BTreeRow extends AbstractRow
|
|||
public static BTreeRow noCellLiveRow(Clustering clustering, LivenessInfo primaryKeyLivenessInfo)
|
||||
{
|
||||
assert !primaryKeyLivenessInfo.isEmpty();
|
||||
return new BTreeRow(clustering, primaryKeyLivenessInfo, DeletionTime.LIVE, BTree.empty(), minDeletionTime(primaryKeyLivenessInfo));
|
||||
return new BTreeRow(clustering, primaryKeyLivenessInfo, Deletion.LIVE, BTree.empty(), minDeletionTime(primaryKeyLivenessInfo));
|
||||
}
|
||||
|
||||
private static int minDeletionTime(Cell cell)
|
||||
|
|
@ -176,7 +182,7 @@ public class BTreeRow extends AbstractRow
|
|||
&& BTree.isEmpty(btree);
|
||||
}
|
||||
|
||||
public DeletionTime deletion()
|
||||
public Deletion deletion()
|
||||
{
|
||||
return deletion;
|
||||
}
|
||||
|
|
@ -234,17 +240,17 @@ public class BTreeRow extends AbstractRow
|
|||
if (filter.includesAllColumns() && (activeDeletion.isLive() || deletion.supersedes(activeDeletion)) && droppedColumns.isEmpty())
|
||||
return this;
|
||||
|
||||
boolean mayHaveShadowed = activeDeletion.supersedes(deletion);
|
||||
boolean mayHaveShadowed = activeDeletion.supersedes(deletion.time());
|
||||
|
||||
LivenessInfo newInfo = primaryKeyLivenessInfo;
|
||||
DeletionTime newDeletion = deletion;
|
||||
Deletion newDeletion = deletion;
|
||||
if (mayHaveShadowed)
|
||||
{
|
||||
if (activeDeletion.deletes(newInfo.timestamp()))
|
||||
newInfo = LivenessInfo.EMPTY;
|
||||
// note that mayHaveShadowed means the activeDeletion shadows the row deletion. So if don't have setActiveDeletionToRow,
|
||||
// the row deletion is shadowed and we shouldn't return it.
|
||||
newDeletion = setActiveDeletionToRow ? activeDeletion : DeletionTime.LIVE;
|
||||
newDeletion = setActiveDeletionToRow ? Deletion.regular(activeDeletion) : Deletion.LIVE;
|
||||
}
|
||||
|
||||
Columns columns = filter.fetchedColumns().columns(isStatic());
|
||||
|
|
@ -307,7 +313,11 @@ public class BTreeRow extends AbstractRow
|
|||
public Row updateAllTimestamp(long newTimestamp)
|
||||
{
|
||||
LivenessInfo newInfo = primaryKeyLivenessInfo.isEmpty() ? primaryKeyLivenessInfo : primaryKeyLivenessInfo.withUpdatedTimestamp(newTimestamp);
|
||||
DeletionTime newDeletion = deletion.isLive() ? deletion : new DeletionTime(newTimestamp - 1, deletion.localDeletionTime());
|
||||
// If the deletion is shadowable and the row has a timestamp, we'll forced the deletion timestamp to be less than the row one, so we
|
||||
// should get rid of said deletion.
|
||||
Deletion newDeletion = deletion.isLive() || (deletion.isShadowable() && !primaryKeyLivenessInfo.isEmpty())
|
||||
? Deletion.LIVE
|
||||
: new Deletion(new DeletionTime(newTimestamp - 1, deletion.time().localDeletionTime()), deletion.isShadowable());
|
||||
|
||||
return transformAndFilter(newInfo, newDeletion, (cd) -> cd.updateAllTimestamp(newTimestamp));
|
||||
}
|
||||
|
|
@ -318,12 +328,12 @@ public class BTreeRow extends AbstractRow
|
|||
return this;
|
||||
|
||||
LivenessInfo newInfo = purger.shouldPurge(primaryKeyLivenessInfo, nowInSec) ? LivenessInfo.EMPTY : primaryKeyLivenessInfo;
|
||||
DeletionTime newDeletion = purger.shouldPurge(deletion) ? DeletionTime.LIVE : deletion;
|
||||
Deletion newDeletion = purger.shouldPurge(deletion.time()) ? Deletion.LIVE : deletion;
|
||||
|
||||
return transformAndFilter(newInfo, newDeletion, (cd) -> cd.purge(purger, nowInSec));
|
||||
}
|
||||
|
||||
private Row transformAndFilter(LivenessInfo info, DeletionTime deletion, Function<ColumnData, ColumnData> function)
|
||||
private Row transformAndFilter(LivenessInfo info, Deletion deletion, Function<ColumnData, ColumnData> function)
|
||||
{
|
||||
Object[] transformed = BTree.transformAndFilter(btree, function);
|
||||
|
||||
|
|
@ -333,7 +343,7 @@ public class BTreeRow extends AbstractRow
|
|||
if (info.isEmpty() && deletion.isLive() && BTree.isEmpty(transformed))
|
||||
return null;
|
||||
|
||||
int minDeletionTime = minDeletionTime(transformed, info, deletion);
|
||||
int minDeletionTime = minDeletionTime(transformed, info, deletion.time());
|
||||
return new BTreeRow(clustering, info, deletion, transformed, minDeletionTime);
|
||||
}
|
||||
|
||||
|
|
@ -523,7 +533,7 @@ public class BTreeRow extends AbstractRow
|
|||
};
|
||||
protected Clustering clustering;
|
||||
protected LivenessInfo primaryKeyLivenessInfo = LivenessInfo.EMPTY;
|
||||
protected DeletionTime deletion = DeletionTime.LIVE;
|
||||
protected Deletion deletion = Deletion.LIVE;
|
||||
|
||||
private final boolean isSorted;
|
||||
private final BTree.Builder<Cell> cells;
|
||||
|
|
@ -565,7 +575,7 @@ public class BTreeRow extends AbstractRow
|
|||
{
|
||||
this.clustering = null;
|
||||
this.primaryKeyLivenessInfo = LivenessInfo.EMPTY;
|
||||
this.deletion = DeletionTime.LIVE;
|
||||
this.deletion = Deletion.LIVE;
|
||||
this.cells.reuse();
|
||||
}
|
||||
|
||||
|
|
@ -574,7 +584,7 @@ public class BTreeRow extends AbstractRow
|
|||
this.primaryKeyLivenessInfo = info;
|
||||
}
|
||||
|
||||
public void addRowDeletion(DeletionTime deletion)
|
||||
public void addRowDeletion(Deletion deletion)
|
||||
{
|
||||
this.deletion = deletion;
|
||||
}
|
||||
|
|
@ -601,7 +611,11 @@ public class BTreeRow extends AbstractRow
|
|||
if (!isSorted | hasComplex)
|
||||
cells.resolve(resolver);
|
||||
Object[] btree = cells.build();
|
||||
int minDeletionTime = minDeletionTime(btree, primaryKeyLivenessInfo, deletion);
|
||||
|
||||
if (deletion.isShadowedBy(primaryKeyLivenessInfo))
|
||||
deletion = Deletion.LIVE;
|
||||
|
||||
int minDeletionTime = minDeletionTime(btree, primaryKeyLivenessInfo, deletion.time());
|
||||
Row row = new BTreeRow(clustering, primaryKeyLivenessInfo, deletion, btree, minDeletionTime);
|
||||
reset();
|
||||
return row;
|
||||
|
|
|
|||
|
|
@ -18,12 +18,14 @@
|
|||
package org.apache.cassandra.db.rows;
|
||||
|
||||
import java.util.*;
|
||||
import java.security.MessageDigest;
|
||||
|
||||
import org.apache.cassandra.config.CFMetaData;
|
||||
import org.apache.cassandra.config.ColumnDefinition;
|
||||
import org.apache.cassandra.db.*;
|
||||
import org.apache.cassandra.db.filter.ColumnFilter;
|
||||
import org.apache.cassandra.service.paxos.Commit;
|
||||
import org.apache.cassandra.utils.FBUtilities;
|
||||
import org.apache.cassandra.utils.MergeIterator;
|
||||
import org.apache.cassandra.utils.SearchIterator;
|
||||
import org.apache.cassandra.utils.btree.BTree;
|
||||
|
|
@ -64,7 +66,7 @@ public interface Row extends Unfiltered, Collection<ColumnData>
|
|||
*
|
||||
* @return the row deletion.
|
||||
*/
|
||||
public DeletionTime deletion();
|
||||
public Deletion deletion();
|
||||
|
||||
/**
|
||||
* Liveness information for the primary key columns of this row.
|
||||
|
|
@ -218,6 +220,130 @@ public interface Row extends Unfiltered, Collection<ColumnData>
|
|||
|
||||
public String toString(CFMetaData metadata, boolean fullDetails);
|
||||
|
||||
/**
|
||||
* A row deletion/tombstone.
|
||||
* <p>
|
||||
* A row deletion mostly consists of the time of said deletion, but there is 2 variants: shadowable
|
||||
* and regular row deletion.
|
||||
* <p>
|
||||
* A shadowable row deletion only exists if the row timestamp ({@code primaryKeyLivenessInfo().timestamp()})
|
||||
* is lower than the deletion timestamp. That is, if a row has a shadowable deletion with timestamp A and an update is made
|
||||
* to that row with a timestamp B such that B > A, then the shadowable deletion is 'shadowed' by that update. A concrete
|
||||
* consequence is that if said update has cells with timestamp lower than A, then those cells are preserved
|
||||
* (since the deletion is removed), and this contrarily to a normal (regular) deletion where the deletion is preserved
|
||||
* and such cells are removed.
|
||||
* <p>
|
||||
* Currently, the only use of shadowable row deletions is Materialized Views, see CASSANDRA-10261.
|
||||
*/
|
||||
public static class Deletion
|
||||
{
|
||||
public static final Deletion LIVE = new Deletion(DeletionTime.LIVE, false);
|
||||
|
||||
private final DeletionTime time;
|
||||
private final boolean isShadowable;
|
||||
|
||||
public Deletion(DeletionTime time, boolean isShadowable)
|
||||
{
|
||||
assert !time.isLive() || !isShadowable;
|
||||
this.time = time;
|
||||
this.isShadowable = isShadowable;
|
||||
}
|
||||
|
||||
public static Deletion regular(DeletionTime time)
|
||||
{
|
||||
return time.isLive() ? LIVE : new Deletion(time, false);
|
||||
}
|
||||
|
||||
public static Deletion shadowable(DeletionTime time)
|
||||
{
|
||||
return new Deletion(time, true);
|
||||
}
|
||||
|
||||
/**
|
||||
* The time of the row deletion.
|
||||
*
|
||||
* @return the time of the row deletion.
|
||||
*/
|
||||
public DeletionTime time()
|
||||
{
|
||||
return time;
|
||||
}
|
||||
|
||||
/**
|
||||
* Whether the deletion is a shadowable one or not.
|
||||
*
|
||||
* @return whether the deletion is a shadowable one. Note that if {@code isLive()}, then this is
|
||||
* guarantee to return {@code false}.
|
||||
*/
|
||||
public boolean isShadowable()
|
||||
{
|
||||
return isShadowable;
|
||||
}
|
||||
|
||||
/**
|
||||
* Wether the deletion is live or not, that is if its an actual deletion or not.
|
||||
*
|
||||
* @return {@code true} if this represents no deletion of the row, {@code false} if that's an actual
|
||||
* deletion.
|
||||
*/
|
||||
public boolean isLive()
|
||||
{
|
||||
return time().isLive();
|
||||
}
|
||||
|
||||
public boolean supersedes(DeletionTime that)
|
||||
{
|
||||
return time.supersedes(that);
|
||||
}
|
||||
|
||||
public boolean supersedes(Deletion that)
|
||||
{
|
||||
return time.supersedes(that.time);
|
||||
}
|
||||
|
||||
public boolean isShadowedBy(LivenessInfo primaryKeyLivenessInfo)
|
||||
{
|
||||
return isShadowable && primaryKeyLivenessInfo.timestamp() > time.markedForDeleteAt();
|
||||
}
|
||||
|
||||
public boolean deletes(LivenessInfo info)
|
||||
{
|
||||
return time.deletes(info);
|
||||
}
|
||||
|
||||
public void digest(MessageDigest digest)
|
||||
{
|
||||
time.digest(digest);
|
||||
FBUtilities.updateWithBoolean(digest, isShadowable);
|
||||
}
|
||||
|
||||
public int dataSize()
|
||||
{
|
||||
return time.dataSize() + 1;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean equals(Object o)
|
||||
{
|
||||
if(!(o instanceof Deletion))
|
||||
return false;
|
||||
Deletion that = (Deletion)o;
|
||||
return this.time.equals(that.time) && this.isShadowable == that.isShadowable;
|
||||
}
|
||||
|
||||
@Override
|
||||
public final int hashCode()
|
||||
{
|
||||
return Objects.hash(time, isShadowable);
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString()
|
||||
{
|
||||
return String.format("%s%s", time, isShadowable ? "(shadowable)" : "");
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Interface for building rows.
|
||||
* <p>
|
||||
|
|
@ -280,9 +406,9 @@ public interface Row extends Unfiltered, Collection<ColumnData>
|
|||
*
|
||||
* This call is optional and can be skipped if the row is not deleted.
|
||||
*
|
||||
* @param deletion the row deletion time, or {@code DeletionTime.LIVE} if the row isn't deleted.
|
||||
* @param deletion the row deletion time, or {@code Deletion.LIVE} if the row isn't deleted.
|
||||
*/
|
||||
public void addRowDeletion(DeletionTime deletion);
|
||||
public void addRowDeletion(Deletion deletion);
|
||||
|
||||
/**
|
||||
* Adds a cell to this builder.
|
||||
|
|
@ -358,7 +484,7 @@ public interface Row extends Unfiltered, Collection<ColumnData>
|
|||
}
|
||||
|
||||
LivenessInfo rowInfo = LivenessInfo.EMPTY;
|
||||
DeletionTime rowDeletion = DeletionTime.LIVE;
|
||||
Deletion rowDeletion = Deletion.LIVE;
|
||||
for (Row row : rows)
|
||||
{
|
||||
if (row == null)
|
||||
|
|
@ -370,10 +496,13 @@ public interface Row extends Unfiltered, Collection<ColumnData>
|
|||
rowDeletion = row.deletion();
|
||||
}
|
||||
|
||||
if (activeDeletion.supersedes(rowDeletion))
|
||||
rowDeletion = DeletionTime.LIVE;
|
||||
if (rowDeletion.isShadowedBy(rowInfo))
|
||||
rowDeletion = Deletion.LIVE;
|
||||
|
||||
if (rowDeletion.supersedes(activeDeletion))
|
||||
activeDeletion = rowDeletion.time();
|
||||
else
|
||||
activeDeletion = rowDeletion;
|
||||
rowDeletion = Deletion.LIVE;
|
||||
|
||||
if (activeDeletion.deletes(rowInfo))
|
||||
rowInfo = LivenessInfo.EMPTY;
|
||||
|
|
|
|||
|
|
@ -50,7 +50,7 @@ public interface RowDiffListener
|
|||
* but the merged result doesn't (i.e. the deletion has been shadowed).
|
||||
* @param original the deletion of input {@code i}. May be {@code null} if input {@code i} had no deletion but the merged row has.
|
||||
*/
|
||||
public void onDeletion(int i, Clustering clustering, DeletionTime merged, DeletionTime original);
|
||||
public void onDeletion(int i, Clustering clustering, Row.Deletion merged, Row.Deletion original);
|
||||
|
||||
/**
|
||||
* Called for every (non-live) complex deletion of any complex column present in either the merged row of input {@code i}.
|
||||
|
|
|
|||
|
|
@ -70,7 +70,7 @@ public abstract class Rows
|
|||
assert !row.isEmpty();
|
||||
|
||||
collector.update(row.primaryKeyLivenessInfo());
|
||||
collector.update(row.deletion());
|
||||
collector.update(row.deletion().time());
|
||||
|
||||
int columnCount = 0;
|
||||
int cellCount = 0;
|
||||
|
|
@ -115,12 +115,12 @@ public abstract class Rows
|
|||
{
|
||||
Clustering clustering = merged.clustering();
|
||||
LivenessInfo mergedInfo = merged.primaryKeyLivenessInfo().isEmpty() ? null : merged.primaryKeyLivenessInfo();
|
||||
DeletionTime mergedDeletion = merged.deletion().isLive() ? null : merged.deletion();
|
||||
Row.Deletion mergedDeletion = merged.deletion().isLive() ? null : merged.deletion();
|
||||
for (int i = 0; i < inputs.length; i++)
|
||||
{
|
||||
Row input = inputs[i];
|
||||
LivenessInfo inputInfo = input == null || input.primaryKeyLivenessInfo().isEmpty() ? null : input.primaryKeyLivenessInfo();
|
||||
DeletionTime inputDeletion = input == null || input.deletion().isLive() ? null : input.deletion();
|
||||
Row.Deletion inputDeletion = input == null || input.deletion().isLive() ? null : input.deletion();
|
||||
|
||||
if (mergedInfo != null || inputInfo != null)
|
||||
diffListener.onPrimaryKeyLivenessInfo(i, clustering, mergedInfo, inputInfo);
|
||||
|
|
@ -237,13 +237,17 @@ public abstract class Rows
|
|||
|
||||
long timeDelta = Math.abs(existingInfo.timestamp() - mergedInfo.timestamp());
|
||||
|
||||
DeletionTime deletion = existing.deletion().supersedes(update.deletion()) ? existing.deletion() : update.deletion();
|
||||
Row.Deletion rowDeletion = existing.deletion().supersedes(update.deletion()) ? existing.deletion() : update.deletion();
|
||||
|
||||
if (deletion.deletes(mergedInfo))
|
||||
if (rowDeletion.deletes(mergedInfo))
|
||||
mergedInfo = LivenessInfo.EMPTY;
|
||||
else if (rowDeletion.isShadowedBy(mergedInfo))
|
||||
rowDeletion = Row.Deletion.LIVE;
|
||||
|
||||
builder.addPrimaryKeyLivenessInfo(mergedInfo);
|
||||
builder.addRowDeletion(deletion);
|
||||
builder.addRowDeletion(rowDeletion);
|
||||
|
||||
DeletionTime deletion = rowDeletion.time();
|
||||
|
||||
Iterator<ColumnData> a = existing.iterator();
|
||||
Iterator<ColumnData> b = update.iterator();
|
||||
|
|
|
|||
|
|
@ -31,8 +31,10 @@ import org.apache.cassandra.io.util.DataOutputPlus;
|
|||
*
|
||||
* The encoded format for an unfiltered is <flags>(<row>|<marker>) where:
|
||||
*
|
||||
* <flags> is a byte whose bits are flags used by the rest of the serialization. Each
|
||||
* flag is defined/explained below as the "Unfiltered flags" constants.
|
||||
* <flags> is a byte (or two) whose bits are flags used by the rest of the serialization. Each
|
||||
* flag is defined/explained below as the "Unfiltered flags" constants. One of those flags
|
||||
* is an extension flag, and if present, trigger the rid of another byte that contains more
|
||||
* flags. If the extension is not set, defaults are assumed for the flags of that 2nd byte.
|
||||
* <row> is <clustering>[<timestamp>][<ttl>][<deletion>]<sc1>...<sci><cc1>...<ccj> where
|
||||
* <clustering> is the row clustering as serialized by
|
||||
* {@code Clustering.serializer}. Note that static row are an exception and
|
||||
|
|
@ -72,12 +74,18 @@ public class UnfilteredSerializer
|
|||
*/
|
||||
private final static int END_OF_PARTITION = 0x01; // Signal the end of the partition. Nothing follows a <flags> field with that flag.
|
||||
private final static int IS_MARKER = 0x02; // Whether the encoded unfiltered is a marker or a row. All following markers applies only to rows.
|
||||
private final static int IS_STATIC = 0x04; // Whether the encoded row is a static.
|
||||
private final static int HAS_TIMESTAMP = 0x08; // Whether the encoded row has a timestamp (i.e. if row.partitionKeyLivenessInfo().hasTimestamp() == true).
|
||||
private final static int HAS_TTL = 0x10; // Whether the encoded row has some expiration info (i.e. if row.partitionKeyLivenessInfo().hasTTL() == true).
|
||||
private final static int HAS_DELETION = 0x20; // Whether the encoded row has some deletion info.
|
||||
private final static int HAS_TIMESTAMP = 0x04; // Whether the encoded row has a timestamp (i.e. if row.partitionKeyLivenessInfo().hasTimestamp() == true).
|
||||
private final static int HAS_TTL = 0x08; // Whether the encoded row has some expiration info (i.e. if row.partitionKeyLivenessInfo().hasTTL() == true).
|
||||
private final static int HAS_DELETION = 0x10; // Whether the encoded row has some deletion info.
|
||||
private final static int HAS_ALL_COLUMNS = 0x20; // Whether the encoded row has all of the columns from the header present.
|
||||
private final static int HAS_COMPLEX_DELETION = 0x40; // Whether the encoded row has some complex deletion for at least one of its columns.
|
||||
private final static int HAS_ALL_COLUMNS = 0x80; // Whether the encoded row has all of the columns from the header present
|
||||
private final static int EXTENSION_FLAG = 0x80; // If present, another byte is read containing the "extended flags" above.
|
||||
|
||||
/*
|
||||
* Extended flags
|
||||
*/
|
||||
private final static int IS_STATIC = 0x01; // Whether the encoded row is a static. If there is no extended flag, the row is assumed not static.
|
||||
private final static int HAS_SHADOWABLE_DELETION = 0x02; // Whether the row deletion is shadowable. If there is no extended flag (or no row deletion), the deletion is assumed not shadowable.
|
||||
|
||||
public void serialize(Unfiltered unfiltered, SerializationHeader header, DataOutputPlus out, int version)
|
||||
throws IOException
|
||||
|
|
@ -96,28 +104,47 @@ public class UnfilteredSerializer
|
|||
throws IOException
|
||||
{
|
||||
int flags = 0;
|
||||
boolean isStatic = row.isStatic();
|
||||
int extendedFlags = 0;
|
||||
boolean hasExtendedFlags = false;
|
||||
|
||||
boolean isStatic = row.isStatic();
|
||||
Columns headerColumns = header.columns(isStatic);
|
||||
LivenessInfo pkLiveness = row.primaryKeyLivenessInfo();
|
||||
DeletionTime deletion = row.deletion();
|
||||
Row.Deletion deletion = row.deletion();
|
||||
boolean hasComplexDeletion = row.hasComplexDeletion();
|
||||
boolean hasAllColumns = (row.size() == headerColumns.size());
|
||||
|
||||
if (isStatic)
|
||||
flags |= IS_STATIC;
|
||||
{
|
||||
hasExtendedFlags = true;
|
||||
extendedFlags |= IS_STATIC;
|
||||
}
|
||||
|
||||
if (!pkLiveness.isEmpty())
|
||||
flags |= HAS_TIMESTAMP;
|
||||
if (pkLiveness.isExpiring())
|
||||
flags |= HAS_TTL;
|
||||
if (!deletion.isLive())
|
||||
{
|
||||
flags |= HAS_DELETION;
|
||||
if (deletion.isShadowable())
|
||||
{
|
||||
hasExtendedFlags = true;
|
||||
extendedFlags |= HAS_SHADOWABLE_DELETION;
|
||||
}
|
||||
}
|
||||
if (hasComplexDeletion)
|
||||
flags |= HAS_COMPLEX_DELETION;
|
||||
if (hasAllColumns)
|
||||
flags |= HAS_ALL_COLUMNS;
|
||||
|
||||
if (hasExtendedFlags)
|
||||
flags |= EXTENSION_FLAG;
|
||||
|
||||
out.writeByte((byte)flags);
|
||||
if (hasExtendedFlags)
|
||||
out.writeByte((byte)extendedFlags);
|
||||
|
||||
if (!isStatic)
|
||||
Clustering.serializer.serialize(row.clustering(), out, version, header.clusteringTypes());
|
||||
|
||||
|
|
@ -129,7 +156,7 @@ public class UnfilteredSerializer
|
|||
header.writeLocalDeletionTime(pkLiveness.localExpirationTime(), out);
|
||||
}
|
||||
if ((flags & HAS_DELETION) != 0)
|
||||
header.writeDeletionTime(deletion, out);
|
||||
header.writeDeletionTime(deletion.time(), out);
|
||||
|
||||
if (!hasAllColumns)
|
||||
Columns.serializer.serializeSubset(Collections2.transform(row, ColumnData::column), headerColumns, out);
|
||||
|
|
@ -186,10 +213,13 @@ public class UnfilteredSerializer
|
|||
boolean isStatic = row.isStatic();
|
||||
Columns headerColumns = header.columns(isStatic);
|
||||
LivenessInfo pkLiveness = row.primaryKeyLivenessInfo();
|
||||
DeletionTime deletion = row.deletion();
|
||||
Row.Deletion deletion = row.deletion();
|
||||
boolean hasComplexDeletion = row.hasComplexDeletion();
|
||||
boolean hasAllColumns = (row.size() == headerColumns.size());
|
||||
|
||||
if (isStatic || deletion.isShadowable())
|
||||
size += 1; // extended flags
|
||||
|
||||
if (!isStatic)
|
||||
size += Clustering.serializer.serializedSize(row.clustering(), version, header.clusteringTypes());
|
||||
|
||||
|
|
@ -201,7 +231,7 @@ public class UnfilteredSerializer
|
|||
size += header.localDeletionTimeSerializedSize(pkLiveness.localExpirationTime());
|
||||
}
|
||||
if (!deletion.isLive())
|
||||
size += header.deletionTimeSerializedSize(deletion);
|
||||
size += header.deletionTimeSerializedSize(deletion.time());
|
||||
|
||||
if (!hasAllColumns)
|
||||
size += Columns.serializer.serializedSubsetSize(Collections2.transform(row, ColumnData::column), header.columns(isStatic));
|
||||
|
|
@ -269,6 +299,8 @@ public class UnfilteredSerializer
|
|||
if (isEndOfPartition(flags))
|
||||
return null;
|
||||
|
||||
int extendedFlags = isExtended(flags) ? in.readUnsignedByte() : 0;
|
||||
|
||||
if (kind(flags) == Unfiltered.Kind.RANGE_TOMBSTONE_MARKER)
|
||||
{
|
||||
RangeTombstone.Bound bound = RangeTombstone.Bound.serializer.deserialize(in, helper.version, header.clusteringTypes());
|
||||
|
|
@ -276,9 +308,9 @@ public class UnfilteredSerializer
|
|||
}
|
||||
else
|
||||
{
|
||||
assert !isStatic(flags); // deserializeStaticRow should be used for that.
|
||||
assert !isStatic(extendedFlags); // deserializeStaticRow should be used for that.
|
||||
builder.newRow(Clustering.serializer.deserialize(in, helper.version, header.clusteringTypes()));
|
||||
return deserializeRowBody(in, header, helper, flags, builder);
|
||||
return deserializeRowBody(in, header, helper, flags, extendedFlags, builder);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -286,10 +318,11 @@ public class UnfilteredSerializer
|
|||
throws IOException
|
||||
{
|
||||
int flags = in.readUnsignedByte();
|
||||
assert !isEndOfPartition(flags) && kind(flags) == Unfiltered.Kind.ROW && isStatic(flags) : flags;
|
||||
assert !isEndOfPartition(flags) && kind(flags) == Unfiltered.Kind.ROW && isExtended(flags) : flags;
|
||||
int extendedFlags = in.readUnsignedByte();
|
||||
Row.Builder builder = BTreeRow.sortedBuilder();
|
||||
builder.newRow(Clustering.STATIC_CLUSTERING);
|
||||
return deserializeRowBody(in, header, helper, flags, builder);
|
||||
return deserializeRowBody(in, header, helper, flags, extendedFlags, builder);
|
||||
}
|
||||
|
||||
public RangeTombstoneMarker deserializeMarkerBody(DataInputPlus in, SerializationHeader header, RangeTombstone.Bound bound)
|
||||
|
|
@ -305,15 +338,17 @@ public class UnfilteredSerializer
|
|||
SerializationHeader header,
|
||||
SerializationHelper helper,
|
||||
int flags,
|
||||
int extendedFlags,
|
||||
Row.Builder builder)
|
||||
throws IOException
|
||||
{
|
||||
try
|
||||
{
|
||||
boolean isStatic = isStatic(flags);
|
||||
boolean isStatic = isStatic(extendedFlags);
|
||||
boolean hasTimestamp = (flags & HAS_TIMESTAMP) != 0;
|
||||
boolean hasTTL = (flags & HAS_TTL) != 0;
|
||||
boolean hasDeletion = (flags & HAS_DELETION) != 0;
|
||||
boolean deletionIsShadowable = (extendedFlags & HAS_SHADOWABLE_DELETION) != 0;
|
||||
boolean hasComplexDeletion = (flags & HAS_COMPLEX_DELETION) != 0;
|
||||
boolean hasAllColumns = (flags & HAS_ALL_COLUMNS) != 0;
|
||||
Columns headerColumns = header.columns(isStatic);
|
||||
|
|
@ -328,7 +363,7 @@ public class UnfilteredSerializer
|
|||
}
|
||||
|
||||
builder.addPrimaryKeyLivenessInfo(rowLiveness);
|
||||
builder.addRowDeletion(hasDeletion ? header.readDeletionTime(in) : DeletionTime.LIVE);
|
||||
builder.addRowDeletion(hasDeletion ? new Row.Deletion(header.readDeletionTime(in), deletionIsShadowable) : Row.Deletion.LIVE);
|
||||
|
||||
Columns columns = hasAllColumns ? headerColumns : Columns.serializer.deserializeSubset(headerColumns, in);
|
||||
for (ColumnDefinition column : columns)
|
||||
|
|
@ -395,9 +430,9 @@ public class UnfilteredSerializer
|
|||
}
|
||||
}
|
||||
|
||||
public void skipRowBody(DataInputPlus in, SerializationHeader header, int flags) throws IOException
|
||||
public void skipRowBody(DataInputPlus in, SerializationHeader header, int flags, int extendedFlags) throws IOException
|
||||
{
|
||||
boolean isStatic = isStatic(flags);
|
||||
boolean isStatic = isStatic(extendedFlags);
|
||||
boolean hasTimestamp = (flags & HAS_TIMESTAMP) != 0;
|
||||
boolean hasTTL = (flags & HAS_TTL) != 0;
|
||||
boolean hasDeletion = (flags & HAS_DELETION) != 0;
|
||||
|
|
@ -430,8 +465,10 @@ public class UnfilteredSerializer
|
|||
public void skipStaticRow(DataInputPlus in, SerializationHeader header, SerializationHelper helper) throws IOException
|
||||
{
|
||||
int flags = in.readUnsignedByte();
|
||||
assert !isEndOfPartition(flags) && kind(flags) == Unfiltered.Kind.ROW && isStatic(flags) : "Flags is " + flags;
|
||||
skipRowBody(in, header, flags);
|
||||
assert !isEndOfPartition(flags) && kind(flags) == Unfiltered.Kind.ROW && isExtended(flags) : "Flags is " + flags;
|
||||
int extendedFlags = in.readUnsignedByte();
|
||||
assert isStatic(extendedFlags);
|
||||
skipRowBody(in, header, flags, extendedFlags);
|
||||
}
|
||||
|
||||
public void skipMarkerBody(DataInputPlus in, SerializationHeader header, boolean isBoundary) throws IOException
|
||||
|
|
@ -468,8 +505,13 @@ public class UnfilteredSerializer
|
|||
return (flags & IS_MARKER) != 0 ? Unfiltered.Kind.RANGE_TOMBSTONE_MARKER : Unfiltered.Kind.ROW;
|
||||
}
|
||||
|
||||
public static boolean isStatic(int flags)
|
||||
public static boolean isStatic(int extendedFlags)
|
||||
{
|
||||
return (flags & IS_MARKER) == 0 && (flags & IS_STATIC) != 0;
|
||||
return (extendedFlags & IS_STATIC) != 0;
|
||||
}
|
||||
|
||||
public static boolean isExtended(int flags)
|
||||
{
|
||||
return (flags & EXTENSION_FLAG) != 0;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -263,14 +263,14 @@ public class MaterializedView
|
|||
*/
|
||||
private PartitionUpdate createTombstone(TemporalRow temporalRow,
|
||||
DecoratedKey partitionKey,
|
||||
DeletionTime deletionTime,
|
||||
Row.Deletion deletion,
|
||||
TemporalRow.Resolver resolver,
|
||||
int nowInSec)
|
||||
{
|
||||
CFMetaData viewCfm = getViewCfs().metadata;
|
||||
Row.Builder builder = BTreeRow.unsortedBuilder(nowInSec);
|
||||
builder.newRow(viewClustering(temporalRow, resolver));
|
||||
builder.addRowDeletion(deletionTime);
|
||||
builder.addRowDeletion(deletion);
|
||||
return PartitionUpdate.singleRowUpdate(viewCfm, partitionKey, builder.build());
|
||||
}
|
||||
|
||||
|
|
@ -342,7 +342,7 @@ public class MaterializedView
|
|||
TemporalRow.Resolver resolver = TemporalRow.earliest;
|
||||
return createTombstone(temporalRow,
|
||||
viewPartitionKey(temporalRow, resolver),
|
||||
new DeletionTime(temporalRow.viewClusteringTimestamp(), temporalRow.nowInSec),
|
||||
Row.Deletion.shadowable(new DeletionTime(temporalRow.viewClusteringTimestamp(), temporalRow.nowInSec)),
|
||||
resolver,
|
||||
temporalRow.nowInSec);
|
||||
}
|
||||
|
|
@ -523,7 +523,7 @@ public class MaterializedView
|
|||
DecoratedKey value = viewPartitionKey(temporalRow, resolver);
|
||||
if (value != null)
|
||||
{
|
||||
PartitionUpdate update = createTombstone(temporalRow, value, deletionTime, resolver, temporalRow.nowInSec);
|
||||
PartitionUpdate update = createTombstone(temporalRow, value, Row.Deletion.regular(deletionTime), resolver, temporalRow.nowInSec);
|
||||
if (update != null)
|
||||
mutations.add(new Mutation(update));
|
||||
}
|
||||
|
|
|
|||
|
|
@ -256,7 +256,7 @@ public class TemporalRow
|
|||
this.nowInSec = nowInSec;
|
||||
|
||||
LivenessInfo liveness = row.primaryKeyLivenessInfo();
|
||||
this.viewClusteringLocalDeletionTime = minValueIfSet(viewClusteringLocalDeletionTime, row.deletion().localDeletionTime(), NO_DELETION_TIME);
|
||||
this.viewClusteringLocalDeletionTime = minValueIfSet(viewClusteringLocalDeletionTime, row.deletion().time().localDeletionTime(), NO_DELETION_TIME);
|
||||
this.viewClusteringTimestamp = minValueIfSet(viewClusteringTimestamp, liveness.timestamp(), NO_TIMESTAMP);
|
||||
this.viewClusteringTtl = minValueIfSet(viewClusteringTtl, liveness.ttl(), NO_TTL);
|
||||
|
||||
|
|
@ -402,7 +402,7 @@ public class TemporalRow
|
|||
return clusterTombstone.deletionTime();
|
||||
|
||||
Row row = partition.getRow(baseClustering);
|
||||
return row == null || row.deletion().isLive() ? DeletionTime.LIVE : row.deletion();
|
||||
return row == null || row.deletion().isLive() ? DeletionTime.LIVE : row.deletion().time();
|
||||
}
|
||||
|
||||
public Collection<org.apache.cassandra.db.rows.Cell> values(ColumnDefinition definition, Resolver resolver)
|
||||
|
|
|
|||
|
|
@ -742,7 +742,7 @@ public class SecondaryIndexManager implements IndexRegistry
|
|||
toInsert.addPrimaryKeyLivenessInfo(merged);
|
||||
}
|
||||
|
||||
public void onDeletion(int i, Clustering clustering, DeletionTime merged, DeletionTime original)
|
||||
public void onDeletion(int i, Clustering clustering, Row.Deletion merged, Row.Deletion original)
|
||||
{
|
||||
}
|
||||
|
||||
|
|
@ -834,7 +834,7 @@ public class SecondaryIndexManager implements IndexRegistry
|
|||
{
|
||||
}
|
||||
|
||||
public void onDeletion(int i, Clustering clustering, DeletionTime merged, DeletionTime original)
|
||||
public void onDeletion(int i, Clustering clustering, Row.Deletion merged, Row.Deletion original)
|
||||
{
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -411,13 +411,13 @@ public abstract class CassandraIndex implements Index
|
|||
|
||||
private void indexPrimaryKey(final Clustering clustering,
|
||||
final LivenessInfo liveness,
|
||||
final DeletionTime deletion)
|
||||
final Row.Deletion deletion)
|
||||
{
|
||||
if (liveness.timestamp() != LivenessInfo.NO_TIMESTAMP)
|
||||
insert(key.getKey(), clustering, null, liveness, opGroup);
|
||||
|
||||
if (!deletion.isLive())
|
||||
delete(key.getKey(), clustering, deletion, opGroup);
|
||||
delete(key.getKey(), clustering, deletion.time(), opGroup);
|
||||
}
|
||||
|
||||
private LivenessInfo getPrimaryKeyIndexLiveness(Row row)
|
||||
|
|
@ -516,7 +516,7 @@ public abstract class CassandraIndex implements Index
|
|||
DeletionTime deletion,
|
||||
OpOrder.Group opGroup)
|
||||
{
|
||||
Row row = BTreeRow.emptyDeletedRow(indexClustering, deletion);
|
||||
Row row = BTreeRow.emptyDeletedRow(indexClustering, Row.Deletion.regular(deletion));
|
||||
PartitionUpdate upd = partitionUpdate(indexKey, row);
|
||||
indexCfs.apply(upd, UpdateTransaction.NO_OP, opGroup, null);
|
||||
logger.debug("Removed index entry for value {}", indexKey);
|
||||
|
|
|
|||
|
|
@ -180,7 +180,7 @@ public class DataResolver extends ResponseResolver
|
|||
currentRow(i, clustering).addPrimaryKeyLivenessInfo(merged);
|
||||
}
|
||||
|
||||
public void onDeletion(int i, Clustering clustering, DeletionTime merged, DeletionTime original)
|
||||
public void onDeletion(int i, Clustering clustering, Row.Deletion merged, Row.Deletion original)
|
||||
{
|
||||
if (merged != null && !merged.equals(original))
|
||||
currentRow(i, clustering).addRowDeletion(merged);
|
||||
|
|
|
|||
|
|
@ -1343,7 +1343,7 @@ public class CassandraServer implements Cassandra.Iface
|
|||
}
|
||||
else if (column_path.super_column != null && column_path.column == null)
|
||||
{
|
||||
Row row = BTreeRow.emptyDeletedRow(new Clustering(column_path.super_column), new DeletionTime(timestamp, nowInSec));
|
||||
Row row = BTreeRow.emptyDeletedRow(new Clustering(column_path.super_column), Row.Deletion.regular(new DeletionTime(timestamp, nowInSec)));
|
||||
update = PartitionUpdate.singleRowUpdate(metadata, dk, row);
|
||||
}
|
||||
else
|
||||
|
|
|
|||
|
|
@ -753,12 +753,12 @@ public abstract class CQLTester
|
|||
Object[] expected = rows[i];
|
||||
UntypedResultSet.Row actual = iter.next();
|
||||
|
||||
Assert.assertEquals(String.format("Invalid number of (expected) values provided for row %d", i), expected.length, meta.size());
|
||||
Assert.assertEquals(String.format("Invalid number of (expected) values provided for row %d", i), expected == null ? 1 : expected.length, meta.size());
|
||||
|
||||
for (int j = 0; j < meta.size(); j++)
|
||||
{
|
||||
ColumnSpecification column = meta.get(j);
|
||||
ByteBuffer expectedByteValue = makeByteBuffer(expected[j], column.type);
|
||||
ByteBuffer expectedByteValue = makeByteBuffer(expected == null ? null : expected[j], column.type);
|
||||
ByteBuffer actualValue = actual.getBytes(column.name.toString());
|
||||
|
||||
if (!Objects.equal(expectedByteValue, actualValue))
|
||||
|
|
|
|||
|
|
@ -49,6 +49,7 @@ import org.apache.cassandra.db.SystemKeyspace;
|
|||
import org.apache.cassandra.serializers.SimpleDateSerializer;
|
||||
import org.apache.cassandra.serializers.TimeSerializer;
|
||||
import org.apache.cassandra.utils.ByteBufferUtil;
|
||||
import org.apache.cassandra.utils.FBUtilities;
|
||||
|
||||
public class MaterializedViewTest extends CQLTester
|
||||
{
|
||||
|
|
@ -340,6 +341,124 @@ public class MaterializedViewTest extends CQLTester
|
|||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void complexTimestampUpdateTestWithFlush() throws Throwable
|
||||
{
|
||||
complexTimestampUpdateTest(true);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void complexTimestampUpdateTestWithoutFlush() throws Throwable
|
||||
{
|
||||
complexTimestampUpdateTest(false);
|
||||
}
|
||||
|
||||
public void complexTimestampUpdateTest(boolean flush) throws Throwable
|
||||
{
|
||||
createTable("CREATE TABLE %s (a int, b int, c int, d int, e int, PRIMARY KEY (a, b))");
|
||||
|
||||
execute("USE " + keyspace());
|
||||
executeNet(protocolVersion, "USE " + keyspace());
|
||||
Keyspace ks = Keyspace.open(keyspace());
|
||||
|
||||
createView("mv", "CREATE MATERIALIZED VIEW %s AS SELECT * FROM %%s WHERE a IS NOT NULL AND b IS NOT NULL AND c IS NOT NULL PRIMARY KEY (c, a, b)");
|
||||
ks.getColumnFamilyStore("mv").disableAutoCompaction();
|
||||
|
||||
//Set initial values TS=0, leaving e null and verify view
|
||||
executeNet(protocolVersion, "INSERT INTO %s (a, b, c, d) VALUES (0, 0, 1, 0) USING TIMESTAMP 0");
|
||||
assertRows(execute("SELECT d from mv WHERE c = ? and a = ? and b = ?", 1, 0, 0), row(0));
|
||||
|
||||
if (flush)
|
||||
FBUtilities.waitOnFutures(ks.flush());
|
||||
|
||||
//update c's timestamp TS=2
|
||||
executeNet(protocolVersion, "UPDATE %s USING TIMESTAMP 2 SET c = ? WHERE a = ? and b = ? ", 1, 0, 0);
|
||||
assertRows(execute("SELECT d from mv WHERE c = ? and a = ? and b = ?", 1, 0, 0), row(0));
|
||||
|
||||
if (flush)
|
||||
FBUtilities.waitOnFutures(ks.flush());
|
||||
|
||||
//change c's value and TS=3, tombstones c=1 and adds c=0 record
|
||||
executeNet(protocolVersion, "UPDATE %s USING TIMESTAMP 3 SET c = ? WHERE a = ? and b = ? ", 0, 0, 0);
|
||||
assertRows(execute("SELECT d from mv WHERE c = ? and a = ? and b = ?", 1, 0, 0));
|
||||
|
||||
if(flush)
|
||||
{
|
||||
ks.getColumnFamilyStore("mv").forceMajorCompaction();
|
||||
FBUtilities.waitOnFutures(ks.flush());
|
||||
}
|
||||
|
||||
|
||||
//change c's value back to 1 with TS=4, check we can see d
|
||||
executeNet(protocolVersion, "UPDATE %s USING TIMESTAMP 4 SET c = ? WHERE a = ? and b = ? ", 1, 0, 0);
|
||||
if (flush)
|
||||
{
|
||||
ks.getColumnFamilyStore("mv").forceMajorCompaction();
|
||||
FBUtilities.waitOnFutures(ks.flush());
|
||||
}
|
||||
|
||||
assertRows(execute("SELECT d,e from mv WHERE c = ? and a = ? and b = ?", 1, 0, 0), row(0, null));
|
||||
|
||||
|
||||
//Add e value @ TS=1
|
||||
executeNet(protocolVersion, "UPDATE %s USING TIMESTAMP 1 SET e = ? WHERE a = ? and b = ? ", 1, 0, 0);
|
||||
assertRows(execute("SELECT d,e from mv WHERE c = ? and a = ? and b = ?", 1, 0, 0), row(0, 1));
|
||||
|
||||
if (flush)
|
||||
FBUtilities.waitOnFutures(ks.flush());
|
||||
|
||||
|
||||
//Change d value @ TS=2
|
||||
executeNet(protocolVersion, "UPDATE %s USING TIMESTAMP 2 SET d = ? WHERE a = ? and b = ? ", 2, 0, 0);
|
||||
assertRows(execute("SELECT d from mv WHERE c = ? and a = ? and b = ?", 1, 0, 0), row(2));
|
||||
|
||||
if (flush)
|
||||
FBUtilities.waitOnFutures(ks.flush());
|
||||
|
||||
|
||||
//Change d value @ TS=3
|
||||
executeNet(protocolVersion, "UPDATE %s USING TIMESTAMP 3 SET d = ? WHERE a = ? and b = ? ", 1, 0, 0);
|
||||
assertRows(execute("SELECT d from mv WHERE c = ? and a = ? and b = ?", 1, 0, 0), row(1));
|
||||
|
||||
|
||||
//Tombstone c
|
||||
executeNet(protocolVersion, "DELETE FROM %s WHERE a = ? and b = ?", 0, 0);
|
||||
assertRows(execute("SELECT d from mv"));
|
||||
|
||||
//Add back without D
|
||||
executeNet(protocolVersion, "INSERT INTO %s (a, b, c) VALUES (0, 0, 1)");
|
||||
|
||||
//Make sure D doesn't pop back in.
|
||||
assertRows(execute("SELECT d from mv WHERE c = ? and a = ? and b = ?", 1, 0, 0), row((Object) null));
|
||||
|
||||
|
||||
//New partition
|
||||
// insert a row with timestamp 0
|
||||
executeNet(protocolVersion, "INSERT INTO %s (a, b, c, d, e) VALUES (?, ?, ?, ?, ?) USING TIMESTAMP 0", 1, 0, 0, 0, 0);
|
||||
|
||||
// overwrite pk and e with timestamp 1, but don't overwrite d
|
||||
executeNet(protocolVersion, "INSERT INTO %s (a, b, c, e) VALUES (?, ?, ?, ?) USING TIMESTAMP 1", 1, 0, 0, 0);
|
||||
|
||||
// delete with timestamp 0 (which should only delete d)
|
||||
executeNet(protocolVersion, "DELETE FROM %s USING TIMESTAMP 0 WHERE a = ? AND b = ?", 1, 0);
|
||||
assertRows(execute("SELECT a, b, c, d, e from mv WHERE c = ? and a = ? and b = ?", 0, 1, 0),
|
||||
row(1, 0, 0, null, 0)
|
||||
);
|
||||
|
||||
executeNet(protocolVersion, "UPDATE %s USING TIMESTAMP 2 SET c = ? WHERE a = ? AND b = ?", 1, 1, 0);
|
||||
executeNet(protocolVersion, "UPDATE %s USING TIMESTAMP 3 SET c = ? WHERE a = ? AND b = ?", 0, 1, 0);
|
||||
assertRows(execute("SELECT a, b, c, d, e from mv WHERE c = ? and a = ? and b = ?", 0, 1, 0),
|
||||
row(1, 0, 0, null, 0)
|
||||
);
|
||||
|
||||
executeNet(protocolVersion, "UPDATE %s USING TIMESTAMP 3 SET d = ? WHERE a = ? AND b = ?", 0, 1, 0);
|
||||
assertRows(execute("SELECT a, b, c, d, e from mv WHERE c = ? and a = ? and b = ?", 0, 1, 0),
|
||||
row(1, 0, 0, 0, 0)
|
||||
);
|
||||
|
||||
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testBuilderWidePartition() throws Throwable
|
||||
{
|
||||
|
|
|
|||
|
|
@ -337,8 +337,8 @@ public class UnfilteredRowIteratorsMergeTest
|
|||
else
|
||||
{
|
||||
Row row = (Row) list.get(index);
|
||||
if (row.deletion() != null && row.deletion().supersedes(def))
|
||||
def = row.deletion();
|
||||
if (row.deletion().supersedes(def))
|
||||
def = row.deletion().time();
|
||||
}
|
||||
|
||||
if (index >= list.size())
|
||||
|
|
|
|||
|
|
@ -395,13 +395,13 @@ public class CustomCassandraIndex implements Index
|
|||
|
||||
private void indexPrimaryKey(final Clustering clustering,
|
||||
final LivenessInfo liveness,
|
||||
final DeletionTime deletion)
|
||||
final Row.Deletion deletion)
|
||||
{
|
||||
if (liveness.timestamp() != LivenessInfo.NO_TIMESTAMP)
|
||||
insert(key.getKey(), clustering, null, liveness, opGroup);
|
||||
|
||||
if (!deletion.isLive())
|
||||
delete(key.getKey(), clustering, deletion, opGroup);
|
||||
delete(key.getKey(), clustering, deletion.time(), opGroup);
|
||||
}
|
||||
|
||||
private LivenessInfo getPrimaryKeyIndexLiveness(Row row)
|
||||
|
|
@ -500,7 +500,7 @@ public class CustomCassandraIndex implements Index
|
|||
DeletionTime deletion,
|
||||
OpOrder.Group opGroup)
|
||||
{
|
||||
Row row = BTreeRow.emptyDeletedRow(indexClustering, deletion);
|
||||
Row row = BTreeRow.emptyDeletedRow(indexClustering, Row.Deletion.regular(deletion));
|
||||
PartitionUpdate upd = partitionUpdate(indexKey, row);
|
||||
indexCfs.apply(upd, UpdateTransaction.NO_OP, opGroup, null);
|
||||
logger.debug("Removed index entry for value {}", indexKey);
|
||||
|
|
|
|||
Loading…
Reference in New Issue