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

merge from 0.6 

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@916962 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2010-02-27 15:18:50 +00:00
parent e8e29e303d
commit 5d4371a0c4
17 changed files with 227 additions and 193 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
conf/storage-conf.xml
View File

@ -83,25 +83,22 @@
~ An optional `Comment` attribute may be used to attach additional
~ human-readable information about the column family to its definition.
~
~ The optional KeysCachedFraction attribute specifies
~ The fraction of keys per sstable whose locations we keep in
~ The optional KeysCached attribute specifies
~ the number of keys per sstable whose locations we keep in
~ memory in "mostly LRU" order. (JUST the key locations, NOT any
~ column values.) The amount of memory used by the default setting of
~ 0.01 is comparable to the amount used by the internal per-sstable key
~ index. Consider increasing this if you have fewer, wider rows.
~ Set to 0 to disable entirely.
~ column values.) Specify a fraction (value less than 1), a percentage
~ (ending in a % sign) or an absolute number of keys to cache.
~
~ The optional RowsCached attribute specifies the number of rows
~ whose entire contents we cache in memory, either as a fixed number
~ of rows or as a percent of rows in the ColumnFamily.
~ Do not use this on ColumnFamilies with large rows, or
~ ColumnFamilies with high write:read ratios. As with key caching,
~ valid values are from 0 to 1. The default 0 disables it entirely.
~ whose entire contents we cache in memory. Do not use this on
~ ColumnFamilies with large rows, or ColumnFamilies with high write:read
~ ratios. Specify a fraction (value less than 1), a percentage (ending in
~ a % sign) or an absolute number of rows to cache.
-->
<ColumnFamily CompareWith="BytesType"
Name="Standard1"
RowsCached="10%"
KeysCachedFraction="0"/>
KeysCached="0"/>
<ColumnFamily CompareWith="UTF8Type" Name="Standard2"/>
<ColumnFamily CompareWith="TimeUUIDType" Name="StandardByUUID1"/>
<ColumnFamily ColumnType="Super"
@ -109,7 +106,7 @@
CompareSubcolumnsWith="UTF8Type"
Name="Super1"
RowsCached="1000"
KeysCachedFraction="0"
KeysCached="50%"
Comment="A column family with supercolumns, whose column and subcolumn names are UTF8 strings"/>
<!--

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

@ -29,6 +29,9 @@ import java.io.InputStream;
public final class CFMetaData
{
public final static double DEFAULT_KEY_CACHE_SIZE = 0.1;
public final static double DEFAULT_ROW_CACHE_SIZE = 0.0;
public final String tableName; // name of table which has this column family
public final String cfName; // name of the column family
public final String columnType; // type: super, standard, etc.
@ -36,9 +39,9 @@ public final class CFMetaData
public final AbstractType subcolumnComparator; // like comparator, for supercolumns
public final String comment; // for humans only
public final double rowCacheSize; // default 0
public final double keysCachedFraction; // default 0.01
public final double keyCacheSize; // default 0.01
CFMetaData(String tableName, String cfName, String columnType, AbstractType comparator, AbstractType subcolumnComparator, String comment, double rowCacheSize, double keysCachedFraction)
CFMetaData(String tableName, String cfName, String columnType, AbstractType comparator, AbstractType subcolumnComparator, String comment, double rowCacheSize, double keyCacheSize)
{
this.tableName = tableName;
this.cfName = cfName;
@ -47,7 +50,7 @@ public final class CFMetaData
this.subcolumnComparator = subcolumnComparator;
this.comment = comment;
this.rowCacheSize = rowCacheSize;
this.keysCachedFraction = keysCachedFraction;
this.keyCacheSize = keyCacheSize;
}
// a quick and dirty pretty printer for describing the column family...
@ -73,7 +76,7 @@ public final class CFMetaData
if (cfm.comment != null)
dout.writeUTF(cfm.comment);
dout.writeDouble(cfm.rowCacheSize);
dout.writeDouble(cfm.keysCachedFraction);
dout.writeDouble(cfm.keyCacheSize);
dout.close();
return bout.toByteArray();
}
@ -105,8 +108,8 @@ public final class CFMetaData
}
String comment = din.readBoolean() ? din.readUTF() : null;
double rowCacheSize = din.readDouble();
double keysCachedFraction = din.readDouble();
return new CFMetaData(tableName, cfName, columnType, comparator, subcolumnComparator, comment, rowCacheSize, keysCachedFraction);
double keyCacheSize = din.readDouble();
return new CFMetaData(tableName, cfName, columnType, comparator, subcolumnComparator, comment, rowCacheSize, keyCacheSize);
}
public boolean equals(Object obj)
@ -121,6 +124,6 @@ public final class CFMetaData
&& FBUtilities.equals(other.subcolumnComparator, subcolumnComparator)
&& FBUtilities.equals(other.comment, comment)
&& other.rowCacheSize == rowCacheSize
&& other.keysCachedFraction == keysCachedFraction;
&& other.keyCacheSize == keyCacheSize;
}
}

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

