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.1' into cassandra-2.2

This commit is contained in:
Robert Stupp 2015-09-16 22:00:25 +02:00
parent 8b2dc1f258 e889ee408b
commit e63dacf793
26 changed files with 759 additions and 297 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -9,6 +9,7 @@
* Fall back to 1/4 commitlog volume for commitlog_total_space on small disks
(CASSANDRA-10199)
Merged from 2.1:
* Fix cache handling of 2i and base tables (CASSANDRA-10155)
* Fix NPE in nodetool compactionhistory (CASSANDRA-9758)
* (Pig) support BulkOutputFormat as a URL parameter (CASSANDRA-7410)
* BATCH statement is broken in cqlsh (CASSANDRA-10272)

193
src/java/org/apache/cassandra/cache/AutoSavingCache.java vendored
View File

@ -19,6 +19,8 @@ package org.apache.cassandra.cache;
import java.io.*;
import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
@ -27,6 +29,10 @@ import org.cliffc.high_scale_lib.NonBlockingHashSet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import org.apache.cassandra.concurrent.ScheduledExecutors;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.DatabaseDescriptor;
@ -62,7 +68,15 @@ public class AutoSavingCache<K extends CacheKey, V> extends InstrumentingCache<K
protected final CacheService.CacheType cacheType;
private final CacheSerializer<K, V> cacheLoader;
private static final String CURRENT_VERSION = "c";
/*
* CASSANDRA-10155 required a format change to fix 2i indexes and caching.
* 2.2 is already at version "c" and 3.0 is at "d".
*
* Since cache versions match exactly and there is no partial fallback just add
* a minor version letter.
*/
private static final String CURRENT_VERSION = "ca";
private static volatile IStreamFactory streamFactory = new IStreamFactory()
{
@ -90,16 +104,14 @@ public class AutoSavingCache<K extends CacheKey, V> extends InstrumentingCache<K
this.cacheLoader = cacheloader;
}
public File getCacheDataPath(UUID cfId, String version)
public File getCacheDataPath(String version)
{
Pair<String, String> names = Schema.instance.getCF(cfId);
return DatabaseDescriptor.getSerializedCachePath(names.left, names.right, cfId, cacheType, version, "db");
return DatabaseDescriptor.getSerializedCachePath( cacheType, version, "db");
}
public File getCacheCrcPath(UUID cfId, String version)
public File getCacheCrcPath(String version)
{
Pair<String, String> names = Schema.instance.getCF(cfId);
return DatabaseDescriptor.getSerializedCachePath(names.left, names.right, cfId, cacheType, version, "crc");
return DatabaseDescriptor.getSerializedCachePath( cacheType, version, "crc");
}
public Writer getWriter(int keysToSave)
@ -130,14 +142,43 @@ public class AutoSavingCache<K extends CacheKey, V> extends InstrumentingCache<K
}
}
public int loadSaved(ColumnFamilyStore cfs)
public ListenableFuture<Integer> loadSavedAsync()
{
final ListeningExecutorService es = MoreExecutors.listeningDecorator(Executors.newSingleThreadExecutor());
final long start = System.nanoTime();
ListenableFuture<Integer> cacheLoad = es.submit(new Callable<Integer>()
{
@Override
public Integer call() throws Exception
{
return loadSaved();
}
});
cacheLoad.addListener(new Runnable() {
@Override
public void run()
{
if (size() > 0)
logger.info("Completed loading ({} ms; {} keys) {} cache",
TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start),
CacheService.instance.keyCache.size(),
cacheType);
es.shutdown();
}
}, MoreExecutors.sameThreadExecutor());
return cacheLoad;
}
public int loadSaved()
{
int count = 0;
long start = System.nanoTime();
// modern format, allows both key and value (so key cache load can be purely sequential)
File dataPath = getCacheDataPath(cfs.metadata.cfId, CURRENT_VERSION);
File crcPath = getCacheCrcPath(cfs.metadata.cfId, CURRENT_VERSION);
File dataPath = getCacheDataPath(CURRENT_VERSION);
File crcPath = getCacheCrcPath(CURRENT_VERSION);
if (dataPath.exists() && crcPath.exists())
{
DataInputStream in = null;
@ -145,18 +186,46 @@ public class AutoSavingCache<K extends CacheKey, V> extends InstrumentingCache<K
{
logger.info(String.format("reading saved cache %s", dataPath));
in = new DataInputStream(new LengthAvailableInputStream(new BufferedInputStream(streamFactory.getInputStream(dataPath, crcPath)), dataPath.length()));
List<Future<Pair<K, V>>> futures = new ArrayList<Future<Pair<K, V>>>();
ArrayDeque<Future<Pair<K, V>>> futures = new ArrayDeque<Future<Pair<K, V>>>();
while (in.available() > 0)
{
Future<Pair<K, V>> entry = cacheLoader.deserialize(in, cfs);
//ksname and cfname are serialized by the serializers in CacheService
//That is delegated there because there are serializer specific conditions
//where a cache key is skipped and not written
String ksname = in.readUTF();
String cfname = in.readUTF();
ColumnFamilyStore cfs = Schema.instance.getColumnFamilyStoreIncludingIndexes(Pair.create(ksname, cfname));
Future<Pair<K, V>> entryFuture = cacheLoader.deserialize(in, cfs);
// Key cache entry can return null, if the SSTable doesn't exist.
if (entry == null)
if (entryFuture == null)
continue;
futures.add(entry);
futures.offer(entryFuture);
count++;
/*
* Kind of unwise to accrue an unbounded number of pending futures
* So now there is this loop to keep a bounded number pending.
*/
do
{
while (futures.peek() != null && futures.peek().isDone())
{
Future<Pair<K, V>> future = futures.poll();
Pair<K, V> entry = future.get();
if (entry != null && entry.right != null)
put(entry.left, entry.right);
}
if (futures.size() > 1000)
Thread.yield();
} while(futures.size() > 1000);
}
for (Future<Pair<K, V>> future : futures)
Future<Pair<K, V>> future = null;
while ((future = futures.poll()) != null)
{
Pair<K, V> entry = future.get();
if (entry != null && entry.right != null)
@ -168,10 +237,10 @@ public class AutoSavingCache<K extends CacheKey, V> extends InstrumentingCache<K
JVMStabilityInspector.inspectThrowable(e);
logger.warn(String.format("Non-fatal checksum error reading saved cache %s", dataPath.getAbsolutePath()), e);
}
catch (Exception e)
catch (Throwable t)
{
JVMStabilityInspector.inspectThrowable(e);
logger.debug(String.format("harmless error reading saved cache %s", dataPath.getAbsolutePath()), e);
JVMStabilityInspector.inspectThrowable(t);
logger.info(String.format("Harmless error reading saved cache %s", dataPath.getAbsolutePath()), t);
}
finally
{
@ -240,7 +309,6 @@ public class AutoSavingCache<K extends CacheKey, V> extends InstrumentingCache<K
return info.forProgress(keysWritten, Math.max(keysWritten, keysEstimate));
}
@SuppressWarnings("resource")
public void saveCache()
{
logger.debug("Deleting old {} files.", cacheType);
@ -254,45 +322,34 @@ public class AutoSavingCache<K extends CacheKey, V> extends InstrumentingCache<K
long start = System.nanoTime();
HashMap<UUID, DataOutputPlus> writers = new HashMap<>();
HashMap<UUID, OutputStream> streams = new HashMap<>();
HashMap<UUID, Pair<File, File>> paths = new HashMap<>();
WrappedDataOutputStreamPlus writer = null;
Pair<File, File> cacheFilePaths = tempCacheFiles();
try
{
try
{
writer = new WrappedDataOutputStreamPlus(streamFactory.getOutputStream(cacheFilePaths.left, cacheFilePaths.right));
}
catch (FileNotFoundException e)
{
throw new RuntimeException(e);
}
while (keyIterator.hasNext())
{
K key = keyIterator.next();
UUID cfId = key.getCFId();
if (!Schema.instance.hasCF(key.getCFId()))
continue; // the table has been dropped.
DataOutputPlus writer = writers.get(cfId);
if (writer == null)
{
Pair<File, File> cacheFilePaths = tempCacheFiles(cfId);
OutputStream stream;
try
{
stream = streamFactory.getOutputStream(cacheFilePaths.left, cacheFilePaths.right);
writer = new WrappedDataOutputStreamPlus(stream);
}
catch (FileNotFoundException e)
{
throw new RuntimeException(e);
}
paths.put(cfId, cacheFilePaths);
streams.put(cfId, stream);
writers.put(cfId, writer);
}
ColumnFamilyStore cfs = Schema.instance.getColumnFamilyStoreIncludingIndexes(key.ksAndCFName);
if (cfs == null)
continue; // the table or 2i has been dropped.
try
{
cacheLoader.serialize(key, writer);
cacheLoader.serialize(key, writer, cfs);
}
catch (IOException e)
{
throw new FSWriteError(e, paths.get(cfId).left);
throw new FSWriteError(e, cacheFilePaths.left);
}
keysWritten++;
@ -302,47 +359,29 @@ public class AutoSavingCache<K extends CacheKey, V> extends InstrumentingCache<K
}
finally
{
if (keyIterator instanceof Closeable)
try
{
((Closeable)keyIterator).close();
}
catch (IOException ignored)
{
// not thrown (by OHC)
}
for (OutputStream writer : streams.values())
{
if (writer != null)
FileUtils.closeQuietly(writer);
}
}
for (Map.Entry<UUID, DataOutputPlus> entry : writers.entrySet())
{
UUID cfId = entry.getKey();
File cacheFile = getCacheDataPath(CURRENT_VERSION);
File crcFile = getCacheCrcPath(CURRENT_VERSION);
Pair<File, File> tmpFiles = paths.get(cfId);
File cacheFile = getCacheDataPath(cfId, CURRENT_VERSION);
File crcFile = getCacheCrcPath(cfId, CURRENT_VERSION);
cacheFile.delete(); // ignore error if it didn't exist
crcFile.delete();
cacheFile.delete(); // ignore error if it didn't exist
crcFile.delete();
if (!cacheFilePaths.left.renameTo(cacheFile))
logger.error("Unable to rename {} to {}", cacheFilePaths.left, cacheFile);
if (!tmpFiles.left.renameTo(cacheFile))
logger.error("Unable to rename {} to {}", tmpFiles.left, cacheFile);
if (!tmpFiles.right.renameTo(crcFile))
logger.error("Unable to rename {} to {}", tmpFiles.right, crcFile);
}
if (!cacheFilePaths.right.renameTo(crcFile))
logger.error("Unable to rename {} to {}", cacheFilePaths.right, crcFile);
logger.info("Saved {} ({} items) in {} ms", cacheType, keysWritten, TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start));
}
private Pair<File, File> tempCacheFiles(UUID cfId)
private Pair<File, File> tempCacheFiles()
{
File dataPath = getCacheDataPath(cfId, CURRENT_VERSION);
File crcPath = getCacheCrcPath(cfId, CURRENT_VERSION);
File dataPath = getCacheDataPath(CURRENT_VERSION);
File crcPath = getCacheCrcPath(CURRENT_VERSION);
return Pair.create(FileUtils.createTempFile(dataPath.getName(), null, dataPath.getParentFile()),
FileUtils.createTempFile(crcPath.getName(), null, crcPath.getParentFile()));
}
@ -377,7 +416,7 @@ public class AutoSavingCache<K extends CacheKey, V> extends InstrumentingCache<K
public interface CacheSerializer<K extends CacheKey, V>
{
void serialize(K key, DataOutputPlus out) throws IOException;
void serialize(K key, DataOutputPlus out, ColumnFamilyStore cfs) throws IOException;
Future<Pair<K, V>> deserialize(DataInputStream in, ColumnFamilyStore cfs) throws IOException;
}

14
src/java/org/apache/cassandra/cache/CacheKey.java vendored
View File

@ -17,12 +17,14 @@
*/
package org.apache.cassandra.cache;
import java.util.UUID;
import org.apache.cassandra.utils.Pair;
public interface CacheKey extends IMeasurableMemory
public abstract class CacheKey implements IMeasurableMemory
{
/**
* @return The cf id of the cache key.
*/
public UUID getCFId();
public final Pair<String, String> ksAndCFName;
public CacheKey(Pair<String, String> ksAndCFName)
{
this.ksAndCFName = ksAndCFName;
}
}

26
src/java/org/apache/cassandra/cache/CounterCacheKey.java vendored
View File

@ -19,36 +19,28 @@ package org.apache.cassandra.cache;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.UUID;
import org.apache.cassandra.db.composites.CellName;
import org.apache.cassandra.db.composites.CellNames;
import org.apache.cassandra.utils.*;
public class CounterCacheKey implements CacheKey
public final class CounterCacheKey extends CacheKey
{
private static final long EMPTY_SIZE = ObjectSizes.measure(new CounterCacheKey(null, ByteBufferUtil.EMPTY_BYTE_BUFFER, CellNames.simpleDense(ByteBuffer.allocate(1))))
+ ObjectSizes.measure(new UUID(0, 0));
private static final long EMPTY_SIZE = ObjectSizes.measure(new CounterCacheKey(null, ByteBufferUtil.EMPTY_BYTE_BUFFER, CellNames.simpleDense(ByteBuffer.allocate(1))));
public final UUID cfId;
public final byte[] partitionKey;
public final byte[] cellName;
private CounterCacheKey(UUID cfId, ByteBuffer partitionKey, CellName cellName)
private CounterCacheKey(Pair<String, String> ksAndCFName, ByteBuffer partitionKey, CellName cellName)
{
this.cfId = cfId;
super(ksAndCFName);
this.partitionKey = ByteBufferUtil.getArray(partitionKey);
this.cellName = ByteBufferUtil.getArray(cellName.toByteBuffer());
}
public static CounterCacheKey create(UUID cfId, ByteBuffer partitionKey, CellName cellName)
public static CounterCacheKey create(Pair<String, String> ksAndCFName, ByteBuffer partitionKey, CellName cellName)
{
return new CounterCacheKey(cfId, partitionKey, cellName);
}
public UUID getCFId()
{
return cfId;
return new CounterCacheKey(ksAndCFName, partitionKey, cellName);
}
public long unsharedHeapSize()
@ -62,7 +54,7 @@ public class CounterCacheKey implements CacheKey
public String toString()
{
return String.format("CounterCacheKey(%s, %s, %s)",
cfId,
ksAndCFName,
ByteBufferUtil.bytesToHex(ByteBuffer.wrap(partitionKey)),
ByteBufferUtil.bytesToHex(ByteBuffer.wrap(cellName)));
}
@ -70,7 +62,7 @@ public class CounterCacheKey implements CacheKey
@Override
public int hashCode()
{
return Arrays.deepHashCode(new Object[]{cfId, partitionKey, cellName});
return Arrays.deepHashCode(new Object[]{ksAndCFName, partitionKey, cellName});
}
@Override
@ -84,7 +76,7 @@ public class CounterCacheKey implements CacheKey
CounterCacheKey cck = (CounterCacheKey) o;
return cfId.equals(cck.cfId)
return ksAndCFName.equals(cck.ksAndCFName)
&& Arrays.equals(partitionKey, cck.partitionKey)
&& Arrays.equals(cellName, cck.cellName);
}

19
src/java/org/apache/cassandra/cache/KeyCacheKey.java vendored
View File

@ -19,15 +19,14 @@ package org.apache.cassandra.cache;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.UUID;
import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.Pair;
public class KeyCacheKey implements CacheKey
public class KeyCacheKey extends CacheKey
{
public final UUID cfId;
public final Descriptor desc;
private static final long EMPTY_SIZE = ObjectSizes.measure(new KeyCacheKey(null, null, ByteBufferUtil.EMPTY_BYTE_BUFFER));
@ -36,19 +35,15 @@ public class KeyCacheKey implements CacheKey
// without extra copies on lookup since client-provided key ByteBuffers will be array-backed already
public final byte[] key;
public KeyCacheKey(UUID cfId, Descriptor desc, ByteBuffer key)
public KeyCacheKey(Pair<String, String> ksAndCFName, Descriptor desc, ByteBuffer key)
{
this.cfId = cfId;
super(ksAndCFName);
this.desc = desc;
this.key = ByteBufferUtil.getArray(key);
assert this.key != null;
}
public UUID getCFId()
{
return cfId;
}
public String toString()
{
return String.format("KeyCacheKey(%s, %s)", desc, ByteBufferUtil.bytesToHex(ByteBuffer.wrap(key)));
@ -67,13 +62,13 @@ public class KeyCacheKey implements CacheKey
KeyCacheKey that = (KeyCacheKey) o;
return cfId.equals(that.cfId) && desc.equals(that.desc) && Arrays.equals(key, that.key);
return ksAndCFName.equals(that.ksAndCFName) && desc.equals(that.desc) && Arrays.equals(key, that.key);
}
@Override
public int hashCode()
{
int result = cfId.hashCode();
int result = ksAndCFName.hashCode();
result = 31 * result + desc.hashCode();
result = 31 * result + Arrays.hashCode(key);
return result;

17
src/java/org/apache/cassandra/cache/OHCProvider.java vendored
View File

@ -23,7 +23,6 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.WritableByteChannel;
import java.util.Iterator;
import java.util.UUID;
import com.google.common.base.Function;
@ -33,6 +32,7 @@ import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.io.util.Memory;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.utils.Pair;
import org.caffinitas.ohc.OHCache;
import org.caffinitas.ohc.OHCacheBuilder;
@ -128,24 +128,27 @@ public class OHCProvider implements CacheProvider<RowCacheKey, IRowCacheEntry>
{
public void serialize(RowCacheKey rowCacheKey, DataOutput dataOutput) throws IOException
{
dataOutput.writeLong(rowCacheKey.cfId.getMostSignificantBits());
dataOutput.writeLong(rowCacheKey.cfId.getLeastSignificantBits());
dataOutput.writeUTF(rowCacheKey.ksAndCFName.left);
dataOutput.writeUTF(rowCacheKey.ksAndCFName.right);
dataOutput.writeInt(rowCacheKey.key.length);
dataOutput.write(rowCacheKey.key);
}
public RowCacheKey deserialize(DataInput dataInput) throws IOException
{
long msb = dataInput.readLong();
long lsb = dataInput.readLong();
String ksName = dataInput.readUTF();
String cfName = dataInput.readUTF();
byte[] key = new byte[dataInput.readInt()];
dataInput.readFully(key);
return new RowCacheKey(new UUID(msb, lsb), key);
return new RowCacheKey(Pair.create(ksName, cfName), key);
}
public int serializedSize(RowCacheKey rowCacheKey)
{
return 20 + rowCacheKey.key.length;
return TypeSizes.NATIVE.sizeof(rowCacheKey.ksAndCFName.left)
+ TypeSizes.NATIVE.sizeof(rowCacheKey.ksAndCFName.right)
+ 4
+ rowCacheKey.key.length;
}
}

34
src/java/org/apache/cassandra/cache/RowCacheKey.java vendored
View File

@ -19,43 +19,36 @@ package org.apache.cassandra.cache;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.UUID;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.Pair;
public class RowCacheKey implements CacheKey, Comparable<RowCacheKey>
public final class RowCacheKey extends CacheKey
{
public final UUID cfId;
public final byte[] key;
private static final long EMPTY_SIZE = ObjectSizes.measure(new RowCacheKey(null, ByteBufferUtil.EMPTY_BYTE_BUFFER));
public RowCacheKey(UUID cfId, byte[] key)
public RowCacheKey(Pair<String, String> ksAndCFName, byte[] key)
{
this.cfId = cfId;
super(ksAndCFName);
this.key = key;
}
public RowCacheKey(UUID cfId, DecoratedKey key)
public RowCacheKey(Pair<String, String> ksAndCFName, DecoratedKey key)
{
this(cfId, key.getKey());
this(ksAndCFName, key.getKey());
}
public RowCacheKey(UUID cfId, ByteBuffer key)
public RowCacheKey(Pair<String, String> ksAndCFName, ByteBuffer key)
{
this.cfId = cfId;
super(ksAndCFName);
this.key = ByteBufferUtil.getArray(key);
assert this.key != null;
}
public UUID getCFId()
{
return cfId;
}
public long unsharedHeapSize()
{
return EMPTY_SIZE + ObjectSizes.sizeOfArray(key);
@ -69,25 +62,20 @@ public class RowCacheKey implements CacheKey, Comparable<RowCacheKey>
RowCacheKey that = (RowCacheKey) o;
return cfId.equals(that.cfId) && Arrays.equals(key, that.key);
return ksAndCFName.equals(that.ksAndCFName) && Arrays.equals(key, that.key);
}
@Override
public int hashCode()
{
int result = cfId.hashCode();
int result = ksAndCFName.hashCode();
result = 31 * result + (key != null ? Arrays.hashCode(key) : 0);
return result;
}
public int compareTo(RowCacheKey otherKey)
{
return (cfId.compareTo(otherKey.cfId) < 0) ? -1 : ((cfId.equals(otherKey.cfId)) ? FBUtilities.compareUnsigned(key, otherKey.key, 0, 0, key.length, otherKey.key.length) : 1);
}
@Override
public String toString()
{
return String.format("RowCacheKey(cfId:%s, key:%s)", cfId, Arrays.toString(key));
return String.format("RowCacheKey(ksAndCFName:%s, key:%s)", ksAndCFName, Arrays.toString(key));
}
}

9
src/java/org/apache/cassandra/config/CFMetaData.java
View File

@ -52,6 +52,7 @@ import org.apache.cassandra.io.util.FileDataInput;
import org.apache.cassandra.schema.LegacySchemaTables;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.UUIDGen;
import org.github.jamm.Unmetered;
@ -167,6 +168,8 @@ public final class CFMetaData
public final UUID cfId; // internal id, never exposed to user
public final String ksName; // name of keyspace
public final String cfName; // name of this column family
public final Pair<String, String> ksAndCFName;
public final byte[] ksAndCFBytes;
public final ColumnFamilyType cfType; // standard, super
public volatile CellNameType comparator; // bytes, long, timeuuid, utf8, etc.
@ -258,6 +261,12 @@ public final class CFMetaData
cfId = id;
ksName = keyspace;
cfName = name;
ksAndCFName = Pair.create(keyspace, name);
byte[] ksBytes = FBUtilities.toWriteUTFBytes(ksName);
byte[] cfBytes = FBUtilities.toWriteUTFBytes(cfName);
ksAndCFBytes = Arrays.copyOf(ksBytes, ksBytes.length + cfBytes.length);
System.arraycopy(cfBytes, 0, ksAndCFBytes, ksBytes.length, cfBytes.length);
cfType = type;
comparator = comp;
}

17
src/java/org/apache/cassandra/config/DatabaseDescriptor.java
View File

@ -1480,20 +1480,11 @@ public class DatabaseDescriptor
return conf.max_hint_window_in_ms;
}
public static File getSerializedCachePath(String ksName,
String cfName,
UUID cfId,
CacheService.CacheType cacheType,
String version,
String extension)
public static File getSerializedCachePath(CacheService.CacheType cacheType, String version, String extension)
{
StringBuilder builder = new StringBuilder();
builder.append(ksName).append('-');
builder.append(cfName).append('-');
builder.append(ByteBufferUtil.bytesToHex(ByteBufferUtil.bytes(cfId))).append('-');
builder.append(cacheType);
builder.append((version == null ? "" : "-" + version + "." + extension));
return new File(conf.saved_caches_directory, builder.toString());
String name = cacheType.toString()
+ (version == null ? "" : "-" + version + "." + extension);
return new File(conf.saved_caches_directory, name);
}
public static int getDynamicUpdateInterval()

56
src/java/org/apache/cassandra/config/Schema.java
View File

@ -34,6 +34,7 @@ import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.commitlog.CommitLog;
import org.apache.cassandra.db.compaction.CompactionManager;
import org.apache.cassandra.db.marshal.UserType;
import org.apache.cassandra.db.index.SecondaryIndex;
import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.schema.LegacySchemaTables;
import org.apache.cassandra.service.MigrationManager;
@ -155,6 +156,53 @@ public class Schema
return keyspaceInstances.get(keyspaceName);
}
/**
* Retrieve a CFS by name even if that CFS is an index
*
* An index is identified by looking for '.' in the CF name and separating to find the base table
* containing the index
* @param ksNameAndCFName
* @return The named CFS or null if the keyspace, base table, or index don't exist
*/
public ColumnFamilyStore getColumnFamilyStoreIncludingIndexes(Pair<String, String> ksNameAndCFName) {
String ksName = ksNameAndCFName.left;
String cfName = ksNameAndCFName.right;
Pair<String, String> baseTable;
/*
* Split does special case a one character regex, and it looks like it can detect
* if you use two characters to escape '.', but it still allocates a useless array.
*/
int indexOfSeparator = cfName.indexOf('.');
if (indexOfSeparator > -1)
baseTable = Pair.create(ksName, cfName.substring(0, indexOfSeparator));
else
baseTable = ksNameAndCFName;
UUID cfId = cfIdMap.get(baseTable);
if (cfId == null)
return null;
Keyspace ks = keyspaceInstances.get(ksName);
if (ks == null)
return null;
ColumnFamilyStore baseCFS = ks.getColumnFamilyStore(cfId);
//Not an index
if (indexOfSeparator == -1)
return baseCFS;
if (baseCFS == null)
return null;
SecondaryIndex index = baseCFS.indexManager.getIndexByName(cfName);
if (index == null)
return null;
return index.getIndexCfs();
}
public ColumnFamilyStore getColumnFamilyStoreInstance(UUID cfId)
{
Pair<String, String> pair = cfIdMap.inverse().get(cfId);
@ -313,12 +361,12 @@ public class Schema
}
/**
* @param cfId The identifier of the ColumnFamily to lookup
* @return true if the CF id is a known one, false otherwise.
* @param ksAndCFName The identifier of the ColumnFamily to lookup
* @return true if the KS and CF pair is a known one, false otherwise.
*/
public boolean hasCF(UUID cfId)
public boolean hasCF(Pair<String, String> ksAndCFName)
{
return cfIdMap.containsValue(cfId);
return cfIdMap.containsKey(ksAndCFName);
}
/**

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

@ -379,8 +379,6 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
fileIndexGenerator.set(generation);
sampleLatencyNanos = DatabaseDescriptor.getReadRpcTimeout() / 2;
CachingOptions caching = metadata.getCaching();
logger.info("Initializing {}.{}", keyspace.getName(), name);
// scan for sstables corresponding to this cf and load them
@ -393,9 +391,6 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
data.addInitialSSTables(sstables);
}
if (caching.keyCache.isEnabled())
CacheService.instance.keyCache.loadSaved(this);
// compaction strategy should be created after the CFS has been prepared
this.compactionStrategyWrapper = new WrappingCompactionStrategy(this);
@ -644,7 +639,6 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
public static void removeUnfinishedCompactionLeftovers(CFMetaData metadata, Map<Integer, UUID> unfinishedCompactions)
{
Directories directories = new Directories(metadata);
Set<Integer> allGenerations = new HashSet<>();
for (Descriptor desc : directories.sstableLister().list().keySet())
allGenerations.add(desc.generation);
@ -714,39 +708,6 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
}
}
// must be called after all sstables are loaded since row cache merges all row versions
public void initRowCache()
{
if (!isRowCacheEnabled())
return;
long start = System.nanoTime();
int cachedRowsRead = CacheService.instance.rowCache.loadSaved(this);
if (cachedRowsRead > 0)
logger.info("Completed loading ({} ms; {} keys) row cache for {}.{}",
TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start),
cachedRowsRead,
keyspace.getName(),
name);
}
public void initCounterCache()
{
if (!metadata.isCounter() || CacheService.instance.counterCache.getCapacity() == 0)
return;
long start = System.nanoTime();
int cachedShardsRead = CacheService.instance.counterCache.loadSaved(this);
if (cachedShardsRead > 0)
logger.info("Completed loading ({} ms; {} shards) counter cache for {}.{}",
TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start),
cachedShardsRead,
keyspace.getName(),
name);
}
/**
* See #{@code StorageService.loadNewSSTables(String, String)} for more info
*
@ -1265,7 +1226,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
if (!isRowCacheEnabled())
return;
RowCacheKey cacheKey = new RowCacheKey(metadata.cfId, key);
RowCacheKey cacheKey = new RowCacheKey(metadata.ksAndCFName, key);
invalidateCachedRow(cacheKey);
}
@ -1667,7 +1628,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
assert isRowCacheEnabled()
: String.format("Row cache is not enabled on table [" + name + "]");
RowCacheKey key = new RowCacheKey(cfId, filter.key);
RowCacheKey key = new RowCacheKey(metadata.ksAndCFName, filter.key);
// attempt a sentinel-read-cache sequence. if a write invalidates our sentinel, we'll return our
// (now potentially obsolete) data, but won't cache it. see CASSANDRA-3862
@ -2080,7 +2041,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
{
RowCacheKey key = keyIter.next();
DecoratedKey dk = partitioner.decorateKey(ByteBuffer.wrap(key.key));
if (key.cfId.equals(metadata.cfId) && !Range.isInRanges(dk.getToken(), ranges))
if (key.ksAndCFName.equals(metadata.ksAndCFName) && !Range.isInRanges(dk.getToken(), ranges))
invalidateCachedRow(dk);
}
@ -2091,7 +2052,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
{
CounterCacheKey key = keyIter.next();
DecoratedKey dk = partitioner.decorateKey(ByteBuffer.wrap(key.partitionKey));
if (key.cfId.equals(metadata.cfId) && !Range.isInRanges(dk.getToken(), ranges))
if (key.ksAndCFName.equals(metadata.ksAndCFName) && !Range.isInRanges(dk.getToken(), ranges))
CacheService.instance.counterCache.remove(key);
}
}
@ -2541,16 +2502,16 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
if (!isRowCacheEnabled())
return null;
IRowCacheEntry cached = CacheService.instance.rowCache.getInternal(new RowCacheKey(metadata.cfId, key));
IRowCacheEntry cached = CacheService.instance.rowCache.getInternal(new RowCacheKey(metadata.ksAndCFName, key));
return cached == null || cached instanceof RowCacheSentinel ? null : (ColumnFamily)cached;
}
private void invalidateCaches()
{
CacheService.instance.invalidateKeyCacheForCf(metadata.cfId);
CacheService.instance.invalidateRowCacheForCf(metadata.cfId);
CacheService.instance.invalidateKeyCacheForCf(metadata.ksAndCFName);
CacheService.instance.invalidateRowCacheForCf(metadata.ksAndCFName);
if (metadata.isCounter())
CacheService.instance.invalidateCounterCacheForCf(metadata.cfId);
CacheService.instance.invalidateCounterCacheForCf(metadata.ksAndCFName);
}
/**
@ -2558,7 +2519,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
*/
public boolean containsCachedRow(DecoratedKey key)
{
return CacheService.instance.rowCache.getCapacity() != 0 && CacheService.instance.rowCache.containsKey(new RowCacheKey(metadata.cfId, key));
return CacheService.instance.rowCache.getCapacity() != 0 && CacheService.instance.rowCache.containsKey(new RowCacheKey(metadata.ksAndCFName, key));
}
public void invalidateCachedRow(RowCacheKey key)
@ -2572,21 +2533,21 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
if (cfId == null)
return; // secondary index
invalidateCachedRow(new RowCacheKey(cfId, key));
invalidateCachedRow(new RowCacheKey(metadata.ksAndCFName, key));
}
public ClockAndCount getCachedCounter(ByteBuffer partitionKey, CellName cellName)
{
if (CacheService.instance.counterCache.getCapacity() == 0L) // counter cache disabled.
return null;
return CacheService.instance.counterCache.get(CounterCacheKey.create(metadata.cfId, partitionKey, cellName));
return CacheService.instance.counterCache.get(CounterCacheKey.create(metadata.ksAndCFName, partitionKey, cellName));
}
public void putCachedCounter(ByteBuffer partitionKey, CellName cellName, ClockAndCount clockAndCount)
{
if (CacheService.instance.counterCache.getCapacity() == 0L) // counter cache disabled.
return;
CacheService.instance.counterCache.put(CounterCacheKey.create(metadata.cfId, partitionKey, cellName), clockAndCount);
CacheService.instance.counterCache.put(CounterCacheKey.create(metadata.ksAndCFName, partitionKey, cellName), clockAndCount);
}
public void forceMajorCompaction() throws InterruptedException, ExecutionException
@ -2971,11 +2932,21 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
return view.sstables.isEmpty() && view.getCurrentMemtable().getOperations() == 0 && view.liveMemtables.size() <= 1 && view.flushingMemtables.size() == 0;
}
private boolean isRowCacheEnabled()
public boolean isRowCacheEnabled()
{
return metadata.getCaching().rowCache.isEnabled() && CacheService.instance.rowCache.getCapacity() > 0;
}
public boolean isCounterCacheEnabled()
{
return metadata.isCounter() && CacheService.instance.counterCache.getCapacity() > 0;
}
public boolean isKeyCacheEnabled()
{
return metadata.getCaching().keyCache.isEnabled() && CacheService.instance.keyCache.getCapacity() > 0;
}
/**
* Discard all SSTables that were created before given timestamp.
*

4
src/java/org/apache/cassandra/db/Keyspace.java
View File

@ -115,10 +115,6 @@ public class Keyspace
// open and store the keyspace
keyspaceInstance = new Keyspace(keyspaceName, loadSSTables);
schema.storeKeyspaceInstance(keyspaceInstance);
// keyspace has to be constructed and in the cache before cacheRow can be called
for (ColumnFamilyStore cfs : keyspaceInstance.getColumnFamilyStores())
cfs.initRowCache();
}
}
}

2
src/java/org/apache/cassandra/db/RowIndexEntry.java
View File

@ -155,7 +155,7 @@ public class RowIndexEntry<T> implements IMeasurableMemory
skipPromotedIndex(in);
}
public static void skipPromotedIndex(DataInput in) throws IOException
private static void skipPromotedIndex(DataInput in) throws IOException
{
int size = in.readInt();
if (size <= 0)

30
src/java/org/apache/cassandra/db/index/SecondaryIndexManager.java
View File

@ -94,7 +94,8 @@ public class SecondaryIndexManager
/**
* Keeps all secondary index instances, either per-column or per-row
*/
private final Set<SecondaryIndex> allIndexes;
private final Collection<SecondaryIndex> allIndexes;
private final Map<String, SecondaryIndex> indexesByName;
/**
@ -106,7 +107,8 @@ public class SecondaryIndexManager
{
indexesByColumn = new ConcurrentSkipListMap<>();
rowLevelIndexMap = new ConcurrentHashMap<>();
allIndexes = Collections.newSetFromMap(new ConcurrentHashMap<SecondaryIndex, Boolean>());
indexesByName = new ConcurrentHashMap<String, SecondaryIndex>();
allIndexes = indexesByName.values();
this.baseCfs = baseCfs;
}
@ -157,7 +159,7 @@ public class SecondaryIndexManager
{
idxNames = filterByColumn(idxNames);
if (idxNames.isEmpty())
return;
return;
logger.info(String.format("Submitting index build of %s for data in %s",
idxNames, StringUtils.join(sstables, ", ")));
@ -171,7 +173,7 @@ public class SecondaryIndexManager
logger.info("Index build of {} complete", idxNames);
}
public boolean indexes(CellName name, Set<SecondaryIndex> indexes)
public boolean indexes(CellName name, Collection<SecondaryIndex> indexes)
{
boolean matching = false;
for (SecondaryIndex index : indexes)
@ -185,7 +187,7 @@ public class SecondaryIndexManager
return matching;
}
public Set<SecondaryIndex> indexFor(CellName name, Set<SecondaryIndex> indexes)
public Set<SecondaryIndex> indexFor(CellName name, Collection<SecondaryIndex> indexes)
{
Set<SecondaryIndex> matching = null;
for (SecondaryIndex index : indexes)
@ -310,7 +312,7 @@ public class SecondaryIndexManager
indexesByColumn.put(cdef.name.bytes, index);
// Add to all indexes set:
allIndexes.add(index);
indexesByName.put(index.getIndexName(), index);
// if we're just linking in the index to indexedColumns on an
// already-built index post-restart, we're done
@ -413,11 +415,16 @@ public class SecondaryIndexManager
/**
* @return all of the secondary indexes without distinction to the (non-)backed by secondary ColumnFamilyStore.
*/
public Set<SecondaryIndex> getIndexes()
public Collection<SecondaryIndex> getIndexes()
{
return allIndexes;
}
public SecondaryIndex getIndexByName(String name)
{
return indexesByName.get(name);
}
/**
* @return if there are ANY indexes for this table..
*/
@ -645,15 +652,6 @@ public class SecondaryIndexManager
return result;
}
public SecondaryIndex getIndexByName(String idxName)
{
for (SecondaryIndex index : allIndexes)
if (idxName.equals(index.getIndexName()))
return index;
return null;
}
public void setIndexBuilt(Set<String> idxNames)
{
for (SecondaryIndex index : getIndexesByNames(idxNames))

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

@ -75,8 +75,8 @@ import static org.apache.cassandra.db.Directories.SECONDARY_INDEX_NAME_SEPARATOR
* to replace some existing sstables. However once created, an sstablereader may also be modified.
*
* A reader's OpenReason describes its current stage in its lifecycle, as follows:
*
*
*
*
* <pre> {@code
* NORMAL
* From: None => Reader has been read from disk, either at startup or from a flushed memtable
@ -1486,7 +1486,7 @@ public abstract class SSTableReader extends SSTable implements SelfRefCounted<SS
public KeyCacheKey getCacheKey(DecoratedKey key)
{
return new KeyCacheKey(metadata.cfId, descriptor, key.getKey());
return new KeyCacheKey(metadata.ksAndCFName, descriptor, key.getKey());
}
public void cacheKey(DecoratedKey key, RowIndexEntry info)
@ -1500,14 +1500,14 @@ public abstract class SSTableReader extends SSTable implements SelfRefCounted<SS
return;
}
KeyCacheKey cacheKey = new KeyCacheKey(metadata.cfId, descriptor, key.getKey());
KeyCacheKey cacheKey = new KeyCacheKey(metadata.ksAndCFName, descriptor, key.getKey());
logger.trace("Adding cache entry for {} -> {}", cacheKey, info);
keyCache.put(cacheKey, info);
}
public RowIndexEntry getCachedPosition(DecoratedKey key, boolean updateStats)
{
return getCachedPosition(new KeyCacheKey(metadata.cfId, descriptor, key.getKey()), updateStats);
return getCachedPosition(new KeyCacheKey(metadata.ksAndCFName, descriptor, key.getKey()), updateStats);
}
protected RowIndexEntry getCachedPosition(KeyCacheKey unifiedKey, boolean updateStats)

