Add SecondaryDB (#15)

* Add secondarydb, initial commit

* copied the wrong test code lol

* restore "except *" catches

Author: wbarnha

rocksdb/_rocksdb.pyx

@@ -1972,13 +1972,17 @@ cdef class DB(object):
     cdef string db_path
 
     def __cinit__(self, db_name, Options opts, dict column_families=None,
-                  read_only=False, *args, **kwargs):
+                  read_only=False, secondary_path=None, *args, **kwargs):
         cdef Status st
         cdef bytes default_cf_name = db.kDefaultColumnFamilyName
         self.wrapped_db = NULL
         self.opts = None
         self.cf_handles = []
         self.cf_options = []
+        if isinstance(secondary_path, str):
+            self.db_path = path_to_string(secondary_path)
+        else:
+            self.db_path = path_to_string(db_name)
 
         if opts.in_use:
             raise InvalidArgument(
@@ -2019,12 +2023,21 @@ cdef class DB(object):
                 self.cf_options.append(cf_options)
         if type(self) != DB:
             return
-        db_path = path_to_string(db_name)
-        if read_only:
+        if isinstance(secondary_path, str):
+            primary_db = path_to_string(db_name)
+            with nogil:
+                st = db.DB_OpenSecondary_ColumnFamilies(
+                    deref(opts.opts),
+                    primary_db,
+                    self.db_path,
+                    self.column_family_descriptors,
+                    &self.column_family_handles,
+                    &self.wrapped_db)
+        elif read_only:
             with nogil:
                 st = db.DB_OpenForReadOnly_ColumnFamilies(
                     deref(opts.opts),
-                    db_path,
+                    self.db_path,
                     self.column_family_descriptors,
                     &self.column_family_handles,
                     &self.wrapped_db,
@@ -2033,7 +2046,7 @@ cdef class DB(object):
             with nogil:
                 st = db.DB_Open_ColumnFamilies(
                     deref(opts.opts),
-                    db_path,
+                    self.db_path,
                     self.column_family_descriptors,
                     &self.column_family_handles,
                     &self.wrapped_db)
@@ -2635,6 +2648,10 @@ cdef class DB(object):
         if copts:
             copts.in_use = False
 
+    def try_catch_up_with_primary(self):
+        with nogil:
+            self.wrapped_db.TryCatchUpWithPrimary()
+
 
 def repair_db(db_name, Options opts):
     cdef Status st

rocksdb/db.pxd

@@ -170,6 +170,7 @@ cdef extern from "rocksdb/db.h" namespace "rocksdb":
         Status DisableFileDeletions() nogil except+
         Status EnableFileDeletions() nogil except+
         Status Close() nogil except+
+        Status TryCatchUpWithPrimary() nogil except+
 
         # TODO: Status GetSortedWalFiles(VectorLogPtr& files)
         # TODO: SequenceNumber GetLatestSequenceNumber()
@@ -195,6 +196,19 @@ cdef extern from "rocksdb/db.h" namespace "rocksdb":
         vector[ColumnFamilyHandle*]*,
         DB**) nogil except+
 
+    cdef Status DB_OpenSecondary "rocksdb::DB::OpenAsSecondary"(
+        const options.Options&,
+        const string&,
+        cpp_bool) nogil except+
+
+    cdef Status DB_OpenSecondary_ColumnFamilies "rocksdb::DB::OpenAsSecondary"(
+        const options.Options&,
+        const string&,  # primary db path
+        const string&,  # secondary db path
+        const vector[ColumnFamilyDescriptor]&,
+        vector[ColumnFamilyHandle*]*,
+        DB**) nogil except+
+
     cdef Status DB_OpenForReadOnly "rocksdb::DB::OpenForReadOnly"(
         const options.Options&,
         const string&,

rocksdb/tests/test_db.py

@@ -735,3 +735,33 @@ def test_compact_range(self):
 
         self.db.compact_range(column_family=self.cf_b)
 
+
+class TestSecondaryDB(TestHelper):
+    def setUp(self):
+        TestHelper.setUp(self)
+        self.db = rocksdb.DB(os.path.join(self.db_loc, "test_primary"), rocksdb.Options(create_if_missing=True))
+        self.secondary_db = rocksdb.DB(os.path.join(self.db_loc, "test_primary"), rocksdb.Options(create_if_missing=True),
+            secondary_path=os.path.join(self.db_loc, "test_secondary"))
+
+    def test_catch_up(self):
+
+        # Just your ordinary test_write_batch test
+
+        batch = rocksdb.WriteBatch()
+        batch.put(b"key", b"v1")
+        batch.delete(b"key")
+        batch.put(b"key", b"v2")
+        batch.put(b"key", b"v3")
+        batch.put(b"a", b"b")
+
+        self.db.write(batch)
+        ref = {b'a': b'b', b'key': b'v3'}
+        ret = self.db.multi_get([b'key', b'a'])
+        self.assertEqual(ref, ret)
+
+        # Now let's try updating the secondary db
+
+        self.secondary_db.try_catch_up_with_primary()
+
+        ret = self.secondary_db.multi_get([b'key', b'a'])
+        self.assertEqual(ref, ret)