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

fix Summary component and caches to use correct partitioner 

patch by Sam Tunnicliffe and Pavel Yaskevich; reviewed by Pavel Yaskevich for CASSANDRA-4289
This commit is contained in:
Pavel Yaskevich 2012-06-01 18:10:21 +03:00
parent 770c201272
commit 267690a140
6 changed files with 41 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -12,6 +12,7 @@
* Set thrift HSHA server thread limit to unlimet by default (CASSANDRA-4277)
* Avoids double serialization of CF id in RowMutation messages
(CASSANDRA-4293)
* fix Summary component and caches to use correct partitioner (CASSANDRA-4289)
1.1.1-dev

5
src/java/org/apache/cassandra/io/sstable/IndexSummary.java
View File

@ -26,6 +26,7 @@ import java.util.List;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.ByteBufferUtil;
@ -110,7 +111,7 @@ public class IndexSummary
}
}
public IndexSummary deserialize(DataInput dis) throws IOException
public IndexSummary deserialize(DataInput dis, IPartitioner partitioner) throws IOException
{
IndexSummary summary = new IndexSummary();
if (dis.readInt() != DatabaseDescriptor.getIndexInterval())
@ -121,7 +122,7 @@ public class IndexSummary
{
long location = dis.readLong();
ByteBuffer key = ByteBufferUtil.readWithLength(dis);
summary.addEntry(StorageService.getPartitioner().decorateKey(key), location);
summary.addEntry(partitioner.decorateKey(key), location);
}
return summary;
}

6
src/java/org/apache/cassandra/io/sstable/SSTableReader.java
View File

@ -401,9 +401,9 @@ public class SSTableReader extends SSTable
try
{
iStream = new DataInputStream(new FileInputStream(summariesFile));
reader.indexSummary = IndexSummary.serializer.deserialize(iStream);
reader.first = decodeKey(StorageService.getPartitioner(), reader.descriptor, ByteBufferUtil.readWithLength(iStream));
reader.last = decodeKey(StorageService.getPartitioner(), reader.descriptor, ByteBufferUtil.readWithLength(iStream));
reader.indexSummary = IndexSummary.serializer.deserialize(iStream, reader.partitioner);
reader.first = decodeKey(reader.partitioner, reader.descriptor, ByteBufferUtil.readWithLength(iStream));
reader.last = decodeKey(reader.partitioner, reader.descriptor, ByteBufferUtil.readWithLength(iStream));
ibuilder.deserializeBounds(iStream);
dbuilder.deserializeBounds(iStream);
}

9
src/java/org/apache/cassandra/service/CacheService.java
View File

@ -309,7 +309,7 @@ public class CacheService implements CacheServiceMBean
public Pair<RowCacheKey, IRowCacheEntry> deserialize(DataInputStream in, ColumnFamilyStore store) throws IOException
{
ByteBuffer buffer = ByteBufferUtil.readWithLength(in);
DecoratedKey key = StorageService.getPartitioner().decorateKey(buffer);
DecoratedKey key = store.partitioner.decorateKey(buffer);
ColumnFamily data = store.getTopLevelColumns(QueryFilter.getIdentityFilter(key, new QueryPath(store.columnFamily)), Integer.MIN_VALUE, true);
return new Pair<RowCacheKey, IRowCacheEntry>(new RowCacheKey(store.metadata.cfId, key), data);
}
@ -319,7 +319,7 @@ public class CacheService implements CacheServiceMBean
{
for (ByteBuffer key : buffers)
{
DecoratedKey dk = StorageService.getPartitioner().decorateKey(key);
DecoratedKey dk = store.partitioner.decorateKey(key);
ColumnFamily data = store.getTopLevelColumns(QueryFilter.getIdentityFilter(dk, new QueryPath(store.columnFamily)), Integer.MIN_VALUE, true);
rowCache.put(new RowCacheKey(store.metadata.cfId, dk), data);
}
@ -356,7 +356,7 @@ public class CacheService implements CacheServiceMBean
if (input.readBoolean())
entry = RowIndexEntry.serializer.deserialize(input, reader.descriptor.version);
else
entry = reader.getPosition(StorageService.getPartitioner().decorateKey(key), Operator.EQ);
entry = reader.getPosition(reader.partitioner.decorateKey(key), Operator.EQ);
return new Pair<KeyCacheKey, RowIndexEntry>(new KeyCacheKey(reader.descriptor, key), entry);
}
@ -375,7 +375,8 @@ public class CacheService implements CacheServiceMBean
{
for (ByteBuffer key : buffers)
{
DecoratedKey dk = StorageService.getPartitioner().decorateKey(key);
DecoratedKey dk = store.partitioner.decorateKey(key);
for (SSTableReader sstable : store.getSSTables())
{
RowIndexEntry entry = sstable.getPosition(dk, Operator.EQ);

BIN
test/data/serialization/1.2/db.RowMutation.bin
View File

Binary file not shown.

29
test/unit/org/apache/cassandra/io/sstable/SSTableReaderTest.java
View File

@ -41,12 +41,15 @@ import org.apache.cassandra.db.columniterator.IdentityQueryFilter;
import org.apache.cassandra.db.compaction.CompactionManager;
import org.apache.cassandra.db.filter.QueryPath;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.LocalPartitioner;
import org.apache.cassandra.dht.LocalToken;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.service.CacheService;
import org.apache.cassandra.io.util.FileDataInput;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.io.util.MmappedSegmentedFile;
import org.apache.cassandra.io.util.SegmentedFile;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.thrift.IndexExpression;
import org.apache.cassandra.thrift.IndexOperator;
@ -284,6 +287,32 @@ public class SSTableReaderTest extends SchemaLoader
assert target.last.equals(lastKey);
}
@Test
public void testLoadingSummaryUsesCorrectPartitioner() throws Exception
{
Table table = Table.open("Keyspace1");
ColumnFamilyStore store = table.getColumnFamilyStore("Indexed1");
ByteBuffer key = ByteBufferUtil.bytes(String.valueOf("k1"));
RowMutation rm = new RowMutation("Keyspace1", key);
rm.add(new QueryPath("Indexed1", null, ByteBufferUtil.bytes("birthdate")), ByteBufferUtil.bytes(1L), System.currentTimeMillis());
rm.apply();
store.forceBlockingFlush();
ColumnFamilyStore indexCfs = store.indexManager.getIndexForColumn(ByteBufferUtil.bytes("birthdate")).getIndexCfs();
assert indexCfs.partitioner instanceof LocalPartitioner;
SSTableReader sstable = indexCfs.getSSTables().iterator().next();
assert sstable.first.token instanceof LocalToken;
SegmentedFile.Builder ibuilder = SegmentedFile.getBuilder(DatabaseDescriptor.getIndexAccessMode());
SegmentedFile.Builder dbuilder = sstable.compression
? SegmentedFile.getCompressedBuilder()
: SegmentedFile.getBuilder(DatabaseDescriptor.getDiskAccessMode());
SSTableReader.saveSummary(sstable, ibuilder, dbuilder);
SSTableReader reopened = SSTableReader.open(sstable.descriptor);
assert reopened.first.token instanceof LocalToken;
}
private void assertIndexQueryWorks(ColumnFamilyStore indexedCFS) throws IOException
{
assert "Indexed1".equals(indexedCFS.getColumnFamilyName());