cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jbel...@apache.org
Subject git commit: Track tombstone expiration patch by yukim; reviewed by jbellis for CASSANDRA-3442
Date Wed, 07 Mar 2012 21:34:40 GMT
Updated Branches:
  refs/heads/trunk 29804fa9a -> 6b2ea2647


Track tombstone expiration
patch by yukim; reviewed by jbellis for CASSANDRA-3442


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/6b2ea264
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/6b2ea264
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/6b2ea264

Branch: refs/heads/trunk
Commit: 6b2ea264702f80518a1147ab0aac44e0cf40dfc3
Parents: 29804fa
Author: Jonathan Ellis <jbellis@apache.org>
Authored: Wed Mar 7 15:21:11 2012 -0600
Committer: Jonathan Ellis <jbellis@apache.org>
Committed: Wed Mar 7 15:34:28 2012 -0600

----------------------------------------------------------------------
 CHANGES.txt                                        |    5 +
 src/java/org/apache/cassandra/db/ColumnFamily.java |   22 ++-
 .../cassandra/db/ColumnFamilySerializer.java       |   13 +-
 src/java/org/apache/cassandra/db/EchoedRow.java    |   12 +-
 .../db/compaction/AbstractCompactedRow.java        |   13 +-
 .../db/compaction/AbstractCompactionTask.java      |   12 +-
 .../cassandra/db/compaction/CompactionTask.java    |    8 -
 .../db/compaction/LazilyCompactedRow.java          |   37 ++--
 .../cassandra/db/compaction/PrecompactedRow.java   |   11 +-
 .../compaction/SizeTieredCompactionStrategy.java   |   25 ++-
 .../apache/cassandra/io/sstable/ColumnStats.java   |   42 +++
 .../apache/cassandra/io/sstable/Descriptor.java    |    5 +-
 .../org/apache/cassandra/io/sstable/SSTable.java   |    2 +
 .../cassandra/io/sstable/SSTableMetadata.java      |   62 ++++-
 .../apache/cassandra/io/sstable/SSTableReader.java |    5 +
 .../apache/cassandra/io/sstable/SSTableWriter.java |   32 +--
 .../cassandra/streaming/IncomingStreamReader.java  |    2 -
 .../apache/cassandra/utils/StreamingHistogram.java |  199 +++++++++++++++
 .../cassandra/db/compaction/CompactionsTest.java   |   62 ++++-
 .../cassandra/utils/StreamingHistogramTest.java    |  123 +++++++++
 20 files changed, 595 insertions(+), 97 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/6b2ea264/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 898e140..16f8f0c 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,3 +1,8 @@
+1.2-dev
+ * Track tombstone expiration and compact when tombstone content is
+   higher than a configurable threshold, default 20% (CASSANDRA-3442)
+
+
 1.1.1-dev
  * optimize commitlog checksumming (CASSANDRA-3610)
  * identify and blacklist corrupted SSTables from future compactions (CASSANDRA-2261)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6b2ea264/src/java/org/apache/cassandra/db/ColumnFamily.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/ColumnFamily.java b/src/java/org/apache/cassandra/db/ColumnFamily.java
index dcf52f4..f7398ee 100644
--- a/src/java/org/apache/cassandra/db/ColumnFamily.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamily.java
@@ -22,6 +22,8 @@ import static org.apache.cassandra.db.DBConstants.*;
 import java.nio.ByteBuffer;
 import java.security.MessageDigest;
 
+import org.apache.cassandra.io.sstable.SSTable;
+import org.apache.cassandra.utils.*;
 import org.apache.commons.lang.builder.HashCodeBuilder;
 
 import org.apache.cassandra.cache.IRowCacheEntry;
@@ -31,10 +33,7 @@ import org.apache.cassandra.db.filter.QueryPath;
 import org.apache.cassandra.db.marshal.AbstractType;
 import org.apache.cassandra.db.marshal.MarshalException;
 import org.apache.cassandra.io.IColumnSerializer;
-import org.apache.cassandra.utils.Allocator;
-import org.apache.cassandra.utils.ByteBufferUtil;
-import org.apache.cassandra.utils.FBUtilities;
-import org.apache.cassandra.utils.HeapAllocator;
+import org.apache.cassandra.io.sstable.ColumnStats;
 
 public class ColumnFamily extends AbstractColumnContainer implements IRowCacheEntry
 {
@@ -364,4 +363,19 @@ public class ColumnFamily extends AbstractColumnContainer implements IRowCacheEn
             column.validateFields(metadata);
         }
     }
+
+    public ColumnStats getColumnStats()
+    {
+        long maxTimestampSeen = Long.MIN_VALUE;
+        StreamingHistogram tombstones = new StreamingHistogram(SSTable.TOMBSTONE_HISTOGRAM_BIN_SIZE);
+
+        for (IColumn column : columns)
+        {
+            maxTimestampSeen = Math.max(maxTimestampSeen, column.maxTimestamp());
+            int deletionTime = column.getLocalDeletionTime();
+            if (deletionTime < Integer.MAX_VALUE)
+                tombstones.update(deletionTime);
+        }
+        return new ColumnStats(getColumnCount(), maxTimestampSeen, tombstones);
+    }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6b2ea264/src/java/org/apache/cassandra/db/ColumnFamilySerializer.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/ColumnFamilySerializer.java b/src/java/org/apache/cassandra/db/ColumnFamilySerializer.java
index 0f6165a..31c8d4b 100644
--- a/src/java/org/apache/cassandra/db/ColumnFamilySerializer.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilySerializer.java
@@ -28,6 +28,7 @@ import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.io.IColumnSerializer;
 import org.apache.cassandra.io.ISerializer;
+import org.apache.cassandra.io.sstable.ColumnStats;
 
 public class ColumnFamilySerializer implements ISerializer<ColumnFamily>
 {
@@ -70,7 +71,7 @@ public class ColumnFamilySerializer implements ISerializer<ColumnFamily>
         serializeForSSTable(columnFamily, dos);
     }
 
