diff --git a/cluster.go b/cluster.go index 4565e91f3..5c5c0e7f1 100644 --- a/cluster.go +++ b/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. diff --git a/fragment.go b/fragment.go index 067814a0b..32b870fb9 100644 --- a/fragment.go +++ b/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) } diff --git a/fragment_internal_test.go b/fragment_internal_test.go index 3a528c840..dc2b786b4 100644 --- a/fragment_internal_test.go +++ b/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) diff --git a/server/config.go b/server/config.go index 6223f07da..9c787f92e 100644 --- a/server/config.go +++ b/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"`