change in logic for xid equality

tools/merge/rocksmerge.go → tools/merge/main.go

@@ -78,7 +78,7 @@ func equalValue(a, b interface{}) bool {
 	aUid := x.Uid()
 	bUid := y.Uid()
 
-	return aUid == bUid ||
+	return (x.ValueBytes() == nil && y.ValueBytes() == nil && (aUid == bUid)) ||
 		((x.Uid() == math.MaxUint64 && y.Uid() == math.MaxUint64) &&
 			bytes.Compare(x.ValueBytes(), y.ValueBytes()) == 0)
 }
@@ -98,12 +98,8 @@ func MergeFolders(mergePath, destPath string) {
 	wopt = rocksdb.NewWriteOptions()
 	wopt.SetSync(true)
 
-	count := 0
-	for range dirList {
-		count++
-	}
-
-	pq = make(PriorityQueue, count)
+	pq = make(PriorityQueue, 0)
+	heap.Init(&pq)
 	var storeIters []*rocksdb.Iterator
 	for i, dir := range dirList {
 		mPath := path.Join(mergePath, dir.Name())
@@ -120,33 +116,24 @@ func MergeFolders(mergePath, destPath string) {
 			glog.WithField("path", mPath).Info("Store empty")
 			continue
 		}
-		pq[i] = &Item{
+		item := &Item{
 			key:      it.Key(),
 			value:    it.Value(),
 			storeIdx: i,
 		}
+		heap.Push(&pq, item)
 		storeIters = append(storeIters, it)
 	}
-	heap.Init(&pq)
-	mergeUsingHeap(destPath, storeIters)
-}
-
-func mergeUsingHeap(destPath string, storeIters []*rocksdb.Iterator) {
-	var opt *rocksdb.Options
-	var wopt *rocksdb.WriteOptions
-	opt = rocksdb.NewOptions()
-	opt.SetCreateIfMissing(true)
-	wopt = rocksdb.NewWriteOptions()
-	wopt.SetSync(true)
 
 	var db *rocksdb.DB
-	db, err := rocksdb.Open(destPath, opt)
+	db, err = rocksdb.Open(destPath, opt)
 	defer db.Close()
 	if err != nil {
 		glog.WithField("filepath", destPath).
 			Fatal("While opening store")
 	}
 
+	heap.Init(&pq)
 	var lastKey, lastValue []byte
 	for pq.Len() > 0 {
 		top := heap.Pop(&pq).(*Item)

tools/merge/rocksmerge_test.go → tools/merge/main_test.go

@@ -11,12 +11,11 @@ import (
 	"github.com/dgraph-io/dgraph/posting"
 	"github.com/dgraph-io/dgraph/store"
 	"github.com/dgraph-io/dgraph/store/rocksdb"
-	"github.com/dgraph-io/dgraph/tools/merge"
 	"github.com/dgraph-io/dgraph/uid"
 	"github.com/dgryski/go-farm"
 )
 
-func TestQuery(t *testing.T) {
+func TestMergeFolders(t *testing.T) {
 	logrus.SetLevel(logrus.DebugLevel)
 	rootDir, err := ioutil.TempDir("", "storetest_")
 	if err != nil {
@@ -82,7 +81,7 @@ func TestQuery(t *testing.T) {
 	ps1.Close()
 	ps2.Close()
 
-	main.MergeFolders(rootDir, destDir)
+	MergeFolders(rootDir, destDir)
 
 	var opt *rocksdb.Options
 	var ropt *rocksdb.ReadOptions