Add a new MergeMultiOperator interface which supports PartialMergeMulti

merge_multi_operator.go

@@ -0,0 +1,26 @@
+package gorocksdb
+
+// #include "rocksdb/c.h"
+import "C"
+
+// MergeMultiOperator implements PartialMergeMulti(key []byte, operands [][]byte) ([]byte, err)
+// When a MergeOperator implements this interface, PartialMergeMulti
+// will be used instead of PartialMerge
+type MergeMultiOperator interface {
+	// PartialMerge performs merge on multiple operands
+	// when all of the operands are themselves merge operation types
+	// that you would have passed to a db.Merge() call in the same order
+	// (i.e.: db.Merge(key,operand[0]), followed by db.Merge(key,operand[1]),
+	// ... db.Merge(key, operand[n])).
+	//
+	// PartialMerge should combine them into a single merge operation.
+	// The return value should be constructed such that a call to
+	// db.Merge(key, new_value) would yield the same result as a call
+	// to db.Merge(key,operand[0]), followed by db.Merge(key,operand[1]),
+	// ... db.Merge(key, operand[n])).
+	//
+	// If it is impossible or infeasible to combine the operations, return false.
+	// The library will internally keep track of the operations, and apply them in the
+	// correct order once a base-value (a Put/Delete/End-of-Database) is seen.
+	PartialMergeMulti(key []byte, operands [][]byte) ([]byte, bool)
+}

merge_multi_operator_test.go

@@ -0,0 +1,82 @@
+package gorocksdb
+
+import (
+	"testing"
+
+	"github.com/facebookgo/ensure"
+)
+
+func TestMergeMultiOperator(t *testing.T) {
+	var (
+		givenKey     = []byte("hello")
+		startingVal  = []byte("foo")
+		mergeVal1    = []byte("bar")
+		mergeVal2    = []byte("baz")
+		fMergeResult = []byte("foobarbaz")
+		pMergeResult = []byte("barbaz")
+	)
+
+	merger := &mockMergeMultiOperator{
+		fullMerge: func(key, existingValue []byte, operands [][]byte) ([]byte, bool) {
+			ensure.DeepEqual(&fatalAsError{t}, key, givenKey)
+			ensure.DeepEqual(&fatalAsError{t}, existingValue, startingVal)
+			ensure.DeepEqual(&fatalAsError{t}, operands[0], pMergeResult)
+			return fMergeResult, true
+		},
+		partialMerge: func(key, leftOperand, rightOperand []byte) ([]byte, bool) {
+			t.FailNow() // this should never be called
+			return nil, false
+		},
+		partialMergeMulti: func(key []byte, operands [][]byte) ([]byte, bool) {
+			ensure.DeepEqual(&fatalAsError{t}, key, givenKey)
+			ensure.DeepEqual(&fatalAsError{t}, operands[0], mergeVal1)
+			ensure.DeepEqual(&fatalAsError{t}, operands[1], mergeVal2)
+			return pMergeResult, true
+		},
+	}
+	db := newTestDB(t, "TestMergeOperator", func(opts *Options) {
+		opts.SetMergeOperator(merger)
+	})
+	defer db.Close()
+
+	wo := NewDefaultWriteOptions()
+	defer wo.Destroy()
+
+	// insert a starting value and compact to trigger merges
+	ensure.Nil(t, db.Put(wo, givenKey, startingVal))
+
+	// trigger a compaction to ensure that a merge is performed
+	db.CompactRange(Range{nil, nil})
+
+	// we expect these two operands to be passed to merge multi
+	ensure.Nil(t, db.Merge(wo, givenKey, mergeVal1))
+	ensure.Nil(t, db.Merge(wo, givenKey, mergeVal2))
+
+	// trigger a compaction to ensure that a
+	// partial and full merge are performed
+	db.CompactRange(Range{nil, nil})
+
+	ro := NewDefaultReadOptions()
+	v1, err := db.Get(ro, givenKey)
+	defer v1.Free()
+	ensure.Nil(t, err)
+	ensure.DeepEqual(t, v1.Data(), fMergeResult)
+
+}
+
+type mockMergeMultiOperator struct {
+	fullMerge         func(key, existingValue []byte, operands [][]byte) ([]byte, bool)
+	partialMerge      func(key, leftOperand, rightOperand []byte) ([]byte, bool)
+	partialMergeMulti func(key []byte, operands [][]byte) ([]byte, bool)
+}
+
+func (m *mockMergeMultiOperator) Name() string { return "gorocksdb.test" }
+func (m *mockMergeMultiOperator) FullMerge(key, existingValue []byte, operands [][]byte) ([]byte, bool) {
+	return m.fullMerge(key, existingValue, operands)
+}
+func (m *mockMergeMultiOperator) PartialMerge(key, leftOperand, rightOperand []byte) ([]byte, bool) {
+	return m.partialMerge(key, leftOperand, rightOperand)
+}
+func (m *mockMergeMultiOperator) PartialMergeMulti(key []byte, operands [][]byte) ([]byte, bool) {
+	return m.partialMergeMulti(key, operands)
+}

