This is an automated email from the ASF dual-hosted git repository. bdeggleston pushed a commit to branch cep-45-mutation-tracking in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 974c7e911fbe40cbd1a9674550d98eadac1254cf Author: Blake Eggleston <[email protected]> AuthorDate: Wed Oct 7 15:02:08 2026 -0700 CEP-45: Track journal segment references via coordinator log offset overlap Patch by Blake Eggleston; Reviewed by Francisco Guerrero for CASSANDRA-21563 --- .../db/virtual/MutationTrackingTables.java | 2 +- src/java/org/apache/cassandra/journal/Journal.java | 2 +- .../replication/CoordinatorLogOffsets.java | 4 + .../ImmutableCoordinatorLogOffsets.java | 6 + .../cassandra/replication/MutationJournal.java | 59 ++- .../MutationJournalSegmentCompactor.java | 2 +- .../org/apache/cassandra/replication/Offsets.java | 22 + .../replication/SegmentReferenceTracker.java | 212 +++++--- .../MutationJournalSegmentRefcountTest.java | 15 + .../cassandra/replication/MutationJournalTest.java | 33 ++ .../apache/cassandra/replication/OffsetsTest.java | 43 ++ .../replication/SegmentReferenceTrackerTest.java | 580 ++++++++++++++------- 12 files changed, 701 insertions(+), 279 deletions(-) diff --git a/src/java/org/apache/cassandra/db/virtual/MutationTrackingTables.java b/src/java/org/apache/cassandra/db/virtual/MutationTrackingTables.java index 4c3b801422..9af071fe65 100644 --- a/src/java/org/apache/cassandra/db/virtual/MutationTrackingTables.java +++ b/src/java/org/apache/cassandra/db/virtual/MutationTrackingTables.java @@ -107,7 +107,7 @@ public class MutationTrackingTables for (Segment<ShortMutationId, Mutation> segment : MutationJournal.instance().getAllSegments()) { - List<String> referringSstables = referenceTracker.referrerDescriptors(segment.id()); + List<String> referringSstables = referenceTracker.referrerDescriptors(segment); result.row(segment.id()) .column(IS_ACTIVE, segment instanceof ActiveSegment) .column(BYTES_ON_DISK, segment.segmentSizeOnDisk()) diff --git a/src/java/org/apache/cassandra/journal/Journal.java b/src/java/org/apache/cassandra/journal/Journal.java index f9569747f2..9f3f66659b 100644 --- a/src/java/org/apache/cassandra/journal/Journal.java +++ b/src/java/org/apache/cassandra/journal/Journal.java @@ -1019,7 +1019,7 @@ public class Journal<K, V> implements Shutdownable return currentSegment; } - @Nullable protected Segment<K, V> getSegment(long timestamp) + @Nullable public Segment<K, V> getSegment(long timestamp) { return segments().get(timestamp); } diff --git a/src/java/org/apache/cassandra/replication/CoordinatorLogOffsets.java b/src/java/org/apache/cassandra/replication/CoordinatorLogOffsets.java index 213413456d..cb78181abb 100644 --- a/src/java/org/apache/cassandra/replication/CoordinatorLogOffsets.java +++ b/src/java/org/apache/cassandra/replication/CoordinatorLogOffsets.java @@ -38,6 +38,10 @@ public interface CoordinatorLogOffsets<O extends Offsets> interface Mutations<O> extends Iterable<Long> { O offsets(long logId); + default O getOffsets(long logId) + { + return offsets(logId); + } int size(); default boolean isEmpty() { diff --git a/src/java/org/apache/cassandra/replication/ImmutableCoordinatorLogOffsets.java b/src/java/org/apache/cassandra/replication/ImmutableCoordinatorLogOffsets.java index 5ebf038923..df851339b0 100644 --- a/src/java/org/apache/cassandra/replication/ImmutableCoordinatorLogOffsets.java +++ b/src/java/org/apache/cassandra/replication/ImmutableCoordinatorLogOffsets.java @@ -301,6 +301,12 @@ public class ImmutableCoordinatorLogOffsets implements CoordinatorLogOffsets<Off return offsets; } + @Override + public Offsets.Immutable getOffsets(long logId) + { + return ids.get(logId); + } + @Override public int size() { diff --git a/src/java/org/apache/cassandra/replication/MutationJournal.java b/src/java/org/apache/cassandra/replication/MutationJournal.java index 8758588bd4..ca64cdcfc1 100644 --- a/src/java/org/apache/cassandra/replication/MutationJournal.java +++ b/src/java/org/apache/cassandra/replication/MutationJournal.java @@ -77,6 +77,7 @@ import org.apache.cassandra.journal.ValueSerializer; import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.schema.Schema; import org.apache.cassandra.schema.TableId; +import org.apache.cassandra.service.StorageService; import org.apache.cassandra.tcm.ClusterMetadata; import org.apache.cassandra.utils.Crc; import org.apache.cassandra.utils.concurrent.OpOrder; @@ -204,7 +205,9 @@ public class MutationJournal } }; this.journal = journal; - segmentReferenceTracker = new SegmentReferenceTracker(() -> journal.compactor().triggerNow()); + segmentReferenceTracker = new SegmentReferenceTracker(() -> journal.compactor().triggerNow(), + StorageService.instance::getLocalHostUUID, + journal::getAllSegments); } /** @@ -744,6 +747,23 @@ public class MutationJournal abstract Map<Long, Long> asMap(); + abstract boolean overlapsLog(long logId, Offsets offsets); + + public boolean overlaps(CoordinatorLogOffsets<?> offsets) + { + if (offsets == null || offsets.isEmpty()) + return false; + + CoordinatorLogOffsets.Mutations<? extends Offsets> mutations = offsets.mutations(); + for (long logId : mutations) + { + Offsets sstOffsets = mutations.getOffsets(logId); + if (sstOffsets != null && !sstOffsets.isEmpty() && overlapsLog(logId, sstOffsets)) + return true; + } + return false; + } + @Override public String toString() { @@ -784,6 +804,17 @@ public class MutationJournal return ranges; } + @Override + boolean overlapsLog(long logId, Offsets offsets) + { + Long val = ranges.get(logId); + if (val == null) + return false; + + long range = val; + return offsets.overlaps(minOffset(range), maxOffset(range)); + } + @Override public boolean mayContain(ShortMutationId id) { @@ -876,6 +907,32 @@ public class MutationJournal return range != NO_VALUE && mayContain(range, id); } + @Override + boolean overlapsLog(long logId, Offsets offsets) + { + long range = ranges.get(logId); + if (range == NO_VALUE) + return false; + return offsets.overlaps(minOffset(range), maxOffset(range)); + } + + @VisibleForTesting + static StaticOffsetRanges of(long logId, int minOffset, int maxOffset) + { + Long2LongHashMap map = new Long2LongHashMap(1, 0.65f, NO_VALUE, false); + map.put(logId, range(minOffset, maxOffset)); + return new StaticOffsetRanges(map); + } + + @VisibleForTesting + static StaticOffsetRanges of(Map<Long, Long> rangesMap) + { + Long2LongHashMap map = new Long2LongHashMap(rangesMap.size(), 0.65f, NO_VALUE, false); + for (Map.Entry<Long, Long> entry : rangesMap.entrySet()) + map.put(entry.getKey().longValue(), entry.getValue().longValue()); + return new StaticOffsetRanges(map); + } + static StaticOffsetRanges read(DataInputPlus in) throws IOException { CRC32 crc = Crc.crc32(); diff --git a/src/java/org/apache/cassandra/replication/MutationJournalSegmentCompactor.java b/src/java/org/apache/cassandra/replication/MutationJournalSegmentCompactor.java index bc16343571..d25e2261c4 100644 --- a/src/java/org/apache/cassandra/replication/MutationJournalSegmentCompactor.java +++ b/src/java/org/apache/cassandra/replication/MutationJournalSegmentCompactor.java @@ -84,7 +84,7 @@ class MutationJournalSegmentCompactor implements SegmentCompactor<ShortMutationI for (StaticSegment<ShortMutationId, Mutation> segment : candidates) { if (!segment.metadata().needsReplay() - && !journal.segmentReferenceTracker().isReferenced(segment.id()) + && !journal.segmentReferenceTracker().isReferenced(segment) && ((MutationJournal.StaticOffsetRanges) segment.keyStats()).isFullyCovered(durablyReconciled)) selected.add(segment); } diff --git a/src/java/org/apache/cassandra/replication/Offsets.java b/src/java/org/apache/cassandra/replication/Offsets.java index 4430a95556..6465953804 100644 --- a/src/java/org/apache/cassandra/replication/Offsets.java +++ b/src/java/org/apache/cassandra/replication/Offsets.java @@ -209,6 +209,28 @@ public abstract class Offsets implements Iterable<ShortMutationId> return maxOffset <= bounds[end]; } + /** + * @return whether any offset in the inclusive range [minOffset, maxOffset] is present in these Offsets + */ + public boolean overlaps(int minOffset, int maxOffset) + { + if (size == 0 || minOffset > maxOffset) + return false; + + if (maxOffset < bounds[0] || minOffset > bounds[size - 1]) + return false; + + int pos = Arrays.binarySearch(bounds, 0, size, minOffset); + if (pos >= 0) + return true; + + int insertionPoint = -pos - 1; + if (insertionPoint % 2 != 0) + return true; + + return bounds[insertionPoint] <= maxOffset; + } + public void digest(Digest digest) { digest.updateWithLong(logId.asLong()); diff --git a/src/java/org/apache/cassandra/replication/SegmentReferenceTracker.java b/src/java/org/apache/cassandra/replication/SegmentReferenceTracker.java index 45c9f95517..b636c24f1b 100644 --- a/src/java/org/apache/cassandra/replication/SegmentReferenceTracker.java +++ b/src/java/org/apache/cassandra/replication/SegmentReferenceTracker.java @@ -20,22 +20,25 @@ package org.apache.cassandra.replication; import java.util.ArrayList; import java.util.Collection; import java.util.Collections; +import java.util.HashMap; import java.util.HashSet; -import java.util.Iterator; import java.util.List; +import java.util.Map; import java.util.Set; import java.util.UUID; import java.util.concurrent.locks.ReentrantLock; -import java.util.function.LongConsumer; import java.util.function.Supplier; import com.google.common.annotations.VisibleForTesting; import org.agrona.collections.Long2ObjectHashMap; +import org.agrona.collections.LongHashSet; +import org.agrona.collections.LongHashSet.LongIterator; -import org.apache.cassandra.db.commitlog.CommitLogPosition; -import org.apache.cassandra.db.commitlog.IntervalSet; +import org.apache.cassandra.db.Mutation; import org.apache.cassandra.io.sstable.format.SSTableReader; +import org.apache.cassandra.journal.KeyStats; +import org.apache.cassandra.journal.Segment; import org.apache.cassandra.notifications.INotification; import org.apache.cassandra.notifications.INotificationConsumer; import org.apache.cassandra.notifications.InitialSSTableAddedNotification; @@ -43,59 +46,47 @@ import org.apache.cassandra.notifications.SSTableAddedNotification; import org.apache.cassandra.notifications.SSTableListChangedNotification; import org.apache.cassandra.notifications.SSTableRepairStatusChanged; import org.apache.cassandra.schema.ReplicationType; -import org.apache.cassandra.service.StorageService; /** - * Tracks, for each mutation journal segment, the set of local unrepaired sstables of tracked tables that - * reference it. + * Tracks, for each mutation journal segment, whether any unreconciled sstables could reference mutation ids it contains * - * <p>A sstable references a segment iff the sstable's {@code StatsMetadata.commitLogIntervals} — - * which for tracked tables stores mutation journal positions, not commit log positions — covers - * any position within that segment. The granularity is coarse: the referenced range is the union - * of the memtable lower/upper bounds for the sstable's lineage. This matches existing commit log - * retention semantics; it can include a segment with no actual data from this sstable, which only - * defers (never incorrectly enables) segment dropping. + * <p>An sstable references a segment if its {@code StatsMetadata.coordinatorLogOffsets} overlap with the segment's + * recorded offset ranges for any coordinator log. * - * <p>Used by {@link MutationJournal} to decide when a static segment may be dropped: a segment may be - * dropped only once it has no unrepaired references and {@code !needsReplay}. This guarantees the - * journal can rebuild any unrepaired sstable from the journal if minority writes need to be filtered - * out (CASSANDRA-21407). + * <p>Used by {@link MutationJournal} to decide when a static segment can be dropped: a segment can be + * dropped only when it has no unrepaired references and {@code !needsReplay}. This guarantees the journal can rebuild + * any unrepaired sstable from the journal if minority writes need to be filtered out (CASSANDRA-21407). * * <p>A per-segment <em>set</em> of referrers (rather than a bare refcount) is kept so that individual * referrers can be located and dropped from a segment's set — needed to reason about keyspaces migrating * to and from tracked replication (CASSANDRA-21406) and to surface which sstables hold a segment. + * */ public class SegmentReferenceTracker implements INotificationConsumer { - // Guards both referrersBySegment and trackedSstables to keep transitions atomic across notifications. private final ReentrantLock lock = new ReentrantLock(); - // segment id -> set of local unrepaired sstables referencing it + // Index mapping each segment id to the set of unrepaired SSTables that overlap it. + // Used by the compactor to quickly determine if a segment can be dropped. private final Long2ObjectHashMap<Set<SSTableReader>> referrersBySegment = new Long2ObjectHashMap<>(); - // Sstables we currently hold refs for. - // Required so SSTableRepairStatusChanged can transition an sstable in/out without the notification - // having to carry the previous repair state. - private final Set<SSTableReader> trackedSSTables = new HashSet<>(); + // Reverse index mapping each tracked SSTable to the journal segment ids it references. + // Used to remove references across all impacted segments when an SSTable is deleted or repaired. + private final Map<SSTableReader, LongHashSet> segmentsBySSTable = new HashMap<>(); - // Invoked whenever a notification drives at least one segment's reference count to zero, - // so the journal can attempt to drop the now-unreferenced segment(s). private final Runnable onSegmentsUnreferenced; - // Resolves the local host id. May return null very early in startup, in which case no sstable is - // treated as locally originated. private final Supplier<UUID> localHostIdSupplier; - public SegmentReferenceTracker(Runnable onSegmentsUnreferenced) - { - this(onSegmentsUnreferenced, StorageService.instance::getLocalHostUUID); - } + private final Supplier<Iterable<Segment<ShortMutationId, Mutation>>> allSegmentsSupplier; - @VisibleForTesting - SegmentReferenceTracker(Runnable onSegmentsUnreferenced, Supplier<UUID> localHostIdSupplier) + public SegmentReferenceTracker(Runnable onSegmentsUnreferenced, + Supplier<UUID> localHostIdSupplier, + Supplier<Iterable<Segment<ShortMutationId, Mutation>>> allSegmentsSupplier) { this.onSegmentsUnreferenced = onSegmentsUnreferenced; this.localHostIdSupplier = localHostIdSupplier; + this.allSegmentsSupplier = allSegmentsSupplier; } @Override @@ -111,6 +102,15 @@ public class SegmentReferenceTracker implements INotificationConsumer onRepairStatusChanged(((SSTableRepairStatusChanged) notification).sstables); } + /** + * @param segment the journal segment + * @return whether any unrepaired local sstable references the given segment + */ + public boolean isReferenced(Segment<ShortMutationId, Mutation> segment) + { + return segment != null && isReferenced(segment.id()); + } + /** * @param segmentId the identifier for the segment * @return whether any unrepaired local sstable references the given segment id @@ -120,7 +120,15 @@ public class SegmentReferenceTracker implements INotificationConsumer lock.lock(); try { - return referrersBySegment.containsKey(segmentId); + Set<SSTableReader> referrers = referrersBySegment.get(segmentId); + if (referrers == null) + return false; + if (referrers.isEmpty()) + { + referrersBySegment.remove(segmentId); + return false; + } + return true; } finally { @@ -148,8 +156,6 @@ public class SegmentReferenceTracker implements INotificationConsumer lock.lock(); try { - // Process additions before removals so refcounts are never observed briefly empty - // between a compaction's input drop and output add when both span the same segment. for (SSTableReader sstable : notification.added) acquireIfTracked(sstable); for (SSTableReader sstable : notification.removed) @@ -186,8 +192,7 @@ public class SegmentReferenceTracker implements INotificationConsumer } /** - * Release every reference currently held for the given sstables, dropping each from every segment's referrer - * set. + * Release every reference currently held for the given sstables. */ public void evict(Iterable<SSTableReader> sstables) { @@ -214,9 +219,12 @@ public class SegmentReferenceTracker implements INotificationConsumer lock.lock(); try { - if (trackedSSTables.isEmpty()) + if (segmentsBySSTable.isEmpty()) + return Set.of(); + segmentsBySSTable.entrySet().removeIf(entry -> entry.getValue() == null || entry.getValue().isEmpty()); + if (segmentsBySSTable.isEmpty()) return Set.of(); - return Set.copyOf(trackedSSTables); + return Set.copyOf(segmentsBySSTable.keySet()); } finally { @@ -224,10 +232,6 @@ public class SegmentReferenceTracker implements INotificationConsumer } } - /** - * @return true when all these conditions are true, the sstable is locally originated, unrepaired, carries - * tracked mutations, and it belongs to a tracked table - */ boolean shouldTrack(SSTableReader sstable) { return isLocallyOriginated(sstable) @@ -251,58 +255,77 @@ public class SegmentReferenceTracker implements INotificationConsumer private void acquireIfTracked(SSTableReader sstable) { - if (shouldTrack(sstable) && trackedSSTables.add(sstable)) - forEachSegment(sstable, segmentId -> - referrersBySegment.computeIfAbsent(segmentId, k -> new HashSet<>()).add(sstable)); + if (!shouldTrack(sstable) || segmentsBySSTable.containsKey(sstable)) + return; + + LongHashSet segmentIds = new LongHashSet(); + if (allSegmentsSupplier != null) + { + Iterable<Segment<ShortMutationId, Mutation>> allSegments = allSegmentsSupplier.get(); + if (allSegments != null) + { + for (Segment<ShortMutationId, Mutation> segment : allSegments) + { + KeyStats<ShortMutationId> keyStats = segment.keyStats(); + if (keyStats instanceof MutationJournal.OffsetRanges) + { + MutationJournal.OffsetRanges segRanges = (MutationJournal.OffsetRanges) keyStats; + if (segRanges.overlaps(sstable.getCoordinatorLogOffsets())) + { + segmentIds.add(segment.id()); + referrersBySegment.computeIfAbsent(segment.id(), k -> new HashSet<>()).add(sstable); + } + } + } + } + } + if (!segmentIds.isEmpty()) + segmentsBySSTable.put(sstable, segmentIds); } - /** - * @return true if releasing this sstable emptied at least one segment's referrer set - */ private boolean releaseIfTracked(SSTableReader sstable) { - if (!trackedSSTables.remove(sstable)) + LongHashSet segmentIds = segmentsBySSTable.remove(sstable); + if (segmentIds == null || segmentIds.isEmpty()) return false; - boolean[] anyEmptied = { false }; - forEachSegment(sstable, segmentId -> { + + boolean anyEmptied = false; + LongIterator idIterator = segmentIds.iterator(); + while (idIterator.hasNext()) + { + long segmentId = idIterator.nextValue(); Set<SSTableReader> referrers = referrersBySegment.get(segmentId); if (referrers != null && referrers.remove(sstable) && referrers.isEmpty()) { referrersBySegment.remove(segmentId); - anyEmptied[0] = true; + anyEmptied = true; } - }); - return anyEmptied[0]; - } - - private static void forEachSegment(SSTableReader sstable, LongConsumer consumer) - { - IntervalSet<CommitLogPosition> intervals = sstable.getSSTableMetadata().commitLogIntervals; - if (intervals.isEmpty()) - return; - - // IntervalSet guarantees starts and ends are returned in matching order. - Iterator<CommitLogPosition> startIt = intervals.starts().iterator(); - Iterator<CommitLogPosition> endIt = intervals.ends().iterator(); - while (startIt.hasNext()) - { - CommitLogPosition start = startIt.next(); - CommitLogPosition end = endIt.next(); - for (long s = start.segmentId; s <= end.segmentId; s++) - consumer.accept(s); } + return anyEmptied; } /** * Number of unrepaired local sstables currently holding the given segment (for diagnostics / the vtable). */ - int referenceCount(long segmentId) + public int referenceCount(Segment<ShortMutationId, Mutation> segment) + { + return segment != null ? referenceCount(segment.id()) : 0; + } + + public int referenceCount(long segmentId) { lock.lock(); try { Set<SSTableReader> referrers = referrersBySegment.get(segmentId); - return referrers == null ? 0 : referrers.size(); + if (referrers == null) + return 0; + if (referrers.isEmpty()) + { + referrersBySegment.remove(segmentId); + return 0; + } + return referrers.size(); } finally { @@ -313,6 +336,11 @@ public class SegmentReferenceTracker implements INotificationConsumer /** * Sorted base filenames of the sstables currently holding the given segment (for diagnostics / the vtable). */ + public List<String> referrerDescriptors(Segment<ShortMutationId, Mutation> segment) + { + return segment != null ? referrerDescriptors(segment.id()) : Collections.emptyList(); + } + public List<String> referrerDescriptors(long segmentId) { lock.lock(); @@ -320,10 +348,14 @@ public class SegmentReferenceTracker implements INotificationConsumer { Set<SSTableReader> referrers = referrersBySegment.get(segmentId); if (referrers == null || referrers.isEmpty()) + { + if (referrers != null) + referrersBySegment.remove(segmentId); return Collections.emptyList(); + } List<String> names = new ArrayList<>(referrers.size()); for (SSTableReader sstable : referrers) - names.add(sstable.descriptor.baseFile().name()); + names.add(sstable.descriptor != null ? sstable.descriptor.baseFile().name() : sstable.getFilename()); Collections.sort(names); return names; } @@ -333,6 +365,12 @@ public class SegmentReferenceTracker implements INotificationConsumer } } + @VisibleForTesting + long referenceCountForTesting(Segment<ShortMutationId, Mutation> segment) + { + return referenceCount(segment); + } + @VisibleForTesting long referenceCountForTesting(long segmentId) { @@ -345,7 +383,8 @@ public class SegmentReferenceTracker implements INotificationConsumer lock.lock(); try { - return trackedSSTables.size(); + segmentsBySSTable.entrySet().removeIf(entry -> entry.getValue() == null || entry.getValue().isEmpty()); + return segmentsBySSTable.size(); } finally { @@ -360,6 +399,7 @@ public class SegmentReferenceTracker implements INotificationConsumer try { referrersBySegment.computeIfAbsent(segmentId, k -> new HashSet<>()).add(referrer); + segmentsBySSTable.computeIfAbsent(referrer, k -> new LongHashSet()).add(segmentId); } finally { @@ -370,16 +410,30 @@ public class SegmentReferenceTracker implements INotificationConsumer @VisibleForTesting void removeReferenceForTesting(long segmentId, SSTableReader referrer) { + boolean removed = false; lock.lock(); try { Set<SSTableReader> referrers = referrersBySegment.get(segmentId); - if (referrers != null && referrers.remove(referrer) && referrers.isEmpty()) - referrersBySegment.remove(segmentId); + if (referrers != null && referrers.remove(referrer)) + { + if (referrers.isEmpty()) + referrersBySegment.remove(segmentId); + removed = true; + } + LongHashSet segmentIds = segmentsBySSTable.get(referrer); + if (segmentIds != null) + { + segmentIds.remove(segmentId); + if (segmentIds.isEmpty()) + segmentsBySSTable.remove(referrer); + } } finally { lock.unlock(); } + if (removed) + onSegmentsUnreferenced.run(); } } diff --git a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationJournalSegmentRefcountTest.java b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationJournalSegmentRefcountTest.java index b21f3bddcd..b5bcd808b4 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationJournalSegmentRefcountTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationJournalSegmentRefcountTest.java @@ -84,6 +84,21 @@ public class MutationJournalSegmentRefcountTest extends TestBaseImpl before, after); })); + int preBounceCount = cluster.get(1).callOnInstance(() -> MutationJournal.instance().countStaticSegmentsForTesting()); + assertTrue(preBounceCount > 0); + + // Restart a replica to confirm segment references survive restart and keep static segments pinned. + cluster.get(1).shutdown().get(); + cluster.get(1).startup(); + cluster.get(1).runOnInstance(() -> { + int postBounceCount = MutationJournal.instance().countStaticSegmentsForTesting(); + assertEquals(preBounceCount, postBounceCount); + + MutationTrackingService.instance().persistLogStateForTesting(); + int postDropCount = MutationJournal.instance().countStaticSegmentsForTesting(); + assertEquals(preBounceCount, postDropCount); + }); + // Restore broadcast, exchange witnesses, and persist so isDurablyReconciled is now true everywhere. cluster.filters().reset(); cluster.forEach(i -> i.runOnInstance(() -> MutationTrackingService.instance().broadcastOffsetsForTesting())); diff --git a/test/unit/org/apache/cassandra/replication/MutationJournalTest.java b/test/unit/org/apache/cassandra/replication/MutationJournalTest.java index b9805b01df..4d75a1e0dd 100644 --- a/test/unit/org/apache/cassandra/replication/MutationJournalTest.java +++ b/test/unit/org/apache/cassandra/replication/MutationJournalTest.java @@ -146,6 +146,39 @@ public class MutationJournalTest } } + @Test + public void testActiveOffsetRangesOverlaps() + { + ActiveOffsetRanges ranges = new ActiveOffsetRanges(); + ranges.update(id(100L, 10)); + ranges.update(id(100L, 50)); + ranges.update(id(200L, 30)); + ranges.update(id(200L, 60)); + + assertFalse(ranges.overlaps(null)); + assertFalse(ranges.overlaps(ImmutableCoordinatorLogOffsets.NONE)); + + ImmutableCoordinatorLogOffsets noMatchLog = new ImmutableCoordinatorLogOffsets.Builder() + .add(300L, 10, 50) + .build(); + assertFalse(ranges.overlaps(noMatchLog)); + + ImmutableCoordinatorLogOffsets noMatchOffset = new ImmutableCoordinatorLogOffsets.Builder() + .add(100L, 60, 80) + .build(); + assertFalse(ranges.overlaps(noMatchOffset)); + + ImmutableCoordinatorLogOffsets match100 = new ImmutableCoordinatorLogOffsets.Builder() + .add(100L, 5, 15) + .build(); + assertTrue(ranges.overlaps(match100)); + + ImmutableCoordinatorLogOffsets match200 = new ImmutableCoordinatorLogOffsets.Builder() + .add(200L, 40, 45) + .build(); + assertTrue(ranges.overlaps(match200)); + } + @Test public void testStaticOffsetRanges() { diff --git a/test/unit/org/apache/cassandra/replication/OffsetsTest.java b/test/unit/org/apache/cassandra/replication/OffsetsTest.java index aff871337f..d6348b7c11 100644 --- a/test/unit/org/apache/cassandra/replication/OffsetsTest.java +++ b/test/unit/org/apache/cassandra/replication/OffsetsTest.java @@ -908,4 +908,47 @@ public class OffsetsTest assertFalse(offsets.containsRange(9, 9)); } } + + @Test + public void testOverlaps() + { + { + Offsets.Mutable offsets = offsets(); + assertFalse(offsets.overlaps(0, 1)); + assertFalse(offsets.overlaps(2, 4)); + } + + { + Offsets.Mutable offsets = offsets(2, 4); + + assertTrue(offsets.overlaps(2, 4)); + assertTrue(offsets.overlaps(3, 4)); + assertTrue(offsets.overlaps(2, 3)); + assertTrue(offsets.overlaps(1, 2)); + assertTrue(offsets.overlaps(1, 3)); + assertTrue(offsets.overlaps(1, 5)); + assertTrue(offsets.overlaps(4, 5)); + + assertFalse(offsets.overlaps(0, 1)); + assertFalse(offsets.overlaps(5, 6)); + assertFalse(offsets.overlaps(4, 2)); + } + + { + Offsets.Mutable offsets = offsets(2, 4, 6, 8); + + assertTrue(offsets.overlaps(2, 4)); + assertTrue(offsets.overlaps(6, 8)); + assertTrue(offsets.overlaps(0, 2)); + assertTrue(offsets.overlaps(3, 5)); + assertTrue(offsets.overlaps(4, 6)); + assertTrue(offsets.overlaps(5, 7)); + assertTrue(offsets.overlaps(7, 9)); + assertTrue(offsets.overlaps(0, 10)); + + assertFalse(offsets.overlaps(0, 1)); + assertFalse(offsets.overlaps(5, 5)); + assertFalse(offsets.overlaps(9, 10)); + } + } } diff --git a/test/unit/org/apache/cassandra/replication/SegmentReferenceTrackerTest.java b/test/unit/org/apache/cassandra/replication/SegmentReferenceTrackerTest.java index c1504ae205..4d68bd99f9 100644 --- a/test/unit/org/apache/cassandra/replication/SegmentReferenceTrackerTest.java +++ b/test/unit/org/apache/cassandra/replication/SegmentReferenceTrackerTest.java @@ -17,7 +17,9 @@ */ package org.apache.cassandra.replication; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.UUID; import java.util.concurrent.atomic.AtomicReference; import java.util.function.BooleanSupplier; @@ -25,13 +27,14 @@ import java.util.function.BooleanSupplier; import org.junit.Test; import org.mockito.Mockito; +import org.apache.cassandra.db.Mutation; import org.apache.cassandra.db.Slice; -import org.apache.cassandra.db.commitlog.CommitLogPosition; import org.apache.cassandra.db.commitlog.IntervalSet; import org.apache.cassandra.db.compaction.OperationType; import org.apache.cassandra.db.rows.Cell; import org.apache.cassandra.io.sstable.format.SSTableReader; import org.apache.cassandra.io.sstable.metadata.StatsMetadata; +import org.apache.cassandra.journal.Segment; import org.apache.cassandra.notifications.InitialSSTableAddedNotification; import org.apache.cassandra.notifications.SSTableAddedNotification; import org.apache.cassandra.notifications.SSTableListChangedNotification; @@ -51,19 +54,24 @@ public class SegmentReferenceTrackerTest { private static final UUID LOCAL_HOST = UUID.randomUUID(); private static final UUID REMOTE_HOST = UUID.randomUUID(); + private static final long LOG_ID = 100L; @Test public void testInitialAddRefsEverySegmentInTheInterval() { - int startSegment = 5; - int endSegment = 7; - SegmentReferenceTracker tracker = newTracker(); - SSTableReader sstable = unrepaired(intervals(startSegment, 0, endSegment, 100)); + Map<Long, Segment<ShortMutationId, Mutation>> segments = new HashMap<>(); + segments.put(4L, mockSegment(4L, LOG_ID, 0, 5)); + segments.put(5L, mockSegment(5L, LOG_ID, 10, 20)); + segments.put(6L, mockSegment(6L, LOG_ID, 25, 40)); + segments.put(7L, mockSegment(7L, LOG_ID, 45, 60)); + segments.put(8L, mockSegment(8L, LOG_ID, 65, 80)); + + SegmentReferenceTracker tracker = newTracker(segments); + SSTableReader sstable = unrepaired(offsets(LOG_ID, 10, 60)); tracker.handleNotification(new InitialSSTableAddedNotification(List.of(sstable)), null); - // Coarse range covers segments 5..7 inclusive. - for (long segment = startSegment; segment <= endSegment; segment++) + for (long segment = 5; segment <= 7; segment++) assertEquals("segment " + segment, 1L, tracker.referenceCountForTesting(segment)); assertFalse(tracker.isReferenced(4)); assertFalse(tracker.isReferenced(8)); @@ -73,14 +81,17 @@ public class SegmentReferenceTrackerTest @Test public void testAddedRepairedSSTableHoldsNoRefs() { - int startSegment = 5; - int endSegment = 7; - SegmentReferenceTracker tracker = newTracker(); - SSTableReader sstable = repaired(intervals(startSegment, 0, endSegment, 100)); + Map<Long, Segment<ShortMutationId, Mutation>> segments = new HashMap<>(); + segments.put(5L, mockSegment(5L, LOG_ID, 10, 20)); + segments.put(6L, mockSegment(6L, LOG_ID, 25, 40)); + segments.put(7L, mockSegment(7L, LOG_ID, 45, 60)); + + SegmentReferenceTracker tracker = newTracker(segments); + SSTableReader sstable = repaired(offsets(LOG_ID, 10, 60)); tracker.handleNotification(new SSTableAddedNotification(List.of(sstable), null), null); - for (long segment = startSegment; segment <= endSegment; segment++) + for (long segment = 5; segment <= 7; segment++) assertFalse("segment " + segment, tracker.isReferenced(segment)); assertEquals(0, tracker.trackedSstableCountForTesting()); } @@ -88,48 +99,45 @@ public class SegmentReferenceTrackerTest @Test public void testUnrepairedSSTableWithoutCoordinatorLogOffsetsHoldsNoRefs() { - SegmentReferenceTracker tracker = newTracker(); - // Unrepaired, but carries no tracked mutations (empty coordinatorLogOffsets) -> not tracked. This is the - // untracked / pre-migration case: such an sstable's commitLogIntervals reference the commit log, not the - // mutation journal, so it must never hold a segment reference. - SSTableReader sstable = sstable(intervals(5, 0, 7, 100), () -> false, coordinatorLogOffsets(false)); + Map<Long, Segment<ShortMutationId, Mutation>> segments = new HashMap<>(); + segments.put(5L, mockSegment(5L, LOG_ID, 10, 20)); + + SegmentReferenceTracker tracker = newTracker(segments); + SSTableReader sstable = sstable(ImmutableCoordinatorLogOffsets.NONE, () -> false, LOCAL_HOST, ReplicationType.tracked); tracker.handleNotification(new SSTableAddedNotification(List.of(sstable), null), null); - for (long segment = 5; segment <= 7; segment++) - assertFalse("segment " + segment, tracker.isReferenced(segment)); + assertFalse(tracker.isReferenced(5)); assertEquals(0, tracker.trackedSstableCountForTesting()); } @Test public void testStreamedSSTableHoldsNoRefs() { - SegmentReferenceTracker tracker = newTracker(); - // Unrepaired with tracked mutations, but streamed from another host: its commitLogIntervals reference that - // host's journal segments, not ours, so it must never hold a local segment reference (CASSANDRA-21406). - SSTableReader sstable = sstable(intervals(5, 0, 7, 100), () -> false, coordinatorLogOffsets(true), REMOTE_HOST); + Map<Long, Segment<ShortMutationId, Mutation>> segments = new HashMap<>(); + segments.put(5L, mockSegment(5L, LOG_ID, 10, 20)); + + SegmentReferenceTracker tracker = newTracker(segments); + SSTableReader sstable = sstable(offsets(LOG_ID, 10, 20), () -> false, REMOTE_HOST, ReplicationType.tracked); tracker.handleNotification(new SSTableAddedNotification(List.of(sstable), null), null); - for (long segment = 5; segment <= 7; segment++) - assertFalse("segment " + segment, tracker.isReferenced(segment)); + assertFalse(tracker.isReferenced(5)); assertEquals(0, tracker.trackedSstableCountForTesting()); } @Test public void testUntrackedTableSSTableHoldsNoRefs() { - SegmentReferenceTracker tracker = newTracker(); - // Local + unrepaired + non-empty coordinatorLogOffsets, but its table has migrated away from tracked. Such - // an sstable (or a compaction output that inherits its offsets) is never promoted to repaired, so it must - // not be counted or its segments would be pinned forever (CASSANDRA-21406). - SSTableReader sstable = sstable(intervals(5, 0, 7, 100), () -> false, coordinatorLogOffsets(true), - LOCAL_HOST, ReplicationType.untracked); + Map<Long, Segment<ShortMutationId, Mutation>> segments = new HashMap<>(); + segments.put(5L, mockSegment(5L, LOG_ID, 10, 20)); + + SegmentReferenceTracker tracker = newTracker(segments); + SSTableReader sstable = sstable(offsets(LOG_ID, 10, 20), () -> false, LOCAL_HOST, ReplicationType.untracked); tracker.handleNotification(new SSTableAddedNotification(List.of(sstable), null), null); - for (long segment = 5; segment <= 7; segment++) - assertFalse("segment " + segment, tracker.isReferenced(segment)); + assertFalse(tracker.isReferenced(5)); assertEquals(0, tracker.trackedSstableCountForTesting()); } @@ -137,14 +145,17 @@ public class SegmentReferenceTrackerTest public void testEvictReleasesReferencesAndFiresCallback() { int[] calls = { 0 }; - SegmentReferenceTracker tracker = newTracker(() -> calls[0]++); - SSTableReader sstable = unrepaired(intervals(5, 0, 7, 0)); + Map<Long, Segment<ShortMutationId, Mutation>> segments = new HashMap<>(); + for (long s = 5; s <= 7; s++) + segments.put(s, mockSegment(s, LOG_ID, (int) s * 10, (int) s * 10 + 9)); + + SegmentReferenceTracker tracker = newTracker(segments, () -> calls[0]++); + SSTableReader sstable = unrepaired(offsets(LOG_ID, 50, 79)); tracker.handleNotification(new SSTableAddedNotification(List.of(sstable), null), null); for (long segment = 5; segment <= 7; segment++) assertEquals(1L, tracker.referenceCountForTesting(segment)); - // Simulates a keyspace migrating away from tracked: evict its sstables' references directly. tracker.evict(List.of(sstable)); for (long segment = 5; segment <= 7; segment++) @@ -156,14 +167,16 @@ public class SegmentReferenceTrackerTest @Test public void testEvictOnlyReleasesGivenSstables() { - SegmentReferenceTracker tracker = newTracker(); - SSTableReader a = unrepaired(intervals(5, 0, 5, 100)); - SSTableReader b = unrepaired(intervals(5, 0, 5, 200)); + Map<Long, Segment<ShortMutationId, Mutation>> segments = new HashMap<>(); + segments.put(5L, mockSegment(5L, LOG_ID, 10, 20)); + + SegmentReferenceTracker tracker = newTracker(segments); + SSTableReader a = unrepaired(offsets(LOG_ID, 10, 15), "a"); + SSTableReader b = unrepaired(offsets(LOG_ID, 16, 20), "b"); tracker.handleNotification(new SSTableAddedNotification(List.of(a, b), null), null); assertEquals(2L, tracker.referenceCountForTesting(5)); - // Evicting only 'a' leaves segment 5 referenced by 'b'; evicting an untracked sstable is a no-op. tracker.evict(List.of(a)); assertTrue(tracker.isReferenced(5)); assertEquals(1L, tracker.referenceCountForTesting(5)); @@ -176,10 +189,11 @@ public class SegmentReferenceTrackerTest @Test public void testAddIsIdempotent() { - int startSegment = 5; - int endSegment = 7; - SegmentReferenceTracker tracker = newTracker(); - SSTableReader sstable = unrepaired(intervals(startSegment, 0, endSegment, 100)); + Map<Long, Segment<ShortMutationId, Mutation>> segments = new HashMap<>(); + segments.put(5L, mockSegment(5L, LOG_ID, 10, 20)); + + SegmentReferenceTracker tracker = newTracker(segments); + SSTableReader sstable = unrepaired(offsets(LOG_ID, 10, 20)); tracker.handleNotification(new SSTableAddedNotification(List.of(sstable), null), null); tracker.handleNotification(new SSTableAddedNotification(List.of(sstable), null), null); @@ -191,79 +205,64 @@ public class SegmentReferenceTrackerTest @Test public void testMultipleDisjointIntervalsRefEachContainedSegment() { - SegmentReferenceTracker tracker = newTracker(); - // Two disjoint intervals: [3:0..3:100] and [9:0..10:50]. - IntervalSet.Builder<CommitLogPosition> builder = new IntervalSet.Builder<>(); - builder.add(new CommitLogPosition(3, 0), new CommitLogPosition(3, 100)); - builder.add(new CommitLogPosition(9, 0), new CommitLogPosition(10, 50)); - SSTableReader sstable = unrepaired(builder.build()); + Map<Long, Segment<ShortMutationId, Mutation>> segments = new HashMap<>(); + segments.put(3L, mockSegment(3L, LOG_ID, 0, 100)); + segments.put(5L, mockSegment(5L, LOG_ID, 200, 300)); + segments.put(9L, mockSegment(9L, LOG_ID, 500, 550)); + segments.put(10L, mockSegment(10L, LOG_ID, 551, 600)); + + SegmentReferenceTracker tracker = newTracker(segments); + ImmutableCoordinatorLogOffsets offsets = new ImmutableCoordinatorLogOffsets.Builder() + .add(LOG_ID, 0, 100) + .add(LOG_ID, 500, 600) + .build(); + SSTableReader sstable = unrepaired(offsets); tracker.handleNotification(new InitialSSTableAddedNotification(List.of(sstable)), null); assertEquals(1L, tracker.referenceCountForTesting(3)); assertEquals(1L, tracker.referenceCountForTesting(9)); assertEquals(1L, tracker.referenceCountForTesting(10)); - // Gap between disjoint intervals is not referenced. - for (long gap = 4; gap <= 8; gap++) - assertFalse("segment " + gap, tracker.isReferenced(gap)); + assertFalse(tracker.isReferenced(5)); } @Test public void testMultipleIntervalsInSameSegmentCountedOnce() { - SegmentReferenceTracker tracker = newTracker(); - // Two disjoint intervals within the SAME segment (different position ranges): forEachSegment emits - // segment 5 twice for this sstable, but the referrer set de-dupes so it is counted exactly once. - IntervalSet.Builder<CommitLogPosition> builder = new IntervalSet.Builder<>(); - builder.add(new CommitLogPosition(5, 0), new CommitLogPosition(5, 100)); - builder.add(new CommitLogPosition(5, 200), new CommitLogPosition(5, 300)); - SSTableReader sstable = unrepaired(builder.build()); - - tracker.handleNotification(new SSTableAddedNotification(List.of(sstable), null), null); - - assertEquals("a segment spanned by multiple intervals of one sstable is referenced once, not n times", - 1L, tracker.referenceCountForTesting(5)); - assertEquals(1, tracker.trackedSstableCountForTesting()); - - // Releasing clears the single reference (no residual double-count). - tracker.handleNotification( - new SSTableListChangedNotification(List.of(), List.of(sstable), OperationType.COMPACTION), - null); - assertFalse(tracker.isReferenced(5)); - assertEquals(0, tracker.trackedSstableCountForTesting()); - } + Map<Long, Segment<ShortMutationId, Mutation>> segments = new HashMap<>(); + segments.put(5L, mockSegment(5L, LOG_ID, 0, 300)); - @Test - public void testEmptyIntervalsHoldNoRefs() - { - SegmentReferenceTracker tracker = newTracker(); - SSTableReader sstable = unrepaired(IntervalSet.empty()); + SegmentReferenceTracker tracker = newTracker(segments); + ImmutableCoordinatorLogOffsets offsets = new ImmutableCoordinatorLogOffsets.Builder() + .add(LOG_ID, 0, 100) + .add(LOG_ID, 200, 300) + .build(); + SSTableReader sstable = unrepaired(offsets); tracker.handleNotification(new SSTableAddedNotification(List.of(sstable), null), null); - // Tracked sstable but no segments to ref. - assertEquals(1, tracker.trackedSstableCountForTesting()); - assertFalse(tracker.isReferenced(0)); + assertEquals(1L, tracker.referenceCountForTesting(5)); + assertEquals(List.of(sstable.getFilename()), tracker.referrerDescriptors(5)); } @Test public void testCompactionPreservesRefsWhenInputAndOutputOverlapSameSegment() { - SegmentReferenceTracker tracker = newTracker(); - SSTableReader input = unrepaired(intervals(5, 0, 5, 100)); - SSTableReader output = unrepaired(intervals(5, 0, 5, 200)); + Map<Long, Segment<ShortMutationId, Mutation>> segments = new HashMap<>(); + segments.put(5L, mockSegment(5L, LOG_ID, 10, 50)); + + SegmentReferenceTracker tracker = newTracker(segments); + SSTableReader input = unrepaired(offsets(LOG_ID, 10, 30), "input"); + SSTableReader output = unrepaired(offsets(LOG_ID, 10, 50), "output"); tracker.handleNotification(new SSTableAddedNotification(List.of(input), null), null); assertEquals(1L, tracker.referenceCountForTesting(5)); - // Compaction emits the SSTableListChangedNotification with added + removed atomically. - tracker.handleNotification( - new SSTableListChangedNotification(List.of(output), - List.of(input), - OperationType.COMPACTION), - null); + tracker.handleNotification(new SSTableListChangedNotification(List.of(output), + List.of(input), + OperationType.COMPACTION), + null); - // Net: still one unrepaired sstable referencing segment 5. assertEquals(1L, tracker.referenceCountForTesting(5)); assertEquals(1, tracker.trackedSstableCountForTesting()); } @@ -271,18 +270,20 @@ public class SegmentReferenceTrackerTest @Test public void testCompactionToRepairedOutputReleasesRefs() { - SegmentReferenceTracker tracker = newTracker(); - SSTableReader input = unrepaired(intervals(5, 0, 5, 100)); - SSTableReader output = repaired(intervals(5, 0, 5, 200)); + Map<Long, Segment<ShortMutationId, Mutation>> segments = new HashMap<>(); + segments.put(5L, mockSegment(5L, LOG_ID, 10, 50)); + + SegmentReferenceTracker tracker = newTracker(segments); + SSTableReader input = unrepaired(offsets(LOG_ID, 10, 30)); + SSTableReader output = repaired(offsets(LOG_ID, 10, 50)); tracker.handleNotification(new SSTableAddedNotification(List.of(input), null), null); assertEquals(1L, tracker.referenceCountForTesting(5)); - tracker.handleNotification( - new SSTableListChangedNotification(List.of(output), - List.of(input), - OperationType.COMPACTION), - null); + tracker.handleNotification(new SSTableListChangedNotification(List.of(output), + List.of(input), + OperationType.COMPACTION), + null); assertFalse(tracker.isReferenced(5)); assertEquals(0, tracker.trackedSstableCountForTesting()); @@ -291,16 +292,53 @@ public class SegmentReferenceTrackerTest @Test public void testRepairPromotionReleasesRefs() { - SegmentReferenceTracker tracker = newTracker(); + Map<Long, Segment<ShortMutationId, Mutation>> segments = new HashMap<>(); + for (long s = 5; s <= 7; s++) + segments.put(s, mockSegment(s, LOG_ID, (int) s * 10, (int) s * 10 + 9)); + + SegmentReferenceTracker tracker = newTracker(segments); + AtomicReference<Boolean> repaired = new AtomicReference<>(false); + SSTableReader sstable = sstableWithRepairSupplier(offsets(LOG_ID, 50, 79), repaired::get); + + tracker.handleNotification(new SSTableAddedNotification(List.of(sstable), null), null); + for (long segment = 5; segment <= 7; segment++) + assertEquals(1L, tracker.referenceCountForTesting(segment)); + + repaired.set(true); + tracker.handleNotification(new SSTableRepairStatusChanged(List.of(sstable)), null); + + for (long segment = 5; segment <= 7; segment++) + assertFalse("segment " + segment, tracker.isReferenced(segment)); + assertEquals(0, tracker.trackedSstableCountForTesting()); + } + + @Test + public void testRepairPromotionWithClearedOffsetsReleasesRefs() + { + Map<Long, Segment<ShortMutationId, Mutation>> segments = new HashMap<>(); + for (long s = 5; s <= 7; s++) + segments.put(s, mockSegment(s, LOG_ID, (int) s * 10, (int) s * 10 + 9)); + + SegmentReferenceTracker tracker = newTracker(segments); AtomicReference<Boolean> repaired = new AtomicReference<>(false); - SSTableReader sstable = sstableWithRepairSupplier(intervals(5, 0, 7, 0), repaired::get); + AtomicReference<ImmutableCoordinatorLogOffsets> offsetsRef = new AtomicReference<>(offsets(LOG_ID, 50, 79)); + + SSTableReader sstable = Mockito.mock(SSTableReader.class); + Mockito.when(sstable.isRepaired()).thenAnswer(inv -> repaired.get()); + Mockito.when(sstable.getSSTableMetadata()).thenReturn(stats(LOCAL_HOST)); + Mockito.when(sstable.getCoordinatorLogOffsets()).thenAnswer(inv -> offsetsRef.get()); + TableMetadata tableMetadata = Mockito.mock(TableMetadata.class); + Mockito.when(tableMetadata.replicationType()).thenReturn(ReplicationType.tracked); + Mockito.when(sstable.metadata()).thenReturn(tableMetadata); + Mockito.when(sstable.getFilename()).thenReturn("sstable"); tracker.handleNotification(new SSTableAddedNotification(List.of(sstable), null), null); for (long segment = 5; segment <= 7; segment++) assertEquals(1L, tracker.referenceCountForTesting(segment)); - // Promote to repaired; repair-status-changed delivers the transition. + // When promoted, SSTableReader.mutatePromotedToRepairedAndReload marks repaired and clears coordinatorLogOffsets. repaired.set(true); + offsetsRef.set(ImmutableCoordinatorLogOffsets.NONE); tracker.handleNotification(new SSTableRepairStatusChanged(List.of(sstable)), null); for (long segment = 5; segment <= 7; segment++) @@ -311,15 +349,16 @@ public class SegmentReferenceTrackerTest @Test public void testRepairStatusFlippingBackToUnrepairedReAcquires() { - SegmentReferenceTracker tracker = newTracker(); + Map<Long, Segment<ShortMutationId, Mutation>> segments = new HashMap<>(); + segments.put(5L, mockSegment(5L, LOG_ID, 10, 20)); + + SegmentReferenceTracker tracker = newTracker(segments); AtomicReference<Boolean> repaired = new AtomicReference<>(true); - SSTableReader sstable = sstableWithRepairSupplier(intervals(5, 0, 5, 100), repaired::get); + SSTableReader sstable = sstableWithRepairSupplier(offsets(LOG_ID, 10, 20), repaired::get); - // Starts repaired -> add is a no-op. tracker.handleNotification(new SSTableAddedNotification(List.of(sstable), null), null); assertEquals(0, tracker.trackedSstableCountForTesting()); - // Flip back to unrepaired (e.g. failed repair session) and deliver repair-status-changed. repaired.set(false); tracker.handleNotification(new SSTableRepairStatusChanged(List.of(sstable)), null); @@ -330,9 +369,14 @@ public class SegmentReferenceTrackerTest @Test public void testMultipleSstablesAccumulateRefsOnSharedSegments() { - SegmentReferenceTracker tracker = newTracker(); - SSTableReader a = unrepaired(intervals(5, 0, 6, 0)); - SSTableReader b = unrepaired(intervals(6, 0, 7, 0)); + Map<Long, Segment<ShortMutationId, Mutation>> segments = new HashMap<>(); + segments.put(5L, mockSegment(5L, LOG_ID, 50, 59)); + segments.put(6L, mockSegment(6L, LOG_ID, 60, 69)); + segments.put(7L, mockSegment(7L, LOG_ID, 70, 79)); + + SegmentReferenceTracker tracker = newTracker(segments); + SSTableReader a = unrepaired(offsets(LOG_ID, 50, 65), "a"); + SSTableReader b = unrepaired(offsets(LOG_ID, 65, 79), "b"); tracker.handleNotification(new SSTableAddedNotification(List.of(a, b), null), null); @@ -340,48 +384,53 @@ public class SegmentReferenceTrackerTest assertEquals(2L, tracker.referenceCountForTesting(6)); assertEquals(1L, tracker.referenceCountForTesting(7)); - tracker.handleNotification( - new SSTableListChangedNotification(List.of(), - List.of(a), - OperationType.COMPACTION), - null); + tracker.handleNotification(new SSTableListChangedNotification(List.of(), + List.of(a), + OperationType.COMPACTION), + null); assertFalse(tracker.isReferenced(5)); - assertEquals("there's still a reference from the shared segment from b", 1L, tracker.referenceCountForTesting(6)); + assertEquals("reference from b remains", 1L, tracker.referenceCountForTesting(6)); assertEquals(1L, tracker.referenceCountForTesting(7)); } @Test public void testReleaseOfUntrackedSstableIsNoOp() { - SegmentReferenceTracker tracker = newTracker(); - SSTableReader sstable = unrepaired(intervals(5, 0, 5, 100)); + Map<Long, Segment<ShortMutationId, Mutation>> segments = new HashMap<>(); + segments.put(5L, mockSegment(5L, LOG_ID, 10, 20)); + + SegmentReferenceTracker tracker = newTracker(segments); + SSTableReader sstable = unrepaired(offsets(LOG_ID, 10, 20)); assertFalse(tracker.isReferenced(5)); - tracker.handleNotification( - new SSTableListChangedNotification(List.of(), - List.of(sstable), - OperationType.COMPACTION), - null); + tracker.handleNotification(new SSTableListChangedNotification(List.of(), + List.of(sstable), + OperationType.COMPACTION), + null); - assertFalse("ensure a never added sstable does not underflow during removal", tracker.isReferenced(5)); + assertFalse(tracker.isReferenced(5)); } @Test public void testCallbackFiredWhenLastReferenceReleased() { int[] calls = { 0 }; - SegmentReferenceTracker tracker = newTracker(() -> calls[0]++); - SSTableReader sstable = unrepaired(intervals(5, 0, 7, 0)); + Map<Long, Segment<ShortMutationId, Mutation>> segments = new HashMap<>(); + for (long s = 5; s <= 7; s++) + segments.put(s, mockSegment(s, LOG_ID, (int) s * 10, (int) s * 10 + 9)); + + SegmentReferenceTracker tracker = newTracker(segments, () -> calls[0]++); + SSTableReader sstable = unrepaired(offsets(LOG_ID, 50, 79)); tracker.handleNotification(new SSTableAddedNotification(List.of(sstable), null), null); - assertEquals("adding never fires the unreferenced callback", 0, calls[0]); + assertEquals("adding does not fire the unreferenced callback", 0, calls[0]); - // Releasing the only referrer drives segments 5..7 to zero -> callback fires (once per notification). - tracker.handleNotification( - new SSTableListChangedNotification(List.of(), List.of(sstable), OperationType.COMPACTION), - null); + tracker.handleNotification(new SSTableListChangedNotification(List.of(), + List.of(sstable), + OperationType.COMPACTION), + null); assertFalse(tracker.isReferenced(5)); assertEquals("releasing the last reference fires the unreferenced callback", 1, calls[0]); @@ -391,125 +440,264 @@ public class SegmentReferenceTrackerTest public void testCallbackNotFiredWhileSegmentStillReferenced() { int[] calls = { 0 }; - SegmentReferenceTracker tracker = newTracker(() -> calls[0]++); - SSTableReader a = unrepaired(intervals(5, 0, 5, 100)); - SSTableReader b = unrepaired(intervals(5, 0, 5, 200)); + Map<Long, Segment<ShortMutationId, Mutation>> segments = new HashMap<>(); + segments.put(5L, mockSegment(5L, LOG_ID, 10, 20)); + + SegmentReferenceTracker tracker = newTracker(segments, () -> calls[0]++); + SSTableReader a = unrepaired(offsets(LOG_ID, 10, 15), "a"); + SSTableReader b = unrepaired(offsets(LOG_ID, 16, 20), "b"); tracker.handleNotification(new SSTableAddedNotification(List.of(a, b), null), null); assertEquals(2L, tracker.referenceCountForTesting(5)); - // Releasing one of two referrers leaves segment 5 still referenced -> no callback. - tracker.handleNotification( - new SSTableListChangedNotification(List.of(), List.of(a), OperationType.COMPACTION), - null); + tracker.handleNotification(new SSTableListChangedNotification(List.of(), + List.of(a), + OperationType.COMPACTION), + null); + + assertTrue(tracker.isReferenced(5)); + assertEquals(0, calls[0]); + } + + @Test + public void testSSTableWithNoOverlappingSegmentsIsNotTracked() + { + Map<Long, Segment<ShortMutationId, Mutation>> segments = new HashMap<>(); + segments.put(5L, mockSegment(5L, LOG_ID, 10, 20)); + + SegmentReferenceTracker tracker = newTracker(segments); + SSTableReader sstable = unrepaired(offsets(LOG_ID, 100, 200)); + + tracker.handleNotification(new SSTableAddedNotification(List.of(sstable), null), null); + + assertEquals(0, tracker.trackedSstableCountForTesting()); + assertTrue(tracker.trackedSSTables().isEmpty()); + assertFalse(tracker.isReferenced(5)); + + tracker.handleNotification(new SSTableListChangedNotification(List.of(), List.of(sstable), OperationType.COMPACTION), + null); + assertEquals(0, tracker.trackedSstableCountForTesting()); + } + + @Test + public void testSegmentOverloadsAndNullHandling() + { + Map<Long, Segment<ShortMutationId, Mutation>> segments = new HashMap<>(); + Segment<ShortMutationId, Mutation> segment5 = mockSegment(5L, LOG_ID, 10, 20); + segments.put(5L, segment5); + + SegmentReferenceTracker tracker = newTracker(segments); + SSTableReader sstable = unrepaired(offsets(LOG_ID, 10, 20), "sstable5"); + + assertFalse(tracker.isReferenced(segment5)); + assertFalse(tracker.isReferenced((Segment<ShortMutationId, Mutation>) null)); + assertEquals(0, tracker.referenceCount(segment5)); + assertEquals(0, tracker.referenceCount((Segment<ShortMutationId, Mutation>) null)); + assertEquals(List.of(), tracker.referrerDescriptors(segment5)); + assertEquals(List.of(), tracker.referrerDescriptors((Segment<ShortMutationId, Mutation>) null)); + + tracker.handleNotification(new SSTableAddedNotification(List.of(sstable), null), null); + assertTrue(tracker.isReferenced(segment5)); + assertEquals(1, tracker.referenceCount(segment5)); + assertEquals(1L, tracker.referenceCountForTesting(segment5)); + assertEquals(List.of("sstable5"), tracker.referrerDescriptors(segment5)); + } + + @Test + public void testDefensivePruneOfEmptyReferrers() + { + Map<Long, Segment<ShortMutationId, Mutation>> segments = new HashMap<>(); + segments.put(5L, mockSegment(5L, LOG_ID, 10, 20)); + + SegmentReferenceTracker tracker = newTracker(segments); + SSTableReader sstable = unrepaired(offsets(LOG_ID, 10, 20), "sstable5"); + + tracker.addReferenceForTesting(5L, sstable); assertTrue(tracker.isReferenced(5)); - assertEquals("callback must not fire while the segment is still referenced", 0, calls[0]); + assertEquals(1, tracker.referenceCount(5)); + + tracker.evict(List.of(sstable)); + assertFalse(tracker.isReferenced(5)); + assertEquals(0, tracker.referenceCount(5)); + assertTrue(tracker.referrerDescriptors(5).isEmpty()); + } + + @Test + public void testTrackingActiveSegment() + { + MutationJournal.ActiveOffsetRanges activeRanges = new MutationJournal.ActiveOffsetRanges(); + activeRanges.update(new ShortMutationId(LOG_ID, 10)); + activeRanges.update(new ShortMutationId(LOG_ID, 50)); + + @SuppressWarnings("unchecked") + Segment<ShortMutationId, Mutation> activeSegment = Mockito.mock(Segment.class); + Mockito.when(activeSegment.id()).thenReturn(5L); + Mockito.when(activeSegment.keyStats()).thenReturn(activeRanges); + + Map<Long, Segment<ShortMutationId, Mutation>> segments = new HashMap<>(); + segments.put(5L, activeSegment); + + SegmentReferenceTracker tracker = newTracker(segments); + SSTableReader sstable = unrepaired(offsets(LOG_ID, 20, 30)); + + tracker.handleNotification(new SSTableAddedNotification(List.of(sstable), null), null); + + assertTrue(tracker.isReferenced(activeSegment)); + assertEquals(1, tracker.referenceCount(activeSegment)); + assertEquals(List.of(sstable.getFilename()), tracker.referrerDescriptors(activeSegment)); + + tracker.handleNotification(new SSTableListChangedNotification(List.of(), List.of(sstable), OperationType.COMPACTION), + null); + + assertFalse(tracker.isReferenced(activeSegment)); + assertEquals(0, tracker.referenceCount(activeSegment)); + } + + @Test + public void testSSTableWithMultipleLogsRefsAllOverlappingSegments() + { + long log1 = 100L; + long log2 = 200L; + + Map<Long, Segment<ShortMutationId, Mutation>> segments = new HashMap<>(); + segments.put(1L, mockSegment(1L, log1, 10, 20)); + segments.put(2L, mockSegment(2L, log2, 30, 40)); + + SegmentReferenceTracker tracker = newTracker(segments); + ImmutableCoordinatorLogOffsets offsets = new ImmutableCoordinatorLogOffsets.Builder() + .add(log1, 15, 25) + .add(log2, 35, 45) + .build(); + SSTableReader sstable = unrepaired(offsets, "sstable-multi-log"); + + tracker.handleNotification(new SSTableAddedNotification(List.of(sstable), null), null); + + assertTrue(tracker.isReferenced(1L)); + assertTrue(tracker.isReferenced(2L)); + assertEquals(1, tracker.referenceCount(1L)); + assertEquals(1, tracker.referenceCount(2L)); + assertEquals(1, tracker.trackedSstableCountForTesting()); + + tracker.handleNotification(new SSTableListChangedNotification(List.of(), List.of(sstable), OperationType.COMPACTION), + null); + + assertFalse(tracker.isReferenced(1L)); + assertFalse(tracker.isReferenced(2L)); + assertEquals(0, tracker.referenceCount(1L)); + assertEquals(0, tracker.referenceCount(2L)); + assertEquals(0, tracker.trackedSstableCountForTesting()); } // -- helpers --------------------------------------------------------- - private static SegmentReferenceTracker newTracker() + private static SegmentReferenceTracker newTracker(Map<Long, Segment<ShortMutationId, Mutation>> segments) + { + return newTracker(segments, () -> {}); + } + + private static SegmentReferenceTracker newTracker(Map<Long, Segment<ShortMutationId, Mutation>> segments, Runnable onSegmentsUnreferenced) + { + return new SegmentReferenceTracker(onSegmentsUnreferenced, () -> LOCAL_HOST, segments::values); + } + + @SuppressWarnings("unchecked") + private static Segment<ShortMutationId, Mutation> mockSegment(long segmentId, long logId, int minOffset, int maxOffset) { - return newTracker(() -> {}); + Segment<ShortMutationId, Mutation> segment = Mockito.mock(Segment.class); + Mockito.when(segment.id()).thenReturn(segmentId); + Mockito.when(segment.keyStats()).thenReturn(MutationJournal.StaticOffsetRanges.of(logId, minOffset, maxOffset)); + return segment; } - private static SegmentReferenceTracker newTracker(Runnable onSegmentsUnreferenced) + private static ImmutableCoordinatorLogOffsets offsets(long logId, int start, int end) { - // Fixed local host id so the mock sstables (originating from LOCAL_HOST) are treated as locally originated. - return new SegmentReferenceTracker(onSegmentsUnreferenced, () -> LOCAL_HOST); + return new ImmutableCoordinatorLogOffsets.Builder().add(logId, start, end).build(); } - private static IntervalSet<CommitLogPosition> intervals(long startSegment, int startPosition, long endSegment, int endPosition) + private static SSTableReader unrepaired(ImmutableCoordinatorLogOffsets offsets) { - assertTrue("startSegment " + startSegment + " is less than or equal to endSegment " + endSegment, startSegment <= endSegment); - assertTrue("startPosition " + startPosition + " is less than or equal to endPosition " + endPosition, startPosition <= endPosition); - return new IntervalSet<>(new CommitLogPosition(startSegment, startPosition), - new CommitLogPosition(endSegment, endPosition)); + return stub(offsets, false, "sstable"); } - private static SSTableReader unrepaired(IntervalSet<CommitLogPosition> intervals) + private static SSTableReader unrepaired(ImmutableCoordinatorLogOffsets offsets, String name) { - return stub(intervals, false); + return stub(offsets, false, name); } - private static SSTableReader repaired(IntervalSet<CommitLogPosition> intervals) + private static SSTableReader repaired(ImmutableCoordinatorLogOffsets offsets) { - return stub(intervals, true); + return stub(offsets, true, "sstable"); } - private static SSTableReader stub(IntervalSet<CommitLogPosition> intervals, boolean isRepaired) + private static SSTableReader stub(ImmutableCoordinatorLogOffsets offsets, boolean isRepaired, String name) { - return sstableWithRepairSupplier(intervals, () -> isRepaired); + return sstableWithRepairSupplier(offsets, () -> isRepaired, name); } - private static SSTableReader sstableWithRepairSupplier(IntervalSet<CommitLogPosition> intervals, + private static SSTableReader sstableWithRepairSupplier(ImmutableCoordinatorLogOffsets offsets, BooleanSupplier isRepairedSupplier) { - return sstable(intervals, isRepairedSupplier, coordinatorLogOffsets(true)); + return sstableWithRepairSupplier(offsets, isRepairedSupplier, "sstable"); } - private static SSTableReader sstable(IntervalSet<CommitLogPosition> intervals, - BooleanSupplier isRepairedSupplier, - ImmutableCoordinatorLogOffsets coordinatorLogOffsets) + private static SSTableReader sstableWithRepairSupplier(ImmutableCoordinatorLogOffsets offsets, + BooleanSupplier isRepairedSupplier, + String name) { - return sstable(intervals, isRepairedSupplier, coordinatorLogOffsets, LOCAL_HOST); + return sstable(offsets, isRepairedSupplier, LOCAL_HOST, ReplicationType.tracked, name); } - private static SSTableReader sstable(IntervalSet<CommitLogPosition> intervals, + private static SSTableReader sstable(ImmutableCoordinatorLogOffsets offsets, BooleanSupplier isRepairedSupplier, - ImmutableCoordinatorLogOffsets coordinatorLogOffsets, - UUID originatingHostId) + UUID originatingHostId, + ReplicationType replicationType) { - return sstable(intervals, isRepairedSupplier, coordinatorLogOffsets, originatingHostId, ReplicationType.tracked); + return sstable(offsets, isRepairedSupplier, originatingHostId, replicationType, "sstable"); } - private static SSTableReader sstable(IntervalSet<CommitLogPosition> intervals, + private static SSTableReader sstable(ImmutableCoordinatorLogOffsets offsets, BooleanSupplier isRepairedSupplier, - ImmutableCoordinatorLogOffsets coordinatorLogOffsets, UUID originatingHostId, - ReplicationType replicationType) + ReplicationType replicationType, + String name) { SSTableReader reader = Mockito.mock(SSTableReader.class); - Mockito.when(reader.isRepaired()).thenAnswer(ref -> isRepairedSupplier.getAsBoolean()); - Mockito.when(reader.getSSTableMetadata()).thenReturn(stats(intervals, originatingHostId)); - Mockito.when(reader.getCoordinatorLogOffsets()).thenReturn(coordinatorLogOffsets); + Mockito.when(reader.isRepaired()).thenAnswer(inv -> isRepairedSupplier.getAsBoolean()); + Mockito.when(reader.getSSTableMetadata()).thenReturn(stats(originatingHostId)); + Mockito.when(reader.getCoordinatorLogOffsets()).thenReturn(offsets); TableMetadata tableMetadata = Mockito.mock(TableMetadata.class); Mockito.when(tableMetadata.replicationType()).thenReturn(replicationType); Mockito.when(reader.metadata()).thenReturn(tableMetadata); - return reader; - } + Mockito.when(reader.getFilename()).thenReturn(name); - private static ImmutableCoordinatorLogOffsets coordinatorLogOffsets(boolean nonEmpty) - { - // A bare mock reports isEmpty()==false (Mockito's default boolean), which is all the tracker inspects. - return nonEmpty ? Mockito.mock(ImmutableCoordinatorLogOffsets.class) : ImmutableCoordinatorLogOffsets.NONE; + return reader; } - private static StatsMetadata stats(IntervalSet<CommitLogPosition> intervals, UUID originatingHostId) + private static StatsMetadata stats(UUID originatingHostId) { - return new StatsMetadata(new EstimatedHistogram(155), // estimatedPartitionSize - new EstimatedHistogram(118), // estimatedCellPerPartitionCount - intervals, // commitLogIntervals - 0L, // minTimestamp - 0L, // maxTimestamp - Cell.NO_DELETION_TIME, // minLocalDeletionTime - Cell.NO_DELETION_TIME, // maxLocalDeletionTime - Cell.NO_TTL, // minTTL - Cell.NO_TTL, // maxTTL - -1.0, // compressionRatio + return new StatsMetadata(new EstimatedHistogram(155), + new EstimatedHistogram(118), + IntervalSet.empty(), + 0L, + 0L, + Cell.NO_DELETION_TIME, + Cell.NO_DELETION_TIME, + Cell.NO_TTL, + Cell.NO_TTL, + -1.0, TombstoneHistogram.createDefault(), - 0, // sstableLevel - List.of(), // clusteringTypes - Slice.ALL, // coveredClustering - false, // hasLegacyCounterShards + 0, + List.of(), + Slice.ALL, + false, ActiveRepairService.UNREPAIRED_SSTABLE, - 0L, // totalColumnsSet - 0L, // totalRows - Double.NaN, // tokenSpaceCoverage + 0L, + 0L, + Double.NaN, originatingHostId, ActiveRepairService.NO_PENDING_REPAIR, - false, // hasPartitionLevelDeletions + false, ImmutableCoordinatorLogOffsets.NONE, ByteBufferUtil.EMPTY_BYTE_BUFFER, ByteBufferUtil.EMPTY_BYTE_BUFFER); --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
