* There is 2 types of markers: bounds (see {@link RangeTombstoneBoundMarker}) and boundaries (see {@link RangeTombstoneBoundaryMarker}). - *
*/ public interface RangeTombstoneMarker extends Unfiltered { diff --git a/src/java/org/apache/cassandra/service/DataResolver.java b/src/java/org/apache/cassandra/service/DataResolver.java index cc7ab2dc83..f8d990906f 100644 --- a/src/java/org/apache/cassandra/service/DataResolver.java +++ b/src/java/org/apache/cassandra/service/DataResolver.java @@ -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))) diff --git a/test/unit/org/apache/cassandra/service/DataResolverTest.java b/test/unit/org/apache/cassandra/service/DataResolverTest.java index 413f032f06..3916a30bae 100644 --- a/test/unit/org/apache/cassandra/service/DataResolverTest.java +++ b/test/unit/org/apache/cassandra/service/DataResolverTest.java @@ -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