-    public int serializeForSSTable(ColumnFamily columnFamily, DataOutput dos)
+    public void serializeForSSTable(ColumnFamily columnFamily, DataOutput dos)
     {
         try
         {
@@ -79,14 +80,8 @@ public class ColumnFamilySerializer implements ISerializer<ColumnFamily>
             Collection<IColumn> columns = columnFamily.getSortedColumns();
             int count = columns.size();
             dos.writeInt(count);
-            int i = 0;
             for (IColumn column : columns)
-            {
                 columnFamily.getColumnSerializer().serialize(column, dos);
-                i++;
-            }
-            assert count == i: "CF size changed during serialization: was " + count + " initially but " + i + " written";
-            return count;
         }
         catch (IOException e)
         {
@@ -100,10 +95,10 @@ public class ColumnFamilySerializer implements ISerializer<ColumnFamily>
         dos.writeLong(columnFamily.getMarkedForDeleteAt());
     }
 
-    public int serializeWithIndexes(ColumnFamily columnFamily, ColumnIndexer.RowHeader index, DataOutput dos)
+    public void serializeWithIndexes(ColumnFamily columnFamily, ColumnIndexer.RowHeader index, DataOutput dos)
     {
         ColumnIndexer.serialize(index, dos);
-        return serializeForSSTable(columnFamily, dos);
+        serializeForSSTable(columnFamily, dos);
     }
 
     public ColumnFamily deserialize(DataInput dis) throws IOException

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6b2ea264/src/java/org/apache/cassandra/db/EchoedRow.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/EchoedRow.java b/src/java/org/apache/cassandra/db/EchoedRow.java
index 9067a10..69ec3d7 100644
--- a/src/java/org/apache/cassandra/db/EchoedRow.java
+++ b/src/java/org/apache/cassandra/db/EchoedRow.java
@@ -22,7 +22,10 @@ import java.io.IOException;
 import java.security.MessageDigest;
 
 import org.apache.cassandra.db.compaction.AbstractCompactedRow;
+import org.apache.cassandra.io.sstable.ColumnStats;
+import org.apache.cassandra.io.sstable.SSTable;
 import org.apache.cassandra.io.sstable.SSTableIdentityIterator;
+import org.apache.cassandra.utils.StreamingHistogram;
 
 /**
  * A CompactedRow implementation that just echos the original row bytes without deserializing.
@@ -60,13 +63,8 @@ public class EchoedRow extends AbstractCompactedRow
         return false;
     }
 
-    public int columnCount()
+    public ColumnStats columnStats()
     {
-        return row.getColumnCount();
-    }
-
-    public long maxTimestamp()
-    {
-        return Long.MIN_VALUE;
+        return new ColumnStats(row.getColumnCount(), Long.MIN_VALUE, new StreamingHistogram(SSTable.TOMBSTONE_HISTOGRAM_BIN_SIZE));
     }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6b2ea264/src/java/org/apache/cassandra/db/compaction/AbstractCompactedRow.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/AbstractCompactedRow.java b/src/java/org/apache/cassandra/db/compaction/AbstractCompactedRow.java
index f04e7c1..5a892c1 100644
--- a/src/java/org/apache/cassandra/db/compaction/AbstractCompactedRow.java
+++ b/src/java/org/apache/cassandra/db/compaction/AbstractCompactedRow.java
@@ -22,6 +22,7 @@ import java.io.IOException;
 import java.security.MessageDigest;
 
 import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.io.sstable.ColumnStats;
 
 /**
  * a CompactedRow is an object that takes a bunch of rows (keys + columnfamilies)
@@ -60,14 +61,8 @@ public abstract class AbstractCompactedRow
     public abstract boolean isEmpty();
 
     /**
-     * @return the number of columns in the row
+     * @return aggregate information about the columns in this row.  Some fields may
+     * contain default values if computing them value would require extra effort we're not willing to make.
      */
-    public abstract int columnCount();
-
-    /**
-     * @return the max column timestamp in the row or Long.MIN_VALUE if
-     * computing this value would require extra effort we're not willing to
-     * make.
-     */
-    public abstract long maxTimestamp();
+    public abstract ColumnStats columnStats();
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6b2ea264/src/java/org/apache/cassandra/db/compaction/AbstractCompactionTask.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/AbstractCompactionTask.java b/src/java/org/apache/cassandra/db/compaction/AbstractCompactionTask.java
index e39a533..e031e07 100644
--- a/src/java/org/apache/cassandra/db/compaction/AbstractCompactionTask.java
+++ b/src/java/org/apache/cassandra/db/compaction/AbstractCompactionTask.java
@@ -29,11 +29,13 @@ public abstract class AbstractCompactionTask
 {
     protected final ColumnFamilyStore cfs;
     protected Collection<SSTableReader> sstables;
+    protected boolean isUserDefined;
 
     public AbstractCompactionTask(ColumnFamilyStore cfs, Collection<SSTableReader> sstables)
     {
         this.cfs = cfs;
         this.sstables = sstables;
+        this.isUserDefined = false;
     }
 
     public abstract int execute(CompactionExecutorStatsCollector collector) throws IOException;
@@ -58,7 +60,9 @@ public abstract class AbstractCompactionTask
      */
     public boolean markSSTablesForCompaction()
     {
-        return markSSTablesForCompaction(cfs.getMinimumCompactionThreshold(), cfs.getMaximumCompactionThreshold());
+        int min = isUserDefined ? 1 : cfs.getMinimumCompactionThreshold();
+        int max = isUserDefined ? Integer.MAX_VALUE : cfs.getMaximumCompactionThreshold();
+        return markSSTablesForCompaction(min, max);
     }
 
     public boolean markSSTablesForCompaction(int min, int max)
@@ -83,4 +87,10 @@ public abstract class AbstractCompactionTask
     // Can be overriden for action that need to be performed if the task won't
     // execute (if sstable can't be marked successfully)
     protected void cancel() {}
+
+    public AbstractCompactionTask isUserDefined(boolean isUserDefined)
+    {
+        this.isUserDefined = isUserDefined;
+        return this;
+    }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6b2ea264/src/java/org/apache/cassandra/db/compaction/CompactionTask.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionTask.java b/src/java/org/apache/cassandra/db/compaction/CompactionTask.java
index f8d123e..4d77af0 100644
--- a/src/java/org/apache/cassandra/db/compaction/CompactionTask.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionTask.java
@@ -42,14 +42,12 @@ public class CompactionTask extends AbstractCompactionTask
 {
     protected static final Logger logger = LoggerFactory.getLogger(CompactionTask.class);
     protected final int gcBefore;
-    protected boolean isUserDefined;
     protected static long totalBytesCompacted = 0;
 
     public CompactionTask(ColumnFamilyStore cfs, Collection<SSTableReader> sstables, final int gcBefore)
     {
         super(cfs, sstables);
         this.gcBefore = gcBefore;
-        this.isUserDefined = false;
     }
 
     public static synchronized long addToTotalBytesCompacted(long bytesCompacted)
@@ -262,10 +260,4 @@ public class CompactionTask extends AbstractCompactionTask
         }
         return max;
     }
-
-    public CompactionTask isUserDefined(boolean isUserDefined)
-    {
-        this.isUserDefined = isUserDefined;
-        return this;
-    }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6b2ea264/src/java/org/apache/cassandra/db/compaction/LazilyCompactedRow.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/LazilyCompactedRow.java b/src/java/org/apache/cassandra/db/compaction/LazilyCompactedRow.java
index 61546f3..eaf401a 100644
--- a/src/java/org/apache/cassandra/db/compaction/LazilyCompactedRow.java
+++ b/src/java/org/apache/cassandra/db/compaction/LazilyCompactedRow.java
@@ -26,6 +26,9 @@ import java.util.List;
 
 import com.google.common.base.Predicates;
 import com.google.common.collect.Iterators;
+import org.apache.cassandra.io.sstable.ColumnStats;
+import org.apache.cassandra.io.sstable.SSTable;
+import org.apache.cassandra.utils.StreamingHistogram;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -58,8 +61,7 @@ public class LazilyCompactedRow extends AbstractCompactedRow implements IIterabl
     private final DataOutputBuffer headerBuffer;
     private ColumnFamily emptyColumnFamily;
     private Reducer reducer;
-    private int columnCount;
-    private long maxTimestamp;
+    private final ColumnStats columnStats;
     private long columnSerializedSize;
     private boolean closed;
 
@@ -85,9 +87,10 @@ public class LazilyCompactedRow extends AbstractCompactedRow implements IIterabl
         ColumnIndexer.serialize(this, headerBuffer);
         // reach into the reducer used during iteration to get column count, size, max column timestamp
         // (however, if there are zero columns, iterator() will not be called by ColumnIndexer and reducer will be null)
-        columnCount = reducer == null ? 0 : reducer.size;
+        columnStats = new ColumnStats(reducer == null ? 0 : reducer.columns, reducer == null ? Long.MIN_VALUE : reducer.maxTimestampSeen,
+                                      reducer == null ? new StreamingHistogram(SSTable.TOMBSTONE_HISTOGRAM_BIN_SIZE) : reducer.tombstones
+        );
         columnSerializedSize = reducer == null ? 0 : reducer.serializedSize;
-        maxTimestamp = reducer == null ? Long.MIN_VALUE : reducer.maxTimestampSeen;
         reducer = null;
     }
 
