CASSANDRA-20068: Add flag to avoid invalidating key cache on sstable deletions

src/java/org/apache/cassandra/config/Config.java

@@ -455,6 +455,7 @@ public static class SSTableConfig
     public volatile DataStorageSpec.IntMebibytesBound sstable_preemptive_open_interval = new DataStorageSpec.IntMebibytesBound("50MiB");
 
     public volatile boolean key_cache_migrate_during_compaction = true;
+    public volatile boolean key_cache_invalidate_after_sstable_deletion = false;
     public volatile int key_cache_keys_to_save = Integer.MAX_VALUE;
     @Replaces(oldName = "key_cache_size_in_mb", converter = Converters.MEBIBYTES_DATA_STORAGE_LONG, deprecated = true)
     public DataStorageSpec.LongMebibytesBound key_cache_size = null;

src/java/org/apache/cassandra/config/DatabaseDescriptor.java

@@ -3760,6 +3760,16 @@ public static void setMigrateKeycacheOnCompaction(boolean migrateCacheEntry)
         conf.key_cache_migrate_during_compaction = migrateCacheEntry;
     }
 
+    public static boolean shouldInvalidateKeycacheOnSSTableDeletion()
+    {
+        return conf.key_cache_invalidate_after_sstable_deletion;
+    }
+
+    public static void setInvalidateKeycacheOnSSTableDeletion(boolean invalidateCacheEntry)
+    {
+        conf.key_cache_invalidate_after_sstable_deletion = invalidateCacheEntry;
+    }
+
     /** This method can return negative number for disabled */
     public static int getSSTablePreemptiveOpenIntervalInMiB()
     {

src/java/org/apache/cassandra/io/sstable/format/big/BigFormat.java

@@ -247,13 +247,16 @@ public void delete(Descriptor desc)
     {
         try
         {
-            // remove key cache entries for the sstable being deleted
-            Iterator<KeyCacheKey> it = CacheService.instance.keyCache.keyIterator();
-            while (it.hasNext())
+            if (DatabaseDescriptor.shouldInvalidateKeycacheOnSSTableDeletion())
             {
-                KeyCacheKey key = it.next();
-                if (key.desc.equals(desc))
-                    it.remove();
+                // remove key cache entries for the sstable being deleted
+                Iterator<KeyCacheKey> it = CacheService.instance.keyCache.keyIterator();
+                while (it.hasNext())
+                {
+                    KeyCacheKey key = it.next();
+                    if (key.desc.equals(desc))
+                        it.remove();
+                }
             }
 
             delete(desc, Lists.newArrayList(Sets.intersection(allComponents(), desc.discoverComponents())));

src/java/org/apache/cassandra/service/StorageService.java

@@ -6646,6 +6646,16 @@ public void setMigrateKeycacheOnCompaction(boolean invalidateKeyCacheOnCompactio
         DatabaseDescriptor.setMigrateKeycacheOnCompaction(invalidateKeyCacheOnCompaction);
     }
 
+    public boolean getInvalidateKeycacheOnSSTableDeletion()
+    {
+        return DatabaseDescriptor.shouldInvalidateKeycacheOnSSTableDeletion();
+    }
+
+    public void setInvalidateKeycacheOnSSTableDeletion(boolean invalidate)
+    {
+        DatabaseDescriptor.setInvalidateKeycacheOnSSTableDeletion(invalidate);
+    }
+
     public int getTombstoneWarnThreshold()
     {
         return DatabaseDescriptor.getTombstoneWarnThreshold();

src/java/org/apache/cassandra/service/StorageServiceMBean.java

@@ -833,6 +833,8 @@ default int upgradeSSTables(String keyspaceName, boolean excludeCurrentVersion,
 
     public boolean getMigrateKeycacheOnCompaction();
     public void setMigrateKeycacheOnCompaction(boolean invalidateKeyCacheOnCompaction);
+    public boolean getInvalidateKeycacheOnSSTableDeletion();
+    public void setInvalidateKeycacheOnSSTableDeletion(boolean invalidate);
 
     public int getConcurrentViewBuilders();
     public void setConcurrentViewBuilders(int value);

test/unit/org/apache/cassandra/io/sstable/keycache/KeyCacheTest.java

@@ -350,14 +350,14 @@ private void testKeyCache(String cf) throws ExecutionException, InterruptedExcep
 
         LifecycleTransaction.waitForDeletions();
 
-        // after releasing the reference this should drop to 2
-        assertKeyCacheSize(2, KEYSPACE1, cf);
+        // after releasing the reference this should drop to 2, unless we don't invalidate keycache on sstable deletion
+        assertKeyCacheSize(DatabaseDescriptor.shouldInvalidateKeycacheOnSSTableDeletion() ? 2 : 4, KEYSPACE1, cf);
 
         // re-read same keys to verify that key cache didn't grow further
         Util.getAll(Util.cmd(cfs, "key1").build());
         Util.getAll(Util.cmd(cfs, "key2").build());
 
-        assertKeyCacheSize( 2, KEYSPACE1, cf);
+        assertKeyCacheSize( DatabaseDescriptor.shouldInvalidateKeycacheOnSSTableDeletion() ? 2 : 4, KEYSPACE1, cf);
     }
 
     @Test