2
src/java/org/apache/cassandra/io/sstable/format/big/BigTableReader.java
View File

@ -125,7 +125,7 @@ public class BigTableReader extends SSTableReader
if ((op == Operator.EQ || op == Operator.GE) && (key instanceof DecoratedKey))
{
DecoratedKey decoratedKey = (DecoratedKey)key;
KeyCacheKey cacheKey = new KeyCacheKey(metadata.cfId, descriptor, decoratedKey.getKey());
KeyCacheKey cacheKey = new KeyCacheKey(metadata.ksAndCFName, descriptor, decoratedKey.getKey());
RowIndexEntry cachedPosition = getCachedPosition(cacheKey, updateCacheAndStats);
if (cachedPosition != null)
{

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

@ -25,7 +25,6 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
@ -267,13 +266,13 @@ public class CacheService implements CacheServiceMBean
keyCache.clear();
}
public void invalidateKeyCacheForCf(UUID cfId)
public void invalidateKeyCacheForCf(Pair<String, String> ksAndCFName)
{
Iterator<KeyCacheKey> keyCacheIterator = keyCache.keyIterator();
while (keyCacheIterator.hasNext())
{
KeyCacheKey key = keyCacheIterator.next();
if (key.cfId.equals(cfId))
if (key.ksAndCFName.equals(ksAndCFName))
keyCacheIterator.remove();
}
}
@ -283,24 +282,24 @@ public class CacheService implements CacheServiceMBean
rowCache.clear();
}
public void invalidateRowCacheForCf(UUID cfId)
public void invalidateRowCacheForCf(Pair<String, String> ksAndCFName)
{
Iterator<RowCacheKey> rowCacheIterator = rowCache.keyIterator();
while (rowCacheIterator.hasNext())
{
RowCacheKey rowCacheKey = rowCacheIterator.next();
if (rowCacheKey.cfId.equals(cfId))
if (rowCacheKey.ksAndCFName.equals(ksAndCFName))
rowCacheIterator.remove();
}
}
public void invalidateCounterCacheForCf(UUID cfId)
public void invalidateCounterCacheForCf(Pair<String, String> ksAndCFName)
{
Iterator<CounterCacheKey> counterCacheIterator = counterCache.keyIterator();
while (counterCacheIterator.hasNext())
{
CounterCacheKey counterCacheKey = counterCacheIterator.next();
if (counterCacheKey.cfId.equals(cfId))
if (counterCacheKey.ksAndCFName.equals(ksAndCFName))
counterCacheIterator.remove();
}
}
@ -353,16 +352,24 @@ public class CacheService implements CacheServiceMBean
public static class CounterCacheSerializer implements CacheSerializer<CounterCacheKey, ClockAndCount>
{
public void serialize(CounterCacheKey key, DataOutputPlus out) throws IOException
public void serialize(CounterCacheKey key, DataOutputPlus out, ColumnFamilyStore cfs) throws IOException
{
assert(cfs.metadata.isCounter());
out.write(cfs.metadata.ksAndCFBytes);
ByteBufferUtil.writeWithLength(key.partitionKey, out);
ByteBufferUtil.writeWithLength(key.cellName, out);
}
public Future<Pair<CounterCacheKey, ClockAndCount>> deserialize(DataInputStream in, final ColumnFamilyStore cfs) throws IOException
{
//Keyspace and CF name are deserialized by AutoSaving cache and used to fetch the CFS provided as a
//parameter so they aren't deserialized here, even though they are serialized by this serializer
final ByteBuffer partitionKey = ByteBufferUtil.readWithLength(in);
final CellName cellName = cfs.metadata.comparator.cellFromByteBuffer(ByteBufferUtil.readWithLength(in));
ByteBuffer cellNameBuffer = ByteBufferUtil.readWithLength(in);
if (cfs == null || !cfs.metadata.isCounter() || !cfs.isCounterCacheEnabled())
return null;
assert(cfs.metadata.isCounter());
final CellName cellName = cfs.metadata.comparator.cellFromByteBuffer(cellNameBuffer);
return StageManager.getStage(Stage.READ).submit(new Callable<Pair<CounterCacheKey, ClockAndCount>>()
{
public Pair<CounterCacheKey, ClockAndCount> call() throws Exception
@ -379,7 +386,7 @@ public class CacheService implements CacheServiceMBean
if (cell == null || !cell.isLive(Long.MIN_VALUE))
return null;
ClockAndCount clockAndCount = CounterContext.instance().getLocalClockAndCount(cell.value());
return Pair.create(CounterCacheKey.create(cfs.metadata.cfId, partitionKey, cellName), clockAndCount);
return Pair.create(CounterCacheKey.create(cfs.metadata.ksAndCFName, partitionKey, cellName), clockAndCount);
}
});
}
@ -387,14 +394,22 @@ public class CacheService implements CacheServiceMBean
public static class RowCacheSerializer implements CacheSerializer<RowCacheKey, IRowCacheEntry>
{
public void serialize(RowCacheKey key, DataOutputPlus out) throws IOException
public void serialize(RowCacheKey key, DataOutputPlus out, ColumnFamilyStore cfs) throws IOException
{
assert(!cfs.isIndex());
out.write(cfs.metadata.ksAndCFBytes);
ByteBufferUtil.writeWithLength(key.key, out);
}
public Future<Pair<RowCacheKey, IRowCacheEntry>> deserialize(DataInputStream in, final ColumnFamilyStore cfs) throws IOException
{
//Keyspace and CF name are deserialized by AutoSaving cache and used to fetch the CFS provided as a
//parameter so they aren't deserialized here, even though they are serialized by this serializer
final ByteBuffer buffer = ByteBufferUtil.readWithLength(in);
if (cfs == null || !cfs.isRowCacheEnabled())
return null;
assert(!cfs.isIndex());
return StageManager.getStage(Stage.READ).submit(new Callable<Pair<RowCacheKey, IRowCacheEntry>>()
{
public Pair<RowCacheKey, IRowCacheEntry> call() throws Exception
@ -402,7 +417,7 @@ public class CacheService implements CacheServiceMBean
DecoratedKey key = cfs.partitioner.decorateKey(buffer);
QueryFilter cacheFilter = new QueryFilter(key, cfs.getColumnFamilyName(), cfs.readFilterForCache(), Integer.MIN_VALUE);
ColumnFamily data = cfs.getTopLevelColumns(cacheFilter, Integer.MIN_VALUE);
return Pair.create(new RowCacheKey(cfs.metadata.cfId, key), (IRowCacheEntry) data);
return Pair.create(new RowCacheKey(cfs.metadata.ksAndCFName, key), (IRowCacheEntry) data);
}
});
}
@ -410,24 +425,23 @@ public class CacheService implements CacheServiceMBean
public static class KeyCacheSerializer implements CacheSerializer<KeyCacheKey, RowIndexEntry>
{
public void serialize(KeyCacheKey key, DataOutputPlus out) throws IOException
public void serialize(KeyCacheKey key, DataOutputPlus out, ColumnFamilyStore cfs) throws IOException
{
RowIndexEntry entry = CacheService.instance.keyCache.getInternal(key);
if (entry == null)
return;
CFMetaData cfm = Schema.instance.getCFMetaData(key.cfId);
if (cfm == null)
return; // the table no longer exists.
out.write(cfs.metadata.ksAndCFBytes);
ByteBufferUtil.writeWithLength(key.key, out);
out.writeInt(key.desc.generation);
out.writeBoolean(true);
key.desc.getFormat().getIndexSerializer(cfm).serialize(entry, out);
key.desc.getFormat().getIndexSerializer(cfs.metadata).serialize(entry, out);
}
public Future<Pair<KeyCacheKey, RowIndexEntry>> deserialize(DataInputStream input, ColumnFamilyStore cfs) throws IOException
{
//Keyspace and CF name are deserialized by AutoSaving cache and used to fetch the CFS provided as a
//parameter so they aren't deserialized here, even though they are serialized by this serializer
int keyLength = input.readInt();
if (keyLength > FBUtilities.MAX_UNSIGNED_SHORT)
{
@ -436,15 +450,15 @@ public class CacheService implements CacheServiceMBean
}
ByteBuffer key = ByteBufferUtil.read(input, keyLength);
int generation = input.readInt();
SSTableReader reader = findDesc(generation, cfs.getSSTables());
input.readBoolean(); // backwards compatibility for "promoted indexes" boolean
if (reader == null)
SSTableReader reader = null;
if (cfs == null || !cfs.isKeyCacheEnabled() || (reader = findDesc(generation, cfs.getSSTables())) == null)
{
RowIndexEntry.Serializer.skipPromotedIndex(input);
RowIndexEntry.Serializer.skip(input);
return null;
}
RowIndexEntry entry = reader.descriptor.getFormat().getIndexSerializer(reader.metadata).deserialize(input, reader.descriptor.version);
return Futures.immediateFuture(Pair.create(new KeyCacheKey(cfs.metadata.cfId, reader.descriptor, key), entry));
return Futures.immediateFuture(Pair.create(new KeyCacheKey(cfs.metadata.ksAndCFName, reader.descriptor, key), entry));
}
private SSTableReader findDesc(int generation, Collection<SSTableReader> collection)

41
src/java/org/apache/cassandra/service/CassandraDaemon.java
View File

@ -26,11 +26,12 @@ import java.net.UnknownHostException;
import java.rmi.registry.LocateRegistry;
import java.rmi.server.RMIServerSocketFactory;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import javax.management.MBeanServer;
import javax.management.MalformedObjectNameException;
import javax.management.ObjectName;
import javax.management.StandardMBean;
import javax.management.remote.JMXConnectorServer;
@ -41,7 +42,8 @@ import com.codahale.metrics.Meter;
import com.codahale.metrics.MetricRegistryListener;
import com.codahale.metrics.SharedMetricRegistries;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Iterables;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.Uninterruptibles;
import org.apache.cassandra.metrics.DefaultNameFactory;
import org.slf4j.Logger;
@ -244,11 +246,16 @@ public class CassandraDaemon
}
}
if (CacheService.instance.keyCache.size() > 0)
logger.info("completed pre-loading ({} keys) key cache.", CacheService.instance.keyCache.size());
if (CacheService.instance.rowCache.size() > 0)
logger.info("completed pre-loading ({} keys) row cache.", CacheService.instance.rowCache.size());
try
{
loadRowAndKeyCacheAsync().get();
}
catch (Throwable t)
{
JVMStabilityInspector.inspectThrowable(t);
logger.warn("Error loading key or row cache", t);
}
try
{
@ -342,6 +349,22 @@ public class CassandraDaemon
completeSetup();
}
/*
* Asynchronously load the row and key cache in one off threads and return a compound future of the result.
* Error handling is pushed into the cache load since cache loads are allowed to fail and are handled by logging.
*/
private ListenableFuture<?> loadRowAndKeyCacheAsync()
{
final ListenableFuture<Integer> keyCacheLoad = CacheService.instance.keyCache.loadSavedAsync();
final ListenableFuture<Integer> rowCacheLoad = CacheService.instance.rowCache.loadSavedAsync();
@SuppressWarnings("unchecked")
ListenableFuture<List<Integer>> retval = Futures.successfulAsList(keyCacheLoad, rowCacheLoad);
return retval;
}
@VisibleForTesting
public void completeSetup()
{
@ -365,13 +388,13 @@ public class CassandraDaemon
{
logger.info("Could not resolve local host");
}
logger.info("JVM vendor/version: {}/{}", System.getProperty("java.vm.name"), System.getProperty("java.version"));
logger.info("Heap size: {}/{}", Runtime.getRuntime().totalMemory(), Runtime.getRuntime().maxMemory());
for(MemoryPoolMXBean pool: ManagementFactory.getMemoryPoolMXBeans())
logger.info("{} {}: {}", pool.getName(), pool.getType(), pool.getPeakUsage());
logger.info("Classpath: {}", System.getProperty("java.class.path"));
}
}

