mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-09-07 00:55:55 +00:00
adding load endpoint to support loading bitmaps from python client
This commit is contained in:
parent
19ccc192ae
commit
93e4a1fc92
2 changed files with 143 additions and 2 deletions
64
core/http.go
64
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
|
||||
|
|
|
|||
81
core/load.go
Normal file
81
core/load.go
Normal file
|
|
@ -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"
|
||||
|
||||
}
|
||||
Loading…
Add table
Reference in a new issue