diff --git a/.gitignore b/.gitignore index 9bf572296..82a553aed 100644 --- a/.gitignore +++ b/.gitignore @@ -5,7 +5,6 @@ vendor .DS_Store build *~ -lattice release-pilosa-fsck.*.*.tar.gz /log.* /tourna.log.* diff --git a/.gitmodules b/.gitmodules new file mode 100644 index 000000000..afb2db2d6 --- /dev/null +++ b/.gitmodules @@ -0,0 +1,3 @@ +[submodule "lattice"] + path = lattice + url = git@github.com:molecula/lattice.git diff --git a/Makefile b/Makefile index 9114fda54..2e6cc7331 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: build check-clean clean build-lattice cover cover-viz default docker docker-build docker-test docker-tag-push generate generate-protoc generate-pql generate-statik gometalinter install install-build-deps install-golangci-lint install-gometalinter install-protoc install-protoc-gen-gofast install-peg install-statik prerelease prerelease-upload release release-build test testv testv-race testvsub testvsub-race test-txstore-rbf +.PHONY: build check-clean clean build-lattice cover cover-viz default docker docker-build docker-test docker-tag-push generate generate-protoc generate-pql generate-statik gometalinter install install-build-deps install-golangci-lint install-gometalinter install-protoc install-protoc-gen-gofast install-peg install-statik prerelease prerelease-upload release release-build test testv testv-race testvsub testvsub-race test-txstore-rbf lattice CLONE_URL=github.com/pilosa/pilosa MOD_VERSION=v2 @@ -155,11 +155,17 @@ install: install-bench: go install -tags='$(BUILD_TAGS)' -ldflags $(LDFLAGS) $(FLAGS) ./cmd/pilosa-bench +# Ensure lattice is cloned and the pinned version is checked out lattice: - git clone git@github.com:molecula/lattice.git + git submodule update --init +# Build the lattice assets build-lattice: lattice require-yarn - cd lattice && git pull && yarn install && yarn build + cd lattice && yarn install && yarn build + +# Upgrade lattice to the latest version +upgrade-lattice: lattice + git submodule update --remote # `go generate` protocol buffers generate-protoc: require-protoc require-protoc-gen-gofast diff --git a/README.md b/README.md index 4543a3524..f8ff52c19 100644 --- a/README.md +++ b/README.md @@ -58,6 +58,15 @@ Check out how the Pilosa [Data Model](https://www.pilosa.com/docs/data-model/) w You can interact with Pilosa directly in the console using the [Pilosa Query Language](https://www.pilosa.com/docs/query-language/) (PQL). +## Upgrading UI + +Lattice is now a submodule of Pilosa, for the purpose of keeping the frontend and backend components of the UI system synchronized. When making a change to the UI that requires changing both the backend and frontend, follow steps in this order: + +1. merge frontend (Lattice) PR +2. run `make upgrade-lattice` +3. run `git add lattice`, `git commit -m"Upgrade Lattice"`, `git push` +4. merge backend (Pilosa) PR + ## Client Libraries There are supported libraries for the following languages: diff --git a/cmd/pilosa-bench/main.go b/cmd/pilosa-bench/main.go index b33bf0fcf..1502dae23 100644 --- a/cmd/pilosa-bench/main.go +++ b/cmd/pilosa-bench/main.go @@ -16,12 +16,14 @@ package main import ( "context" + "expvar" "flag" "fmt" "io/ioutil" "log" "math/rand" "net/http" + _ "net/http/pprof" "os" "sort" "strings" @@ -32,6 +34,14 @@ import ( "golang.org/x/sync/errgroup" ) +var ( + requestCountVar = expvar.NewInt("request_count") + requestCurrentLatencyVar = expvar.NewFloat("request_current_latency") // seconds + requestAvgLatencyVar = expvar.NewFloat("request_avg_latency") // seconds + requestTotalLatencyVar = expvar.NewFloat("request_total_latency") // seconds + requestPerSecVar = expvar.NewFloat("request_per_sec") +) + func main() { if err := run(context.Background(), os.Args[1:]); err == flag.ErrHelp { os.Exit(1) @@ -86,6 +96,13 @@ func run(ctx context.Context, args []string) (err error) { return err } + // Set up HTTP endpoint to provide /debug endpoints. + fmt.Println("Serving debug endpoint at http://localhost:7070/debug") + go func() { _ = http.ListenAndServe(":7070", nil) }() + + // Run separate goroutine to calculate the current req/sec & latency. + go monitor() + // Load all id/keys for each field. log.Printf("loading field identifiers") fieldIDMap, err := loadFields(ctx, client) @@ -145,10 +162,15 @@ func run(ctx context.Context, args []string) (err error) { log.Printf("[query] %s", q) g.Go(func() error { + t := time.Now() _, err = client.Query(ctx, key.index, &pilosa.QueryRequest{Index: key.index, Query: q}) if err != nil { return err } + elapsed := time.Since(t).Seconds() + requestCountVar.Add(1) + requestTotalLatencyVar.Add(elapsed) + requestAvgLatencyVar.Set(requestTotalLatencyVar.Value() / float64(requestCountVar.Value())) return nil }) } @@ -156,6 +178,29 @@ func run(ctx context.Context, args []string) (err error) { return g.Wait() } +// monitor runs in a separate goroutine and updates metrics. +func monitor() { + ticker := time.NewTicker(1 * time.Second) + defer ticker.Stop() + + var lastTime time.Time + var lastN int64 + var lastLatency float64 + for range ticker.C { + now, n := time.Now(), requestCountVar.Value() + latency := requestTotalLatencyVar.Value() + + if !lastTime.IsZero() { + elapsed := lastTime.Sub(now).Seconds() + if n > 0 { + requestCurrentLatencyVar.Set((lastLatency - latency) / float64(n)) + } + requestPerSecVar.Set(float64(lastN-n) / elapsed) + } + lastTime, lastN, lastLatency = now, n, latency + } +} + func generateQuery(typ, index, field string, info *pilosa.FieldInfo, identifiers *pilosa.RowIdentifiers, opt queryOptions) (string, error) { switch typ { case "row": diff --git a/executor.go b/executor.go index ae1071b33..b1a0ae1eb 100644 --- a/executor.go +++ b/executor.go @@ -2797,6 +2797,12 @@ func (e *executor) executeGroupBy(ctx context.Context, qcx *Qcx, index string, c } if hasLimit || hasCol { // we need to perform this query cluster-wide ahead of executeGroupByShard + if idx, ok := child.Args["valueidx"].(int64); ok { + // The rows query was already completed on the initiating node. + childRows[i] = opt.EmbeddedData[idx].Columns() + continue + } + childRows[i], err = e.executeRows(ctx, qcx, index, child, shards, opt) if err != nil { return nil, errors.Wrap(err, "getting rows for ") @@ -2804,6 +2810,13 @@ func (e *executor) executeGroupBy(ctx context.Context, qcx *Qcx, index string, c if len(childRows[i]) == 0 { // there are no results because this field has no values. return &GroupCounts{}, nil } + + // Stuff the result into opt.EmbeddedData so that it gets sent to other nodes in the map-reduce. + // This is flagged as "NoSplit" to ensure that the entire row gets sent out. + rowsRow := NewRow(childRows[i]...) + rowsRow.NoSplit = true + child.Args["valueidx"] = int64(len(opt.EmbeddedData)) + opt.EmbeddedData = append(opt.EmbeddedData, rowsRow) } } @@ -5527,6 +5540,10 @@ func makeEmbeddedDataForShards(allRows []*Row, shards []uint64) []*Row { if row == nil || len(row.segments) == 0 { continue } + if row.NoSplit { + newRows[i] = row + continue + } segments := row.segments segmentIndex := 0 newRows[i] = &Row{ @@ -7381,9 +7398,11 @@ func newGroupByIterator(executor *executor, qcx *Qcx, rowIDs []RowIDs, children idx := holder.Index(index) var ( - fieldName string - viewName string - ok bool + fieldName string + viewName string + ok bool + views []string + isTimeField bool ) ignorePrev := false for i, call := range children { @@ -7397,9 +7416,42 @@ func newGroupByIterator(executor *executor, qcx *Qcx, rowIDs []RowIDs, children gbi.fields[i].Field = fieldName switch field.Type() { - case FieldTypeSet, FieldTypeTime, FieldTypeMutex, FieldTypeBool: + case FieldTypeSet, FieldTypeMutex, FieldTypeBool: viewName = viewStandard + case FieldTypeTime: + var ( + err error + v interface{} + ) + // Parse "from" time, if set. + var ( + hasFrom bool + fromTime time.Time + ) + if v, hasFrom = call.Args["from"]; hasFrom { + if fromTime, err = parseTime(v); err != nil { + return nil, errors.Wrap(err, "parsing from time") + } + } + + // Parse "to" time, if set. + var ( + hasTo bool + toTime time.Time + ) + if v, hasTo = call.Args["to"]; hasTo { + if toTime, err = parseTime(v); err != nil { + return nil, errors.Wrap(err, "parsing to time") + } + } + + if hasTo || hasFrom { + views = viewsByTimeRange(viewStandard, fromTime, toTime, field.TimeQuantum()) + isTimeField = true + } else { + viewName = viewStandard + } case FieldTypeInt: viewName = viewBSIGroupPrefix + fieldName @@ -7408,11 +7460,6 @@ func newGroupByIterator(executor *executor, qcx *Qcx, rowIDs []RowIDs, children call.Name, strings.Join([]string{FieldTypeSet, FieldTypeTime, FieldTypeMutex, FieldTypeBool, FieldTypeInt}, ",")) } - // Fetch fragment. - frag := holder.fragment(index, fieldName, viewName, shard) - if frag == nil { // this means this whole shard doesn't have all it needs to continue - return nil, nil - } filters := []roaring.BitmapFilter{} if len(rowIDs[i]) > 0 { filters = append(filters, roaring.NewBitmapRowsFilter(rowIDs[i])) @@ -7424,9 +7471,35 @@ func newGroupByIterator(executor *executor, qcx *Qcx, rowIDs []RowIDs, children } defer finisher(&err0) - gbi.rowIters[i], err = frag.rowIterator(tx, i != 0, filters...) - if err != nil { - return nil, err + // Fetch fragment(s), get rowIterator + if isTimeField { + var fragments []*fragment + for _, viewName := range views { + fragment := holder.fragment(index, fieldName, viewName, shard) + if fragment != nil { + fragments = append(fragments, fragment) + } + } + if len(fragments) == 0 { + // whole shard doesn't have all it needs to continue ? + return nil, nil + } + + gbi.rowIters[i], err = timeFragmentsRowIterator(fragments, tx, i != 0, filters...) + if err != nil { + return nil, err + } + } else { + frag := holder.fragment(index, fieldName, viewName, shard) + if frag == nil { // this means this whole shard doesn't have all it needs to continue + return nil, nil + } + + gbi.rowIters[i], err = frag.rowIterator(tx, i != 0, filters...) + if err != nil { + return nil, err + } + } prev, hasPrev, err := call.UintArg("previous") diff --git a/executor_test.go b/executor_test.go index 552aed964..6a202ca3a 100644 --- a/executor_test.go +++ b/executor_test.go @@ -5517,7 +5517,7 @@ func TestExecutor_Execute_DistinctFailure(t *testing.T) { func TestExecutor_Execute_GroupBy(t *testing.T) { groupByTest := func(t *testing.T, clusterSize int) { - c := test.MustRunCluster(t, 1) + c := test.MustRunCluster(t, clusterSize) defer c.Close() c.CreateField(t, "i", pilosa.IndexOptions{}, "general") c.CreateField(t, "i", pilosa.IndexOptions{}, "sub") @@ -5924,7 +5924,7 @@ func TestExecutor_Execute_GroupBy(t *testing.T) { }) } - for size := range []int{1, 3} { + for _, size := range []int{1, 3} { t.Run(fmt.Sprintf("%d_nodes", size), func(t *testing.T) { groupByTest(t, size) }) @@ -6863,6 +6863,30 @@ func variousQueries(t *testing.T, clusterSize int) { {"icecream", "userF"}, }) + // Create and populate "places_visited" time field. + c.CreateField(t, "users", pilosa.IndexOptions{Keys: true, TrackExistence: true}, "places_visited", pilosa.OptFieldKeys(), pilosa.OptFieldTypeTime(pilosa.TimeQuantum("YM"))) + ts2019Jan01 := int64(1546300800) * 1e+9 // 2019 January 1st 0:00:00 + ts2019Aug01 := int64(1564617600) * 1e+9 // 2019 August 1st 0:00:00 + ts2020Jan01 := int64(1577836800) * 1e+9 // 2020 January 1st 0:00:00 + c.ImportTimeQuantumKey(t, "users", "places_visited", []test.TimeQuantumKey{ + // 2019 January: nairobi, paris, austin, toronto + {RowKey: "nairobi", ColKey: "userB", Ts: ts2019Jan01}, + {RowKey: "paris", ColKey: "userC", Ts: ts2019Jan01}, + {RowKey: "austin", ColKey: "userF", Ts: ts2019Jan01}, + {RowKey: "toronto", ColKey: "userA", Ts: ts2019Jan01}, + // 2019 August: toronto only + {RowKey: "toronto", ColKey: "userB", Ts: ts2019Aug01}, + {RowKey: "toronto", ColKey: "userC", Ts: ts2019Aug01}, + // 2020: toronto, mombasa, sydney, nairobi + {RowKey: "toronto", ColKey: "userB", Ts: ts2020Jan01}, + {RowKey: "toronto", ColKey: "userD", Ts: ts2020Jan01}, + {RowKey: "toronto", ColKey: "userE", Ts: ts2020Jan01}, + {RowKey: "toronto", ColKey: "userF", Ts: ts2020Jan01}, + {RowKey: "mombasa", ColKey: "userA", Ts: ts2020Jan01}, + {RowKey: "sydney", ColKey: "userD", Ts: ts2020Jan01}, + {RowKey: "nairobi", ColKey: "userE", Ts: ts2020Jan01}, + }) + // Create and populate "affinity" int field with negative, positive, zero and null values. c.CreateField(t, "users", pilosa.IndexOptions{Keys: true, TrackExistence: true}, "affinity", pilosa.OptFieldTypeInt(-1000, 1000)) c.ImportIntKey(t, "users", "affinity", []test.IntKey{ @@ -6900,6 +6924,66 @@ func variousQueries(t *testing.T, clusterSize int) { qrVerifier func(t *testing.T, resp pilosa.QueryResponse) csvVerifier string }{ + { // 2020 & 2019 All + query: `GroupBy(Rows(places_visited, from='2019-01-01T00:00', to='2020-12-31T23:59'))`, + csvVerifier: `nairobi,2 +paris,1 +austin,1 +toronto,6 +mombasa,1 +sydney,1 +`, + }, + { // 2019 January only + query: `GroupBy(Rows(places_visited, from='2019-01-01T00:00', to='2019-02-01T00:00'))`, + csvVerifier: `nairobi,1 +paris,1 +austin,1 +toronto,1 +`, + }, + { // 2019 All + query: `GroupBy(Rows(places_visited, from='2019-01-01T00:00', to='2019-12-31T23:59'))`, + csvVerifier: `nairobi,1 +paris,1 +austin,1 +toronto,3 +`, + }, + { // 2019 All, this excludes userC (who likes pangolin & icecream) from the count. + // UserC visited Paris and Toronto in 2019 + query: `GroupBy( + Rows(places_visited, from='2019-01-01T00:00', to='2019-12-31T23:59'), + filter=Not(Intersect(Row(likes='pangolin'), Row(likes='icecream'))) + )`, + csvVerifier: `nairobi,1 +austin,1 +toronto,2 +`, + }, + { // After excluding UserC, this gets the sum of the networth of everyone per cities travelled + query: `GroupBy( + Rows(places_visited, from='2019-01-01T00:00', to='2019-12-31T23:59'), + filter=Not(Intersect(Row(likes='pangolin'), Row(likes='icecream'))), + aggregate=Sum(field=net_worth) + )`, + csvVerifier: `nairobi,1,10 +austin,1,100000 +toronto,2,11 +`, + }, + { // 2020 & 2019 All + query: `Rows(places_visited, from='2019-01-01T00:00', to='2020-12-31T23:59')`, + csvVerifier: "nairobi\nparis\naustin\ntoronto\nmombasa\nsydney\n", + }, + { // 2019 All + query: `Rows(places_visited, from='2019-01-01T00:00', to='2019-12-31T23:59')`, + csvVerifier: "nairobi\nparis\naustin\ntoronto\n", + }, + { // 2019 January only + query: `Rows(places_visited, from='2019-01-01T00:00', to='2019-02-01T00:00')`, + csvVerifier: "nairobi\nparis\naustin\ntoronto\n", + }, { query: "Count(All())", qrVerifier: func(t *testing.T, resp pilosa.QueryResponse) { diff --git a/fragment.go b/fragment.go index aa93dcc9a..8592182f4 100644 --- a/fragment.go +++ b/fragment.go @@ -3329,6 +3329,95 @@ func (f *fragment) rowIterator(tx Tx, wrap bool, filters ...roaring.BitmapFilter return f.setRowIterator(tx, wrap, filters...) } +type timeRowIterator struct { + tx Tx + cur int + wrap bool + allRowIDs []uint64 + rowIDToFragments map[uint64][]*fragment +} + +func timeFragmentsRowIterator(fragments []*fragment, tx Tx, wrap bool, filters ...roaring.BitmapFilter) (rowIterator, error) { + if len(fragments) == 0 { + return nil, fmt.Errorf("there should be at least 1 fragment") + } else if len(fragments) == 1 { + return fragments[0].setRowIterator(tx, wrap, filters...) + } + + it := &timeRowIterator{ + tx: tx, + cur: 0, + wrap: wrap, + } + + // create a sort of inverted index that maps each + // rowID back to the fragments that have that rowID + rowIDToFragments := make(map[uint64][]*fragment) + for _, f := range fragments { + rowIDs, err := f.rows(context.Background(), tx, 0, filters...) + if err != nil { + return nil, err + } + for _, rowID := range rowIDs { + fs := append(rowIDToFragments[rowID], f) + rowIDToFragments[rowID] = fs + } + } + + // if len(rowIDToFragments) == 0 what to do ?? + // ie all fragments returned empty rowIDs, is this possible + // is this an error + + // collect all rowIDs from inverted index to a slice + allRowIDs := make([]uint64, len(rowIDToFragments)) + i := 0 + for rowID := range rowIDToFragments { + allRowIDs[i] = rowID + i++ + } + sort.Slice(allRowIDs, func(i, j int) bool { return allRowIDs[i] < allRowIDs[j] }) + + it.rowIDToFragments = rowIDToFragments + it.allRowIDs = allRowIDs + + return it, nil +} + +func (it *timeRowIterator) Seek(rowID uint64) { + idx := sort.Search(len(it.allRowIDs), func(i int) bool { + return it.allRowIDs[i] >= rowID + }) + it.cur = idx +} + +func (it *timeRowIterator) Next() (r *Row, rowID uint64, _ *int64, wrapped bool, err error) { + if it.cur >= len(it.allRowIDs) { + if !it.wrap || len(it.allRowIDs) == 0 { + return nil, 0, nil, true, nil + } + it.Seek(0) + wrapped = true + } + + // gather rows + rowID = it.allRowIDs[it.cur] + fragments := it.rowIDToFragments[rowID] + rows := make([]*Row, 0, len(fragments)) + for _, fragment := range fragments { + row, err := fragment.row(it.tx, rowID) + if err != nil { + return row, rowID, nil, wrapped, err + } + rows = append(rows, row) + } + + // union rows + r = rows[0].Union(rows[1:]...) + + it.cur++ + return r, rowID, nil, wrapped, nil +} + type intRowIterator struct { f *fragment values int64Slice // sorted slice of int values diff --git a/lattice b/lattice new file mode 160000 index 000000000..36f453c1e --- /dev/null +++ b/lattice @@ -0,0 +1 @@ +Subproject commit 36f453c1ea3bf86c546a8ad4a88f2a926724d683 diff --git a/pql/ast.go b/pql/ast.go index 40e2acaba..54ae2dee7 100644 --- a/pql/ast.go +++ b/pql/ast.go @@ -388,6 +388,7 @@ var callInfoByFunc = map[string]callInfo{ "from": nil, "to": nil, "like": "", + "valueidx": int64(0), }, }, "Shift": {allowUnknown: false, diff --git a/rbf/rbf.go b/rbf/rbf.go index f13a5a7fc..365907121 100644 --- a/rbf/rbf.go +++ b/rbf/rbf.go @@ -462,7 +462,7 @@ func (c *leafCell) lastValue(tx *Tx) uint16 { // We have to take int32 rather than uint16 because the interval is [start, end), // and otherwise we have no way to ask to count the entire container (the // high bit will be missed). -func (c *leafCell) countRange(start, end int32) (n int) { +func (c *leafCell) countRange(tx *Tx, start, end int32) (n int) { // If the full range is being queried, simply use the precalculated count. if start == 0 && end > math.MaxUint16 { return c.BitN @@ -475,6 +475,10 @@ func (c *leafCell) countRange(start, end int32) (n int) { return int(roaring.RunCountRange(toInterval16(c.Data), start, end)) case ContainerTypeBitmap: return int(roaring.BitmapCountRange(toArray64(c.Data), start, end)) + case ContainerTypeBitmapPtr: + _, a, err := tx.leafCellBitmap(toPgno(c.Data)) + panicOn(err) + return int(roaring.BitmapCountRange(a, start, end)) default: panic(fmt.Sprintf("invalid container type: %d", c.Type)) } diff --git a/rbf/tx.go b/rbf/tx.go index 705edfc5b..a952104ec 100644 --- a/rbf/tx.go +++ b/rbf/tx.go @@ -1318,7 +1318,6 @@ func (tx *Tx) CountRange(name string, start, end uint64) (uint64, error) { } else if err != nil { return 0, err } - var n uint64 for { if err := csr.Next(); err == io.EOF { @@ -1341,7 +1340,7 @@ func (tx *Tx) CountRange(name string, start, end uint64) (uint64, error) { // If range is entirely in one container then just count that range. if skey == ekey { - return uint64(c.countRange(int32(lowbits(start)), ebits)), nil + return uint64(c.countRange(tx, int32(lowbits(start)), ebits)), nil } // INVAR: skey < ekey @@ -1351,7 +1350,7 @@ func (tx *Tx) CountRange(name string, start, end uint64) (uint64, error) { break } if k == skey { - n += uint64(c.countRange(int32(lowbits(start)), roaring.MaxContainerVal+1)) + n += uint64(c.countRange(tx, int32(lowbits(start)), roaring.MaxContainerVal+1)) continue } if k < ekey { @@ -1359,7 +1358,7 @@ func (tx *Tx) CountRange(name string, start, end uint64) (uint64, error) { continue } if k == ekey && ebits > 0 { - n += uint64(c.countRange(0, ebits)) + n += uint64(c.countRange(tx, 0, ebits)) break } } diff --git a/roaring/roaring.go b/roaring/roaring.go index 2705a6ba3..9c3a97aa9 100644 --- a/roaring/roaring.go +++ b/roaring/roaring.go @@ -4144,26 +4144,26 @@ func intersectionAnyRunBitmap(a, b *Container) bool { bb := b.bitmap()[:1024] runs := a.runs() for _, r := range runs { - loWord, loBit := r.Start/64, r.Start%64 - hiWord, hiBit := r.Last/64, r.Last%64 - if loBit != 0 { - w := bb[loWord] - mask := (uint64(1) << loBit) - 1 - if w&^mask != 0 { + if r.Start/64 == r.Last/64 { + mask := (^uint64(0) << (r.Start % 64)) &^ + (^uint64(0) << ((r.Last % 64) + 1)) + if mask&bb[r.Start/64] != 0 { return true } + continue } - for i := loWord; i < hiWord; i++ { + + firstWord, lastWord := r.Start/64, r.Last/64 + for i := firstWord + 1; i < lastWord; i++ { if bb[i] != 0 { return true } } - if hiBit != 0 { - w := bb[hiWord] - mask := (uint64(1) << hiBit) - 1 - if w&mask != 0 { - return true - } + + firstMask := ^uint64(0) << (r.Start % 64) + lastMask := ^(^uint64(0) << ((r.Last % 64) + 1)) + if (firstMask&bb[firstWord])|(lastMask&bb[lastWord]) != 0 { + return true } } return false diff --git a/roaring/roaring_container_test.go b/roaring/roaring_container_test.go index 3b747b5c9..9fbf4b214 100644 --- a/roaring/roaring_container_test.go +++ b/roaring/roaring_container_test.go @@ -97,3 +97,15 @@ func TestIntersectVariants(t *testing.T) { } } } + +func TestIntersectionAnyRunBitmapSingleWordRegression(t *testing.T) { + // In a previous version, single-word runs would match any bit within the word. + // Verify that this no longer happens. + any := intersectionAnyRunBitmap( + NewContainerRun([]Interval16{{1, 2}}), + NewContainerBitmapN([]uint64{0b1001}, 2), + ) + if any { + t.Errorf("matched an exclusive single-word run") + } +} diff --git a/row.go b/row.go index 46a0ec478..3ff47cba5 100644 --- a/row.go +++ b/row.go @@ -42,6 +42,10 @@ type Row struct { // query. Knowing the index and field, we can figure out how to // interpret the row data. Field string + + // NoSplit indicates that this row may not be split. + // This is used for `Rows` calls in a GroupBy. + NoSplit bool } // NewRow returns a new instance of Row. diff --git a/scripts/bench_read.sh b/scripts/bench_read.sh index bfbf6baac..a070bfb94 100755 --- a/scripts/bench_read.sh +++ b/scripts/bench_read.sh @@ -25,14 +25,21 @@ for TYPE in row row-bsi row-range count intersect union difference xor groupby t do WORKFLOW_PATH="${BASH_SOURCE%/*}/etc/gloat/query.${TYPE}.yml" WORKFLOW_NAME="$(gloat workflow name $WORKFLOW_PATH)" - TITLE="$WORKFLOW_NAME, $DATE ($SHA)" - + # Execute RBF/Roaring benchmark. + STARTTIME=$(date +%s) RBF_PATH=gloat/data/query/${TYPE}/rbf/${DATE}.tar.gz TXSRC=rbf gloat run -v -o "$RBF_PATH" $WORKFLOW_PATH + RBF_ELAPSED=$(($(date +%s) - $STARTTIME)) + RBF_LATENCY=$(gloat metric -n -name request_avg_latency "$RBF_PATH") + STARTTIME=$(date +%s) ROARING_PATH=gloat/data/query/${TYPE}/roaring/${DATE}.tar.gz TXSRC=roaring gloat run -v -o "$ROARING_PATH" $WORKFLOW_PATH + ROARING_ELAPSED=$(($(date +%s) - $STARTTIME)) + ROARING_LATENCY=$(gloat metric -n -name request_avg_latency "$ROARING_PATH") + + TITLE="$WORKFLOW_NAME, $DATE ($SHA) elapsed rbf=$RBF_ELAPSEDroaring=$ROARING_ELAPSED> latency rbf=$RBF_LATENCY roaring=$ROARING_LATENCY" # Generate graph from results. gloat graph -layout 2,5 -size 5120,820 -title "$TITLE" -name utime,stime,heap_alloc,heap_inuse,heap_objects,num_gc,rchar,wchar,syscr,syscw -series rbf,roaring -o /tmp/output.png $RBF_PATH $ROARING_PATH diff --git a/scripts/etc/gloat/query.count.yml b/scripts/etc/gloat/query.count.yml index 5690c13a8..b0215d62c 100644 --- a/scripts/etc/gloat/query.count.yml +++ b/scripts/etc/gloat/query.count.yml @@ -8,3 +8,4 @@ health_regexp: "NORMAL" vars_urls: - http://localhost:10101/debug/vars + - http://localhost:7070/debug/vars diff --git a/scripts/etc/gloat/query.difference.yml b/scripts/etc/gloat/query.difference.yml index 3a73a8e46..d3b17694f 100644 --- a/scripts/etc/gloat/query.difference.yml +++ b/scripts/etc/gloat/query.difference.yml @@ -8,3 +8,4 @@ health_regexp: "NORMAL" vars_urls: - http://localhost:10101/debug/vars + - http://localhost:7070/debug/vars diff --git a/scripts/etc/gloat/query.groupby.yml b/scripts/etc/gloat/query.groupby.yml index a5c4c5d00..25b97fce4 100644 --- a/scripts/etc/gloat/query.groupby.yml +++ b/scripts/etc/gloat/query.groupby.yml @@ -8,3 +8,4 @@ health_regexp: "NORMAL" vars_urls: - http://localhost:10101/debug/vars + - http://localhost:7070/debug/vars diff --git a/scripts/etc/gloat/query.intersect.yml b/scripts/etc/gloat/query.intersect.yml index 00bbfbd2e..4cb0993e9 100644 --- a/scripts/etc/gloat/query.intersect.yml +++ b/scripts/etc/gloat/query.intersect.yml @@ -8,3 +8,4 @@ health_regexp: "NORMAL" vars_urls: - http://localhost:10101/debug/vars + - http://localhost:7070/debug/vars diff --git a/scripts/etc/gloat/query.row-bsi.yml b/scripts/etc/gloat/query.row-bsi.yml index 04fe472eb..7c86b5a41 100644 --- a/scripts/etc/gloat/query.row-bsi.yml +++ b/scripts/etc/gloat/query.row-bsi.yml @@ -8,3 +8,4 @@ health_regexp: "NORMAL" vars_urls: - http://localhost:10101/debug/vars + - http://localhost:7070/debug/vars diff --git a/scripts/etc/gloat/query.row-range.yml b/scripts/etc/gloat/query.row-range.yml index f05b5cb4c..75ab2c6d6 100644 --- a/scripts/etc/gloat/query.row-range.yml +++ b/scripts/etc/gloat/query.row-range.yml @@ -8,3 +8,4 @@ health_regexp: "NORMAL" vars_urls: - http://localhost:10101/debug/vars + - http://localhost:7070/debug/vars diff --git a/scripts/etc/gloat/query.row.yml b/scripts/etc/gloat/query.row.yml index 142f12004..7f6997ea3 100644 --- a/scripts/etc/gloat/query.row.yml +++ b/scripts/etc/gloat/query.row.yml @@ -8,3 +8,4 @@ health_regexp: "NORMAL" vars_urls: - http://localhost:10101/debug/vars + - http://localhost:7070/debug/vars diff --git a/scripts/etc/gloat/query.topk.yml b/scripts/etc/gloat/query.topk.yml index 10da63427..25ef7fe50 100644 --- a/scripts/etc/gloat/query.topk.yml +++ b/scripts/etc/gloat/query.topk.yml @@ -8,3 +8,4 @@ health_regexp: "NORMAL" vars_urls: - http://localhost:10101/debug/vars + - http://localhost:7070/debug/vars diff --git a/scripts/etc/gloat/query.union.yml b/scripts/etc/gloat/query.union.yml index a2270d974..1b2777b90 100644 --- a/scripts/etc/gloat/query.union.yml +++ b/scripts/etc/gloat/query.union.yml @@ -8,3 +8,4 @@ health_regexp: "NORMAL" vars_urls: - http://localhost:10101/debug/vars + - http://localhost:7070/debug/vars diff --git a/scripts/etc/gloat/query.xor.yml b/scripts/etc/gloat/query.xor.yml index f2eb857b8..3328fb0b1 100644 --- a/scripts/etc/gloat/query.xor.yml +++ b/scripts/etc/gloat/query.xor.yml @@ -8,3 +8,4 @@ health_regexp: "NORMAL" vars_urls: - http://localhost:10101/debug/vars + - http://localhost:7070/debug/vars diff --git a/test/cluster.go b/test/cluster.go index 006729ced..3af5c9b23 100644 --- a/test/cluster.go +++ b/test/cluster.go @@ -164,6 +164,36 @@ func (c *Cluster) ImportKeyKey(t testing.TB, index, field string, valAndRecKeys } } +// TimeQuantumKey is a string key and a string+key value +type TimeQuantumKey struct { + RowKey string + ColKey string + Ts int64 +} + +// ImportTimeQuantumKey imports data into an index where the index is keyd +// and the field is a time-quantum +func (c *Cluster) ImportTimeQuantumKey(t testing.TB, index, field string, entries []TimeQuantumKey) { + t.Helper() + importRequest := &pilosa.ImportRequest{ + Index: index, + Field: field, + RowKeys: make([]string, len(entries)), + ColumnKeys: make([]string, len(entries)), + Timestamps: make([]int64, len(entries)), + } + for i, entry := range entries { + importRequest.ColumnKeys[i] = entry.ColKey + importRequest.RowKeys[i] = entry.RowKey + importRequest.Timestamps[i] = entry.Ts + + } + err := c.Nodes[0].API.Import(context.Background(), nil, importRequest) + if err != nil { + t.Fatalf("importing keykey data: %v", err) + } +} + // IntKey is a string key and a signed integer value. type IntKey struct { Val int64 diff --git a/tx_internal_test.go b/tx_internal_test.go index a5da9cccd..3ff4ddd63 100644 --- a/tx_internal_test.go +++ b/tx_internal_test.go @@ -37,20 +37,29 @@ func requireCountRangeSampleData(tb testing.TB) (*fragment, Tx) { // request that each container get its own copy of the bitmap. var bitmapSample [1025]uint64 for i := range arraySample { - arraySample[i] = uint16(i) + arraySample[i] = uint16(i * 2) } - for i := 0; i < 4096/64; i++ { - bitmapSample[i] = ^uint64(0) + // Put corresponding bits in the bitmap... + for i := 0; i < 4096/32; i++ { + // bit 0 is 0x1, bit 2 is 0x4, so even-numbered bits + // are 0x5555.... + bitmapSample[i] = 0x5555555555555555 } bm := roaring.NewSliceBitmap() for n := 0; n < 4096 && n < countRangeMaxN; n++ { c := roaring.NewContainerArray(arraySample[:n]) bm.Put(uint64(n), c) } - for n := 4096; n < countRangeMaxN; n++ { + // Start filling in the missing bits. This starts us out with + // bitmap containers, but then eventually converts to things + // that are more likely to be run containers. At the end of this, + // we should have exactly the first 8,192 bits set, for a single + // run of 8k. + for n := 4096; n < 8192; n++ { c := roaring.NewContainerBitmapN(bitmapSample[:], int32(n)) bm.Put(uint64(n), c) - bitmapSample[n/64] |= 1 << (n % 64) + w := n - 4096 + bitmapSample[w/32] |= 1 << (((n % 32) * 2) + 1) } var asBytes bytes.Buffer n, err := bm.WriteTo(&asBytes) @@ -90,11 +99,14 @@ func TestTx_CountRange(t *testing.T) { expected := uint64(0) j := uint64(0) for i := uint64(0); i < countRangeMaxN; i += 7 { + expected += i if i%4 == 3 { expected -= (j * 7) + 21 j += 7 } - got, err := tx.CountRange("i", "f", viewStandard, 0, uint64(j)<<16, uint64(i)<<16) + // Every other bit gets set, for a total of i bits in container + // i, so they're all in the first (i*2) bits of the container. + got, err := tx.CountRange("i", "f", viewStandard, 0, uint64(j)<<16, (uint64(i)<<16)+(i*2)) if err != nil { t.Fatalf("counting range: %v", err) } @@ -102,7 +114,8 @@ func TestTx_CountRange(t *testing.T) { t.Fatalf("counting from container %d to %d, expected %d, got %d", j, i, expected, got) } - expected += (i * 7) + 21 + // The -i here undoes the +i at the top of this loop. + expected += (i * 7) + 21 - i } }