31
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -688,9 +688,16 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
prepareToJoin();
// Has to be called after the host id has potentially changed in prepareToJoin().
for (ColumnFamilyStore cfs : ColumnFamilyStore.all())
if (cfs.metadata.isCounter())
cfs.initCounterCache();
try
{
CacheService.instance.counterCache.loadSavedAsync().get();
}
catch (Throwable t)
{
JVMStabilityInspector.inspectThrowable(t);
logger.warn("Error loading counter cache", t);
}
if (Boolean.parseBoolean(System.getProperty("cassandra.join_ring", "true")))
{
@ -2617,8 +2624,8 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
/**
* Takes the snapshot of a multiple column family from different keyspaces. A snapshot name must be specified.
*
*
*
*
* @param tag
* the tag given to the snapshot; may not be null or empty
* @param columnFamilyList
@ -3825,7 +3832,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
public synchronized void drain() throws IOException, InterruptedException, ExecutionException
{
inShutdownHook = true;
ExecutorService counterMutationStage = StageManager.getStage(Stage.COUNTER_MUTATION);
ExecutorService mutationStage = StageManager.getStage(Stage.MUTATION);
if (mutationStage.isTerminated() && counterMutationStage.isTerminated())
@ -3955,32 +3962,32 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
*/
public LinkedHashMap<InetAddress, Float> effectiveOwnership(String keyspace) throws IllegalStateException
{
if (keyspace != null)
{
Keyspace keyspaceInstance = Schema.instance.getKeyspaceInstance(keyspace);
if(keyspaceInstance == null)
throw new IllegalArgumentException("The keyspace " + keyspace + ", does not exist");
if(keyspaceInstance.getReplicationStrategy() instanceof LocalStrategy)
throw new IllegalStateException("Ownership values for keyspaces with LocalStrategy are meaningless");
}
else
{
List<String> nonSystemKeyspaces = Schema.instance.getNonSystemKeyspaces();
//system_traces is a non-system keyspace however it needs to be counted as one for this process
int specialTableCount = 0;
if (nonSystemKeyspaces.contains("system_traces"))
{
specialTableCount += 1;
}
if (nonSystemKeyspaces.size() > specialTableCount)
if (nonSystemKeyspaces.size() > specialTableCount)
throw new IllegalStateException("Non-system keyspaces don't have the same replication settings, effective ownership information is meaningless");
keyspace = "system_traces";
}
TokenMetadata metadata = tokenMetadata.cloneOnlyTokenMap();
Collection<Collection<InetAddress>> endpointsGroupedByDc = new ArrayList<>();

16
src/java/org/apache/cassandra/utils/FBUtilities.java
View File

@ -792,4 +792,20 @@ public class FBUtilities
digest.update((byte) ((val >>> 8) & 0xFF));
digest.update((byte) ((val >>> 0) & 0xFF));
}
public static byte[] toWriteUTFBytes(String s)
{
try
{
ByteArrayOutputStream baos = new ByteArrayOutputStream();
DataOutputStream dos = new DataOutputStream(baos);
dos.writeUTF(s);
dos.flush();
return baos.toByteArray();
}
catch (IOException e)
{
throw new RuntimeException(e);
}
}
}

