Make RocksDB instance responsible for closing associated ColumnFamilyHandle instances (#7428)

Summary:
- Takes the burden off developer to close ColumnFamilyHandle instances before closing RocksDB instance
- The change is backward-compatible

----
Previously the pattern for working with Column Families was:

```java
try (final ColumnFamilyOptions cfOpts = new ColumnFamilyOptions().optimizeUniversalStyleCompaction()) {

  // list of column family descriptors, first entry must always be default column family
  final List<ColumnFamilyDescriptor> cfDescriptors = Arrays.asList(
      new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY, cfOpts),
      new ColumnFamilyDescriptor("my-first-columnfamily".getBytes(), cfOpts)
  );

  // a list which will hold the handles for the column families once the db is opened
  final List<ColumnFamilyHandle> columnFamilyHandleList =
      new ArrayList<>();

  try (final DBOptions options = new DBOptions()
      .setCreateIfMissing(true)
      .setCreateMissingColumnFamilies(true);
       final RocksDB db = RocksDB.open(options,
           "path/to/do", cfDescriptors,
           columnFamilyHandleList)) {

    try {

      // do something

    } finally {

      // NOTE user must explicitly frees the column family handles before freeing the db
      for (final ColumnFamilyHandle columnFamilyHandle :
          columnFamilyHandleList) {
        columnFamilyHandle.close();
      }
    } // frees the column family options
  }
} // frees the db and the db options
```

With the changes in this PR, the Java user no longer has to worry about manually closing the Column Families, which allows them to write simpler symmetrical create/free oriented code like this:

```java
try (final ColumnFamilyOptions cfOpts = new ColumnFamilyOptions().optimizeUniversalStyleCompaction()) {

  // list of column family descriptors, first entry must always be default column family
  final List<ColumnFamilyDescriptor> cfDescriptors = Arrays.asList(
      new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY, cfOpts),
      new ColumnFamilyDescriptor("my-first-columnfamily".getBytes(), cfOpts)
  );

  // a list which will hold the handles for the column families once the db is opened
  final List<ColumnFamilyHandle> columnFamilyHandleList =
      new ArrayList<>();

  try (final DBOptions options = new DBOptions()
      .setCreateIfMissing(true)
      .setCreateMissingColumnFamilies(true);
       final RocksDB db = RocksDB.open(options,
           "path/to/do", cfDescriptors,
           columnFamilyHandleList)) {

        // do something

    } // frees the column family options, then frees the db and the db options
  }
}
```

**NOTE**: The changes in this PR are backwards API compatible, which means existing code using the original approach will also continue to function correctly.

Pull Request resolved: https://github.com/facebook/rocksdb/pull/7428

Reviewed By: cheng-chang

Differential Revision: D24063348

Pulled By: jay-zhuang

fbshipit-source-id: 648d7526669923128c863ead94516bf4d50ac658

java/src/main/java/org/rocksdb/ColumnFamilyHandle.java

@@ -13,6 +13,12 @@ import java.util.Objects;
  * ColumnFamily Pointers.
  */
 public class ColumnFamilyHandle extends RocksObject {
+  /**
+   * Constructs column family Java object, which operates on underlying native object.
+   *
+   * @param rocksDB db instance associated with this column family
+   * @param nativeHandle native handle to underlying native ColumnFamily object
+   */
   ColumnFamilyHandle(final RocksDB rocksDB,
       final long nativeHandle) {
     super(nativeHandle);

java/src/main/java/org/rocksdb/RocksDB.java

@@ -38,6 +38,8 @@ public class RocksDB extends RocksObject {
     RocksDB.loadLibrary();
   }
 
+  private List<ColumnFamilyHandle> ownedColumnFamilyHandles = new ArrayList<>();
+
   /**
    * Loads the necessary library files.
    * Calling this method twice will have no effect.
@@ -307,9 +309,12 @@ public class RocksDB extends RocksObject {
     db.storeOptionsInstance(options);
 
     for (int i = 1; i < handles.length; i++) {
-      columnFamilyHandles.add(new ColumnFamilyHandle(db, handles[i]));
+      final ColumnFamilyHandle columnFamilyHandle = new ColumnFamilyHandle(db, handles[i]);
+      columnFamilyHandles.add(columnFamilyHandle);
     }
 
+    db.ownedColumnFamilyHandles.addAll(columnFamilyHandles);
+
     return db;
   }
 
@@ -484,9 +489,12 @@ public class RocksDB extends RocksObject {
     db.storeOptionsInstance(options);
 
     for (int i = 1; i < handles.length; i++) {
-      columnFamilyHandles.add(new ColumnFamilyHandle(db, handles[i]));
+      final ColumnFamilyHandle columnFamilyHandle = new ColumnFamilyHandle(db, handles[i]);
+      columnFamilyHandles.add(columnFamilyHandle);
     }
 
+    db.ownedColumnFamilyHandles.addAll(columnFamilyHandles);
+
     return db;
   }
 
@@ -577,9 +585,12 @@ public class RocksDB extends RocksObject {
     db.storeOptionsInstance(options);
 
     for (int i = 1; i < handles.length; i++) {
-      columnFamilyHandles.add(new ColumnFamilyHandle(db, handles[i]));
+      final ColumnFamilyHandle columnFamilyHandle = new ColumnFamilyHandle(db, handles[i]);
+      columnFamilyHandles.add(columnFamilyHandle);
     }
 
+    db.ownedColumnFamilyHandles.addAll(columnFamilyHandles);
+
     return db;
   }
 
@@ -597,6 +608,11 @@ public class RocksDB extends RocksObject {
    * @throws RocksDBException if an error occurs whilst closing.
    */
   public void closeE() throws RocksDBException {
+    for (final ColumnFamilyHandle columnFamilyHandle : ownedColumnFamilyHandles) {
+      columnFamilyHandle.close();
+    }
+    ownedColumnFamilyHandles.clear();
+
     if (owningHandle_.compareAndSet(true, false)) {
       try {
         closeDatabase(nativeHandle_);
@@ -619,6 +635,11 @@ public class RocksDB extends RocksObject {
    */
   @Override
   public void close() {
+    for (final ColumnFamilyHandle columnFamilyHandle : ownedColumnFamilyHandles) {
+      columnFamilyHandle.close();
+    }
+    ownedColumnFamilyHandles.clear();
+
     if (owningHandle_.compareAndSet(true, false)) {
       try {
         closeDatabase(nativeHandle_);
@@ -661,10 +682,12 @@ public class RocksDB extends RocksObject {
   public ColumnFamilyHandle createColumnFamily(
       final ColumnFamilyDescriptor columnFamilyDescriptor)
       throws RocksDBException {
-    return new ColumnFamilyHandle(this, createColumnFamily(nativeHandle_,
-        columnFamilyDescriptor.getName(),
-        columnFamilyDescriptor.getName().length,
-        columnFamilyDescriptor.getOptions().nativeHandle_));
+    final ColumnFamilyHandle columnFamilyHandle = new ColumnFamilyHandle(this,
+        createColumnFamily(nativeHandle_, columnFamilyDescriptor.getName(),
+            columnFamilyDescriptor.getName().length,
+            columnFamilyDescriptor.getOptions().nativeHandle_));
+    ownedColumnFamilyHandles.add(columnFamilyHandle);
+    return columnFamilyHandle;
   }
 
   /**
@@ -688,8 +711,10 @@ public class RocksDB extends RocksObject {
     final List<ColumnFamilyHandle> columnFamilyHandles =
         new ArrayList<>(cfHandles.length);
     for (int i = 0; i < cfHandles.length; i++) {
-      columnFamilyHandles.add(new ColumnFamilyHandle(this, cfHandles[i]));
+      final ColumnFamilyHandle columnFamilyHandle = new ColumnFamilyHandle(this, cfHandles[i]);
+      columnFamilyHandles.add(columnFamilyHandle);
     }
+    ownedColumnFamilyHandles.addAll(columnFamilyHandles);
     return columnFamilyHandles;
   }
 
@@ -719,8 +744,10 @@ public class RocksDB extends RocksObject {
     final List<ColumnFamilyHandle> columnFamilyHandles =
         new ArrayList<>(cfHandles.length);
     for (int i = 0; i < cfHandles.length; i++) {
-      columnFamilyHandles.add(new ColumnFamilyHandle(this, cfHandles[i]));
+      final ColumnFamilyHandle columnFamilyHandle = new ColumnFamilyHandle(this, cfHandles[i]);
+      columnFamilyHandles.add(columnFamilyHandle);
     }
+    ownedColumnFamilyHandles.addAll(columnFamilyHandles);
     return columnFamilyHandles;
   }
 
@@ -753,7 +780,22 @@ public class RocksDB extends RocksObject {
     dropColumnFamilies(nativeHandle_, cfHandles);
   }
 
-  //TODO(AR) what about DestroyColumnFamilyHandle
+  /**
+   * Deletes native column family handle of given {@link ColumnFamilyHandle} Java object
+   * and removes reference from {@link RocksDB#ownedColumnFamilyHandles}.
+   *
+   * @param columnFamilyHandle column family handle object.
+   */
+  public void destroyColumnFamilyHandle(final ColumnFamilyHandle columnFamilyHandle) {
+    for (int i = 0; i < ownedColumnFamilyHandles.size(); ++i) {
+      final ColumnFamilyHandle ownedHandle = ownedColumnFamilyHandles.get(i);
+      if (ownedHandle.equals(columnFamilyHandle)) {
+        columnFamilyHandle.close();
+        ownedColumnFamilyHandles.remove(i);
+        return;
+      }
+    }
+  }
 
   /**
    * Set the database entry for "key" to "value".
@@ -4479,7 +4521,6 @@ public class RocksDB extends RocksObject {
       final long handle, final long cfHandle) throws RocksDBException;
   private native void dropColumnFamilies(final long handle,
       final long[] cfHandles) throws RocksDBException;
-  //TODO(AR) best way to express DestroyColumnFamilyHandle? ...maybe in ColumnFamilyHandle?
   private native void put(final long handle, final byte[] key,
       final int keyOffset, final int keyLength, final byte[] value,
       final int valueOffset, int valueLength) throws RocksDBException;

java/src/test/java/org/rocksdb/CompactionFilterFactoryTest.java

@@ -39,29 +39,22 @@ public class CompactionFilterFactoryTest {
 
       final List<ColumnFamilyHandle> cfHandles = new ArrayList<>();
 
-      try (final RocksDB rocksDb = RocksDB.open(options,
-               dbFolder.getRoot().getAbsolutePath(), cfNames, cfHandles);
-      ) {
-        try {
-          final byte[] key1 = "key1".getBytes();
-          final byte[] key2 = "key2".getBytes();
+      try (final RocksDB rocksDb =
+               RocksDB.open(options, dbFolder.getRoot().getAbsolutePath(), cfNames, cfHandles)) {
+        final byte[] key1 = "key1".getBytes();
+        final byte[] key2 = "key2".getBytes();
 
-          final byte[] value1 = "value1".getBytes();
-          final byte[] value2 = new byte[0];
+        final byte[] value1 = "value1".getBytes();
+        final byte[] value2 = new byte[0];
 
-          rocksDb.put(cfHandles.get(1), key1, value1);
-          rocksDb.put(cfHandles.get(1), key2, value2);
+        rocksDb.put(cfHandles.get(1), key1, value1);
+        rocksDb.put(cfHandles.get(1), key2, value2);
 
-          rocksDb.compactRange(cfHandles.get(1));
+        rocksDb.compactRange(cfHandles.get(1));
 
-          assertThat(rocksDb.get(cfHandles.get(1), key1)).isEqualTo(value1);
-          final boolean exists = rocksDb.keyMayExist(cfHandles.get(1), key2, null);
-          assertThat(exists).isFalse();
-        } finally {
-          for (final ColumnFamilyHandle cfHandle : cfHandles) {
-            cfHandle.close();
-          }
-        }
+        assertThat(rocksDb.get(cfHandles.get(1), key1)).isEqualTo(value1);
+        final boolean exists = rocksDb.keyMayExist(cfHandles.get(1), key2, null);
+        assertThat(exists).isFalse();
       }
     }
   }