@ -26,10 +26,10 @@ import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.BytesType;
import org.apache.cassandra.db.marshal.UTF8Type;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.locator.EndPointSnitch;
import org.apache.cassandra.locator.IEndPointSnitch;
import org.apache.cassandra.locator.AbstractReplicationStrategy;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.XMLUtils;
import org.apache.log4j.Logger;
import org.w3c.dom.Node;
@ -457,8 +457,8 @@ public class DatabaseDescriptor
new UTF8Type(),
null,
"persistent metadata for the local node",
0d,
0.01d));
0.0,
0.01));
systemMeta.cfMetaData.put(HintedHandOffManager.HINTS_CF, new CFMetaData(Table.SYSTEM_TABLE,
HintedHandOffManager.HINTS_CF,
@ -466,8 +466,8 @@ public class DatabaseDescriptor
new UTF8Type(),
new BytesType(),
"hinted handoff data",
0d,
0.01d));
0.0,
0.01));
/* Load the seeds for node contact points */
String[] seeds = xmlUtils.getNodeValues("/Storage/Seeds/Seed");
@ -642,23 +642,22 @@ public class DatabaseDescriptor
throw new ConfigurationException("CompareSubcolumnsWith is only a valid attribute on super columnfamilies (not regular columnfamily " + cfName + ")");
}
double keysCachedFraction = 0.01d;
double keyCacheSize = CFMetaData.DEFAULT_KEY_CACHE_SIZE;
if ((value = XMLUtils.getAttributeValue(columnFamily, "KeysCachedFraction")) != null)
{
keysCachedFraction = Double.valueOf(value);
keyCacheSize = Double.valueOf(value);
// TODO: KeysCachedFraction deprecated: remove in 1.0
logger_.warn("KeysCachedFraction is deprecated: use KeysCached instead.");
}
if ((value = XMLUtils.getAttributeValue(columnFamily, "KeysCached")) != null)
{
keyCacheSize = FBUtilities.parseDoubleOrPercent(value);
}
double rowCacheSize = 0;
double rowCacheSize = CFMetaData.DEFAULT_ROW_CACHE_SIZE;
if ((value = XMLUtils.getAttributeValue(columnFamily, "RowsCached")) != null)
{
if (value.endsWith("%"))
{
rowCacheSize = Double.valueOf(value.substring(0, value.length() - 1)) / 100;
}
else
{
rowCacheSize = Double.valueOf(value);
}
rowCacheSize = FBUtilities.parseDoubleOrPercent(value);
}
// Parse out user-specified logical names for the various dimensions
@ -666,7 +665,7 @@ public class DatabaseDescriptor
String comment = xmlUtils.getNodeValue(xqlCF + "Comment");
// insert it into the table dictionary.
meta.cfMetaData.put(cfName, new CFMetaData(tableName, cfName, columnType, comparator, subcolumnComparator, comment, rowCacheSize, keysCachedFraction));
meta.cfMetaData.put(cfName, new CFMetaData(tableName, cfName, columnType, comparator, subcolumnComparator, comment, rowCacheSize, keyCacheSize));
}
tables_.put(meta.name, meta);
@ -1085,20 +1084,24 @@ public class DatabaseDescriptor
return getCFMetaData(tableName, cfName).subcolumnComparator;
}
public static double getKeysCachedFraction(String tableName, String columnFamilyName)
/**
* @return The absolute number of keys that should be cached per table.
*/
public static int getKeysCachedFor(String tableName, String columnFamilyName, long expectedKeys)
{
CFMetaData cfm = getCFMetaData(tableName, columnFamilyName);
if (cfm == null)
return 0.01d;
return cfm.keysCachedFraction;
double v = (cfm == null) ? CFMetaData.DEFAULT_KEY_CACHE_SIZE : cfm.keyCacheSize;
return (int)Math.min(FBUtilities.absoluteFromFraction(v, expectedKeys), Integer.MAX_VALUE);
}
public static double getRowsCachedFraction(String tableName, String columnFamilyName)
/**
* @return The absolute number of rows that should be cached for the columnfamily.
*/
public static int getRowsCachedFor(String tableName, String columnFamilyName, long expectedRows)
{
CFMetaData cfm = getCFMetaData(tableName, columnFamilyName);
if (cfm == null)
return 0.01d;
return cfm.rowCacheSize;
double v = (cfm == null) ? CFMetaData.DEFAULT_ROW_CACHE_SIZE : cfm.rowCacheSize;
return (int)Math.min(FBUtilities.absoluteFromFraction(v, expectedRows), Integer.MAX_VALUE);
}
private static class ConfigurationException extends Exception

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