5
test/unit/org/apache/cassandra/cache/AutoSavingCacheTest.java vendored
View File

@ -76,9 +76,8 @@ public class AutoSavingCacheTest
Assert.assertEquals(0, keyCache.size());
// then load saved
keyCache.loadSaved(cfs);
Assert.assertEquals(2, keyCache.size());
keyCache.loadSavedAsync().get();
for (SSTableReader sstable : cfs.getSSTables())
Assert.assertNotNull(keyCache.get(new KeyCacheKey(cfs.metadata.cfId, sstable.descriptor, ByteBufferUtil.bytes("key1"))));
Assert.assertNotNull(keyCache.get(new KeyCacheKey(cfs.metadata.ksAndCFName, sstable.descriptor, ByteBufferUtil.bytes("key1"))));
}
}

17
test/unit/org/apache/cassandra/cache/CacheProviderTest.java vendored
View File

@ -1,4 +1,3 @@
package org.apache.cassandra.cache;
/*
*
* Licensed to the Apache Software Foundation (ASF) under one
@ -19,12 +18,12 @@ package org.apache.cassandra.cache;
* under the License.
*
*/
package org.apache.cassandra.cache;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import org.junit.BeforeClass;
import org.junit.Test;
@ -35,6 +34,7 @@ import org.apache.cassandra.db.ArrayBackedSortedColumns;
import org.apache.cassandra.db.ColumnFamily;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.locator.SimpleStrategy;
import org.apache.cassandra.utils.Pair;
import com.googlecode.concurrentlinkedhashmap.Weighers;
@ -128,21 +128,20 @@ public class CacheProviderTest
simpleCase(cf, cache);
concurrentCase(cf, cache);
}
@Test
public void testKeys()
{
UUID cfId = UUID.randomUUID();
Pair<String, String> ksAndCFName = Pair.create(KEYSPACE1, CF_STANDARD1);
byte[] b1 = {1, 2, 3, 4};
RowCacheKey key1 = new RowCacheKey(cfId, ByteBuffer.wrap(b1));
RowCacheKey key1 = new RowCacheKey(ksAndCFName, ByteBuffer.wrap(b1));
byte[] b2 = {1, 2, 3, 4};
RowCacheKey key2 = new RowCacheKey(cfId, ByteBuffer.wrap(b2));
RowCacheKey key2 = new RowCacheKey(ksAndCFName, ByteBuffer.wrap(b2));
assertEquals(key1, key2);
assertEquals(key1.hashCode(), key2.hashCode());
byte[] b3 = {1, 2, 3, 5};
RowCacheKey key3 = new RowCacheKey(cfId, ByteBuffer.wrap(b3));
RowCacheKey key3 = new RowCacheKey(ksAndCFName, ByteBuffer.wrap(b3));
assertNotSame(key1, key3);
assertNotSame(key1.hashCode(), key3.hashCode());
}

