"all bitmap" multi-field, single-shard ingest

This adds a shard-based import endpoint which takes bitmap data for
all field types and imports data for the whole shard transactionally.

It uses the BitmapRewriter interface to try to intelligently allow for
setting and clearing bits simultaneously without multiple writes which
is especially helpful when ingesting into int-like fields, but also
allows clear-and-then-set behavior for set fields.
This commit is contained in:
Matthew Jaffee 2022-03-31 16:53:54 -05:00 committed by Matthew Jaffee
parent 964f14a3a1
commit 772496b440
28 changed files with 2456 additions and 214 deletions

View file

@ -334,6 +334,10 @@ install-protoc-gen-go:
install-protoc:
@echo This tool cannot automatically install protoc. Please download and install protoc from https://google.github.io/proto-lens/installing-protoc.html
@echo On mac, brew install protobuf seems to work.
@echo As of the commit that added this line, protoc-gen-gofast was at 226206f39bd7, and the protoc version in use was:
@echo $$ protoc --version
@echo libprotoc 3.19.4
install-peg:
GO111MODULE=off $(GO) get github.com/pointlander/peg

106
api.go
View file

@ -1603,6 +1603,112 @@ func (api *API) ImportWithTx(ctx context.Context, qcx *Qcx, req *ImportRequest,
return errors.Wrap(err, "committing")
}
func (api *API) ImportRoaringShard(ctx context.Context, indexName string, shard uint64, req *ImportRoaringShardRequest) error {
index, err := api.Index(ctx, indexName)
if err != nil {
return errors.Wrap(err, "getting index")
}
// we really only need a Tx, but getting a Qcx so that there's only one path for getting a Tx
qcx := api.Txf().NewQcx()
qcx.write = true
tx, finisher, err := qcx.GetTx(Txo{Write: true, Index: index, Shard: shard})
if err != nil {
return errors.Wrap(err, "getting Tx")
}
defer qcx.Finish()
var err1 error
defer finisher(&err1)
if !req.Remote {
return errors.New("forwarding unimplemented on this endpoint")
}
for _, viewUpdate := range req.Views {
field := index.Field(viewUpdate.Field)
if field == nil {
err1 = errors.Errorf("no field named '%s' found.", viewUpdate.Field)
return err1
}
fieldType := field.Options().Type
if err1 = cleanupView(fieldType, &viewUpdate); err1 != nil {
return err1
}
view, err := field.createViewIfNotExists(viewUpdate.View)
if err != nil {
err1 = errors.Wrap(err, "getting view")
return err1
}
frag, err := view.CreateFragmentIfNotExists(shard)
if err != nil {
err1 = errors.Wrap(err, "getting fragment")
return err1
}
switch fieldType {
case FieldTypeSet, FieldTypeTime:
if !viewUpdate.ClearRecords {
err1 = frag.ImportRoaringClearAndSet(ctx, tx, viewUpdate.Clear, viewUpdate.Set)
} else {
err1 = frag.ImportRoaringSingleValued(ctx, tx, viewUpdate.Clear, viewUpdate.Set)
}
case FieldTypeInt, FieldTypeTimestamp, FieldTypeDecimal:
err1 = frag.ImportRoaringBSI(ctx, tx, viewUpdate.Clear, viewUpdate.Set)
case FieldTypeMutex, FieldTypeBool:
err1 = frag.ImportRoaringSingleValued(ctx, tx, viewUpdate.Clear, viewUpdate.Set)
default:
err1 = errors.Errorf("field type %s is not supported", fieldType)
}
if err1 != nil {
return err1
}
// need to update field/bsiGroup bitDepth value if this is an int-like field.
//
// TODO get rid of cached bitDepth entirely because the fact
// that we have to do this is weird and since this state isn't
// in RBF might have transactional issues.
if len(field.bsiGroups) > 0 {
maxRowID, _, err := frag.maxRow(tx, nil)
if err != nil {
err1 = errors.Wrapf(err, "getting fragment max row id")
return err1
}
var bd uint64
if maxRowID+1 > bsiOffsetBit {
bd = maxRowID + 1 - bsiOffsetBit
}
field.cacheBitDepth(bd) // updating bitDepth shouldn't harm anything even if we roll back... only might make some ops slightly more inefficient
}
}
return nil
}
func cleanupView(fieldType string, viewUpdate *RoaringUpdate) error {
// TODO wouldn't hurt to have consolidated logic somewhere for validating view names.
switch fieldType {
case FieldTypeSet, FieldTypeTime:
if viewUpdate.View == "" {
viewUpdate.View = "standard"
}
// add 'standard_' if we just have a time... this is how IDK works by default
if fieldType == FieldTypeTime && !strings.HasPrefix(viewUpdate.View, viewStandard) {
viewUpdate.View = fmt.Sprintf("%s_%s", viewStandard, viewUpdate.View)
}
case FieldTypeInt, FieldTypeDecimal, FieldTypeTimestamp:
if viewUpdate.View == "" {
viewUpdate.View = "bsig_" + viewUpdate.Field
} else if viewUpdate.View != "bsig_"+viewUpdate.Field {
return NewBadRequestError(errors.Errorf("invalid view name (%s) for field %s of type %s", viewUpdate.View, viewUpdate.Field, fieldType))
}
}
return nil
}
// ImportValue is a wrapper around the common code in ImportValueWithTx, which
// currently just translates req.Clear into a clear ImportOption.
func (api *API) ImportValue(ctx context.Context, qcx *Qcx, req *ImportValueRequest, opts ...ImportOption) error {

View file

@ -25,6 +25,7 @@ import (
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/authn"
"github.com/molecula/featurebase/v3/boltdb"
"github.com/molecula/featurebase/v3/roaring"
"github.com/molecula/featurebase/v3/server"
"github.com/molecula/featurebase/v3/shardwidth"
"github.com/molecula/featurebase/v3/test"
@ -541,17 +542,13 @@ func TestAPI_Ingest(t *testing.T) {
)},
)
defer c.Close()
coord := c.GetPrimary()
// m0 := c.GetNode(0)
// m1 := c.GetNode(1)
// m2 := c.GetNode(2)
index := "ingest"
setField := "set"
timeField := "tq"
intField := "int"
_, err := coord.API.CreateIndex(ctx, index, pilosa.IndexOptions{Keys: false})
_, err := coord.API.CreateIndex(ctx, index, pilosa.IndexOptions{Keys: false, TrackExistence: true})
if err != nil {
t.Fatalf("creating index: %v", err)
}
@ -563,69 +560,167 @@ func TestAPI_Ingest(t *testing.T) {
if err != nil {
t.Fatalf("creating field: %v", err)
}
sampleJson := []byte(`
[
{
"action": "set",
"records": {
"2": {
"set": [2],
"tq": { "time": "2006-01-02T15:04:05.999999999Z", "values": [6] }
},
"5": { "set": [3] },
"8": { "set": [3] },
"1": {
"set": [2],
"tq": { "time": "2006-01-02T15:04:05.999999999Z", "values": [3, 4] }
},
"4": { "set": [3, 7] }
}
},
{
"action": "clear",
"record_ids": [ 5, 6, 7 ],
"fields": [ "tq", "set" ]
},
{
"action": "write",
"records": {
"8": { "tq": { "time": "2006-01-02T15:04:05.999999999Z", "values": [3, 4] } },
"9": { "set": [7, 3] }
}
},
{
"action": "delete",
"record_ids": [ 9 ]
}
]
`)
// just for set row 3:
// first operation should set it for 4, 5, and 8.
// clear operation should clear it for 5, 6, and 7, leaving it still set for 4 and 8.
// the write operation should clear set for record 8, even though record 8 doesn't
// contain that field in that op, because set is present in record 9, which also
// gets row 3 set. but then we delete 9.
// so after all that we expect Row(set=3) to be 4...
sampleBuf := bytes.NewBuffer(sampleJson)
qcx := coord.API.Txf().NewQcx()
defer func() {
if err := qcx.Finish(); err != nil {
t.Fatalf("finishing qcx: %v", err)
_, err = coord.API.CreateField(ctx, index, intField, pilosa.OptFieldTypeInt(0, 100000))
if err != nil {
t.Fatalf("creating field: %v", err)
}
t.Run("IngestAPI", func(t *testing.T) {
sampleJson := []byte(`
[
{
"action": "set",
"records": {
"2": {
"set": [2],
"tq": { "time": "2006-01-02T15:04:05.999999999Z", "values": [6] }
},
"5": { "set": [3] },
"8": { "set": [3] },
"1": {
"set": [2],
"tq": { "time": "2006-01-02T15:04:05.999999999Z", "values": [3, 4] }
},
"4": { "set": [3, 7] }
}
},
{
"action": "clear",
"record_ids": [ 5, 6, 7 ],
"fields": [ "tq", "set" ]
},
{
"action": "write",
"records": {
"8": { "tq": { "time": "2006-01-02T15:04:05.999999999Z", "values": [3, 4] } },
"9": { "set": [7, 3] }
}
},
{
"action": "delete",
"record_ids": [ 9 ]
}
]
`)
// just for set row 3:
// first operation should set it for 4, 5, and 8.
// clear operation should clear it for 5, 6, and 7, leaving it still set for 4 and 8.
// the write operation should clear set for record 8, even though record 8 doesn't
// contain that field in that op, because set is present in record 9, which also
// gets row 3 set. but then we delete 9.
// so after all that we expect Row(set=3) to be 4...
sampleBuf := bytes.NewBuffer(sampleJson)
qcx := coord.API.Txf().NewQcx()
defer func() {
if err := qcx.Finish(); err != nil {
t.Fatalf("finishing qcx: %v", err)
}
}()
err = coord.API.IngestOperations(ctx, qcx, index, sampleBuf)
if err != nil {
t.Fatalf("importing data: %v", err)
}
}()
err = coord.API.IngestOperations(ctx, qcx, index, sampleBuf)
if err != nil {
t.Fatalf("importing data: %v", err)
}
query := "Row(set=3)"
res, err := coord.API.Query(context.Background(), &pilosa.QueryRequest{Index: index, Query: query})
if err != nil {
t.Errorf("query: %v", err)
}
r := res.Results[0].(*pilosa.Row).Columns()
if len(r) != 1 || r[0] != 4 {
t.Fatalf("expected row with 4 set, got %d", r)
}
query := "Row(set=3)"
res, err := coord.API.Query(context.Background(), &pilosa.QueryRequest{Index: index, Query: query})
if err != nil {
t.Errorf("query: %v", err)
}
r := res.Results[0].(*pilosa.Row).Columns()
if len(r) != 1 || r[0] != 4 {
t.Fatalf("expected row with 4 set, got %d", r)
}
})
t.Run("ImportRoaringShard", func(t *testing.T) {
setBuf := &bytes.Buffer{}
setBits := roaring.NewBitmap(7, pilosa.ShardWidth+7)
_, _ = setBits.WriteTo(setBuf) // bytes.Buffer never errors
intBuf := &bytes.Buffer{}
intBits := roaring.NewBitmap(7, pilosa.ShardWidth*2+7)
_, _ = intBits.WriteTo(intBuf) // bytes.Buffer never errors
request := &pilosa.ImportRoaringShardRequest{
Remote: true,
Views: []pilosa.RoaringUpdate{
{
Field: setField,
View: "standard",
Set: setBuf.Bytes(),
},
{
Field: intField,
View: "bsig_" + intField,
Set: intBuf.Bytes(),
},
},
}
if err := coord.API.ImportRoaringShard(context.Background(), "ingest", 8, request); err != nil {
t.Fatalf("ingesting: %v", err)
}
mustQuery := func(t *testing.T, index, query string) pilosa.QueryResponse {
res, err := coord.API.Query(context.Background(), &pilosa.QueryRequest{Index: index, Query: query})
if err != nil {
t.Fatalf("querying: %v", err)
}
return res
}
res := mustQuery(t, "ingest", "Row(set=0)")
r := res.Results[0].(*pilosa.Row).Columns()
if len(r) != 1 || r[0] != pilosa.ShardWidth*8+7 {
t.Fatalf("expected row with pilosa.ShardWidth*8+7 set, got %d", r)
}
res = mustQuery(t, "ingest", "Row(set=1)")
r = res.Results[0].(*pilosa.Row).Columns()
if len(r) != 1 || r[0] != pilosa.ShardWidth*8+7 {
t.Fatalf("expected row with pilosa.ShardWidth*8+7 set, got %d", r)
}
res = mustQuery(t, "ingest", "Row(int==1)")
r = res.Results[0].(*pilosa.Row).Columns()
if len(r) != 1 || r[0] != pilosa.ShardWidth*8+7 {
t.Fatalf("expected row with, pilosa.ShardWidth*8+7 set, got %d", r)
}
request = &pilosa.ImportRoaringShardRequest{
Remote: true,
Views: []pilosa.RoaringUpdate{
{
Field: setField,
View: "standard",
Clear: setBuf.Bytes(),
},
{
Field: intField,
View: "bsig_" + intField,
Clear: intBuf.Bytes(),
},
},
}
if err := coord.API.ImportRoaringShard(context.Background(), "ingest", 8, request); err != nil {
t.Fatalf("ingesting: %v", err)
}
res = mustQuery(t, "ingest", "Row(set=0)")
r = res.Results[0].(*pilosa.Row).Columns()
if len(r) != 0 {
t.Fatalf("expected no values after clearing, got: %v", r)
}
res = mustQuery(t, "ingest", "Row(set=1)")
r = res.Results[0].(*pilosa.Row).Columns()
if len(r) != 0 {
t.Fatalf("expected no values after clearing, got: %v", r)
}
res = mustQuery(t, "ingest", "Row(int==1)")
r = res.Results[0].(*pilosa.Row).Columns()
if len(r) != 0 {
t.Fatalf("expected no values after clearing, got: %v", r)
}
})
}
// ingestBenchmarkHelper makes it easier to exclude this from benchmark computations

View file

@ -2,10 +2,13 @@
package client
import (
"bytes"
"math/bits"
"sort"
"sync"
"time"
featurebase "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/client/egpool"
"github.com/molecula/featurebase/v3/logger"
"github.com/molecula/featurebase/v3/roaring"
@ -164,6 +167,8 @@ type Batch struct {
splitBatchMode bool
frags fragments
clearFrags fragments
useShardTransactionalEndpoint bool
}
func (b *Batch) Len() int { return len(b.ids) }
@ -206,6 +211,13 @@ func OptKeyTranslateBatchSize(v int) BatchOption {
}
}
func OptUseShardTransactionalEndpoint(use bool) BatchOption {
return func(b *Batch) error {
b.useShardTransactionalEndpoint = use
return nil
}
}
// NewBatch initializes a new Batch object which will use the given
// Pilosa client, index, set of fields, and will take "size" records
// before returning ErrBatchNowFull. The positions of the Fields in
@ -687,6 +699,14 @@ func (b *Batch) Import() error {
if err != nil {
return errors.Wrap(err, "making fragments (flush)")
}
if b.useShardTransactionalEndpoint {
// TODO handle bool?
frags, clearFrags, err = b.makeSingleValFragments(frags, clearFrags)
if err != nil {
return errors.Wrap(err, "making single val fragments")
}
}
makeTime := time.Now()
b.log.Printf("making fragments for batch of %d took %v", size, makeTime.Sub(transTime))
@ -698,11 +718,18 @@ func (b *Batch) Import() error {
b.clearFrags = make(fragments)
// create bitmaps out of each field in b.rowIDs and import. Also
// import int data.
err = b.doImport(frags, clearFrags)
if err != nil {
return errors.Wrap(err, "doing import")
if !b.useShardTransactionalEndpoint {
err = b.doImport(frags, clearFrags)
if err != nil {
return errors.Wrap(err, "doing import")
}
b.log.Printf("importing fragments took %v", time.Since(makeTime))
} else {
err = b.doImportShardTransactional(frags, clearFrags)
if err != nil {
return errors.Wrap(err, "doing shard transactional import")
}
}
b.log.Printf("importing fragments took %v", time.Since(makeTime))
}
b.reset()
@ -749,7 +776,7 @@ func (b *Batch) doTranslation() error {
// Translate the column keys.
eg.Go(func() error {
// Dedupliucate keys to translate.
// Deduplicate keys to translate.
dedup := make(map[string]struct{})
var keys []string
for _, key := range b.toTranslateID {
@ -1026,6 +1053,78 @@ func (b *Batch) createFieldKeys(field *Field, keys ...string) (map[string]uint64
return results, nil
}
func (b *Batch) doImportShardTransactional(frags, clearFrags fragments) error {
start := time.Now()
requests := make(map[uint64]*featurebase.ImportRoaringShardRequest)
getOrCreate := func(requests map[uint64]*featurebase.ImportRoaringShardRequest, shard uint64) *featurebase.ImportRoaringShardRequest {
request, ok := requests[shard]
if !ok {
request = &featurebase.ImportRoaringShardRequest{
Remote: true, // the client will send to all replicas TODO probably rename before merge
Views: make([]featurebase.RoaringUpdate, 0, 1),
}
requests[shard] = request
}
return request
}
for fragKey, viewMap := range frags {
request := getOrCreate(requests, fragKey.shard)
for view, bitmap := range viewMap {
buf := &bytes.Buffer{}
_, err := bitmap.WriteTo(buf)
if err != nil {
return errors.Wrap(err, "serializing bitmap")
}
request.Views = append(request.Views, featurebase.RoaringUpdate{Field: fragKey.field, View: view, Set: buf.Bytes()})
// handle clear bitmap now if it exists so we don't have to go searching later
if clearVM := clearFrags.GetViewMap(fragKey.shard, fragKey.field); clearVM != nil {
if clearBitmap, ok := clearVM[view]; ok {
clearBuf := &bytes.Buffer{}
_, err := clearBitmap.WriteTo(clearBuf)
if err != nil {
return errors.Wrap(err, "serializing clear bitmap")
}
request.Views[len(request.Views)-1].Clear = clearBuf.Bytes()
// delete from clearFrags so any remaining we know for sure must be added new
clearFrags.DeleteView(fragKey.shard, fragKey.field, view)
}
}
}
}
for fragKey, viewMap := range clearFrags {
request := getOrCreate(requests, fragKey.shard)
for view, bitmap := range viewMap {
buf := &bytes.Buffer{}
_, err := bitmap.WriteTo(buf)
if err != nil {
return errors.Wrap(err, "serializing bitmap")
}
request.Views = append(request.Views, featurebase.RoaringUpdate{Field: fragKey.field, View: view, Clear: buf.Bytes()})
}
}
b.client.Stats.Timing(MetricBatchShardImportBuildRequestsSeconds, time.Since(start), 1.0)
start = time.Now()
eg := egpool.Group{PoolSize: 20}
for shard, request := range requests {
shard := shard
request := request
eg.Go(func() error {
return b.client.ImportRoaringShard(b.index.Name(), shard, request)
})
}
err := eg.Wait()
dur := time.Since(start)
b.client.Stats.Timing(MetricBatchShardImportDurationSeconds, dur, 1.0)
b.log.Printf("import shard took: %v\n", dur)
return errors.Wrap(err, "doing shard-transactional imports")
}
func (b *Batch) doImport(frags, clearFrags fragments) error {
start := time.Now()
@ -1088,6 +1187,14 @@ func anyCause(cause error, errs ...error) error {
return nil
}
func (b *Batch) shardWidth() uint64 {
shardWidth := b.index.ShardWidth()
if shardWidth == 0 {
shardWidth = DefaultShardWidth
}
return shardWidth
}
// this is kind of bad as it means we can never import column id
// ^uint64(0) which is a valid column ID. I think it's unlikely to
// matter much in practice (we could maybe special case it somewhere
@ -1095,10 +1202,7 @@ func anyCause(cause error, errs ...error) error {
var nilSentinel = ^uint64(0)
func (b *Batch) makeFragments(frags, clearFrags fragments) (fragments, fragments, error) {
shardWidth := b.index.ShardWidth()
if shardWidth == 0 {
shardWidth = DefaultShardWidth
}
shardWidth := b.shardWidth()
emptyClearRows := make(map[int]uint64)
// create _exists fragments if needed
@ -1219,6 +1323,133 @@ func (b *Batch) makeFragments(frags, clearFrags fragments) (fragments, fragments
return frags, clearFrags, nil
}
func (b *Batch) makeSingleValFragments(frags, clearFrags fragments) (fragments, fragments, error) {
shardWidth := b.shardWidth()
ids := make([]uint64, len(b.ids))
// -------------------------
// int-like fields
// -------------------------
for fieldName, bvalues := range b.values {
ids = ids[:len(b.ids)]
// trim out null values from ids and values.
nullIndices := b.nullIndices[fieldName]
i, n := uint64(0), 0
for _, nullIndex := range nullIndices {
copy(ids[n:], b.ids[i:nullIndex])
n += copy(bvalues[n:], bvalues[i:nullIndex])
i = nullIndex + 1
}
copy(ids[n:], b.ids[i:])
n += copy(bvalues[n:], bvalues[i:])
ids, bvalues = ids[:n], bvalues[:n]
if len(ids) == 0 {
continue
}
sc := &valsByIDsSortable{ids: ids, vals: bvalues, width: shardWidth}
if !sort.IsSorted(sc) {
sort.Stable(sc)
}
field := b.headerMap[fieldName]
base := field.Options().base
shard := ids[0] / shardWidth
bitmap := frags.GetOrCreate(shard, fieldName, "bsig_"+fieldName) // TODO... grab bsig_ prefix from elsewhere
for i, id := range ids {
if i+1 < len(ids) {
// we only want the last value set for each id
if ids[i+1] == id {
continue
}
}
if shard != id/shardWidth {
shard = id / shardWidth
bitmap = frags.GetOrCreate(shard, fieldName, "bsig_"+fieldName)
}
fragmentColumn := id % shardWidth
bitmap.Add(fragmentColumn) // existence bit
svalue := bvalues[i] - base
negative := svalue < 0
var value uint64
if negative {
bitmap.Add(shardWidth + fragmentColumn) // set sign bit
value = uint64(svalue * -1)
} else {
value = uint64(svalue)
}
lz := bits.LeadingZeros64(value)
row := uint64(2)
for mask := uint64(0x1); mask <= 1<<(64-lz) && mask != 0; mask = mask << 1 {
if value&mask > 0 {
bitmap.Add(row*shardWidth + fragmentColumn)
}
row++
}
}
}
// -------------------------
// mutex fields
// -------------------------
for findex, rowIDs := range b.rowIDs {
field := b.header[findex]
if field.Opts().Type() != FieldTypeMutex {
continue
}
ids = ids[:0]
// get slice of column ids for non-nil rowIDs and cut nil row
// IDs out of rowIDs.
idsIndex := 0
for i, id := range b.ids {
rowID := rowIDs[i]
if rowID == nilSentinel {
continue
}
rowIDs[idsIndex] = rowID
ids = append(ids, id)
idsIndex++
}
rowIDs = rowIDs[:idsIndex]
if len(ids) == 0 {
continue
}
sc := &rowsByIDsSortable{ids: ids, rows: rowIDs, width: shardWidth}
if !sort.IsSorted(sc) {
sort.Stable(sc)
}
shard := ids[0] / shardWidth
bitmap := frags.GetOrCreate(shard, field.Name(), "standard")
clearBM := clearFrags.GetOrCreate(shard, field.Name(), "standard")
for i, id := range ids {
if i+1 < len(ids) {
// we only want the last value set for each id
if ids[i+1] == id {
continue
}
}
row := rowIDs[i]
if shard != id/shardWidth {
shard = id / shardWidth
bitmap = frags.GetOrCreate(shard, field.Name(), "standard")
}
fragmentColumn := id % shardWidth
clearBM.Add(fragmentColumn) // Will use this to clear columns.
bitmap.Add(row*shardWidth + fragmentColumn)
}
}
return frags, clearFrags, nil
}
type valsByIDsSortable struct {
ids []uint64
vals []int64
@ -1268,7 +1499,7 @@ func (b *Batch) importValueData() error {
sc := &valsByIDsSortable{ids: ids, vals: bvalues, width: shardWidth}
if !sort.IsSorted(sc) {
sort.Sort(sc)
sort.Stable(sc) // TODO(jaffee) this was sort.Sort which I think is a bug. If we get multiple of the same record w/in a batch with different int values, the last one needs to win. We need a test for this.
}
curShard := ids[0] / shardWidth
@ -1320,7 +1551,7 @@ type rowsByIDsSortable struct {
func (v *rowsByIDsSortable) Len() int { return len(v.ids) }
// comparing on shard rather than ID was twice as fast in informal tests
func (v *rowsByIDsSortable) Less(i, j int) bool { return v.ids[i]/v.width < v.ids[j]/v.width }
func (v *rowsByIDsSortable) Less(i, j int) bool { return v.ids[i] < v.ids[j] }
func (v *rowsByIDsSortable) Swap(i, j int) {
v.ids[i], v.ids[j] = v.ids[j], v.ids[i]
v.rows[i], v.rows[j] = v.rows[j], v.rows[i]
@ -1363,7 +1594,7 @@ func (b *Batch) importMutexData() error {
sc := &rowsByIDsSortable{ids: ids, rows: rowIDs, width: shardWidth}
if !sort.IsSorted(sc) {
sort.Sort(sc)
sort.Stable(sc)
}
curShard := ids[0] / shardWidth
@ -1507,3 +1738,11 @@ func (f fragments) GetViewMap(shard uint64, field string) map[string]*roaring.Bi
}
return viewMap
}
func (f fragments) DeleteView(shard uint64, field, view string) {
vm := f.GetViewMap(shard, field)
if vm == nil {
return
}
delete(vm, view)
}

View file

@ -38,6 +38,7 @@ func TestAgainstCluster(t *testing.T) {
t.Run("test-batches", func(t *testing.T) { testBatches(t, c, client) })
t.Run("batches-strings-ids", func(t *testing.T) { testBatchesStringIDs(t, c, client) })
t.Run("test-batch-staleness", func(t *testing.T) { testBatchStaleness(t, c, client) })
t.Run("test-import-batch-multiple-ints", func(t *testing.T) { testImportBatchMultipleInts(t, c, client) })
}
func testStringSliceCombos(t *testing.T, c *test.Cluster, client *Client) {
@ -1348,3 +1349,41 @@ func testBatchStaleness(t *testing.T, c *test.Cluster, client *Client) {
t.Fatal("batch expected to be stale")
}
}
func testImportBatchMultipleInts(t *testing.T, c *test.Cluster, client *Client) {
schema := NewSchema()
idx := schema.Index("test-import-batch-multi-int")
field := idx.Field("anint", OptFieldTypeInt())
err := client.SyncSchema(schema)
if err != nil {
t.Fatalf("syncing schema: %v", err)
}
b, err := NewBatch(client, 6, idx, []*Field{field}, OptUseShardTransactionalEndpoint(true))
if err != nil {
t.Fatalf("getting batch: %v", err)
}
r := Row{Values: make([]interface{}, 1)}
vals := []int64{16, 8, 32, 1, 2, 4}
for i := uint64(0); i < 6; i++ {
r.ID = uint64(1)
r.Values[0] = vals[i]
err := b.Add(r)
if err != nil && err != ErrBatchNowFull {
t.Fatalf("adding to batch: %v", err)
}
}
err = b.Import()
if err != nil {
t.Fatalf("importing: %v", err)
}
if resp, err := client.Query(field.Equals(4)); err != nil {
t.Fatalf("querying: %v", err)
} else if res := resp.Results()[0].Row().Columns; len(res) != 1 || res[0] != 1 {
t.Fatalf("unepxected result: %v", res)
}
}

View file

@ -21,6 +21,7 @@ import (
"github.com/golang/protobuf/proto" //nolint:staticcheck
pilosa "github.com/molecula/featurebase/v3"
fbproto "github.com/molecula/featurebase/v3/encoding/proto" // TODO use this everywhere and get rid of proto import
"github.com/molecula/featurebase/v3/logger"
pnet "github.com/molecula/featurebase/v3/net"
"github.com/molecula/featurebase/v3/pb"
@ -646,6 +647,27 @@ func (c *Client) importData(uri *pnet.URI, path string, data []byte) error {
return nil
}
func (c *Client) ImportRoaringShard(index string, shard uint64, request *pilosa.ImportRoaringShardRequest) error {
uris, err := c.getURIsForShard(index, shard)
if err != nil {
return errors.Wrap(err, "getting URIs for import")
}
data, err := fbproto.DefaultSerializer.Marshal(request)
if err != nil {
return errors.Wrap(err, "marshaling")
}
eg := errgroup.Group{}
for _, uri := range uris {
uri := uri
eg.Go(func() error {
return c.importData(uri, fmt.Sprintf("/index/%s/shard/%d/import-roaring", index, shard), data)
})
}
err = eg.Wait()
return errors.Wrap(err, "importing")
}
// ImportRoaringBitmap can import pre-made bitmaps for a number of
// different views into the given field/shard. If the view name in the
// map is an empty string, the standard view will be used.

View file

@ -2,13 +2,16 @@
package client
import (
"bytes"
"fmt"
"io/ioutil"
"testing"
"time"
featurebase "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/disco"
pnet "github.com/molecula/featurebase/v3/net"
"github.com/molecula/featurebase/v3/roaring"
"github.com/molecula/featurebase/v3/shardwidth"
"github.com/molecula/featurebase/v3/test"
"github.com/stretchr/testify/require"
@ -23,6 +26,8 @@ var (
testIndexKeyTranslation *Index
testField *Field
testFieldTimestamp *Field
testFieldInt *Field
testFieldTimeQuantum *Field
testFieldInt0 *Field
testFieldInt1 *Field
@ -40,6 +45,8 @@ func setup(t *testing.T, cli *Client) {
)
testField = testIndex.Field("test-field")
testFieldTimeQuantum = testIndex.Field("test-field-timequantum", OptFieldTypeTime(TimeQuantumYear))
testFieldTimestamp = testIndex.Field("test-field-timestamp", OptFieldTypeTimestamp(time.Date(1970, time.January, 1, 0, 0, 0, 0, time.UTC), "s"))
testFieldInt = testIndex.Field("test-field-int", OptFieldTypeInt(0, 100000))
testIndexKeyTranslation = testSchema.Index("test-index-key-translation", OptIndexKeys(true))
testIndexAtomicRecord = testSchema.Index("test-index-atomic-record")
@ -433,20 +440,20 @@ func TestClientAgainstCluster(t *testing.T) {
require.NoError(t, err)
_, err = cli.Query(testIndex.RawQuery(`
Set(0, test-field-group-by-int=1)
Set(1, test-field-group-by-int=2)
Set(0, test-field-group-by-int=1)
Set(1, test-field-group-by-int=2)
Set(2, test-field-group-by-int=-2)
Set(3, test-field-group-by-int=-1)
Set(2, test-field-group-by-int=-2)
Set(3, test-field-group-by-int=-1)
Set(4, test-field-group-by-int=4)
Set(4, test-field-group-by-int=4)
Set(10, test-field-group-by-int=0)
Set(100, test-field-group-by-int=0)
Set(1000, test-field-group-by-int=0)
Set(10000, test-field-group-by-int=0)
Set(100000, test-field-group-by-int=0)
`))
Set(10, test-field-group-by-int=0)
Set(100, test-field-group-by-int=0)
Set(1000, test-field-group-by-int=0)
Set(10000, test-field-group-by-int=0)
Set(100000, test-field-group-by-int=0)
`))
require.NoError(t, err, "Set(0..100000)")
resp, err := cli.Query(testIndex.GroupBy(testFieldGroupBy.Rows()))
@ -738,6 +745,118 @@ func TestClientAgainstCluster(t *testing.T) {
require.Equalf(t, time.Minute, trns.Timeout, "TranslateColumnKeys Timeout")
require.Truef(t, trns.Active, "TranslateColumnKeys Active")
})
t.Run("ImportRoaringShard", func(t *testing.T) {
setup(t, cli)
shardWidth := uint64(1 << shardwidth.Exponent)
bitmap := roaring.NewBitmap(1, shardWidth*2+1, shardWidth*3+1)
buf := &bytes.Buffer{}
_, err := bitmap.WriteTo(buf)
if err != nil {
t.Fatalf("serializing bitmap: %v", err)
}
request := &featurebase.ImportRoaringShardRequest{
Remote: true,
Views: []featurebase.RoaringUpdate{
{
Field: "test-field",
View: "standard",
Set: buf.Bytes(),
},
{
Field: "test-field-timestamp",
View: "bsig_test-field-timestamp",
Set: buf.Bytes(),
},
{
Field: "test-field-int",
View: "bsig_test-field-int",
Set: buf.Bytes(),
},
},
}
err = cli.ImportRoaringShard("test-index", 3, request)
if err != nil {
t.Fatalf("import-roaring-shard: %v", err)
}
if resp, err := cli.Query(testField.Row(2)); err != nil {
t.Fatalf("querying: %v", err)
} else if res := resp.ResultList[0].Row().Columns; len(res) != 1 || res[0] != shardWidth*3+1 {
t.Fatalf("unexpected result: %v", res)
}
if resp, err := cli.Query(testFieldInt.NotNull()); err != nil {
t.Fatalf("querying: %v", err)
} else if res := resp.ResultList[0].Row().Columns; len(res) != 1 || res[0] != shardWidth*3+1 {
t.Fatalf("unexpected result: %v", res)
}
if resp, err := cli.Query(testFieldTimestamp.NotNull()); err != nil {
t.Fatalf("querying: %v", err)
} else if res := resp.ResultList[0].Row().Columns; len(res) != 1 || res[0] != shardWidth*3+1 {
t.Fatalf("unexpected result: %v", res)
}
if resp, err := cli.Query(testIndex.RawQuery("Row(test-field-timestamp>'1969-12-31T23:59:59Z')")); err != nil {
t.Fatalf("querying: %v", err)
} else if res := resp.ResultList[0].Row().Columns; len(res) != 1 || res[0] != shardWidth*3+1 {
t.Fatalf("unexpected result: %v", res)
}
// now write more data
bitmap = roaring.NewBitmap(1, 2, shardWidth*3+1, shardWidth*3+2)
buf = &bytes.Buffer{}
_, err = bitmap.WriteTo(buf)
if err != nil {
t.Fatalf("serializing bitmap: %v", err)
}
request = &featurebase.ImportRoaringShardRequest{
Remote: true,
Views: []featurebase.RoaringUpdate{
{
Field: "test-field",
View: "standard",
Set: buf.Bytes(),
},
{
Field: "test-field-timestamp",
View: "bsig_test-field-timestamp",
Set: buf.Bytes(),
},
{
Field: "test-field-int",
View: "bsig_test-field-int",
Set: buf.Bytes(),
},
},
}
err = cli.ImportRoaringShard("test-index", 3, request)
if err != nil {
t.Fatalf("import-roaring-shard: %v", err)
}
if resp, err := cli.Query(testField.Row(3)); err != nil {
t.Errorf("querying: %v", err)
} else if res := resp.ResultList[0].Row().Columns; len(res) != 2 || res[0] != shardWidth*3+1 || res[1] != shardWidth*3+2 {
t.Errorf("unexpected result: %v", res)
}
if resp, err := cli.Query(testFieldInt.NotNull()); err != nil {
t.Errorf("querying: %v", err)
} else if res := resp.ResultList[0].Row().Columns; len(res) != 2 || res[0] != shardWidth*3+1 || res[1] != shardWidth*3+2 {
t.Errorf("unexpected result: %v", res)
}
if resp, err := cli.Query(testFieldTimestamp.NotNull()); err != nil {
t.Errorf("querying: %v", err)
} else if res := resp.ResultList[0].Row().Columns; len(res) != 2 || res[0] != shardWidth*3+1 || res[1] != shardWidth*3+2 {
t.Errorf("unexpected result: %v", res)
}
if resp, err := cli.Query(testIndex.RawQuery("Row(test-field-timestamp>'1969-12-31T23:59:59Z')")); err != nil {
t.Errorf("querying: %v", err)
} else if res := resp.ResultList[0].Row().Columns; len(res) != 2 || res[0] != shardWidth*3+1 || res[1] != shardWidth*3+2 {
t.Errorf("unexpected result: %v", res)
}
})
})
}
}

View file

@ -13,4 +13,15 @@ const (
// starting and finishing a transaction, importing all data, and
// resetting internal structures.
MetricBatchFlushDurationSeconds = "batch_flush_duration_seconds"
// MetricBatchShardImportBuildRequestsSeconds is the time it takes
// after making fragments to build the shard-transactional request
// objects (but not actually import them or do any network activity).
MetricBatchShardImportBuildRequestsSeconds = "batch_shard_import_build_requests_seconds"
// MetricBatchShardImportDurationSeconds is the time it takes to
// import all data for all shards in the batch using the
// shard-transactional endpoint. This does not include the time it
// takes to build the requests locally.
MetricBatchShardImportDurationSeconds = "batch_shard_import_duration_seconds"
)

View file

@ -24,6 +24,7 @@ Provides a set of commands for inspecting RBF data files.
cmd.AddCommand(newRBFDumpCommand(stdin, stdout, stderr))
cmd.AddCommand(newRBFPagesCommand(stdin, stdout, stderr))
cmd.AddCommand(newRBFPageCommand(stdin, stdout, stderr))
cmd.AddCommand(newRBFVizCommand(stdin, stdout, stderr))
return cmd
}
@ -145,3 +146,27 @@ Prints the header & cell data for one or more pages.
}
return cmd
}
func newRBFVizCommand(stdin io.Reader, stdout, stderr io.Writer) *cobra.Command {
c := ctl.NewRBFVizCommand(stdin, stdout, stderr)
cmd := &cobra.Command{
Use: "viz [flags] PATH",
Short: "Show visualization of RBF data. Experimental.",
Long: `
Show visualization of RBF data. Experimental, do not depend on specifics of the output or the flags of this command.
`,
Args: func(cmd *cobra.Command, args []string) error {
if len(args) == 0 {
return fmt.Errorf("data directory path required")
} else if len(args) > 1 {
return fmt.Errorf("too many command line arguments")
}
c.Path = args[0]
return nil
},
RunE: func(cmd *cobra.Command, args []string) error {
return c.Run(context.Background())
},
}
return cmd
}

52
ctl/rbf_viz.go Normal file
View file

@ -0,0 +1,52 @@
// Copyright 2021 Molecula Corp. All rights reserved.
package ctl
import (
"context"
"fmt"
"io"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/rbf"
)
// RBFVizCommand represents a command for doing a visualisation of the RBF tree.
type RBFVizCommand struct {
// Filepath to the RBF database.
Path string
// Standard input/output
*pilosa.CmdIO
}
// NewRBFVizCommand returns a new instance of RBFVizCommand.
func NewRBFVizCommand(stdin io.Reader, stdout, stderr io.Writer) *RBFVizCommand {
return &RBFVizCommand{
CmdIO: pilosa.NewCmdIO(stdin, stdout, stderr),
}
}
// Run executes a consistency viz of an RBF database.
func (cmd *RBFVizCommand) Run(ctx context.Context) error {
// Open database.
db := rbf.NewDB(cmd.Path, nil)
if err := db.Open(); err != nil {
return err
}
defer db.Close()
// Run viz on the database.
if err := db.Viz(cmd.Stdout); err != nil {
switch err := err.(type) {
case rbf.ErrorList:
for i := range err {
fmt.Fprintln(cmd.Stdout, err[i])
}
default:
fmt.Fprintln(cmd.Stdout, err)
}
return fmt.Errorf("viz failed")
}
return nil
}

View file

@ -222,6 +222,14 @@ func (s Serializer) Unmarshal(buf []byte, m pilosa.Message) error {
}
s.decodeImportRoaringRequest(msg, mt)
return nil
case *pilosa.ImportRoaringShardRequest:
msg := &pb.ImportRoaringShardRequest{}
err := proto.Unmarshal(buf, msg)
if err != nil {
return errors.Wrap(err, "unmarshaling ImportRoaringShardRequest")
}
s.decodeImportRoaringShardRequest(msg, mt)
return nil
case *pilosa.ImportResponse:
msg := &pb.ImportResponse{}
err := proto.Unmarshal(buf, msg)
@ -385,6 +393,8 @@ func (s Serializer) encodeToProto(m pilosa.Message) proto.Message {
return s.encodeImportValueRequest(mt)
case *pilosa.ImportRoaringRequest:
return s.encodeImportRoaringRequest(mt)
case *pilosa.ImportRoaringShardRequest:
return s.encodeImportRoaringShardRequest(mt)
case *pilosa.ImportResponse:
return s.encodeImportResponse(mt)
case *pilosa.BlockDataRequest:
@ -488,6 +498,27 @@ func (s Serializer) encodeImportRoaringRequest(m *pilosa.ImportRoaringRequest) *
}
}
func (s Serializer) encodeImportRoaringShardRequest(m *pilosa.ImportRoaringShardRequest) *pb.ImportRoaringShardRequest {
views := make([]*pb.RoaringUpdate, len(m.Views))
for i, view := range m.Views {
views[i] = s.encodeRoaringUpdate(view)
}
return &pb.ImportRoaringShardRequest{
Remote: m.Remote,
Views: views,
}
}
func (s Serializer) encodeRoaringUpdate(m pilosa.RoaringUpdate) *pb.RoaringUpdate {
return &pb.RoaringUpdate{
Field: m.Field,
View: m.View,
Clear: m.Clear,
Set: m.Set,
ClearRecords: m.ClearRecords,
}
}
func (s Serializer) encodeQueryRequest(m *pilosa.QueryRequest) *pb.QueryRequest {
r := &pb.QueryRequest{
Query: m.Query,
@ -1321,6 +1352,23 @@ func (s Serializer) decodeImportRoaringRequest(pb *pb.ImportRoaringRequest, m *p
m.UpdateExistence = pb.UpdateExistence
}
func (s Serializer) decodeImportRoaringShardRequest(pb *pb.ImportRoaringShardRequest, m *pilosa.ImportRoaringShardRequest) {
m.Remote = pb.Remote
for _, viewUpdate := range pb.Views {
pru := &pilosa.RoaringUpdate{}
s.decodeRoaringUpdate(viewUpdate, pru)
m.Views = append(m.Views, *pru)
}
}
func (s Serializer) decodeRoaringUpdate(pb *pb.RoaringUpdate, m *pilosa.RoaringUpdate) {
m.Field = pb.Field
m.View = pb.View
m.Clear = pb.Clear
m.Set = pb.Set
m.ClearRecords = pb.ClearRecords
}
func (s Serializer) decodeImportResponse(pb *pb.ImportResponse, m *pilosa.ImportResponse) {
m.Err = pb.Err
}

View file

@ -2384,6 +2384,82 @@ func (f *fragment) importRoaring(ctx context.Context, tx Tx, data []byte, clear
return nil
}
// ImportRoaringClearAndSet simply clears the bits in clear and sets the bits in set.
func (f *fragment) ImportRoaringClearAndSet(ctx context.Context, tx Tx, clear, set []byte) error {
clearIter, err := roaring.NewContainerIterator(clear)
if err != nil {
return errors.Wrap(err, "getting clear iterator")
}
setIter, err := roaring.NewContainerIterator(set)
if err != nil {
return errors.Wrap(err, "getting set iterator")
}
rewriter, err := roaring.NewClearAndSetRewriter(clearIter, setIter)
if err != nil {
return errors.Wrap(err, "getting rewriter")
}
err = tx.ApplyRewriter(f.index(), f.field(), f.view(), f.shard, 0, rewriter)
return errors.Wrap(err, "applying rewriter")
}
// ImportRoaringBSI interprets "clear" as a single row specifying
// records to be cleared, and "set" as specifying the values to be set
// which implies clearing any other values in those columns.
func (f *fragment) ImportRoaringBSI(ctx context.Context, tx Tx, clear, set []byte) error {
// In this first block, we take the first row of clear as records
// we want to unconditionally clear, and the first row of set as
// records we also want to clear because they're going to get set
// and Union the two together into a single clearing iterator.
clearclearIter, err := roaring.NewRepeatedRowIteratorFromBytes(clear)
if err != nil {
return errors.Wrap(err, "getting clear iterator")
}
setClearIter, err := roaring.NewRepeatedRowIteratorFromBytes(set)
if err != nil {
return errors.Wrap(err, "getting set/clear iterator")
}
clearIter := roaring.NewUnionContainerIterator(clearclearIter, setClearIter)
// Then we get the set iterator and create the rewriter.
setIter, err := roaring.NewContainerIterator(set)
if err != nil {
return errors.Wrap(err, "getting set iterator")
}
rewriter, err := roaring.NewClearAndSetRewriter(clearIter, setIter)
if err != nil {
return errors.Wrap(err, "getting rewriter")
}
err = tx.ApplyRewriter(f.index(), f.field(), f.view(), f.shard, 0, rewriter)
return errors.Wrap(err, "applying rewriter")
}
// ImportRoaringSingleValued treats "clear" as a single row and clears
// all the columns specified, then sets all the bits in set. It's very
// similar to ImportRoaringBSI, but doesn't treate the first row of
// "set" as the existence row to also be cleared. Essentially it's for
// FieldTypeMutex.
func (f *fragment) ImportRoaringSingleValued(ctx context.Context, tx Tx, clear, set []byte) error {
clearIter, err := roaring.NewRepeatedRowIteratorFromBytes(clear)
if err != nil {
return errors.Wrap(err, "getting cleariterator")
}
setIter, err := roaring.NewContainerIterator(set)
if err != nil {
return errors.Wrap(err, "getting set iterator")
}
rewriter, err := roaring.NewClearAndSetRewriter(clearIter, setIter)
if err != nil {
return errors.Wrap(err, "getting rewriter")
}
err = tx.ApplyRewriter(f.index(), f.field(), f.view(), f.shard, 0, rewriter)
return errors.Wrap(err, "applying rewriter")
}
func (f *fragment) doImportRoaring(ctx context.Context, tx Tx, data []byte, clear bool) (map[uint64]int, bool, error) {
f.mu.RLock()
defer f.mu.RUnlock()

View file

@ -349,6 +349,37 @@ type ImportRoaringRequest struct {
UpdateExistence bool
}
type ImportRoaringShardRequest struct {
// Has this request already been forwarded to all replicas? If
// Remote=false, then the handling server is responsible for
// ensuring this request is sent to all repliacs before returning
// a successful response to the client.
Remote bool
Views []RoaringUpdate
}
// RoaringUpdate represents the bits to clear and then set in a particular view.
type RoaringUpdate struct {
Field string
View string
// Clear is a roaring encoded bitmatrix of bits to clear. For
// mutex or int-like fields, only the first row is looked at and
// the bits in that row are cleared from every row.
Clear []byte
// Set is the roaring encoded bitmatrix of bits to set. If this is
// a mutex or int-like field, we'll assume the first shard width
// of containers is the exists row and we will first clear all
// bits in those columns and then set
Set []byte
// ClearRecords, when true, denotes that Clear should be
// interpreted as a single row which will be subtracted from every
// row in this view.
ClearRecords bool
}
// ValidateWithTimestamp ensures that the payload of the request is valid.
func (irr *ImportRoaringRequest) ValidateWithTimestamp(indexCreatedAt, fieldCreatedAt int64) error {
if (irr.IndexCreatedAt != 0 && irr.IndexCreatedAt != indexCreatedAt) ||

View file

@ -454,6 +454,7 @@ func newRouter(handler *Handler) http.Handler {
router.HandleFunc("/index/{index}/field/{field}/import", handler.chkAuthZ(handler.handlePostImport, authz.Write)).Methods("POST").Name("PostImport")
router.HandleFunc("/index/{index}/field/{field}/mutex-check", handler.chkAuthZ(handler.handleGetMutexCheck, authz.Read)).Methods("GET").Name("GetMutexCheck")
router.HandleFunc("/index/{index}/field/{field}/import-roaring/{shard}", handler.chkAuthZ(handler.handlePostImportRoaring, authz.Write)).Methods("POST").Name("PostImportRoaring")
router.HandleFunc("/index/{index}/shard/{shard}/import-roaring", handler.chkAuthZ(handler.handlePostShardImportRoaring, authz.Write)).Methods("POST").Name("PostImportRoaring")
router.HandleFunc("/index/{index}/query", handler.chkAuthZ(handler.handlePostQuery, authz.Read)).Methods("POST").Name("PostQuery")
router.HandleFunc("/info", handler.chkAuthZ(handler.handleGetInfo, authz.Admin)).Methods("GET").Name("GetInfo")
router.HandleFunc("/recalculate-caches", handler.chkAuthZ(handler.handleRecalculateCaches, authz.Admin)).Methods("POST").Name("RecalculateCaches")
@ -3336,6 +3337,81 @@ func (h *Handler) handlePostImportRoaring(w http.ResponseWriter, r *http.Request
}
}
// handlePostShardImportRoaring takes data for multiple fields for a
// particular shard and imports it all in a single transaction. It was
// developed in the post-RBF world and should probably ultimately
// replace most of the other import endpoints.
func (h *Handler) handlePostShardImportRoaring(w http.ResponseWriter, r *http.Request) {
// Verify that request is only communicating over protobufs.
if error, code := validateProtobufHeader(r); error != "" {
http.Error(w, error, code)
return
}
// Get index and field type to determine how to handle the
// import data.
indexName := mux.Vars(r)["index"]
ctx := r.Context()
// Read entire body.
span, _ := tracing.StartSpanFromContext(ctx, "ioutil.ReadAll-Body")
body, err := readBody(r)
span.LogKV("bodySize", len(body))
span.Finish()
if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
req := &ImportRoaringShardRequest{}
span, _ = tracing.StartSpanFromContext(ctx, "Unmarshal")
err = h.serializer.Unmarshal(body, req)
span.Finish()
if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
urlVars := mux.Vars(r)
shard, err := strconv.ParseUint(urlVars["shard"], 10, 64)
if err != nil {
http.Error(w, "shard should be an unsigned integer", http.StatusBadRequest)
return
}
resp := &ImportResponse{}
// TODO give meaningful stats for import
err = h.api.ImportRoaringShard(ctx, indexName, shard, req)
if err != nil {
resp.Err = err.Error()
if errors.Is(err, ErrIndexNotFound) {
w.WriteHeader(http.StatusNotFound)
} else if errors.As(err, &BadRequestError{}) {
w.WriteHeader(http.StatusBadRequest)
} else if _, ok := errors.Cause(err).(NotFoundError); ok {
w.WriteHeader(http.StatusNotFound)
} else if errors.As(err, &PreconditionFailedError{}) {
w.WriteHeader(http.StatusPreconditionFailed)
} else {
w.WriteHeader(http.StatusInternalServerError)
}
}
// Marshal response object.
buf, err := h.serializer.Marshal(resp)
if err != nil {
http.Error(w, fmt.Sprintf("marshal shard-import-roaring response: %v", err), http.StatusInternalServerError)
return
}
// Write response.
_, err = w.Write(buf)
if err != nil {
h.logger.Errorf("writing shard-import-roaring response: %v", err)
return
}
}
// handlePostIngestNode is the internal endpoint taking already-translated
// ingest operations, sorted by shard, for a single node.
func (h *Handler) handlePostIngestNode(w http.ResponseWriter, r *http.Request) {

View file

@ -2425,6 +2425,140 @@ func (m *ImportRoaringRequest) GetUpdateExistence() bool {
return false
}
type RoaringUpdate struct {
Field string `protobuf:"bytes,1,opt,name=Field,proto3" json:"Field,omitempty"`
View string `protobuf:"bytes,2,opt,name=View,proto3" json:"View,omitempty"`
Clear []byte `protobuf:"bytes,3,opt,name=Clear,proto3" json:"Clear,omitempty"`
Set []byte `protobuf:"bytes,4,opt,name=Set,proto3" json:"Set,omitempty"`
ClearRecords bool `protobuf:"varint,5,opt,name=ClearRecords,proto3" json:"ClearRecords,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
}
func (m *RoaringUpdate) Reset() { *m = RoaringUpdate{} }
func (m *RoaringUpdate) String() string { return proto.CompactTextString(m) }
func (*RoaringUpdate) ProtoMessage() {}
func (*RoaringUpdate) Descriptor() ([]byte, []int) {
return fileDescriptor_413a91106d7bcce8, []int{34}
}
func (m *RoaringUpdate) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
}
func (m *RoaringUpdate) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
if deterministic {
return xxx_messageInfo_RoaringUpdate.Marshal(b, m, deterministic)
} else {
b = b[:cap(b)]
n, err := m.MarshalToSizedBuffer(b)
if err != nil {
return nil, err
}
return b[:n], nil
}
}
func (m *RoaringUpdate) XXX_Merge(src proto.Message) {
xxx_messageInfo_RoaringUpdate.Merge(m, src)
}
func (m *RoaringUpdate) XXX_Size() int {
return m.Size()
}
func (m *RoaringUpdate) XXX_DiscardUnknown() {
xxx_messageInfo_RoaringUpdate.DiscardUnknown(m)
}
var xxx_messageInfo_RoaringUpdate proto.InternalMessageInfo
func (m *RoaringUpdate) GetField() string {
if m != nil {
return m.Field
}
return ""
}
func (m *RoaringUpdate) GetView() string {
if m != nil {
return m.View
}
return ""
}
func (m *RoaringUpdate) GetClear() []byte {
if m != nil {
return m.Clear
}
return nil
}
func (m *RoaringUpdate) GetSet() []byte {
if m != nil {
return m.Set
}
return nil
}
func (m *RoaringUpdate) GetClearRecords() bool {
if m != nil {
return m.ClearRecords
}
return false
}
type ImportRoaringShardRequest struct {
Remote bool `protobuf:"varint,1,opt,name=Remote,proto3" json:"Remote,omitempty"`
Views []*RoaringUpdate `protobuf:"bytes,2,rep,name=Views,proto3" json:"Views,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
}
func (m *ImportRoaringShardRequest) Reset() { *m = ImportRoaringShardRequest{} }
func (m *ImportRoaringShardRequest) String() string { return proto.CompactTextString(m) }
func (*ImportRoaringShardRequest) ProtoMessage() {}
func (*ImportRoaringShardRequest) Descriptor() ([]byte, []int) {
return fileDescriptor_413a91106d7bcce8, []int{35}
}
func (m *ImportRoaringShardRequest) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
}
func (m *ImportRoaringShardRequest) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
if deterministic {
return xxx_messageInfo_ImportRoaringShardRequest.Marshal(b, m, deterministic)
} else {
b = b[:cap(b)]
n, err := m.MarshalToSizedBuffer(b)
if err != nil {
return nil, err
}
return b[:n], nil
}
}
func (m *ImportRoaringShardRequest) XXX_Merge(src proto.Message) {
xxx_messageInfo_ImportRoaringShardRequest.Merge(m, src)
}
func (m *ImportRoaringShardRequest) XXX_Size() int {
return m.Size()
}
func (m *ImportRoaringShardRequest) XXX_DiscardUnknown() {
xxx_messageInfo_ImportRoaringShardRequest.DiscardUnknown(m)
}
var xxx_messageInfo_ImportRoaringShardRequest proto.InternalMessageInfo
func (m *ImportRoaringShardRequest) GetRemote() bool {
if m != nil {
return m.Remote
}
return false
}
func (m *ImportRoaringShardRequest) GetViews() []*RoaringUpdate {
if m != nil {
return m.Views
}
return nil
}
type GroupCounts struct {
Aggregate string `protobuf:"bytes,1,opt,name=Aggregate,proto3" json:"Aggregate,omitempty"`
Groups []*GroupCount `protobuf:"bytes,2,rep,name=Groups,proto3" json:"Groups,omitempty"`
@ -2437,7 +2571,7 @@ func (m *GroupCounts) Reset() { *m = GroupCounts{} }
func (m *GroupCounts) String() string { return proto.CompactTextString(m) }
func (*GroupCounts) ProtoMessage() {}
func (*GroupCounts) Descriptor() ([]byte, []int) {
return fileDescriptor_413a91106d7bcce8, []int{34}
return fileDescriptor_413a91106d7bcce8, []int{36}
}
func (m *GroupCounts) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -2515,115 +2649,121 @@ func init() {
proto.RegisterType((*TranslateIDsResponse)(nil), "pb.TranslateIDsResponse")
proto.RegisterType((*ImportRoaringRequestView)(nil), "pb.ImportRoaringRequestView")
proto.RegisterType((*ImportRoaringRequest)(nil), "pb.ImportRoaringRequest")
proto.RegisterType((*RoaringUpdate)(nil), "pb.RoaringUpdate")
proto.RegisterType((*ImportRoaringShardRequest)(nil), "pb.ImportRoaringShardRequest")
proto.RegisterType((*GroupCounts)(nil), "pb.GroupCounts")
}
func init() { proto.RegisterFile("public.proto", fileDescriptor_413a91106d7bcce8) }
var fileDescriptor_413a91106d7bcce8 = []byte{
// 1618 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x58, 0x5d, 0x6f, 0x1b, 0x45,
0x17, 0xce, 0xee, 0xfa, 0xf3, 0xd8, 0x71, 0x92, 0x69, 0xda, 0x77, 0xdf, 0xbe, 0xa9, 0x5f, 0x77,
0x85, 0x2a, 0x97, 0xa0, 0x54, 0x18, 0xa8, 0x50, 0x25, 0xa8, 0xe2, 0x38, 0x25, 0x56, 0x9b, 0xb4,
0x4c, 0x42, 0xe0, 0x82, 0x9b, 0x8d, 0x3d, 0xb8, 0x2b, 0xd6, 0x5e, 0xb3, 0x5e, 0xd7, 0xc9, 0x2f,
0x80, 0x9f, 0xc0, 0x1d, 0xbf, 0x06, 0xc1, 0x1d, 0x5c, 0x72, 0x89, 0xca, 0x1d, 0xbf, 0x02, 0x9d,
0x33, 0x33, 0xfb, 0x65, 0xb7, 0xaa, 0x2a, 0xee, 0xf6, 0x7c, 0xcc, 0x99, 0x39, 0xcf, 0xf9, 0xb4,
0xa1, 0x3e, 0x9d, 0x5f, 0xf8, 0xde, 0x60, 0x6f, 0x1a, 0x06, 0x51, 0xc0, 0xcc, 0xe9, 0x85, 0x73,
0x05, 0x16, 0x0f, 0x16, 0xcc, 0x86, 0xf2, 0x41, 0xe0, 0xcf, 0xc7, 0x93, 0x99, 0x6d, 0xb4, 0xac,
0x76, 0x81, 0x6b, 0x92, 0x31, 0x28, 0x3c, 0x16, 0x57, 0x33, 0xdb, 0x6a, 0x59, 0xed, 0x2a, 0xa7,
0x6f, 0xd4, 0xe6, 0x81, 0x1b, 0x7a, 0x93, 0x91, 0x5d, 0x68, 0x19, 0xed, 0x3a, 0xd7, 0x24, 0xdb,
0x86, 0x62, 0x7f, 0x32, 0x14, 0x97, 0x76, 0xb1, 0x65, 0xb4, 0xab, 0x5c, 0x12, 0xc8, 0x7d, 0xe4,
0x09, 0x7f, 0x68, 0x97, 0x24, 0x97, 0x08, 0xa7, 0x0d, 0x55, 0x1e, 0x2c, 0x8e, 0xdd, 0x28, 0xf4,
0x2e, 0xd9, 0xff, 0xa0, 0xc0, 0x83, 0x85, 0xbc, 0xbd, 0xd6, 0x29, 0xef, 0x4d, 0x2f, 0xf6, 0x78,
0xb0, 0xe0, 0xc4, 0x74, 0xf6, 0xa1, 0x7a, 0xea, 0x8d, 0x26, 0x62, 0x88, 0x4f, 0xfd, 0x2f, 0x58,
0xcf, 0x02, 0x54, 0x34, 0xd2, 0x8a, 0xc8, 0x43, 0xd1, 0x89, 0x18, 0xd9, 0x66, 0x4e, 0x74, 0x22,
0x46, 0xce, 0xc7, 0xd0, 0xe0, 0xc1, 0xa2, 0x3f, 0x14, 0x93, 0xc8, 0xfb, 0xc6, 0x13, 0x21, 0x39,
0x16, 0xdf, 0x58, 0x90, 0x17, 0xc5, 0xce, 0x9a, 0x89, 0xb3, 0xce, 0x4d, 0x28, 0xf5, 0x7b, 0x4f,
0xbc, 0x59, 0xc4, 0x36, 0xc1, 0xea, 0xf7, 0xf4, 0x01, 0xfc, 0x74, 0x0e, 0x60, 0xeb, 0xf0, 0x32,
0x0a, 0xdd, 0x41, 0x24, 0x86, 0xfd, 0x9e, 0x84, 0x8c, 0x35, 0xc0, 0xec, 0xf7, 0xe8, 0x7d, 0x05,
0x6e, 0xf6, 0x7b, 0xac, 0x09, 0x85, 0x73, 0xd7, 0x97, 0x46, 0x6b, 0x1d, 0xc0, 0x67, 0x49, 0x83,
0x9c, 0xf8, 0xce, 0xd7, 0x19, 0x23, 0x0a, 0x8f, 0x1b, 0x50, 0x22, 0x94, 0xe4, 0x75, 0x55, 0xae,
0x28, 0x76, 0x2f, 0x09, 0x94, 0xb4, 0x77, 0x1d, 0xed, 0x2d, 0x3d, 0x22, 0x8e, 0x9f, 0x73, 0x0b,
0xca, 0x8f, 0xc5, 0x15, 0xbd, 0x5f, 0x7b, 0x67, 0xa4, 0xbc, 0xfb, 0xcd, 0x80, 0x6b, 0xf1, 0xe9,
0x33, 0xf7, 0xc2, 0x17, 0xe7, 0xae, 0x3f, 0x17, 0xac, 0xa9, 0x7d, 0x35, 0xb2, 0x6f, 0x3e, 0x5a,
0x23, 0xcf, 0xd9, 0xed, 0x18, 0x29, 0x54, 0xa8, 0xa1, 0x82, 0xba, 0xe6, 0x68, 0x4d, 0x65, 0xc9,
0x0e, 0x54, 0xba, 0xa7, 0x7d, 0x32, 0x67, 0x5b, 0x2d, 0xa3, 0x6d, 0x1d, 0xad, 0xf1, 0x98, 0xc3,
0x6e, 0x42, 0xf9, 0x78, 0x1e, 0x89, 0xcb, 0x7e, 0x8f, 0x72, 0xa8, 0x70, 0xb4, 0xc6, 0x35, 0x03,
0x4f, 0xd2, 0xe7, 0x63, 0x71, 0x25, 0x13, 0x09, 0x4f, 0x6a, 0x0e, 0xdb, 0x86, 0x42, 0x37, 0x08,
0x7c, 0x4a, 0xa6, 0x0a, 0xde, 0x86, 0x54, 0xb7, 0x0c, 0x45, 0x32, 0xec, 0x5c, 0xc2, 0x76, 0xd6,
0x21, 0x15, 0x16, 0x06, 0x16, 0xda, 0x33, 0x94, 0x3d, 0x24, 0xd8, 0x26, 0x85, 0xca, 0x54, 0xf7,
0x63, 0xb0, 0xee, 0x41, 0x89, 0xcc, 0xc8, 0x84, 0xaf, 0x75, 0xfe, 0x93, 0x81, 0x37, 0x01, 0x88,
0x2b, 0xb5, 0x6e, 0x95, 0xf0, 0x7d, 0x1a, 0xf6, 0x7b, 0xce, 0x27, 0x79, 0x28, 0x29, 0x66, 0x08,
0xfb, 0x89, 0x3b, 0x16, 0xf2, 0x66, 0x4e, 0xdf, 0xc8, 0x3b, 0xbb, 0x9a, 0x0a, 0xba, 0xba, 0xca,
0xe9, 0xdb, 0x99, 0x43, 0x23, 0x7b, 0x1c, 0x1f, 0x93, 0x4a, 0x82, 0x95, 0x8f, 0x21, 0x79, 0x9c,
0x1d, 0x9d, 0x7c, 0x76, 0xd8, 0xcb, 0x27, 0xf2, 0x09, 0xf2, 0x29, 0x14, 0x9e, 0xb9, 0x5e, 0xb8,
0x94, 0xb6, 0x9b, 0x12, 0x2f, 0x8b, 0x5e, 0x68, 0x49, 0xe0, 0x8b, 0x07, 0xc1, 0x7c, 0x12, 0x49,
0xc0, 0xb8, 0x24, 0x9c, 0x87, 0x50, 0xc5, 0xf3, 0xd2, 0xd7, 0x1d, 0x69, 0x4c, 0xe5, 0x4d, 0x05,
0x6f, 0x47, 0x9a, 0xcb, 0x2b, 0xe2, 0x3e, 0x60, 0xa6, 0xfb, 0x40, 0x17, 0x00, 0xa5, 0x33, 0x69,
0xa1, 0x09, 0x45, 0xa2, 0x94, 0xcb, 0x89, 0x09, 0xc9, 0x7e, 0x85, 0x8d, 0x5b, 0xd8, 0x77, 0xa2,
0xfb, 0x1f, 0xa2, 0x58, 0x66, 0x1c, 0xbe, 0xc0, 0xe2, 0x2a, 0x27, 0x02, 0xa8, 0x48, 0xa0, 0x82,
0x45, 0x62, 0xc0, 0x48, 0x19, 0x40, 0x2e, 0xf6, 0x87, 0x9e, 0xf6, 0x8d, 0x08, 0xac, 0x42, 0x1e,
0x2c, 0x12, 0x18, 0x14, 0xc5, 0xfe, 0xaf, 0x6f, 0x29, 0x90, 0x9f, 0x55, 0xaa, 0x0f, 0xbc, 0x5f,
0x5f, 0xf8, 0x15, 0xc0, 0x67, 0x61, 0x30, 0x9f, 0x12, 0x44, 0xcc, 0x81, 0x22, 0x51, 0xca, 0xa7,
0x3a, 0xaa, 0xeb, 0xf7, 0x70, 0x29, 0x5a, 0x0d, 0x2e, 0x06, 0x61, 0x7f, 0x34, 0x92, 0xe5, 0xc3,
0xf1, 0xd3, 0xf9, 0xc9, 0x80, 0xca, 0xb9, 0xeb, 0xc7, 0xe2, 0x73, 0xd7, 0x57, 0xbe, 0xe2, 0x67,
0xd6, 0x8c, 0xa5, 0xcd, 0xdc, 0x84, 0xca, 0x23, 0x3f, 0x70, 0x23, 0x54, 0x46, 0x5b, 0x06, 0x8f,
0x69, 0xb6, 0x0b, 0xd0, 0x13, 0x03, 0x6f, 0xec, 0xfa, 0x28, 0x2d, 0x24, 0xf5, 0xac, 0xb8, 0x3c,
0x25, 0x66, 0x0e, 0xd4, 0xcf, 0xbc, 0xb1, 0x98, 0x45, 0xee, 0x78, 0x8a, 0xea, 0xb2, 0xcd, 0x67,
0x78, 0xce, 0x47, 0x50, 0x56, 0x27, 0x56, 0x47, 0x03, 0xb9, 0xa7, 0x03, 0xd7, 0x17, 0xfa, 0x8d,
0x44, 0x38, 0x0f, 0x61, 0xab, 0xe7, 0xcd, 0x22, 0x6f, 0x32, 0x88, 0x62, 0x73, 0x18, 0x00, 0x55,
0x8e, 0xaa, 0x0d, 0x4a, 0x2a, 0xae, 0x29, 0x33, 0xa9, 0x29, 0xe7, 0x67, 0x03, 0xea, 0x9f, 0xcf,
0x45, 0x78, 0xc5, 0xc5, 0x77, 0x73, 0x31, 0x8b, 0xf0, 0x1e, 0xa2, 0x75, 0xa4, 0x89, 0x40, 0x93,
0xa7, 0xcf, 0xdd, 0x70, 0x28, 0x4b, 0xa4, 0xc0, 0x15, 0x45, 0xb1, 0x16, 0xe3, 0x20, 0x12, 0xe4,
0x54, 0x85, 0x2b, 0x8a, 0xed, 0x42, 0xfd, 0x70, 0x7c, 0x21, 0x86, 0x43, 0x31, 0xec, 0xb9, 0x91,
0x6b, 0x57, 0xb2, 0x13, 0x2a, 0x23, 0x64, 0xef, 0xc0, 0xfa, 0xb3, 0x50, 0x9c, 0x85, 0xee, 0x64,
0xe6, 0xbb, 0x91, 0x18, 0xda, 0x55, 0xb2, 0x95, 0x65, 0xb2, 0x1d, 0xa8, 0x1e, 0xbb, 0x97, 0xc7,
0x62, 0x1c, 0x84, 0x57, 0x36, 0x10, 0x08, 0x09, 0xc3, 0x79, 0x02, 0xeb, 0xca, 0x8d, 0xd9, 0x34,
0x98, 0xcc, 0x04, 0x46, 0xf9, 0x30, 0x0c, 0x95, 0x17, 0xf8, 0xc9, 0xee, 0x42, 0x99, 0x8b, 0xd9,
0xdc, 0x8f, 0x74, 0x9d, 0x6f, 0xe0, 0x73, 0xf4, 0xa9, 0xb9, 0x1f, 0x71, 0x2d, 0x77, 0xfe, 0x2e,
0x42, 0x2d, 0x25, 0x88, 0x3b, 0x0f, 0x76, 0xcf, 0x75, 0xd9, 0x79, 0x70, 0x6e, 0xf2, 0x60, 0xb1,
0x34, 0x52, 0xb1, 0x5a, 0xea, 0x60, 0x9c, 0xa8, 0x94, 0x34, 0x4e, 0x92, 0xe2, 0xb4, 0x56, 0x17,
0x27, 0xae, 0x11, 0xcf, 0xdd, 0xc9, 0x48, 0x0c, 0x29, 0x91, 0x2a, 0x5c, 0x93, 0xac, 0x9d, 0x64,
0x2d, 0xe1, 0xab, 0xaa, 0x40, 0xf3, 0x78, 0x92, 0xd3, 0xb2, 0xe6, 0x70, 0xf8, 0x94, 0x65, 0x7c,
0x24, 0xc5, 0xee, 0x43, 0xe3, 0xa9, 0x3f, 0x4c, 0xaa, 0x6a, 0xa6, 0x22, 0xd1, 0x40, 0x3b, 0x09,
0x9b, 0xe7, 0xb4, 0xd8, 0x83, 0xfc, 0xe4, 0xa7, 0x98, 0xd4, 0x3a, 0x4c, 0xf9, 0x99, 0x92, 0xf0,
0xfc, 0x8e, 0xb0, 0x9b, 0x5a, 0x3c, 0x28, 0x50, 0xb5, 0xce, 0x3a, 0x1e, 0x8b, 0x99, 0x3c, 0xb5,
0x98, 0xec, 0xa5, 0xfb, 0x98, 0x5d, 0x23, 0xed, 0x86, 0x46, 0x48, 0x72, 0x79, 0xba, 0xd3, 0xed,
0xa6, 0x1a, 0xa7, 0x5d, 0x4f, 0x8c, 0xc7, 0x4c, 0x9e, 0x6a, 0xac, 0x07, 0x2b, 0x96, 0x04, 0x7b,
0x9d, 0x0e, 0xe5, 0x37, 0x00, 0x29, 0xe4, 0x2b, 0x96, 0x8a, 0x07, 0xf9, 0x09, 0x63, 0x37, 0x12,
0x28, 0xb2, 0x12, 0x9e, 0x9f, 0x45, 0xbb, 0xa9, 0x6d, 0xcd, 0xde, 0x48, 0x5e, 0x1b, 0x33, 0x79,
0x6a, 0x9b, 0x7b, 0x1f, 0x6a, 0xe9, 0x40, 0x6d, 0x92, 0xfa, 0x46, 0x36, 0x50, 0x33, 0x9e, 0xd6,
0x41, 0x07, 0x97, 0xca, 0xdf, 0xde, 0x4a, 0x1c, 0x5c, 0x12, 0xf2, 0x65, 0x7d, 0xe7, 0x17, 0x13,
0xd6, 0xfb, 0xe3, 0x69, 0x10, 0x46, 0xa9, 0x1e, 0x20, 0x17, 0x52, 0x63, 0xe5, 0x42, 0x6a, 0xe6,
0x66, 0x00, 0xf5, 0x02, 0x6a, 0x91, 0x05, 0x2e, 0x89, 0x54, 0x3e, 0x16, 0x32, 0xf9, 0xb8, 0x03,
0x55, 0x39, 0x42, 0x51, 0x54, 0x24, 0x51, 0xc2, 0x90, 0x2b, 0xf2, 0x82, 0x56, 0xa4, 0x32, 0x75,
0x2e, 0x4d, 0xb2, 0x26, 0x80, 0x54, 0x23, 0x61, 0x85, 0x84, 0x29, 0x0e, 0xca, 0x63, 0x87, 0x66,
0x76, 0xa9, 0x65, 0xb5, 0x2d, 0x9e, 0xe2, 0xb0, 0x3b, 0xd0, 0x20, 0x27, 0x0e, 0x42, 0x81, 0xcd,
0x64, 0x3f, 0xa2, 0x7c, 0xb6, 0x78, 0x8e, 0x8b, 0x7a, 0xe4, 0x56, 0xa2, 0x27, 0x3b, 0x4d, 0x8e,
0x4b, 0x13, 0xc3, 0x17, 0x6e, 0x48, 0x19, 0x5b, 0xe1, 0x92, 0x70, 0xfe, 0x30, 0x81, 0x49, 0x24,
0xe5, 0xba, 0xf3, 0xaf, 0xc1, 0xf9, 0x7a, 0xd8, 0xb2, 0xe0, 0x94, 0x97, 0xc0, 0x49, 0xe6, 0x81,
0x04, 0x46, 0xcf, 0x83, 0x16, 0xd4, 0xf4, 0x40, 0x43, 0x21, 0xa2, 0x6a, 0xf0, 0x34, 0x0b, 0x27,
0xd7, 0x69, 0x84, 0xbf, 0x51, 0x94, 0x4a, 0x95, 0x6c, 0x67, 0x78, 0x2b, 0xa0, 0x85, 0x37, 0x84,
0xb6, 0xf6, 0x7a, 0x68, 0xeb, 0x69, 0x68, 0xbf, 0x37, 0xa0, 0xbe, 0x1f, 0x05, 0x63, 0x6f, 0xc0,
0xc5, 0x20, 0x08, 0x87, 0xaf, 0x06, 0x55, 0xc2, 0x67, 0xa6, 0xe1, 0x6b, 0x83, 0xd5, 0x7f, 0x11,
0xaa, 0xfe, 0x7b, 0x83, 0xf6, 0x8e, 0xa5, 0x28, 0x71, 0x54, 0x61, 0xb7, 0xc1, 0xec, 0x87, 0x94,
0xb3, 0xb5, 0xce, 0x56, 0xa2, 0xa8, 0x75, 0xcc, 0x7e, 0xe8, 0xbc, 0x07, 0xdb, 0xf2, 0x21, 0x5a,
0xa4, 0x06, 0xce, 0x36, 0x14, 0x0f, 0xc3, 0x30, 0xd0, 0x23, 0x47, 0x12, 0xb8, 0x58, 0xc7, 0x33,
0x0c, 0x83, 0xf1, 0x36, 0x39, 0xb1, 0xea, 0xd7, 0x64, 0x0b, 0x6a, 0x27, 0x41, 0xf4, 0x65, 0xe8,
0x45, 0xd4, 0x92, 0xe4, 0xe0, 0x48, 0xb3, 0x9c, 0xbb, 0x70, 0x3d, 0x77, 0x73, 0x32, 0x19, 0x31,
0x8d, 0xac, 0xe4, 0x17, 0xd9, 0x29, 0x5c, 0x8b, 0x55, 0xfb, 0xbd, 0xb7, 0x7a, 0xe3, 0xb2, 0xd1,
0x77, 0x53, 0x9e, 0x93, 0x51, 0x75, 0xfd, 0x0a, 0x6f, 0x9c, 0x2e, 0xd8, 0x0a, 0x4d, 0xf9, 0x93,
0x58, 0xbd, 0xe0, 0xdc, 0x13, 0x8b, 0x57, 0xfd, 0x12, 0xa0, 0xb5, 0xc2, 0xa4, 0x1f, 0xd2, 0xf4,
0xed, 0xfc, 0x60, 0xc2, 0xf6, 0x2a, 0x23, 0x49, 0x42, 0x19, 0xa9, 0x84, 0x62, 0x1d, 0x28, 0xbe,
0xf0, 0xc4, 0x42, 0xef, 0x02, 0x3b, 0xa9, 0x60, 0x2f, 0xbd, 0x81, 0x4b, 0x55, 0x2c, 0xa4, 0xfd,
0x41, 0xe4, 0x05, 0x13, 0xbd, 0xd9, 0x4a, 0x0a, 0x6f, 0xe8, 0xfa, 0xc1, 0xe0, 0x5b, 0xf9, 0xa3,
0x8c, 0x4b, 0x62, 0x45, 0x61, 0x14, 0xdf, 0xb0, 0x30, 0x4a, 0x2b, 0x0b, 0xa3, 0x0d, 0x1b, 0x5f,
0x4c, 0x87, 0x6e, 0x24, 0x0e, 0x2f, 0xbd, 0x59, 0x24, 0x26, 0x03, 0x61, 0x97, 0xc9, 0xa3, 0x3c,
0xdb, 0x39, 0xcd, 0x4c, 0x12, 0xec, 0x1e, 0xfb, 0xa3, 0x51, 0x28, 0x46, 0x6e, 0xa4, 0x61, 0x4c,
0x18, 0xec, 0x0e, 0x94, 0x48, 0x59, 0x23, 0x91, 0x5f, 0x0d, 0x94, 0xb4, 0xbb, 0xf9, 0xeb, 0xcb,
0xa6, 0xf1, 0xfb, 0xcb, 0xa6, 0xf1, 0xe7, 0xcb, 0xa6, 0xf1, 0xe3, 0x5f, 0xcd, 0xb5, 0x8b, 0x12,
0xfd, 0x23, 0xf2, 0xc1, 0x3f, 0x01, 0x00, 0x00, 0xff, 0xff, 0xec, 0x75, 0x7f, 0x8e, 0x21, 0x11,
0x00, 0x00,
// 1694 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x58, 0x5b, 0x6f, 0x23, 0x49,
0x15, 0x4e, 0x5f, 0x7c, 0x3b, 0x76, 0x9c, 0xa4, 0x36, 0xbb, 0xf4, 0x0e, 0x59, 0xe3, 0x6d, 0xa1,
0xc5, 0x4b, 0x50, 0x56, 0x18, 0x18, 0xa1, 0x91, 0x60, 0x14, 0xc7, 0x19, 0x62, 0xcd, 0x24, 0x33,
0x94, 0x43, 0xe0, 0x61, 0x5e, 0x3a, 0x76, 0xe1, 0x69, 0xd1, 0x76, 0x9b, 0xee, 0xf6, 0x38, 0xf9,
0x01, 0x08, 0x7e, 0x02, 0x6f, 0xfc, 0x1a, 0x04, 0x6f, 0xf0, 0xc8, 0x23, 0x1a, 0xde, 0xf8, 0x15,
0xe8, 0x9c, 0xaa, 0xea, 0x9b, 0x9d, 0xd1, 0x68, 0xc4, 0x5b, 0x9f, 0x4b, 0x9d, 0xaa, 0xf3, 0x9d,
0xab, 0x0d, 0xad, 0xe5, 0xea, 0x36, 0xf0, 0x27, 0x27, 0xcb, 0x28, 0x4c, 0x42, 0x66, 0x2e, 0x6f,
0xdd, 0x7b, 0xb0, 0x78, 0xb8, 0x66, 0x0e, 0xd4, 0xce, 0xc2, 0x60, 0x35, 0x5f, 0xc4, 0x8e, 0xd1,
0xb5, 0x7a, 0x36, 0xd7, 0x24, 0x63, 0x60, 0x3f, 0x17, 0xf7, 0xb1, 0x63, 0x75, 0xad, 0x5e, 0x83,
0xd3, 0x37, 0x6a, 0xf3, 0xd0, 0x8b, 0xfc, 0xc5, 0xcc, 0xb1, 0xbb, 0x46, 0xaf, 0xc5, 0x35, 0xc9,
0x0e, 0xa1, 0x32, 0x5a, 0x4c, 0xc5, 0x9d, 0x53, 0xe9, 0x1a, 0xbd, 0x06, 0x97, 0x04, 0x72, 0x9f,
0xf9, 0x22, 0x98, 0x3a, 0x55, 0xc9, 0x25, 0xc2, 0xed, 0x41, 0x83, 0x87, 0xeb, 0x4b, 0x2f, 0x89,
0xfc, 0x3b, 0xf6, 0x6d, 0xb0, 0x79, 0xb8, 0x96, 0xb7, 0x37, 0xfb, 0xb5, 0x93, 0xe5, 0xed, 0x09,
0x0f, 0xd7, 0x9c, 0x98, 0xee, 0x29, 0x34, 0xc6, 0xfe, 0x6c, 0x21, 0xa6, 0xf8, 0xd4, 0xcf, 0xc1,
0x7a, 0x15, 0xa2, 0xa2, 0x91, 0x57, 0x44, 0x1e, 0x8a, 0xae, 0xc4, 0xcc, 0x31, 0x4b, 0xa2, 0x2b,
0x31, 0x73, 0x7f, 0x0a, 0x6d, 0x1e, 0xae, 0x47, 0x53, 0xb1, 0x48, 0xfc, 0xdf, 0xfa, 0x22, 0x22,
0xc7, 0xd2, 0x1b, 0x6d, 0x79, 0x51, 0xea, 0xac, 0x99, 0x39, 0xeb, 0x3e, 0x82, 0xea, 0x68, 0xf8,
0xc2, 0x8f, 0x13, 0xb6, 0x0f, 0xd6, 0x68, 0xa8, 0x0f, 0xe0, 0xa7, 0x7b, 0x06, 0x07, 0xe7, 0x77,
0x49, 0xe4, 0x4d, 0x12, 0x31, 0x1d, 0x0d, 0x25, 0x64, 0xac, 0x0d, 0xe6, 0x68, 0x48, 0xef, 0xb3,
0xb9, 0x39, 0x1a, 0xb2, 0x0e, 0xd8, 0x37, 0x5e, 0x20, 0x8d, 0x36, 0xfb, 0x80, 0xcf, 0x92, 0x06,
0x39, 0xf1, 0xdd, 0xd7, 0x05, 0x23, 0x0a, 0x8f, 0xcf, 0xa0, 0x4a, 0x28, 0xc9, 0xeb, 0x1a, 0x5c,
0x51, 0xec, 0x9b, 0x2c, 0x50, 0xd2, 0xde, 0xa7, 0x68, 0x6f, 0xe3, 0x11, 0x69, 0xfc, 0xdc, 0x2f,
0xa0, 0xf6, 0x5c, 0xdc, 0xd3, 0xfb, 0xb5, 0x77, 0x46, 0xce, 0xbb, 0x7f, 0x18, 0xf0, 0x49, 0x7a,
0xfa, 0xda, 0xbb, 0x0d, 0xc4, 0x8d, 0x17, 0xac, 0x04, 0xeb, 0x68, 0x5f, 0x8d, 0xe2, 0x9b, 0x2f,
0x76, 0xc8, 0x73, 0xf6, 0x65, 0x8a, 0x14, 0x2a, 0x34, 0x51, 0x41, 0x5d, 0x73, 0xb1, 0xa3, 0xb2,
0xe4, 0x08, 0xea, 0x83, 0xf1, 0x88, 0xcc, 0x39, 0x56, 0xd7, 0xe8, 0x59, 0x17, 0x3b, 0x3c, 0xe5,
0xb0, 0x47, 0x50, 0xbb, 0x5c, 0x25, 0xe2, 0x6e, 0x34, 0xa4, 0x1c, 0xb2, 0x2f, 0x76, 0xb8, 0x66,
0xe0, 0x49, 0xfa, 0x7c, 0x2e, 0xee, 0x65, 0x22, 0xe1, 0x49, 0xcd, 0x61, 0x87, 0x60, 0x0f, 0xc2,
0x30, 0xa0, 0x64, 0xaa, 0xe3, 0x6d, 0x48, 0x0d, 0x6a, 0x50, 0x21, 0xc3, 0xee, 0x1d, 0x1c, 0x16,
0x1d, 0x52, 0x61, 0x61, 0x60, 0xa1, 0x3d, 0x43, 0xd9, 0x43, 0x82, 0xed, 0x53, 0xa8, 0x4c, 0x75,
0x3f, 0x06, 0xeb, 0x1b, 0xa8, 0x92, 0x19, 0x99, 0xf0, 0xcd, 0xfe, 0xb7, 0x0a, 0xf0, 0x66, 0x00,
0x71, 0xa5, 0x36, 0x68, 0x10, 0xbe, 0x2f, 0xa3, 0xd1, 0xd0, 0xfd, 0x59, 0x19, 0x4a, 0x8a, 0x19,
0xc2, 0x7e, 0xe5, 0xcd, 0x85, 0xbc, 0x99, 0xd3, 0x37, 0xf2, 0xae, 0xef, 0x97, 0x82, 0xae, 0x6e,
0x70, 0xfa, 0x76, 0x57, 0xd0, 0x2e, 0x1e, 0xc7, 0xc7, 0xe4, 0x92, 0x60, 0xeb, 0x63, 0x48, 0x9e,
0x66, 0x47, 0xbf, 0x9c, 0x1d, 0xce, 0xe6, 0x89, 0x72, 0x82, 0xfc, 0x1c, 0xec, 0x57, 0x9e, 0x1f,
0x6d, 0xa4, 0xed, 0xbe, 0xc4, 0xcb, 0xa2, 0x17, 0x5a, 0x12, 0xf8, 0xca, 0x59, 0xb8, 0x5a, 0x24,
0x12, 0x30, 0x2e, 0x09, 0xf7, 0x29, 0x34, 0xf0, 0xbc, 0xf4, 0xf5, 0x48, 0x1a, 0x53, 0x79, 0x53,
0xc7, 0xdb, 0x91, 0xe6, 0xf2, 0x8a, 0xb4, 0x0f, 0x98, 0xf9, 0x3e, 0x30, 0x00, 0x40, 0x69, 0x2c,
0x2d, 0x74, 0xa0, 0x42, 0x94, 0x72, 0x39, 0x33, 0x21, 0xd9, 0x0f, 0xd8, 0xf8, 0x02, 0xfb, 0x4e,
0xf2, 0xf8, 0xc7, 0x28, 0x96, 0x19, 0x87, 0x2f, 0xb0, 0xb8, 0xca, 0x89, 0x10, 0xea, 0x12, 0xa8,
0x70, 0x9d, 0x19, 0x30, 0x72, 0x06, 0x90, 0x8b, 0xfd, 0x61, 0xa8, 0x7d, 0x23, 0x02, 0xab, 0x90,
0x87, 0xeb, 0x0c, 0x06, 0x45, 0xb1, 0xef, 0xe8, 0x5b, 0x6c, 0xf2, 0xb3, 0x41, 0xf5, 0x81, 0xf7,
0xeb, 0x0b, 0x7f, 0x03, 0xf0, 0x8b, 0x28, 0x5c, 0x2d, 0x09, 0x22, 0xe6, 0x42, 0x85, 0x28, 0xe5,
0x53, 0x0b, 0xd5, 0xf5, 0x7b, 0xb8, 0x14, 0x6d, 0x07, 0x17, 0x83, 0x70, 0x3a, 0x9b, 0xc9, 0xf2,
0xe1, 0xf8, 0xe9, 0xfe, 0xc5, 0x80, 0xfa, 0x8d, 0x17, 0xa4, 0xe2, 0x1b, 0x2f, 0x50, 0xbe, 0xe2,
0x67, 0xd1, 0x8c, 0xa5, 0xcd, 0x3c, 0x82, 0xfa, 0xb3, 0x20, 0xf4, 0x12, 0x54, 0x46, 0x5b, 0x06,
0x4f, 0x69, 0x76, 0x0c, 0x30, 0x14, 0x13, 0x7f, 0xee, 0x05, 0x28, 0xb5, 0xb3, 0x7a, 0x56, 0x5c,
0x9e, 0x13, 0x33, 0x17, 0x5a, 0xd7, 0xfe, 0x5c, 0xc4, 0x89, 0x37, 0x5f, 0xa2, 0xba, 0x6c, 0xf3,
0x05, 0x9e, 0xfb, 0x13, 0xa8, 0xa9, 0x13, 0xdb, 0xa3, 0x81, 0xdc, 0xf1, 0xc4, 0x0b, 0x84, 0x7e,
0x23, 0x11, 0xee, 0x53, 0x38, 0x18, 0xfa, 0x71, 0xe2, 0x2f, 0x26, 0x49, 0x6a, 0x0e, 0x03, 0xa0,
0xca, 0x51, 0xb5, 0x41, 0x49, 0xa5, 0x35, 0x65, 0x66, 0x35, 0xe5, 0xfe, 0xd5, 0x80, 0xd6, 0x2f,
0x57, 0x22, 0xba, 0xe7, 0xe2, 0xf7, 0x2b, 0x11, 0x27, 0x78, 0x0f, 0xd1, 0x3a, 0xd2, 0x44, 0xa0,
0xc9, 0xf1, 0x1b, 0x2f, 0x9a, 0xca, 0x12, 0xb1, 0xb9, 0xa2, 0x28, 0xd6, 0x62, 0x1e, 0x26, 0x82,
0x9c, 0xaa, 0x73, 0x45, 0xb1, 0x63, 0x68, 0x9d, 0xcf, 0x6f, 0xc5, 0x74, 0x2a, 0xa6, 0x43, 0x2f,
0xf1, 0x9c, 0x7a, 0x71, 0x42, 0x15, 0x84, 0xec, 0xbb, 0xb0, 0xfb, 0x2a, 0x12, 0xd7, 0x91, 0xb7,
0x88, 0x03, 0x2f, 0x11, 0x53, 0xa7, 0x41, 0xb6, 0x8a, 0x4c, 0x76, 0x04, 0x8d, 0x4b, 0xef, 0xee,
0x52, 0xcc, 0xc3, 0xe8, 0xde, 0x01, 0x02, 0x21, 0x63, 0xb8, 0x2f, 0x60, 0x57, 0xb9, 0x11, 0x2f,
0xc3, 0x45, 0x2c, 0x30, 0xca, 0xe7, 0x51, 0xa4, 0xbc, 0xc0, 0x4f, 0xf6, 0x35, 0xd4, 0xb8, 0x88,
0x57, 0x41, 0xa2, 0xeb, 0x7c, 0x0f, 0x9f, 0xa3, 0x4f, 0xad, 0x82, 0x84, 0x6b, 0xb9, 0xfb, 0xdf,
0x0a, 0x34, 0x73, 0x82, 0xb4, 0xf3, 0x60, 0xf7, 0xdc, 0x95, 0x9d, 0x07, 0xe7, 0x26, 0x0f, 0xd7,
0x1b, 0x23, 0x15, 0xab, 0xa5, 0x05, 0xc6, 0x95, 0x4a, 0x49, 0xe3, 0x2a, 0x2b, 0x4e, 0x6b, 0x7b,
0x71, 0xe2, 0x1a, 0xf1, 0xc6, 0x5b, 0xcc, 0xc4, 0x94, 0x12, 0xa9, 0xce, 0x35, 0xc9, 0x7a, 0x59,
0xd6, 0x12, 0xbe, 0xaa, 0x0a, 0x34, 0x8f, 0x67, 0x39, 0x2d, 0x6b, 0x0e, 0x87, 0x4f, 0x4d, 0xc6,
0x47, 0x52, 0xec, 0x31, 0xb4, 0x5f, 0x06, 0xd3, 0xac, 0xaa, 0x62, 0x15, 0x89, 0x36, 0xda, 0xc9,
0xd8, 0xbc, 0xa4, 0xc5, 0x9e, 0x94, 0x27, 0x3f, 0xc5, 0xa4, 0xd9, 0x67, 0xca, 0xcf, 0x9c, 0x84,
0x97, 0x77, 0x84, 0xe3, 0xdc, 0xe2, 0x41, 0x81, 0x6a, 0xf6, 0x77, 0xf1, 0x58, 0xca, 0xe4, 0xb9,
0xc5, 0xe4, 0x24, 0xdf, 0xc7, 0x9c, 0x26, 0x69, 0xb7, 0x35, 0x42, 0x92, 0xcb, 0xf3, 0x9d, 0xee,
0x38, 0xd7, 0x38, 0x9d, 0x56, 0x66, 0x3c, 0x65, 0xf2, 0x5c, 0x63, 0x3d, 0xdb, 0xb2, 0x24, 0x38,
0xbb, 0x74, 0xa8, 0xbc, 0x01, 0x48, 0x21, 0xdf, 0xb2, 0x54, 0x3c, 0x29, 0x4f, 0x18, 0xa7, 0x9d,
0x41, 0x51, 0x94, 0xf0, 0xf2, 0x2c, 0x3a, 0xce, 0x6d, 0x6b, 0xce, 0x5e, 0xf6, 0xda, 0x94, 0xc9,
0x73, 0xdb, 0xdc, 0x0f, 0xa1, 0x99, 0x0f, 0xd4, 0x3e, 0xa9, 0xef, 0x15, 0x03, 0x15, 0xf3, 0xbc,
0x0e, 0x3a, 0xb8, 0x51, 0xfe, 0xce, 0x41, 0xe6, 0xe0, 0x86, 0x90, 0x6f, 0xea, 0xbb, 0x7f, 0x33,
0x61, 0x77, 0x34, 0x5f, 0x86, 0x51, 0x92, 0xeb, 0x01, 0x72, 0x21, 0x35, 0xb6, 0x2e, 0xa4, 0x66,
0x69, 0x06, 0x50, 0x2f, 0xa0, 0x16, 0x69, 0x73, 0x49, 0xe4, 0xf2, 0xd1, 0x2e, 0xe4, 0xe3, 0x11,
0x34, 0xe4, 0x08, 0x45, 0x51, 0x85, 0x44, 0x19, 0x43, 0xae, 0xc8, 0x6b, 0x5a, 0x91, 0x6a, 0xd4,
0xb9, 0x34, 0xc9, 0x3a, 0x00, 0x52, 0x8d, 0x84, 0x75, 0x12, 0xe6, 0x38, 0x28, 0x4f, 0x1d, 0x8a,
0x9d, 0x6a, 0xd7, 0xea, 0x59, 0x3c, 0xc7, 0x61, 0x5f, 0x41, 0x9b, 0x9c, 0x38, 0x8b, 0x04, 0x36,
0x93, 0xd3, 0x84, 0xf2, 0xd9, 0xe2, 0x25, 0x2e, 0xea, 0x91, 0x5b, 0x99, 0x9e, 0xec, 0x34, 0x25,
0x2e, 0x4d, 0x8c, 0x40, 0x78, 0x11, 0x65, 0x6c, 0x9d, 0x4b, 0xc2, 0xfd, 0x97, 0x09, 0x4c, 0x22,
0x29, 0xd7, 0x9d, 0xff, 0x1b, 0x9c, 0xef, 0x87, 0xad, 0x08, 0x4e, 0x6d, 0x03, 0x9c, 0x6c, 0x1e,
0x48, 0x60, 0xf4, 0x3c, 0xe8, 0x42, 0x53, 0x0f, 0x34, 0x14, 0x22, 0xaa, 0x06, 0xcf, 0xb3, 0x70,
0x72, 0x8d, 0x13, 0xfc, 0x8d, 0xa2, 0x54, 0x1a, 0x64, 0xbb, 0xc0, 0xdb, 0x02, 0x2d, 0x7c, 0x20,
0xb4, 0xcd, 0xf7, 0x43, 0xdb, 0xca, 0x43, 0xfb, 0x47, 0x03, 0x5a, 0xa7, 0x49, 0x38, 0xf7, 0x27,
0x5c, 0x4c, 0xc2, 0x68, 0xfa, 0x30, 0xa8, 0x12, 0x3e, 0x33, 0x0f, 0x5f, 0x0f, 0xac, 0xd1, 0xdb,
0x48, 0xf5, 0xdf, 0xcf, 0x68, 0xef, 0xd8, 0x88, 0x12, 0x47, 0x15, 0xf6, 0x25, 0x98, 0xa3, 0x88,
0x72, 0xb6, 0xd9, 0x3f, 0xc8, 0x14, 0xb5, 0x8e, 0x39, 0x8a, 0xdc, 0x1f, 0xc0, 0xa1, 0x7c, 0x88,
0x16, 0xa9, 0x81, 0x73, 0x08, 0x95, 0xf3, 0x28, 0x0a, 0xf5, 0xc8, 0x91, 0x04, 0x2e, 0xd6, 0xe9,
0x0c, 0xc3, 0x60, 0x7c, 0x4c, 0x4e, 0x6c, 0xfb, 0x35, 0xd9, 0x85, 0xe6, 0x55, 0x98, 0xfc, 0x3a,
0xf2, 0x13, 0x6a, 0x49, 0x72, 0x70, 0xe4, 0x59, 0xee, 0xd7, 0xf0, 0x69, 0xe9, 0xe6, 0x6c, 0x32,
0x62, 0x1a, 0x59, 0xd9, 0x2f, 0xb2, 0x31, 0x7c, 0x92, 0xaa, 0x8e, 0x86, 0x1f, 0xf5, 0xc6, 0x4d,
0xa3, 0xdf, 0xcf, 0x79, 0x4e, 0x46, 0xd5, 0xf5, 0x5b, 0xbc, 0x71, 0x07, 0xe0, 0x28, 0x34, 0xe5,
0x4f, 0x62, 0xf5, 0x82, 0x1b, 0x5f, 0xac, 0x1f, 0xfa, 0x25, 0x40, 0x6b, 0x85, 0x49, 0x3f, 0xa4,
0xe9, 0xdb, 0xfd, 0x93, 0x09, 0x87, 0xdb, 0x8c, 0x64, 0x09, 0x65, 0xe4, 0x12, 0x8a, 0xf5, 0xa1,
0xf2, 0xd6, 0x17, 0x6b, 0xbd, 0x0b, 0x1c, 0xe5, 0x82, 0xbd, 0xf1, 0x06, 0x2e, 0x55, 0xb1, 0x90,
0x4e, 0x27, 0x89, 0x1f, 0x2e, 0xf4, 0x66, 0x2b, 0x29, 0xbc, 0x61, 0x10, 0x84, 0x93, 0xdf, 0xc9,
0x1f, 0x65, 0x5c, 0x12, 0x5b, 0x0a, 0xa3, 0xf2, 0x81, 0x85, 0x51, 0xdd, 0x5a, 0x18, 0x3d, 0xd8,
0xfb, 0xd5, 0x72, 0xea, 0x25, 0xe2, 0xfc, 0xce, 0x8f, 0x13, 0xb1, 0x98, 0x08, 0xa7, 0x46, 0x1e,
0x95, 0xd9, 0xee, 0x1f, 0x0c, 0xd8, 0x55, 0x5e, 0x48, 0xd1, 0x03, 0xfb, 0x3b, 0x03, 0x1b, 0xdd,
0xd3, 0x0b, 0x21, 0xc1, 0x9d, 0xa2, 0x65, 0x11, 0xb6, 0x0a, 0xad, 0x7d, 0xb0, 0xc6, 0x22, 0x51,
0x7f, 0x5c, 0xe0, 0x27, 0xb6, 0x06, 0x12, 0xc9, 0x72, 0x8c, 0xd5, 0xfe, 0x57, 0xe0, 0xb9, 0xaf,
0xe1, 0xf3, 0x02, 0xa4, 0x54, 0x8d, 0x3a, 0x2c, 0xd9, 0xea, 0x68, 0x14, 0x56, 0xc7, 0xef, 0x41,
0xe5, 0x26, 0x17, 0x98, 0x03, 0x39, 0x2f, 0x73, 0xce, 0x70, 0x29, 0x77, 0xc7, 0x85, 0x79, 0x89,
0x3d, 0xf2, 0x74, 0x36, 0x8b, 0xc4, 0xcc, 0x4b, 0x74, 0xb2, 0x64, 0x0c, 0xf6, 0x15, 0x54, 0x49,
0x59, 0x9b, 0x2d, 0x2f, 0x40, 0x4a, 0x3a, 0xd8, 0xff, 0xfb, 0xbb, 0x8e, 0xf1, 0xcf, 0x77, 0x1d,
0xe3, 0xdf, 0xef, 0x3a, 0xc6, 0x9f, 0xff, 0xd3, 0xd9, 0xb9, 0xad, 0xd2, 0xff, 0x3e, 0x3f, 0xfa,
0x5f, 0x00, 0x00, 0x00, 0xff, 0xff, 0x48, 0x84, 0x6e, 0xe4, 0x07, 0x12, 0x00, 0x00,
}
func (m *Row) Marshal() (dAtA []byte, err error) {
@ -4813,6 +4953,122 @@ func (m *ImportRoaringRequest) MarshalToSizedBuffer(dAtA []byte) (int, error) {
return len(dAtA) - i, nil
}
func (m *RoaringUpdate) Marshal() (dAtA []byte, err error) {
size := m.Size()
dAtA = make([]byte, size)
n, err := m.MarshalToSizedBuffer(dAtA[:size])
if err != nil {
return nil, err
}
return dAtA[:n], nil
}
func (m *RoaringUpdate) MarshalTo(dAtA []byte) (int, error) {
size := m.Size()
return m.MarshalToSizedBuffer(dAtA[:size])
}
func (m *RoaringUpdate) MarshalToSizedBuffer(dAtA []byte) (int, error) {
i := len(dAtA)
_ = i
var l int
_ = l
if m.XXX_unrecognized != nil {
i -= len(m.XXX_unrecognized)
copy(dAtA[i:], m.XXX_unrecognized)
}
if m.ClearRecords {
i--
if m.ClearRecords {
dAtA[i] = 1
} else {
dAtA[i] = 0
}
i--
dAtA[i] = 0x28
}
if len(m.Set) > 0 {
i -= len(m.Set)
copy(dAtA[i:], m.Set)
i = encodeVarintPublic(dAtA, i, uint64(len(m.Set)))
i--
dAtA[i] = 0x22
}
if len(m.Clear) > 0 {
i -= len(m.Clear)
copy(dAtA[i:], m.Clear)
i = encodeVarintPublic(dAtA, i, uint64(len(m.Clear)))
i--
dAtA[i] = 0x1a
}
if len(m.View) > 0 {
i -= len(m.View)
copy(dAtA[i:], m.View)
i = encodeVarintPublic(dAtA, i, uint64(len(m.View)))
i--
dAtA[i] = 0x12
}
if len(m.Field) > 0 {
i -= len(m.Field)
copy(dAtA[i:], m.Field)
i = encodeVarintPublic(dAtA, i, uint64(len(m.Field)))
i--
dAtA[i] = 0xa
}
return len(dAtA) - i, nil
}
func (m *ImportRoaringShardRequest) Marshal() (dAtA []byte, err error) {
size := m.Size()
dAtA = make([]byte, size)
n, err := m.MarshalToSizedBuffer(dAtA[:size])
if err != nil {
return nil, err
}
return dAtA[:n], nil
}
func (m *ImportRoaringShardRequest) MarshalTo(dAtA []byte) (int, error) {
size := m.Size()
return m.MarshalToSizedBuffer(dAtA[:size])
}
func (m *ImportRoaringShardRequest) MarshalToSizedBuffer(dAtA []byte) (int, error) {
i := len(dAtA)
_ = i
var l int
_ = l
if m.XXX_unrecognized != nil {
i -= len(m.XXX_unrecognized)
copy(dAtA[i:], m.XXX_unrecognized)
}
if len(m.Views) > 0 {
for iNdEx := len(m.Views) - 1; iNdEx >= 0; iNdEx-- {
{
size, err := m.Views[iNdEx].MarshalToSizedBuffer(dAtA[:i])
if err != nil {
return 0, err
}
i -= size
i = encodeVarintPublic(dAtA, i, uint64(size))
}
i--
dAtA[i] = 0x12
}
}
if m.Remote {
i--
if m.Remote {
dAtA[i] = 1
} else {
dAtA[i] = 0
}
i--
dAtA[i] = 0x8
}
return len(dAtA) - i, nil
}
func (m *GroupCounts) Marshal() (dAtA []byte, err error) {
size := m.Size()
dAtA = make([]byte, size)
@ -5880,6 +6136,58 @@ func (m *ImportRoaringRequest) Size() (n int) {
return n
}
func (m *RoaringUpdate) Size() (n int) {
if m == nil {
return 0
}
var l int
_ = l
l = len(m.Field)
if l > 0 {
n += 1 + l + sovPublic(uint64(l))
}
l = len(m.View)
if l > 0 {
n += 1 + l + sovPublic(uint64(l))
}
l = len(m.Clear)
if l > 0 {
n += 1 + l + sovPublic(uint64(l))
}
l = len(m.Set)
if l > 0 {
n += 1 + l + sovPublic(uint64(l))
}
if m.ClearRecords {
n += 2
}
if m.XXX_unrecognized != nil {
n += len(m.XXX_unrecognized)
}
return n
}
func (m *ImportRoaringShardRequest) Size() (n int) {
if m == nil {
return 0
}
var l int
_ = l
if m.Remote {
n += 2
}
if len(m.Views) > 0 {
for _, e := range m.Views {
l = e.Size()
n += 1 + l + sovPublic(uint64(l))
}
}
if m.XXX_unrecognized != nil {
n += len(m.XXX_unrecognized)
}
return n
}
func (m *GroupCounts) Size() (n int) {
if m == nil {
return 0
@ -11756,6 +12064,314 @@ func (m *ImportRoaringRequest) Unmarshal(dAtA []byte) error {
}
return nil
}
func (m *RoaringUpdate) Unmarshal(dAtA []byte) error {
l := len(dAtA)
iNdEx := 0
for iNdEx < l {
preIndex := iNdEx
var wire uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPublic
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
wire |= uint64(b&0x7F) << shift
if b < 0x80 {
break
}
}
fieldNum := int32(wire >> 3)
wireType := int(wire & 0x7)
if wireType == 4 {
return fmt.Errorf("proto: RoaringUpdate: wiretype end group for non-group")
}
if fieldNum <= 0 {
return fmt.Errorf("proto: RoaringUpdate: illegal tag %d (wire type %d)", fieldNum, wire)
}
switch fieldNum {
case 1:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field Field", wireType)
}
var stringLen uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPublic
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
stringLen |= uint64(b&0x7F) << shift
if b < 0x80 {
break
}
}
intStringLen := int(stringLen)
if intStringLen < 0 {
return ErrInvalidLengthPublic
}
postIndex := iNdEx + intStringLen
if postIndex < 0 {
return ErrInvalidLengthPublic
}
if postIndex > l {
return io.ErrUnexpectedEOF
}
m.Field = string(dAtA[iNdEx:postIndex])
iNdEx = postIndex
case 2:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field View", wireType)
}
var stringLen uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPublic
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
stringLen |= uint64(b&0x7F) << shift
if b < 0x80 {
break
}
}
intStringLen := int(stringLen)
if intStringLen < 0 {
return ErrInvalidLengthPublic
}
postIndex := iNdEx + intStringLen
if postIndex < 0 {
return ErrInvalidLengthPublic
}
if postIndex > l {
return io.ErrUnexpectedEOF
}
m.View = string(dAtA[iNdEx:postIndex])
iNdEx = postIndex
case 3:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field Clear", wireType)
}
var byteLen int
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPublic
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
byteLen |= int(b&0x7F) << shift
if b < 0x80 {
break
}
}
if byteLen < 0 {
return ErrInvalidLengthPublic
}
postIndex := iNdEx + byteLen
if postIndex < 0 {
return ErrInvalidLengthPublic
}
if postIndex > l {
return io.ErrUnexpectedEOF
}
m.Clear = append(m.Clear[:0], dAtA[iNdEx:postIndex]...)
if m.Clear == nil {
m.Clear = []byte{}
}
iNdEx = postIndex
case 4:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field Set", wireType)
}
var byteLen int
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPublic
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
byteLen |= int(b&0x7F) << shift
if b < 0x80 {
break
}
}
if byteLen < 0 {
return ErrInvalidLengthPublic
}
postIndex := iNdEx + byteLen
if postIndex < 0 {
return ErrInvalidLengthPublic
}
if postIndex > l {
return io.ErrUnexpectedEOF
}
m.Set = append(m.Set[:0], dAtA[iNdEx:postIndex]...)
if m.Set == nil {
m.Set = []byte{}
}
iNdEx = postIndex
case 5:
if wireType != 0 {
return fmt.Errorf("proto: wrong wireType = %d for field ClearRecords", wireType)
}
var v int
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPublic
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
v |= int(b&0x7F) << shift
if b < 0x80 {
break
}
}
m.ClearRecords = bool(v != 0)
default:
iNdEx = preIndex
skippy, err := skipPublic(dAtA[iNdEx:])
if err != nil {
return err
}
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) > l {
return io.ErrUnexpectedEOF
}
m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
iNdEx += skippy
}
}
if iNdEx > l {
return io.ErrUnexpectedEOF
}
return nil
}
func (m *ImportRoaringShardRequest) Unmarshal(dAtA []byte) error {
l := len(dAtA)
iNdEx := 0
for iNdEx < l {
preIndex := iNdEx
var wire uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPublic
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
wire |= uint64(b&0x7F) << shift
if b < 0x80 {
break
}
}
fieldNum := int32(wire >> 3)
wireType := int(wire & 0x7)
if wireType == 4 {
return fmt.Errorf("proto: ImportRoaringShardRequest: wiretype end group for non-group")
}
if fieldNum <= 0 {
return fmt.Errorf("proto: ImportRoaringShardRequest: illegal tag %d (wire type %d)", fieldNum, wire)
}
switch fieldNum {
case 1:
if wireType != 0 {
return fmt.Errorf("proto: wrong wireType = %d for field Remote", wireType)
}
var v int
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPublic
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
v |= int(b&0x7F) << shift
if b < 0x80 {
break
}
}
m.Remote = bool(v != 0)
case 2:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field Views", wireType)
}
var msglen int
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPublic
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
msglen |= int(b&0x7F) << shift
if b < 0x80 {
break
}
}
if msglen < 0 {
return ErrInvalidLengthPublic
}
postIndex := iNdEx + msglen
if postIndex < 0 {
return ErrInvalidLengthPublic
}
if postIndex > l {
return io.ErrUnexpectedEOF
}
m.Views = append(m.Views, &RoaringUpdate{})
if err := m.Views[len(m.Views)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
return err
}
iNdEx = postIndex
default:
iNdEx = preIndex
skippy, err := skipPublic(dAtA[iNdEx:])
if err != nil {
return err
}
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) > l {
return io.ErrUnexpectedEOF
}
m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
iNdEx += skippy
}
}
if iNdEx > l {
return io.ErrUnexpectedEOF
}
return nil
}
func (m *GroupCounts) Unmarshal(dAtA []byte) error {
l := len(dAtA)
iNdEx := 0

View file

@ -239,6 +239,20 @@ message ImportRoaringRequest {
bool UpdateExistence = 7;
}
message RoaringUpdate {
string Field = 1;
string View = 2;
bytes Clear = 3;
bytes Set = 4;
bool ClearRecords = 5;
}
message ImportRoaringShardRequest {
bool Remote = 1;
repeated RoaringUpdate Views = 2;
}
message GroupCounts{
string Aggregate = 1;
repeated GroupCount Groups = 2;

View file

@ -2,9 +2,17 @@
set -e
for i in {41..44}; do
if [ -f /data/datagen_linux_arm64 ]; then
datagen_loc=/data/datagen_linux_arm64
else
datagen_loc=`which datagen`
fi
declare -i end=${2:-44} # end at 44 or whatever the second argument is
for (( c=41; c<=$end; c++ )); do
echo "!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!"
echo "ROUND $i"
echo "ROUND $c"
echo "!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!"
/data/datagen_linux_arm64 -s texas_health --pilosa.index thr --end-at 1048575 --pilosa.batch-size 1048576 --concurrency 1 --seed=$i --pilosa.hosts=$1
$datagen_loc -s texas_health --pilosa.index thr --end-at 1048575 --pilosa.batch-size 1048576 --concurrency 1 --seed=$c --pilosa.hosts=$1
done

View file

@ -1017,10 +1017,7 @@ func (c *Cursor) First() error {
}
}
// Last moves to the last element of the btree. Returns io.EOF if there are no
// elements. The first call to Prev() will not move the position but subsequent
// calls will move the position backward until it reaches the beginning and
// return io.EOF.
// Last moves to the last element of the btree.
func (c *Cursor) Last() error {
// c.stack.elems[0].pgno = c.root
c.buffered = true
@ -1058,6 +1055,7 @@ func (c *Cursor) Last() error {
// Seek moves to the specified container of the btree.
// If the container does not exist then it moves to the next container after the key.
// TODO: what happens if there are no more containers?!?!
func (c *Cursor) Seek(key uint64) (exact bool, err error) {
// c.stack.elems[0].pgno = c.bitmap.root
c.buffered = true
@ -1248,6 +1246,7 @@ func (se *stackElem) clear() {
se.key = 0
}
// TODO wtf does this do?
var _ = (&stackElem{}).clear
var _ = (&stackElem{}).String
var _ = (&stackElem{}).equal
@ -1454,7 +1453,7 @@ func (c *Cursor) difference(key uint64, data *roaring.Container) (bool, error) {
}
res := roaring.Difference(container, data)
if res == nil {
if res.N() == 0 {
return true, c.deleteLeafCell(cell.Key)
}

View file

@ -2,7 +2,6 @@
package rbf
import (
"errors"
"fmt"
"io"
"os"
@ -13,6 +12,8 @@ import (
"syscall"
"unsafe"
"github.com/pkg/errors"
"github.com/benbjohnson/immutable"
"github.com/molecula/featurebase/v3/logger"
rbfcfg "github.com/molecula/featurebase/v3/rbf/cfg"
@ -806,6 +807,18 @@ func (db *DB) Check() error {
return tx.Check()
}
// Viz writes a graphiz(dot) formatted visualisation of the RBF tree
// to w. At the time of writing this is very preliminary... feel free
// to hack on it and make changes.
func (db *DB) Viz(w io.Writer) error {
tx, err := db.Begin(false)
if err != nil {
return errors.Wrap(err, "beginning transaction")
}
defer tx.Rollback()
return tx.Viz(w)
}
// writeDBPage writes a page to the data file.
func (db *DB) writeDBPage(pgno uint32, page []byte) error {
_, err := db.file.WriteAt(page, int64(pgno)*PageSize)

View file

@ -43,6 +43,8 @@ const (
const maxBranchCellsPerPage = int((PageSize - branchPageHeaderSize) / (branchCellIndexElemSize + unsafe.Sizeof(branchCell{})))
type PageType uint32
// Page types.
const (
PageTypeRootRecord = 1
@ -52,6 +54,24 @@ const (
PageTypeBitmap = 16 // Only used internally when walking the b-tree
)
func (typ PageType) String() string {
switch typ {
case PageTypeRootRecord:
return "root-record"
case PageTypeLeaf:
return "leaf"
case PageTypeBranch:
return "branch"
case PageTypeBitmapHeader:
return "bitmap-header"
case PageTypeBitmap:
return "bitmap"
default:
return fmt.Sprintf("unknown<%d>", typ)
}
}
// Meta commit/rollback flags.
const (
MetaPageFlagCommit = 1

View file

@ -3,7 +3,6 @@ package rbf
import (
"bufio"
"errors"
"fmt"
"io"
"math"
@ -15,6 +14,7 @@ import (
"github.com/molecula/featurebase/v3/roaring"
txkey "github.com/molecula/featurebase/v3/short_txkey"
"github.com/molecula/featurebase/v3/vprint"
"github.com/pkg/errors"
)
var _ = txkey.ToString
@ -678,7 +678,7 @@ func (tx *Tx) Depth(name string) (int, error) {
}
defer c.Close()
if err := c.First(); err != nil {
if err := c.First(); err != nil { // TODO, EOF check?
return 0, err
}
return c.stack.top + 1, nil

131
rbf/viz.go Normal file
View file

@ -0,0 +1,131 @@
package rbf
import (
"fmt"
"io"
"github.com/pkg/errors"
)
func (tx *Tx) Viz(w io.Writer) error {
b := &builder{Writer: w}
b.start()
defer b.finish()
roots, err := tx.RootRecords()
if err != nil {
return errors.Wrap(err, "root records")
}
itr := roots.Iterator()
// we use page numbers for graphviz node IDs, but for cells we
// need something different so we start very high assuming it
// won't overlap
cid := &cellID{id: 1 << 48}
for name, pgno := itr.Next(); name != nil; name, pgno = itr.Next() {
err := tx.walkTree(pgno.(uint32), 0, func(pgno, parent, typ uint32, err error) error {
page, _, err := tx.readPage(pgno)
if err != nil {
return errors.Wrap(err, "reading page")
}
node := &vizNode{
pgno: pgno,
typ: PageType(typ).String(),
}
if parent == 0 {
node.name = name.(string)
}
node.cellN = readCellN(page)
switch typ {
case PageTypeBitmap:
return nil
case PageTypeLeaf:
numBitmaps := 0
for i, n := 0, readCellN(page); i < n; i++ {
if cell := readLeafCell(page, i); cell.Type == ContainerTypeBitmapPtr {
numBitmaps++
} else {
nodeid := cid.Next()
b.addCell(&cell, nodeid)
b.addEdge(int(pgno), nodeid)
}
}
node.numBitmaps = numBitmaps
}
b.addNode(node)
if parent != 0 {
b.addEdge(int(parent), int(pgno))
}
return nil
})
if err != nil {
return errors.Wrapf(err, "walking %s", name.(string))
}
}
return nil
}
type cellID struct {
id int
}
func (c *cellID) Next() int {
c.id++
return c.id
}
type vizNode struct {
name string
pgno uint32
typ string
cellN int
numBitmaps int
}
// builder wraps an io.Writer and understands how to compose DOT formatted elements.
type builder struct {
io.Writer
}
// start generates a title and initial node in DOT format.
func (b *builder) start() {
graphname := "unnamed"
fmt.Fprintln(b, `digraph "`+graphname+`" {`)
fmt.Fprintln(b, `node [style=filled fillcolor="#f8f8f8"]`)
}
// finish closes the opening curly bracket in the constructed DOT buffer.
func (b *builder) finish() {
fmt.Fprintln(b, "}")
}
// addNode generates a graph node in DOT format.
func (b *builder) addNode(node *vizNode) {
label := fmt.Sprintf("%s %s", node.name, node.typ)
if node.typ != "bitmap" {
label = label + fmt.Sprintf(" N=%d", node.cellN)
}
if node.numBitmaps > 0 {
label = label + fmt.Sprintf(" bitMapCells=%d", node.numBitmaps)
}
// Create DOT attribute for node.
attr := fmt.Sprintf(`label="%s" id="node%d" shape="rectangle"`,
label, node.pgno)
fmt.Fprintf(b, "N%d [%s]\n", node.pgno, attr)
}
// addEdge generates a graph edge in DOT format.
func (b *builder) addEdge(from, to int) {
fmt.Fprintf(b, "N%d -> N%d []\n", from, to)
}
func (b *builder) addCell(cell *leafCell, id int) {
label := fmt.Sprintf(`k%d type:%s elemn:%d bitn:%d`, cell.Key, cell.Type, cell.ElemN, cell.BitN)
fmt.Fprintf(b, `N%d [label="%s" id="node%d" shape="rectangle"]`, id, label, id)
}

View file

@ -2,8 +2,10 @@
package roaring
import (
"errors"
"fmt"
"math"
"github.com/pkg/errors"
"github.com/molecula/featurebase/v3/shardwidth"
)
@ -30,7 +32,8 @@ type FilterKey uint64
// key, or data. The values are represented as exclusive upper bounds on
// a series of matches followed by a series of rejections. So for instance,
// if called on key 23, the result {YesKey: 23, NoKey: 24} indicates that
// key 23 is a "no". This may seem confusing but it makes the math a lot
// key 23 is a "no". (TODO what about key 24, presumably that's a no as well?)
// This may seem confusing but it makes the math a lot
// easier to write. It can also report an error, which indicates that the
// entire operation should be stopped with that error.
type FilterResult struct {
@ -1137,3 +1140,209 @@ func NewBitmapBSICountFilter(filter *Bitmap) *BitmapBSICountFilter {
return b
}
// getNextFromIterator is a convenience function which calls Next and then Value
// on a ContainerIterator and changes the key to a FilterKey, and
// returns KEY_DONE if the iterator is done.
func getNextFromIterator(contIter ContainerIterator) (FilterKey, *Container) {
if !contIter.Next() {
return KEY_DONE, nil
}
key, val := contIter.Value()
return FilterKey(key), val
}
// NewRepeatedRowIteratorFromBytes interprets "data" as a roaring
// bitmap and returns a ContainerIterator which will repeatedly return
// the first "row" of data as determined by shard width, but with
// increasing keys. It is essentially a conveniece wrapper around
// NewContainerIterator and NewRepeatedRowContainerIterator. It treats
// empty "data" as valid and returns a no-op iterator.
func NewRepeatedRowIteratorFromBytes(data []byte) (ContainerIterator, error) {
iter, err := NewContainerIterator(data)
if err != nil {
return nil, errors.Wrap(err, "getting container iterator")
}
return NewRepeatedRowContainerIterator(iter), nil
}
// NewRepeatedRowContainerIterator returns a ContainerIterator which
// reads the first "row" of containers out of iter (as determined by
// shard width, so up to and including key 15 by default). It then
// returns a ContainerIterator which will repeatedly emit the
// containers in that row, but with increasing keys such that each
// time a particular container is emitted it has its key from the last
// time plus number of containers in a row (16 by default).
func NewRepeatedRowContainerIterator(iter ContainerIterator) *repeatedRowIterator {
conts := getFirstRowAsContainers(iter)
return &repeatedRowIterator{
containers: conts,
cur: -1,
}
}
type containerWithKey struct {
*Container
key FilterKey
}
type repeatedRowIterator struct {
containers []containerWithKey
row uint64
cur int
}
func (r *repeatedRowIterator) Next() bool {
r.cur++
if r.cur >= len(r.containers) {
r.cur = 0
r.row++
}
return len(r.containers) > 0
}
func (r *repeatedRowIterator) Value() (uint64, *Container) {
if len(r.containers) == 0 {
return 0, nil
}
return r.row*rowWidth + uint64(r.containers[r.cur].key%rowWidth), r.containers[r.cur].Container
}
func (r *repeatedRowIterator) Close() {}
func getFirstRowAsContainers(citer ContainerIterator) []containerWithKey {
citerContainers := make([]containerWithKey, 0)
for citer.Next() {
key, cont := citer.Value()
if key > 15 {
continue
}
citerContainers = append(citerContainers, containerWithKey{Container: cont, key: FilterKey(key)})
}
return citerContainers
}
// NewClearAndSetRewriter instantiates a ClearAndSetRewriter
func NewClearAndSetRewriter(clear, set ContainerIterator) (*ClearAndSetRewriter, error) {
curSetKey, curSet := getNextFromIterator(set)
curClearKey, curClear := getNextFromIterator(clear)
return &ClearAndSetRewriter{
curSetKey: curSetKey,
curSet: curSet,
curClearKey: curClearKey,
curClear: curClear,
clearIter: clear,
setIter: set,
}, nil
}
const KEY_DONE FilterKey = math.MaxUint64
// ClearAndSetRewriter is a BitmapRewriter which can operate on two
// ContainerIterators, clearing bits from one and setting bits from
// the other. It tries to do this pretty efficiently such that it
// doesn't look at the clear iterator unless there is actually a
// container that might need bits cleared, and it doesn't write a
// container unless bits have actually changed.
type ClearAndSetRewriter struct {
curSetKey FilterKey
curSet *Container
curClearKey FilterKey
curClear *Container
clearIter ContainerIterator
setIter ContainerIterator
}
func (csr *ClearAndSetRewriter) nextClear() (FilterKey, *Container) {
csr.curClearKey, csr.curClear = getNextFromIterator(csr.clearIter)
return csr.curClearKey, csr.curClear
}
func (csr *ClearAndSetRewriter) nextSet() (FilterKey, *Container) {
csr.curSetKey, csr.curSet = getNextFromIterator(csr.setIter)
return csr.curSetKey, csr.curSet
}
func (csr *ClearAndSetRewriter) lowestKey() FilterKey {
if csr.curSetKey < csr.curClearKey {
return csr.curSetKey
}
return csr.curClearKey
}
func (csr *ClearAndSetRewriter) ConsiderKey(key FilterKey, n int32) FilterResult {
if key < csr.lowestKey() {
return key.RejectUntil(csr.lowestKey())
}
return key.NeedData()
}
// writeCurrent writes the current clear and set if they match the
// key. It takes some care to try to determine if any bits actually
// changed to avoid unnecessary writes.
func (csr *ClearAndSetRewriter) writeCurrent(key FilterKey, data *Container, writeback ContainerWriteback) error {
if key == KEY_DONE {
return nil
}
changed := false
startN := data.N()
if csr.curClearKey == key && csr.curSetKey == key {
actualClear := csr.curClear.Difference(csr.curSet) // don't need to clear bits that we're going to set
data = data.DifferenceInPlace(actualClear)
clearedN := data.N()
changed = clearedN != startN
data = data.UnionInPlace(csr.curSet)
data.Repair()
setN := data.N()
changed = changed || clearedN != setN
} else if csr.curClearKey == key {
data = data.DifferenceInPlace(csr.curClear)
clearedN := data.N()
changed = clearedN != startN
} else if csr.curSetKey == key {
data = data.UnionInPlace(csr.curSet)
data.Repair()
setN := data.N()
changed = startN != setN
}
if changed {
if err := writeback(key, data); err != nil {
return err
}
}
return nil
}
func (csr *ClearAndSetRewriter) RewriteData(key FilterKey, data *Container, writeback ContainerWriteback) FilterResult {
if data == nil {
key = KEY_DONE
}
// when we're called with a container, we need to process all the
// sets up to that container that we haven't done yet. Then we
// need to do any clears on that container, then any sets on that
// container. Then we can reject until the lowest next container.
// When we enter this function, curSet and curClear are the next
// things we need to process. When we leave this function they
// must be set to the next things we need to process.
for ; csr.curSetKey < key; csr.nextSet() {
if err := writeback(csr.curSetKey, csr.curSet); err != nil {
return key.Fail(errors.Wrapf(err, "writing set container at %d", csr.curSetKey))
}
}
// fast forward clears to this key
for ; csr.curClearKey < key && key != KEY_DONE; csr.nextClear() {
}
err := csr.writeCurrent(key, data, writeback)
if err != nil {
return key.Fail(errors.Wrapf(err, "writing current key %d", key))
}
if csr.curSetKey == key {
csr.nextSet()
}
if csr.curClearKey == key {
csr.nextClear()
}
return key.RejectUntil(csr.lowestKey())
}

View file

@ -382,3 +382,99 @@ func TestMutexDupFilter(t *testing.T) {
})
}
}
const (
shardWidth = 1 << shardwidth.Exponent
)
func TestGetNextFromIteratorEmpty(t *testing.T) {
data := NewBitmap().Roaring()
cit, err := NewContainerIterator(data)
if err != nil {
t.Fatalf("getting roaring iterator: %v", err)
}
key, cont := getNextFromIterator(cit)
if key != KEY_DONE || cont != nil {
t.Fatalf("expected iterator done, but got: %d, %v", key, cont.Slice())
}
}
func TestGetNextFromIterator(t *testing.T) {
data := NewBitmap(1, 7, 100000, 500000, shardWidth, shardWidth+3, shardWidth+containerWidth-1, shardWidth+containerWidth).Roaring()
cit, err := NewContainerIterator(data)
if err != nil {
t.Fatalf("getting roaring iterator: %v", err)
}
expected := []struct {
key FilterKey
slice []uint16
}{
{0, []uint16{1, 7}},
{1, []uint16{100000 % containerWidth}},
{500000 / containerWidth, []uint16{500000 % containerWidth}},
{shardWidth / containerWidth, []uint16{0, 3, 65535}},
{shardWidth/containerWidth + 1, []uint16{0}},
}
for i, exp := range expected {
key, cont := getNextFromIterator(cit)
if key != exp.key {
t.Errorf("key mismatch at %d, got: %d, exp: %d", i, key, exp.key)
}
if !reflect.DeepEqual(cont.Slice(), exp.slice) {
t.Fatalf("data misatch at %d, got: %v, exp: %v", i, cont.Slice(), exp.slice)
}
}
key, cont := getNextFromIterator(cit)
if key != KEY_DONE || cont != nil {
t.Fatalf("expected iterator done, but got: %d, %v", key, cont.Slice())
}
}
func TestClearColumnsAndSetRewriter(t *testing.T) {
data := NewBitmap(1, 7, 100000, 500000,
shardWidth, shardWidth+1, shardWidth+7, shardWidth+containerWidth-1, shardWidth+containerWidth,
shardWidth*2+7, shardWidth*2+9).Roaring()
clearIter, err := NewRepeatedRowIteratorFromBytes(data)
if err != nil {
t.Fatalf("getting clear iterator: %v", err)
}
setIter, err := NewContainerIterator(data)
if err != nil {
t.Fatalf("getting set iterator: %v", err)
}
rewriter, err := NewClearAndSetRewriter(clearIter, setIter)
if err != nil {
t.Fatalf("getting rewriter: %v", err)
}
expClearKeys := []FilterKey{
0, 1, 500000 / containerWidth,
shardWidth / containerWidth, shardWidth/containerWidth + 1, (shardWidth + 500000) / containerWidth,
shardWidth * 2 / containerWidth, shardWidth*2/containerWidth + 1, (shardWidth*2 + 500000) / containerWidth,
}
for i, key := range expClearKeys {
if rewriter.curClearKey != key {
t.Errorf("unexpected clear key at %d, exp: %d, got: %d", i, key, rewriter.curClearKey)
}
rewriter.nextClear()
}
expSetKeys := []FilterKey{
0, 1, 500000 / containerWidth,
shardWidth / containerWidth, shardWidth/containerWidth + 1,
shardWidth * 2 / containerWidth, KEY_DONE,
}
for i, key := range expSetKeys {
if rewriter.curSetKey != key {
t.Errorf("unexpected clear key at %d, exp: %d, got: %d", i, key, rewriter.curClearKey)
}
rewriter.nextSet()
}
}

View file

@ -3,6 +3,7 @@
package roaring
import (
"bytes"
"encoding/binary"
"fmt"
"hash/fnv"
@ -164,6 +165,67 @@ type ContainerIterator interface {
Close()
}
type nopContainerIterator struct{}
func (n nopContainerIterator) Next() bool { return false }
func (n nopContainerIterator) Value() (uint64, *Container) { return 0, nil }
func (n nopContainerIterator) Close() {}
type unionContainerIterator struct {
iters []ContainerIterator
curs []containerWithKey
cur FilterKey
}
func NewUnionContainerIterator(iters ...ContainerIterator) ContainerIterator {
return &unionContainerIterator{
iters: iters,
curs: make([]containerWithKey, len(iters)),
cur: 0,
}
}
func (u *unionContainerIterator) Next() bool {
if u.cur == KEY_DONE {
return false
}
// Next all iters that are at cur and save lowest
lowest := uint64(KEY_DONE)
for i, iter := range u.iters {
if u.curs[i].key == u.cur {
if iter.Next() {
key, c := iter.Value()
u.curs[i].key, u.curs[i].Container = FilterKey(key), c
if key < lowest {
lowest = key
}
} else {
u.curs[i].key = KEY_DONE
u.curs[i].Container = nil
}
}
}
u.cur = FilterKey(lowest)
return u.cur != KEY_DONE
}
func (u *unionContainerIterator) Value() (uint64, *Container) {
if u.cur == KEY_DONE {
return uint64(KEY_DONE), nil
}
var ret *Container
for _, cur := range u.curs {
if cur.key == u.cur {
ret = ret.UnionInPlace(cur.Container)
}
}
ret.Repair()
return uint64(u.cur), ret
}
func (u *unionContainerIterator) Close() {}
// Bitmap represents a roaring bitmap.
type Bitmap struct {
Containers Containers
@ -1715,6 +1777,43 @@ func (b *Bitmap) writeToUnoptimized(w io.Writer) (n int64, err error) {
return n, nil
}
// NewContainerIterator takes a byte slice which is either standard
// roaring or pilosa roaring and returns a ContainerIterator.
func NewContainerIterator(data []byte) (ContainerIterator, error) {
if len(data) == 0 {
return nopContainerIterator{}, nil
}
ri, err := NewRoaringIterator(data)
if err != nil {
return nil, errors.Wrap(err, "getting roaring iterator")
}
return &containerIteratorRoaringIteratorWrapper{
r: ri,
}, nil
}
// containerIteratorRoaringIteratorWrapper wraps a RoaringIterator to
// make it play like a ContainerIterator.
type containerIteratorRoaringIteratorWrapper struct {
r RoaringIterator
nextKey uint64
nextCont *Container
}
func (c *containerIteratorRoaringIteratorWrapper) Next() bool {
c.nextKey, c.nextCont = c.r.NextContainer()
if c.nextCont == nil {
return false
}
return true
}
func (c *containerIteratorRoaringIteratorWrapper) Value() (uint64, *Container) {
return c.nextKey, c.nextCont
}
func (c *containerIteratorRoaringIteratorWrapper) Close() {}
// RoaringIterator represents something which can iterate through a roaring
// bitmap and yield information about containers, including type, size, and
// the location of their data structures.
@ -1732,6 +1831,7 @@ type RoaringIterator interface {
// allocate a Container from the output of its internal call to Next(),
// and return the key and container rc. If Next returns an error, then
// NextContainer will return 0, nil.
// TODO: have this reuse the *Container?
NextContainer() (key uint64, rc *Container)
// Data returns the underlying data, esp for the Ops log.
@ -7425,11 +7525,23 @@ func differenceRunRunInPlace(c, other *Container) *Container {
return c
}
// Roaring encodes the bitmap in the Pilosa roaring
// format. Convenience wrapper around WriteTo.
func (b *Bitmap) Roaring() []byte {
buf := &bytes.Buffer{}
_, err := b.WriteTo(buf)
if err != nil {
panic(err) // I don't believe this can happen when writing to a bytes.Buffer
}
return buf.Bytes()
}
//RBF exports to be reconsidered as we progress
func (b *Bitmap) Put(key uint64, c *Container) {
b.Containers.Put(key, c)
}
func AsBitmap(c *Container) []uint64 {
return c.bitmap()
}
@ -7504,6 +7616,7 @@ func (c *Container) CountRange(start, end int32) (n int32) {
// c or other. It may, or may not, modify c. The resulting container's
// count, as returned by c.N(), may be incorrect; see (*Container).Repair().
// Do not freeze a container produced by this operation before repairing it.
// TODO(jaffee): why don't we just call Repair in here?!?!
func (c *Container) UnionInPlace(other *Container) (r *Container) {
return c.unionInPlace(other)
}

View file

@ -5,7 +5,7 @@ import "sync"
///////////////////////////////////////////////////////////////////////////
var containerWidth uint64 = 65536
const containerWidth = 1 << 16
////////////////// array
func arrayEmpty() []uint16 {

View file

@ -2845,19 +2845,19 @@ func TestBitmapClone(t *testing.T) {
// sets 100 bits per true
func rleCont(num int, left, mid, right bool) []uint64 {
ret := make([]uint64, 0)
base := containerWidth * uint64(num)
base := uint64(containerWidth * uint64(num))
if left {
for i := uint64(0); i < 100; i++ {
ret = append(ret, base+i)
}
}
if mid {
for i := containerWidth / 2; i < containerWidth/2+100; i++ {
for i := uint64(containerWidth / 2); i < containerWidth/2+100; i++ {
ret = append(ret, base+i)
}
}
if right {
for i := containerWidth - 100; i < containerWidth; i++ {
for i := uint64(containerWidth - 100); i < containerWidth; i++ {
ret = append(ret, base+i)
}
}
@ -2872,7 +2872,7 @@ func arrCont(num int, left, mid, right bool) []uint64 {
ret = append(ret, base+0, base+2)
}
if mid {
half := containerWidth / 2
half := uint64(containerWidth / 2)
ret = append(ret, base+half, base+half+2)
}
if right {

View file

@ -11,7 +11,7 @@ import (
"testing/quick"
"time"
"github.com/molecula/featurebase/v3"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/generator"
"github.com/molecula/featurebase/v3/roaring"
_ "github.com/molecula/featurebase/v3/test"
@ -2295,3 +2295,83 @@ func TestContainer_UnionInPlace_TwoBigArrays(t *testing.T) {
panic("should be NOT be an array now")
}
}
func TestContainerIteratorRoaringIteratorWrapper(t *testing.T) {
data := roaring.NewBitmap(1, 7, 100000, 500000).Roaring()
cit, err := roaring.NewContainerIterator(data)
if err != nil {
t.Fatalf("getting roaring iterator: %v", err)
}
// First Container (1, 7)
if !cit.Next() {
t.Fail()
}
if k, v := cit.Value(); k != 0 || !reflect.DeepEqual(v.Slice(), []uint16{1, 7}) {
t.Fail()
}
// Second Container (100,000)
if !cit.Next() {
t.Fail()
}
if k, v := cit.Value(); k != 1 || !reflect.DeepEqual(v.Slice(), []uint16{100000 % (1 << 16)}) {
t.Fail()
}
// Third Container (500,000)
if !cit.Next() {
t.Fail()
}
if k, v := cit.Value(); k != 500000/(1<<16) || !reflect.DeepEqual(v.Slice(), []uint16{500000 % (1 << 16)}) {
t.Fail()
}
// Next should now return false
if cit.Next() {
t.Fail()
}
}
func TestUnionContainerIterator(t *testing.T) {
data1 := roaring.NewBitmap(1, 7, 100000, 500000).Roaring()
d1iter, err := roaring.NewContainerIterator(data1)
if err != nil {
t.Fatalf("getting iterator: %v", err)
}
data2 := roaring.NewBitmap(1, 2, 100000, 200000, 500000, 700000).Roaring()
d2iter, err := roaring.NewContainerIterator(data2)
if err != nil {
t.Fatalf("getting iterator: %v", err)
}
ci := roaring.NewUnionContainerIterator(d1iter, d2iter)
exps := []struct {
key uint64
vals []uint16
}{
{0, []uint16{1, 2, 7}},
{1, []uint16{100000 % (1 << 16)}},
{3, []uint16{200000 % (1 << 16)}},
{7, []uint16{500000 % (1 << 16)}},
{10, []uint16{700000 % (1 << 16)}},
}
for i, exp := range exps {
if !ci.Next() {
t.Fatalf("Next unexpected false at %d", i)
}
k, c := ci.Value()
if exp.key != k {
t.Errorf("key mismatch at %d, exp: %d, got: %d", i, exp.key, k)
}
if !reflect.DeepEqual(exp.vals, c.Slice()) {
t.Errorf("data mismatch at %d, exp: %v, got: %v", i, exp.vals, c.Slice())
}
}
if ci.Next() {
k, c := ci.Value()
t.Fatalf("unexpected data at the end: %d, %v", k, c)
}
}