@ -128,7 +128,7 @@ public class BinaryMemtable implements IFlushable
assert bytes.length > 0;
writer.append(key, bytes);
}
SSTableReader sstable = writer.closeAndOpenReader(DatabaseDescriptor.getKeysCachedFraction(cfs.getTable().name, cfs.getColumnFamilyName()));
SSTableReader sstable = writer.closeAndOpenReader();
logger.info("Completed flushing " + writer.getFilename());
return sstable;
}

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

@ -18,51 +18,45 @@
package org.apache.cassandra.db;
import java.io.Closeable;
import java.io.File;
import java.io.IOException;
import java.io.Closeable;
import java.lang.management.ManagementFactory;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Condition;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import com.google.common.collect.AbstractIterator;
import com.google.common.collect.Iterables;
import org.apache.cassandra.cache.IAggregatableCacheProvider;
import org.apache.cassandra.cache.InstrumentedCache;
import org.apache.cassandra.cache.JMXAggregatingCache;
import org.apache.cassandra.cache.JMXInstrumentedCache;
import org.apache.log4j.Logger;
import org.apache.commons.collections.IteratorUtils;
import com.google.common.base.Predicate;
import com.google.common.collect.Iterables;
import com.google.common.collect.Iterators;
import org.apache.cassandra.concurrent.JMXEnabledThreadPoolExecutor;
import org.apache.cassandra.concurrent.NamedThreadFactory;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.commitlog.CommitLog;
import org.apache.cassandra.db.commitlog.CommitLogSegment;
import org.apache.cassandra.db.filter.*;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.dht.AbstractBounds;
import org.apache.cassandra.dht.Bounds;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.*;
import org.apache.cassandra.io.SSTable;
import org.apache.cassandra.io.SSTableReader;
import org.apache.cassandra.io.SSTableScanner;
import org.apache.cassandra.io.SSTableTracker;
import org.apache.cassandra.io.util.FileUtils;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.thrift.SliceRange;
import org.apache.cassandra.utils.*;
import org.apache.cassandra.concurrent.JMXEnabledThreadPoolExecutor;
import org.apache.cassandra.concurrent.NamedThreadFactory;
import org.apache.cassandra.db.filter.*;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.commons.collections.IteratorUtils;
import com.google.common.collect.Iterators;
import com.google.common.base.Predicate;
public class ColumnFamilyStore implements ColumnFamilyStoreMBean
{
@ -115,8 +109,6 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
/* active memtable associated with this ColumnFamilyStore. */
private Memtable memtable_;
private final JMXInstrumentedCache<String, ColumnFamily> rowCache;
// TODO binarymemtable ops are not threadsafe (do they need to be?)
private AtomicReference<BinaryMemtable> binaryMemtable_;
@ -190,51 +182,8 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
}
sstables.add(sstable);
}
ssTables_ = new SSTableTracker();
ssTables_ = new SSTableTracker(table, columnFamilyName);
ssTables_.add(sstables);
double v = DatabaseDescriptor.getRowsCachedFraction(table, columnFamilyName);
int cacheSize;
if (0 < v && v < 1)
cacheSize = Math.max(1, (int)(v * ssTables_.estimatedKeys()));
else
cacheSize = (int)v;
if (logger_.isDebugEnabled())
logger_.debug("row cache capacity for " + columnFamilyName + " is " + cacheSize);
rowCache = new JMXInstrumentedCache<String, ColumnFamily>(table, columnFamilyName + "RowCache", cacheSize);
// we don't need to keep a reference to the key cache aggregator, just create it so it registers itself w/ JMX
new JMXAggregatingCache(new Iterable<IAggregatableCacheProvider>()
{
public Iterator<IAggregatableCacheProvider> iterator()
{
final Iterator<SSTableReader> iter = ssTables_.iterator();
return new AbstractIterator<IAggregatableCacheProvider>()
{
@Override
protected IAggregatableCacheProvider computeNext()
{
if (!iter.hasNext())
return endOfData();
return new IAggregatableCacheProvider()
{
SSTableReader sstable = iter.next();
public InstrumentedCache getCache()
{
return sstable.getKeyCache();
}
public long getObjectCount()
{
return sstable.getIndexPositions().size() * SSTableReader.indexInterval();
}
};
}
};
}
}, table, columnFamilyName + "KeyCache");
}
public static ColumnFamilyStore createColumnFamilyStore(String table, String columnFamily) throws IOException
@ -726,12 +675,12 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
private ColumnFamily cacheRow(String key) throws IOException
{
ColumnFamily cached;
if ((cached = rowCache.get(key)) == null)
if ((cached = ssTables_.getRowCache().get(key)) == null)
{
cached = getTopLevelColumns(new IdentityQueryFilter(key, new QueryPath(columnFamily_)), Integer.MIN_VALUE);
if (cached == null)
return null;
rowCache.put(key, cached);
ssTables_.getRowCache().put(key, cached);
}
return cached;
}
@ -750,7 +699,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
{
if (filter.path.superColumnName == null)
{
if (rowCache.getCapacity() == 0)
if (ssTables_.getRowCache().getCapacity() == 0)
return removeDeleted(getTopLevelColumns(filter, gcBefore), gcBefore);
ColumnFamily cached = cacheRow(filter.key);
@ -763,7 +712,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
// we are querying subcolumns of a supercolumn: fetch the supercolumn with NQF, then filter in-memory.
ColumnFamily cf;
SuperColumn sc;
if (rowCache.getCapacity() == 0)
if (ssTables_.getRowCache().getCapacity() == 0)
{
QueryFilter nameFilter = new NamesQueryFilter(filter.key, new QueryPath(columnFamily_), filter.path.superColumnName);
cf = getTopLevelColumns(nameFilter, gcBefore);
@ -1100,13 +1049,13 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
/** raw cached row -- does not fetch the row if it is not present. not counted in cache statistics. */
public ColumnFamily getRawCachedRow(String key)
{
return rowCache.getCapacity() == 0 ? null : rowCache.getInternal(key);
return ssTables_.getRowCache().getCapacity() == 0 ? null : ssTables_.getRowCache().getInternal(key);
}
void invalidateCachedRow(String key)
{
if (rowCache != null)
rowCache.remove(key);
if (ssTables_.getRowCache() != null)
ssTables_.getRowCache().remove(key);
}
public void forceMajorCompaction()

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

@ -310,7 +310,7 @@ public class CompactionManager implements CompactionManagerMBean
ci.close();
}
SSTableReader ssTable = writer.closeAndOpenReader(DatabaseDescriptor.getKeysCachedFraction(table.name, cfs.getColumnFamilyName()));
SSTableReader ssTable = writer.closeAndOpenReader();
cfs.replaceCompactedSSTables(sstables, Arrays.asList(ssTable));
submitMinorIfNeeded(cfs);
@ -384,7 +384,7 @@ public class CompactionManager implements CompactionManagerMBean
if (writer != null)
{
results.add(writer.closeAndOpenReader(DatabaseDescriptor.getKeysCachedFraction(table.name, cfs.getColumnFamilyName()), target != null));
results.add(writer.closeAndOpenReader());
String format = "AntiCompacted to %s. %d/%d bytes for %d keys. Time: %dms.";
long dTime = System.currentTimeMillis() - startTime;
logger.info(String.format(format, writer.getFilename(), SSTable.getTotalBytes(sstables), results.get(0).length(), totalkeysWritten, dTime));

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

@ -158,7 +158,7 @@ public class Memtable implements Comparable<Memtable>, IFlushable
writer.append(entry.getKey(), buffer);
}
SSTableReader ssTable = writer.closeAndOpenReader(DatabaseDescriptor.getKeysCachedFraction(getTableName(), cfs.getColumnFamilyName()));
SSTableReader ssTable = writer.closeAndOpenReader();
logger.info("Completed flushing " + ssTable.getFilename());
return ssTable;
}

62
src/java/org/apache/cassandra/io/SSTableReader.java
View File

@ -31,8 +31,8 @@ import org.apache.commons.lang.StringUtils;
import org.apache.cassandra.cache.InstrumentedCache;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.utils.BloomFilter;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.*;
@ -42,8 +42,6 @@ import org.apache.cassandra.io.util.FileDataInput;
import org.apache.cassandra.io.util.MappedFileDataInput;
import org.cliffc.high_scale_lib.NonBlockingHashMap;
import com.google.common.base.Predicate;
import com.google.common.base.Predicates;
/**
* SSTableReaders are open()ed by Table.onStart; after that they are created by SSTableWriter.renameAndOpen.
@ -116,12 +114,10 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
public static SSTableReader open(String dataFileName) throws IOException
{
return open(dataFileName,
StorageService.getPartitioner(),
DatabaseDescriptor.getKeysCachedFraction(parseTableName(dataFileName), parseColumnFamilyName(dataFileName)));
return open(dataFileName, StorageService.getPartitioner());
}
public static SSTableReader open(String dataFileName, IPartitioner partitioner, double keysCacheFraction) throws IOException
public static SSTableReader open(String dataFileName, IPartitioner partitioner) throws IOException
{
assert partitioner != null;
@ -130,10 +126,7 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
logger.info("Sampling index for " + dataFileName);
sstable.loadIndexFile();
sstable.loadBloomFilter();
if (keysCacheFraction > 0)
{
sstable.keyCache = createKeyCache((int)((sstable.getIndexPositions().size() + 1) * INDEX_INTERVAL * keysCacheFraction));
}
if (logger.isDebugEnabled())
logger.debug("INDEX LOAD TIME for " + dataFileName + ": " + (System.currentTimeMillis() - start) + " ms.");
@ -145,23 +138,15 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
private final MappedByteBuffer[] indexBuffers;
private final MappedByteBuffer[] buffers;
public static InstrumentedCache<DecoratedKey, PositionSize> createKeyCache(int size)
{
return new InstrumentedCache<DecoratedKey, PositionSize>(size);
}
private InstrumentedCache<DecoratedKey, PositionSize> keyCache;
private InstrumentedCache<Pair<Descriptor,DecoratedKey>,PositionSize> keyCache;
SSTableReader(String filename,
IPartitioner partitioner,
List<KeyPosition> indexPositions, Map<KeyPosition, PositionSize> spannedIndexDataPositions,
BloomFilter bloomFilter,
InstrumentedCache<DecoratedKey, PositionSize> keyCache)
throws IOException
BloomFilter bloomFilter)
throws IOException
{
super(filename, partitioner);
assert keyCache != null;
if (DatabaseDescriptor.getIndexAccessMode() == DatabaseDescriptor.DiskAccessMode.mmap)
{
@ -201,13 +186,13 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
this.indexPositions = indexPositions;
this.spannedIndexDataPositions = spannedIndexDataPositions;
this.bf = bloomFilter;
this.keyCache = keyCache;
}
public void addFinalizingReference(SSTableTracker tracker)
public void setTrackedBy(SSTableTracker tracker)
{
phantomReference = new SSTableDeletingReference(tracker, this, finalizerQueue);
finalizers.add(phantomReference);
keyCache = tracker.getKeyCache();
}
private static MappedByteBuffer mmap(String filename, long start, int size) throws IOException
@ -234,7 +219,7 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
private SSTableReader(String filename, IPartitioner partitioner) throws IOException
{
this(filename, partitioner, null, null, null, SSTableReader.createKeyCache(0));
this(filename, partitioner, null, null, null);
}
public List<KeyPosition> getIndexPositions()
@ -242,6 +227,11 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
return indexPositions;
}
public long estimatedKeys()
{
return indexPositions.size() * INDEX_INTERVAL;
}
void loadBloomFilter() throws IOException
{
DataInputStream stream = new DataInputStream(new FileInputStream(filterFilename()));
@ -329,21 +319,29 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
*/
public PositionSize getPosition(DecoratedKey decoratedKey) throws IOException
{
// first, check bloom filter
if (!bf.isPresent(partitioner.convertToDiskFormat(decoratedKey)))
return null;
if (keyCache.getCapacity() > 0)
// next, the key cache
Pair<Descriptor, DecoratedKey> unifiedKey = new Pair<Descriptor, DecoratedKey>(desc, decoratedKey);
if (keyCache != null && keyCache.getCapacity() > 0)
{
PositionSize cachedPosition = keyCache.get(decoratedKey);
PositionSize cachedPosition = keyCache.get(unifiedKey);
if (cachedPosition != null)
{
return cachedPosition;
}
}
// next, see if the sampled index says it's impossible for the key to be present
KeyPosition sampledPosition = getIndexScanPosition(decoratedKey);
if (sampledPosition == null)
{
return null;
}
// handle exact sampled index hit
if (spannedIndexDataPositions != null)
{
PositionSize info = spannedIndexDataPositions.get(sampledPosition);
@ -351,6 +349,7 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
return info;
}
// scan the on-disk index, starting at the nearest sampled position
long p = sampledPosition.position;
FileDataInput input;
if (indexBuffers == null)
@ -392,8 +391,8 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
{
info = new PositionSize(position, length() - position);
}
if (keyCache.getCapacity() > 0)
keyCache.put(decoratedKey, info);
if (keyCache != null && keyCache.getCapacity() > 0)
keyCache.put(unifiedKey, info);
return info;
}
if (v > 0)
@ -511,11 +510,6 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
return ColumnFamily.create(getTableName(), getColumnFamilyName());
}
public InstrumentedCache<DecoratedKey, PositionSize> getKeyCache()
{
return keyCache;
}
public ICompactSerializer2<IColumn> getColumnSerializer()
{
return DatabaseDescriptor.getColumnFamilyType(getTableName(), getColumnFamilyName()).equals("Standard")

63
src/java/org/apache/cassandra/io/SSTableTracker.java
View File

@ -25,16 +25,35 @@ import java.util.*;
import java.io.IOException;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.cassandra.cache.JMXInstrumentedCache;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ColumnFamily;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.utils.Pair;
import org.apache.log4j.Logger;
public class SSTableTracker implements Iterable<SSTableReader>
{
private static final Logger logger = Logger.getLogger(SSTableTracker.class);
private volatile Set<SSTableReader> sstables;
private final AtomicLong liveSize = new AtomicLong();
private final AtomicLong totalSize = new AtomicLong();
public SSTableTracker()
private final String ksname;
private final String cfname;
private final JMXInstrumentedCache<Pair<SSTable.Descriptor,DecoratedKey>,SSTable.PositionSize> keyCache;
private final JMXInstrumentedCache<String, ColumnFamily> rowCache;
public SSTableTracker(String ksname, String cfname)
{
this.sstables = Collections.<SSTableReader>emptySet();
this.ksname = ksname;
this.cfname = cfname;
sstables = Collections.emptySet();
keyCache = new JMXInstrumentedCache<Pair<SSTable.Descriptor,DecoratedKey>,SSTable.PositionSize>(ksname, cfname + "KeyCache", 0);
rowCache = new JMXInstrumentedCache<String, ColumnFamily>(ksname, cfname + "RowCache", 0);
}
public synchronized void replace(Collection<SSTableReader> oldSSTables, Iterable<SSTableReader> replacements) throws IOException
@ -48,7 +67,7 @@ public class SSTableTracker implements Iterable<SSTableReader>
long size = sstable.bytesOnDisk();
liveSize.addAndGet(size);
totalSize.addAndGet(size);
sstable.addFinalizingReference(this);
sstable.setTrackedBy(this);
}
for (SSTableReader sstable : oldSSTables)
@ -60,6 +79,7 @@ public class SSTableTracker implements Iterable<SSTableReader>
}
sstables = Collections.unmodifiableSet(sstablesNew);
updateCacheSizes();
}
public synchronized void add(Iterable<SSTableReader> sstables)
@ -80,6 +100,31 @@ public class SSTableTracker implements Iterable<SSTableReader>
replace(compacted, Collections.<SSTableReader>emptyList());
}
/**
* Resizes the key and row caches based on the current key estimate.
*/
public synchronized void updateCacheSizes()
{
long keys = estimatedKeys();
int keyCacheSize = DatabaseDescriptor.getKeysCachedFor(ksname, cfname, keys);
if (keyCacheSize != keyCache.getCapacity())
{
// update cache size for the new key volume
if (logger.isDebugEnabled())
logger.debug("key cache capacity for " + cfname + " is " + keyCacheSize);
keyCache.setCapacity(keyCacheSize);
}
int rowCacheSize = DatabaseDescriptor.getRowsCachedFor(ksname, cfname, keys);
if (rowCacheSize != rowCache.getCapacity())
{
if (logger.isDebugEnabled())
logger.debug("row cache capacity for " + cfname + " is " + rowCacheSize);
rowCache.setCapacity(rowCacheSize);
}
}
// the modifiers create new, unmodifiable objects each time; the volatile fences the assignment
// so we don't need any further synchronization for the common case here
public Set<SSTableReader> getSSTables()
@ -102,12 +147,17 @@ public class SSTableTracker implements Iterable<SSTableReader>
sstables = Collections.emptySet();
}
public JMXInstrumentedCache<String, ColumnFamily> getRowCache()
{
return rowCache;
}
public long estimatedKeys()
{
long n = 0;
for (SSTableReader sstable : this)
{
n += sstable.getIndexPositions().size() * SSTableReader.INDEX_INTERVAL;
n += sstable.estimatedKeys();
}
return n;
}
@ -126,5 +176,10 @@ public class SSTableTracker implements Iterable<SSTableReader>
{
totalSize.addAndGet(-size);
}
public JMXInstrumentedCache<Pair<SSTable.Descriptor, DecoratedKey>, SSTable.PositionSize> getKeyCache()
{
return keyCache;
}
}

