add batch loader

commands/pilosa-loader/loader.go

@@ -9,13 +9,22 @@ import (
 	"net/http"
 	"net/url"
 	"os"
+	"pilosa/db"
+	"pilosa/index"
+	"strconv"
 	"strings"
+
+	"github.com/davecgh/go-spew/spew"
 )
 
-func post(database, base_url, id, pid, fragment_type string) {
+func post(database, base_url, id, compressed_string, fragment_type string, slice int) {
 	values := make(url.Values)
 	values.Set("db", database)
-	values.Set("pql", fmt.Sprintf("set(%s,%s,%s)", id, fragment_type, pid))
+	values.Set("id", id)
+	values.Set("frame", fragment_type)
+	values.Set("slice", fmt.Sprintf("%d", slice))
+	values.Set("bitmap", compressed_string)
+	spew.Dump(values)
 	r, err := http.PostForm(base_url, values)
 	if err != nil {
 		log.Printf("error posting stat to stathat: %s", err)
@@ -38,10 +47,31 @@ func Load(database, url, fullpath string, fragment_type string) error {
 		recs := strings.Split(line, "|")
 		id := recs[0]
 		log.Println(id)
-		for _, profile_id := range recs[1:] {
-			post(database, url, id, profile_id, fragment_type)
+		bm := index.NewBitmap()
+		slice := -1
+		last_slice := -1
+		for _, pid := range recs[1:] {
+			profile_id, _ := strconv.ParseUint(strings.TrimSpace(pid), 10, 64)
+
+			if slice < 0 {
+				slice = db.GetSlice(profile_id)
+				last_slice = slice
+			} else {
+				slice = db.GetSlice(profile_id)
+			}
+			if last_slice != slice {
+				post(database, url, id, bm.ToCompressString(), fragment_type, last_slice)
+				bm = index.NewBitmap()
+				last_slice = slice
+			}
+
+			index.SetBit(bm, profile_id)
 
 		}
+		if slice != -1 {
+			post(database, url, id, bm.ToCompressString(), fragment_type, slice)
+		}
+
 		line, e = Readln(r)
 	}
 	return nil
@@ -68,7 +98,7 @@ func Readln(r *bufio.Reader) (string, error) {
 
 func main() {
 	flag.Parse()
-	full_url := fmt.Sprintf("http://%s/query", *host_port)
+	full_url := fmt.Sprintf("http://%s/bulk", *host_port)
 	//fun(*file)
 	Load(*database, full_url, *file, *fragment)
 }

core/http.go

@@ -34,6 +34,7 @@ func (self *WebService) Run() {
 	mux.HandleFunc("/test", self.HandleTest)
 	mux.HandleFunc("/version", self.HandleVersion)
 	mux.HandleFunc("/ping", self.HandlePing)
+	mux.HandleFunc("/batch", self.HandleBatch)
 	s := &http.Server{
 		Addr:    ":" + port_string,
 		Handler: mux,
@@ -55,6 +56,66 @@ func (self *WebService) HandleMessage(w http.ResponseWriter, r *http.Request) {
 	//service.Inbox <- &message
 }
 
+func (self *WebService) HandleBatch(w http.ResponseWriter, r *http.Request) {
+	if r.Method != "POST" {
+		http.Error(w, "Only POST allowed", http.StatusMethodNotAllowed)
+		return
+	}
+	/*   	values.Set("db", database)
+	values.Set("id", id)
+	values.Set("frame", fragment_type)
+	values.Set("slice", fmt.Sprintf("%d", slice))
+	values.Set("bitmap", compressed_string)
+	*/
+
+	err := r.ParseForm()
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+
+	database_name := r.Form.Get("db")
+	if database_name == "" {
+		http.Error(w, "Provide a database (db)", http.StatusNotFound)
+		return
+	}
+
+	bitmap_id_string := r.Form.Get("id")
+	if bitmap_id_string == "" {
+		http.Error(w, "Provide a bitmap id (id)", http.StatusNotFound)
+		return
+	}
+	bitmap_id, _ := strconv.ParseUint(bitmap_id_string, 10, 64)
+
+	slice_string := r.Form.Get("slice")
+	if bitmap_id_string == "" {
+		http.Error(w, "Provide a slice (slice)", http.StatusNotFound)
+		return
+	}
+	slice, _ := strconv.ParseInt(slice_string, 10, 32)
+
+	frame := r.Form.Get("frame")
+	if bitmap_id_string == "" {
+		http.Error(w, "Provide a frame (frame)", http.StatusNotFound)
+		return
+	}
+
+	compressed_bitmap := r.Form.Get("bitmap")
+	if bitmap_id_string == "" {
+		http.Error(w, "Provide a compressed base64 bitmap (bitmap)", http.StatusNotFound)
+		return
+	}
+
+	results := self.service.Batch(database_name, frame, compressed_bitmap, bitmap_id, int(slice))
+
+	encoder := json.NewEncoder(w)
+	err = encoder.Encode(results)
+	if err != nil {
+		log.Fatal("Error encoding results")
+	}
+
+}
+
 func (self *WebService) HandleQuery(w http.ResponseWriter, r *http.Request) {
 	if r.Method != "POST" {
 		http.Error(w, "Only POST allowed", http.StatusMethodNotAllowed)

db/topology.go

@@ -390,13 +390,16 @@ func (d *Database) GetOrCreateFragment(frame *Frame, slice *Slice, fragment_id u
 func (f *Fragment) SetProcess(process *Process) {
 	f.process = process
 }
+func GetSlice(profile_id uint64) int {
+	return int(profile_id / SLICE_WIDTH)
+}
 
 ///////////////////////////////////////////////////////////////////////////////////////////////
 
 // Get a slice from a database
+
 func (d *Database) GetSliceForProfile(profile_id uint64) (*Slice, error) {
-	slice_id := profile_id / SLICE_WIDTH
-	return d.getSlice(int(slice_id))
+	return d.getSlice(GetSlice(profile_id))
 }
 
 type Bitmap struct {

dispatch/dispatch.go

@@ -27,6 +27,10 @@ func (self *Dispatch) Run() {
 	for {
 		message := self.service.Transport.Receive()
 		switch data := message.Data.(type) {
+		case core.BatchRequest:
+			response := db.Message{Data: core.BatchResponse{Id: data.Id}}
+			self.service.Index.LoadBitmap(data.Fragment_id, data.Bitmap_id, data.Compressed_bitmap)
+			self.service.Transport.Send(&response, data.Source)
 		case core.PingRequest:
 			pong := db.Message{Data: core.PongRequest{Id: data.Id}}
 			self.service.Transport.Send(&pong, data.Source)

index/bitmap.go

@@ -4,7 +4,10 @@ package index
 
 import (
 	"bytes"
+	"compress/gzip"
+	"encoding/base64"
 	"encoding/gob"
+	"io/ioutil"
 	"log"
 
 	"github.com/yasushi-saito/rbtree"
@@ -354,6 +357,8 @@ type IBitmap interface {
 	BuildFromBits(bits []uint64)
 	ToBytes() []byte
 	FromBytes([]byte)
+	ToCompressString() string
+	FromCompressString(string)
 }
 
 func NewRB() *rbtree.Tree {
@@ -363,8 +368,29 @@ func NewRB() *rbtree.Tree {
 func CreateRBBitmap() IBitmap {
 	return &Bitmap{nodes: NewRB(), bcount: 0}
 }
-func (b *Bitmap) AddChunk(a *Chunk) {
-	b.nodes.Insert(a)
+
+func (self *Bitmap) FromCompressString(str string) {
+	compressed_data, err := base64.StdEncoding.DecodeString(str)
+	if err != nil {
+		log.Println(err)
+		return
+	}
+	reader, _ := gzip.NewReader(bytes.NewReader(compressed_data))
+	data, _ := ioutil.ReadAll(reader)
+	self.FromBytes(data)
+}
+
+func (self *Bitmap) ToCompressString() string {
+	var b bytes.Buffer
+	w := gzip.NewWriter(&b)
+	w.Write(self.ToBytes())
+	w.Flush()
+	w.Close()
+	return base64.StdEncoding.EncodeToString(b.Bytes())
+}
+
+func (self *Bitmap) AddChunk(a *Chunk) {
+	self.nodes.Insert(a)
 }
 func (b *Bitmap) Min() ChunkIterator {
 	return &RBNodeIterator{b.nodes.Min()}

index/brand.go

@@ -125,6 +125,12 @@ func packagePairs(r RankList) []Pair {
 	}
 	return res
 }
+func (self *Brand) Store(bitmap_id uint64, bm IBitmap) {
+	//oldbm:=self.Get(bitmap_id)
+	//nbm = Union(oldbm, bm)
+	self.storage.Store(int64(bitmap_id), self.db, self.slice, bm.(*Bitmap))
+	self.cache_it(bm, bitmap_id)
+}
 
 func (self *Brand) TopN(src_bitmap IBitmap, n int) []Pair {
 	breakout := 500

index/commands.go

@@ -202,3 +202,19 @@ func NewClear() *CmdClear {
 func (self *CmdClear) Execute(f *Fragment) Calculation {
 	return f.impl.Clear()
 }
+
+type CmdLoader struct {
+	*Responder
+	bitmap_id         uint64
+	compressed_bitmap string
+}
+
+func NewLoader(bitmap_id uint64, compressed_bitmap string) *CmdLoader {
+	return &CmdLoader{NewResponder("Loader"), bitmap_id, compressed_bitmap}
+}
+func (self *CmdLoader) Execute(f *Fragment) Calculation {
+	nbm := NewBitmap()
+	nbm.FromCompressString(self.compressed_bitmap)
+	f.impl.Store(self.bitmap_id, nbm)
+	return "ok"
+}

index/fragment_container.go

@@ -25,6 +25,13 @@ func init() {
 	gob.Register(vh)
 }
 
+func (self *FragmentContainer) LoadBitmap(frag_id SUUID, bitmap_id uint64, compressed_bitmap string) {
+	if fragment, found := self.GetFragment(frag_id); found {
+		loader := NewLoader(bitmap_id, compressed_bitmap)
+		fragment.requestChan <- loader
+	}
+}
+
 func (self *FragmentContainer) GetFragment(frag_id SUUID) (*Fragment, bool) {
 	//lock
 	c, v := self.fragments[frag_id]
@@ -144,6 +151,7 @@ type Pilosa interface {
 	SetBit(id uint64, bit_pos uint64) bool
 	TopN(b IBitmap, n int) []Pair
 	Clear() bool
+	Store(bitmap_id uint64, bm IBitmap)
 }
 
 type Fragment struct {

index/general.go

@@ -48,3 +48,9 @@ func (self *General) TopN(b IBitmap, n int) []Pair {
 
 	return nil
 }
+func (self *General) Store(bitmap_id uint64, bm IBitmap) {
+	//oldbm:=self.Get(bitmap_id)
+	//nbm = Union(oldbm, bm)
+	self.storage.Store(int64(bitmap_id), self.db, self.slice, bm.(*Bitmap))
+	self.bitmap_cache.Add(bitmap_id, bm)
+}

util/util_test.go

@@ -46,6 +46,13 @@ func TestId(t *testing.T) {
 		b2 := Hex_to_SUUID(s)
 		So(b1, ShouldEqual, b2)
 	})
+	Convey("Gen 10", t, func() {
+		for i := 0; i < 10; i++ {
+			bc1 := Id()
+			println(SUUID_to_Hex(bc1))
+		}
+		So(1, ShouldEqual, 1)
+	})
 
 }