From 93e4a1fc92d2cbd6465dbff047c793e10c2e9bf7 Mon Sep 17 00:00:00 2001
From: Todd Gruben <tgruben@umbel.com>
Date: Thu, 26 Feb 2015 23:01:17 +0000
Subject: [PATCH] adding load endpoint to support loading bitmaps from python
 client

---
 core/http.go | 64 +++++++++++++++++++++++++++++++++++++++--
 core/load.go | 81 ++++++++++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 143 insertions(+), 2 deletions(-)
 create mode 100644 core/load.go

diff --git a/core/http.go b/core/http.go
index 8c8f90cf3..730567b5b 100644
--- a/core/http.go
+++ b/core/http.go
@@ -162,6 +162,7 @@ func (self *WebService) Run() {
 	mux.HandleFunc("/version", self.HandleVersion)
 	mux.HandleFunc("/ping", self.HandlePing)
 	mux.HandleFunc("/batch", self.HandleBatch)
+	mux.HandleFunc("/load", self.HandleLoad)
 	log_set_bit := config.GetIntDefault("log_set_bit_request", 0)
 	if log_set_bit == 1 {
 		mux.HandleFunc("/set_bits", NewRequestLogger(self.HandleSetBit, logger_chan))
@@ -198,6 +199,64 @@ func (self *WebService) HandleMessage(w http.ResponseWriter, r *http.Request) {
 	//service.Inbox <- &message
 }
 
+func (self *WebService) HandleLoad(w http.ResponseWriter, r *http.Request) {
+	if r.Method != "POST" {
+		http.Error(w, "Only POST allowed", http.StatusMethodNotAllowed)
+		return
+	}
+
+	decoder := json.NewDecoder(r.Body)
+	var obj JsonObject
+
+	err := decoder.Decode(&obj)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+	database_name, ok := obj["db"]
+	if !ok {
+		http.Error(w, "Provide a database (db)", http.StatusNotFound)
+		return
+	}
+
+	_, ok = obj["id"]
+	if !ok {
+		http.Error(w, "Provide a bitmap id (id)", http.StatusNotFound)
+		return
+	}
+	t := float64(obj["id"].(float64))
+	bitmap_id := uint64(t)
+
+	frame, ok := obj["frame"]
+	if !ok {
+		http.Error(w, "Provide a frame (frame)", http.StatusNotFound)
+		return
+	}
+	api_string, ok := obj["bitmap"]
+	if !ok {
+		http.Error(w, "Provide a compressed base64 bitmap (bitmap)", http.StatusNotFound)
+		return
+	}
+
+	_, ok = obj["filter"]
+	if !ok {
+		http.Error(w, "Provide a filter for categories", http.StatusNotFound)
+		return
+	}
+	t = float64(obj["filter"].(float64))
+	filter := uint64(t)
+
+	results := FromApiString(self.service, database_name.(string), frame.(string), api_string.(string), bitmap_id, filter)
+
+	encoder := json.NewEncoder(w)
+	err = encoder.Encode(results)
+	if err != nil {
+		log.Warn("Error Load results")
+		log.Warn(spew.Sdump(r.Form))
+		err = encoder.Encode("Bad Batch Request")
+	}
+
+}
 func (self *WebService) HandleBatch(w http.ResponseWriter, r *http.Request) {
 	if r.Method != "POST" {
 		http.Error(w, "Only POST allowed", http.StatusMethodNotAllowed)
@@ -360,8 +419,9 @@ func bitmaps(frame string, obj JsonObject) chan uint64 {
 		base_id := uint64(t)
 
 		if strings.HasSuffix(frame, ".t") {
-			timestamp := obj["timestamp"].(string)
-			if timestamp == "2014-01-01 00:00:00" { //skip the default timestamp
+			timestamp, present := obj["timestamp"].(string)
+
+			if !present || timestamp == "2014-01-01 00:00:00" { //skip the default timestamp
 				c <- base_id
 			} else {
 				quantum := index.YMDH
diff --git a/core/load.go b/core/load.go
new file mode 100644
index 000000000..75b96ea67
--- /dev/null
+++ b/core/load.go
@@ -0,0 +1,81 @@
+package core
+
+// #cgo  CFLAGS:-mpopcnt
+
+import (
+	"bytes"
+	"compress/gzip"
+	"encoding/base64"
+	"encoding/binary"
+	log "github.com/cihub/seelog"
+	"pilosa/index"
+)
+
+func copy_raw(src [32]uint64) index.BlockArray {
+	var o = make([]uint64, 32, 32)
+	for k, v := range src {
+		o[k] = v
+	}
+	return index.BlockArray{o}
+}
+func sendBitmap(service *Service, bitmap index.IBitmap, db string, frame string, bitmap_id, filter uint64, slice int, finish chan error) {
+	compressed_bitmap := bitmap.ToCompressString()
+	results := service.Batch(db, frame, compressed_bitmap, bitmap_id, slice, filter)
+	finish <- results
+}
+
+func FromApiString(service *Service, db string, frame string, api_string string, bitmap_id, filter uint64) string {
+	compressed_data, err := base64.StdEncoding.DecodeString(api_string)
+	if err != nil {
+		log.Warn(err)
+		return "Bad"
+	}
+	reader, err := gzip.NewReader(bytes.NewReader(compressed_data))
+	if err != nil {
+		log.Warn(err)
+		return "Bad"
+	}
+	var numChunks uint64
+	err = binary.Read(reader, binary.BigEndian, &numChunks)
+	if err != nil {
+		log.Warn(err)
+		return "Bad"
+	}
+	first := true
+	bitmap := index.NewBitmap()
+	last_slice := index.COUNTERMASK
+	sent_count := 0
+	finish := make(chan error)
+
+	for i := uint64(0); i < numChunks; i++ {
+		var raw struct {
+			Key   uint64
+			Block [32]uint64
+		}
+		binary.Read(reader, binary.BigEndian, &raw)
+		slice := raw.Key >> 5
+		if slice != last_slice {
+			if first {
+				first = false
+			} else {
+				//make async latter
+				sent_count += 1
+				go sendBitmap(service, bitmap, db, frame, bitmap_id, filter, int(last_slice), finish)
+				bitmap = index.NewBitmap()
+			}
+			last_slice = slice
+		}
+		o := copy_raw(raw.Block)
+		chunk := &index.Chunk{raw.Key, o}
+		println("KEY", raw.Key)
+		bitmap.AddChunk(chunk)
+
+	}
+	sent_count += 1
+	go sendBitmap(service, bitmap, db, frame, bitmap_id, filter, int(last_slice), finish)
+	for i := 0; i < sent_count; i++ {
+		<-finish
+	}
+	return "OK"
+
+}