cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From marc...@apache.org
Subject git commit: Improve compaction of repaired/unrepaired sstables
Date Wed, 05 Nov 2014 08:50:32 GMT
Repository: cassandra
Updated Branches:
  refs/heads/cassandra-2.1 0a0ba84b8 -> e16f584e6


Improve compaction of repaired/unrepaired sstables

Patch by marcuse; reviewed by Aleksey Yeschenko and Sankalp Kohli for CASSANDRA-8004


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

Branch: refs/heads/cassandra-2.1
Commit: e16f584e61dccd243656269a4f305c7a50b7e433
Parents: 0a0ba84
Author: Marcus Eriksson <marcuse@apache.org>
Authored: Wed Oct 15 10:46:01 2014 +0200
Committer: Marcus Eriksson <marcuse@apache.org>
Committed: Wed Nov 5 09:35:24 2014 +0100

----------------------------------------------------------------------
 CHANGES.txt                                     |   1 +
 .../org/apache/cassandra/config/CFMetaData.java |   3 +
 .../cassandra/db/CollationController.java       |   2 +-
 .../apache/cassandra/db/ColumnFamilyStore.java  |  57 +---
 .../compaction/AbstractCompactionStrategy.java  |  14 +
 .../db/compaction/CompactionManager.java        |   5 +-
 .../DateTieredCompactionStrategy.java           |  46 +--
 .../compaction/LeveledCompactionStrategy.java   |  60 +---
 .../db/compaction/LeveledManifest.java          | 110 ++----
 .../SizeTieredCompactionStrategy.java           |  63 ++--
 .../compaction/WrappingCompactionStrategy.java  | 331 +++++++++++++++++++
 .../cassandra/io/sstable/SSTableReader.java     |   3 +-
 .../db/compaction/CompactionsTest.java          |   4 +-
 .../LeveledCompactionStrategyTest.java          |  88 +++--
 14 files changed, 487 insertions(+), 300 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/e16f584e/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 42cef8c..80f4c8f 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 2.1.2
+ * Improve compaction of repaired/unrepaired sstables (CASSANDRA-8004)
  * Make cache serializers pluggable (CASSANDRA-8096)
  * Fix issues with CONTAINS (KEY) queries on secondary indexes
    (CASSANDRA-8147)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/e16f584e/src/java/org/apache/cassandra/config/CFMetaData.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/config/CFMetaData.java b/src/java/org/apache/cassandra/config/CFMetaData.java
index d986c40..7f0e7cb 100644
--- a/src/java/org/apache/cassandra/config/CFMetaData.java
+++ b/src/java/org/apache/cassandra/config/CFMetaData.java
@@ -73,6 +73,7 @@ import org.apache.cassandra.db.SystemKeyspace;
 import org.apache.cassandra.db.compaction.AbstractCompactionStrategy;
 import org.apache.cassandra.db.compaction.LeveledCompactionStrategy;
 import org.apache.cassandra.db.compaction.SizeTieredCompactionStrategy;
+import org.apache.cassandra.db.compaction.WrappingCompactionStrategy;
 import org.apache.cassandra.db.composites.CType;
 import org.apache.cassandra.db.composites.CellName;
 import org.apache.cassandra.db.composites.CellNameType;
@@ -1296,6 +1297,8 @@ public final class CFMetaData
     {
         className = className.contains(".") ? className : "org.apache.cassandra.db.compaction." + className;
         Class<AbstractCompactionStrategy> strategyClass = FBUtilities.classForName(className, "compaction strategy");
+        if (strategyClass.equals(WrappingCompactionStrategy.class))
+            throw new ConfigurationException("You can't set WrappingCompactionStrategy as the compaction strategy!");
         if (!AbstractCompactionStrategy.class.isAssignableFrom(strategyClass))
             throw new ConfigurationException(String.format("Specified compaction strategy class (%s) is not derived from AbstractReplicationStrategy", className));
 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/e16f584e/src/java/org/apache/cassandra/db/CollationController.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/CollationController.java b/src/java/org/apache/cassandra/db/CollationController.java
index f9d5daa..1b22e70 100644
--- a/src/java/org/apache/cassandra/db/CollationController.java
+++ b/src/java/org/apache/cassandra/db/CollationController.java
@@ -150,7 +150,7 @@ public class CollationController
             // "hoist up" the requested data into a more recent sstable
             if (sstablesIterated > cfs.getMinimumCompactionThreshold()
                 && !cfs.isAutoCompactionDisabled()
-                && cfs.getCompactionStrategy() instanceof SizeTieredCompactionStrategy)
+                && cfs.getCompactionStrategy().shouldDefragment())
             {
                 Tracing.trace("Defragmenting requested data");
                 Mutation mutation = new Mutation(cfs.keyspace.getName(), filter.key.getKey(), returnCF.cloneMe());

http://git-wip-us.apache.org/repos/asf/cassandra/blob/e16f584e/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
index 0541608..0fa50bb 100644
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@ -127,7 +127,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
     /* These are locally held copies to be changed from the config during runtime */
     private volatile DefaultInteger minCompactionThreshold;
     private volatile DefaultInteger maxCompactionThreshold;
-    private volatile AbstractCompactionStrategy compactionStrategy;
+    private final WrappingCompactionStrategy compactionStrategyWrapper;
 
     public final Directories directories;
 
@@ -146,7 +146,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
             for (ColumnFamilyStore cfs : concatWithIndexes())
                 cfs.maxCompactionThreshold = new DefaultInteger(metadata.getMaxCompactionThreshold());
 
-        maybeReloadCompactionStrategy();
+        compactionStrategyWrapper.maybeReloadCompactionStrategy(metadata);
 
         scheduleFlush();
 
@@ -158,22 +158,6 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
             switchMemtable();
     }
 
