PostingList to Flatbuffer and back for AddTriple

posting/list.go

+/*
+ * Copyright 2015 Manish R Jain <manishrjain@gmail.com>
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * 		http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package posting
+
+import (
+	"github.com/google/flatbuffers/go"
+	"github.com/manishrjain/dgraph/posting/types"
+	"github.com/manishrjain/dgraph/x"
+)
+
+type List struct {
+	TList *types.PostingList
+}
+
+func addTripleToPosting(b *flatbuffers.Builder,
+	t x.Triple) flatbuffers.UOffsetT {
+
+	// Do this before posting start.
+	so := b.CreateString(t.Source)
+	types.PostingStart(b)
+	types.PostingAddUid(b, t.ValueId)
+
+	// so := b.CreateString(t.Source)
+	types.PostingAddSource(b, so)
+
+	types.PostingAddTs(b, t.Timestamp.UnixNano())
+	return types.PostingEnd(b)
+}
+
+func addPosting(b *flatbuffers.Builder, p types.Posting) flatbuffers.UOffsetT {
+	// Do this before posting start.
+	so := b.CreateByteString(p.Source())
+
+	types.PostingStart(b)
+	types.PostingAddUid(b, p.Uid())
+
+	types.PostingAddSource(b, so)
+
+	types.PostingAddTs(b, p.Ts())
+	return types.PostingEnd(b)
+}
+
+func (l *List) Init() {
+	b := flatbuffers.NewBuilder(0)
+	types.PostingListStart(b)
+	of := types.PostingListEnd(b)
+	b.Finish(of)
+
+	l.TList = types.GetRootAsPostingList(b.Bytes, b.Head())
+}
+
+func (l *List) AddTriple(t x.Triple) {
+	b := flatbuffers.NewBuilder(0)
+
+	num := l.TList.PostingsLength()
+	var offsets []flatbuffers.UOffsetT
+
+	if num == 0 {
+		offsets = append(offsets, addTripleToPosting(b, t))
+
+	} else {
+		added := false
+		for i := 0; i < num; i++ {
+			var p types.Posting
+			l.TList.Postings(&p, i)
+
+			// Put the triple just before the first posting which has a greater
+			// uid than itself.
+			if !added && p.Uid() > t.ValueId {
+				offsets = append(offsets, addTripleToPosting(b, t))
+				added = true
+			}
+			offsets = append(offsets, addPosting(b, p))
+		}
+		if !added {
+			// t.ValueId is the largest. So, add at end.
+			offsets = append(offsets, addTripleToPosting(b, t))
+			added = true // useless, but consistent w/ behavior.
+		}
+	}
+
+	types.PostingListStartPostingsVector(b, len(offsets))
+	for i := len(offsets) - 1; i >= 0; i-- {
+		b.PrependUOffsetT(offsets[i])
+	}
+	vend := b.EndVector(len(offsets))
+
+	types.PostingListStart(b)
+	types.PostingListAddPostings(b, vend)
+	end := types.PostingListEnd(b)
+	b.Finish(end)
+
+	l.TList = types.GetRootAsPostingList(b.Bytes, b.Head())
+}

posting/list_test.go

+/*
+ * Copyright 2015 Manish R Jain <manishrjain@gmail.com>
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * 		http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package posting
+
+import (
+	"testing"
+	"time"
+
+	"github.com/manishrjain/dgraph/posting/types"
+	"github.com/manishrjain/dgraph/x"
+)
+
+var uids = [...]uint64{
+	9, 49, 81,
+}
+
+func TestAddTriple(t *testing.T) {
+	var l List
+	l.Init()
+
+	triple := x.Triple{
+		ValueId:   9,
+		Source:    "testing",
+		Timestamp: time.Now(),
+	}
+	l.AddTriple(triple)
+
+	if l.TList.PostingsLength() != 1 {
+		t.Error("Unable to find added elements in posting list")
+	}
+	var p types.Posting
+	if ok := l.TList.Postings(&p, 0); !ok {
+		t.Error("Unable to retrieve posting at 1st iter")
+		t.Fail()
+	}
+	if p.Uid() != 9 {
+		t.Errorf("Expected 9. Got: %v", p.Uid)
+	}
+	if string(p.Source()) != "testing" {
+		t.Errorf("Expected testing. Got: %v", string(p.Source()))
+	}
+
+	// Add another triple now.
+	triple.ValueId = 81
+	l.AddTriple(triple)
+	if l.TList.PostingsLength() != 2 {
+		t.Errorf("Length: %d", l.TList.PostingsLength())
+		t.Fail()
+	}
+
+	var uid uint64
+	uid = 1
+	for i := 0; i < l.TList.PostingsLength(); i++ {
+		if ok := l.TList.Postings(&p, i); !ok {
+			t.Error("Unable to retrieve posting at 2nd iter")
+		}
+		uid *= 9
+		if p.Uid() != uid {
+			t.Errorf("Expected: %v. Got: %v", uid, p.Uid())
+		}
+	}
+
+	// Add another triple, in between the two above.
+	triple.ValueId = 49
+	l.AddTriple(triple)
+	if l.TList.PostingsLength() != 3 {
+		t.Errorf("Length: %d", l.TList.PostingsLength())
+		t.Fail()
+	}
+	for i := 0; i < len(uids); i++ {
+		if ok := l.TList.Postings(&p, i); !ok {
+			t.Error("Unable to retrieve posting at 2nd iter")
+		}
+		if p.Uid() != uids[i] {
+			t.Errorf("Expected: %v. Got: %v", uids[i], p.Uid())
+		}
+	}
+}

types.fbs → posting/types.fbs

 namespace types;
 
+table Posting {
+	uid:ulong;
+	source:string;
+	ts:long;
+}
+
 table PostingList {
-	ids:[ulong];
+	postings:[Posting];
 }
 
 root_type PostingList;

posting/types/Posting.go

+// automatically generated, do not modify
+
+package types
+
+import (
+	flatbuffers "github.com/google/flatbuffers/go"
+)
+type Posting struct {
+	_tab flatbuffers.Table
+}
+
+func (rcv *Posting) Init(buf []byte, i flatbuffers.UOffsetT) {
+	rcv._tab.Bytes = buf
+	rcv._tab.Pos = i
+}
+
+func (rcv *Posting) Uid() uint64 {
+	o := flatbuffers.UOffsetT(rcv._tab.Offset(4))
+	if o != 0 {
+		return rcv._tab.GetUint64(o + rcv._tab.Pos)
+	}
+	return 0
+}
+
+func (rcv *Posting) Source() []byte {
+	o := flatbuffers.UOffsetT(rcv._tab.Offset(6))
+	if o != 0 {
+		return rcv._tab.ByteVector(o + rcv._tab.Pos)
+	}
+	return nil
+}
+
+func (rcv *Posting) Ts() int64 {
+	o := flatbuffers.UOffsetT(rcv._tab.Offset(8))
+	if o != 0 {
+		return rcv._tab.GetInt64(o + rcv._tab.Pos)
+	}
+	return 0
+}
+
+func PostingStart(builder *flatbuffers.Builder) { builder.StartObject(3) }
+func PostingAddUid(builder *flatbuffers.Builder, uid uint64) { builder.PrependUint64Slot(0, uid, 0) }
+func PostingAddSource(builder *flatbuffers.Builder, source flatbuffers.UOffsetT) { builder.PrependUOffsetTSlot(1, flatbuffers.UOffsetT(source), 0) }
+func PostingAddTs(builder *flatbuffers.Builder, ts int64) { builder.PrependInt64Slot(2, ts, 0) }
+func PostingEnd(builder *flatbuffers.Builder) flatbuffers.UOffsetT { return builder.EndObject() }

types/PostingList.go → posting/types/PostingList.go

@@ -21,16 +21,22 @@ func (rcv *PostingList) Init(buf []byte, i flatbuffers.UOffsetT) {
 	rcv._tab.Pos = i
 }
 
-func (rcv *PostingList) Ids(j int) uint64 {
+func (rcv *PostingList) Postings(obj *Posting, j int) bool {
 	o := flatbuffers.UOffsetT(rcv._tab.Offset(4))
 	if o != 0 {
-		a := rcv._tab.Vector(o)
-		return rcv._tab.GetUint64(a + flatbuffers.UOffsetT(j * 8))
+		x := rcv._tab.Vector(o)
+		x += flatbuffers.UOffsetT(j) * 4
+		x = rcv._tab.Indirect(x)
+	if obj == nil {
+		obj = new(Posting)
 	}
-	return 0
+		obj.Init(rcv._tab.Bytes, x)
+		return true
+	}
+	return false
 }
 
-func (rcv *PostingList) IdsLength() int {
+func (rcv *PostingList) PostingsLength() int {
 	o := flatbuffers.UOffsetT(rcv._tab.Offset(4))
 	if o != 0 {
 		return rcv._tab.VectorLen(o)
@@ -39,7 +45,7 @@ func (rcv *PostingList) IdsLength() int {
 }
 
 func PostingListStart(builder *flatbuffers.Builder) { builder.StartObject(1) }
-func PostingListAddIds(builder *flatbuffers.Builder, ids flatbuffers.UOffsetT) { builder.PrependUOffsetTSlot(0, flatbuffers.UOffsetT(ids), 0) }
-func PostingListStartIdsVector(builder *flatbuffers.Builder, numElems int) flatbuffers.UOffsetT { return builder.StartVector(8, numElems, 8)
+func PostingListAddPostings(builder *flatbuffers.Builder, postings flatbuffers.UOffsetT) { builder.PrependUOffsetTSlot(0, flatbuffers.UOffsetT(postings), 0) }
+func PostingListStartPostingsVector(builder *flatbuffers.Builder, numElems int) flatbuffers.UOffsetT { return builder.StartVector(4, numElems, 4)
 }
 func PostingListEnd(builder *flatbuffers.Builder) flatbuffers.UOffsetT { return builder.EndObject() }

x/x.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"net/http"
+	"time"
 
 	"github.com/Sirupsen/logrus"
 )
@@ -27,6 +28,16 @@ type Status struct {
 	Message string `json:"message"`
 }
 
+type Triple struct {
+	Entity    uint64
+	EntityEid string
+	Attribute string
+	Value     interface{}
+	ValueId   uint64
+	Source    string
+	Timestamp time.Time
+}
+
 func Log(p string) *logrus.Entry {
 	l := logrus.WithFields(logrus.Fields{
 		"package": p,