@@ -106,7 +109,7 @@ public class LazilyCompactedRow extends AbstractCompactedRow implements IIterabl
         out.writeLong(dataSize);
         out.write(headerBuffer.getData(), 0, headerBuffer.getLength());
         out.write(clockOut.getData(), 0, clockOut.getLength());
-        out.writeInt(columnCount);
+        out.writeInt(columnStats.columnCount);
 
         Iterator<IColumn> iter = iterator();
         while (iter.hasNext())
@@ -133,7 +136,7 @@ public class LazilyCompactedRow extends AbstractCompactedRow implements IIterabl
         try
         {
             ColumnFamily.serializer().serializeCFInfo(emptyColumnFamily, out);
-            out.writeInt(columnCount);
+            out.writeInt(columnStats.columnCount);
             digest.update(out.getData(), 0, out.getLength());
         }
         catch (IOException e)
@@ -154,7 +157,7 @@ public class LazilyCompactedRow extends AbstractCompactedRow implements IIterabl
         boolean cfIrrelevant = shouldPurge
                              ? ColumnFamilyStore.removeDeletedCF(emptyColumnFamily, controller.gcBefore) == null
                              : !emptyColumnFamily.isMarkedForDelete(); // tombstones are relevant
-        return cfIrrelevant && columnCount == 0;
+        return cfIrrelevant && columnStats.columnCount == 0;
     }
 
     public int getEstimatedColumnCount()
@@ -179,14 +182,9 @@ public class LazilyCompactedRow extends AbstractCompactedRow implements IIterabl
         return Iterators.filter(iter, Predicates.notNull());
     }
 
-    public int columnCount()
+    public ColumnStats columnStats()
     {
-        return columnCount;
-    }
-
-    public long maxTimestamp()
-    {
-        return maxTimestamp;
+        return columnStats;
     }
 
     private void close()
@@ -209,8 +207,9 @@ public class LazilyCompactedRow extends AbstractCompactedRow implements IIterabl
     {
         ColumnFamily container = emptyColumnFamily.cloneMeShallow();
         long serializedSize = 4; // int for column count
-        int size = 0;
+        int columns = 0;
         long maxTimestampSeen = Long.MIN_VALUE;
+        StreamingHistogram tombstones = new StreamingHistogram(SSTable.TOMBSTONE_HISTOGRAM_BIN_SIZE);
 
         public void reduce(IColumn current)
         {
@@ -227,9 +226,15 @@ public class LazilyCompactedRow extends AbstractCompactedRow implements IIterabl
             }
             IColumn reduced = purged.iterator().next();
             container.clear();
+
             serializedSize += reduced.serializedSize();
-            size++;
+            columns++;
             maxTimestampSeen = Math.max(maxTimestampSeen, reduced.maxTimestamp());
+            int deletionTime = reduced.getLocalDeletionTime();
+            if (deletionTime < Integer.MAX_VALUE)
+            {
+                tombstones.update(deletionTime);
+            }
             return reduced;
         }
     }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6b2ea264/src/java/org/apache/cassandra/db/compaction/PrecompactedRow.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/PrecompactedRow.java b/src/java/org/apache/cassandra/db/compaction/PrecompactedRow.java
index 8522ea6..35cd33d 100644
--- a/src/java/org/apache/cassandra/db/compaction/PrecompactedRow.java
+++ b/src/java/org/apache/cassandra/db/compaction/PrecompactedRow.java
@@ -23,6 +23,7 @@ import java.io.IOException;
 import java.security.MessageDigest;
 import java.util.List;
 
+import org.apache.cassandra.io.sstable.ColumnStats;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -65,6 +66,7 @@ public class PrecompactedRow extends AbstractCompactedRow
 
         if (cf.hasExpiredTombstones(controller.gcBefore))
             shouldPurge = controller.shouldPurge(key);
+
         // We should only gc tombstone if shouldPurge == true. But otherwise,
         // it is still ok to collect column that shadowed by their (deleted)
         // container, which removeDeleted(cf, Integer.MAX_VALUE) will do
@@ -161,14 +163,9 @@ public class PrecompactedRow extends AbstractCompactedRow
         return compactedCf == null;
     }
 
