From ceed3a20ef78b402a7a734e63d758aff105fa2de Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Jimmy=20M=C3=A5rdell?= <yarin@spotify.com>
Date: Thu, 4 Dec 2014 09:59:34 -0600
Subject: [PATCH] Use live sstables in snapshot repair if possible
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

patch by Jimmy Mårdell; reviewed by yukim for CASSANDRA-8312
---
 CHANGES.txt                                   |  1 +
 .../cassandra/db/ColumnFamilyStore.java       | 36 +++++++++++++++++--
 .../db/compaction/CompactionManager.java      | 13 +++----
 3 files changed, 38 insertions(+), 12 deletions(-)

diff --git a/CHANGES.txt b/CHANGES.txt
index dc3896d54b..79c2d81d27 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -30,6 +30,7 @@
  * Fix totalDiskSpaceUsed calculation (CASSANDRA-8205)
  * Add DC-aware sequential repair (CASSANDRA-8193)
  * Improve JBOD disk utilization (CASSANDRA-7386)
+ * Use live sstables in snapshot repair if possible (CASSANDRA-8312)
 
 
 2.0.11:
diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
index 6cdf9e901b..b5c6c98c59 100644
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@ -1840,10 +1840,40 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
 
     public List<SSTableReader> getSnapshotSSTableReader(String tag) throws IOException
     {
+        Map<Integer, SSTableReader> active = new HashMap<>();
+        for (SSTableReader sstable : data.getView().sstables)
+            active.put(sstable.descriptor.generation, sstable);
         Map<Descriptor, Set<Component>> snapshots = directories.sstableLister().snapshots(tag).list();
-        List<SSTableReader> readers = new ArrayList<SSTableReader>(snapshots.size());
-        for (Map.Entry<Descriptor, Set<Component>> entries : snapshots.entrySet())
-            readers.add(SSTableReader.open(entries.getKey(), entries.getValue(), metadata, partitioner));
+        List<SSTableReader> readers = new ArrayList<>(snapshots.size());
+        try
+        {
+            for (Map.Entry<Descriptor, Set<Component>> entries : snapshots.entrySet())
+            {
+                // Try acquire reference to an active sstable instead of snapshot if it exists,
+                // to avoid opening new sstables. If it fails, use the snapshot reference instead.
+                SSTableReader sstable = active.get(entries.getKey().generation);
+                if (sstable == null || !sstable.acquireReference())
+                {
+                    if (logger.isDebugEnabled())
+                        logger.debug("using snapshot sstable " + entries.getKey());
+                    sstable = SSTableReader.open(entries.getKey(), entries.getValue(), metadata, partitioner);
+                    // This is technically not necessary since it's a snapshot but makes things easier
+                    sstable.acquireReference();
+                }
+                else if (logger.isDebugEnabled())
+                {
+                    logger.debug("using active sstable " + entries.getKey());
+                }
+                readers.add(sstable);
+            }
+        }
+        catch (IOException | RuntimeException e)
+        {
+            // In case one of the snapshot sstables fails to open,
+            // we must release the references to the ones we opened so far
+            SSTableReader.releaseReferences(readers);
+            throw e;
+        }
         return readers;
     }
 
diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
index d298e72e6b..19dedb0ff9 100644
--- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
@@ -765,8 +765,8 @@ public class CompactionManager implements CompactionManagerMBean
             sstables = cfs.getSnapshotSSTableReader(snapshotName);
 
             // Computing gcbefore based on the current time wouldn't be very good because we know each replica will execute
-            // this at a different time (that's the whole purpose of repair with snaphsot). So instead we take the creation
-            // time of the snapshot, which should give us roughtly the same time on each replica (roughtly being in that case
+            // this at a different time (that's the whole purpose of repair with snapshot). So instead we take the creation
+            // time of the snapshot, which should give us roughly the same time on each replica (roughly being in that case
             // 'as good as in the non-snapshot' case)
             gcBefore = cfs.gcBefore(cfs.getSnapshotCreationTime(snapshotName));
         }
@@ -803,16 +803,11 @@ public class CompactionManager implements CompactionManagerMBean
         finally
         {
             iter.close();
+            SSTableReader.releaseReferences(sstables);
             if (isSnapshotValidation)
             {
-                for (SSTableReader sstable : sstables)
-                    FileUtils.closeQuietly(sstable);
                 cfs.clearSnapshot(snapshotName);
             }
-            else
-            {
-                SSTableReader.releaseReferences(sstables);
-            }
 
             metrics.finishCompaction(ci);
         }
@@ -956,7 +951,7 @@ public class CompactionManager implements CompactionManagerMBean
         public void afterExecute(Runnable r, Throwable t)
         {
             DebuggableThreadPoolExecutor.maybeResetTraceSessionWrapper(r);
-    
+
             if (t == null)
                 t = DebuggableThreadPoolExecutor.extractThrowable(r);