Fix cleanup not deleting index entries

patch by jbellis; reviewed by slebresne for CASSANDRA-4379
apache · Jun 27, 2012 · 8a3bb80 · 8a3bb80
1 parent 54c647f
commit 8a3bb80
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 3 deletions.
diff --git a/CHANGES.txt b/CHANGES.txt
@@ -1,4 +1,5 @@
 1.1.2
+ * Fix cleanup not deleting index entries (CASSANDRA-4379)
  * Use correct partitioner when saving + loading caches (CASSANDRA-4331)
  * Check schema before trying to export sstable (CASSANDRA-2760)
  * Raise a meaningful exception instead of NPE when PFS encounters

diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
@@ -683,16 +683,19 @@ private void doCleanupCompaction(ColumnFamilyStore cfs, Collection<SSTableReader
         assert !cfs.isIndex();
         Table table = cfs.table;
         Collection<Range<Token>> ranges = StorageService.instance.getLocalRanges(table.name);
-        boolean isCommutative = cfs.metadata.getDefaultValidator().isCommutative();
         if (ranges.isEmpty())
         {
             logger.info("Cleanup cannot run before a node has joined the ring");
             return;
         }
 
+        boolean isCommutative = cfs.metadata.getDefaultValidator().isCommutative();
+        Collection<ByteBuffer> indexedColumns = cfs.indexManager.getIndexedColumns();
+
         for (SSTableReader sstable : sstables)
         {
-            if (!new Bounds<Token>(sstable.first.token, sstable.last.token).intersects(ranges))
+            if (indexedColumns.isEmpty()
+                && !new Bounds<Token>(sstable.first.token, sstable.last.token).intersects(ranges))
             {
                 cfs.replaceCompactedSSTables(Arrays.asList(sstable), Collections.<SSTableReader>emptyList(), OperationType.CLEANUP);
                 continue;
@@ -720,7 +723,6 @@ private void doCleanupCompaction(ColumnFamilyStore cfs, Collection<SSTableReader
 
             SSTableScanner scanner = sstable.getDirectScanner();
             long rowsRead = 0;
-            Collection<ByteBuffer> indexedColumns = cfs.indexManager.getIndexedColumns();
             List<IColumn> indexedColumnsInRow = null;
 
             CleanupInfo ci = new CleanupInfo(sstable, scanner);