Merge pull request #1892 from jaffee/import-roaring-union

smallWrite path for import-roaring

cluster.go

@@ -184,6 +184,7 @@ type cluster struct { // nolint: maligned
 	ReplicaN int
 
 	// Threshold for logging long-running queries
+	// TODO(2.0) move this out of cluster. (why is it here??)
 	longQueryTime time.Duration
 
 	// Maximum number of Set() or Clear() commands per request.

fragment.go

@@ -1729,11 +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 := uint64(0)
 	for iter.Next() {
-		key, _ := iter.Value()
+		key, c := iter.Value()
+		incomingCnt += uint64(c.N())
 
 		// virtual row for the current container
 		vRow := key >> shardVsContainerExponent
@@ -1742,19 +1744,30 @@ 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.Count() > 0 {
-			bm = f.storage.Union(bm)
-		}
+	} else if f.storage.Any() {
+		bm = f.storage.Union(bm)
 	}
 
-	for _, rowID := range rowSet {
+	for rowID := range rowSet {
 		n := bm.CountRange(rowID*ShardWidth, (rowID+1)*ShardWidth)
 		f.cache.BulkAdd(rowID, n)
 	}

fragment_internal_test.go

@@ -766,6 +766,40 @@ func BenchmarkFragment_RepeatedSmallImports(b *testing.B) {
 	}
 }
 
+func BenchmarkFragment_RepeatedSmallImportsRoaring(b *testing.B) {
+	for _, numUpdates := range []int{100} {
+		for _, bitsPerUpdate := range []uint64{100, 1000} {
+			for _, numRows := range []uint64{1000, 100000, 1000000} {
+				for _, opN := range []int{1, 5000, 50000} {
+					b.Run(fmt.Sprintf("Rows%dUpdates%dBits%dOpN%d", numRows, numUpdates, bitsPerUpdate, opN), func(b *testing.B) {
+						for a := 0; a < b.N; a++ {
+							b.StopTimer()
+							// build the update data set all at once - this will get applied
+							// to a fragment in numUpdates batches
+							f := mustOpenFragment("i", "f", viewStandard, 0, "")
+							f.MaxOpN = opN
+							defer f.Clean(b)
+							err := f.importRoaring(getZipfRowsSliceRoaring(numRows, 1), false)
+							if err != nil {
+								b.Fatalf("importing base data for benchmark: %v", err)
+							}
+							for i := 0; i < numUpdates; i++ {
+								data := getUpdataRoaring(numRows, bitsPerUpdate, int64(i))
+								b.StartTimer()
+								err := f.importRoaring(data, false)
+								b.StopTimer()
+								if err != nil {
+									b.Fatalf("doing small roaring import: %v", err)
+								}
+							}
+						}
+					})
+				}
+			}
+		}
+	}
+}
+
 func BenchmarkFragment_RepeatedSmallValueImports(b *testing.B) {
 	initialCols := make([]uint64, 0, ShardWidth)
 	initialVals := make([]uint64, 0, ShardWidth)
@@ -1970,6 +2004,9 @@ func BenchmarkImportRoaring(b *testing.B) {
 }
 
 func BenchmarkImportRoaringConcurrent(b *testing.B) {
+	if testing.Short() {
+		b.SkipNow()
+	}
 	for _, numRows := range rowCases {
 		data := getZipfRowsSliceRoaring(numRows, 1)
 		b.Logf("%dRows: %.2fMB\n", numRows, float64(len(data))/1024/1024)
@@ -2003,6 +2040,9 @@ func BenchmarkImportRoaringConcurrent(b *testing.B) {
 	}
 }
 func BenchmarkImportRoaringUpdateConcurrent(b *testing.B) {
+	if testing.Short() {
+		b.SkipNow()
+	}
 	for _, numRows := range rowCases {
 		for _, numCols := range colCases {
 			data := getZipfRowsSliceRoaring(numRows, 1)
@@ -2162,6 +2202,41 @@ func BenchmarkImportIntoLargeFragment(b *testing.B) {
 	}
 }
 
+func BenchmarkImportRoaringIntoLargeFragment(b *testing.B) {
+	b.StopTimer()
+	initBigFrag()
+	updata := getUpdataRoaring(10000000, 11000, 0)
+	for i := 0; i < b.N; i++ {
+		origF, err := os.Open(bigFrag)
+		if err != nil {
+			b.Fatalf("opening frag file: %v", err)
+		}
+		fi, err := ioutil.TempFile(*TempDir, "")
+		if err != nil {
+			b.Fatalf("getting temp file: %v", err)
+		}
+		_, err = io.Copy(fi, origF)
+		if err != nil {
+			b.Fatalf("copying fragment file: %v", err)
+		}
+		origF.Close()
+		fi.Close()
+		nf := newFragment(fi.Name(), "i", "f", viewStandard, 0)
+		err = nf.Open()
+		if err != nil {
+			b.Fatalf("opening fragment: %v", err)
+		}
+		b.StartTimer()
+		err = nf.importRoaring(updata, false)
+		b.StopTimer()
+		if err != nil {
+			b.Fatalf("bulkImport: %v", err)
+		}
+
+		nf.Clean(b)
+	}
+}
+
 func TestGetZipfRowsSliceRoaring(t *testing.T) {
 	f := mustOpenFragment("i", "f", viewStandard, 0, DefaultCacheType)
 	data := getZipfRowsSliceRoaring(10, 1)

server/config.go

@@ -75,10 +75,11 @@ type Config struct {
 
 	Cluster struct {
 		// Disabled controls whether clustering functionality is enabled.
-		Disabled      bool          `toml:"disabled"`
-		Coordinator   bool          `toml:"coordinator"`
-		ReplicaN      int           `toml:"replicas"`
-		Hosts         []string      `toml:"hosts"`
+		Disabled    bool     `toml:"disabled"`
+		Coordinator bool     `toml:"coordinator"`
+		ReplicaN    int      `toml:"replicas"`
+		Hosts       []string `toml:"hosts"`
+		// TODO(2.0) move this out of cluster. (why is it here??)
 		LongQueryTime toml.Duration `toml:"long-query-time"`
 	} `toml:"cluster"`