-    private void maybeReloadCompactionStrategy()
-    {
-        // Check if there is a need for reloading
-        if (metadata.compactionStrategyClass.equals(compactionStrategy.getClass()) && metadata.compactionStrategyOptions.equals(compactionStrategy.options))
-            return;
-
-        // synchronize vs runWithCompactionsDisabled calling pause/resume.  otherwise, letting old compactions
-        // finish should be harmless and possibly useful.
-        synchronized (this)
-        {
-            compactionStrategy.shutdown();
-            compactionStrategy = metadata.createCompactionStrategyInstance(this);
-            compactionStrategy.startup();
-        }
-    }
-
     void scheduleFlush()
     {
         int period = metadata.getMemtableFlushPeriod();
@@ -213,7 +197,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
         try
         {
             metadata.compactionStrategyClass = CFMetaData.createCompactionStrategy(compactionStrategyClass);
-            maybeReloadCompactionStrategy();
+            compactionStrategyWrapper.maybeReloadCompactionStrategy(metadata);
         }
         catch (ConfigurationException e)
         {
@@ -297,13 +281,12 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
             CacheService.instance.keyCache.loadSaved(this);
 
         // compaction strategy should be created after the CFS has been prepared
-        this.compactionStrategy = metadata.createCompactionStrategyInstance(this);
-        this.compactionStrategy.startup();
+        this.compactionStrategyWrapper = new WrappingCompactionStrategy(this);
 
         if (maxCompactionThreshold.value() <= 0 || minCompactionThreshold.value() <=0)
         {
             logger.warn("Disabling compaction strategy by setting compaction thresholds to 0 is deprecated, set the compaction option 'enabled' to 'false' instead.");
-            this.compactionStrategy.disable();
+            this.compactionStrategyWrapper.disable();
         }
 
         // create the private ColumnFamilyStores for the secondary column indexes
@@ -367,7 +350,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
             logger.warn("Failed unregistering mbean: {}", mbeanName, e);
         }
 
-        compactionStrategy.shutdown();
+        compactionStrategyWrapper.shutdown();
 
         SystemKeyspace.removeTruncationRecord(metadata.cfId);
         data.unreferenceSSTables();
@@ -1370,7 +1353,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
 
     void replaceFlushed(Memtable memtable, SSTableReader sstable)
     {
-        compactionStrategy.replaceFlushed(memtable, sstable);
+        compactionStrategyWrapper.replaceFlushed(memtable, sstable);
     }
 
     public boolean isValid()
@@ -1810,7 +1793,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
         {
             public List<SSTableReader> apply(DataTracker.View view)
             {
-                return compactionStrategy.filterSSTablesForReads(view.intervalTree.search(key));
+                return compactionStrategyWrapper.filterSSTablesForReads(view.intervalTree.search(key));
             }
         };
     }
@@ -1825,7 +1808,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
         {
             public List<SSTableReader> apply(DataTracker.View view)
             {
-                return compactionStrategy.filterSSTablesForReads(view.sstablesInBounds(rowBounds));
+                return compactionStrategyWrapper.filterSSTablesForReads(view.sstablesInBounds(rowBounds));
             }
         };
     }
@@ -2587,7 +2570,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
     {
         // we don't use CompactionStrategy.pause since we don't want users flipping that on and off
         // during runWithCompactionsDisabled
-        this.compactionStrategy.disable();
+        this.compactionStrategyWrapper.disable();
     }
 
     public void enableAutoCompaction()
@@ -2602,7 +2585,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
     @VisibleForTesting
     public void enableAutoCompaction(boolean waitForFutures)
     {
-        this.compactionStrategy.enable();
+        this.compactionStrategyWrapper.enable();
         List<Future<?>> futures = CompactionManager.instance.submitBackground(this);
         if (waitForFutures)
             FBUtilities.waitOnFutures(futures);
@@ -2610,7 +2593,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
 
     public boolean isAutoCompactionDisabled()
     {
-        return !this.compactionStrategy.isEnabled();
+        return !this.compactionStrategyWrapper.isEnabled();
     }
 
     /*
@@ -2624,8 +2607,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
 
     public AbstractCompactionStrategy getCompactionStrategy()
     {
-        assert compactionStrategy != null : "No compaction strategy set yet";
-        return compactionStrategy;
+        return compactionStrategyWrapper;
     }
 
     public void setCompactionThresholds(int minThreshold, int maxThreshold)
@@ -2634,10 +2616,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
 
         minCompactionThreshold.set(minThreshold);
         maxCompactionThreshold.set(maxThreshold);
-
-        // this is called as part of CompactionStrategy constructor; avoid circular dependency by checking for null
-        if (compactionStrategy != null)
-            CompactionManager.instance.submitBackground(this);
+        CompactionManager.instance.submitBackground(this);
     }
 
     public int getMinimumCompactionThreshold()
@@ -2725,16 +2704,12 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
 
     public int getUnleveledSSTables()
     {
-        return this.compactionStrategy instanceof LeveledCompactionStrategy
-               ? ((LeveledCompactionStrategy) this.compactionStrategy).getLevelSize(0)
-               : 0;
+        return this.compactionStrategyWrapper.getUnleveledSSTables();
     }
 
     public int[] getSSTableCountPerLevel()
     {
-        return compactionStrategy instanceof LeveledCompactionStrategy
-               ? ((LeveledCompactionStrategy) compactionStrategy).getAllLevelSize()
-               : null;
+        return compactionStrategyWrapper.getSSTableCountPerLevel();
     }
 
     public static class ViewFragment

http://git-wip-us.apache.org/repos/asf/cassandra/blob/e16f584e/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java b/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java
index 288c475..bf136b9 100644
--- a/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java
+++ b/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java
@@ -290,6 +290,20 @@ public abstract class AbstractCompactionStrategy
         return new ScannerList(scanners);
     }
 
+    public boolean shouldDefragment()
+    {
+        return false;
+    }
+
+    public String getName()
+    {
+        return getClass().getSimpleName();
+    }
+
+    public abstract void addSSTable(SSTableReader added);
+
+    public abstract void removeSSTable(SSTableReader sstable);
+
     public static class ScannerList implements AutoCloseable
     {
         public final List<ICompactionScanner> scanners;

http://git-wip-us.apache.org/repos/asf/cassandra/blob/e16f584e/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
index 84c3cb5..272b533 100644
--- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
@@ -179,7 +179,7 @@ public class CompactionManager implements CompactionManagerMBean
         logger.debug("Scheduling a background task check for {}.{} with {}",
                      cfs.keyspace.getName(),
                      cfs.name,
-                     cfs.getCompactionStrategy().getClass().getSimpleName());
+                     cfs.getCompactionStrategy().getName());
         List<Future<?>> futures = new ArrayList<Future<?>>();
 
         // we must schedule it at least once, otherwise compaction will stop for a CF until next flush
@@ -998,8 +998,7 @@ public class CompactionManager implements CompactionManagerMBean
             SSTableRewriter repairedSSTableWriter = new SSTableRewriter(cfs, sstableAsSet, sstable.maxDataAge, false);
             SSTableRewriter unRepairedSSTableWriter = new SSTableRewriter(cfs, sstableAsSet, sstable.maxDataAge, false);
 
-            AbstractCompactionStrategy strategy = cfs.getCompactionStrategy();
-            try (AbstractCompactionStrategy.ScannerList scanners = strategy.getScanners(new HashSet<>(Collections.singleton(sstable)));
+            try (AbstractCompactionStrategy.ScannerList scanners = cfs.getCompactionStrategy().getScanners(new HashSet<>(Collections.singleton(sstable)));
                  CompactionController controller = new CompactionController(cfs, sstableAsSet, CFMetaData.DEFAULT_GC_GRACE_SECONDS))
             {
                 repairedSSTableWriter.switchWriter(CompactionManager.createWriter(cfs, destination, expectedBloomFilterSize, repairedAt, sstable));

http://git-wip-us.apache.org/repos/asf/cassandra/blob/e16f584e/src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategy.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategy.java b/src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategy.java
index 8c997ed..7c1b514 100644
--- a/src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategy.java
+++ b/src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategy.java
@@ -35,8 +35,9 @@ public class DateTieredCompactionStrategy extends AbstractCompactionStrategy
 {
     private static final Logger logger = LoggerFactory.getLogger(DateTieredCompactionStrategy.class);
 
-    protected DateTieredCompactionStrategyOptions options;
+    private final DateTieredCompactionStrategyOptions options;
     protected volatile int estimatedRemainingTasks;
+    private final Set<SSTableReader> sstables = new HashSet<>();
 
     public DateTieredCompactionStrategy(ColumnFamilyStore cfs, Map<String, String> options)
     {
@@ -75,36 +76,12 @@ public class DateTieredCompactionStrategy extends AbstractCompactionStrategy
 
         int base = cfs.getMinimumCompactionThreshold();
         long now = getNow();
-        Iterable<SSTableReader> candidates = filterSuspectSSTables(cfs.getUncompactingSSTables());
+        Iterable<SSTableReader> candidates = filterSuspectSSTables(Sets.intersection(cfs.getUncompactingSSTables(), sstables));
 
-        Set<SSTableReader> repairedCandidates = new HashSet<>();
-        Set<SSTableReader> unRepairedCandidates = new HashSet<>();
-        for (SSTableReader sstable : candidates)
-        {
-            if (sstable.isRepaired())
-            {
-                repairedCandidates.add(sstable);
-            }
-            else
-            {
-                unRepairedCandidates.add(sstable);
-            }
-        }
-
-
-        List<SSTableReader> mostInterestingRepaired = getCompactionCandidates(repairedCandidates, now, base);
-        List<SSTableReader> mostInterestingUnrepaired = getCompactionCandidates(unRepairedCandidates, now, base);
-        if (mostInterestingRepaired != null && mostInterestingUnrepaired != null)
-        {
-            return mostInterestingRepaired.size() > mostInterestingUnrepaired.size() ? mostInterestingRepaired : mostInterestingUnrepaired;
-        }
-        else if (mostInterestingRepaired != null)
-        {
-            return mostInterestingRepaired;
-        }
-        else if (mostInterestingUnrepaired != null)
+        List<SSTableReader> mostInteresting = getCompactionCandidates(candidates, now, base);
+        if (mostInteresting != null)
         {
-            return mostInterestingUnrepaired;
+            return mostInteresting;
         }
 
         // if there is no sstable to compact in standard way, try compacting single sstable whose droppable tombstone
@@ -185,8 +162,17 @@ public class DateTieredCompactionStrategy extends AbstractCompactionStrategy
             sstableMinTimestampPairs.add(Pair.create(sstable, sstable.getMinTimestamp()));
         return sstableMinTimestampPairs;
     }
+    @Override
+    public void addSSTable(SSTableReader sstable)
+    {
+        sstables.add(sstable);
+    }
 
-
+    @Override
+    public void removeSSTable(SSTableReader sstable)
+    {
+        sstables.remove(sstable);
+    }
     /**
      * A target time span used for bucketing SSTables based on timestamps.
      */

http://git-wip-us.apache.org/repos/asf/cassandra/blob/e16f584e/src/java/org/apache/cassandra/db/compaction/LeveledCompactionStrategy.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/LeveledCompactionStrategy.java b/src/java/org/apache/cassandra/db/compaction/LeveledCompactionStrategy.java
index 7f2d881..a560234 100644
--- a/src/java/org/apache/cassandra/db/compaction/LeveledCompactionStrategy.java
+++ b/src/java/org/apache/cassandra/db/compaction/LeveledCompactionStrategy.java
@@ -33,13 +33,8 @@ import org.apache.cassandra.dht.Range;
 import org.apache.cassandra.dht.Token;
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.io.sstable.SSTableReader;
-import org.apache.cassandra.notifications.INotification;
-import org.apache.cassandra.notifications.INotificationConsumer;
-import org.apache.cassandra.notifications.SSTableAddedNotification;
-import org.apache.cassandra.notifications.SSTableListChangedNotification;
-import org.apache.cassandra.notifications.SSTableRepairStatusChanged;
 
-public class LeveledCompactionStrategy extends AbstractCompactionStrategy implements INotificationConsumer
+public class LeveledCompactionStrategy extends AbstractCompactionStrategy
 {
     private static final Logger logger = LoggerFactory.getLogger(LeveledCompactionStrategy.class);
     private static final String SSTABLE_SIZE_OPTION = "sstable_size_in_mb";
@@ -71,26 +66,10 @@ public class LeveledCompactionStrategy extends AbstractCompactionStrategy implem
         }
         maxSSTableSizeInMB = configuredMaxSSTableSize;
 
-        manifest = LeveledManifest.create(cfs, this.maxSSTableSizeInMB, cfs.getSSTables(), localOptions);
+        manifest = new LeveledManifest(cfs, this.maxSSTableSizeInMB, localOptions);
         logger.debug("Created {}", manifest);
     }
 
