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
This commit is contained in:
Matt Jaffee 2019-02-20 09:45:49 -06:00
parent 5027a7a883
commit 0de419c95e
No known key found for this signature in database
GPG key ID: 08A3DFFF987B11BF
2 changed files with 77 additions and 2 deletions

View file

@ -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()

View file

@ -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, "")