-    public int columnCount()
-    {
-        return compactedCf == null ? 0 : compactedCf.getColumnCount();
-    }
-
-    public long maxTimestamp()
+    public ColumnStats columnStats()
     {
-        return compactedCf.maxTimestamp();
+        return compactedCf.getColumnStats();
     }
 
     /**

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6b2ea264/src/java/org/apache/cassandra/db/compaction/SizeTieredCompactionStrategy.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/SizeTieredCompactionStrategy.java b/src/java/org/apache/cassandra/db/compaction/SizeTieredCompactionStrategy.java
index 97396e4..d6f4025 100644
--- a/src/java/org/apache/cassandra/db/compaction/SizeTieredCompactionStrategy.java
+++ b/src/java/org/apache/cassandra/db/compaction/SizeTieredCompactionStrategy.java
@@ -32,9 +32,12 @@ public class SizeTieredCompactionStrategy extends AbstractCompactionStrategy
 {
     private static final Logger logger = LoggerFactory.getLogger(SizeTieredCompactionStrategy.class);
     protected static final long DEFAULT_MIN_SSTABLE_SIZE = 50L * 1024L * 1024L;
+    protected static final float DEFAULT_TOMBSTONE_THRESHOLD = 0.2f;
     protected static final String MIN_SSTABLE_SIZE_KEY = "min_sstable_size";
+    protected static final String TOMBSTONE_THRESHOLD_KEY = "tombstone_threshold";
     protected long minSSTableSize;
     protected volatile int estimatedRemainingTasks;
+    protected float tombstoneThreshold;
 
     public SizeTieredCompactionStrategy(ColumnFamilyStore cfs, Map<String, String> options)
     {
@@ -44,6 +47,8 @@ public class SizeTieredCompactionStrategy extends AbstractCompactionStrategy
         minSSTableSize = (null != optionValue) ? Long.parseLong(optionValue) : DEFAULT_MIN_SSTABLE_SIZE;
         cfs.setMaximumCompactionThreshold(cfs.metadata.getMaxCompactionThreshold());
         cfs.setMinimumCompactionThreshold(cfs.metadata.getMinCompactionThreshold());
+        optionValue = options.get(TOMBSTONE_THRESHOLD_KEY);
+        tombstoneThreshold = (null != optionValue) ? Float.parseFloat(optionValue) : DEFAULT_TOMBSTONE_THRESHOLD;
     }
 
     public AbstractCompactionTask getNextBackgroundTask(final int gcBefore)
@@ -75,7 +80,21 @@ public class SizeTieredCompactionStrategy extends AbstractCompactionStrategy
         }
 
         if (prunedBuckets.isEmpty())
-            return null;
+        {
+            // if there is no sstable to compact in standard way, try compacting single sstable whose droppable tombstone
+            // ratio is greater than threshold.
+            for (List<SSTableReader> bucket : buckets)
+            {
+                for (SSTableReader table : bucket)
+                {
+                    if (table.getEstimatedDroppableTombstoneRatio(gcBefore) > tombstoneThreshold)
+                        prunedBuckets.add(Collections.singletonList(table));
+                }
+            }
+
+            if (prunedBuckets.isEmpty())
+                return null;
+        }
 
         List<SSTableReader> smallestBucket = Collections.min(prunedBuckets, new Comparator<List<SSTableReader>>()
         {
@@ -97,7 +116,8 @@ public class SizeTieredCompactionStrategy extends AbstractCompactionStrategy
                 return n / sstables.size();
             }
         });
-        return new CompactionTask(cfs, smallestBucket, gcBefore);
+        // when bucket only contains just one sstable, set userDefined to true to force single sstable compaction
+        return new CompactionTask(cfs, smallestBucket, gcBefore).isUserDefined(smallestBucket.size() == 1);
     }
 
     public AbstractCompactionTask getMaximalTask(final int gcBefore)
@@ -174,7 +194,6 @@ public class SizeTieredCompactionStrategy extends AbstractCompactionStrategy
                 buckets.put(bucket, size);
             }
         }
-
         return new LinkedList<List<T>>(buckets.keySet());
     }
 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6b2ea264/src/java/org/apache/cassandra/io/sstable/ColumnStats.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/sstable/ColumnStats.java b/src/java/org/apache/cassandra/io/sstable/ColumnStats.java
new file mode 100644
index 0000000..a7dcfec
--- /dev/null
+++ b/src/java/org/apache/cassandra/io/sstable/ColumnStats.java
@@ -0,0 +1,42 @@
+/*
+ * 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.io.sstable;
+
+import org.apache.cassandra.utils.StreamingHistogram;
+
+/**
+ * ColumnStats holds information about the columns for one row inside sstable
+ */
+public class ColumnStats
+{
+    /** how many columns are there in the row */
+    public final int columnCount;
+
+    /** the largest (client-supplied) timestamp in the row */
+    public final long maxTimestamp;
+
+    /** histogram of tombstone drop time */
+    public final StreamingHistogram tombstoneHistogram;
+
+    public ColumnStats(int columnCount, long maxTimestamp, StreamingHistogram tombstoneHistogram)
+    {
+        this.maxTimestamp = maxTimestamp;
+        this.columnCount = columnCount;
+        this.tombstoneHistogram = tombstoneHistogram;
+    }
+}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6b2ea264/src/java/org/apache/cassandra/io/sstable/Descriptor.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/sstable/Descriptor.java b/src/java/org/apache/cassandra/io/sstable/Descriptor.java
index a4ff975..734d742 100644
--- a/src/java/org/apache/cassandra/io/sstable/Descriptor.java
+++ b/src/java/org/apache/cassandra/io/sstable/Descriptor.java
@@ -53,7 +53,8 @@ public class Descriptor
     // h (1.0): tracks max client timestamp in metadata component
     // hb (1.0.3): records compression ration in metadata component
     // hc (1.0.4): records partitioner in metadata component
-    public static final String CURRENT_VERSION = "hc";
+    // hd (1.2): records estimated histogram of deletion times in tombstones
+    public static final String CURRENT_VERSION = "hd";
 
     public final File directory;
     /** version has the following format: <code>[a-z]+</code> */
@@ -73,6 +74,7 @@ public class Descriptor
     public final boolean tracksMaxTimestamp;
     public final boolean hasCompressionRatio;
     public final boolean hasPartitioner;