-    @Override
-    public void startup()
-    {
-        super.startup();
-        cfs.getDataTracker().subscribe(this);
-        logger.debug("{} subscribed to the data tracker.", this);
-    }
-
-    @Override
-    public void shutdown()
-    {
-        super.shutdown();
-        cfs.getDataTracker().unsubscribe(this);
-        logger.debug("{} unsubscribed from the data tracker.", this);
-    }
-
     public int getLevelSize(int i)
     {
         return manifest.getLevelSize(i);
@@ -175,24 +154,6 @@ public class LeveledCompactionStrategy extends AbstractCompactionStrategy implem
         return manifest.getEstimatedTasks();
     }
 
-    public void handleNotification(INotification notification, Object sender)
-    {
-        if (notification instanceof SSTableAddedNotification)
-        {
-            SSTableAddedNotification flushedNotification = (SSTableAddedNotification) notification;
-            manifest.add(flushedNotification.added);
-        }
-        else if (notification instanceof SSTableListChangedNotification)
-        {
-            SSTableListChangedNotification listChangedNotification = (SSTableListChangedNotification) notification;
-            manifest.replace(listChangedNotification.removed, listChangedNotification.added);
-        }
-        else if (notification instanceof SSTableRepairStatusChanged)
-        {
-            manifest.repairStatusChanged(((SSTableRepairStatusChanged) notification).sstable);
-        }
-    }
-
     public long getMaxSSTableBytes()
     {
         return maxSSTableSizeInMB * 1024L * 1024L;
@@ -203,10 +164,7 @@ public class LeveledCompactionStrategy extends AbstractCompactionStrategy implem
         Multimap<Integer, SSTableReader> byLevel = ArrayListMultimap.create();
         for (SSTableReader sstable : sstables)
         {
-            if (manifest.hasRepairedData() && !sstable.isRepaired())
-                byLevel.get(0).add(sstable);
-            else
-                byLevel.get(sstable.getSSTableLevel()).add(sstable);
+            byLevel.get(sstable.getSSTableLevel()).add(sstable);
         }
 
         List<ICompactionScanner> scanners = new ArrayList<ICompactionScanner>(sstables.size());
@@ -247,6 +205,18 @@ public class LeveledCompactionStrategy extends AbstractCompactionStrategy implem
         return new ScannerList(scanners);
     }
 
+    @Override
+    public void addSSTable(SSTableReader added)
+    {
+        manifest.add(added);
+    }
+
+    @Override
+    public void removeSSTable(SSTableReader sstable)
+    {
+        manifest.remove(sstable);
+    }
+
     // Lazily creates SSTableBoundedScanner for sstable that are assumed to be from the
     // same level (e.g. non overlapping) - see #4142
     private static class LeveledScanner extends AbstractIterator<OnDiskAtomIterator> implements ICompactionScanner

