From 735a7287b79ee3bd7e5841b05aed13b2fd8b7beb Mon Sep 17 00:00:00 2001 From: Linh Vo Date: Thu, 22 Jun 2017 21:57:00 -0500 Subject: [PATCH] merge action's processing --- handler.go | 63 +++++++++++++++++++++++++++++++++++++++++++ server/server_test.go | 32 ++++++++++++++++++++++ 2 files changed, 95 insertions(+) diff --git a/handler.go b/handler.go index 5dd2347e8..def9f8436 100644 --- a/handler.go +++ b/handler.go @@ -29,6 +29,7 @@ import ( "net/http" _ "net/http/pprof" "os" + "sort" "strconv" "strings" "time" @@ -1639,3 +1640,65 @@ func (h *Handler) handlePostInput(w http.ResponseWriter, r *http.Request) { } } } + +// MapAction Process the input data and set a bit +func (h *Handler) MapAction(a *Action, value string, colID uint64) (*Bit, error) { + var bit Bit + var ok bool + bit.ColumnID = colID + bit.RowID, ok = a.ValueMap[value] + if !ok { + return nil, fmt.Errorf("Value %s does not exist in definition map", value) + } + + // _, err := i.Frame(f).SetBit(ViewStandard, rowID, colID, nil) + return &bit, nil +} + +// ValueToRow Sets a bitmap with rowID from the input value +func (h *Handler) ValueToRow(a *Action, value string, colID uint64) (*Bit, error) { + var bit Bit + var err error + bit.ColumnID = colID + bit.RowID, err = strconv.ParseUint(value, 10, 64) + if err != nil { + return nil, err + } + // _, err = i.Frame(f).SetBit(ViewStandard, rowID, colID, nil) + return &bit, err +} + +// SingleRowBoolean Sets a bitmap with rowID from the Action defintion +func (h *Handler) SingleRowBoolean(a *Action, value string, colID uint64) (*Bit, error) { + var bit Bit + var err error + bit.ColumnID = colID + bit.RowID, err = strconv.ParseUint(value, 10, 64) + if err != nil { + return nil, err + } + // _, err = i.Frame(f).SetBit(ViewStandard, rowID, colID, nil) + return &bit, err +} + +// InputBits Process and Sort the Input Bits and route to appropriate nodes. +func (h *Handler) InputBits(index, frame string, bits []Bit) error { + client, err := NewClient(h.Host) + if err != nil { + return err + } + + bitsBySlice := Bits(bits).GroupBySlice() + + // Parse path into bits. + for slice, bits := range bitsBySlice { + sort.Sort(BitsByPos(bits)) + + h.logger().Printf("inputing slice: %d, n=%d", slice, len(bits)) + if err := client.Import(context.Background(), index, frame, slice, bits); err != nil { + return err + } + } + + return nil +} diff --git a/server/server_test.go b/server/server_test.go index cd82ba007..a92da70a8 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -494,6 +494,29 @@ func TestMain_SendReceiveMessage(t *testing.T) { if maxSlices1["i"] != 2 { t.Fatalf("unexpected maxSlice on node1: %d", maxSlices1["i"]) } + + // Write input definition to the first node. + if _, err := m0.CreateDefinition("i", "test", `{ + "frames": [{"name": "event-time", + "options": { + "cacheType": "ranked", + "timeQuantum": "YMD" + }}], + "fields": [{"name": "id", + "primaryKey": true + }]} + `); err != nil { + t.Fatal(err) + } + + frame0 := m0.Server.Holder.Frame("i", "event-time") + if frame0 == nil { + t.Fatal("frame not found") + } + frame1 := m1.Server.Holder.Frame("i", "event-time") + if frame1 == nil { + t.Fatal("frame not found") + } } // availablePorts returns a slice of ports that can be used for testing. @@ -615,6 +638,15 @@ func (m *Main) Query(index, rawQuery, query string) (string, error) { return resp.Body, nil } +// CreateDefinition. +func (m *Main) CreateDefinition(index, def, query string) (string, error) { + resp := MustDo("POST", m.URL()+fmt.Sprintf("/index/%s/input-definition/%s", index, def), query) + if resp.StatusCode != http.StatusOK { + return "", fmt.Errorf("invalid status: %d, body=%s", resp.StatusCode, resp.Body) + } + return resp.Body, nil +} + // SetCommand represents a command to set a bit. type SetCommand struct { ID uint64