+    public final boolean tracksTombstones;
 
     /**
      * A descriptor that assumes CURRENT_VERSION.
@@ -101,6 +103,7 @@ public class Descriptor
         tracksMaxTimestamp = version.compareTo("h") >= 0;
         hasCompressionRatio = version.compareTo("hb") >= 0;
         hasPartitioner = version.compareTo("hc") >= 0;
+        tracksTombstones = version.compareTo("hd") >= 0;
         isLatestVersion = version.compareTo(CURRENT_VERSION) == 0;
     }
 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6b2ea264/src/java/org/apache/cassandra/io/sstable/SSTable.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/sstable/SSTable.java b/src/java/org/apache/cassandra/io/sstable/SSTable.java
index 5ac3298..074c7fd 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTable.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTable.java
@@ -61,6 +61,8 @@ public abstract class SSTable
 
     public static final String TEMPFILE_MARKER = "tmp";
 
+    public static final int TOMBSTONE_HISTOGRAM_BIN_SIZE = 100;
+
     public static final Comparator<SSTableReader> maxTimestampComparator = new Comparator<SSTableReader>()
     {
         public int compare(SSTableReader o1, SSTableReader o2)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6b2ea264/src/java/org/apache/cassandra/io/sstable/SSTableMetadata.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableMetadata.java b/src/java/org/apache/cassandra/io/sstable/SSTableMetadata.java
index d1d514f..42a6c73 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTableMetadata.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableMetadata.java
@@ -24,6 +24,7 @@ import java.io.File;
 import java.io.FileInputStream;
 import java.io.IOException;
 
+import org.apache.cassandra.utils.StreamingHistogram;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -40,6 +41,7 @@ import org.apache.cassandra.utils.EstimatedHistogram;
  *  - max column timestamp
  *  - compression ratio
  *  - partitioner
+ *  - tombstone drop time histogram
  *
  * An SSTableMetadata should be instantiated via the Collector, openFromDescriptor()
  * or createDefaultInstance()
@@ -56,6 +58,7 @@ public class SSTableMetadata
     public final long maxTimestamp;
     public final double compressionRatio;
     public final String partitioner;
+    public final StreamingHistogram estimatedTombstoneDropTime;
 
     private SSTableMetadata()
     {
@@ -64,10 +67,12 @@ public class SSTableMetadata
              ReplayPosition.NONE,
              Long.MIN_VALUE,
              Double.MIN_VALUE,
-             null);
+             null,
+             defaultTombstoneDropTimeHistogram());
     }
 
-    private SSTableMetadata(EstimatedHistogram rowSizes, EstimatedHistogram columnCounts, ReplayPosition replayPosition, long maxTimestamp, double cr, String partitioner)
+    private SSTableMetadata(EstimatedHistogram rowSizes, EstimatedHistogram columnCounts, ReplayPosition replayPosition, long maxTimestamp,
+                            double cr, String partitioner, StreamingHistogram estimatedTombstoneDropTime)
     {
         this.estimatedRowSize = rowSizes;
         this.estimatedColumnCount = columnCounts;
@@ -75,6 +80,7 @@ public class SSTableMetadata
         this.maxTimestamp = maxTimestamp;
         this.compressionRatio = cr;
         this.partitioner = partitioner;
+        this.estimatedTombstoneDropTime = estimatedTombstoneDropTime;
     }
 
     public static SSTableMetadata createDefaultInstance()
@@ -99,6 +105,26 @@ public class SSTableMetadata
         return new EstimatedHistogram(150);
     }
 
+    static StreamingHistogram defaultTombstoneDropTimeHistogram()
+    {
+        return new StreamingHistogram(SSTable.TOMBSTONE_HISTOGRAM_BIN_SIZE);
+    }
+
+    /**
+     * @param gcBefore
+     * @return estimated droppable tombstone ratio at given gcBefore time.
+     */
+    public double getEstimatedDroppableTombstoneRatio(int gcBefore)
+    {
+        long estimatedColumnCount = this.estimatedColumnCount.mean() * this.estimatedColumnCount.count();
+        if (estimatedColumnCount > 0)
+        {
+            double droppable = estimatedTombstoneDropTime.sum(gcBefore);
+            return droppable / estimatedColumnCount;
+        }
+        return 0.0f;
+    }
+
     public static class Collector
     {
         protected EstimatedHistogram estimatedRowSize = defaultRowSizeHistogram();
@@ -106,6 +132,7 @@ public class SSTableMetadata
         protected ReplayPosition replayPosition = ReplayPosition.NONE;
         protected long maxTimestamp = Long.MIN_VALUE;
         protected double compressionRatio = Double.MIN_VALUE;
+        protected StreamingHistogram estimatedTombstoneDropTime = defaultTombstoneDropTimeHistogram();
 
         public void addRowSize(long rowSize)
         {
@@ -117,6 +144,11 @@ public class SSTableMetadata
             estimatedColumnCount.add(columnCount);
         }
 
+        public void mergeTombstoneHistogram(StreamingHistogram histogram)
+        {
+            estimatedTombstoneDropTime.merge(histogram);
+        }
+
         /**
          * Ratio is compressed/uncompressed and it is
          * if you have 1.x then compression isn't helping
@@ -138,7 +170,8 @@ public class SSTableMetadata
                                        replayPosition,
                                        maxTimestamp,
                                        compressionRatio,
-                                       partitioner);
+                                       partitioner,
+                                       estimatedTombstoneDropTime);
         }
 
         public Collector estimatedRowSize(EstimatedHistogram estimatedRowSize)
@@ -158,6 +191,21 @@ public class SSTableMetadata
             this.replayPosition = replayPosition;
             return this;
         }
+
+        void update(long size, ColumnStats stats)
+        {
+            /*
+             * The max timestamp is not always collected here (more precisely, row.maxTimestamp() may return Long.MIN_VALUE),
+             * to avoid deserializing an EchoedRow.
+             * This is the reason why it is collected first when calling ColumnFamilyStore.createCompactionWriter
+             * However, for old sstables without timestamp, we still want to update the timestamp (and we know
+             * that in this case we will not use EchoedRow, since CompactionControler.needsDeserialize() will be true).
+            */
+            updateMaxTimestamp(stats.maxTimestamp);
+            addRowSize(size);
+            addColumnCount(stats.columnCount);
+            mergeTombstoneHistogram(stats.tombstoneHistogram);
+        }
     }
 
     public static class SSTableMetadataSerializer
@@ -174,6 +222,7 @@ public class SSTableMetadata
             dos.writeLong(sstableStats.maxTimestamp);
             dos.writeDouble(sstableStats.compressionRatio);
             dos.writeUTF(sstableStats.partitioner);
+            StreamingHistogram.serializer.serialize(sstableStats.estimatedTombstoneDropTime, dos);
         }
 
         public SSTableMetadata deserialize(Descriptor descriptor) throws IOException
@@ -207,9 +256,12 @@ public class SSTableMetadata
             long maxTimestamp = desc.tracksMaxTimestamp ? dis.readLong() : Long.MIN_VALUE;
             double compressionRatio = desc.hasCompressionRatio
                                     ? dis.readDouble()
-                                    : Double.MIN_VALUE;
+                                              : Double.MIN_VALUE;
             String partitioner = desc.hasPartitioner ? dis.readUTF() : null;