25
src/java/org/apache/cassandra/io/SSTableWriter.java
View File

@ -21,22 +21,23 @@ package org.apache.cassandra.io;
*/
import java.io.*;
import java.io.DataOutputStream;
import java.io.FileOutputStream;
import java.io.IOError;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import org.apache.log4j.Logger;
import org.apache.cassandra.cache.InstrumentedCache;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.io.util.BufferedRandomAccessFile;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.BloomFilter;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.io.util.BufferedRandomAccessFile;
import org.apache.cassandra.io.util.DataOutputBuffer;
import com.reardencommerce.kernel.collections.shared.evictable.ConcurrentLinkedHashMap;
public class SSTableWriter extends SSTable
{
@ -128,16 +129,16 @@ public class SSTableWriter extends SSTable
afterAppend(decoratedKey, currentPosition, value.length);
}
public SSTableReader closeAndOpenReader(double cacheFraction) throws IOException
public SSTableReader closeAndOpenReader() throws IOException
{
return closeAndOpenReader(cacheFraction, false);
return closeAndOpenReader(false);
}
/**
* Renames temporary SSTable files to valid data, index, and bloom filter files.
* The SSTableWriter object will no longer be valid.
*/
public SSTableReader closeAndOpenReader(double cacheFraction, boolean temporary) throws IOException
public SSTableReader closeAndOpenReader(boolean temporary) throws IOException
{
// bloom filter
FileOutputStream fos = new FileOutputStream(filterFilename());
@ -162,8 +163,7 @@ public class SSTableWriter extends SSTable
newpath = rename(newpath); // important to do this last since index & filter file names are derived from it
}
InstrumentedCache<DecoratedKey, PositionSize> keyCache = SSTableReader.createKeyCache((int)(cacheFraction * keysWritten));
return new SSTableReader(newpath, partitioner, indexPositions, spannedIndexDataPositions, bf, keyCache);
return new SSTableReader(newpath, partitioner, indexPositions, spannedIndexDataPositions, bf);
}
static String rename(String tmpFilename)
@ -186,8 +186,7 @@ public class SSTableWriter extends SSTable
SSTableWriter.rename(filterFilename(dataFileName));
dataFileName = SSTableWriter.rename(dataFileName);
return SSTableReader.open(dataFileName,
StorageService.getPartitioner(),
DatabaseDescriptor.getKeysCachedFraction(parseTableName(dataFileName), parseColumnFamilyName(dataFileName)));
StorageService.getPartitioner());
}
}

