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

Merge branch 'cassandra-2.2' into cassandra-3.0

This commit is contained in:
Yuki Morishita 2015-11-11 16:16:23 -06:00
parent 9a90e9894e e487553575
commit 0de23f20ae
10 changed files with 298 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -5,6 +5,7 @@ Merged from 2.2:
* (Hadoop) fix splits calculation (CASSANDRA-10640)
* (Hadoop) ensure that Cluster instances are always closed (CASSANDRA-10058)
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)

34
src/java/org/apache/cassandra/db/ColumnFamilyStore.java
View File

@ -1739,6 +1739,40 @@ 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 = decorateKey(ByteBuffer.wrap(key.key));
if (key.ksAndCFName.equals(metadata.ksAndCFName) && Bounds.isInBounds(dk.getToken(), boundsToInvalidate))
{
invalidateCachedPartition(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 = 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
*/

5
src/java/org/apache/cassandra/db/compaction/CompactionController.java
View File

@ -198,11 +198,6 @@ public class CompactionController implements AutoCloseable
return min;
}
public void invalidateCachedPartition(DecoratedKey key)
{
cfs.invalidateCachedPartition(key);
}
public void close()
{
overlappingSSTables.release();

62
src/java/org/apache/cassandra/dht/Bounds.java
View File

@ -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.PartitionPosition;
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;
}
public boolean isStartInclusive()
{
return true;
@ -124,4 +147,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;
}
}

12
src/java/org/apache/cassandra/streaming/StreamReader.java
View File

@ -106,7 +106,7 @@ public class StreamReader
writer = createWriter(cfs, totalSize, repairedAt, format);
while (in.getBytesRead() < totalSize)
{
writePartition(deserializer, writer, cfs);
writePartition(deserializer, writer);
// TODO move this to BytesReadTracker
session.progress(desc, ProgressInfo.Direction.IN, in.getBytesRead(), totalSize);
}
@ -167,12 +167,10 @@ public class StreamReader
return size;
}
protected void writePartition(StreamDeserializer deserializer, SSTableMultiWriter writer, ColumnFamilyStore cfs) throws IOException
protected void writePartition(StreamDeserializer deserializer, SSTableMultiWriter writer) throws IOException
{
DecoratedKey key = deserializer.newPartition();
writer.append(deserializer);
writer.append(deserializer.newPartition());
deserializer.checkForExceptions();
cfs.invalidateCachedPartition(key);
}
public static class StreamDeserializer extends UnmodifiableIterator<Unfiltered> implements UnfilteredRowIterator
@ -197,13 +195,13 @@ public class StreamReader
this.header = header;
}
public DecoratedKey newPartition() throws IOException
public StreamDeserializer newPartition() throws IOException
{
key = metadata.decorateKey(ByteBufferUtil.readWithShortLength(in));
partitionLevelDeletion = DeletionTime.serializer.deserialize(in);
iterator = SSTableSimpleIterator.create(metadata, in, header, helper, partitionLevelDeletion);
staticRow = iterator.readStaticRow();
return key;
return this;
}
public CFMetaData metadata()

37
src/java/org/apache/cassandra/streaming/StreamReceiveTask.java
View File

@ -17,7 +17,14 @@
*/
package org.apache.cassandra.streaming;
import java.util.*;
import java.io.File;
import java.io.IOError;
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;
@ -35,6 +42,8 @@ import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
import org.apache.cassandra.db.view.View;
import org.apache.cassandra.dht.Bounds;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.sstable.ISSTableScanner;
import org.apache.cassandra.io.sstable.SSTableMultiWriter;
import org.apache.cassandra.io.sstable.format.SSTableReader;
@ -172,6 +181,32 @@ public class StreamReceiveTask extends StreamTask
// add sstables and build secondary indexes
cfs.addSSTables(readers);
cfs.indexManager.buildAllIndexesBlocking(readers);
//invalidate row and counter cache
if (cfs.isRowCacheEnabled() || cfs.metadata.isCounter())
{
List<Bounds<Token>> boundsToInvalidate = new ArrayList<>(readers.size());
readers.forEach(sstable -> 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.getTableName());
}
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.getTableName());
}
}
}
}
catch (Throwable t)

2
src/java/org/apache/cassandra/streaming/compress/CompressedStreamReader.java
View File

@ -93,7 +93,7 @@ public class CompressedStreamReader extends StreamReader
while (in.getBytesRead() < sectionLength)
{
writePartition(deserializer, writer, cfs);
writePartition(deserializer, writer);
// when compressed, report total bytes of compressed chunks read since remoteFile.size is the sum of chunks transferred
session.progress(desc, ProgressInfo.Direction.IN, cis.getTotalCompressedBytesRead(), totalSize);
}

