Collection Merge Operator

CMakeLists.txt

@@ -793,6 +793,7 @@ set(SOURCES
         utilities/leveldb_options/leveldb_options.cc
         utilities/memory/memory_util.cc
         utilities/merge_operators/bytesxor.cc
+        utilities/merge_operators/collection/collection_merge_operator.cc
         utilities/merge_operators/max.cc
         utilities/merge_operators/put.cc
         utilities/merge_operators/sortlist.cc
@@ -1213,6 +1214,7 @@ if(WITH_TESTS)
         utilities/cassandra/cassandra_serialize_test.cc
         utilities/checkpoint/checkpoint_test.cc
         utilities/memory/memory_test.cc
+        utilities/merge_operators/collection/collection_merge_operator_test.cc
         utilities/merge_operators/string_append/stringappend_test.cc
         utilities/object_registry_test.cc
         utilities/option_change_migration/option_change_migration_test.cc

Makefile

@@ -598,6 +598,7 @@ ifdef ASSERT_STATUS_CHECKED
 		cleanable_test \
 		checkpoint_test \
 		coding_test \
+		collection_merge_operator_test \
 		crc32c_test \
 		dbformat_test \
 		db_basic_test \
@@ -1487,6 +1488,9 @@ cache_test: $(OBJ_DIR)/cache/cache_test.o $(TEST_LIBRARY) $(LIBRARY)
 coding_test: $(OBJ_DIR)/util/coding_test.o $(TEST_LIBRARY) $(LIBRARY)
 	$(AM_LINK)
 
+collection_merge_operator_test:  $(OBJ_DIR)/utilities/merge_operators/collection/collection_merge_operator_test.o $(TEST_LIBRARY) $(LIBRARY)
+	$(AM_LINK)
+
 hash_test: $(OBJ_DIR)/util/hash_test.o $(TEST_LIBRARY) $(LIBRARY)
 	$(AM_LINK)
 

java/CMakeLists.txt

@@ -13,6 +13,7 @@ set(JNI_NATIVE_SOURCES
         rocksjni/cassandra_value_operator.cc
         rocksjni/checkpoint.cc
         rocksjni/clock_cache.cc
+        rocksjni/collection_merge_operator.cc
         rocksjni/columnfamilyhandle.cc
         rocksjni/compaction_filter.cc
         rocksjni/compaction_filter_factory.cc
@@ -116,6 +117,7 @@ set(JAVA_MAIN_CLASSES
   src/main/java/org/rocksdb/Checkpoint.java
   src/main/java/org/rocksdb/ChecksumType.java
   src/main/java/org/rocksdb/ClockCache.java
+  src/main/java/org/rocksdb/CollectionMergeOperator.java
   src/main/java/org/rocksdb/ColumnFamilyDescriptor.java
   src/main/java/org/rocksdb/ColumnFamilyHandle.java
   src/main/java/org/rocksdb/ColumnFamilyMetaData.java
@@ -424,6 +426,7 @@ if(${CMAKE_VERSION} VERSION_LESS "3.11.4" OR (${Java_VERSION_MINOR} STREQUAL "7"
           org.rocksdb.CassandraValueMergeOperator
           org.rocksdb.Checkpoint
           org.rocksdb.ClockCache
+          org.rocksdb.CollectionMergeOperator
           org.rocksdb.ColumnFamilyHandle
           org.rocksdb.ColumnFamilyOptions
           org.rocksdb.CompactionJobInfo

java/Makefile

@@ -16,6 +16,7 @@ NATIVE_JAVA_CLASSES = \
 	org.rocksdb.ClockCache\
 	org.rocksdb.CassandraCompactionFilter\
 	org.rocksdb.CassandraValueMergeOperator\
+	org.rocksdb.CollectionMergeOperator\
 	org.rocksdb.ColumnFamilyHandle\
 	org.rocksdb.ColumnFamilyOptions\
 	org.rocksdb.CompactionJobInfo\
@@ -115,6 +116,7 @@ JAVA_TESTS = \
 	org.rocksdb.util.BytewiseComparatorIntTest\
 	org.rocksdb.CheckPointTest\
 	org.rocksdb.ClockCacheTest\
+	org.rocksdb.CollectionMergeOperatorTest\
 	org.rocksdb.ColumnFamilyOptionsTest\
 	org.rocksdb.ColumnFamilyTest\
 	org.rocksdb.CompactionFilterFactoryTest\

java/rocksjni/collection_merge_operator.cc

@@ -0,0 +1,68 @@
+//  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).
+
+#include <jni.h>
+#include <memory>
+
+#include "include/org_rocksdb_CollectionMergeOperator.h"
+
+#include "rocksjni/portal.h"
+#include "utilities/merge_operators/collection/collection_merge_operator.h"
+
+/*
+ * Class:     org_rocksdb_CollectionMergeOperator
+ * Method:    newCollectionMergeOperator
+ * Signature: (SJBB)J
+ */
+jlong Java_org_rocksdb_CollectionMergeOperator_newCollectionMergeOperator__SJBB(
+    JNIEnv*, jclass, jshort jfixed_record_len,
+    jlong jcomparator_handle, jbyte jcomparator_type,
+    jbyte junique_constraint) {
+  ROCKSDB_NAMESPACE::Comparator* comparator = nullptr;
+  if (jcomparator_handle != 0) {
+    comparator =
+        ROCKSDB_NAMESPACE::AbstractComparatorJni::castCppComparator(jcomparator_handle, jcomparator_type);
+  }
+
+  const ROCKSDB_NAMESPACE::UniqueConstraint unique_constraint =
+      ROCKSDB_NAMESPACE::UniqueConstraintJni::toCppUniqueConstraint(junique_constraint);
+  auto* sptr_collection_merge_operator =
+      new std::shared_ptr<ROCKSDB_NAMESPACE::MergeOperator>(
+        std::make_shared<ROCKSDB_NAMESPACE::CollectionMergeOperator>(
+          jfixed_record_len, comparator, unique_constraint));
+  return reinterpret_cast<jlong>(sptr_collection_merge_operator);
+}
+
+/*
+ * Class:     org_rocksdb_CollectionMergeOperator
+ * Method:    newCollectionMergeOperator
+ * Signature: (SBB)J
+ */
+jlong Java_org_rocksdb_CollectionMergeOperator_newCollectionMergeOperator__SBB(
+    JNIEnv*, jclass, jshort jfixed_record_len, jbyte jbuiltin_comparator,
+    jbyte junique_constraint) {
+  auto* comparator =
+    ROCKSDB_NAMESPACE::BuiltinComparatorJni::toCppBuiltinComparator(jbuiltin_comparator);
+
+  const ROCKSDB_NAMESPACE::UniqueConstraint unique_constraint =
+      ROCKSDB_NAMESPACE::UniqueConstraintJni::toCppUniqueConstraint(junique_constraint);
+  auto* sptr_collection_merge_operator =
+      new std::shared_ptr<ROCKSDB_NAMESPACE::MergeOperator>(
+        std::make_shared<ROCKSDB_NAMESPACE::CollectionMergeOperator>(
+          jfixed_record_len, comparator, unique_constraint));
+  return reinterpret_cast<jlong>(sptr_collection_merge_operator);
+}
+
+/*
+ * Class:     org_rocksdb_CollectionMergeOperator
+ * Method:    disposeInternal
+ * Signature: (J)V
+ */
+void Java_org_rocksdb_CollectionMergeOperator_disposeInternal(
+    JNIEnv*, jobject, jlong jhandle) {
+  auto* sptr_collection_merge_operator =
+      reinterpret_cast<std::shared_ptr<ROCKSDB_NAMESPACE::MergeOperator>*>(jhandle);
+  delete sptr_collection_merge_operator;
+}

java/rocksjni/merge_operator.cc

@@ -8,34 +8,22 @@
 // for ROCKSDB_NAMESPACE::MergeOperator.
 
 #include <jni.h>
-#include <stdio.h>
-#include <stdlib.h>
 #include <memory>
-#include <string>
 
 #include "include/org_rocksdb_StringAppendOperator.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"
-#include "rocksdb/table.h"
-#include "rocksjni/portal.h"
 #include "utilities/merge_operators.h"
 
 /*
  * Class:     org_rocksdb_StringAppendOperator
  * Method:    newSharedStringAppendOperator
  * Signature: (C)J
  */
-jlong Java_org_rocksdb_StringAppendOperator_newSharedStringAppendOperator(
-    JNIEnv* /*env*/, jclass /*jclazz*/, jchar jdelim) {
-  auto* sptr_string_append_op =
-      new std::shared_ptr<ROCKSDB_NAMESPACE::MergeOperator>(
-          ROCKSDB_NAMESPACE::MergeOperators::CreateStringAppendOperator(
-              (char)jdelim));
+jlong Java_org_rocksdb_StringAppendOperator_newSharedStringAppendOperator
+(JNIEnv*, jclass, jchar jdelim) {
+  auto* sptr_string_append_op = new std::shared_ptr<ROCKSDB_NAMESPACE::MergeOperator>(
+    ROCKSDB_NAMESPACE::MergeOperators::CreateStringAppendOperator((char)jdelim));
   return reinterpret_cast<jlong>(sptr_string_append_op);
 }
 
@@ -44,9 +32,8 @@ jlong Java_org_rocksdb_StringAppendOperator_newSharedStringAppendOperator(
  * Method:    disposeInternal
  * Signature: (J)V
  */
-void Java_org_rocksdb_StringAppendOperator_disposeInternal(JNIEnv* /*env*/,
-                                                           jobject /*jobj*/,
-                                                           jlong jhandle) {
+void Java_org_rocksdb_StringAppendOperator_disposeInternal(
+    JNIEnv*, jobject, jlong jhandle) {
   auto* sptr_string_append_op =
       reinterpret_cast<std::shared_ptr<ROCKSDB_NAMESPACE::MergeOperator>*>(
           jhandle);
@@ -59,10 +46,9 @@ void Java_org_rocksdb_StringAppendOperator_disposeInternal(JNIEnv* /*env*/,
  * Signature: ()J
  */
 jlong Java_org_rocksdb_UInt64AddOperator_newSharedUInt64AddOperator(
-    JNIEnv* /*env*/, jclass /*jclazz*/) {
-  auto* sptr_uint64_add_op =
-      new std::shared_ptr<ROCKSDB_NAMESPACE::MergeOperator>(
-          ROCKSDB_NAMESPACE::MergeOperators::CreateUInt64AddOperator());
+    JNIEnv*, jclass) {
+  auto* sptr_uint64_add_op = new std::shared_ptr<ROCKSDB_NAMESPACE::MergeOperator>(
+      ROCKSDB_NAMESPACE::MergeOperators::CreateUInt64AddOperator());
   return reinterpret_cast<jlong>(sptr_uint64_add_op);
 }
 
@@ -71,9 +57,8 @@ jlong Java_org_rocksdb_UInt64AddOperator_newSharedUInt64AddOperator(
  * Method:    disposeInternal
  * Signature: (J)V
  */
-void Java_org_rocksdb_UInt64AddOperator_disposeInternal(JNIEnv* /*env*/,
-                                                        jobject /*jobj*/,
-                                                        jlong jhandle) {
+void Java_org_rocksdb_UInt64AddOperator_disposeInternal(
+    JNIEnv*, jobject, jlong jhandle) {
   auto* sptr_uint64_add_op =
       reinterpret_cast<std::shared_ptr<ROCKSDB_NAMESPACE::MergeOperator>*>(
           jhandle);

java/rocksjni/options.cc

@@ -146,20 +146,14 @@ jboolean Java_org_rocksdb_Options_createMissingColumnFamilies(
 /*
  * Class:     org_rocksdb_Options
  * Method:    setComparatorHandle
- * Signature: (JI)V
+ * Signature: (JB)V
  */
-void Java_org_rocksdb_Options_setComparatorHandle__JI(
-    JNIEnv*, jobject, jlong jhandle, jint builtinComparator) {
-  switch (builtinComparator) {
-    case 1:
-      reinterpret_cast<ROCKSDB_NAMESPACE::Options*>(jhandle)->comparator =
-          ROCKSDB_NAMESPACE::ReverseBytewiseComparator();
-      break;
-    default:
-      reinterpret_cast<ROCKSDB_NAMESPACE::Options*>(jhandle)->comparator =
-          ROCKSDB_NAMESPACE::BytewiseComparator();
-      break;
-  }
+void Java_org_rocksdb_Options_setComparatorHandle__JB(
+    JNIEnv*, jobject, jlong jhandle, jbyte jbuiltin_comparator) {
+  auto* comparator =
+      ROCKSDB_NAMESPACE::BuiltinComparatorJni::toCppBuiltinComparator(jbuiltin_comparator);
+  reinterpret_cast<ROCKSDB_NAMESPACE::Options*>(jhandle)->comparator =
+      comparator;
 }
 
 /*
@@ -170,20 +164,8 @@ void Java_org_rocksdb_Options_setComparatorHandle__JI(
 void Java_org_rocksdb_Options_setComparatorHandle__JJB(
     JNIEnv*, jobject, jlong jopt_handle, jlong jcomparator_handle,
     jbyte jcomparator_type) {
-  ROCKSDB_NAMESPACE::Comparator* comparator = nullptr;
-  switch (jcomparator_type) {
-    // JAVA_COMPARATOR
-    case 0x0:
-      comparator = reinterpret_cast<ROCKSDB_NAMESPACE::ComparatorJniCallback*>(
-          jcomparator_handle);
-      break;
-
-    // JAVA_NATIVE_COMPARATOR_WRAPPER
-    case 0x1:
-      comparator =
-          reinterpret_cast<ROCKSDB_NAMESPACE::Comparator*>(jcomparator_handle);
-      break;
-  }
+  ROCKSDB_NAMESPACE::Comparator* comparator =
+      ROCKSDB_NAMESPACE::AbstractComparatorJni::castCppComparator(jcomparator_handle, jcomparator_type);
   auto* opt = reinterpret_cast<ROCKSDB_NAMESPACE::Options*>(jopt_handle);
   opt->comparator = comparator;
 }
@@ -3864,20 +3846,14 @@ void Java_org_rocksdb_ColumnFamilyOptions_optimizeUniversalStyleCompaction(
 /*
  * Class:     org_rocksdb_ColumnFamilyOptions
  * Method:    setComparatorHandle
- * Signature: (JI)V
+ * Signature: (JB)V
  */
-void Java_org_rocksdb_ColumnFamilyOptions_setComparatorHandle__JI(
-    JNIEnv*, jobject, jlong jhandle, jint builtinComparator) {
-  switch (builtinComparator) {
-    case 1:
-      reinterpret_cast<ROCKSDB_NAMESPACE::ColumnFamilyOptions*>(jhandle)
-          ->comparator = ROCKSDB_NAMESPACE::ReverseBytewiseComparator();
-      break;
-    default:
-      reinterpret_cast<ROCKSDB_NAMESPACE::ColumnFamilyOptions*>(jhandle)
-          ->comparator = ROCKSDB_NAMESPACE::BytewiseComparator();
-      break;
-  }
+void Java_org_rocksdb_ColumnFamilyOptions_setComparatorHandle__JB(
+    JNIEnv*, jobject, jlong jhandle, jbyte jbuiltin_comparator) {
+  auto* comparator =
+        ROCKSDB_NAMESPACE::BuiltinComparatorJni::toCppBuiltinComparator(jbuiltin_comparator);
+  reinterpret_cast<ROCKSDB_NAMESPACE::ColumnFamilyOptions*>(jhandle)->comparator =
+      comparator;
 }
 
 /*
@@ -3888,22 +3864,9 @@ void Java_org_rocksdb_ColumnFamilyOptions_setComparatorHandle__JI(
 void Java_org_rocksdb_ColumnFamilyOptions_setComparatorHandle__JJB(
     JNIEnv*, jobject, jlong jopt_handle, jlong jcomparator_handle,
     jbyte jcomparator_type) {
-  ROCKSDB_NAMESPACE::Comparator* comparator = nullptr;
-  switch (jcomparator_type) {
-    // JAVA_COMPARATOR
-    case 0x0:
-      comparator = reinterpret_cast<ROCKSDB_NAMESPACE::ComparatorJniCallback*>(
-          jcomparator_handle);
-      break;
-
-    // JAVA_NATIVE_COMPARATOR_WRAPPER
-    case 0x1:
-      comparator =
-          reinterpret_cast<ROCKSDB_NAMESPACE::Comparator*>(jcomparator_handle);
-      break;
-  }
-  auto* opt =
-      reinterpret_cast<ROCKSDB_NAMESPACE::ColumnFamilyOptions*>(jopt_handle);
+  ROCKSDB_NAMESPACE::Comparator* comparator =
+      ROCKSDB_NAMESPACE::AbstractComparatorJni::castCppComparator(jcomparator_handle, jcomparator_type);
+  auto* opt = reinterpret_cast<ROCKSDB_NAMESPACE::ColumnFamilyOptions*>(jopt_handle);
   opt->comparator = comparator;
 }