Merge branch 'cassandra-2.1' into cassandra-2.2

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)

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<Bounds<Token>> boundsToInvalidate)
+    {
+        int invalidatedKeys = 0;
+        for (Iterator<RowCacheKey> 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<Bounds<Token>> boundsToInvalidate)
+    {
+        int invalidatedKeys = 0;
+        for (Iterator<CounterCacheKey> 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
      */

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();

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<T extends RingPosition<T>> extends AbstractBounds<T>
         return "]";
     }
 
+    public static <T extends RingPosition<T>> boolean isInBounds(T token, Iterable<Bounds<T>> bounds)
+    {
+        assert bounds != null;
+
+        for (Bounds<T> 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<T extends RingPosition<T>> extends AbstractBounds<T>
     {
         return new Bounds<T>(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 <T extends RingPosition<T>> Set<Bounds<T>> getNonOverlappingBounds(Iterable<Bounds<T>> bounds)
+    {
+        ArrayList<Bounds<T>> sortedBounds = Lists.newArrayList(bounds);
+        Collections.sort(sortedBounds, new Comparator<Bounds<T>>()
+        {
+            public int compare(Bounds<T> o1, Bounds<T> o2)
+            {
+                return o1.left.compareTo(o2.left);
+            }
+        });
+
+        Set<Bounds<T>> nonOverlappingBounds = Sets.newHashSet();
+
+        PeekingIterator<Bounds<T>> it = Iterators.peekingIterator(sortedBounds.iterator());
+        while (it.hasNext())
+        {
+            Bounds<T> beginBound = it.next();
+            Bounds<T> 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;
+    }
 }

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;

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);
     }
 }

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<Bounds<Token>> boundsToInvalidate = new ArrayList<>(readers.size());
+                    for (SSTableReader sstable : readers)
+                        boundsToInvalidate.add(new Bounds<Token>(sstable.first.getToken(), sstable.last.getToken()));
+                    Set<Bounds<Token>> 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);

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<Token>(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
     {

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<Bounds<Token>> subranges = getBounds(20);
+
+        //invalidate 3 of the 5 ranges
+        ArrayList<Bounds<Token>> 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<Bounds<Token>> getBounds(int nElements)
+    {
+        ColumnFamilyStore store = Keyspace.open(KEYSPACE_CACHED).getColumnFamilyStore(CF_CACHED);
+        TreeSet<DecoratedKey> orderedKeys = new TreeSet<>();
+
+        for(Iterator<RowCacheKey> it = CacheService.instance.rowCache.keyIterator();it.hasNext();)
+            orderedKeys.add(store.partitioner.decorateKey(ByteBuffer.wrap(it.next().key)));
+
+        ArrayList<Bounds<Token>> boundsToInvalidate = new ArrayList<>();
+        Iterator<DecoratedKey> 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
     {

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<Token> bounds(long left, long right)
+    {
+        return new Bounds<Token>(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<Token>> 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<Bounds<Token>> 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)));
+    }
+}