48
test/unit/org/apache/cassandra/db/CounterCacheTest.java
View File

@ -17,10 +17,13 @@
*/
package org.apache.cassandra.db;
import java.util.Collections;
import java.util.concurrent.ExecutionException;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.ColumnDefinition;
import org.apache.cassandra.dht.Bounds;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.schema.KeyspaceParams;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.junit.AfterClass;
@ -94,6 +97,51 @@ public class CounterCacheTest
assertEquals(ClockAndCount.create(2L, 2L), cfs.getCachedCounter(bytes(2), c2, cd, null));
}
@Test
public void testCounterCacheInvalidate()
{
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE1).getColumnFamilyStore(COUNTER1);
cfs.truncateBlocking();
CacheService.instance.invalidateCounterCache();
Clustering c1 = CBuilder.create(cfs.metadata.comparator).add(ByteBufferUtil.bytes(1)).build();
Clustering c2 = CBuilder.create(cfs.metadata.comparator).add(ByteBufferUtil.bytes(2)).build();
ColumnDefinition cd = cfs.metadata.getColumnDefinition(ByteBufferUtil.bytes("c"));
assertEquals(0, CacheService.instance.counterCache.size());
assertNull(cfs.getCachedCounter(bytes(1), c1, cd, null));
assertNull(cfs.getCachedCounter(bytes(1), c2, cd, null));
assertNull(cfs.getCachedCounter(bytes(2), c1, cd, null));
assertNull(cfs.getCachedCounter(bytes(2), c2, cd, null));
assertNull(cfs.getCachedCounter(bytes(3), c1, cd, null));
assertNull(cfs.getCachedCounter(bytes(3), c2, cd, null));
cfs.putCachedCounter(bytes(1), c1, cd, null, ClockAndCount.create(1L, 1L));
cfs.putCachedCounter(bytes(1), c2, cd, null, ClockAndCount.create(1L, 2L));
cfs.putCachedCounter(bytes(2), c1, cd, null, ClockAndCount.create(2L, 1L));
cfs.putCachedCounter(bytes(2), c2, cd, null, ClockAndCount.create(2L, 2L));
cfs.putCachedCounter(bytes(3), c1, cd, null, ClockAndCount.create(3L, 1L));
cfs.putCachedCounter(bytes(3), c2, cd, null, ClockAndCount.create(3L, 2L));
assertEquals(ClockAndCount.create(1L, 1L), cfs.getCachedCounter(bytes(1), c1, cd, null));
assertEquals(ClockAndCount.create(1L, 2L), cfs.getCachedCounter(bytes(1), c2, cd, null));
assertEquals(ClockAndCount.create(2L, 1L), cfs.getCachedCounter(bytes(2), c1, cd, null));
assertEquals(ClockAndCount.create(2L, 2L), cfs.getCachedCounter(bytes(2), c2, cd, null));
assertEquals(ClockAndCount.create(3L, 1L), cfs.getCachedCounter(bytes(3), c1, cd, null));
assertEquals(ClockAndCount.create(3L, 2L), cfs.getCachedCounter(bytes(3), c2, cd, null));
cfs.invalidateCounterCache(Collections.singleton(new Bounds<Token>(cfs.decorateKey(bytes(1)).getToken(),
cfs.decorateKey(bytes(2)).getToken())));
assertEquals(2, CacheService.instance.counterCache.size());
assertNull(cfs.getCachedCounter(bytes(1), c1, cd, null));
assertNull(cfs.getCachedCounter(bytes(1), c2, cd, null));
assertNull(cfs.getCachedCounter(bytes(2), c1, cd, null));
assertNull(cfs.getCachedCounter(bytes(2), c2, cd, null));
assertEquals(ClockAndCount.create(3L, 1L), cfs.getCachedCounter(bytes(3), c1, cd, null));
assertEquals(ClockAndCount.create(3L, 2L), cfs.getCachedCounter(bytes(3), c2, cd, null));
}
@Test
public void testSaveLoad() throws ExecutionException, InterruptedException, WriteTimeoutException
{

50
test/unit/org/apache/cassandra/db/RowCacheTest.java
View File

@ -20,9 +20,12 @@ package org.apache.cassandra.db;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
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 +39,8 @@ import org.apache.cassandra.db.compaction.CompactionManager;
import org.apache.cassandra.db.filter.ColumnFilter;
import org.apache.cassandra.db.marshal.IntegerType;
import org.apache.cassandra.db.partitions.CachedPartition;
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;
@ -229,6 +234,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 bounds of 20 elements each
ArrayList<Bounds<Token>> subranges = getBounds(20);
//invalidate 3 of the 5 bounds
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.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
{

61
test/unit/org/apache/cassandra/dht/BoundsTest.java Normal file
View File

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