natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'cassandra-3.0' into trunk 

* cassandra-3.0:
  Legacy deserializer can create unexpected boundary range tombstones
This commit is contained in:
Sylvain Lebresne 2017-02-23 15:21:47 +01:00
parent cd29d44dfe ab71748495
commit 831c05b1c0
4 changed files with 142 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -44,6 +44,7 @@
* More fixes to the TokenAllocator (CASSANDRA-12990)
* NoReplicationTokenAllocator should work with zero replication factor (CASSANDRA-12983)
Merged from 3.0:
* Legacy deserializer can create unexpected boundary range tombstones (CASSANDRA-13237)
* Remove unnecessary assertion from AntiCompactionTest (CASSANDRA-13070)
* Fix cqlsh COPY for dates before 1900 (CASSANDRA-13185)
* Use keyspace replication settings on system.size_estimates table (CASSANDRA-9639)

1
src/java/org/apache/cassandra/db/rows/RangeTombstoneMarker.java
View File

@ -27,7 +27,6 @@ import org.apache.cassandra.utils.memory.AbstractAllocator;
* A marker for a range tombstone bound.
* <p>
* There is 2 types of markers: bounds (see {@link RangeTombstoneBoundMarker}) and boundaries (see {@link RangeTombstoneBoundaryMarker}).
* </p>
*/
public interface RangeTombstoneMarker extends Unfiltered
{

31
src/java/org/apache/cassandra/service/DataResolver.java
View File

@ -313,26 +313,33 @@ public class DataResolver extends ResponseResolver
// active after that point. Further whatever deletion was open or is open by this marker on the
// source, that deletion cannot supersedes the current one.
//
// What we want to know here is if the source deletion and merged deletion was or will be equal,
// because in that case we don't want to include any repair for the source, and otherwise we do.
// But while the marker deletion (before and/or after this point) cannot supersed the current
// deletion, we want to know if it's equal to it (both before and after), because in that case
// the source is up to date and we don't want to include repair.
//
// Note further that if the marker is a boundary, as both side of that boundary will have a
// different deletion time, only one side might be equal to the merged deletion. This means we
// can only be in one of 2 cases:
// 1) the source was up-to-date on deletion up to that point (markerToRepair[i] == null), and then
// it won't be from that point on.
// So in practice we have 2 possible case:
// 1) the source was up-to-date on deletion up to that point (markerToRepair[i] == null). Then
// it won't be from that point on unless it's a boundary and the new opened deletion time
// is also equal to the current deletion (note that this implies the boundary has the same
// closing and opening deletion time, which should generally not happen, but can due to legacy
// reading code not avoiding this for a while, see CASSANDRA-13237).
// 2) the source wasn't up-to-date on deletion up to that point (markerToRepair[i] != null), and
// it may now be (if it isn't we just have nothing to do for that marker).
assert !currentDeletion.isLive();
assert !currentDeletion.isLive() : currentDeletion.toString();
if (markerToRepair[i] == null)
{
// Since there is an ongoing merged deletion, the only way we don't have an open repair for
// this source is that it had a range open with the same deletion as current and it's
// closing it. This imply we need to open a deletion for the source from that point.
assert marker.isClose(isReversed) && currentDeletion.equals(marker.closeDeletionTime(isReversed));
assert !marker.isOpen(isReversed) || currentDeletion.supersedes(marker.openDeletionTime(isReversed));
markerToRepair[i] = marker.closeBound(isReversed).invert();
// closing it.
assert marker.isClose(isReversed) && currentDeletion.equals(marker.closeDeletionTime(isReversed))
: String.format("currentDeletion=%s, marker=%s", currentDeletion, marker.toString(command.metadata()));
// and so unless it's a boundary whose opening deletion time is still equal to the current
// deletion (see comment above for why this can actually happen), we have to repair the source
// from that point on.
if (!(marker.isOpen(isReversed) && currentDeletion.equals(marker.openDeletionTime(isReversed))))
markerToRepair[i] = marker.closeBound(isReversed).invert();
}
// In case 2) above, we only have something to do if the source is up-to-date after that point
else if (marker.isOpen(isReversed) && currentDeletion.equals(marker.openDeletionTime(isReversed)))

129
test/unit/org/apache/cassandra/service/DataResolverTest.java
View File

@ -556,6 +556,73 @@ public class DataResolverTest
assertRepairContainsDeletions(msg2, null, one_two, withExclusiveEndIf(three_four, timestamp2 >= timestamp1), five_six);
}
/**
* Test cases where a boundary of a source is covered by another source deletion and timestamp on one or both side
* of the boundary are equal to the "merged" deletion.
* This is a test for CASSANDRA-13237 to make sure we handle this case properly.
*/
@Test
public void testRepairRangeTombstoneBoundary() throws UnknownHostException
{
testRepairRangeTombstoneBoundary(1, 0, 1);
messageRecorder.sent.clear();
testRepairRangeTombstoneBoundary(1, 1, 0);
messageRecorder.sent.clear();
testRepairRangeTombstoneBoundary(1, 1, 1);
}
/**
* Test for CASSANDRA-13237, checking we don't fail (and handle correctly) the case where a RT boundary has the
* same deletion on both side (while is useless but could be created by legacy code pre-CASSANDRA-13237 and could
* thus still be sent).
*/
public void testRepairRangeTombstoneBoundary(int timestamp1, int timestamp2, int timestamp3) throws UnknownHostException
{
DataResolver resolver = new DataResolver(ks, command, ConsistencyLevel.ALL, 2, System.nanoTime());
InetAddress peer1 = peer();
InetAddress peer2 = peer();
// 1st "stream"
RangeTombstone one_nine = tombstone("0", true , "9", true, timestamp1, nowInSec);
UnfilteredPartitionIterator iter1 = iter(new RowUpdateBuilder(cfm, nowInSec, 1L, dk)
.addRangeTombstone(one_nine)
.buildUpdate());
// 2nd "stream" (build more manually to ensure we have the boundary we want)
RangeTombstoneBoundMarker open_one = marker("0", true, true, timestamp2, nowInSec);
RangeTombstoneBoundaryMarker boundary_five = boundary("5", false, timestamp2, nowInSec, timestamp3, nowInSec);
RangeTombstoneBoundMarker close_nine = marker("9", false, true, timestamp3, nowInSec);
UnfilteredPartitionIterator iter2 = iter(dk, open_one, boundary_five, close_nine);
resolver.preprocess(readResponseMessage(peer1, iter1));
resolver.preprocess(readResponseMessage(peer2, iter2));
boolean shouldHaveRepair = timestamp1 != timestamp2 || timestamp1 != timestamp3;
// No results, we've only reconciled tombstones.
try (PartitionIterator data = resolver.resolve())
{
assertFalse(data.hasNext());
assertRepairFuture(resolver, shouldHaveRepair ? 1 : 0);
}
assertEquals(shouldHaveRepair? 1 : 0, messageRecorder.sent.size());
if (!shouldHaveRepair)
return;
MessageOut msg = getSentMessage(peer2);
assertRepairMetadata(msg);
assertRepairContainsNoColumns(msg);
RangeTombstone expected = timestamp1 != timestamp2
// We've repaired the 1st part
? tombstone("0", true, "5", false, timestamp1, nowInSec)
// We've repaired the 2nd part
: tombstone("5", true, "9", true, timestamp1, nowInSec);
assertRepairContainsDeletions(msg, null, expected);
}
// Forces the start to be exclusive if the condition holds
private static RangeTombstone withExclusiveStartIf(RangeTombstone rt, boolean condition)
{
@ -883,17 +950,43 @@ public class DataResolverTest
private RangeTombstone tombstone(Object start, boolean inclusiveStart, Object end, boolean inclusiveEnd, long markedForDeleteAt, int localDeletionTime)
{
Kind startKind = inclusiveStart ? Kind.INCL_START_BOUND : Kind.EXCL_START_BOUND;
Kind endKind = inclusiveEnd ? Kind.INCL_END_BOUND : Kind.EXCL_END_BOUND;
ClusteringBound startBound = ClusteringBound.create(startKind, cfm.comparator.make(start).getRawValues());
ClusteringBound endBound = ClusteringBound.create(endKind, cfm.comparator.make(end).getRawValues());
ClusteringBound startBound = rtBound(start, true, inclusiveStart);
ClusteringBound endBound = rtBound(end, false, inclusiveEnd);
return new RangeTombstone(Slice.make(startBound, endBound), new DeletionTime(markedForDeleteAt, localDeletionTime));
}
private UnfilteredPartitionIterator fullPartitionDelete(TableMetadata cfm, DecoratedKey dk, long timestamp, int nowInSec)
private ClusteringBound rtBound(Object value, boolean isStart, boolean inclusive)
{
return new SingletonUnfilteredPartitionIterator(PartitionUpdate.fullPartitionDelete(cfm, dk, timestamp, nowInSec).unfilteredIterator());
ClusteringBound.Kind kind = isStart
? (inclusive ? Kind.INCL_START_BOUND : Kind.EXCL_START_BOUND)
: (inclusive ? Kind.INCL_END_BOUND : Kind.EXCL_END_BOUND);
return ClusteringBound.create(kind, cfm.comparator.make(value).getRawValues());
}
private ClusteringBoundary rtBoundary(Object value, boolean inclusiveOnEnd)
{
ClusteringBound.Kind kind = inclusiveOnEnd
? Kind.INCL_END_EXCL_START_BOUNDARY
: Kind.EXCL_END_INCL_START_BOUNDARY;
return ClusteringBoundary.create(kind, cfm.comparator.make(value).getRawValues());
}
private RangeTombstoneBoundMarker marker(Object value, boolean isStart, boolean inclusive, long markedForDeleteAt, int localDeletionTime)
{
return new RangeTombstoneBoundMarker(rtBound(value, isStart, inclusive), new DeletionTime(markedForDeleteAt, localDeletionTime));
}
private RangeTombstoneBoundaryMarker boundary(Object value, boolean inclusiveOnEnd, long markedForDeleteAt1, int localDeletionTime1, long markedForDeleteAt2, int localDeletionTime2)
{
return new RangeTombstoneBoundaryMarker(rtBoundary(value, inclusiveOnEnd),
new DeletionTime(markedForDeleteAt1, localDeletionTime1),
new DeletionTime(markedForDeleteAt2, localDeletionTime2));
}
private UnfilteredPartitionIterator fullPartitionDelete(TableMetadata table, DecoratedKey dk, long timestamp, int nowInSec)
{
return new SingletonUnfilteredPartitionIterator(PartitionUpdate.fullPartitionDelete(table, dk, timestamp, nowInSec).unfilteredIterator());
}
private static class MessageRecorder implements IMessageSink
@ -915,4 +1008,26 @@ public class DataResolverTest
{
return new SingletonUnfilteredPartitionIterator(update.unfilteredIterator());
}
private UnfilteredPartitionIterator iter(DecoratedKey key, Unfiltered... unfiltereds)
{
SortedSet<Unfiltered> s = new TreeSet<>(cfm.comparator);
Collections.addAll(s, unfiltereds);
final Iterator<Unfiltered> iterator = s.iterator();
UnfilteredRowIterator rowIter = new AbstractUnfilteredRowIterator(cfm,
key,
DeletionTime.LIVE,
cfm.regularAndStaticColumns(),
Rows.EMPTY_STATIC_ROW,
false,
EncodingStats.NO_STATS)
{
protected Unfiltered computeNext()
{
return iterator.hasNext() ? iterator.next() : endOfData();
}
};
return new SingletonUnfilteredPartitionIterator(rowIter);
}
}