Extend StringAppendTESTOperator to use string delimiter of variable length (#2)

java/CMakeLists.txt

@@ -213,6 +213,7 @@ set(JAVA_MAIN_CLASSES
   src/main/java/org/rocksdb/StatsLevel.java
   src/main/java/org/rocksdb/Status.java
   src/main/java/org/rocksdb/StringAppendOperator.java
+  src/main/java/org/rocksdb/StringAppendOperatorWithVariableDelimitor.java
   src/main/java/org/rocksdb/TableFilter.java
   src/main/java/org/rocksdb/TableProperties.java
   src/main/java/org/rocksdb/TableFormatConfig.java
@@ -456,6 +457,7 @@ if(${CMAKE_VERSION} VERSION_LESS "3.11.4" OR (${Java_VERSION_MINOR} STREQUAL "7"
           org.rocksdb.SstFileReaderIterator
           org.rocksdb.Statistics
           org.rocksdb.StringAppendOperator
+          org.rocksdb.StringAppendOperatorWithVariableDelimitor
           org.rocksdb.TableFormatConfig
           org.rocksdb.ThreadStatus
           org.rocksdb.TimedEnv

java/Makefile

@@ -76,6 +76,7 @@ NATIVE_JAVA_CLASSES = \
 	org.rocksdb.VectorMemTableConfig\
 	org.rocksdb.Snapshot\
 	org.rocksdb.StringAppendOperator\
+	org.rocksdb.StringAppendOperatorWithVariableDelimitor\
 	org.rocksdb.UInt64AddOperator\
 	org.rocksdb.WriteBatch\
 	org.rocksdb.WriteBatch.Handler\

java/rocksjni/merge_operator.cc

@@ -7,17 +7,20 @@
 // This file implements the "bridge" between Java and C++
 // for ROCKSDB_NAMESPACE::MergeOperator.
 
+#include "rocksdb/merge_operator.h"
+
 #include <jni.h>
 #include <stdio.h>
 #include <stdlib.h>
+
 #include <memory>
 #include <string>
 
 #include "include/org_rocksdb_StringAppendOperator.h"
+#include "include/org_rocksdb_StringAppendOperatorWithVariableDelimitor.h"
 #include "include/org_rocksdb_UInt64AddOperator.h"
 #include "rocksdb/db.h"
 #include "rocksdb/memtablerep.h"
-#include "rocksdb/merge_operator.h"
 #include "rocksdb/options.h"
 #include "rocksdb/slice_transform.h"
 #include "rocksdb/statistics.h"
@@ -79,3 +82,40 @@ void Java_org_rocksdb_UInt64AddOperator_disposeInternal(JNIEnv* /*env*/,
           jhandle);
   delete sptr_uint64_add_op;  // delete std::shared_ptr
 }
+
+/*
+ * Class:     org_rocksdb_StringAppendOperatorWithVariableDelimitor
+ * Method:    newSharedStringAppendTESTOperator
+ * Signature: ([B)J
+ */
+jlong Java_org_rocksdb_StringAppendOperatorWithVariableDelimitor_newSharedStringAppendTESTOperator(
+    JNIEnv* env, jclass /*jclazz*/, jbyteArray jdelim) {
+  jboolean has_exception = JNI_FALSE;
+  std::string delim = ROCKSDB_NAMESPACE::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_NAMESPACE::MergeOperator>(
+          ROCKSDB_NAMESPACE::MergeOperators::CreateStringAppendTESTOperator(
+              delim));
+  return reinterpret_cast<jlong>(sptr_string_append_test_op);
+}
+
+/*
+ * Class:     org_rocksdb_StringAppendOperatorWithVariableDelimitor
+ * Method:    disposeInternal
+ * Signature: (J)V
+ */
+void Java_org_rocksdb_StringAppendOperatorWithVariableDelimitor_disposeInternal(
+    JNIEnv* /*env*/, jobject /*jobj*/, jlong jhandle) {
+  auto* sptr_string_append_test_op =
+      reinterpret_cast<std::shared_ptr<ROCKSDB_NAMESPACE::MergeOperator>*>(
+          jhandle);
+  delete sptr_string_append_test_op;  // delete std::shared_ptr
+}

java/src/main/java/org/rocksdb/StringAppendOperatorWithVariableDelimitor.java

+/*
+ * 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.
+ */
+
+package org.rocksdb;
+
+import java.nio.charset.Charset;
+
+/**
+ * Merge operator that concatenates two strings with a delimiter of variable length, string or byte
+ * array.
+ */
+public class StringAppendOperatorWithVariableDelimitor extends MergeOperator {
+  public StringAppendOperatorWithVariableDelimitor() {
+    this(',');
+  }
+
+  public StringAppendOperatorWithVariableDelimitor(char delim) {
+    this(Character.toString(delim));
+  }
+
+  public StringAppendOperatorWithVariableDelimitor(byte[] delim) {
+    super(newSharedStringAppendTESTOperator(delim));
+  }
+
+  public StringAppendOperatorWithVariableDelimitor(String delim) {
+    this(delim.getBytes());
+  }
+
+  public StringAppendOperatorWithVariableDelimitor(String delim, Charset charset) {
+    this(delim.getBytes(charset));
+  }
+
+  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

@@ -205,6 +205,34 @@ public class MergeTest {
     }
   }
 
+  @Test
+  public void stringDelimiter() throws RocksDBException {
+    stringDelimiter("DELIM");
+    stringDelimiter("");
+  }
+
+  private void stringDelimiter(String delim) throws RocksDBException {
+    try (final MergeOperator stringAppendOperator =
+             new StringAppendOperatorWithVariableDelimitor(delim.getBytes());
+         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(), "aa".getBytes());
+
+      // Writing bb under key
+      db.merge("key".getBytes(), "bb".getBytes());
+
+      // Writing empty under key
+      db.merge("key".getBytes(), "".getBytes());
+
+      final byte[] value = db.get("key".getBytes());
+      final String strValue = new String(value);
+
+      assertThat(strValue).isEqualTo("aa" + delim + "bb" + delim);
+    }
+  }
+
   @Test
   public void cFOperatorOption()
       throws InterruptedException, RocksDBException {

utilities/merge_operators.h

@@ -21,6 +21,8 @@ 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();
   static std::shared_ptr<MergeOperator> CreateSortOperator();

utilities/merge_operators/string_append/stringappend2.cc

@@ -15,11 +15,6 @@
 
 namespace ROCKSDB_NAMESPACE {
 
-// 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,21 @@ 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 +83,16 @@ 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;
-       it != operand_list.end(); ++it) {
-    new_value->append(1, delim_);
+  for (auto it = operand_list.begin() + 1; it != operand_list.end(); ++it) {
+    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_NAMESPACE

utilities/merge_operators/string_append/stringappend2.h

@@ -14,6 +14,7 @@
 #pragma once
 #include <deque>
 #include <string>
+#include <utility>
 
 #include "rocksdb/merge_operator.h"
 #include "rocksdb/slice.h"
@@ -22,18 +23,22 @@ namespace ROCKSDB_NAMESPACE {
 
 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)){};
 
-  virtual bool FullMergeV2(const MergeOperationInput& merge_in,
-                           MergeOperationOutput* merge_out) const override;
+  // Constructor with char delimiter
+  explicit StringAppendTESTOperator(char delim_char)
+      : delim_(std::string(1, delim_char)){};
 
-  virtual bool PartialMergeMulti(const Slice& key,
-                                 const std::deque<Slice>& operand_list,
-                                 std::string* new_value, Logger* logger) const
-      override;
+  bool FullMergeV2(const MergeOperationInput& merge_in,
+                   MergeOperationOutput* merge_out) const override;
 
-  virtual const char* Name() const override;
+  bool PartialMergeMulti(const Slice& key,
+                         const std::deque<Slice>& operand_list,
+                         std::string* new_value, Logger* logger) const override;
+
+  const char* Name() const override;
 
  private:
   // A version of PartialMerge that actually performs "partial merging".
@@ -42,8 +47,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
 };
 
 }  // namespace ROCKSDB_NAMESPACE