http://git-wip-us.apache.org/repos/asf/cassandra/blob/e16f584e/src/java/org/apache/cassandra/db/compaction/LeveledManifest.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/LeveledManifest.java b/src/java/org/apache/cassandra/db/compaction/LeveledManifest.java
index a4d2115..4b26d23 100644
--- a/src/java/org/apache/cassandra/db/compaction/LeveledManifest.java
+++ b/src/java/org/apache/cassandra/db/compaction/LeveledManifest.java
@@ -62,18 +62,14 @@ public class LeveledManifest
     private final ColumnFamilyStore cfs;
     @VisibleForTesting
     protected final List<SSTableReader>[] generations;
-    @VisibleForTesting
-    protected final List<SSTableReader> unrepairedL0;
     private final RowPosition[] lastCompactedKeys;
     private final int maxSSTableSizeInBytes;
     private final SizeTieredCompactionStrategyOptions options;
-    private boolean hasRepairedData = false;
     private final int [] compactionCounter;
 
-    private LeveledManifest(ColumnFamilyStore cfs, int maxSSTableSizeInMB, SizeTieredCompactionStrategyOptions options)
+    LeveledManifest(ColumnFamilyStore cfs, int maxSSTableSizeInMB, SizeTieredCompactionStrategyOptions options)
     {
         this.cfs = cfs;
-        this.hasRepairedData = cfs.getRepairedSSTables().size() > 0;
         this.maxSSTableSizeInBytes = maxSSTableSizeInMB * 1024 * 1024;
         this.options = options;
 
@@ -88,7 +84,6 @@ public class LeveledManifest
             generations[i] = new ArrayList<>();
             lastCompactedKeys[i] = cfs.partitioner.getMinimumToken().minKeyBound();
         }
-        unrepairedL0 = new ArrayList<>();
         compactionCounter = new int[n];
     }
 
@@ -115,73 +110,39 @@ public class LeveledManifest
 
     public synchronized void add(SSTableReader reader)
     {
-        if (!hasRepairedData && reader.isRepaired())
-        {
-            // this is the first repaired sstable we get - we need to
-            // rebuild the entire manifest, unrepaired data should be
-            // in unrepairedL0. Note that we keep the sstable level in
-            // the sstable metadata since we are likely to be able to
-            // re-add it at a good level later (during anticompaction
-            // for example).
-            hasRepairedData = true;
-            rebuildManifestAfterFirstRepair();
-        }
-
         int level = reader.getSSTableLevel();
-        if (hasRepairedData && !reader.isRepaired())
+
+        assert level < generations.length : "Invalid level " + level + " out of " + (generations.length - 1);
+        logDistribution();
+        if (canAddSSTable(reader))
         {
-            logger.debug("Adding unrepaired {} to unrepaired L0", reader);
-            unrepairedL0.add(reader);
+            // adding the sstable does not cause overlap in the level
+            logger.debug("Adding {} to L{}", reader, level);
+            generations[level].add(reader);
         }
         else
         {
-            assert level < generations.length : "Invalid level " + level + " out of " + (generations.length - 1);
-            logDistribution();
-            if (canAddSSTable(reader))
+            // this can happen if:
+            // * a compaction has promoted an overlapping sstable to the given level, or
+            //   was also supposed to add an sstable at the given level.
+            // * we are moving sstables from unrepaired to repaired and the sstable
+            //   would cause overlap
+            //
+            // The add(..):ed sstable will be sent to level 0
+            try
             {
-                // adding the sstable does not cause overlap in the level
-                logger.debug("Adding {} to L{}", reader, level);
-                generations[level].add(reader);
+                reader.descriptor.getMetadataSerializer().mutateLevel(reader.descriptor, 0);
+                reader.reloadSSTableMetadata();
             }
-            else
+            catch (IOException e)
             {
-                // this can happen if:
-                // * a compaction has promoted an overlapping sstable to the given level, or
-                // * we promote a non-repaired sstable to repaired at level > 0, but an ongoing compaction
-                //   was also supposed to add an sstable at the given level.
-                //
-                // The add(..):ed sstable will be sent to level 0
-                try
-                {
-                    reader.descriptor.getMetadataSerializer().mutateLevel(reader.descriptor, 0);
-                    reader.reloadSSTableMetadata();
-                }
-                catch (IOException e)
-                {
-                    logger.error("Could not change sstable level - adding it at level 0 anyway, we will find it at restart.", e);
-                }
-                generations[0].add(reader);
+                logger.error("Could not change sstable level - adding it at level 0 anyway, we will find it at restart.", e);
             }
+            generations[0].add(reader);
         }
-
     }
 
 
