From 0de419c95ed5db10dd0e280dffc7117a07856b66 Mon Sep 17 00:00:00 2001
From: Matt Jaffee <matthew.jaffee@gmail.com>
Date: Wed, 20 Feb 2019 09:45:49 -0600
Subject: [PATCH] add a SetBit/ClearBit path to bulkImport for small updates

also add benchmarks for this situation and set default MaxOpN higher which
benchmarks suggest is a good idea
---
 fragment.go               | 27 ++++++++++++++++++--
 fragment_internal_test.go | 52 +++++++++++++++++++++++++++++++++++++++
 2 files changed, 77 insertions(+), 2 deletions(-)

diff --git a/fragment.go b/fragment.go
index 58680e655..fe7de7b49 100644
--- a/fragment.go
+++ b/fragment.go
@@ -76,7 +76,7 @@ const (
 	HashBlockSize = 100
 
 	// defaultFragmentMaxOpN is the default value for Fragment.MaxOpN.
-	defaultFragmentMaxOpN = 2000
+	defaultFragmentMaxOpN = 5000
 
 	// Row ids used for boolean fields.
 	falseRowID = uint64(0)
@@ -1455,7 +1455,30 @@ func (f *fragment) bulkImport(rowIDs, columnIDs []uint64, options *ImportOptions
 }
 
 // bulkImportStandard performs a bulk import on a standard fragment.
-func (f *fragment) bulkImportStandard(rowIDs, columnIDs []uint64, options *ImportOptions) error {
+func (f *fragment) bulkImportStandard(rowIDs, columnIDs []uint64, options *ImportOptions) (err error) {
+	// first we'll try the "small update" path. If there aren't many bits being
+	// set, it isn't worth it to do a full import and snapshot. Instead we
+	// leverage individual set/clear bits which get set in memory and appended
+	// to the op log.
+	if len(columnIDs)+f.opN < f.MaxOpN {
+		for i := range rowIDs {
+			rowID, columnID := rowIDs[i], columnIDs[i]
+			if options.Clear {
+				_, err = f.clearBit(rowID, columnID)
+			} else {
+				_, err = f.setBit(rowID, columnID)
+			}
+			if err != nil {
+				return errors.Wrapf(err, "importing %dth bit clear:%v", i, options.Clear)
+			}
+		}
+		// forcibly recalculate the cache - setbit just calls invalidate, but in
+		// order to maintain parity with the "normal" bulkimport path, we want
+		// to recalculate it at the end of the import
+		f.cache.Recalculate()
+		return nil
+	} // end "small update" path - after this is the real bulk import path
+
 	// Create a temporary bitmap which will be populated by rowIDs and columnIDs
 	// and then merged into the existing fragment's bitmap.
 	localBitmap := roaring.NewBitmap()
diff --git a/fragment_internal_test.go b/fragment_internal_test.go
index ce7e06da9..2d0e35972 100644
--- a/fragment_internal_test.go
+++ b/fragment_internal_test.go
@@ -713,6 +713,58 @@ func BenchmarkFragment_ImportValue(b *testing.B) {
 	}
 }
 
+// BenchmarkFragment_RepeatedSmallImports tests the situation where updates are
+// constantly coming in across a large column space so that each fragment gets
+// only a few updates during each import.
+//
+// We test a variety of combinations of the number of separate updates(imports),
+// the number of bits in the import, the number of rows in the fragment (which
+// is a pretty good proxy for fragment size on disk), and the MaxOpN on the
+// fragment which controls how many set bits occur before a snapshot is done. If
+// the number of bits in a given import is greater than MaxOpN, bulkImport will
+// always go through the standard snapshotting import path.
+func BenchmarkFragment_RepeatedSmallImports(b *testing.B) {
+	for _, numUpdates := range []int{100} {
+		for _, bitsPerUpdate := range []int{100, 1000} {
+			for _, numRows := range []int{1000, 100000, 1000000} {
+				// build the update data set all at once - this will get applied
+				// to a fragment in numUpdates batches
+				updateRows := make([]uint64, numUpdates*bitsPerUpdate)
+				updateCols := make([]uint64, numUpdates*bitsPerUpdate)
+				for i := 0; i < numUpdates*bitsPerUpdate; i++ {
+					updateRows[i] = uint64(rand.Int63n(int64(numRows))) // row id
+					updateCols[i] = uint64(rand.Int63n(ShardWidth))     // column id
+				}
+				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()
+							f := mustOpenFragment("i", "f", viewStandard, 0, "")
+							f.MaxOpN = opN
+							defer f.Clean(b)
+							err := f.importRoaring(getZipfRowsSliceRoaring(uint64(numRows), 1), false)
+							if err != nil {
+								b.Fatalf("importing base data for benchmark: %v", err)
+							}
+							b.StartTimer()
+							for i := 0; i < numUpdates; i++ {
+								err := f.bulkImportStandard(
+									updateRows[bitsPerUpdate*i:bitsPerUpdate*(i+1)],
+									updateRows[bitsPerUpdate*i:bitsPerUpdate*(i+1)],
+									&ImportOptions{},
+								)
+								if err != nil {
+									b.Fatalf("doing small bulk import: %v", err)
+								}
+							}
+						}
+					})
+				}
+			}
+		}
+	}
+}
+
 // Ensure a fragment can snapshot correctly.
 func TestFragment_Snapshot(t *testing.T) {
 	f := mustOpenFragment("i", "f", viewStandard, 0, "")