-            return new SSTableMetadata(rowSizes, columnCounts, replayPosition, maxTimestamp, compressionRatio, partitioner);
+            StreamingHistogram tombstoneHistogram = desc.tracksTombstones
+                                                   ? StreamingHistogram.serializer.deserialize(dis)
+                                                   : defaultTombstoneDropTimeHistogram();
+            return new SSTableMetadata(rowSizes, columnCounts, replayPosition, maxTimestamp, compressionRatio, partitioner, tombstoneHistogram);
         }
     }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6b2ea264/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableReader.java b/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
index a33c89a..072b6f7 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
@@ -955,6 +955,11 @@ public class SSTableReader extends SSTable
         return sstableMetadata.estimatedColumnCount;
     }
 
+    public double getEstimatedDroppableTombstoneRatio(int gcBefore)
+    {
+        return sstableMetadata.getEstimatedDroppableTombstoneRatio(gcBefore);
+    }
+
     public double getCompressionRatio()
     {
         return sstableMetadata.compressionRatio;

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6b2ea264/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java b/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java
index b146d81..e07e151 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java
@@ -158,16 +158,8 @@ public class SSTableWriter extends SSTable
         long dataSize = row.write(dataFile.stream);
         assert dataSize == dataFile.getFilePointer() - (dataStart + 8)
                 : "incorrect row data size " + dataSize + " written to " + dataFile.getPath() + "; correct is " + (dataFile.getFilePointer() - (dataStart + 8));
-        /*
-         * The max timestamp is not always collected here (more precisely, row.maxTimestamp() may return Long.MIN_VALUE),
-         * to avoid deserializing an EchoedRow.
-         * This is the reason why it is collected first when calling ColumnFamilyStore.createCompactionWriter
-         * However, for old sstables without timestamp, we still want to update the timestamp (and we know
-         * that in this case we will not use EchoedRow, since CompactionControler.needsDeserialize() will be true).
-        */
-        sstableMetadataCollector.updateMaxTimestamp(row.maxTimestamp());
-        sstableMetadataCollector.addRowSize(dataFile.getFilePointer() - currentPosition);
-        sstableMetadataCollector.addColumnCount(row.columnCount());
+
+        sstableMetadataCollector.update(dataFile.getFilePointer() - currentPosition, row.columnStats());
         afterAppend(row.key, currentPosition);
         return currentPosition;
     }
@@ -184,13 +176,10 @@ public class SSTableWriter extends SSTable
         dataFile.stream.writeLong(header.serializedSize() + cf.serializedSizeForSSTable());
 
         // write out row header and data
-        int columnCount = ColumnFamily.serializer().serializeWithIndexes(cf, header, dataFile.stream);
+        ColumnFamily.serializer().serializeWithIndexes(cf, header, dataFile.stream);
         afterAppend(decoratedKey, startPosition);
 
-        // track max column timestamp
-        sstableMetadataCollector.updateMaxTimestamp(cf.maxTimestamp());
-        sstableMetadataCollector.addRowSize(dataFile.getFilePointer() - startPosition);
-        sstableMetadataCollector.addColumnCount(columnCount);
+        sstableMetadataCollector.update(dataFile.getFilePointer() - startPosition, cf.getColumnStats());
     }
 
     public void append(DecoratedKey decoratedKey, ByteBuffer value) throws IOException
@@ -234,6 +223,7 @@ public class SSTableWriter extends SSTable
 
         // deserialize each column to obtain maxTimestamp and immediately serialize it.
         long maxTimestamp = Long.MIN_VALUE;
+        StreamingHistogram tombstones = new StreamingHistogram(TOMBSTONE_HISTOGRAM_BIN_SIZE);
         ColumnFamily cf = ColumnFamily.create(metadata, ArrayBackedSortedColumns.factory());
         for (int i = 0; i < columnCount; i++)
         {
@@ -256,6 +246,12 @@ public class SSTableWriter extends SSTable
                     }
                 }
             }
+
+            int deletionTime = column.getLocalDeletionTime();
+            if (deletionTime < Integer.MAX_VALUE)
+            {
+                tombstones.update(deletionTime);
+            }
             maxTimestamp = Math.max(maxTimestamp, column.maxTimestamp());
             cf.getColumnSerializer().serialize(column, dataFile.stream);
         }
@@ -265,15 +261,11 @@ public class SSTableWriter extends SSTable
         sstableMetadataCollector.updateMaxTimestamp(maxTimestamp);
         sstableMetadataCollector.addRowSize(dataFile.getFilePointer() - currentPosition);
         sstableMetadataCollector.addColumnCount(columnCount);
+        sstableMetadataCollector.mergeTombstoneHistogram(tombstones);
         afterAppend(key, currentPosition);
         return currentPosition;
     }
 
-    public void updateMaxTimestamp(long timestamp)
-    {
-        sstableMetadataCollector.updateMaxTimestamp(timestamp);
-    }
-
     /**
      * After failure, attempt to close the index writer and data file before deleting all temp components for the sstable
      */

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6b2ea264/src/java/org/apache/cassandra/streaming/IncomingStreamReader.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/IncomingStreamReader.java b/src/java/org/apache/cassandra/streaming/IncomingStreamReader.java
index a42ff85..64ef2f8 100644
--- a/src/java/org/apache/cassandra/streaming/IncomingStreamReader.java
+++ b/src/java/org/apache/cassandra/streaming/IncomingStreamReader.java
@@ -132,8 +132,6 @@ public class IncomingStreamReader
                         // We don't expire anything so the row shouldn't be empty
                         assert !row.isEmpty();
                         writer.append(row);