-    /**
-     * Since we run standard LCS when we have no repaired data
-     * we need to move all sstables from the leveling
-     * to unrepairedL0.
-     */
-    private void rebuildManifestAfterFirstRepair()
-    {
-        for (int i = 0; i < getAllLevelSize().length; i++)
-        {
-            List<SSTableReader> oldLevel = generations[i];
-            generations[i] = new ArrayList<>();
-            for (SSTableReader sstable : oldLevel)
-                add(sstable);
-        }
-    }
 
     public synchronized void replace(Collection<SSTableReader> removed, Collection<SSTableReader> added)
     {
@@ -216,7 +177,7 @@ public class LeveledManifest
     {
         SSTableReader previous = null;
         Collections.sort(generations[level], SSTableReader.sstableComparator);
-        List<SSTableReader> outOfOrderSSTables = new ArrayList<SSTableReader>();
+        List<SSTableReader> outOfOrderSSTables = new ArrayList<>();
         for (SSTableReader current : generations[level])
         {
             if (previous != null && current.first.compareTo(previous.last) <= 0)
@@ -279,15 +240,6 @@ public class LeveledManifest
         }
     }
 
-    public synchronized void repairStatusChanged(Collection<SSTableReader> sstables)
-    {
-        for(SSTableReader sstable : sstables)
-        {
-            remove(sstable);
-            add(sstable);
-        }
-    }
-
     private String toString(Collection<SSTableReader> sstables)
     {
         StringBuilder builder = new StringBuilder();
@@ -320,18 +272,6 @@ public class LeveledManifest
      */
     public synchronized CompactionCandidate getCompactionCandidates()
     {
-        // if we don't have any repaired data, continue as usual
-        if (hasRepairedData)
-        {
-            Collection<SSTableReader> unrepairedMostInterresting = getSSTablesForSTCS(unrepairedL0);
-            if (!unrepairedMostInterresting.isEmpty())
-            {
-                logger.info("Unrepaired data is most interresting, compacting {} sstables with STCS", unrepairedMostInterresting.size());
-                for (SSTableReader reader : unrepairedMostInterresting)
-                    assert !reader.isRepaired();
-                return new CompactionCandidate(unrepairedMostInterresting, 0, Long.MAX_VALUE);
-            }
-        }
         // LevelDB gives each level a score of how much data it contains vs its ideal amount, and
         // compacts the level with the highest score. But this falls apart spectacularly once you
         // get behind.  Consider this set of levels:
@@ -519,7 +459,6 @@ public class LeveledManifest
         int level = reader.getSSTableLevel();
         assert level >= 0 : reader + " not present in manifest: "+level;
         generations[level].remove(reader);
-        unrepairedL0.remove(reader);
         return level;
     }
 
@@ -759,11 +698,6 @@ public class LeveledManifest
 
     }
 
-    public boolean hasRepairedData()
-    {
-        return hasRepairedData;
-    }
-
     public static class CompactionCandidate
     {
         public final Collection<SSTableReader> sstables;

http://git-wip-us.apache.org/repos/asf/cassandra/blob/e16f584e/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 461c5e1..b72737a 100644
--- a/src/java/org/apache/cassandra/db/compaction/SizeTieredCompactionStrategy.java
+++ b/src/java/org/apache/cassandra/db/compaction/SizeTieredCompactionStrategy.java
@@ -31,7 +31,6 @@ import org.apache.cassandra.cql3.statements.CFPropDefs;
 import org.apache.cassandra.db.ColumnFamilyStore;
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.io.sstable.SSTableReader;
-import org.apache.cassandra.service.ActiveRepairService;
 import org.apache.cassandra.utils.Pair;
 
 public class SizeTieredCompactionStrategy extends AbstractCompactionStrategy
@@ -62,6 +61,7 @@ public class SizeTieredCompactionStrategy extends AbstractCompactionStrategy
 
     protected SizeTieredCompactionStrategyOptions options;
     protected volatile int estimatedRemainingTasks;
+    private final Set<SSTableReader> sstables = new HashSet<>();
 
     public SizeTieredCompactionStrategy(ColumnFamilyStore cfs, Map<String, String> options)
     {
@@ -79,17 +79,8 @@ public class SizeTieredCompactionStrategy extends AbstractCompactionStrategy
         int minThreshold = cfs.getMinimumCompactionThreshold();
         int maxThreshold = cfs.getMaximumCompactionThreshold();
 
-        Iterable<SSTableReader> candidates = filterSuspectSSTables(cfs.getUncompactingSSTables());
+        Iterable<SSTableReader> candidates = filterSuspectSSTables(Sets.intersection(cfs.getUncompactingSSTables(), sstables));
         candidates = filterColdSSTables(Lists.newArrayList(candidates), options.coldReadsToOmit);
-        Pair<Set<SSTableReader>,Set<SSTableReader>> repairedUnrepaired = splitInRepairedAndUnrepaired(candidates);
-        if (repairedUnrepaired.left.size() > repairedUnrepaired.right.size())
-        {
-            candidates = repairedUnrepaired.left;
-        }
-        else
-        {
-            candidates = repairedUnrepaired.right;
-        }
 
         List<List<SSTableReader>> buckets = getBuckets(createSSTableAndLengthPairs(candidates), options.bucketHigh, options.bucketLow, options.minSSTableSize);
         logger.debug("Compaction buckets are {}", buckets);
@@ -113,20 +104,6 @@ public class SizeTieredCompactionStrategy extends AbstractCompactionStrategy
         return Collections.singletonList(sstablesWithTombstones.get(0));
     }
 
-    private static Pair<Set<SSTableReader>, Set<SSTableReader>> splitInRepairedAndUnrepaired(Iterable<SSTableReader> candidates)
-    {
-        Set<SSTableReader> repaired = new HashSet<>();
-        Set<SSTableReader> unRepaired = new HashSet<>();
-        for(SSTableReader candidate : candidates)
-        {
-            if (!candidate.isRepaired())
-                unRepaired.add(candidate);
-            else
-                repaired.add(candidate);
-        }
-        return Pair.create(repaired, unRepaired);
-    }
-
     /**
      * Removes as many cold sstables as possible while retaining at least 1-coldReadsToOmit of the total reads/sec
      * across all sstables
@@ -276,20 +253,12 @@ public class SizeTieredCompactionStrategy extends AbstractCompactionStrategy
 
     public Collection<AbstractCompactionTask> getMaximalTask(final int gcBefore)
     {
-        Iterable<SSTableReader> allSSTables = cfs.markAllCompacting();
-        if (allSSTables == null || Iterables.isEmpty(allSSTables))
+        Iterable<SSTableReader> filteredSSTables = filterSuspectSSTables(sstables);
+        if (Iterables.isEmpty(sstables))
             return null;
-        Set<SSTableReader> sstables = Sets.newHashSet(allSSTables);
-        Set<SSTableReader> repaired = new HashSet<>();
-        Set<SSTableReader> unrepaired = new HashSet<>();
-        for (SSTableReader sstable : sstables)
-        {
-            if (sstable.isRepaired())
-                repaired.add(sstable);
-            else
-                unrepaired.add(sstable);
-        }
-        return Arrays.<AbstractCompactionTask>asList(new CompactionTask(cfs, repaired, gcBefore, false), new CompactionTask(cfs, unrepaired, gcBefore, false));
+        if (!cfs.getDataTracker().markCompacting(filteredSSTables))
+            return null;
+        return Arrays.<AbstractCompactionTask>asList(new CompactionTask(cfs, filteredSSTables, gcBefore, false));
     }
 
     public AbstractCompactionTask getUserDefinedTask(Collection<SSTableReader> sstables, final int gcBefore)
@@ -396,6 +365,24 @@ public class SizeTieredCompactionStrategy extends AbstractCompactionStrategy
         return uncheckedOptions;
     }
 
+    @Override
+    public boolean shouldDefragment()
+    {
+        return true;
+    }
+
+    @Override
+    public void addSSTable(SSTableReader added)
+    {
+        sstables.add(added);
+    }
+
+    @Override
+    public void removeSSTable(SSTableReader sstable)
+    {
+        sstables.remove(sstable);
+    }
+
     public String toString()
     {
         return String.format("SizeTieredCompactionStrategy[%s/%s]",

http://git-wip-us.apache.org/repos/asf/cassandra/blob/e16f584e/src/java/org/apache/cassandra/db/compaction/WrappingCompactionStrategy.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/WrappingCompactionStrategy.java b/src/java/org/apache/cassandra/db/compaction/WrappingCompactionStrategy.java
new file mode 100644
index 0000000..1d713ef
--- /dev/null
+++ b/src/java/org/apache/cassandra/db/compaction/WrappingCompactionStrategy.java
@@ -0,0 +1,331 @@
+/*
+ * 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.db.compaction;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.Callable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.config.CFMetaData;
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.dht.Range;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.io.sstable.SSTableReader;
+import org.apache.cassandra.notifications.INotification;
+import org.apache.cassandra.notifications.INotificationConsumer;
+import org.apache.cassandra.notifications.SSTableAddedNotification;
+import org.apache.cassandra.notifications.SSTableDeletingNotification;
+import org.apache.cassandra.notifications.SSTableListChangedNotification;
+import org.apache.cassandra.notifications.SSTableRepairStatusChanged;
+
+public final class WrappingCompactionStrategy extends AbstractCompactionStrategy implements INotificationConsumer
+{
+    private static final Logger logger = LoggerFactory.getLogger(WrappingCompactionStrategy.class);
+    private volatile AbstractCompactionStrategy repaired;
+    private volatile AbstractCompactionStrategy unrepaired;
+    public WrappingCompactionStrategy(ColumnFamilyStore cfs)
+    {
+        super(cfs, cfs.metadata.compactionStrategyOptions);
+        reloadCompactionStrategy(cfs.metadata);
+        cfs.getDataTracker().subscribe(this);
+        logger.debug("{} subscribed to the data tracker.", this);
+    }
+
+    @Override
+    public synchronized AbstractCompactionTask getNextBackgroundTask(int gcBefore)
+    {
+        if (!isEnabled())
+            return null;
+
+        if (repaired.getEstimatedRemainingTasks() > unrepaired.getEstimatedRemainingTasks())
+        {
+            AbstractCompactionTask repairedTask = repaired.getNextBackgroundTask(gcBefore);
+            if (repairedTask != null)
+                return repairedTask;
+            return unrepaired.getNextBackgroundTask(gcBefore);
+        }
+        else
+        {
+            AbstractCompactionTask unrepairedTask = unrepaired.getNextBackgroundTask(gcBefore);
+            if (unrepairedTask != null)
+                return unrepairedTask;
+            return repaired.getNextBackgroundTask(gcBefore);
+        }
+
+    }
+
+    @Override
+    public Collection<AbstractCompactionTask> getMaximalTask(final int gcBefore)
+    {
+        // runWithCompactionsDisabled cancels active compactions and disables them, then we are able
+        // to make the repaired/unrepaired strategies mark their own sstables as compacting. Once the
+        // sstables are marked the compactions are re-enabled
+        return cfs.runWithCompactionsDisabled(new Callable<Collection<AbstractCompactionTask>>()
+        {
+            @Override
+            public Collection<AbstractCompactionTask> call() throws Exception
+            {
+                synchronized (WrappingCompactionStrategy.this)
+                {
+                    Collection<AbstractCompactionTask> repairedTasks = repaired.getMaximalTask(gcBefore);
+                    Collection<AbstractCompactionTask> unrepairedTasks = unrepaired.getMaximalTask(gcBefore);
+
+                    if (repairedTasks == null && unrepairedTasks == null)
+                        return null;
+
+                    if (repairedTasks == null)
+                        return unrepairedTasks;
+                    if (unrepairedTasks == null)
+                        return repairedTasks;
+
+                    List<AbstractCompactionTask> tasks = new ArrayList<>();
+                    tasks.addAll(repairedTasks);
+                    tasks.addAll(unrepairedTasks);
+                    return tasks;
+                }
+            }
+        }, false);
+    }
+
+    @Override
+    public synchronized AbstractCompactionTask getUserDefinedTask(Collection<SSTableReader> sstables, int gcBefore)
+    {
+        assert !sstables.isEmpty();
+        boolean userDefinedInRepaired = sstables.iterator().next().isRepaired();
+        for (SSTableReader sstable : sstables)
+        {
+            if (userDefinedInRepaired != sstable.isRepaired())
+            {
+                logger.error("You can't mix repaired and unrepaired sstables in a user defined compaction");
+                return null;
+            }
+        }
+        if (userDefinedInRepaired)
+            return repaired.getUserDefinedTask(sstables, gcBefore);
+        else
+            return unrepaired.getUserDefinedTask(sstables, gcBefore);
+    }
+
+    @Override
+    public synchronized int getEstimatedRemainingTasks()
+    {
+        assert repaired.getClass().equals(unrepaired.getClass());
+        return repaired.getEstimatedRemainingTasks() + unrepaired.getEstimatedRemainingTasks();
+    }
+
+    @Override
+    public synchronized long getMaxSSTableBytes()
+    {
+        assert repaired.getClass().equals(unrepaired.getClass());
+        return unrepaired.getMaxSSTableBytes();
+    }
+
+    public synchronized void maybeReloadCompactionStrategy(CFMetaData metadata)
+    {
+        if (repaired != null && repaired.getClass().equals(metadata.compactionStrategyClass)
+            && unrepaired != null && unrepaired.getClass().equals(metadata.compactionStrategyClass)
+            && repaired.options.equals(metadata.compactionStrategyOptions)
+            && unrepaired.options.equals(metadata.compactionStrategyOptions))
+            return;
+
+        reloadCompactionStrategy(metadata);
+    }
+
+    public synchronized void reloadCompactionStrategy(CFMetaData metadata)
+    {
+        if (repaired != null)
+            repaired.shutdown();
+        if (unrepaired != null)
+            unrepaired.shutdown();
+        repaired = metadata.createCompactionStrategyInstance(cfs);
+        unrepaired = metadata.createCompactionStrategyInstance(cfs);
+        startup();
+    }
+
+    public synchronized int getUnleveledSSTables()
+    {
+        if (this.repaired instanceof LeveledCompactionStrategy && this.unrepaired instanceof LeveledCompactionStrategy)
+        {
+            return ((LeveledCompactionStrategy)repaired).getLevelSize(0) + ((LeveledCompactionStrategy)unrepaired).getLevelSize(0);
+        }
+        return 0;
+    }
+
+    public synchronized int[] getSSTableCountPerLevel()
+    {
+        if (this.repaired instanceof LeveledCompactionStrategy && this.unrepaired instanceof LeveledCompactionStrategy)
+        {
+            int [] repairedCountPerLevel = ((LeveledCompactionStrategy) repaired).getAllLevelSize();
+            int [] unrepairedCountPerLevel = ((LeveledCompactionStrategy) unrepaired).getAllLevelSize();
+            return sumArrays(repairedCountPerLevel, unrepairedCountPerLevel);
+        }
+        return null;
+    }
+
+    public static int [] sumArrays(int[] a, int [] b)
+    {
+        int [] res = new int[Math.max(a.length, b.length)];
+        for (int i = 0; i < res.length; i++)
+        {
+            if (i < a.length && i < b.length)
+                res[i] = a[i] + b[i];
+            else if (i < a.length)
+                res[i] = a[i];
+            else
+                res[i] = b[i];
+        }
+        return res;
+    }
+
+    @Override
+    public boolean shouldDefragment()
+    {
+        assert repaired.getClass().equals(unrepaired.getClass());
+        return repaired.shouldDefragment();
+    }
+
+    @Override
+    public String getName()
+    {
+        assert repaired.getClass().equals(unrepaired.getClass());
+        return repaired.getName();
+    }
+
+    @Override
+    public void addSSTable(SSTableReader added)
+    {
+        throw new UnsupportedOperationException("Can't add sstables to the wrapping compaction strategy");
+    }
+
+    @Override
+    public void removeSSTable(SSTableReader sstable)
+    {
+        throw new UnsupportedOperationException("Can't remove sstables from the wrapping compaction strategy");
+    }
+
+    public synchronized void handleNotification(INotification notification, Object sender)
+    {
+        if (notification instanceof SSTableAddedNotification)
+        {
+            SSTableAddedNotification flushedNotification = (SSTableAddedNotification) notification;
+            if (flushedNotification.added.isRepaired())
+                repaired.addSSTable(flushedNotification.added);
+            else
+                unrepaired.addSSTable(flushedNotification.added);
+        }
+        else if (notification instanceof SSTableListChangedNotification)
+        {
+            SSTableListChangedNotification listChangedNotification = (SSTableListChangedNotification) notification;
+            for (SSTableReader sstable : listChangedNotification.removed)
+            {
+                if (sstable.isRepaired())
+                    repaired.removeSSTable(sstable);
+                else
+                    unrepaired.removeSSTable(sstable);
+            }
+            for (SSTableReader sstable : listChangedNotification.added)
+            {
+                if (sstable.isRepaired())
+                    repaired.addSSTable(sstable);
+                else
+                    unrepaired.addSSTable(sstable);
+            }
+        }
+        else if (notification instanceof SSTableRepairStatusChanged)
+        {
+            for (SSTableReader sstable : ((SSTableRepairStatusChanged) notification).sstable)
+            {
+                if (sstable.isRepaired())
+                {
+                    unrepaired.removeSSTable(sstable);
+                    repaired.addSSTable(sstable);
+                }
+                else
+                {
+                    repaired.removeSSTable(sstable);
+                    unrepaired.addSSTable(sstable);
+                }
+            }
+        }
+        else if (notification instanceof SSTableDeletingNotification)
+        {
+            SSTableReader sstable = ((SSTableDeletingNotification)notification).deleting;
+            if (sstable.isRepaired())
+                repaired.removeSSTable(sstable);
+            else
+                unrepaired.removeSSTable(sstable);
+        }
+    }
+
+    @Override
+    public List<SSTableReader> filterSSTablesForReads(List<SSTableReader> sstables)
+    {
+        // todo: union of filtered sstables or intersection?
+        return unrepaired.filterSSTablesForReads(repaired.filterSSTablesForReads(sstables));
+    }
+
+    @Override
+    public synchronized void startup()
+    {
+        super.startup();
+        for (SSTableReader sstable : cfs.getSSTables())
+        {
+            if (sstable.isRepaired())
+                repaired.addSSTable(sstable);
+            else
+                unrepaired.addSSTable(sstable);
+        }
+        repaired.startup();
+        unrepaired.startup();
+    }
+
+    @Override
+    public synchronized void shutdown()
+    {
+        super.shutdown();
+        repaired.shutdown();
+        unrepaired.shutdown();
+    }
+
+    @Override
+    public synchronized ScannerList getScanners(Collection<SSTableReader> sstables, Range<Token> range)
+    {
+        List<SSTableReader> repairedSSTables = new ArrayList<>();
+        List<SSTableReader> unrepairedSSTables = new ArrayList<>();
+        for (SSTableReader sstable : sstables)
+            if (sstable.isRepaired())
+                repairedSSTables.add(sstable);
+            else
+                unrepairedSSTables.add(sstable);
+        ScannerList repairedScanners = repaired.getScanners(repairedSSTables, range);
+        ScannerList unrepairedScanners = unrepaired.getScanners(unrepairedSSTables, range);
+        List<ICompactionScanner> scanners = new ArrayList<>(repairedScanners.scanners.size() + unrepairedScanners.scanners.size());
+        scanners.addAll(repairedScanners.scanners);
+        scanners.addAll(unrepairedScanners.scanners);
+        return new ScannerList(scanners);
+    }
+
+    public List<AbstractCompactionStrategy> getWrappedStrategies()
+    {
+        return Arrays.asList(repaired, unrepaired);
+    }
+}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/e16f584e/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 40e708d..8f302f3 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
@@ -240,6 +240,7 @@ public class SSTableReader extends SSTable
                 try
                 {
                     CompactionMetadata metadata = (CompactionMetadata) sstable.descriptor.getMetadataSerializer().deserialize(sstable.descriptor, MetadataType.COMPACTION);
+                    assert metadata != null : sstable.getFilename();
                     if (cardinality == null)
                         cardinality = metadata.cardinalityEstimator;
                     else
@@ -1633,7 +1634,7 @@ public class SSTableReader extends SSTable
 
         synchronized (replaceLock)
         {
-            assert replacedBy == null;
+            assert replacedBy == null : getFilename();
         }
         return !isCompacted.getAndSet(true);
     }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/e16f584e/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 979b079..a1ecfab 100644
--- a/test/unit/org/apache/cassandra/db/compaction/CompactionsTest.java
+++ b/test/unit/org/apache/cassandra/db/compaction/CompactionsTest.java
@@ -126,9 +126,9 @@ public class CompactionsTest extends SchemaLoader
     public void testSingleSSTableCompactionWithLeveledCompaction() throws Exception
     {
         ColumnFamilyStore store = testSingleSSTableCompaction(LeveledCompactionStrategy.class.getCanonicalName());
-        LeveledCompactionStrategy strategy = (LeveledCompactionStrategy) store.getCompactionStrategy();
+        WrappingCompactionStrategy strategy = (WrappingCompactionStrategy) store.getCompactionStrategy();
         // tombstone removal compaction should not promote level
-        assert strategy.getLevelSize(0) == 1;
+        assert strategy.getSSTableCountPerLevel()[0] == 1;
     }
 
     @Test

http://git-wip-us.apache.org/repos/asf/cassandra/blob/e16f584e/test/unit/org/apache/cassandra/db/compaction/LeveledCompactionStrategyTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/db/compaction/LeveledCompactionStrategyTest.java b/test/unit/org/apache/cassandra/db/compaction/LeveledCompactionStrategyTest.java
index 65c7b69..ebc6e86 100644
--- a/test/unit/org/apache/cassandra/db/compaction/LeveledCompactionStrategyTest.java
+++ b/test/unit/org/apache/cassandra/db/compaction/LeveledCompactionStrategyTest.java
@@ -35,6 +35,8 @@ import org.apache.cassandra.db.*;
 import org.apache.cassandra.dht.Range;
 import org.apache.cassandra.dht.Token;
 import org.apache.cassandra.io.sstable.SSTableReader;
+import org.apache.cassandra.notifications.SSTableAddedNotification;
+import org.apache.cassandra.notifications.SSTableRepairStatusChanged;
 import org.apache.cassandra.repair.RepairJobDesc;
 import org.apache.cassandra.repair.Validator;
 import org.apache.cassandra.service.ActiveRepairService;
@@ -93,10 +95,10 @@ public class LeveledCompactionStrategyTest extends SchemaLoader
         }
 
         waitForLeveling(cfs);
-        LeveledCompactionStrategy strategy = (LeveledCompactionStrategy) cfs.getCompactionStrategy();
+        WrappingCompactionStrategy strategy = (WrappingCompactionStrategy) cfs.getCompactionStrategy();
         // Checking we're not completely bad at math
-        assert strategy.getLevelSize(1) > 0;
-        assert strategy.getLevelSize(2) > 0;
+        assert strategy.getSSTableCountPerLevel()[1] > 0;
+        assert strategy.getSSTableCountPerLevel()[2] > 0;
 
         Range<Token> range = new Range<>(Util.token(""), Util.token(""));
         int gcBefore = keyspace.getColumnFamilyStore(cfname).gcBefore(System.currentTimeMillis());
@@ -112,9 +114,9 @@ public class LeveledCompactionStrategyTest extends SchemaLoader
      */
     private void waitForLeveling(ColumnFamilyStore cfs) throws InterruptedException
     {
-        LeveledCompactionStrategy strategy = (LeveledCompactionStrategy) cfs.getCompactionStrategy();
+        WrappingCompactionStrategy strategy = (WrappingCompactionStrategy) cfs.getCompactionStrategy();
         // L0 is the lowest priority, so when that's done, we know everything is done
-        while (strategy.getLevelSize(0) > 1)
+        while (strategy.getSSTableCountPerLevel()[0] > 1)
             Thread.sleep(100);
     }
 
@@ -138,7 +140,7 @@ public class LeveledCompactionStrategyTest extends SchemaLoader
         }
 
         waitForLeveling(cfs);
-        LeveledCompactionStrategy strategy = (LeveledCompactionStrategy) cfs.getCompactionStrategy();
+        LeveledCompactionStrategy strategy = (LeveledCompactionStrategy) ((WrappingCompactionStrategy) cfs.getCompactionStrategy()).getWrappedStrategies().get(1);
         assert strategy.getLevelSize(1) > 0;
 
         // get LeveledScanner for level 1 sstables
@@ -177,7 +179,7 @@ public class LeveledCompactionStrategyTest extends SchemaLoader
         }
         waitForLeveling(cfs);
         cfs.forceBlockingFlush();
