Optimize disk usage for disk-backed compaction (#3493)

Delete RocksDB files for every corfu table right after
the checkpointing finishes.

runtime/src/main/java/org/corfudb/runtime/CheckpointWriter.java

@@ -107,6 +107,7 @@ public class CheckpointWriter<T extends ICorfuTable<?, ?>> {
     /** Local ref to the object that we're dumping.
      *  TODO: generalize to all SMR objects.
      */
+    @Getter
     private final T corfuTable;
 
     @Getter

runtime/src/main/java/org/corfudb/runtime/DistributedCheckpointer.java

@@ -172,6 +172,7 @@ private CheckpointingStatus appendCheckpoint(TableName tableName,
             try {
                 CheckpointWriter<ICorfuTable<?,?>> cpw = checkpointWriterFn.apply(tableName);
                 cpw.appendCheckpoint(Optional.of(livenessUpdater));
+                cpw.getCorfuTable().close();
                 returnStatus = StatusType.COMPLETED;
                 break;
             } catch (RuntimeException re) {

runtime/src/main/java/org/corfudb/runtime/collections/ICorfuTable.java

@@ -49,6 +49,8 @@ public interface ICorfuTable<K, V> {
 
     void clear();
 
+    void close();
+
     boolean isTableCached();
 
 }

runtime/src/main/java/org/corfudb/runtime/collections/PersistedStreamingMap.java

@@ -91,12 +91,18 @@ public static Options getPersistedStreamingMapOptions() {
     private final ISerializer serializer;
     private final RocksDB rocksDb;
 
+    private final String absolutePathString;
+
+    private Options options;
+
     public PersistedStreamingMap(@NonNull Path dataPath,
                                  @NonNull Options options,
                                  @NonNull ISerializer serializer,
                                  @NonNull CorfuRuntime corfuRuntime) {
+        this.absolutePathString = dataPath.toFile().getAbsolutePath();
+        this.options = options;
         try {
-            RocksDB.destroyDB(dataPath.toFile().getAbsolutePath(), options);
+            RocksDB.destroyDB(this.absolutePathString, this.options);
             this.rocksDb = RocksDB.open(options, dataPath.toFile().getAbsolutePath());
         } catch (RocksDBException e) {
             throw new UnrecoverableCorfuError(e);
@@ -311,5 +317,13 @@ public Stream<Entry<K, V>> entryStream() {
     @Override
     public void close() {
         this.rocksDb.close();
+
+        // Release disk space
+        try {
+            RocksDB.destroyDB(this.absolutePathString, this.options);
+            log.info("Cleared RocksDB data on {}", absolutePathString);
+        } catch (RocksDBException e) {
+            throw new UnrecoverableCorfuError(e);
+        }
     }
 }

runtime/src/main/java/org/corfudb/runtime/collections/PersistentCorfuTable.java

@@ -52,6 +52,11 @@ public void clear() {
         proxy.logUpdate("clear", false, null);
     }
 
+    @Override
+    public void close() {
+        // NO OP
+    }
+
     @Override
     public boolean isTableCached() {
         return ((MVOCorfuCompileProxy)proxy).isObjectCached();

test/src/test/java/org/corfudb/runtime/DistributedCheckpointerUnitTest.java

@@ -68,6 +68,7 @@ public void setup() {
         doNothing().when(txn).putRecord(any(), any(), any(), any());
         doNothing().when(txn).delete(anyString(), any(TableName.class));
         when(txn.commit()).thenReturn(Timestamp.getDefaultInstance());
+        when(cpw.getCorfuTable()).thenReturn(mock(CorfuTable.class));
     }
 
     @Test

test/src/test/java/org/corfudb/runtime/collections/DiskBackedCorfuClientTest.java

@@ -481,6 +481,21 @@ void verifyCommitAddressInterleavingTxn(@ForAll @Size(SAMPLE_SIZE) Set<String> i
         }
     }
 
+    /**
+     * Verify RocksDB persisted cache is cleaned up
+     */
+    @Property(tries = NUM_OF_TRIES)
+    void verifyPersistedCacheCleanUp() {
+        resetTests();
+        assertThat(persistedCacheLocation).doesNotExist();
+        try (final CorfuTable<String, String> table1 = setupTable(alternateMapName)) {
+            table1.put(defaultNewMapEntry, defaultNewMapEntry);
+            assertThat(persistedCacheLocation).exists();
+            table1.close();
+            assertThat(persistedCacheLocation).doesNotExist();
+        }
+    }
+
     /**
      * A custom generator for a set of {@link Uuid}.
      */