diff --git a/CHANGES.txt b/CHANGES.txt
index 124b3e4c70..9ed00e236a 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -3,6 +3,7 @@
  * Allow where clause conditions to be in parenthesis (CASSANDRA-6037)
  * Do not open non-ssl storage port if encryption option is all (CASSANDRA-3916)
  * Improve memory usage of metadata min/max column names (CASSANDRA-6077)
+ * Add tombstone debug threshold and histogram (CASSANDRA-6042, 6057)
 
 
 1.2.10
diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml
index 27ac09bd5c..a3cdec7f9f 100644
--- a/conf/cassandra.yaml
+++ b/conf/cassandra.yaml
@@ -440,6 +440,11 @@ snapshot_before_compaction: false
 # lose data on truncation or drop.
 auto_snapshot: true
 
+# Log a debug message if more than this many tombstones are scanned
+# in a single-partition query.  Set the threshold on SliceQueryFilter
+# to debug to enable.
+tombstone_debug_threshold: 10000
+
 # Add column indexes to a row after its contents reach this size.
 # Increase if your column values are large, or if you have a very large
 # number of columns.  The competing causes are, Cassandra has to
diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java
index a924a4c8b4..292161b2a3 100644
--- a/src/java/org/apache/cassandra/config/Config.java
+++ b/src/java/org/apache/cassandra/config/Config.java
@@ -176,6 +176,8 @@ public class Config
     private static boolean loadYaml = true;
     private static boolean outboundBindAny = false;
 
+    public volatile int tombstone_debug_threshold = 10000;
+
     public static boolean getOutboundBindAny()
     {
         return outboundBindAny;
diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
index 22de2d6613..2c999ef7bd 100644
--- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
+++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
@@ -929,6 +929,20 @@ public class DatabaseDescriptor
         return conf.commitlog_directory;
     }
 
+    /**
+     * How many tombstones need to be scanned before we log a
+     * debug message
+     */
+    public static int getTombstoneDebugThreshold()
+    {
+        return conf.tombstone_debug_threshold;
+    }
+
+    public static void setTombstoneDebugThreshold(int tombstoneDebugThreshold)
+    {
+        conf.tombstone_debug_threshold = tombstoneDebugThreshold;
+    }
+
     /**
      * size of commitlog segments to allocate
      */
diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
index c6464616db..a41c15707e 100644
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@ -49,10 +49,7 @@ import org.apache.cassandra.db.compaction.AbstractCompactionStrategy;
 import org.apache.cassandra.db.compaction.CompactionManager;
 import org.apache.cassandra.db.compaction.LeveledCompactionStrategy;
 import org.apache.cassandra.db.compaction.OperationType;
-import org.apache.cassandra.db.filter.ExtendedFilter;
-import org.apache.cassandra.db.filter.IDiskAtomFilter;
-import org.apache.cassandra.db.filter.QueryFilter;
-import org.apache.cassandra.db.filter.QueryPath;
+import org.apache.cassandra.db.filter.*;
 import org.apache.cassandra.db.index.SecondaryIndex;
 import org.apache.cassandra.db.index.SecondaryIndexManager;
 import org.apache.cassandra.db.marshal.AbstractType;
@@ -1221,6 +1218,13 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
                 result = cf.isSuper() ? removeDeleted(cf, gcBefore) : removeDeletedCF(cf, gcBefore);
 
             }
+
+            if (filter.filter instanceof SliceQueryFilter)
+            {
+                // Log the number of tombstones scanned on single key queries
+                metric.tombstoneScannedHistogram.update(((SliceQueryFilter) filter.filter).lastIgnored());
+                metric.liveScannedHistogram.update(((SliceQueryFilter) filter.filter).lastLive());
+            }
         }
         finally
         {
@@ -1907,6 +1911,22 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
         return getMinimumCompactionThreshold() <= 0 || getMaximumCompactionThreshold() <= 0;
     }
 
+    public long getTombstonesPerLastRead()
+    {
+        return metric.tombstoneScannedHistogram.count();
+    }
+
+    public float getPercentageTombstonesPerLastRead()
+    {
+        long total = metric.tombstoneScannedHistogram.count() + metric.liveScannedHistogram.count();
+        return (metric.tombstoneScannedHistogram.count() / total);
+    }
+
+    public long getLiveCellsPerLastRead()
+    {
+        return metric.liveScannedHistogram.count();
+    }
+
     // End JMX get/set.
 
     public long estimateKeys()
diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStoreMBean.java b/src/java/org/apache/cassandra/db/ColumnFamilyStoreMBean.java
index 554c2045b3..f937032754 100644
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStoreMBean.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStoreMBean.java
@@ -286,6 +286,15 @@ public interface ColumnFamilyStoreMBean
      */
     public void disableAutoCompaction();
 
+    /** Number of tombstoned cells retreived during the last slicequery */
+    public long getTombstonesPerLastRead();
+
+    /** Percentage of tombstoned cells retreived during the last slicequery */
+    public float getPercentageTombstonesPerLastRead();
+
+    /** Number of live cells retreived during the last slicequery */
+    public long getLiveCellsPerLastRead();
+
     public long estimateKeys();
 
     /**
diff --git a/src/java/org/apache/cassandra/db/filter/SliceQueryFilter.java b/src/java/org/apache/cassandra/db/filter/SliceQueryFilter.java
index 39d1ca14a9..7e1ec6bae0 100644
--- a/src/java/org/apache/cassandra/db/filter/SliceQueryFilter.java
+++ b/src/java/org/apache/cassandra/db/filter/SliceQueryFilter.java
@@ -17,7 +17,9 @@
  */
 package org.apache.cassandra.db.filter;
 