-        LeveledCompactionStrategy strategy = (LeveledCompactionStrategy) cfs.getCompactionStrategy();
+        LeveledCompactionStrategy strategy = (LeveledCompactionStrategy) ((WrappingCompactionStrategy) cfs.getCompactionStrategy()).getWrappedStrategies().get(1);
         cfs.disableAutoCompaction();
 
         while(CompactionManager.instance.isCompacting(Arrays.asList(cfs)))
@@ -227,61 +229,45 @@ public class LeveledCompactionStrategyTest extends SchemaLoader
         while(CompactionManager.instance.isCompacting(Arrays.asList(cfs)))
             Thread.sleep(100);
 
-        LeveledCompactionStrategy strategy = (LeveledCompactionStrategy) cfs.getCompactionStrategy();
-        assertTrue(strategy.getLevelSize(1) > 0);
-        assertTrue(strategy.getLevelSize(2) > 0);
+        WrappingCompactionStrategy strategy = (WrappingCompactionStrategy) cfs.getCompactionStrategy();
+        List<AbstractCompactionStrategy> strategies = strategy.getWrappedStrategies();
+        LeveledCompactionStrategy repaired = (LeveledCompactionStrategy) strategies.get(0);
+        LeveledCompactionStrategy unrepaired = (LeveledCompactionStrategy) strategies.get(1);
+        assertEquals(0, repaired.manifest.getLevelCount() );
+        assertEquals(2, unrepaired.manifest.getLevelCount());
+        assertTrue(strategy.getSSTableCountPerLevel()[1] > 0);
+        assertTrue(strategy.getSSTableCountPerLevel()[2] > 0);
 
         for (SSTableReader sstable : cfs.getSSTables())
