add hash prefixing to rocksdb

src/storage/rocksdb.cpp

@@ -3,6 +3,8 @@
 #include <sstream>
 #include <mutex>
 #include <unistd.h>
+#include "../server.h"
+#include "../cluster.h"
 #include "rocksdbfactor_internal.h"
 
 static const char keyprefix[] = INTERNAL_KEY_PREFIX;
@@ -10,6 +12,13 @@ static const char keyprefix[] = INTERNAL_KEY_PREFIX;
 rocksdb::Options DefaultRocksDBOptions();
 extern "C" pid_t gettid();
 
+std::string prefixKey(const char *key, size_t cchKey)
+{
+    unsigned int hash = keyHashSlot(key, cchKey);
+    char *hash_char = (char *)&hash;
+    return std::string(hash_char + (sizeof(unsigned int) - 2), 2) + std::string(key, cchKey);
+}
+
 bool FInternalKey(const char *key, size_t cch)
 {
     if (cch >= sizeof(INTERNAL_KEY_PREFIX))
@@ -32,10 +41,11 @@ void RocksDBStorageProvider::insert(const char *key, size_t cchKey, void *data,
 {
     rocksdb::Status status;
     std::unique_lock<fastlock> l(m_lock);
+    std::string prefixed_key = prefixKey(key, cchKey);
     if (m_spbatch != nullptr)
-        status = m_spbatch->Put(m_spcolfamily.get(), rocksdb::Slice(key, cchKey), rocksdb::Slice((const char*)data, cb));
+        status = m_spbatch->Put(m_spcolfamily.get(), rocksdb::Slice(prefixed_key), rocksdb::Slice((const char*)data, cb));
     else
-        status = m_spdb->Put(WriteOptions(), m_spcolfamily.get(), rocksdb::Slice(key, cchKey), rocksdb::Slice((const char*)data, cb));
+        status = m_spdb->Put(WriteOptions(), m_spcolfamily.get(), rocksdb::Slice(prefixed_key), rocksdb::Slice((const char*)data, cb));
     if (!status.ok())
         throw status.ToString();
 
@@ -59,7 +69,8 @@ void RocksDBStorageProvider::bulkInsert(char **rgkeys, size_t *rgcbkeys, char **
         // Insert rows into the SST file, note that inserted keys must be 
         // strictly increasing (based on options.comparator)
         for (size_t ielem = 0; ielem < celem; ++ielem) {
-            s = sst_file_writer.Put(rocksdb::Slice(rgkeys[ielem], rgcbkeys[ielem]), rocksdb::Slice(rgvals[ielem], rgcbvals[ielem]));
+            std::string prefixed_key = prefixKey(rgkeys[ielem], rgcbkeys[ielem]);
+            s = sst_file_writer.Put(rocksdb::Slice(prefixed_key), rocksdb::Slice(rgvals[ielem], rgcbvals[ielem]));
             if (!s.ok()) {
                 unlink(file_path.c_str());
                 goto LFallback;
@@ -87,7 +98,8 @@ void RocksDBStorageProvider::bulkInsert(char **rgkeys, size_t *rgcbkeys, char **
     LFallback:
         auto spbatch = std::make_unique<rocksdb::WriteBatch>();
         for (size_t ielem = 0; ielem < celem; ++ielem) {
-            spbatch->Put(m_spcolfamily.get(), rocksdb::Slice(rgkeys[ielem], rgcbkeys[ielem]), rocksdb::Slice(rgvals[ielem], rgcbvals[ielem]));
+            std::string prefixed_key = prefixKey(rgkeys[ielem], rgcbkeys[ielem]);
+            spbatch->Put(m_spcolfamily.get(), rocksdb::Slice(prefixed_key), rocksdb::Slice(rgvals[ielem], rgcbvals[ielem]));
         }
         m_spdb->Write(WriteOptions(), spbatch.get());
     }
@@ -100,15 +112,16 @@ bool RocksDBStorageProvider::erase(const char *key, size_t cchKey)
 {
     rocksdb::Status status;
     std::unique_lock<fastlock> l(m_lock);
-    if (!FKeyExists(key, cchKey))
+    std::string prefixed_key = prefixKey(key, cchKey);
+    if (!FKeyExists(prefixed_key))
         return false;
     if (m_spbatch != nullptr)
     {
-        status = m_spbatch->Delete(m_spcolfamily.get(), rocksdb::Slice(key, cchKey));
+        status = m_spbatch->Delete(m_spcolfamily.get(), rocksdb::Slice(prefixed_key));
     }
     else
     {
-        status = m_spdb->Delete(WriteOptions(), m_spcolfamily.get(), rocksdb::Slice(key, cchKey));
+        status = m_spdb->Delete(WriteOptions(), m_spcolfamily.get(), rocksdb::Slice(prefixed_key));
     }
     if (status.ok())
         --m_count;
@@ -118,7 +131,8 @@ bool RocksDBStorageProvider::erase(const char *key, size_t cchKey)
 void RocksDBStorageProvider::retrieve(const char *key, size_t cchKey, callbackSingle fn) const
 {
     rocksdb::PinnableSlice slice;
-    auto status = m_spdb->Get(ReadOptions(), m_spcolfamily.get(), rocksdb::Slice(key, cchKey), &slice);
+    std::string prefixed_key = prefixKey(key, cchKey);
+    auto status = m_spdb->Get(ReadOptions(), m_spcolfamily.get(), rocksdb::Slice(prefixed_key), &slice);
     if (status.ok())
         fn(key, cchKey, slice.data(), slice.size());
 }
@@ -151,10 +165,10 @@ bool RocksDBStorageProvider::enumerate(callback fn) const
     std::unique_ptr<rocksdb::Iterator> it = std::unique_ptr<rocksdb::Iterator>(m_spdb->NewIterator(ReadOptions(), m_spcolfamily.get()));
     size_t count = 0;
     for (it->SeekToFirst(); it->Valid(); it->Next()) {
-        if (FInternalKey(it->key().data(), it->key().size()))
+        if (FInternalKey(it->key().data()+2, it->key().size()-2))
             continue;
         ++count;
-        bool fContinue = fn(it->key().data(), it->key().size(), it->value().data(), it->value().size());
+        bool fContinue = fn(it->key().data()+2, it->key().size()-2, it->value().data(), it->value().size());
         if (!fContinue)
             break;
     }
@@ -227,10 +241,10 @@ void RocksDBStorageProvider::flush()
     m_spdb->SyncWAL();
 }
 
-bool RocksDBStorageProvider::FKeyExists(const char *key, size_t cch) const
+bool RocksDBStorageProvider::FKeyExists(std::string& key) const
 {
     rocksdb::PinnableSlice slice;
     if (m_spbatch)
-        return m_spbatch->GetFromBatchAndDB(m_spdb.get(), ReadOptions(), m_spcolfamily.get(), rocksdb::Slice(key, cch), &slice).ok();
-    return m_spdb->Get(ReadOptions(), m_spcolfamily.get(), rocksdb::Slice(key, cch), &slice).ok();
+        return m_spbatch->GetFromBatchAndDB(m_spdb.get(), ReadOptions(), m_spcolfamily.get(), rocksdb::Slice(key), &slice).ok();
+    return m_spdb->Get(ReadOptions(), m_spcolfamily.get(), rocksdb::Slice(key), &slice).ok();
 }

src/storage/rocksdb.h

@@ -48,7 +48,7 @@ public:
     size_t count() const override;
 
 protected:
-    bool FKeyExists(const char *key, size_t cchKey) const;
+    bool FKeyExists(std::string&) const;
 
     const rocksdb::ReadOptions &ReadOptions() const { return m_readOptionsTemplate; }
     rocksdb::WriteOptions WriteOptions() const;