From 52d43fb4e2076e4ccfafa241e6573766a725dec3 Mon Sep 17 00:00:00 2001
From: Matt Jaffee <matthew.jaffee@gmail.com>
Date: Mon, 11 Mar 2019 12:03:42 -0500
Subject: [PATCH] add smallPath for importRoaring

this converts the rowSet to a map from a slice which might be bad... benchmarks
will tell.
---
 fragment.go | 23 ++++++++++++++++++-----
 1 file changed, 18 insertions(+), 5 deletions(-)

diff --git a/fragment.go b/fragment.go
index 0c4ff5dd6..0801c8001 100644
--- a/fragment.go
+++ b/fragment.go
@@ -1729,13 +1729,13 @@ func (f *fragment) importRoaring(data []byte, clear bool) error {
 
 	// get a list of keys in order to update the cache
 	iter, _ := bm.Containers.Iterator(0)
-	rowSet := make([]uint64, 0)
+	rowSet := make(map[uint64]struct{})
 	var lastRow uint64 = math.MaxUint64
 
-	incomingCnt := 0
+	incomingCnt := uint64(0)
 	for iter.Next() {
 		key, c := iter.Value()
-		incomingCnt += int(c.N())
+		incomingCnt += uint64(c.N())
 
 		// virtual row for the current container
 		vRow := key >> shardVsContainerExponent
@@ -1744,10 +1744,23 @@ func (f *fragment) importRoaring(data []byte, clear bool) error {
 		if vRow == lastRow {
 			continue
 		}
-		rowSet = append(rowSet, vRow)
+		rowSet[vRow] = struct{}{}
 		lastRow = vRow
 	}
 
+	// take smallPath? TODO - ideally instead of checking f.storage.Any(), the
+	// test here would be if the storage size (in bytes) is significantly
+	// greater than the size of the incoming bits serialized as append
+	// operations. Getting the storage size might be a bit expensive though
+	// especially if the fragment isn't mapped.
+	if incomingCnt+uint64(f.opN) <= uint64(f.MaxOpN) && f.storage.Any() {
+		toSet, toClear := bm.Slice(), []uint64{}
+		if clear {
+			toSet, toClear = toClear, toSet
+		}
+		return f.importPositions(toSet, toClear, rowSet)
+	}
+
 	if clear {
 		bm = f.storage.Difference(bm)
 	} else if f.storage.Containers.Size() >= bm.Containers.Size() {
@@ -1757,7 +1770,7 @@ func (f *fragment) importRoaring(data []byte, clear bool) error {
 		bm.UnionInPlace(f.storage)
 	}
 
-	for _, rowID := range rowSet {
+	for rowID := range rowSet {
 		n := bm.CountRange(rowID*ShardWidth, (rowID+1)*ShardWidth)
 		f.cache.BulkAdd(rowID, n)
 	}