From ed1890b3a2211ad991a77b68f5de3f50f92822a1 Mon Sep 17 00:00:00 2001 From: Todd Gruben Date: Wed, 24 Feb 2016 14:00:41 -0600 Subject: [PATCH] adjust SetBit/ClearBit to releably report bit change --- executor.go | 7 ++-- executor_test.go | 23 +++++++++--- fragment.go | 19 +++++++--- handler.go | 2 ++ internal/internal.pb.go | 62 ++++++++++++++++++-------------- internal/internal.proto | 1 + roaring/roaring.go | 79 +++++++++++++++++++++++------------------ 7 files changed, 120 insertions(+), 73 deletions(-) diff --git a/executor.go b/executor.go index 3e419c949..caa816af1 100644 --- a/executor.go +++ b/executor.go @@ -336,10 +336,11 @@ func (e *Executor) executeSetBit(db string, c *pql.SetBit) (bool, error) { } // Forward call to remote node otherwise. - if _, err := e.exec(node, db, &pql.Query{Root: c}, nil); err != nil { + if res, err := e.exec(node, db, &pql.Query{Root: c}, nil); err != nil { return false, err + } else { + ret = res.(bool) } - fmt.Println("NEED TO IMPLEMENT REMOTE SETBIT") } return ret, nil } @@ -444,7 +445,7 @@ func (e *Executor) exec(node *Node, db string, q *pql.Query, slices []uint64) (r case *pql.Count: return pb.GetN(), nil case *pql.SetBit: - return nil, nil + return pb.GetChanged(), nil default: panic(fmt.Sprintf("invalid node for remote exec: %T", q.Root)) } diff --git a/executor_test.go b/executor_test.go index 703e16bc1..70e780c80 100644 --- a/executor_test.go +++ b/executor_test.go @@ -123,14 +123,29 @@ func TestExecutor_Execute_SetBit(t *testing.T) { defer idx.Close() e := NewExecutor(idx.Index, NewCluster(1)) - if _, err := e.Execute("d", MustParse(`SetBit(id=10, frame=f, profileID=1)`), nil); err != nil { - t.Fatal(err) + f := idx.MustCreateFragmentIfNotExists("d", "f", 0) + if n := f.Bitmap(11).Count(); n != 0 { + t.Fatalf("unexpected bitmap count: %d", n) } - f := idx.MustCreateFragmentIfNotExists("d", "f", 0) - if n := f.Bitmap(10).Count(); n != 1 { + if res, err := e.Execute("d", MustParse(`SetBit(id=11, frame=f, profileID=1)`), nil); err != nil { + t.Fatal(err) + } else { + if !res.(bool) { + t.Fatalf("expected bit changed") + } + } + + if n := f.Bitmap(11).Count(); n != 1 { t.Fatalf("unexpected bitmap count: %d", n) } + if res, err := e.Execute("d", MustParse(`SetBit(id=11, frame=f, profileID=1)`), nil); err != nil { + t.Fatal(err) + } else { + if res.(bool) { + t.Fatalf("expected bit unchanged") + } + } } // Ensure a SetBitmapAttrs() query can be executed. diff --git a/fragment.go b/fragment.go index 176585ddf..8d60c0004 100644 --- a/fragment.go +++ b/fragment.go @@ -320,18 +320,23 @@ func (f *Fragment) SetBit(bitmapID, profileID uint64) (changed bool, err error) func (f *Fragment) setBit(bitmapID, profileID uint64) (bool, error) { // Determine the position of the bit in the storage. + ret := false pos, err := f.pos(bitmapID, profileID) if err != nil { return false, err } // Write to storage. - if err := f.storage.Add(pos); err != nil { + + if ret, err = f.storage.Add(pos); err != nil { return false, err } // Update the cache. - return f.bitmap(bitmapID).setBit(profileID), nil + if f.bitmap(bitmapID).setBit(profileID) { + ret = true + } + return ret, nil } @@ -348,12 +353,16 @@ func (f *Fragment) ClearBit(bitmapID, profileID uint64) (bool, error) { } // Write to storage. - if err := f.storage.Remove(pos); err != nil { + changed, err := f.storage.Remove(pos) + if err != nil { return false, err } // Update the cache. - return f.bitmap(bitmapID).clearBit(profileID), nil + if f.bitmap(bitmapID).clearBit(profileID) { + return true, nil + } + return changed, nil } @@ -509,7 +518,7 @@ func (f *Fragment) Import(bitmapIDs, profileIDs []uint64) error { } // Write to storage. - if err := f.storage.Add(pos); err != nil { + if _, err := f.storage.Add(pos); err != nil { return err } } diff --git a/handler.go b/handler.go index c6927a14e..a1e65878f 100644 --- a/handler.go +++ b/handler.go @@ -404,6 +404,8 @@ func encodeQueryResponse(resp *QueryResponse) *internal.QueryResponse { pb.Pairs = encodePairs(result) case uint64: pb.N = proto.Uint64(result) + case bool: + pb.Changed = proto.Bool(result) default: panic(fmt.Sprintf("invalid query result type: %T", resp.Result)) } diff --git a/internal/internal.pb.go b/internal/internal.pb.go index ef49d0d53..82d20783f 100644 --- a/internal/internal.pb.go +++ b/internal/internal.pb.go @@ -25,6 +25,7 @@ It has these top-level messages: package internal import proto "github.com/gogo/protobuf/proto" + import fmt "fmt" import math "math" @@ -263,6 +264,7 @@ type QueryResponse struct { N *uint64 `protobuf:"varint,3,opt,name=N" json:"N,omitempty"` Pairs []*Pair `protobuf:"bytes,4,rep,name=Pairs" json:"Pairs,omitempty"` Profiles []*Profile `protobuf:"bytes,5,rep,name=Profiles" json:"Profiles,omitempty"` + Changed *bool `protobuf:"varint,6,opt,name=Changed" json:"Changed,omitempty"` XXX_unrecognized []byte `json:"-"` } @@ -306,6 +308,13 @@ func (m *QueryResponse) GetProfiles() []*Profile { return nil } +func (m *QueryResponse) GetChanged() bool { + if m != nil && m.Changed != nil { + return *m.Changed + } + return false +} + type ImportRequest struct { DB *string `protobuf:"bytes,1,req,name=DB" json:"DB,omitempty"` Frame *string `protobuf:"bytes,2,req,name=Frame" json:"Frame,omitempty"` @@ -405,30 +414,31 @@ func init() { } var fileDescriptor0 = []byte{ - // 398 bytes of a gzipped FileDescriptorProto - 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x09, 0x6e, 0x88, 0x02, 0xff, 0x84, 0x52, 0x4d, 0x8f, 0xda, 0x30, - 0x14, 0x54, 0x88, 0x03, 0xe4, 0xa5, 0xa4, 0xe0, 0x5e, 0x50, 0x25, 0x54, 0x64, 0x2e, 0xa8, 0x07, - 0x0e, 0xa8, 0x7f, 0xa0, 0x40, 0xab, 0x22, 0x54, 0x44, 0x8b, 0xda, 0x73, 0x23, 0xe4, 0x96, 0xa8, - 0x21, 0xce, 0x3a, 0xce, 0x81, 0x1f, 0xb1, 0xff, 0x79, 0x9f, 0x3f, 0x12, 0xd8, 0x5d, 0x56, 0x7b, - 0x8a, 0x3c, 0x1e, 0xbf, 0x99, 0x79, 0x13, 0x88, 0xd3, 0x5c, 0x71, 0x99, 0x27, 0xd9, 0xac, 0x90, - 0x42, 0x09, 0xda, 0xad, 0xcf, 0xec, 0x1b, 0xb4, 0x17, 0xa9, 0x3a, 0x25, 0x05, 0xfd, 0x00, 0xed, - 0xe5, 0xb1, 0xca, 0xff, 0x97, 0x43, 0x6f, 0xec, 0x4f, 0xa3, 0xf9, 0xdb, 0x59, 0xf3, 0xc8, 0xe0, - 0x74, 0x04, 0xc1, 0x67, 0xa5, 0x64, 0x39, 0x6c, 0x99, 0xfb, 0xf8, 0x72, 0xaf, 0x61, 0x36, 0x81, - 0xc0, 0xf2, 0x22, 0xf0, 0x37, 0xfc, 0x8c, 0x53, 0x5a, 0x53, 0x42, 0x7b, 0x10, 0xfc, 0x4e, 0xb2, - 0x8a, 0x9b, 0x47, 0x84, 0x31, 0x20, 0xbb, 0x24, 0x95, 0xcf, 0x38, 0x4b, 0x51, 0xe5, 0x0a, 0x39, - 0x78, 0x64, 0x1f, 0xc1, 0x47, 0x4b, 0xb4, 0x0f, 0x5d, 0xeb, 0x6c, 0xbd, 0x72, 0xbc, 0x01, 0x84, - 0x3b, 0x29, 0xfe, 0xa6, 0x19, 0x47, 0xc8, 0x72, 0x3f, 0x41, 0xc7, 0x41, 0x14, 0xa0, 0xd5, 0x30, - 0x5f, 0xb1, 0xba, 0x05, 0xa2, 0xbf, 0xd7, 0x2e, 0x42, 0xfa, 0x0e, 0xa2, 0xbd, 0x92, 0x69, 0xfe, - 0xaf, 0xf6, 0xeb, 0x21, 0x88, 0x92, 0xbf, 0xf0, 0xad, 0x85, 0x7c, 0x84, 0x8c, 0x8b, 0x85, 0x10, - 0x99, 0x85, 0x08, 0x42, 0x5d, 0x36, 0x85, 0x8e, 0x9e, 0xf7, 0x1d, 0xb7, 0xd8, 0x28, 0x7b, 0x37, - 0x95, 0x37, 0xf0, 0xe6, 0x47, 0xc5, 0xe5, 0xf9, 0x27, 0xbf, 0xab, 0x78, 0xa9, 0xb4, 0xe9, 0xd5, - 0xc2, 0x19, 0xc0, 0x35, 0x98, 0x3b, 0x13, 0x2d, 0xa4, 0x31, 0xb4, 0xf7, 0x59, 0x7a, 0xe0, 0x25, - 0xea, 0xe2, 0xea, 0xf4, 0x3e, 0x5c, 0xd4, 0xd2, 0xc9, 0xde, 0x7b, 0xd0, 0x73, 0xd3, 0xca, 0x42, - 0xe4, 0x25, 0xd7, 0x81, 0xbe, 0x48, 0x89, 0xf3, 0xb4, 0xf7, 0x71, 0x5d, 0xad, 0xc9, 0x12, 0xcd, - 0xfb, 0x17, 0x2f, 0xae, 0xf2, 0x10, 0xbc, 0xad, 0x4b, 0x85, 0xbe, 0x75, 0x31, 0x7a, 0xf4, 0x13, - 0xdf, 0xa6, 0xaf, 0xc9, 0x95, 0x78, 0x60, 0x18, 0x83, 0x2b, 0x86, 0xbd, 0x61, 0x7f, 0xa0, 0xb7, - 0x3e, 0x15, 0x42, 0xaa, 0x17, 0xd2, 0x7d, 0x95, 0xc9, 0x89, 0xbb, 0x74, 0x78, 0x34, 0xe9, 0x50, - 0xde, 0x55, 0x5b, 0x97, 0x6d, 0x2d, 0x10, 0x8a, 0xaf, 0x9b, 0xb6, 0xad, 0x28, 0x61, 0x23, 0x88, - 0x6b, 0x85, 0x1b, 0x89, 0xd9, 0x7b, 0xfc, 0x91, 0x92, 0xc3, 0x91, 0x3f, 0x1e, 0xa7, 0x9b, 0x20, - 0x0f, 0x01, 0x00, 0x00, 0xff, 0xff, 0x4e, 0xbb, 0x74, 0xfe, 0x03, 0x03, 0x00, 0x00, + // 410 bytes of a gzipped FileDescriptorProto + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x09, 0x6e, 0x88, 0x02, 0xff, 0x84, 0x92, 0xc1, 0x8e, 0xda, 0x30, + 0x10, 0x86, 0x15, 0xe2, 0x04, 0x32, 0x29, 0x01, 0xdc, 0x0b, 0xaa, 0x84, 0x8a, 0xcc, 0x05, 0xf5, + 0xc0, 0x01, 0xf5, 0x05, 0x0a, 0xb4, 0x2a, 0x42, 0x45, 0xb4, 0xa8, 0x3d, 0x37, 0xa2, 0x2e, 0x44, + 0x0d, 0x71, 0xea, 0x38, 0x07, 0x5e, 0xa6, 0xcf, 0xda, 0xb1, 0xe3, 0x04, 0x76, 0x97, 0xd5, 0x9e, + 0x22, 0xff, 0x1e, 0xcf, 0x7c, 0xff, 0xfc, 0x81, 0x28, 0xc9, 0x14, 0x97, 0x59, 0x9c, 0xce, 0x72, + 0x29, 0x94, 0xa0, 0x9d, 0xfa, 0xcc, 0x3e, 0x83, 0xbf, 0x48, 0xd4, 0x39, 0xce, 0xe9, 0x5b, 0xf0, + 0x97, 0xa7, 0x32, 0xfb, 0x53, 0x0c, 0x9d, 0xb1, 0x3b, 0x0d, 0xe7, 0xbd, 0x59, 0xf3, 0xc8, 0xe8, + 0x74, 0x04, 0xde, 0x07, 0xa5, 0x64, 0x31, 0x6c, 0x99, 0xfb, 0xe8, 0x7a, 0xaf, 0x65, 0x36, 0x01, + 0xaf, 0xaa, 0x0b, 0xc1, 0xdd, 0xf0, 0x0b, 0x76, 0x69, 0x4d, 0x09, 0xed, 0x82, 0xf7, 0x23, 0x4e, + 0x4b, 0x6e, 0x1e, 0x11, 0xc6, 0x80, 0xec, 0xe2, 0x44, 0x3e, 0xa9, 0x59, 0x8a, 0x32, 0x53, 0x58, + 0x83, 0x47, 0xf6, 0x0e, 0x5c, 0x44, 0xa2, 0x7d, 0xe8, 0x54, 0x64, 0xeb, 0x95, 0xad, 0x1b, 0x40, + 0xb0, 0x93, 0xe2, 0x77, 0x92, 0x72, 0x94, 0xaa, 0xda, 0xf7, 0xd0, 0xb6, 0x12, 0x05, 0x68, 0x35, + 0x95, 0x2f, 0xa0, 0x6e, 0x81, 0xe8, 0xef, 0x2d, 0x45, 0x40, 0x5f, 0x43, 0xb8, 0x57, 0x32, 0xc9, + 0x8e, 0x35, 0xaf, 0x83, 0x22, 0x8e, 0xfc, 0x8e, 0x6f, 0x2b, 0xc9, 0x45, 0xc9, 0x50, 0x2c, 0x84, + 0x48, 0x2b, 0x89, 0xa0, 0xd4, 0x61, 0x53, 0x68, 0xeb, 0x7e, 0x5f, 0x70, 0x8b, 0xcd, 0x64, 0xe7, + 0xee, 0xe4, 0x0d, 0xbc, 0xfa, 0x5a, 0x72, 0x79, 0xf9, 0xc6, 0xff, 0x96, 0xbc, 0x50, 0x1a, 0x7a, + 0xb5, 0xb0, 0x00, 0xb8, 0x06, 0x73, 0x67, 0xac, 0x05, 0x34, 0x02, 0x7f, 0x9f, 0x26, 0x07, 0x5e, + 0xe0, 0x5c, 0x5c, 0x9d, 0xde, 0x87, 0xb5, 0x5a, 0xd8, 0xb1, 0xff, 0x1c, 0xe8, 0xda, 0x6e, 0x45, + 0x2e, 0xb2, 0x82, 0x6b, 0x43, 0x1f, 0xa5, 0xc4, 0x7e, 0x9a, 0x7d, 0x5c, 0x47, 0x6b, 0xbc, 0x84, + 0xf3, 0xfe, 0x95, 0xc5, 0x46, 0x1e, 0x80, 0xb3, 0xb5, 0xae, 0x90, 0x5b, 0x07, 0xa3, 0x5b, 0x3f, + 0xe2, 0x36, 0x79, 0x4d, 0x6e, 0x86, 0x7b, 0xa6, 0x62, 0x70, 0x53, 0x61, 0x13, 0xe8, 0x41, 0x7b, + 0x79, 0x8a, 0xb3, 0x23, 0xff, 0x35, 0xf4, 0x0d, 0xe0, 0x4f, 0xe8, 0xae, 0xcf, 0xb9, 0x90, 0xea, + 0x19, 0xbb, 0x9f, 0x64, 0x7c, 0xe6, 0xd6, 0x2e, 0x1e, 0x8d, 0x5d, 0xe4, 0xb1, 0x59, 0xd7, 0xe9, + 0x57, 0x4c, 0x84, 0xe2, 0xeb, 0x26, 0xfe, 0x8a, 0x82, 0xb0, 0x11, 0x44, 0xf5, 0x84, 0x3b, 0x2b, + 0x60, 0x6f, 0xf0, 0xcf, 0x8a, 0x0f, 0x27, 0xfe, 0xb0, 0x9d, 0x8e, 0x86, 0xfc, 0x0f, 0x00, 0x00, + 0xff, 0xff, 0x98, 0x36, 0x6d, 0xac, 0x14, 0x03, 0x00, 0x00, } diff --git a/internal/internal.proto b/internal/internal.proto index 3fe5b6b83..91ebb3128 100644 --- a/internal/internal.proto +++ b/internal/internal.proto @@ -49,6 +49,7 @@ message QueryResponse { optional uint64 N = 3; repeated Pair Pairs = 4; repeated Profile Profiles = 5; + optional bool Changed = 6; } message ImportRequest { diff --git a/roaring/roaring.go b/roaring/roaring.go index 6aa49579c..020bbf62a 100644 --- a/roaring/roaring.go +++ b/roaring/roaring.go @@ -41,24 +41,28 @@ func NewBitmap(a ...uint64) *Bitmap { } // Add adds values to the bitmap. -func (b *Bitmap) Add(a ...uint64) error { +func (b *Bitmap) Add(a ...uint64) (bool, error) { + ret := false for _, v := range a { // Create an add operation. op := &op{typ: opTypeAdd, value: v} // Write operation to op log. if err := b.writeOp(op); err != nil { - return err + return false, err } // Apply to the in-memory bitmap. - op.apply(b) + if op.apply(b) { + ret = true + + } } - return nil + return ret, nil } -func (b *Bitmap) add(v uint64) { +func (b *Bitmap) add(v uint64) bool { hb := highbits(v) i := search64(b.keys, hb) @@ -69,7 +73,7 @@ func (b *Bitmap) add(v uint64) { i = -i - 1 } - b.containers[i].add(lowbits(v)) + return b.containers[i].add(lowbits(v)) } // Contains returns true if v is in the bitmap. @@ -82,29 +86,32 @@ func (b *Bitmap) Contains(v uint64) bool { } // Remove removes values from the bitmap. -func (b *Bitmap) Remove(a ...uint64) error { +func (b *Bitmap) Remove(a ...uint64) (bool, error) { + ret := false for _, v := range a { // Create an add operation. op := &op{typ: opTypeRemove, value: v} // Write operation to op log. if err := b.writeOp(op); err != nil { - return err + return false, err } // Apply operation to the bitmap. - op.apply(b) + if op.apply(b) { + ret = true + } } - return nil + return ret, nil } -func (b *Bitmap) remove(v uint64) { +func (b *Bitmap) remove(v uint64) bool { hb := highbits(v) i := search64(b.keys, hb) if i < 0 { - return + return false } - b.containers[i].remove(lowbits(v)) + return b.containers[i].remove(lowbits(v)) } // Slice returns a slice of all integers in the bitmap. @@ -429,34 +436,32 @@ func (c *container) unmap() { } // add adds a value to the container. -func (c *container) add(v uint16) { +func (c *container) add(v uint16) bool { if c.isArray() { - c.arrayAdd(v) - return + return c.arrayAdd(v) } - c.bitmapAdd(v) + return c.bitmapAdd(v) } -func (c *container) arrayAdd(v uint16) { +func (c *container) arrayAdd(v uint16) bool { // Optimize appending to the end of an array container. if c.n > 0 && c.n < arrayMaxSize && c.isArray() && c.array[c.n-1] < v { c.unmap() c.array = append(c.array, v) c.n++ - return + return true } // Find index of the integer in the container. Exit if it already exists. i := search16(c.array, v) if i >= 0 { - return + return false } // Convert to a bitmap container if too many values are in an array container. if c.n >= arrayMaxSize { c.convertToBitmap() - c.bitmapAdd(v) - return + return c.bitmapAdd(v) } // Otherwise insert into array. @@ -466,15 +471,17 @@ func (c *container) arrayAdd(v uint16) { copy(c.array[i+1:], c.array[i:]) c.array[i] = v c.n++ + return true } -func (c *container) bitmapAdd(v uint16) { +func (c *container) bitmapAdd(v uint16) bool { if c.bitmapContains(v) { - return + return false } c.unmap() c.bitmap[v/64] |= (1 << uint64(v%64)) c.n++ + return true } // contains returns true if v is in the container. @@ -494,28 +501,28 @@ func (c *container) bitmapContains(v uint16) bool { } // remove adds a value to the container. -func (c *container) remove(v uint16) { +func (c *container) remove(v uint16) bool { if c.isArray() { - c.arrayRemove(v) - return + return c.arrayRemove(v) } - c.bitmapRemove(v) + return c.bitmapRemove(v) } -func (c *container) arrayRemove(v uint16) { +func (c *container) arrayRemove(v uint16) bool { i := search16(c.array, v) if i < 0 { - return + return false } c.unmap() c.n-- c.array = append(c.array[:i], c.array[i+1:]...) + return true } -func (c *container) bitmapRemove(v uint16) { +func (c *container) bitmapRemove(v uint16) bool { if !c.bitmapContains(v) { - return + return false } c.unmap() @@ -527,6 +534,7 @@ func (c *container) bitmapRemove(v uint16) { if c.n == arrayMaxSize { c.convertToArray() } + return true } // convertToArray converts the values in the bitmap to array values. @@ -594,15 +602,16 @@ type op struct { } // apply executes the operation against a bitmap. -func (op *op) apply(b *Bitmap) { +func (op *op) apply(b *Bitmap) bool { switch op.typ { case opTypeAdd: - b.add(op.value) + return b.add(op.value) case opTypeRemove: - b.remove(op.value) + return b.remove(op.value) default: panic(fmt.Sprintf("invalid op type: %d", op.typ)) } + return false } // WriteTo writes op to the w.