4
src/java/org/apache/cassandra/tools/SSTableImport.java
View File

@ -161,7 +161,7 @@ public class SSTableImport
cfamily.clear();
}
writer.closeAndOpenReader(0);
writer.closeAndOpenReader();
}
catch (ClassCastException cce)
{
@ -209,4 +209,4 @@ public class SSTableImport
System.exit(0);
}
}
}

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

@ -63,6 +63,21 @@ public class FBUtilities
return result.toArray( new String[0] );
}
/**
* Parses a string representing either a fraction, absolute value or percentage.
*/
public static double parseDoubleOrPercent(String value)
{
if (value.endsWith("%"))
{
return Double.valueOf(value.substring(0, value.length() - 1)) / 100;
}
else
{
return Double.valueOf(value);
}
}
public static InetAddress getLocalAddress()
{
if (localInetAddress_ == null)
@ -79,6 +94,27 @@ public class FBUtilities
return localInetAddress_;
}
/**
* @param fractOrAbs A double that may represent a fraction or absolute value.
* @param total If fractionOrAbs is a fraction, the total to take the fraction from
* @return An absolute value which may be larger than the total.
*/
public static long absoluteFromFraction(double fractOrAbs, long total)
{
if (fractOrAbs < 0)
throw new UnsupportedOperationException("unexpected negative value " + fractOrAbs);
if (0 < fractOrAbs && fractOrAbs < 1)
{
// fraction
return Math.max(1, (long)(fractOrAbs * total));
}
// absolute
assert fractOrAbs >= 1 || fractOrAbs == 0;
return (long)fractOrAbs;
}
/**
* Given two bit arrays represented as BigIntegers, containing the given
* number of significant bits, calculate a midpoint.

4
test/conf/storage-conf.xml
View File

@ -40,11 +40,11 @@
<MemtableOperationsInMillions>0.00002</MemtableOperationsInMillions> <!-- 20 -->
<Keyspaces>
<Keyspace Name = "Keyspace1">
<ColumnFamily Name="Standard1" RowsCached="10%" KeysCachedFraction="0"/>
<ColumnFamily Name="Standard1" RowsCached="10%" KeysCached="0"/>
<ColumnFamily Name="Standard2"/>
<ColumnFamily CompareWith="LongType" Name="StandardLong1"/>
<ColumnFamily CompareWith="LongType" Name="StandardLong2"/>
<ColumnFamily ColumnType="Super" CompareSubcolumnsWith="LongType" Name="Super1" RowsCached="1000" KeysCachedFraction="0"/>
<ColumnFamily ColumnType="Super" CompareSubcolumnsWith="LongType" Name="Super1" RowsCached="1000" KeysCached="0"/>
<ColumnFamily ColumnType="Super" CompareSubcolumnsWith="LongType" Name="Super2"/>
<ColumnFamily ColumnType="Super" CompareSubcolumnsWith="LongType" Name="Super3"/>
<ColumnFamily ColumnType="Super" CompareSubcolumnsWith="UTF8Type" Name="Super4"/>

4
test/unit/org/apache/cassandra/io/SSTableTest.java
View File

@ -43,7 +43,7 @@ public class SSTableTest extends CleanupHelper
TreeMap<String, byte[]> map = new TreeMap<String,byte[]>();
map.put(key, bytes);
SSTableReader ssTable = SSTableUtils.writeRawSSTable("table", "singlewrite", map);
SSTableReader ssTable = SSTableUtils.writeRawSSTable("Keyspace1", "Standard1", map);
// verify
verifySingle(ssTable, bytes, key);
@ -71,7 +71,7 @@ public class SSTableTest extends CleanupHelper
}
// write
SSTableReader ssTable = SSTableUtils.writeRawSSTable("table", "manywrites", map);
SSTableReader ssTable = SSTableUtils.writeRawSSTable("Keyspace1", "Standard2", map);
// verify
verifyMany(ssTable, map);

2
test/unit/org/apache/cassandra/io/SSTableUtils.java
View File

@ -104,6 +104,6 @@ public class SSTableUtils
entry.getValue());
new File(writer.indexFilename()).deleteOnExit();
new File(writer.filterFilename()).deleteOnExit();
return writer.closeAndOpenReader(1.0);
return writer.closeAndOpenReader();
}
}

11
test/unit/org/apache/cassandra/tools/SSTableExportTest.java
View File

@ -28,7 +28,6 @@ import org.apache.cassandra.db.ColumnFamily;
import org.apache.cassandra.db.filter.NamesQueryFilter;
import org.apache.cassandra.db.filter.QueryPath;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.io.SSTableAccessor;
import org.apache.cassandra.io.SSTableReader;
import org.apache.cassandra.io.SSTableWriter;
import org.apache.cassandra.io.util.DataOutputBuffer;
@ -67,7 +66,7 @@ public class SSTableExportTest
dob.reset();
cfamily.clear();
writer.closeAndOpenReader(0);
writer.closeAndOpenReader();
// Enumerate and verify
File temp = File.createTempFile("Standard1", ".txt");
@ -105,7 +104,7 @@ public class SSTableExportTest
dob.reset();
cfamily.clear();
SSTableReader reader = writer.closeAndOpenReader(0);
SSTableReader reader = writer.closeAndOpenReader();
// Export to JSON and verify
File tempJson = File.createTempFile("Standard1", ".json");
@ -145,7 +144,7 @@ public class SSTableExportTest
dob.reset();
cfamily.clear();
SSTableReader reader = writer.closeAndOpenReader(0);
SSTableReader reader = writer.closeAndOpenReader();
// Export to JSON and verify
File tempJson = File.createTempFile("Super4", ".json");
@ -178,7 +177,7 @@ public class SSTableExportTest
dob.reset();
cfamily.clear();
SSTableReader reader = writer.closeAndOpenReader(0);
SSTableReader reader = writer.closeAndOpenReader();
// Export to JSON and verify
File tempJson = File.createTempFile("Standard1", ".json");
@ -188,7 +187,7 @@ public class SSTableExportTest
File tempSS2 = tempSSTableFile("Keyspace1", "Standard1");
SSTableImport.importJson(tempJson.getPath(), "Keyspace1", "Standard1", tempSS2.getPath());
reader = SSTableAccessor.getSSTableReader(tempSS2.getPath(), DatabaseDescriptor.getPartitioner());
reader = SSTableReader.open(tempSS2.getPath(), DatabaseDescriptor.getPartitioner());
NamesQueryFilter qf = new NamesQueryFilter("rowA", new QueryPath("Standard1", null, null), "name".getBytes());
ColumnFamily cf = qf.getSSTableColumnIterator(reader).getColumnFamily();
assertTrue(cf != null);

5
test/unit/org/apache/cassandra/tools/SSTableImportTest.java
View File

@ -26,7 +26,6 @@ import org.apache.cassandra.db.ColumnFamily;
import org.apache.cassandra.db.IColumn;
import org.apache.cassandra.db.filter.NamesQueryFilter;
import org.apache.cassandra.db.filter.QueryPath;
import org.apache.cassandra.io.SSTableAccessor;
import org.apache.cassandra.io.SSTableReader;
import static org.apache.cassandra.utils.FBUtilities.hexToBytes;
import static org.apache.cassandra.io.SSTableUtils.tempSSTableFile;
@ -44,7 +43,7 @@ public class SSTableImportTest
SSTableImport.importJson(jsonUrl, "Keyspace1", "Standard1", tempSS.getPath());
// Verify results
SSTableReader reader = SSTableAccessor.getSSTableReader(tempSS.getPath(), DatabaseDescriptor.getPartitioner());
SSTableReader reader = SSTableReader.open(tempSS.getPath(), DatabaseDescriptor.getPartitioner());
NamesQueryFilter qf = new NamesQueryFilter("rowA", new QueryPath("Standard1", null, null), "colAA".getBytes());
ColumnFamily cf = qf.getSSTableColumnIterator(reader).getColumnFamily();
assert Arrays.equals(cf.getColumn("colAA".getBytes()).value(), hexToBytes("76616c4141"));
@ -58,7 +57,7 @@ public class SSTableImportTest
SSTableImport.importJson(jsonUrl, "Keyspace1", "Super4", tempSS.getPath());
// Verify results
SSTableReader reader = SSTableAccessor.getSSTableReader(tempSS.getPath(), DatabaseDescriptor.getPartitioner());
SSTableReader reader = SSTableReader.open(tempSS.getPath(), DatabaseDescriptor.getPartitioner());
NamesQueryFilter qf = new NamesQueryFilter("rowA", new QueryPath("Super4", null, null), "superA".getBytes());
ColumnFamily cf = qf.getSSTableColumnIterator(reader).getColumnFamily();
IColumn superCol = cf.getColumn("superA".getBytes());