-import java.io.*;
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.*;
 
@@ -26,9 +28,10 @@ import com.google.common.collect.Lists;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.*;
-import org.apache.cassandra.db.columniterator.OnDiskAtomIterator;
 import org.apache.cassandra.db.columniterator.ISSTableColumnIterator;
+import org.apache.cassandra.db.columniterator.OnDiskAtomIterator;
 import org.apache.cassandra.db.columniterator.SSTableSliceIterator;
 import org.apache.cassandra.db.marshal.AbstractType;
 import org.apache.cassandra.db.marshal.CompositeType;
@@ -172,6 +175,8 @@ public class SliceQueryFilter implements IDiskAtomFilter
         }
 
         Tracing.trace("Read {} live and {} tombstoned cells", columnCounter.live(), columnCounter.ignored());
+        if (columnCounter.ignored() > DatabaseDescriptor.getTombstoneDebugThreshold())
+            logger.debug("Read {} live and {} tombstoned cells", columnCounter.live(), columnCounter.ignored());
     }
 
     public int getLiveCount(ColumnFamily cf)
@@ -249,6 +254,16 @@ public class SliceQueryFilter implements IDiskAtomFilter
         return columnCounter == null ? 0 : columnCounter.live();
     }
 
+    public int lastIgnored()
+    {
+        return columnCounter == null ? 0 : columnCounter.ignored();
+    }
+
+    public int lastLive()
+    {
+        return columnCounter == null ? 0 : columnCounter.live();
+    }
+
     @Override
     public String toString()
     {
diff --git a/src/java/org/apache/cassandra/metrics/ColumnFamilyMetrics.java b/src/java/org/apache/cassandra/metrics/ColumnFamilyMetrics.java
index d111667b0f..9bd90a97a6 100644
--- a/src/java/org/apache/cassandra/metrics/ColumnFamilyMetrics.java
+++ b/src/java/org/apache/cassandra/metrics/ColumnFamilyMetrics.java
@@ -79,6 +79,10 @@ public class ColumnFamilyMetrics
     public final Gauge<Long> bloomFilterDiskSpaceUsed;
     /** Key cache hit rate  for this CF */
     public final Gauge<Double> keyCacheHitRate;
+    /** Tombstones scanned in queries on this CF */
+    public final Histogram tombstoneScannedHistogram;
+    /** Live cells scanned in queries on this CF */
+    public final Histogram liveScannedHistogram;
 
     private final MetricNameFactory factory;
 
@@ -295,6 +299,8 @@ public class ColumnFamilyMetrics
                 return Math.max(requests, 1); // to avoid NaN.
             }
         });
+        tombstoneScannedHistogram = Metrics.newHistogram(factory.createMetricName("TombstoneScannedHistogram"));
+        liveScannedHistogram = Metrics.newHistogram(factory.createMetricName("LiveScannedHistogram"));
     }
 
     public void updateSSTableIterated(int count)
@@ -331,6 +337,8 @@ public class ColumnFamilyMetrics
         Metrics.defaultRegistry().removeMetric(factory.createMetricName("RecentBloomFilterFalseRatio"));
         Metrics.defaultRegistry().removeMetric(factory.createMetricName("BloomFilterDiskSpaceUsed"));
         Metrics.defaultRegistry().removeMetric(factory.createMetricName("KeyCacheHitRate"));
+        Metrics.defaultRegistry().removeMetric(factory.createMetricName("TombstoneScannedHistogram"));
+        Metrics.defaultRegistry().removeMetric(factory.createMetricName("LiveScannedHistogram"));
     }
 
     class ColumnFamilyMetricNameFactory implements MetricNameFactory
diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java
index 7967deec73..50719fd45a 100644
--- a/src/java/org/apache/cassandra/service/StorageService.java
+++ b/src/java/org/apache/cassandra/service/StorageService.java
@@ -3934,4 +3934,14 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
     {
         return DatabaseDescriptor.getPartitionerName();
     }
+
+    public int getTombstoneDebugThreshold()
+    {
+        return DatabaseDescriptor.getTombstoneDebugThreshold();
+    }
+
+    public void setTombstoneDebugThreshold(int tombstoneDebugThreshold)
+    {
+        DatabaseDescriptor.setTombstoneDebugThreshold(tombstoneDebugThreshold);
+    }
 }
diff --git a/src/java/org/apache/cassandra/service/StorageServiceMBean.java b/src/java/org/apache/cassandra/service/StorageServiceMBean.java
index 4fbed9c0ec..63add12d0b 100644
--- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java
+++ b/src/java/org/apache/cassandra/service/StorageServiceMBean.java
@@ -476,4 +476,9 @@ public interface StorageServiceMBean extends NotificationEmitter
     public String getClusterName();
     /** Returns the cluster partitioner */
     public String getPartitionerName();
+
+    /** Returns the threshold for returning debugging queries with many tombstones */
+    public int getTombstoneDebugThreshold();
+    /** Sets the threshold for returning debugging queries with many tombstones */
+    public void setTombstoneDebugThreshold(int tombstoneDebugThreshold);
 }