Extend StringAppendTESTOperator to use string delimiter of variable length

java/rocksjni/merge_operator.cc

@@ -36,6 +36,28 @@ jlong Java_org_rocksdb_StringAppendOperator_newSharedStringAppendOperator(
   return reinterpret_cast<jlong>(sptr_string_append_op);
 }
 
+/*
+ * Class:     org_rocksdb_StringAppendOperator
+ * Method:    newSharedStringAppendTESTOperator
+ * Signature: ([B)J
+ */
+jlong Java_org_rocksdb_StringAppendOperator_newSharedStringAppendTESTOperator(
+        JNIEnv* env, jclass, jbyteArray jdelim) {
+  jboolean has_exception = JNI_FALSE;
+  std::string delim = rocksdb::JniUtil::byteString<std::string>(
+          env, jdelim,
+          [](const char* str, const size_t len) { return std::string(str, len); },
+          &has_exception);
+  if (has_exception == JNI_TRUE) {
+    // exception occurred
+    return 0;
+  }
+
+  auto* sptr_string_append_test_op = new std::shared_ptr<rocksdb::MergeOperator>(
+          rocksdb::MergeOperators::CreateStringAppendTESTOperator(delim));
+  return reinterpret_cast<jlong>(sptr_string_append_test_op);
+}
+
 /*
  * Class:     org_rocksdb_StringAppendOperator
  * Method:    disposeInternal

java/src/main/java/org/rocksdb/StringAppendOperator.java

@@ -14,10 +14,15 @@ public StringAppendOperator() {
         this(',');
     }
 
-    public StringAppendOperator(char delim) {
+    public StringAppendOperator(final char delim) {
         super(newSharedStringAppendOperator(delim));
     }
 
+    public StringAppendOperator(final byte[] delim) {
+        super(newSharedStringAppendTESTOperator(delim));
+    }
+
     private native static long newSharedStringAppendOperator(final char delim);
+    private native static long newSharedStringAppendTESTOperator(final byte[] delim);
     @Override protected final native void disposeInternal(final long handle);
 }

java/src/test/java/org/rocksdb/MergeTest.java

@@ -15,6 +15,7 @@
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
 
+import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.assertj.core.api.Assertions.assertThat;
 
 public class MergeTest {
@@ -183,6 +184,35 @@ public void operatorOption()
     }
   }
 
+  @Test
+  public void emptyStringDelimiter() throws RocksDBException {
+    stringDelimiter("");
+  }
+
+  @Test
+  public void stringDelimiter() throws RocksDBException {
+    stringDelimiter("DELIM");
+  }
+
+  private void stringDelimiter(String delim) throws RocksDBException {
+    try (final StringAppendOperator stringAppendOperator = new StringAppendOperator(delim.getBytes(UTF_8));
+         final Options opt = new Options()
+                 .setCreateIfMissing(true)
+                 .setMergeOperator(stringAppendOperator);
+         final RocksDB db = RocksDB.open(opt, dbFolder.getRoot().getAbsolutePath())) {
+      // Writing aa under key
+      db.put("key".getBytes(UTF_8), "aa".getBytes(UTF_8));
+
+      // Writing bb under key
+      db.merge("key".getBytes(UTF_8), "bb".getBytes(UTF_8));
+
+      final byte[] value = db.get("key".getBytes(UTF_8));
+      final String strValue = new String(value, UTF_8);
+
+      assertThat(strValue).isEqualTo("aa" + delim + "bb");
+    }
+  }
+
   @Test
   public void uint64AddOperatorOption()
       throws InterruptedException, RocksDBException {

utilities/merge_operators.h

@@ -21,6 +21,7 @@ class MergeOperators {
   static std::shared_ptr<MergeOperator> CreateStringAppendOperator();
   static std::shared_ptr<MergeOperator> CreateStringAppendOperator(char delim_char);
   static std::shared_ptr<MergeOperator> CreateStringAppendTESTOperator();
+  static std::shared_ptr<MergeOperator> CreateStringAppendTESTOperator(std::string delim);
   static std::shared_ptr<MergeOperator> CreateMaxOperator();
   static std::shared_ptr<MergeOperator> CreateBytesXOROperator();
 

utilities/merge_operators/string_append/stringappend2.cc

@@ -15,11 +15,6 @@
 
 namespace rocksdb {
 
-// Constructor: also specify the delimiter character.
-StringAppendTESTOperator::StringAppendTESTOperator(char delim_char)
-    : delim_(delim_char) {
-}
-
 // Implementation for the merge operation (concatenates two strings)
 bool StringAppendTESTOperator::FullMergeV2(
     const MergeOperationInput& merge_in,
@@ -37,7 +32,7 @@ bool StringAppendTESTOperator::FullMergeV2(
   size_t numBytes = 0;
   for (auto it = merge_in.operand_list.begin();
        it != merge_in.operand_list.end(); ++it) {
-    numBytes += it->size() + 1;   // Plus 1 for the delimiter
+    numBytes += it->size() + delim_.size();   // Plus one delimiter
   }
 
   // Only print the delimiter after the first entry has been printed
@@ -48,20 +43,20 @@ bool StringAppendTESTOperator::FullMergeV2(
     merge_out->new_value.reserve(numBytes + merge_in.existing_value->size());
     merge_out->new_value.append(merge_in.existing_value->data(),
                                 merge_in.existing_value->size());
-    printDelim = true;
+    printDelim = !delim_.empty();
   } else if (numBytes) {
     merge_out->new_value.reserve(
-        numBytes - 1);  // Minus 1 since we have one less delimiter
+        numBytes - delim_.size());  // Minus 1 delimiter since we have one less delimiter
   }
 
   // Concatenate the sequence of strings (and add a delimiter between each)
   for (auto it = merge_in.operand_list.begin();
        it != merge_in.operand_list.end(); ++it) {
     if (printDelim) {
-      merge_out->new_value.append(1, delim_);
+      merge_out->new_value.append(delim_);
     }
     merge_out->new_value.append(it->data(), it->size());
-    printDelim = true;
+    printDelim = !delim_.empty();
   }
 
   return true;
@@ -87,17 +82,17 @@ bool StringAppendTESTOperator::_AssocPartialMergeMulti(
   // Determine and reserve correct size for *new_value.
   size_t size = 0;
   for (const auto& operand : operand_list) {
-    size += operand.size();
+    size += operand.size() + delim_.size();
   }
-  size += operand_list.size() - 1;  // Delimiters
+  size -= delim_.size();  // since we have one less delimiter
   new_value->reserve(size);
 
   // Apply concatenation
   new_value->assign(operand_list.front().data(), operand_list.front().size());
 
-  for (std::deque<Slice>::const_iterator it = operand_list.begin() + 1;
+  for (auto it = operand_list.begin() + 1;
        it != operand_list.end(); ++it) {
-    new_value->append(1, delim_);
+    new_value->append(delim_);
     new_value->append(it->data(), it->size());
   }
 
@@ -114,4 +109,9 @@ MergeOperators::CreateStringAppendTESTOperator() {
   return std::make_shared<StringAppendTESTOperator>(',');
 }
 
+std::shared_ptr<MergeOperator>
+MergeOperators::CreateStringAppendTESTOperator(std::string delim) {
+  return std::make_shared<StringAppendTESTOperator>(delim);
+}
+
 } // namespace rocksdb

utilities/merge_operators/string_append/stringappend2.h

@@ -13,26 +13,28 @@
 #pragma once
 #include <deque>
 #include <string>
-
+#include <utility>
 #include "rocksdb/merge_operator.h"
 #include "rocksdb/slice.h"
 
 namespace rocksdb {
 
 class StringAppendTESTOperator : public MergeOperator {
  public:
-  // Constructor with delimiter
-  explicit StringAppendTESTOperator(char delim_char);
+  // Constructor with string delimiter
+  explicit StringAppendTESTOperator(std::string delim_str) : delim_(std::move(delim_str)) {};
+
+  // Constructor with char delimiter
+  explicit StringAppendTESTOperator(char delim_char) : delim_(std::string(1, delim_char)) {};
 
-  virtual bool FullMergeV2(const MergeOperationInput& merge_in,
-                           MergeOperationOutput* merge_out) const override;
+  bool FullMergeV2(const MergeOperationInput& merge_in,
+                   MergeOperationOutput* merge_out) const override;
 
-  virtual bool PartialMergeMulti(const Slice& key,
-                                 const std::deque<Slice>& operand_list,
-                                 std::string* new_value, Logger* logger) const
-      override;
+  bool PartialMergeMulti(const Slice& key,
+                         const std::deque<Slice>& operand_list,
+                         std::string* new_value, Logger* logger) const override;
 
-  virtual const char* Name() const override;
+  const char* Name() const override;
 
  private:
   // A version of PartialMerge that actually performs "partial merging".
@@ -41,7 +43,7 @@ class StringAppendTESTOperator : public MergeOperator {
                                const std::deque<Slice>& operand_list,
                                std::string* new_value, Logger* logger) const;
 
-  char delim_;         // The delimiter is inserted between elements
+  std::string delim_;         // The delimiter is inserted between elements
 
 };