add merge operator

Author: ltagliamonte-dd

src/storage/hash_merge_operator.cc

@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "storage/hash_merge_operator.h"
+
+#include "storage/redis_metadata.h"
+
+namespace engine {
+
+bool HashMergeOperator::Merge(const rocksdb::Slice& key, const rocksdb::Slice* existing_value,
+                              const rocksdb::Slice& value, std::string* new_value, rocksdb::Logger* logger) const {
+  if (new_value == nullptr) {
+    return false;
+  }
+
+  new_value->assign(value.data(), value.size());
+
+  return true;
+}
+
+const char* HashMergeOperator::Name() const { return "HashMergeOperator"; }
+
+}  // namespace engine

src/storage/hash_merge_operator.h

@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#pragma once
+
+#include <rocksdb/merge_operator.h>
+
+namespace engine {
+
+class HashMergeOperator : public rocksdb::AssociativeMergeOperator {
+ public:
+  bool Merge(const rocksdb::Slice& key, const rocksdb::Slice* existing_value, const rocksdb::Slice& value,
+             std::string* new_value, rocksdb::Logger* logger) const override;
+
+  const char* Name() const override;
+};
+
+}  // namespace engine

src/storage/storage.cc

@@ -48,6 +48,7 @@
 #include "rocksdb_crc32c.h"
 #include "server/server.h"
 #include "storage/batch_indexer.h"
+#include "storage/hash_merge_operator.h"
 #include "string_util.h"
 #include "table_properties_collector.h"
 #include "time_util.h"
@@ -338,6 +339,7 @@ Status Storage::Open(DBOpenMode mode) {
   subkey_opts.disable_auto_compactions = config_->rocks_db.disable_auto_compactions;
   subkey_opts.table_properties_collector_factories.emplace_back(
       NewCompactOnExpiredTableCollectorFactory(std::string(kPrimarySubkeyColumnFamilyName), 0.3));
+  subkey_opts.merge_operator = std::make_shared<HashMergeOperator>();
   SetBlobDB(&subkey_opts);
 
   rocksdb::BlockBasedTableOptions pubsub_table_opts = InitTableOptions();