Merge branch 'cassandra-1.2' into cassandra-2.0.0

Conflicts:
	CHANGES.txt
	src/java/org/apache/cassandra/io/sstable/SSTableReader.java
	src/java/org/apache/cassandra/metrics/ColumnFamilyMetrics.java
	src/java/org/apache/cassandra/service/CacheService.java

CHANGES.txt

@@ -3,6 +3,7 @@
  * fix HsHa to respect max frame size (CASSANDRA-4573)
 Merged from 1.2:
  * Correctly validate sparse composite cells in scrub (CASSANDRA-5855)
+ * Add KeyCacheHitRate metric to CF metrics (CASSANDRA-5868)
 
 
 2.0.0-rc1

src/java/org/apache/cassandra/db/compaction/CompactionController.java

@@ -161,7 +161,7 @@ public class CompactionController
             {
                 // if we don't have bloom filter(bf_fp_chance=1.0 or filter file is missing),
                 // we check index file instead.
-                if (sstable.getBloomFilter() instanceof AlwaysPresentFilter && sstable.getPosition(key, SSTableReader.Operator.EQ) != null)
+                if (sstable.getBloomFilter() instanceof AlwaysPresentFilter && sstable.getPosition(key, SSTableReader.Operator.EQ, false) != null)
                     return false;
                 else if (sstable.getBloomFilter().isPresent(key.key))
                     return false;

src/java/org/apache/cassandra/io/sstable/SSTableReader.java

@@ -25,6 +25,7 @@ import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
 
 import com.google.common.primitives.Longs;
 import com.google.common.util.concurrent.RateLimiter;
@@ -104,6 +105,9 @@ public class SSTableReader extends SSTable
     // not final since we need to be able to change level on a file.
     private volatile SSTableMetadata sstableMetadata;
 
+    private final AtomicLong keyCacheHit = new AtomicLong(0);
+    private final AtomicLong keyCacheRequest = new AtomicLong(0);
+
     public static long getApproximateKeyCount(Iterable<SSTableReader> sstables, CFMetaData metadata)
     {
         long count = 0;
@@ -789,8 +793,20 @@ public class SSTableReader extends SSTable
 
     private RowIndexEntry getCachedPosition(KeyCacheKey unifiedKey, boolean updateStats)
     {
-        if (keyCache != null && keyCache.getCapacity() > 0)
-            return updateStats ? keyCache.get(unifiedKey) : keyCache.getInternal(unifiedKey);
+        if (keyCache != null && keyCache.getCapacity() > 0) {
+            if (updateStats)
+            {
+                RowIndexEntry cachedEntry = keyCache.get(unifiedKey);
+                keyCacheRequest.incrementAndGet();
+                if (cachedEntry != null)
+                    keyCacheHit.incrementAndGet();
+                return cachedEntry;
+            }
+            else
+            {
+                return keyCache.getInternal(unifiedKey);
+            }
+        }
         return null;
     }
 
@@ -1260,6 +1276,22 @@ public class SSTableReader extends SSTable
         return new File(descriptor.filenameFor(component)).lastModified();
     }
 
+    /**
+     * @return Number of key cache hit
+     */
+    public long getKeyCacheHit()
+    {
+        return keyCacheHit.get();
+    }
+
+    /**
+     * @return Number of key cache request
+     */
+    public long getKeyCacheRequest()
+    {
+        return keyCacheRequest.get();
+    }
+
     /**
      * @param sstables
      * @return true if all desired references were acquired.  Otherwise, it will unreference any partial acquisition, and return false.

src/java/org/apache/cassandra/metrics/ColumnFamilyMetrics.java

@@ -22,6 +22,7 @@ import com.yammer.metrics.core.Counter;
 import com.yammer.metrics.core.Gauge;
 import com.yammer.metrics.core.Histogram;
 import com.yammer.metrics.core.MetricName;
+import com.yammer.metrics.util.RatioGauge;
 
 import org.apache.cassandra.db.ColumnFamilyStore;
 import org.apache.cassandra.db.Keyspace;
@@ -76,6 +77,8 @@ public class ColumnFamilyMetrics
     public final Gauge<Double> recentBloomFilterFalseRatio;
     /** Disk space used by bloom filter */
     public final Gauge<Long> bloomFilterDiskSpaceUsed;
+    /** Key cache hit rate  for this CF */
+    public final Gauge<Double> keyCacheHitRate;
 
     private final MetricNameFactory factory;
 
@@ -277,6 +280,24 @@ public class ColumnFamilyMetrics
             }
         });
         speculativeRetry = Metrics.newCounter(factory.createMetricName("SpeculativeRetry"));
+        keyCacheHitRate = Metrics.newGauge(factory.createMetricName("KeyCacheHitRate"), new RatioGauge()
+        {
+            protected double getNumerator()
+            {
+                long hits = 0L;
+                for (SSTableReader sstable : cfs.getSSTables())
+                    hits += sstable.getKeyCacheHit();
+                return hits;
+            }
+
+            protected double getDenominator()
+            {
+                long requests = 0L;
+                for (SSTableReader sstable : cfs.getSSTables())
+                    requests += sstable.getKeyCacheRequest();
+                return Math.max(requests, 1); // to avoid NaN.
+            }
+        });
     }
 
     public void updateSSTableIterated(int count)

src/java/org/apache/cassandra/service/CacheService.java

@@ -365,7 +365,7 @@ public class CacheService implements CacheServiceMBean
 
                 for (SSTableReader sstable : cfs.getSSTables())
                 {
-                    RowIndexEntry entry = sstable.getPosition(dk, Operator.EQ);
+                    RowIndexEntry entry = sstable.getPosition(dk, Operator.EQ, false);
                     if (entry != null)
                         keyCache.put(new KeyCacheKey(sstable.descriptor, key), entry);
                 }