fix re-using index CF sstable names after drop/recreate

patch by jbellis; reviewed by slebresne for CASSANDRA-2872 git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1148466 13f79535-47bb-0310-9956-ffa450edef68
apache · Jul 19, 2011 · 7f9f796 · 7f9f796
1 parent 83f9f45
commit 7f9f796
Show file tree

Hide file tree

Showing 2 changed files with 45 additions and 22 deletions.
diff --git a/CHANGES.txt b/CHANGES.txt
@@ -3,6 +3,7 @@
    reads with dynamic snitch + read repair disabled (CASSANDRA-2870)
  * support spaces in path to log4j configuration (CASSANDRA-2383)
  * avoid including inferred types in CF update (CASSANDRA-2809)
+ * fix re-using index CF sstable names after drop/recreate (CASSANDRA-2872)
 
 
 0.7.7

diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@ -466,13 +466,24 @@ public static synchronized ColumnFamilyStore createColumnFamilyStore(Table table
     {
         // get the max generation number, to prevent generation conflicts
         List<Integer> generations = new ArrayList<Integer>();
-        for (Descriptor desc : files(table.name, columnFamily, true).keySet())
+        for (String path : DatabaseDescriptor.getAllDataFileLocationsForTable(table.name))
         {
-            generations.add(desc.generation);
-            if (desc.isFromTheFuture())
+            Iterable<Pair<Descriptor, Component>> pairs = files(new File(path));
+            File incrementalsPath = new File(path, "backups");
+            if (incrementalsPath.exists())
+                pairs = Iterables.concat(pairs, files(incrementalsPath));
+
+            for (Pair<Descriptor, Component> pair : pairs)
             {
-                throw new RuntimeException(String.format("Can't open sstables from the future! Current version %s, found file: %s",
-                                                         Descriptor.CURRENT_VERSION, desc));
+                Descriptor desc = pair.left;
+                if (!desc.cfname.equals(columnFamily))
+                    continue;
+                generations.add(desc.generation);
+                if (desc.isFromTheFuture())
+                {
+                    throw new RuntimeException(String.format("Can't open sstables from the future! Current version %s, found file: %s",
+                                                             Descriptor.CURRENT_VERSION, desc));
+                }
             }
         }
         Collections.sort(generations);
@@ -622,34 +633,45 @@ private static Map<Descriptor,Set<Component>> files(String keyspace, final Strin
         final Map<Descriptor,Set<Component>> sstables = new HashMap<Descriptor,Set<Component>>();
         for (String directory : DatabaseDescriptor.getAllDataFileLocationsForTable(keyspace))
         {
-            // NB: we never "accept" a file in the FilenameFilter sense: they are added to the sstable map
-            new File(directory).list(new FilenameFilter()
+            for (Pair<Descriptor, Component> component : files(new File(directory)))
             {
-                public boolean accept(File dir, String name)
+                if (component != null && component.left.cfname.equals(columnFamily))
                 {
-                    Pair<Descriptor,Component> component = SSTable.tryComponentFromFilename(dir, name);
-                    if (component != null && component.left.cfname.equals(columnFamily))
+                    if (includeCompacted || !new File(component.left.filenameFor(Component.COMPACTED_MARKER)).exists())
                     {
-                        if (includeCompacted || !new File(component.left.filenameFor(Component.COMPACTED_MARKER)).exists())
+                        Set<Component> components = sstables.get(component.left);
+                        if (components == null)
                         {
-                            Set<Component> components = sstables.get(component.left);
-                            if (components == null)
-                            {
-                                components = new HashSet<Component>();
-                                sstables.put(component.left, components);
-                            }
-                            components.add(component.right);
+                            components = new HashSet<Component>();
+                            sstables.put(component.left, components);
                         }
-                        else
-                            logger.debug("not including compacted sstable " + component.left.cfname + "-" + component.left.generation);
+                        components.add(component.right);
                     }
-                    return false;
+                    else
+                        logger.debug("not including compacted sstable " + component.left.cfname + "-" + component.left.generation);
                 }
-            });
+            }
         }
         return sstables;
     }
 
+    private static List<Pair<Descriptor, Component>> files(File path)
+    {
+        final List<Pair<Descriptor, Component>> sstables = new ArrayList<Pair<Descriptor, Component>>();
+        // NB: we never "accept" a file in the FilenameFilter sense: they are added to the sstable map
+        path.list(new FilenameFilter()
+        {
+            public boolean accept(File dir, String name)
+            {
+                Pair<Descriptor, Component> pair = SSTable.tryComponentFromFilename(dir, name);
+                if (pair != null)
+                    sstables.add(pair);
+                return false;
+            }
+        });
+        return sstables;
+    }
+
     /**
      * @return the name of the column family
      */