-                        // row append does not update the max timestamp on its own
-                        writer.updateMaxTimestamp(row.maxTimestamp());
 
                         // update cache
                         ColumnFamily cf = row.getFullColumnFamily();

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6b2ea264/src/java/org/apache/cassandra/utils/StreamingHistogram.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/utils/StreamingHistogram.java b/src/java/org/apache/cassandra/utils/StreamingHistogram.java
new file mode 100644
index 0000000..a95c259
--- /dev/null
+++ b/src/java/org/apache/cassandra/utils/StreamingHistogram.java
@@ -0,0 +1,199 @@
+/*
+ * 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.utils;
+
+import org.apache.cassandra.db.DBConstants;
+import org.apache.cassandra.io.ISerializer;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.*;
+
+/**
+ * Histogram that can be constructed from streaming of data.
+ *
+ * The algorithm is taken from following paper:
+ * Yael Ben-Haim and Elad Tom-Tov, "A Streaming Parallel Decision Tree Algorithm" (2010)
+ * http://jmlr.csail.mit.edu/papers/volume11/ben-haim10a/ben-haim10a.pdf
+ */
+public class StreamingHistogram
+{
+    public static final StreamingHistogramSerializer serializer = new StreamingHistogramSerializer();
+
+    // TreeMap to hold bins of histogram.
+    private final TreeMap<Double, Long> bin;
+
+    // maximum bin size for this histogram
+    private final int maxBinSize;
+
+    /**
+     * Creates a new histogram with max bin size of maxBinSize
+     * @param maxBinSize maximum number of bins this histogram can have
+     */
+    public StreamingHistogram(int maxBinSize)
+    {
+        this.maxBinSize = maxBinSize;
+        bin = new TreeMap<Double, Long>();
+    }
+
+    private StreamingHistogram(int maxBinSize, Map<Double, Long> bin)
+    {
+        this.maxBinSize = maxBinSize;
+        this.bin = new TreeMap<Double, Long>(bin);
+    }
+
+    /**
+     * Adds new point p to this histogram.
+     * @param p
+     */
+    public void update(double p)
+    {
+        update(p, 1);
+    }
+
+    /**
+     * Adds new point p with value m to this histogram.
+     * @param p
+     * @param m
+     */
+    public void update(double p, long m)
+    {
+        Long mi = bin.get(p);
+        if (mi != null)
+        {
+            // we found the same p so increment that counter
+            bin.put(p, mi + m);
+        }
+        else
+        {
+            bin.put(p, m);
+            // if bin size exceeds maximum bin size then trim down to max size
+            while (bin.size() > maxBinSize)
+            {
+                // find points p1, p2 which have smallest difference
+                Iterator<Double> keys = bin.keySet().iterator();
+                double p1 = keys.next();
+                double p2 = keys.next();
+                double smallestDiff = p2 - p1;
+                double q1 = p1, q2 = p2;
+                while (keys.hasNext())
+                {
+                    p1 = p2;
+                    p2 = keys.next();
+                    double diff = p2 - p1;
+                    if (diff < smallestDiff)
+                    {
+                        smallestDiff = diff;
+                        q1 = p1;
+                        q2 = p2;
+                    }
+                }
+                // merge those two
+                long k1 = bin.remove(q1);
+                long k2 = bin.remove(q2);
+                bin.put((q1 * k1 + q2 * k2) / (k1 + k2), k1 + k2);
+            }
+        }
+    }
+
+    /**
+     * Merges given histogram with this histogram.
+     *
+     * @param other histogram to merge
+     */
+    public void merge(StreamingHistogram other)
+    {
+        if (other == null)
+            return;
+
+        for (Map.Entry<Double, Long> entry : other.getAsMap().entrySet())
+            update(entry.getKey(), entry.getValue());
+    }
+
+    /**
+     * Calculates estimated number of points in interval [-∞,b].
+     *
+     * @param b upper bound of a interval to calculate sum
+     * @return estimated number of points in a interval [-∞,b].
+     */
+    public double sum(double b)
+    {
+        double sum = 0;
+        // find the points pi, pnext which satisfy pi <= b < pnext
+        Map.Entry<Double, Long> pnext = bin.higherEntry(b);
+        if (pnext == null)
+        {
+            // if b is greater than any key in this histogram,
+            // just count all appearance and return
+            for (Long value : bin.values())
+                sum += value;
+        }
+        else
+        {
+            Map.Entry<Double, Long> pi = bin.floorEntry(b);
+            // calculate estimated count mb for point b
+            double weight = (b - pi.getKey()) / (pnext.getKey() - pi.getKey());
+            double mb = pi.getValue() + (pnext.getValue() - pi.getValue()) * weight;
+            sum += (pi.getValue() + mb) * weight / 2;
+
+            sum += pi.getValue() / 2;
+            for (Long value : bin.headMap(pi.getKey(), false).values())
+                sum += value;
+        }
+        return sum;
+    }
+
+    public Map<Double, Long> getAsMap()
+    {
+        return Collections.unmodifiableMap(bin);
+    }
+
+    public static class StreamingHistogramSerializer implements ISerializer<StreamingHistogram>
+    {
+        public void serialize(StreamingHistogram histogram, DataOutput dos) throws IOException
+        {
+            dos.writeInt(histogram.maxBinSize);
+            Map<Double, Long> entries = histogram.getAsMap();
+            dos.writeInt(entries.size());
+            for (Map.Entry<Double, Long> entry : entries.entrySet())
+            {
+                dos.writeDouble(entry.getKey());
+                dos.writeLong(entry.getValue());
+            }
+        }
+
+        public StreamingHistogram deserialize(DataInput dis) throws IOException
+        {
+            int maxBinSize = dis.readInt();
+            int size = dis.readInt();
+            Map<Double, Long> tmp = new HashMap<Double, Long>(size);
+            for (int i = 0; i < size; i++)
+            {
+                tmp.put(dis.readDouble(), dis.readLong());
+            }
+
+            return new StreamingHistogram(maxBinSize, tmp);
+        }
+
+        public long serializedSize(StreamingHistogram histogram)
+        {
+            throw new UnsupportedOperationException();
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6b2ea264/test/unit/org/apache/cassandra/db/compaction/CompactionsTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/db/compaction/CompactionsTest.java b/test/unit/org/apache/cassandra/db/compaction/CompactionsTest.java
index 23a2657..a9c66bf 100644
--- a/test/unit/org/apache/cassandra/db/compaction/CompactionsTest.java
+++ b/test/unit/org/apache/cassandra/db/compaction/CompactionsTest.java
@@ -20,17 +20,15 @@ package org.apache.cassandra.db.compaction;
 
 import java.io.*;
 import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.HashSet;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.Set;
+import java.util.*;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 
 import org.junit.Test;
 import static junit.framework.Assert.assertEquals;
 import static junit.framework.Assert.assertNotNull;
+import static junit.framework.Assert.assertTrue;
 
 import org.apache.cassandra.CleanupHelper;
 import org.apache.cassandra.Util;
@@ -100,6 +98,60 @@ public class CompactionsTest extends CleanupHelper
         store.truncate();
     }
 
+    /**
+     * Test to see if sstable has enough expired columns, it is compacted itself.
+     */
+    @Test
+    public void testSingleSSTableCompaction() throws Exception
+    {
+        Table table = Table.open(TABLE1);
+        ColumnFamilyStore store = table.getColumnFamilyStore("Standard1");
+        store.clearUnsafe();
+        store.metadata.gcGraceSeconds(5);
+
+        // update SizeTieredCompactionStrategy's min_sstable_size to something small
+        // to split bucket for ttl'd sstable from others
+        Map<String, String> opts = new HashMap<String, String>();
+        opts.put(SizeTieredCompactionStrategy.MIN_SSTABLE_SIZE_KEY, "512");
+        store.metadata.compactionStrategyOptions(opts);
+        store.setCompactionStrategyClass(SizeTieredCompactionStrategy.class.getCanonicalName());
+        // disable compaction while flushing
+        store.disableAutoCompaction();
+
+        long timestamp = System.currentTimeMillis();
+        for (int i = 0; i < 10; i++)
+        {
+            DecoratedKey key = Util.dk(Integer.toString(i));
+            RowMutation rm = new RowMutation(TABLE1, key.key);
+            for (int j = 0; j < 10; j++)
+                rm.add(new QueryPath("Standard1", null, ByteBufferUtil.bytes(Integer.toString(j))),
+                       ByteBufferUtil.EMPTY_BYTE_BUFFER,
+                       timestamp,
+                       j > 0 ? 3 : 0); // let first column never expire, since deleting all columns does not produce sstable
+            rm.apply();
+        }
+        store.forceBlockingFlush();
+        assertEquals(1, store.getSSTables().size());
+        long originalSize = store.getSSTables().iterator().next().uncompressedLength();
+
+        // wait enough to force single compaction
+        TimeUnit.SECONDS.sleep(5);
+
+        // enable compaction, submit background and wait for it to complete
+        store.setMinimumCompactionThreshold(2);
+        store.setMaximumCompactionThreshold(4);
+        FBUtilities.waitOnFuture(CompactionManager.instance.submitBackground(store));
+        while (CompactionManager.instance.getPendingTasks() > 0 || CompactionManager.instance.getActiveCompactions() > 0)
+            TimeUnit.SECONDS.sleep(1);
+
+        // and sstable with ttl should be compacted
+        assertEquals(1, store.getSSTables().size());
+        long size = store.getSSTables().iterator().next().uncompressedLength();
+        assertTrue("should be less than " + originalSize + ", but was " + size, size < originalSize);
+
+        // make sure max timestamp of compacted sstables is recorded properly after compaction.
+        assertMaxTimestamp(store, timestamp);
+    }
 
     @Test
     public void testSuperColumnCompactions() throws IOException, ExecutionException, InterruptedException

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6b2ea264/test/unit/org/apache/cassandra/utils/StreamingHistogramTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/utils/StreamingHistogramTest.java b/test/unit/org/apache/cassandra/utils/StreamingHistogramTest.java
new file mode 100644
index 0000000..0da6849
--- /dev/null
+++ b/test/unit/org/apache/cassandra/utils/StreamingHistogramTest.java
@@ -0,0 +1,123 @@
+/*
+ * 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.utils;
+
+import com.google.common.io.ByteArrayDataOutput;
+import org.junit.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.util.*;
+
+import static junit.framework.Assert.assertEquals;
+
+public class StreamingHistogramTest
+{
+    @Test
+    public void testFunction() throws Exception
+    {
+        StreamingHistogram hist = new StreamingHistogram(5);
+        long[] samples = new long[]{23, 19, 10, 16, 36, 2, 9, 32, 30, 45};
+
+        // add 7 points to histogram of 5 bins
+        for (int i = 0; i < 7; i++)
+        {
+            hist.update(samples[i]);
+        }
+
+        // should end up (2,1),(9.5,2),(17.5,2),(23,1),(36,1)
+        Map<Double, Long> expected1 = new LinkedHashMap<Double, Long>(5);
+        expected1.put(2.0, 1L);
+        expected1.put(9.5, 2L);
+        expected1.put(17.5, 2L);
+        expected1.put(23.0, 1L);
+        expected1.put(36.0, 1L);
+
+        Iterator<Map.Entry<Double, Long>> expectedItr = expected1.entrySet().iterator();
+        for (Map.Entry<Double, Long> actual : hist.getAsMap().entrySet())
+        {
+            Map.Entry<Double, Long> entry = expectedItr.next();
+            assertEquals(entry.getKey(), actual.getKey(), 0.01);
+            assertEquals(entry.getValue(), actual.getValue());
+        }
+
+        // merge test
+        StreamingHistogram hist2 = new StreamingHistogram(3);
+        for (int i = 7; i < samples.length; i++)
+        {
+            hist2.update(samples[i]);
+        }
+        hist.merge(hist2);
+        // should end up (2,1),(9.5,2),(19.33,3),(32.67,3),(45,1)
+        Map<Double, Long> expected2 = new LinkedHashMap<Double, Long>(5);
+        expected2.put(2.0, 1L);
+        expected2.put(9.5, 2L);
+        expected2.put(19.33, 3L);
+        expected2.put(32.67, 3L);
+        expected2.put(45.0, 1L);
+        expectedItr = expected2.entrySet().iterator();
+        for (Map.Entry<Double, Long> actual : hist.getAsMap().entrySet())
+        {
+            Map.Entry<Double, Long> entry = expectedItr.next();
+            assertEquals(entry.getKey(), actual.getKey(), 0.01);
+            assertEquals(entry.getValue(), actual.getValue());
+        }
+
+        // sum test
+        assertEquals(3.28, hist.sum(15), 0.01);
+        // sum test (b > max(hist))
+        assertEquals(10.0, hist.sum(50), 0.01);
+    }
+
+    @Test
+    public void testSerDe() throws Exception
+    {
+        StreamingHistogram hist = new StreamingHistogram(5);
+        long[] samples = new long[]{23, 19, 10, 16, 36, 2, 9};
+
+        // add 7 points to histogram of 5 bins
+        for (int i = 0; i < samples.length; i++)
+        {
+            hist.update(samples[i]);
+        }
+
+        ByteArrayOutputStream out = new ByteArrayOutputStream();
+        StreamingHistogram.serializer.serialize(hist, new DataOutputStream(out));
+        byte[] bytes = out.toByteArray();
+
+        StreamingHistogram deserialized = StreamingHistogram.serializer.deserialize(new DataInputStream(new ByteArrayInputStream(bytes)));
+
+        // deserialized histogram should have following values
+        Map<Double, Long> expected1 = new LinkedHashMap<Double, Long>(5);
+        expected1.put(2.0, 1L);
+        expected1.put(9.5, 2L);
+        expected1.put(17.5, 2L);
+        expected1.put(23.0, 1L);
+        expected1.put(36.0, 1L);
+
+        Iterator<Map.Entry<Double, Long>> expectedItr = expected1.entrySet().iterator();
+        for (Map.Entry<Double, Long> actual : deserialized.getAsMap().entrySet())
+        {
+            Map.Entry<Double, Long> entry = expectedItr.next();
+            assertEquals(entry.getKey(), actual.getKey(), 0.01);
+            assertEquals(entry.getValue(), actual.getValue());
+        }
+    }
+}


Mime
View raw message