diff --git a/CHANGES.txt b/CHANGES.txt index 05577860c0..0fcf0373a3 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -10,6 +10,7 @@ * Deprecate Pig support (CASSANDRA-10542) * Reduce contention getting instances of CompositeType (CASSANDRA-10433) Merged from 2.1: + * Invalidate cache after stream receive task is completed (CASSANDRA-10341) * Reject counter writes in CQLSSTableWriter (CASSANDRA-10258) * Remove superfluous COUNTER_MUTATION stage mapping (CASSANDRA-10605) * Improve json2sstable error reporting on nonexistent columns (CASSANDRA-10401) diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index d553f4db58..2d58219399 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -2519,6 +2519,41 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean CacheService.instance.invalidateCounterCacheForCf(metadata.ksAndCFName); } + public int invalidateRowCache(Collection> boundsToInvalidate) + { + int invalidatedKeys = 0; + for (Iterator keyIter = CacheService.instance.rowCache.keyIterator(); + keyIter.hasNext(); ) + { + RowCacheKey key = keyIter.next(); + DecoratedKey dk = partitioner.decorateKey(ByteBuffer.wrap(key.key)); + if (key.ksAndCFName.equals(metadata.ksAndCFName) && Bounds.isInBounds(dk.getToken(), boundsToInvalidate)) + { + invalidateCachedRow(dk); + invalidatedKeys++; + } + } + + return invalidatedKeys; + } + + public int invalidateCounterCache(Collection> boundsToInvalidate) + { + int invalidatedKeys = 0; + for (Iterator keyIter = CacheService.instance.counterCache.keyIterator(); + keyIter.hasNext(); ) + { + CounterCacheKey key = keyIter.next(); + DecoratedKey dk = partitioner.decorateKey(ByteBuffer.wrap(key.partitionKey)); + if (key.ksAndCFName.equals(metadata.ksAndCFName) && Bounds.isInBounds(dk.getToken(), boundsToInvalidate)) + { + CacheService.instance.counterCache.remove(key); + invalidatedKeys++; + } + } + return invalidatedKeys; + } + /** * @return true if @param key is contained in the row cache */ diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionController.java b/src/java/org/apache/cassandra/db/compaction/CompactionController.java index 5f0a198689..e0278c98df 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionController.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionController.java @@ -189,11 +189,6 @@ public class CompactionController implements AutoCloseable return min; } - public void invalidateCachedRow(DecoratedKey key) - { - cfs.invalidateCachedRow(key); - } - public void close() { overlappingSSTables.release(); diff --git a/src/java/org/apache/cassandra/dht/Bounds.java b/src/java/org/apache/cassandra/dht/Bounds.java index 9060bcfd59..73414cd2aa 100644 --- a/src/java/org/apache/cassandra/dht/Bounds.java +++ b/src/java/org/apache/cassandra/dht/Bounds.java @@ -17,8 +17,17 @@ */ package org.apache.cassandra.dht; +import java.util.ArrayList; +import java.util.Collection; import java.util.Collections; +import java.util.Comparator; import java.util.List; +import java.util.Set; + +import com.google.common.collect.Iterators; +import com.google.common.collect.Lists; +import com.google.common.collect.PeekingIterator; +import com.google.common.collect.Sets; import org.apache.cassandra.db.RowPosition; import org.apache.cassandra.utils.Pair; @@ -102,6 +111,20 @@ public class Bounds> extends AbstractBounds return "]"; } + public static > boolean isInBounds(T token, Iterable> bounds) + { + assert bounds != null; + + for (Bounds bound : bounds) + { + if (bound.contains(token)) + { + return true; + } + } + return false; + } + /** * Compute a bounds of keys corresponding to a given bounds of token. */ @@ -114,4 +137,43 @@ public class Bounds> extends AbstractBounds { return new Bounds(left, newRight); } + + /** + * Retrieves non-overlapping bounds for the list of input bounds + * + * Assume we have the following bounds + * (brackets representing left/right bound): + * [ ] [ ] [ ] [ ] + * [ ] [ ] + * This method will return the following bounds: + * [ ] [ ] + * + * @param bounds unsorted bounds to find overlaps + * @return the non-overlapping bounds + */ + public static > Set> getNonOverlappingBounds(Iterable> bounds) + { + ArrayList> sortedBounds = Lists.newArrayList(bounds); + Collections.sort(sortedBounds, new Comparator>() + { + public int compare(Bounds o1, Bounds o2) + { + return o1.left.compareTo(o2.left); + } + }); + + Set> nonOverlappingBounds = Sets.newHashSet(); + + PeekingIterator> it = Iterators.peekingIterator(sortedBounds.iterator()); + while (it.hasNext()) + { + Bounds beginBound = it.next(); + Bounds endBound = beginBound; + while (it.hasNext() && endBound.right.compareTo(it.peek().left) >= 0) + endBound = it.next(); + nonOverlappingBounds.add(new Bounds<>(beginBound.left, endBound.right)); + } + + return nonOverlappingBounds; + } } diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableRewriter.java b/src/java/org/apache/cassandra/io/sstable/SSTableRewriter.java index dc4fe75bd8..c62b827300 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableRewriter.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableRewriter.java @@ -20,7 +20,6 @@ package org.apache.cassandra.io.sstable; import java.util.*; import com.google.common.annotations.VisibleForTesting; -import com.google.common.util.concurrent.Runnables; import org.apache.cassandra.cache.InstrumentingCache; import org.apache.cassandra.cache.KeyCacheKey; diff --git a/src/java/org/apache/cassandra/streaming/StreamReader.java b/src/java/org/apache/cassandra/streaming/StreamReader.java index 1ccebb0901..fe3b13d5d0 100644 --- a/src/java/org/apache/cassandra/streaming/StreamReader.java +++ b/src/java/org/apache/cassandra/streaming/StreamReader.java @@ -171,6 +171,5 @@ public class StreamReader { DecoratedKey key = StorageService.getPartitioner().decorateKey(ByteBufferUtil.readWithShortLength(in)); writer.appendFromStream(key, cfs.metadata, in, inputVersion); - cfs.invalidateCachedRow(key); } } diff --git a/src/java/org/apache/cassandra/streaming/StreamReceiveTask.java b/src/java/org/apache/cassandra/streaming/StreamReceiveTask.java index d4d49b3913..846524b1e5 100644 --- a/src/java/org/apache/cassandra/streaming/StreamReceiveTask.java +++ b/src/java/org/apache/cassandra/streaming/StreamReceiveTask.java @@ -23,14 +23,20 @@ import java.io.IOException; import java.util.ArrayList; import java.util.Collection; import java.util.List; +import java.util.Set; import java.util.UUID; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import org.apache.cassandra.concurrent.NamedThreadFactory; import org.apache.cassandra.config.Schema; import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.db.Keyspace; +import org.apache.cassandra.dht.Bounds; +import org.apache.cassandra.dht.Token; import org.apache.cassandra.io.sstable.format.SSTableReader; import org.apache.cassandra.io.sstable.format.SSTableWriter; import org.apache.cassandra.utils.Pair; @@ -43,6 +49,7 @@ import org.apache.cassandra.utils.concurrent.Refs; public class StreamReceiveTask extends StreamTask { private static final ExecutorService executor = Executors.newCachedThreadPool(new NamedThreadFactory("StreamReceiveTask")); + private static final Logger logger = LoggerFactory.getLogger(StreamReceiveTask.class); // number of files to receive private final int totalFiles; @@ -76,6 +83,7 @@ public class StreamReceiveTask extends StreamTask assert cfId.equals(sstable.metadata.cfId); sstables.add(sstable); + if (sstables.size() == totalFiles) { done = true; @@ -131,6 +139,33 @@ public class StreamReceiveTask extends StreamTask // add sstables and build secondary indexes cfs.addSSTables(readers); cfs.indexManager.maybeBuildSecondaryIndexes(readers, cfs.indexManager.allIndexesNames()); + + //invalidate row and counter cache + if (cfs.isRowCacheEnabled() || cfs.metadata.isCounter()) + { + List> boundsToInvalidate = new ArrayList<>(readers.size()); + for (SSTableReader sstable : readers) + boundsToInvalidate.add(new Bounds(sstable.first.getToken(), sstable.last.getToken())); + Set> nonOverlappingBounds = Bounds.getNonOverlappingBounds(boundsToInvalidate); + + if (cfs.isRowCacheEnabled()) + { + int invalidatedKeys = cfs.invalidateRowCache(nonOverlappingBounds); + if (invalidatedKeys > 0) + logger.debug("[Stream #{}] Invalidated {} row cache entries on table {}.{} after stream " + + "receive task completed.", task.session.planId(), invalidatedKeys, + cfs.keyspace.getName(), cfs.getColumnFamilyName()); + } + + if (cfs.metadata.isCounter()) + { + int invalidatedKeys = cfs.invalidateCounterCache(nonOverlappingBounds); + if (invalidatedKeys > 0) + logger.debug("[Stream #{}] Invalidated {} counter cache entries on table {}.{} after stream " + + "receive task completed.", task.session.planId(), invalidatedKeys, + cfs.keyspace.getName(), cfs.getColumnFamilyName()); + } + } } task.session.taskCompleted(task); diff --git a/test/unit/org/apache/cassandra/db/CounterCacheTest.java b/test/unit/org/apache/cassandra/db/CounterCacheTest.java index 5b37b2c273..ed7921e567 100644 --- a/test/unit/org/apache/cassandra/db/CounterCacheTest.java +++ b/test/unit/org/apache/cassandra/db/CounterCacheTest.java @@ -17,6 +17,7 @@ */ package org.apache.cassandra.db; +import java.util.Collections; import java.util.concurrent.ExecutionException; import org.junit.AfterClass; @@ -26,6 +27,8 @@ import org.junit.Test; import org.apache.cassandra.SchemaLoader; import org.apache.cassandra.config.KSMetaData; import org.apache.cassandra.db.marshal.CounterColumnType; +import org.apache.cassandra.dht.Bounds; +import org.apache.cassandra.dht.Token; import org.apache.cassandra.exceptions.ConfigurationException; import org.apache.cassandra.config.Schema; import org.apache.cassandra.exceptions.WriteTimeoutException; @@ -85,6 +88,48 @@ public class CounterCacheTest assertEquals(ClockAndCount.create(2L, 2L), cfs.getCachedCounter(bytes(2), cellname(2))); } + @Test + public void testCounterCacheInvalidate() + { + ColumnFamilyStore cfs = Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF); + cfs.truncateBlocking(); + CacheService.instance.invalidateCounterCache(); + + assertEquals(0, CacheService.instance.counterCache.size()); + assertNull(cfs.getCachedCounter(bytes(1), cellname(1))); + assertNull(cfs.getCachedCounter(bytes(1), cellname(2))); + assertNull(cfs.getCachedCounter(bytes(2), cellname(1))); + assertNull(cfs.getCachedCounter(bytes(2), cellname(2))); + assertNull(cfs.getCachedCounter(bytes(3), cellname(1))); + assertNull(cfs.getCachedCounter(bytes(3), cellname(2))); + + cfs.putCachedCounter(bytes(1), cellname(1), ClockAndCount.create(1L, 1L)); + cfs.putCachedCounter(bytes(1), cellname(2), ClockAndCount.create(1L, 2L)); + cfs.putCachedCounter(bytes(2), cellname(1), ClockAndCount.create(2L, 1L)); + cfs.putCachedCounter(bytes(2), cellname(2), ClockAndCount.create(2L, 2L)); + cfs.putCachedCounter(bytes(3), cellname(1), ClockAndCount.create(3L, 1L)); + cfs.putCachedCounter(bytes(3), cellname(2), ClockAndCount.create(3L, 2L)); + + assertEquals(6, CacheService.instance.counterCache.size()); + assertEquals(ClockAndCount.create(1L, 1L), cfs.getCachedCounter(bytes(1), cellname(1))); + assertEquals(ClockAndCount.create(1L, 2L), cfs.getCachedCounter(bytes(1), cellname(2))); + assertEquals(ClockAndCount.create(2L, 1L), cfs.getCachedCounter(bytes(2), cellname(1))); + assertEquals(ClockAndCount.create(2L, 2L), cfs.getCachedCounter(bytes(2), cellname(2))); + assertEquals(ClockAndCount.create(3L, 1L), cfs.getCachedCounter(bytes(3), cellname(1))); + assertEquals(ClockAndCount.create(3L, 2L), cfs.getCachedCounter(bytes(3), cellname(2))); + + cfs.invalidateCounterCache(Collections.singleton(new Bounds(cfs.partitioner.decorateKey(bytes(1)).getToken(), + cfs.partitioner.decorateKey(bytes(2)).getToken()))); + + assertEquals(2, CacheService.instance.counterCache.size()); + assertNull(cfs.getCachedCounter(bytes(1), cellname(1))); + assertNull(cfs.getCachedCounter(bytes(1), cellname(2))); + assertNull(cfs.getCachedCounter(bytes(2), cellname(1))); + assertNull(cfs.getCachedCounter(bytes(2), cellname(2))); + assertEquals(ClockAndCount.create(3L, 1L), cfs.getCachedCounter(bytes(3), cellname(1))); + assertEquals(ClockAndCount.create(3L, 2L), cfs.getCachedCounter(bytes(3), cellname(2))); + } + @Test public void testSaveLoad() throws ExecutionException, InterruptedException, WriteTimeoutException { diff --git a/test/unit/org/apache/cassandra/db/RowCacheTest.java b/test/unit/org/apache/cassandra/db/RowCacheTest.java index 5912d7c3bd..9fb322b53a 100644 --- a/test/unit/org/apache/cassandra/db/RowCacheTest.java +++ b/test/unit/org/apache/cassandra/db/RowCacheTest.java @@ -20,8 +20,12 @@ package org.apache.cassandra.db; import java.net.InetAddress; import java.nio.ByteBuffer; +import java.util.ArrayList; import java.util.Collection; +import java.util.Iterator; +import java.util.TreeSet; +import com.google.common.collect.Lists; import org.junit.AfterClass; import org.junit.BeforeClass; import org.junit.Test; @@ -36,6 +40,8 @@ import org.apache.cassandra.db.composites.*; import org.apache.cassandra.db.compaction.CompactionManager; import org.apache.cassandra.db.filter.QueryFilter; import org.apache.cassandra.db.marshal.IntegerType; +import org.apache.cassandra.dht.Bounds; +import org.apache.cassandra.dht.Token; import org.apache.cassandra.exceptions.ConfigurationException; import org.apache.cassandra.dht.ByteOrderedPartitioner.BytesToken; import org.apache.cassandra.locator.TokenMetadata; @@ -171,6 +177,51 @@ public class RowCacheTest CacheService.instance.setRowCacheCapacityInMB(0); } + @Test + public void testInvalidateRowCache() throws Exception + { + StorageService.instance.initServer(0); + CacheService.instance.setRowCacheCapacityInMB(1); + rowCacheLoad(100, Integer.MAX_VALUE, 1000); + + ColumnFamilyStore store = Keyspace.open(KEYSPACE_CACHED).getColumnFamilyStore(CF_CACHED); + assertEquals(CacheService.instance.rowCache.size(), 100); + + //construct 5 ranges of 20 elements each + ArrayList> subranges = getBounds(20); + + //invalidate 3 of the 5 ranges + ArrayList> boundsToInvalidate = Lists.newArrayList(subranges.get(0), subranges.get(2), subranges.get(4)); + int invalidatedKeys = store.invalidateRowCache(boundsToInvalidate); + assertEquals(60, invalidatedKeys); + + //now there should be only 40 cached entries left + assertEquals(CacheService.instance.rowCache.size(), 40); + CacheService.instance.setRowCacheCapacityInMB(0); + } + + private ArrayList> getBounds(int nElements) + { + ColumnFamilyStore store = Keyspace.open(KEYSPACE_CACHED).getColumnFamilyStore(CF_CACHED); + TreeSet orderedKeys = new TreeSet<>(); + + for(Iterator it = CacheService.instance.rowCache.keyIterator();it.hasNext();) + orderedKeys.add(store.partitioner.decorateKey(ByteBuffer.wrap(it.next().key))); + + ArrayList> boundsToInvalidate = new ArrayList<>(); + Iterator iterator = orderedKeys.iterator(); + + while (iterator.hasNext()) + { + Token startRange = iterator.next().getToken(); + for (int i = 0; i < nElements-2; i++) + iterator.next(); + Token endRange = iterator.next().getToken(); + boundsToInvalidate.add(new Bounds<>(startRange, endRange)); + } + return boundsToInvalidate; + } + @Test public void testRowCachePartialLoad() throws Exception { diff --git a/test/unit/org/apache/cassandra/dht/BoundsTest.java b/test/unit/org/apache/cassandra/dht/BoundsTest.java new file mode 100644 index 0000000000..2ac06d9022 --- /dev/null +++ b/test/unit/org/apache/cassandra/dht/BoundsTest.java @@ -0,0 +1,61 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.dht; + +import java.util.LinkedList; +import java.util.List; +import java.util.Set; + +import org.junit.Test; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +public class BoundsTest +{ + + private Bounds bounds(long left, long right) + { + return new Bounds(new Murmur3Partitioner.LongToken(left), new Murmur3Partitioner.LongToken(right)); + } + + @Test + /** + * [0,1],[0,5],[1,8],[4,10] = [0, 10] + * [15,19][19,20] = [15,20] + * [21, 22] = [21,22] + */ + public void testGetNonOverlappingBounds() + { + List> bounds = new LinkedList<>(); + bounds.add(bounds(19, 20)); + bounds.add(bounds(0, 1)); + bounds.add(bounds(4, 10)); + bounds.add(bounds(15, 19)); + bounds.add(bounds(0, 5)); + bounds.add(bounds(21, 22)); + bounds.add(bounds(1, 8)); + + Set> nonOverlappingBounds = Bounds.getNonOverlappingBounds(bounds); + assertEquals(3, nonOverlappingBounds.size()); + assertTrue(nonOverlappingBounds.contains(bounds(0, 10))); + assertTrue(nonOverlappingBounds.contains(bounds(15,20))); + assertTrue(nonOverlappingBounds.contains(bounds(21,22))); + } +}