Add a new int64 add merge operator, also with Java API

BUCK

@@ -310,6 +310,7 @@ cpp_library_wrapper(name="rocksdb_lib", srcs=[
         "utilities/memory/memory_util.cc",
         "utilities/merge_operators.cc",
         "utilities/merge_operators/bytesxor.cc",
+        "utilities/merge_operators/int64add/int64_add.cc",
         "utilities/merge_operators/max.cc",
         "utilities/merge_operators/put.cc",
         "utilities/merge_operators/sortlist.cc",
@@ -5292,6 +5293,12 @@ cpp_unittest_wrapper(name="merge_helper_test",
             extra_compiler_flags=[])
 
 
+cpp_unittest_wrapper(name="merge_operators_test",
+            srcs=["utilities/merge_operators/test/merge_operators_test.cc"],
+            deps=[":rocksdb_test_lib"],
+            extra_compiler_flags=[])
+
+
 cpp_unittest_wrapper(name="merge_test",
             srcs=["db/merge_test.cc"],
             deps=[":rocksdb_test_lib"],
@@ -5532,12 +5539,6 @@ cpp_unittest_wrapper(name="stats_history_test",
             extra_compiler_flags=[])
 
 
-cpp_unittest_wrapper(name="stringappend_test",
-            srcs=["utilities/merge_operators/string_append/stringappend_test.cc"],
-            deps=[":rocksdb_test_lib"],
-            extra_compiler_flags=[])
-
-
 cpp_unittest_wrapper(name="table_properties_collector_test",
             srcs=["db/table_properties_collector_test.cc"],
             deps=[":rocksdb_test_lib"],

CMakeLists.txt

@@ -961,6 +961,7 @@ set(SOURCES
         utilities/merge_operators/string_append/stringappend.cc
         utilities/merge_operators/string_append/stringappend2.cc
         utilities/merge_operators/uint64add.cc
+        utilities/merge_operators/int64add/int64_add.cc
         utilities/object_registry.cc
         utilities/option_change_migration/option_change_migration.cc
         utilities/options/options_util.cc
@@ -1526,7 +1527,7 @@ if(WITH_TESTS)
         utilities/checkpoint/checkpoint_test.cc
         utilities/env_timed_test.cc
         utilities/memory/memory_test.cc
-        utilities/merge_operators/string_append/stringappend_test.cc
+        utilities/merge_operators/test/merge_operators_test.cc
         utilities/object_registry_test.cc
         utilities/option_change_migration/option_change_migration_test.cc
         utilities/options/options_util_test.cc

Makefile

@@ -1349,7 +1349,7 @@ option_change_migration_test: $(OBJ_DIR)/utilities/option_change_migration/optio
 agg_merge_test: $(OBJ_DIR)/utilities/agg_merge/agg_merge_test.o $(TEST_LIBRARY) $(LIBRARY)
 	$(AM_LINK)
 
-stringappend_test: $(OBJ_DIR)/utilities/merge_operators/string_append/stringappend_test.o $(TEST_LIBRARY) $(LIBRARY)
+merge_operators_test: $(OBJ_DIR)/utilities/merge_operators/test/merge_operators_test.o $(OBJ_DIR)/utilities/merge_operators/test/stringappend_test.o $(OBJ_DIR)/utilities/merge_operators/test/int64add_test.o $(TEST_LIBRARY) $(LIBRARY)
 	$(AM_LINK)
 
 cassandra_format_test: $(OBJ_DIR)/utilities/cassandra/cassandra_format_test.o $(TEST_LIBRARY) $(LIBRARY)

build_tools/run_ci_db_test.ps1

@@ -41,7 +41,7 @@ $RunOnly = New-Object System.Collections.Generic.HashSet[string]
 $RunOnly.Add("c_test") | Out-Null
 $RunOnly.Add("compact_on_deletion_collector_test") | Out-Null
 $RunOnly.Add("merge_test") | Out-Null
-$RunOnly.Add("stringappend_test") | Out-Null # Apparently incorrectly written
+$RunOnly.Add("merge_operators_test") | Out-Null # Apparently incorrectly written
 $RunOnly.Add("backup_engine_test") | Out-Null # Disabled
 $RunOnly.Add("timer_queue_test") | Out-Null # Not a gtest
 

java/CMakeLists.txt

@@ -185,6 +185,7 @@ set(JAVA_MAIN_CLASSES
   src/main/java/org/rocksdb/IndexType.java
   src/main/java/org/rocksdb/InfoLogLevel.java
   src/main/java/org/rocksdb/IngestExternalFileOptions.java
+  src/main/java/org/rocksdb/Int64AddOperator.java
   src/main/java/org/rocksdb/KeyMayExist.java
   src/main/java/org/rocksdb/LRUCache.java
   src/main/java/org/rocksdb/LevelMetaData.java
@@ -298,6 +299,7 @@ set(JAVA_MAIN_CLASSES
   src/main/java/org/rocksdb/util/BytewiseComparator.java
   src/main/java/org/rocksdb/util/Environment.java
   src/main/java/org/rocksdb/util/IntComparator.java
+  src/main/java/org/rocksdb/util/MergeEncodings.java
   src/main/java/org/rocksdb/util/ReverseBytewiseComparator.java
   src/main/java/org/rocksdb/util/SizeUnit.java
   src/main/java/org/rocksdb/util/StdErrLogger.java
@@ -426,6 +428,7 @@ set(JAVA_TEST_CLASSES
   src/test/java/org/rocksdb/util/ReverseBytewiseComparatorIntTest.java
   src/test/java/org/rocksdb/util/SizeUnitTest.java
   src/test/java/org/rocksdb/util/StdErrLoggerTest.java
+  src/test/java/org/rocksdb/util/MergeEncodingsTest.java
   src/test/java/org/rocksdb/util/TestUtil.java
 )
 
@@ -716,6 +719,7 @@ if(${CMAKE_VERSION} VERSION_LESS "3.11.4")
           org.rocksdb.HashSkipListMemTableConfig
           org.rocksdb.HyperClockCache
           org.rocksdb.IngestExternalFileOptions
+          org.rocksdb.Int64AddOperator
           org.rocksdb.Logger
           org.rocksdb.LRUCache
           org.rocksdb.MemoryUtil

java/Makefile

@@ -35,6 +35,7 @@ NATIVE_JAVA_CLASSES = \
 	org.rocksdb.FlushOptions\
 	org.rocksdb.Filter\
 	org.rocksdb.IngestExternalFileOptions\
+	org.rocksdb.Int64AddOperator\
 	org.rocksdb.HashLinkedListMemTableConfig\
 	org.rocksdb.HashSkipListMemTableConfig\
 	org.rocksdb.ConcurrentTaskLimiter\
@@ -140,6 +141,7 @@ JAVA_TESTS = \
 	org.rocksdb.IngestExternalFileOptionsTest\
 	org.rocksdb.util.IntComparatorTest\
 	org.rocksdb.util.JNIComparatorTest\
+	org.rocksdb.util.MergeEncodingsTest\
 	org.rocksdb.FilterTest\
 	org.rocksdb.FlushTest\
 	org.rocksdb.ImportColumnFamilyTest\

java/rocksjni/merge_operator.cc

@@ -16,6 +16,7 @@
 #include <memory>
 #include <string>
 
+#include "include/org_rocksdb_Int64AddOperator.h"
 #include "include/org_rocksdb_StringAppendOperator.h"
 #include "include/org_rocksdb_UInt64AddOperator.h"
 #include "rocksdb/db.h"
@@ -96,3 +97,30 @@ void Java_org_rocksdb_UInt64AddOperator_disposeInternalJni(JNIEnv* /*env*/,
           jhandle);
   delete sptr_uint64_add_op;  // delete std::shared_ptr
 }
+
+/*
+ * Class:     org_rocksdb_Int64AddOperator
+ * Method:    newSharedInt64AddOperator
+ * Signature: ()J
+ */
+jlong Java_org_rocksdb_Int64AddOperator_newSharedInt64AddOperator(
+    JNIEnv* /*env*/, jclass /*jclazz*/) {
+  auto* sptr_int64_add_op =
+      new std::shared_ptr<ROCKSDB_NAMESPACE::MergeOperator>(
+          ROCKSDB_NAMESPACE::MergeOperators::CreateInt64AddOperator());
+  return GET_CPLUSPLUS_POINTER(sptr_int64_add_op);
+}
+
+/*
+ * Class:     org_rocksdb_Int64AddOperator
+ * Method:    disposeInternal
+ * Signature: (J)V
+ */
+void Java_org_rocksdb_Int64AddOperator_disposeInternal(JNIEnv* /*env*/,
+                                                       jobject /*jobj*/,
+                                                       jlong jhandle) {
+  auto* sptr_int64_add_op =
+      reinterpret_cast<std::shared_ptr<ROCKSDB_NAMESPACE::MergeOperator>*>(
+          jhandle);
+  delete sptr_int64_add_op;  // delete this instance of the std::shared_ptr
+}

java/src/main/java/org/rocksdb/Int64AddOperator.java

@@ -0,0 +1,16 @@
+// Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+// This source code is licensed under both the GPLv2 (found in the
+// COPYING file in the root directory) and Apache 2.0 License
+// (found in the LICENSE.Apache file in the root directory).
+
+package org.rocksdb;
+
+public class Int64AddOperator extends MergeOperator {
+  protected Int64AddOperator() {
+    super(newSharedInt64AddOperator());
+  }
+
+  @Override protected native void disposeInternal(final long handle);
+
+  private static native long newSharedInt64AddOperator();
+}

java/src/main/java/org/rocksdb/util/MergeEncodings.java

@@ -0,0 +1,150 @@
+// Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under both the GPLv2 (found in the
+//  COPYING file in the root directory) and Apache 2.0 License
+//  (found in the LICENSE.Apache file in the root directory).
+
+package org.rocksdb.util;
+
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+
+/**
+ * Java implementations of merge encodings for merge operators including {@link
+ * org.rocksdb.Int64AddOperator}
+ */
+public class MergeEncodings {
+  /**
+   * Zigzag decode an unsigned long value to a signed long value
+   * Use the parameter to index into the list [0,-1,1,-2,2,-3,3,...]
+   * unsigned integers are in 1:1 mapping to the signed integers
+   *
+   * @param zigzag to decode
+   * @return the value at {@code zigzag}'s index in the list [0,-1,1,-2,2,-3,3,...]
+   */
+  public static long fromZigzag(final long zigzag) {
+    final long half = zigzag / 2;
+    return (half + half == zigzag) ? half : -half - 1;
+  }
+
+  /**
+   * Zigzag encode a signed long value to an unsigned long value
+   * Each signed long integer is represented by its position in the list [0,-1,1,-2,2,-3,3,...]
+   * unsigned integers are in 1:1 mapping to the signed integers
+   *
+   * @param value to encode
+   * @return {@code value}'s position in the list [0,-1,1,-2,2,-3,3,...]
+   */
+  public static long toZigzag(final long value) {
+    final long MAX_ZIGZAG = Long.MAX_VALUE / 2;
+    final long MIN_ZIGZAG = Long.MIN_VALUE / 2;
+    if (value < MIN_ZIGZAG || value > MAX_ZIGZAG) {
+      throw new IllegalArgumentException(
+          "Zigzag can only be applied to Long.MIN_VALUE/2..Long.MAX_VALUE/2, parameter is "
+          + value);
+    }
+    final long twice = value + value;
+    return (value >= 0) ? twice : -twice - 1;
+  }
+
+  /**
+   * Create a variable length encoding of a signed integer
+   * The encoding is little-endian (least significant byte first)
+   * and each byte encodes a full 8 bits of the supplied {@code value}
+   *
+   * This is possible because the length of the {@code byte[]} is known/stored elsewhere,
+   * such as in the value length of a {@code (key,value)}-pair, so we always know how many
+   * bytes to decode.
+   *
+   * The MSB is 2s complement, and is sign-extended on decoding.
+   *
+   * @param value to encode as a sequence of bytes
+   * @return a byte array encoding the value, and which is just exactly as long as necessary
+   */
+  @SuppressWarnings("PMD.AvoidReassigningParameters")
+  public static byte[] encodeVarintSigned(long value) {
+    final byte[] bytes = new byte[Long.BYTES];
+    int i = 0;
+    if (value < 0) {
+      while (value < -0x80) {
+        bytes[i++] = (byte) (value & 0xff);
+        value >>= 8;
+      }
+    } else {
+      while (value > 0x7f) {
+        bytes[i++] = (byte) (value & 0xff);
+        value >>= 8;
+      }
+    }
+    bytes[i++] = (byte) (value & 0xff);
+    return Arrays.copyOfRange(bytes, 0, i);
+  }
+
+  /**
+   * Decode a variable length encoding of a signed integer
+   * The encoding is little-endian (least significant byte first)
+   * and each byte encodes a full 8 bits of the supplied {@code value},
+   * except for the MSB, which is 2s complement
+   *
+   * This is possible because the length of {@code byte[] bytes} is known/stored elsewhere,
+   * such as in the value length of a {@code (key,value)}-pair, so we always know how many
+   * bytes to decode.
+   *
+   * @param bytes which encode the value as a sequence of bytes
+   * @return the value decoded from {@code bytes}
+   */
+  public static long decodeVarintSigned(final byte[] bytes) {
+    long acc = 0;
+    if (bytes.length > 0) {
+      int pos = bytes.length;
+      acc = bytes[--pos];
+      while (pos > 0) {
+        acc = (acc << 8) | (bytes[--pos] & 0xff);
+      }
+    }
+
+    return acc;
+  }
+
+  /**
+   * Encode a signed value
+   * @param value to encode
+   * @return the signed value's encoding as a variable-length run of bytes
+   */
+  public static byte[] encodeSigned(final long value) {
+    return encodeVarintSigned(value);
+  }
+
+  /**
+   * Decode a run of bytes into a signed values
+   * @param bytes to decode
+   * @return the signed value represented by the encoding
+   */
+  public static long decodeSigned(final byte[] bytes) {
+    return decodeVarintSigned(bytes);
+  }
+
+  /**
+   * Fixed-length encode a long into a byte array
+   * @param l the long value to encode
+   * @return the byte array into which the value is encoded
+   */
+  public static byte[] longToByteArray(final long l) {
+    final ByteBuffer buf = ByteBuffer.allocate(Long.BYTES).order(ByteOrder.LITTLE_ENDIAN);
+    buf.putLong(l);
+    return buf.array();
+  }
+
+  /**
+   * Decode a fixed-length byte array into a long value
+   * @param a the byte-array to decode
+   * @return the long value encoded in the input array
+   */
+  public static long longFromByteArray(final byte[] a) {
+    final ByteBuffer buf = ByteBuffer.allocate(Long.BYTES).order(ByteOrder.LITTLE_ENDIAN);
+    buf.put(a);
+    buf.position(Math.max(buf.position(), Long.BYTES)); // guard against BufferOverflowException
+    buf.flip();
+    return buf.getLong();
+  }
+}