merge_operator.go

@@ -28,6 +28,14 @@ type MergeOperator interface {
 	// internal corruption. This will be treated as an error by the library.
 	FullMerge(key, existingValue []byte, operands [][]byte) ([]byte, bool)
 
+	// The name of the MergeOperator.
+	Name() string
+}
+
+// PartialMerger implements PartialMerge(key, leftOperand, rightOperand []byte) ([]byte, err)
+// When a MergeOperator implements this interface, PartialMerge will be called in addition
+// to FullMerge for compactions across levels
+type PartialMerger interface {
 	// This function performs merge(left_op, right_op)
 	// when both the operands are themselves merge operation types
 	// that you would have passed to a db.Merge() call in the same order
@@ -42,9 +50,28 @@ type MergeOperator interface {
 	// The library will internally keep track of the operations, and apply them in the
 	// correct order once a base-value (a Put/Delete/End-of-Database) is seen.
 	PartialMerge(key, leftOperand, rightOperand []byte) ([]byte, bool)
+}
 
-	// The name of the MergeOperator.
-	Name() string
+// MultiMerger implements PartialMergeMulti(key []byte, operands [][]byte) ([]byte, err)
+// When a MergeOperator implements this interface, PartialMergeMulti will be called in addition
+// to FullMerge for compactions across levels
+type MultiMerger interface {
+	// PartialMerge performs merge on multiple operands
+	// when all of the operands are themselves merge operation types
+	// that you would have passed to a db.Merge() call in the same order
+	// (i.e.: db.Merge(key,operand[0]), followed by db.Merge(key,operand[1]),
+	// ... db.Merge(key, operand[n])).
+	//
+	// PartialMerge should combine them into a single merge operation.
+	// The return value should be constructed such that a call to
+	// db.Merge(key, new_value) would yield the same result as a call
+	// to db.Merge(key,operand[0]), followed by db.Merge(key,operand[1]),
+	// ... db.Merge(key, operand[n])).
+	//
+	// If it is impossible or infeasible to combine the operations, return false.
+	// The library will internally keep track of the operations, and apply them in the
+	// correct order once a base-value (a Put/Delete/End-of-Database) is seen.
+	PartialMergeMulti(key []byte, operands [][]byte) ([]byte, bool)
 }
 
 // NewNativeMergeOperator creates a MergeOperator object.
@@ -110,13 +137,22 @@ func gorocksdb_mergeoperator_partial_merge_multi(idx int, cKey *C.char, cKeyLen
 	success := true
 
 	merger := mergeOperators.Get(idx).(mergeOperatorWrapper).mergeOperator
-	leftOperand := operands[0]
-	for i := 1; i < int(cNumOperands); i++ {
-		newValue, success = merger.PartialMerge(key, leftOperand, operands[i])
-		if !success {
-			break
+
+	// check if this MergeOperator supports partial or multi merges
+	switch v := merger.(type) {
+	case MultiMerger:
+		newValue, success = v.PartialMergeMulti(key, operands)
+	case PartialMerger:
+		leftOperand := operands[0]
+		for i := 1; i < int(cNumOperands); i++ {
+			newValue, success = v.PartialMerge(key, leftOperand, operands[i])
+			if !success {
+				break
+			}
+			leftOperand = newValue
 		}
-		leftOperand = newValue
+	default:
+		success = false
 	}
 
 	newValueLen := len(newValue)

merge_operator_test.go

@@ -40,15 +40,146 @@ func TestMergeOperator(t *testing.T) {
 	ensure.DeepEqual(t, v1.Data(), givenMerged)
 }
 
+func TestPartialMergeOperator(t *testing.T) {
+	var (
+		givenKey     = []byte("hello")
+		startingVal  = []byte("foo")
+		mergeVal1    = []byte("bar")
+		mergeVal2    = []byte("baz")
+		fMergeResult = []byte("foobarbaz")
+		pMergeResult = []byte("barbaz")
+	)
+
+	merger := &mockMergePartialOperator{
+		fullMerge: func(key, existingValue []byte, operands [][]byte) ([]byte, bool) {
+			ensure.DeepEqual(&fatalAsError{t}, key, givenKey)
+			ensure.DeepEqual(&fatalAsError{t}, existingValue, startingVal)
+			ensure.DeepEqual(&fatalAsError{t}, operands[0], pMergeResult)
+			return fMergeResult, true
+		},
+		partialMerge: func(key, leftOperand, rightOperand []byte) ([]byte, bool) {
+			ensure.DeepEqual(&fatalAsError{t}, key, givenKey)
+			ensure.DeepEqual(&fatalAsError{t}, leftOperand, mergeVal1)
+			ensure.DeepEqual(&fatalAsError{t}, rightOperand, mergeVal2)
+			return pMergeResult, true
+		},
+	}
+	db := newTestDB(t, "TestMergeOperator", func(opts *Options) {
+		opts.SetMergeOperator(merger)
+	})
+	defer db.Close()
+
+	wo := NewDefaultWriteOptions()
+	defer wo.Destroy()
+
+	// insert a starting value and compact to trigger merges
+	ensure.Nil(t, db.Put(wo, givenKey, startingVal))
+
+	// trigger a compaction to ensure that a merge is performed
+	db.CompactRange(Range{nil, nil})
+
+	// we expect these two operands to be passed to merge partial
+	ensure.Nil(t, db.Merge(wo, givenKey, mergeVal1))
+	ensure.Nil(t, db.Merge(wo, givenKey, mergeVal2))
+
+	// trigger a compaction to ensure that a
+	// partial and full merge are performed
+	db.CompactRange(Range{nil, nil})
+
+	ro := NewDefaultReadOptions()
+	v1, err := db.Get(ro, givenKey)
+	defer v1.Free()
+	ensure.Nil(t, err)
+	ensure.DeepEqual(t, v1.Data(), fMergeResult)
+
+}
+
+func TestMergeMultiOperator(t *testing.T) {
+	var (
+		givenKey     = []byte("hello")
+		startingVal  = []byte("foo")
+		mergeVal1    = []byte("bar")
+		mergeVal2    = []byte("baz")
+		fMergeResult = []byte("foobarbaz")
+		pMergeResult = []byte("barbaz")
+	)
+
+	merger := &mockMergeMultiOperator{
+		fullMerge: func(key, existingValue []byte, operands [][]byte) ([]byte, bool) {
+			ensure.DeepEqual(&fatalAsError{t}, key, givenKey)
+			ensure.DeepEqual(&fatalAsError{t}, existingValue, startingVal)
+			ensure.DeepEqual(&fatalAsError{t}, operands[0], pMergeResult)
+			return fMergeResult, true
+		},
+		partialMergeMulti: func(key []byte, operands [][]byte) ([]byte, bool) {
+			ensure.DeepEqual(&fatalAsError{t}, key, givenKey)
+			ensure.DeepEqual(&fatalAsError{t}, operands[0], mergeVal1)
+			ensure.DeepEqual(&fatalAsError{t}, operands[1], mergeVal2)
+			return pMergeResult, true
+		},
+	}
+	db := newTestDB(t, "TestMergeOperator", func(opts *Options) {
+		opts.SetMergeOperator(merger)
+	})
+	defer db.Close()
+
+	wo := NewDefaultWriteOptions()
+	defer wo.Destroy()
+
+	// insert a starting value and compact to trigger merges
+	ensure.Nil(t, db.Put(wo, givenKey, startingVal))
+
+	// trigger a compaction to ensure that a merge is performed
+	db.CompactRange(Range{nil, nil})
+
+	// we expect these two operands to be passed to merge multi
+	ensure.Nil(t, db.Merge(wo, givenKey, mergeVal1))
+	ensure.Nil(t, db.Merge(wo, givenKey, mergeVal2))
+
+	// trigger a compaction to ensure that a
+	// partial and full merge are performed
+	db.CompactRange(Range{nil, nil})
+
+	ro := NewDefaultReadOptions()
+	v1, err := db.Get(ro, givenKey)
+	defer v1.Free()
+	ensure.Nil(t, err)
+	ensure.DeepEqual(t, v1.Data(), fMergeResult)
+
+}
+
+// Mock Objects
 type mockMergeOperator struct {
-	fullMerge    func(key, existingValue []byte, operands [][]byte) ([]byte, bool)
-	partialMerge func(key, leftOperand, rightOperand []byte) ([]byte, bool)
+	fullMerge func(key, existingValue []byte, operands [][]byte) ([]byte, bool)
 }
 
 func (m *mockMergeOperator) Name() string { return "gorocksdb.test" }
 func (m *mockMergeOperator) FullMerge(key, existingValue []byte, operands [][]byte) ([]byte, bool) {
 	return m.fullMerge(key, existingValue, operands)
 }
-func (m *mockMergeOperator) PartialMerge(key, leftOperand, rightOperand []byte) ([]byte, bool) {
+
+type mockMergeMultiOperator struct {
+	fullMerge         func(key, existingValue []byte, operands [][]byte) ([]byte, bool)
+	partialMergeMulti func(key []byte, operands [][]byte) ([]byte, bool)
+}
+
+func (m *mockMergeMultiOperator) Name() string { return "gorocksdb.multi" }
+func (m *mockMergeMultiOperator) FullMerge(key, existingValue []byte, operands [][]byte) ([]byte, bool) {
+	return m.fullMerge(key, existingValue, operands)
+}
+func (m *mockMergeMultiOperator) PartialMergeMulti(key []byte, operands [][]byte) ([]byte, bool) {
+	return m.partialMergeMulti(key, operands)
+}
+
+type mockMergePartialOperator struct {
+	fullMerge    func(key, existingValue []byte, operands [][]byte) ([]byte, bool)
+	partialMerge func(key, leftOperand, rightOperand []byte) ([]byte, bool)
+}
+
+func (m *mockMergePartialOperator) Name() string { return "gorocksdb.partial" }
+func (m *mockMergePartialOperator) FullMerge(key, existingValue []byte, operands [][]byte) ([]byte, bool) {
+	return m.fullMerge(key, existingValue, operands)
+}
+func (m *mockMergePartialOperator) PartialMerge(key, leftOperand, rightOperand []byte) ([]byte, bool) {
 	return m.partialMerge(key, leftOperand, rightOperand)
 }