Add DB::OpenAsSecondary to RocksJava (#7047)

Summary:
Closes https://github.com/facebook/rocksdb/issues/5852
Pull Request resolved: https://github.com/facebook/rocksdb/pull/7047

Reviewed By: cheng-chang

Differential Revision: D22335162

Pulled By: pdillinger

fbshipit-source-id: 75f3c524deccea7ebc0ad288da41f1ea81406c1c

java/Makefile

@@ -159,6 +159,7 @@ JAVA_TESTS = \
 	org.rocksdb.RocksIteratorTest\
 	org.rocksdb.RocksMemEnvTest\
 	org.rocksdb.util.SizeUnitTest\
+	org.rocksdb.SecondaryDBTest\
 	org.rocksdb.SliceTest\
 	org.rocksdb.SnapshotTest\
 	org.rocksdb.SstFileManagerTest\

java/rocksjni/rocksjni.cc

@@ -208,6 +208,72 @@ jlongArray Java_org_rocksdb_RocksDB_open__JLjava_lang_String_2_3_3B_3J(
           ROCKSDB_NAMESPACE::DB::Open);
 }
 
+/*
+ * Class:     org_rocksdb_RocksDB
+ * Method:    openAsSecondary
+ * Signature: (JLjava/lang/String;Ljava/lang/String;)J
+ */
+jlong Java_org_rocksdb_RocksDB_openAsSecondary__JLjava_lang_String_2Ljava_lang_String_2(
+    JNIEnv* env, jclass, jlong jopt_handle, jstring jdb_path,
+    jstring jsecondary_db_path) {
+  const char* secondary_db_path =
+      env->GetStringUTFChars(jsecondary_db_path, nullptr);
+  if (secondary_db_path == nullptr) {
+    // exception thrown: OutOfMemoryError
+    return 0;
+  }
+
+  jlong db_handle = rocksdb_open_helper(
+      env, jopt_handle, jdb_path,
+      [secondary_db_path](const ROCKSDB_NAMESPACE::Options& options,
+                          const std::string& db_path,
+                          ROCKSDB_NAMESPACE::DB** db) {
+        return ROCKSDB_NAMESPACE::DB::OpenAsSecondary(options, db_path,
+                                                      secondary_db_path, db);
+      });
+
+  // we have now finished with secondary_db_path
+  env->ReleaseStringUTFChars(jsecondary_db_path, secondary_db_path);
+
+  return db_handle;
+}
+
+/*
+ * Class:     org_rocksdb_RocksDB
+ * Method:    openAsSecondary
+ * Signature: (JLjava/lang/String;Ljava/lang/String;[[B[J)[J
+ */
+jlongArray
+Java_org_rocksdb_RocksDB_openAsSecondary__JLjava_lang_String_2Ljava_lang_String_2_3_3B_3J(
+    JNIEnv* env, jclass, jlong jopt_handle, jstring jdb_path,
+    jstring jsecondary_db_path, jobjectArray jcolumn_names,
+    jlongArray jcolumn_options) {
+  const char* secondary_db_path =
+      env->GetStringUTFChars(jsecondary_db_path, nullptr);
+  if (secondary_db_path == nullptr) {
+    // exception thrown: OutOfMemoryError
+    return nullptr;
+  }
+
+  jlongArray jhandles = rocksdb_open_helper(
+      env, jopt_handle, jdb_path, jcolumn_names, jcolumn_options,
+      [secondary_db_path](
+          const ROCKSDB_NAMESPACE::DBOptions& options,
+          const std::string& db_path,
+          const std::vector<ROCKSDB_NAMESPACE::ColumnFamilyDescriptor>&
+              column_families,
+          std::vector<ROCKSDB_NAMESPACE::ColumnFamilyHandle*>* handles,
+          ROCKSDB_NAMESPACE::DB** db) {
+        return ROCKSDB_NAMESPACE::DB::OpenAsSecondary(
+            options, db_path, secondary_db_path, column_families, handles, db);
+      });
+
+  // we have now finished with secondary_db_path
+  env->ReleaseStringUTFChars(jsecondary_db_path, secondary_db_path);
+
+  return jhandles;
+}
+
 /*
  * Class:     org_rocksdb_RocksDB
  * Method:    disposeInternal
@@ -3320,6 +3386,20 @@ void Java_org_rocksdb_RocksDB_endTrace(JNIEnv* env, jobject, jlong jdb_handle) {
   }
 }
 
+/*
+ * Class:     org_rocksdb_RocksDB
+ * Method:    tryCatchUpWithPrimary
+ * Signature: (J)V
+ */
+void Java_org_rocksdb_RocksDB_tryCatchUpWithPrimary(JNIEnv* env, jobject,
+                                                    jlong jdb_handle) {
+  auto* db = reinterpret_cast<ROCKSDB_NAMESPACE::DB*>(jdb_handle);
+  auto s = db->TryCatchUpWithPrimary();
+  if (!s.ok()) {
+    ROCKSDB_NAMESPACE::RocksDBExceptionJni::ThrowNew(env, s);
+  }
+}
+
 /*
  * Class:     org_rocksdb_RocksDB
  * Method:    destroyDB

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

@@ -437,6 +437,99 @@ public class RocksDB extends RocksObject {
     return db;
   }
 
+  /**
+   * Open DB as secondary instance with only the default column family.
+   *
+   * The secondary instance can dynamically tail the MANIFEST of
+   * a primary that must have already been created. User can call
+   * {@link #tryCatchUpWithPrimary()} to make the secondary instance catch up
+   * with primary (WAL tailing is NOT supported now) whenever the user feels
+   * necessary. Column families created by the primary after the secondary
+   * instance starts are currently ignored by the secondary instance.
+   * Column families opened by secondary and dropped by the primary will be
+   * dropped by secondary as well. However the user of the secondary instance
+   * can still access the data of such dropped column family as long as they
+   * do not destroy the corresponding column family handle.
+   * WAL tailing is not supported at present, but will arrive soon.
+   *
+   * @param options the options to open the secondary instance.
+   * @param path the path to the primary RocksDB instance.
+   * @param secondaryPath points to a directory where the secondary instance
+   *    stores its info log
+   *
+   * @return a {@link RocksDB} instance on success, null if the specified
+   *     {@link RocksDB} can not be opened.
+   *
+   * @throws RocksDBException thrown if error happens in underlying
+   *    native library.
+   */
+  public static RocksDB openAsSecondary(final Options options, final String path,
+      final String secondaryPath) throws RocksDBException {
+    // when non-default Options is used, keeping an Options reference
+    // in RocksDB can prevent Java to GC during the life-time of
+    // the currently-created RocksDB.
+    final RocksDB db = new RocksDB(openAsSecondary(options.nativeHandle_, path, secondaryPath));
+    db.storeOptionsInstance(options);
+    return db;
+  }
+
+  /**
+   * Open DB as secondary instance with column families.
+   * You can open a subset of column families in secondary mode.
+   *
+   * The secondary instance can dynamically tail the MANIFEST of
+   * a primary that must have already been created. User can call
+   * {@link #tryCatchUpWithPrimary()} to make the secondary instance catch up
+   * with primary (WAL tailing is NOT supported now) whenever the user feels
+   * necessary. Column families created by the primary after the secondary
+   * instance starts are currently ignored by the secondary instance.
+   * Column families opened by secondary and dropped by the primary will be
+   * dropped by secondary as well. However the user of the secondary instance
+   * can still access the data of such dropped column family as long as they
+   * do not destroy the corresponding column family handle.
+   * WAL tailing is not supported at present, but will arrive soon.
+   *
+   * @param options the options to open the secondary instance.
+   * @param path the path to the primary RocksDB instance.
+   * @param secondaryPath points to a directory where the secondary instance
+   *    stores its info log.
+   * @param columnFamilyDescriptors list of column family descriptors
+   * @param columnFamilyHandles will be filled with ColumnFamilyHandle instances
+   *     on open.
+   *
+   * @return a {@link RocksDB} instance on success, null if the specified
+   *     {@link RocksDB} can not be opened.
+   *
+   * @throws RocksDBException thrown if error happens in underlying
+   *    native library.
+   */
+  public static RocksDB openAsSecondary(final DBOptions options, final String path,
+      final String secondaryPath, final List<ColumnFamilyDescriptor> columnFamilyDescriptors,
+      final List<ColumnFamilyHandle> columnFamilyHandles) throws RocksDBException {
+    // when non-default Options is used, keeping an Options reference
+    // in RocksDB can prevent Java to GC during the life-time of
+    // the currently-created RocksDB.
+
+    final byte[][] cfNames = new byte[columnFamilyDescriptors.size()][];
+    final long[] cfOptionHandles = new long[columnFamilyDescriptors.size()];
+    for (int i = 0; i < columnFamilyDescriptors.size(); i++) {
+      final ColumnFamilyDescriptor cfDescriptor = columnFamilyDescriptors.get(i);
+      cfNames[i] = cfDescriptor.getName();
+      cfOptionHandles[i] = cfDescriptor.getOptions().nativeHandle_;
+    }
+
+    final long[] handles =
+        openAsSecondary(options.nativeHandle_, path, secondaryPath, cfNames, cfOptionHandles);
+    final RocksDB db = new RocksDB(handles[0]);
+    db.storeOptionsInstance(options);
+
+    for (int i = 1; i < handles.length; i++) {
+      columnFamilyHandles.add(new ColumnFamilyHandle(db, handles[i]));
+    }
+
+    return db;
+  }
+
   /**
    * This is similar to {@link #close()} except that it
    * throws an exception if any error occurs.
@@ -4166,6 +4259,25 @@ public class RocksDB extends RocksObject {
     endTrace(nativeHandle_);
   }
 
+  /**
+   * Make the secondary instance catch up with the primary by tailing and
+   * replaying the MANIFEST and WAL of the primary.
+   * Column families created by the primary after the secondary instance starts
+   * will be ignored unless the secondary instance closes and restarts with the
+   * newly created column families.
+   * Column families that exist before secondary instance starts and dropped by
+   * the primary afterwards will be marked as dropped. However, as long as the
+   * secondary instance does not delete the corresponding column family
+   * handles, the data of the column family is still accessible to the
+   * secondary.
+   *
+   * @throws RocksDBException thrown if error happens in underlying
+   *     native library.
+   */
+  public void tryCatchUpWithPrimary() throws RocksDBException {
+    tryCatchUpWithPrimary(nativeHandle_);
+  }
+
   /**
    * Delete files in multiple ranges at once.
    * Delete files in a lot of ranges one at a time can be slow, use this API for
@@ -4289,6 +4401,13 @@ public class RocksDB extends RocksObject {
       final long[] columnFamilyOptions
   ) throws RocksDBException;
 
+  private native static long openAsSecondary(final long optionsHandle, final String path,
+      final String secondaryPath) throws RocksDBException;
+
+  private native static long[] openAsSecondary(final long optionsHandle, final String path,
+      final String secondaryPath, final byte[][] columnFamilyNames,
+      final long[] columnFamilyOptions) throws RocksDBException;
+
   @Override protected native void disposeInternal(final long handle);
 
   private native static void closeDatabase(final long handle)
@@ -4535,6 +4654,7 @@ public class RocksDB extends RocksObject {
   private native void startTrace(final long handle, final long maxTraceFileSize,
       final long traceWriterHandle) throws RocksDBException;
   private native void endTrace(final long handle) throws RocksDBException;
+  private native void tryCatchUpWithPrimary(final long handle) throws RocksDBException;
   private native void deleteFilesInRanges(long handle, long cfHandle, final byte[][] ranges,
       boolean include_end) throws RocksDBException;
 

java/src/test/java/org/rocksdb/SecondaryDBTest.java

@@ -0,0 +1,135 @@
+// Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under both the GPLv2 (found in the
+//  COPYING file in the root directory) and Apache 2.0 License
+//  (found in the LICENSE.Apache file in the root directory).
+
+package org.rocksdb;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.junit.ClassRule;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+public class SecondaryDBTest {
+  @ClassRule
+  public static final RocksNativeLibraryResource ROCKS_NATIVE_LIBRARY_RESOURCE =
+      new RocksNativeLibraryResource();
+
+  @Rule public TemporaryFolder dbFolder = new TemporaryFolder();
+
+  @Rule public TemporaryFolder secondaryDbFolder = new TemporaryFolder();
+
+  @Test
+  public void openAsSecondary() throws RocksDBException {
+    try (final Options options = new Options().setCreateIfMissing(true);
+         final RocksDB db = RocksDB.open(options, dbFolder.getRoot().getAbsolutePath())) {
+      db.put("key1".getBytes(), "value1".getBytes());
+      db.put("key2".getBytes(), "value2".getBytes());
+      db.put("key3".getBytes(), "value3".getBytes());
+
+      // open secondary
+      try (final Options secondaryOptions = new Options();
+           final RocksDB secondaryDb =
+               RocksDB.openAsSecondary(secondaryOptions, dbFolder.getRoot().getAbsolutePath(),
+                   secondaryDbFolder.getRoot().getAbsolutePath())) {
+        assertThat(secondaryDb.get("key1".getBytes())).isEqualTo("value1".getBytes());
+        assertThat(secondaryDb.get("key2".getBytes())).isEqualTo("value2".getBytes());
+        assertThat(secondaryDb.get("key3".getBytes())).isEqualTo("value3".getBytes());
+
+        // write to primary
+        db.put("key4".getBytes(), "value4".getBytes());
+        db.put("key5".getBytes(), "value5".getBytes());
+        db.put("key6".getBytes(), "value6".getBytes());
+
+        // tell secondary to catch up
+        secondaryDb.tryCatchUpWithPrimary();
+
+        db.put("key7".getBytes(), "value7".getBytes());
+
+        // check secondary
+        assertThat(secondaryDb.get("key4".getBytes())).isEqualTo("value4".getBytes());
+        assertThat(secondaryDb.get("key5".getBytes())).isEqualTo("value5".getBytes());
+        assertThat(secondaryDb.get("key6".getBytes())).isEqualTo("value6".getBytes());
+
+        assertThat(secondaryDb.get("key7".getBytes())).isNull();
+      }
+    }
+  }
+
+  @Test
+  public void openAsSecondaryColumnFamilies() throws RocksDBException {
+    try (final ColumnFamilyOptions cfOpts = new ColumnFamilyOptions()) {
+      final List<ColumnFamilyDescriptor> cfDescriptors = new ArrayList<>();
+      cfDescriptors.add(new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY, cfOpts));
+      cfDescriptors.add(new ColumnFamilyDescriptor("cf1".getBytes(), cfOpts));
+
+      final List<ColumnFamilyHandle> cfHandles = new ArrayList<>();
+
+      try (final DBOptions options =
+               new DBOptions().setCreateIfMissing(true).setCreateMissingColumnFamilies(true);
+           final RocksDB db = RocksDB.open(
+               options, dbFolder.getRoot().getAbsolutePath(), cfDescriptors, cfHandles)) {
+        try {
+          final ColumnFamilyHandle cf1 = cfHandles.get(1);
+
+          db.put(cf1, "key1".getBytes(), "value1".getBytes());
+          db.put(cf1, "key2".getBytes(), "value2".getBytes());
+          db.put(cf1, "key3".getBytes(), "value3".getBytes());
+
+          final List<ColumnFamilyHandle> secondaryCfHandles = new ArrayList<>();
+
+          // open secondary
+          try (final DBOptions secondaryOptions = new DBOptions();
+               final RocksDB secondaryDb =
+                   RocksDB.openAsSecondary(secondaryOptions, dbFolder.getRoot().getAbsolutePath(),
+                       secondaryDbFolder.getRoot().getAbsolutePath(), cfDescriptors,
+                       secondaryCfHandles)) {
+            try {
+              final ColumnFamilyHandle secondaryCf1 = secondaryCfHandles.get(1);
+
+              assertThat(secondaryDb.get(secondaryCf1, "key1".getBytes()))
+                  .isEqualTo("value1".getBytes());
+              assertThat(secondaryDb.get(secondaryCf1, "key2".getBytes()))
+                  .isEqualTo("value2".getBytes());
+              assertThat(secondaryDb.get(secondaryCf1, "key3".getBytes()))
+                  .isEqualTo("value3".getBytes());
+
+              // write to primary
+              db.put(cf1, "key4".getBytes(), "value4".getBytes());
+              db.put(cf1, "key5".getBytes(), "value5".getBytes());
+              db.put(cf1, "key6".getBytes(), "value6".getBytes());
+
+              // tell secondary to catch up
+              secondaryDb.tryCatchUpWithPrimary();
+
+              db.put(cf1, "key7".getBytes(), "value7".getBytes());
+
+              // check secondary
+              assertThat(secondaryDb.get(secondaryCf1, "key4".getBytes()))
+                  .isEqualTo("value4".getBytes());
+              assertThat(secondaryDb.get(secondaryCf1, "key5".getBytes()))
+                  .isEqualTo("value5".getBytes());
+              assertThat(secondaryDb.get(secondaryCf1, "key6".getBytes()))
+                  .isEqualTo("value6".getBytes());
+
+              assertThat(secondaryDb.get(secondaryCf1, "key7".getBytes())).isNull();
+
+            } finally {
+              for (final ColumnFamilyHandle secondaryCfHandle : secondaryCfHandles) {
+                secondaryCfHandle.close();
+              }
+            }
+          }
+        } finally {
+          for (final ColumnFamilyHandle cfHandle : cfHandles) {
+            cfHandle.close();
+          }
+        }
+      }
+    }
+  }
+}