-        {
             assertFalse(sstable.isRepaired());
-        }
+
         int sstableCount = 0;
-        for (List<SSTableReader> level : strategy.manifest.generations)
+        for (List<SSTableReader> level : unrepaired.manifest.generations)
             sstableCount += level.size();
-
+        // we only have unrepaired sstables:
         assertEquals(sstableCount, cfs.getSSTables().size());
 
-        assertFalse(strategy.manifest.hasRepairedData());
-        assertTrue(strategy.manifest.unrepairedL0.size() == 0);
-
-        SSTableReader sstable1 = strategy.manifest.generations[2].get(0);
-        SSTableReader sstable2 = strategy.manifest.generations[1].get(0);
+        SSTableReader sstable1 = unrepaired.manifest.generations[2].get(0);
+        SSTableReader sstable2 = unrepaired.manifest.generations[1].get(0);
 
-        // "repair" an sstable:
-        strategy.manifest.remove(sstable1);
         sstable1.descriptor.getMetadataSerializer().mutateRepairedAt(sstable1.descriptor, System.currentTimeMillis());
         sstable1.reloadSSTableMetadata();
         assertTrue(sstable1.isRepaired());
 
-        // make sure adding a repaired sstable makes the manifest contain only repaired data;
-        strategy.manifest.add(sstable1);
-        assertTrue(strategy.manifest.hasRepairedData());
-        assertTrue(strategy.manifest.generations[2].contains(sstable1));
-        assertFalse(strategy.manifest.generations[1].contains(sstable2));
-        assertTrue(strategy.manifest.unrepairedL0.contains(sstable2));
-        sstableCount = 0;
-        for (int i = 0; i < strategy.manifest.generations.length; i++)
-        {
-            sstableCount += strategy.manifest.generations[i].size();
-            if (i != 2)
-                assertEquals(strategy.manifest.generations[i].size(), 0);
-            else
-                assertEquals(strategy.manifest.generations[i].size(), 1);
-        }
-        assertEquals(1, sstableCount);
-
-        // make sure adding an unrepaired sstable puts it in unrepairedL0:
-        strategy.manifest.remove(sstable2);
-        strategy.manifest.add(sstable2);
-        assertTrue(strategy.manifest.unrepairedL0.contains(sstable2));
-        assertEquals(strategy.manifest.unrepairedL0.size(), cfs.getSSTables().size() - 1);
-
-        // make sure repairing an sstable takes it away from unrepairedL0 and puts it in the correct level:
-        strategy.manifest.remove(sstable2);
-        sstable2.descriptor.getMetadataSerializer().mutateRepairedAt(sstable2.descriptor, System.currentTimeMillis());
-        sstable2.reloadSSTableMetadata();
-        strategy.manifest.add(sstable2);
-        assertFalse(strategy.manifest.unrepairedL0.contains(sstable2));
-        assertTrue(strategy.manifest.generations[1].contains(sstable2));
+        strategy.handleNotification(new SSTableRepairStatusChanged(Arrays.asList(sstable1)), this);
+
+        int repairedSSTableCount = 0;
+        for (List<SSTableReader> level : repaired.manifest.generations)
+            repairedSSTableCount += level.size();
+        assertEquals(1, repairedSSTableCount);
+        // make sure the repaired sstable ends up in the same level in the repaired manifest:
+        assertTrue(repaired.manifest.generations[2].contains(sstable1));
+        // and that it is gone from unrepaired
+        assertFalse(unrepaired.manifest.generations[2].contains(sstable1));
+
+        unrepaired.removeSSTable(sstable2);
+        strategy.handleNotification(new SSTableAddedNotification(sstable2), this);
+        assertTrue(unrepaired.manifest.getLevel(1).contains(sstable2));
+        assertFalse(repaired.manifest.getLevel(1).contains(sstable2));
     }
 }


Mime
View raw message