266
test/unit/org/apache/cassandra/cql3/KeyCacheCqlTest.java Normal file
View File

@ -0,0 +1,266 @@
/*
* 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.cql3;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import org.junit.Assert;
import org.junit.Test;
import com.google.common.collect.ImmutableSet;
import org.apache.cassandra.cache.KeyCacheKey;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.metrics.CacheMetrics;
import org.apache.cassandra.metrics.CassandraMetricsRegistry;
import org.apache.cassandra.service.CacheService;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.Pair;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
public class KeyCacheCqlTest extends CQLTester
{
static final String commonColumnsDef =
"part_key_a int," +
"part_key_b text," +
"clust_key_a int," +
"clust_key_b text," +
"clust_key_c frozen<list<text>>," + // to make it really big
"col_text text," +
"col_int int," +
"col_long bigint,";
static final String commonColumns =
"part_key_a," +
"part_key_b," +
"clust_key_a," +
"clust_key_b," +
"clust_key_c," + // to make it really big
"col_text," +
"col_int," +
"col_long";
@Test
public void test2iKeyCachePaths() throws Throwable
{
String table = createTable("CREATE TABLE %s ("
+ commonColumnsDef
+ "PRIMARY KEY ((part_key_a, part_key_b),clust_key_a,clust_key_b,clust_key_c))");
createIndex("CREATE INDEX some_index ON %s (col_int)");
insertData(table, "some_index", true);
clearCache();
CacheMetrics metrics = CacheService.instance.keyCache.getMetrics();
for (int i = 0; i < 10; i++)
{
UntypedResultSet result = execute("SELECT part_key_a FROM %s WHERE col_int = ?", i);
assertEquals(500, result.size());
}
long hits = metrics.hits.getCount();
long requests = metrics.requests.getCount();
assertEquals(4900, hits);
assertEquals(5250, requests);
//
for (int i = 0; i < 10; i++)
{
UntypedResultSet result = execute("SELECT part_key_a FROM %s WHERE col_int = ?", i);
// 100 part-keys * 50 clust-keys
// indexed on part-key % 10 = 10 index partitions
// (50 clust-keys * 100-part-keys / 10 possible index-values) = 500
assertEquals(500, result.size());
}
metrics = CacheService.instance.keyCache.getMetrics();
hits = metrics.hits.getCount();
requests = metrics.requests.getCount();
assertEquals(10000, hits);
assertEquals(10500, requests);
CacheService.instance.keyCache.submitWrite(Integer.MAX_VALUE).get();
int beforeSize = CacheService.instance.keyCache.size();
CacheService.instance.keyCache.clear();
Assert.assertEquals(0, CacheService.instance.keyCache.size());
// then load saved
CacheService.instance.keyCache.loadSaved();
assertEquals(beforeSize, CacheService.instance.keyCache.size());
for (int i = 0; i < 10; i++)
{
UntypedResultSet result = execute("SELECT part_key_a FROM %s WHERE col_int = ?", i);
// 100 part-keys * 50 clust-keys
// indexed on part-key % 10 = 10 index partitions
// (50 clust-keys * 100-part-keys / 10 possible index-values) = 500
assertEquals(500, result.size());
}
//Test Schema.getColumnFamilyStoreIncludingIndexes, several null check paths
//are defensive and unreachable
assertNull(Schema.instance.getColumnFamilyStoreIncludingIndexes(Pair.create("foo", "bar")));
assertNull(Schema.instance.getColumnFamilyStoreIncludingIndexes(Pair.create(KEYSPACE, "bar")));
dropTable("DROP TABLE %s");
//Test loading for a dropped 2i/table
CacheService.instance.keyCache.clear();
// then load saved
CacheService.instance.keyCache.loadSaved();
assertEquals(0, CacheService.instance.keyCache.size());
}
@Test
public void test2iKeyCachePathsSaveKeysForDroppedTable() throws Throwable
{
String table = createTable("CREATE TABLE %s ("
+ commonColumnsDef
+ "PRIMARY KEY ((part_key_a, part_key_b),clust_key_a,clust_key_b,clust_key_c))");
createIndex("CREATE INDEX some_index ON %s (col_int)");
insertData(table, "some_index", true);
clearCache();
CacheMetrics metrics = CacheService.instance.keyCache.getMetrics();
for (int i = 0; i < 10; i++)
{
UntypedResultSet result = execute("SELECT part_key_a FROM %s WHERE col_int = ?", i);
assertEquals(500, result.size());
}
long hits = metrics.hits.getCount();
long requests = metrics.requests.getCount();
assertEquals(4900, hits);
assertEquals(5250, requests);
//
for (int i = 0; i < 10; i++)
{
UntypedResultSet result = execute("SELECT part_key_a FROM %s WHERE col_int = ?", i);
// 100 part-keys * 50 clust-keys
// indexed on part-key % 10 = 10 index partitions
// (50 clust-keys * 100-part-keys / 10 possible index-values) = 500
assertEquals(500, result.size());
}
metrics = CacheService.instance.keyCache.getMetrics();
hits = metrics.hits.getCount();
requests = metrics.requests.getCount();
assertEquals(10000, hits);
assertEquals(10500, requests);
dropTable("DROP TABLE %s");
CacheService.instance.keyCache.submitWrite(Integer.MAX_VALUE).get();
CacheService.instance.keyCache.clear();
Assert.assertEquals(0, CacheService.instance.keyCache.size());
// then load saved
CacheService.instance.keyCache.loadSaved();
Iterator<KeyCacheKey> iter = CacheService.instance.keyCache.keyIterator();
while(iter.hasNext())
{
KeyCacheKey key = iter.next();
Assert.assertFalse(key.ksAndCFName.left.equals("KEYSPACE"));
Assert.assertFalse(key.ksAndCFName.right.startsWith(table));
}
}
// Inserts 100 partitions split over 10 sstables (flush after 10 partitions).
// Clustered tables receive 50 CQL rows per partition.
private void insertData(String table, String index, boolean withClustering) throws Throwable
{
StorageService.instance.disableAutoCompaction(KEYSPACE, table);
Keyspace.open(KEYSPACE).getColumnFamilyStore(table).forceFlush().get();
Keyspace.open(KEYSPACE).getColumnFamilyStore(table).truncateBlocking();
if (index != null)
{
StorageService.instance.disableAutoCompaction(KEYSPACE, table + '.' + index);
Keyspace.open(KEYSPACE).getColumnFamilyStore(table).indexManager.getIndexesByNames(ImmutableSet.of(table + "." + index)).iterator().next().forceBlockingFlush();
}
for (int i = 0; i < 100; i++)
{
int partKeyA = i;
String partKeyB = Integer.toOctalString(i);
for (int c = 0; c < (withClustering ? 50 : 1); c++)
{
int clustKeyA = c;
String clustKeyB = Integer.toOctalString(c);
List<String> clustKeyC = makeList(clustKeyB);
String colText = String.valueOf(i) + '-' + String.valueOf(c);
int colInt = i % 10;
long colLong = c;
execute("INSERT INTO %s (" + commonColumns + ") VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
partKeyA, partKeyB,
clustKeyA, clustKeyB, clustKeyC,
colText, colInt, colLong);
}
if (i % 10 == 9)
{
Keyspace.open(KEYSPACE).getColumnFamilyStore(table).forceFlush().get();
if (index != null)
Keyspace.open(KEYSPACE).getColumnFamilyStore(table).indexManager.getIndexesByNames(ImmutableSet.of(table + "." + index)).iterator().next().forceBlockingFlush();
}
}
}
private static List<String> makeList(String value)
{
List<String> list = new ArrayList<>(50);
for (int i = 0; i < 50; i++)
{
list.add(value + i);
}
return list;
}
private static void clearCache()
{
for (String name : ImmutableSet.copyOf(CassandraMetricsRegistry.Metrics.getMetrics().keySet()))
{
CassandraMetricsRegistry.Metrics.remove(name);
}
CacheService.instance.keyCache.clear();
CacheMetrics metrics = CacheService.instance.keyCache.getMetrics();
Assert.assertEquals(0, metrics.entries.getValue().intValue());
Assert.assertEquals(0L, metrics.hits.getCount());
Assert.assertEquals(0L, metrics.requests.getCount());
Assert.assertEquals(0L, metrics.size.getValue().longValue());
}
}

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

@ -27,6 +27,7 @@ import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.config.KSMetaData;
import org.apache.cassandra.db.marshal.CounterColumnType;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.exceptions.WriteTimeoutException;
import org.apache.cassandra.locator.SimpleStrategy;
import org.apache.cassandra.service.CacheService;
@ -63,6 +64,7 @@ public class CounterCacheTest
public void testReadWrite()
{
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF);
cfs.truncateBlocking();
CacheService.instance.invalidateCounterCache();
assertEquals(0, CacheService.instance.counterCache.size());
@ -87,6 +89,7 @@ public class CounterCacheTest
public void testSaveLoad() throws ExecutionException, InterruptedException, WriteTimeoutException
{
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF);
cfs.truncateBlocking();
CacheService.instance.invalidateCounterCache();
ColumnFamily cells = ArrayBackedSortedColumns.factory.create(cfs.metadata);
@ -101,11 +104,76 @@ public class CounterCacheTest
assertEquals(0, CacheService.instance.counterCache.size());
// load from cache and validate
CacheService.instance.counterCache.loadSaved(cfs);
CacheService.instance.counterCache.loadSaved();
assertEquals(4, 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(1L, 1L), cfs.getCachedCounter(bytes(2), cellname(1)));
assertEquals(ClockAndCount.create(1L, 2L), cfs.getCachedCounter(bytes(2), cellname(2)));
}
@Test
public void testDroppedSaveLoad() throws ExecutionException, InterruptedException, WriteTimeoutException
{
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF);
cfs.truncateBlocking();
CacheService.instance.invalidateCounterCache();
ColumnFamily cells = ArrayBackedSortedColumns.factory.create(cfs.metadata);
cells.addColumn(new BufferCounterUpdateCell(cellname(1), 1L, FBUtilities.timestampMicros()));
cells.addColumn(new BufferCounterUpdateCell(cellname(2), 2L, FBUtilities.timestampMicros()));
new CounterMutation(new Mutation(KEYSPACE1, bytes(1), cells), ConsistencyLevel.ONE).apply();
new CounterMutation(new Mutation(KEYSPACE1, bytes(2), cells), ConsistencyLevel.ONE).apply();
// flush the counter cache and invalidate
CacheService.instance.counterCache.submitWrite(Integer.MAX_VALUE).get();
CacheService.instance.invalidateCounterCache();
assertEquals(0, CacheService.instance.counterCache.size());
Keyspace ks = Schema.instance.removeKeyspaceInstance(KEYSPACE1);
try
{
// load from cache and validate
CacheService.instance.counterCache.loadSaved();
assertEquals(0, CacheService.instance.counterCache.size());
}
finally
{
Schema.instance.storeKeyspaceInstance(ks);
}
}
@Test
public void testDisabledSaveLoad() throws ExecutionException, InterruptedException, WriteTimeoutException
{
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF);
cfs.truncateBlocking();
CacheService.instance.invalidateCounterCache();
ColumnFamily cells = ArrayBackedSortedColumns.factory.create(cfs.metadata);
cells.addColumn(new BufferCounterUpdateCell(cellname(1), 1L, FBUtilities.timestampMicros()));
cells.addColumn(new BufferCounterUpdateCell(cellname(2), 2L, FBUtilities.timestampMicros()));
new CounterMutation(new Mutation(KEYSPACE1, bytes(1), cells), ConsistencyLevel.ONE).apply();
new CounterMutation(new Mutation(KEYSPACE1, bytes(2), cells), ConsistencyLevel.ONE).apply();
// flush the counter cache and invalidate
CacheService.instance.counterCache.submitWrite(Integer.MAX_VALUE).get();
CacheService.instance.invalidateCounterCache();
assertEquals(0, CacheService.instance.counterCache.size());
CacheService.instance.setCounterCacheCapacityInMB(0);
try
{
// load from cache and validate
CacheService.instance.counterCache.loadSaved();
assertEquals(0, CacheService.instance.counterCache.size());
}
finally
{
CacheService.instance.setCounterCacheCapacityInMB(1);
}
}
}

2
test/unit/org/apache/cassandra/db/KeyCacheTest.java
View File

@ -105,7 +105,7 @@ public class KeyCacheTest
CacheService.instance.invalidateKeyCache();
assertKeyCacheSize(0, KEYSPACE1, COLUMN_FAMILY2);
CacheService.instance.keyCache.loadSaved(store);
CacheService.instance.keyCache.loadSaved();
assertKeyCacheSize(savedMap.size(), KEYSPACE1, COLUMN_FAMILY2);
// probably it's better to add equals/hashCode to RowIndexEntry...

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

@ -31,6 +31,7 @@ import org.apache.cassandra.Util;
import org.apache.cassandra.cache.CachingOptions;
import org.apache.cassandra.cache.RowCacheKey;
import org.apache.cassandra.config.KSMetaData;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.db.composites.*;
import org.apache.cassandra.db.compaction.CompactionManager;
import org.apache.cassandra.db.filter.QueryFilter;
@ -177,6 +178,42 @@ public class RowCacheTest
rowCacheLoad(100, 50, 0);
CacheService.instance.setRowCacheCapacityInMB(0);
}
@Test
public void testRowCacheDropSaveLoad() throws Exception
{
CacheService.instance.setRowCacheCapacityInMB(1);
rowCacheLoad(100, 50, 0);
CacheService.instance.rowCache.submitWrite(Integer.MAX_VALUE).get();
Keyspace instance = Schema.instance.removeKeyspaceInstance(KEYSPACE_CACHED);
try
{
CacheService.instance.rowCache.size();
CacheService.instance.rowCache.clear();
CacheService.instance.rowCache.loadSaved();
int after = CacheService.instance.rowCache.size();
assertEquals(0, after);
}
finally
{
Schema.instance.storeKeyspaceInstance(instance);
}
}
@Test
public void testRowCacheDisabled() throws Exception
{
CacheService.instance.setRowCacheCapacityInMB(1);
rowCacheLoad(100, 50, 0);
CacheService.instance.rowCache.submitWrite(Integer.MAX_VALUE).get();
CacheService.instance.setRowCacheCapacityInMB(0);
CacheService.instance.rowCache.size();
CacheService.instance.rowCache.clear();
CacheService.instance.rowCache.loadSaved();
int after = CacheService.instance.rowCache.size();
assertEquals(0, after);
}
@Test
public void testRowCacheRange()
{
@ -195,7 +232,7 @@ public class RowCacheTest
ByteBuffer key = ByteBufferUtil.bytes("rowcachekey");
DecoratedKey dk = cachedStore.partitioner.decorateKey(key);
RowCacheKey rck = new RowCacheKey(cachedStore.metadata.cfId, dk);
RowCacheKey rck = new RowCacheKey(cachedStore.metadata.ksAndCFName, dk);
Mutation mutation = new Mutation(KEYSPACE_CACHED, key);
for (int i = 0; i < 200; i++)
mutation.add(cf, Util.cellname(i), ByteBufferUtil.bytes("val" + i), System.currentTimeMillis());
@ -272,6 +309,6 @@ public class RowCacheTest
// empty the cache again to make sure values came from disk
CacheService.instance.invalidateRowCache();
assertEquals(0, CacheService.instance.rowCache.size());
assertEquals(keysToSave == Integer.MAX_VALUE ? totalKeys : keysToSave, CacheService.instance.rowCache.loadSaved(store));
assertEquals(keysToSave == Integer.MAX_VALUE ? totalKeys : keysToSave, CacheService.instance.rowCache.loadSaved());
}
}