diff --git a/handler.go b/handler.go index 2474c537a..a40bd8766 100644 --- a/handler.go +++ b/handler.go @@ -1643,7 +1643,7 @@ func (h *Handler) handlePostInput(w http.ResponseWriter, r *http.Request) { return } for _, req := range reqs { - err = h.JSONParser(req.(map[string]interface{}), index, inputDefName) + bits, err := h.JSONParser(req.(map[string]interface{}), index, inputDefName) if err == ErrInputDefinitionNotFound { http.Error(w, err.Error(), http.StatusNotFound) return @@ -1651,18 +1651,24 @@ func (h *Handler) handlePostInput(w http.ResponseWriter, r *http.Request) { http.Error(w, err.Error(), http.StatusBadRequest) return } + for fr, bs := range bits { + err := index.InputBits(fr, bs) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + } } - if err := json.NewEncoder(w).Encode(postInputDefinitionResponse{}); err != nil { h.logger().Printf("response encoding error: %s", err) } } // JSONParser validate input json file and execute SetBit -func (h *Handler) JSONParser(req map[string]interface{}, index *Index, name string) error { +func (h *Handler) JSONParser(req map[string]interface{}, index *Index, name string) (map[string][]*Bit, error) { inputDef := index.inputDefinition(name) if inputDef == nil { - return ErrInputDefinitionNotFound + return nil, ErrInputDefinitionNotFound } // if field in input data is not in defined definition, return error var columnLabel string @@ -1676,11 +1682,12 @@ func (h *Handler) JSONParser(req map[string]interface{}, index *Index, name stri for key, _ := range req { _, ok := validFields[key] if !ok { - return fmt.Errorf("field not found: %s", key) + return nil, fmt.Errorf("field not found: %s", key) } } var bits []*Bit + setBits := make(map[string][]*Bit) for _, field := range inputDef.Fields() { // skip field that defined in definition but not in input data //var colValue uint64 @@ -1689,24 +1696,22 @@ func (h *Handler) JSONParser(req map[string]interface{}, index *Index, name stri } value, ok := req[columnLabel] if !ok { - return fmt.Errorf("columnLabel required") + return nil, fmt.Errorf("columnLabel required") } colValue, ok := value.(float64) if !ok { - return fmt.Errorf("float64 require, got value:%s, type: %s", value, reflect.TypeOf(value)) + return nil, fmt.Errorf("float64 require, got value:%s, type: %s", value, reflect.TypeOf(value)) } for _, action := range field.Actions { - bit, err := h.HandleAction(action, req[field.Name], uint64(colValue)) + frame := action.Frame + bit, err := HandleAction(action, req[field.Name], uint64(colValue)) if err != nil { - return fmt.Errorf("error handling action: %s", action.ValueDestination) + return nil, fmt.Errorf("error handling action: %s, err: %s", action.ValueDestination, err) } - bits = append(bits, bit) + //bits = append(bits, bit) + setBits[frame] = append(bits, bit) } } - return nil -} - -func (h *Handler) HandleAction(a Action, value interface{}, colID uint64) (*Bit, error) { - return nil, nil + return setBits, nil } diff --git a/handler_test.go b/handler_test.go index bb9b5359d..cc71c76ff 100644 --- a/handler_test.go +++ b/handler_test.go @@ -19,6 +19,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "github.com/gogo/protobuf/proto" "github.com/pilosa/pilosa" "github.com/pilosa/pilosa/internal" @@ -1304,12 +1305,27 @@ var defaultBody = ` { "frames":[ { - "name":"event-time", + "name":"cab-type", "options": { "timeQuantum":"YMD", "inverseEnabled":false, "cacheType":"ranked" } + }, + { + "name":"add-ons", + "options": { + "timeQuantum":"YMD", + "inverseEnabled":false, + "cacheType":"ranked" + } + }, + { + "name":"distance-miles", + "options": { + "timeQuantum":"YMD", + "cacheType":"ranked" + } } ], "fields":[ @@ -1324,8 +1340,8 @@ var defaultBody = ` "frame":"cab-type", "valueDestination":"mapping", "valueMap":{ - "Green":1, - "Yellow":2 + "green":1, + "yellow":2 } } ] @@ -1372,13 +1388,14 @@ func TestHandler_CreateInput(t *testing.T) { "id": 1, "cabType": "yellow", "distanceMiles": 8, - "with-pet": true + "withPet": true }]`) h := NewHandler() h.Holder = hldr.Holder h.Cluster = NewCluster(1) w := httptest.NewRecorder() h.ServeHTTP(w, MustNewHTTPRequest("POST", "/index/i0/input/input1", bytes.NewBuffer(inputBody))) + fmt.Print(w.Body.String()) if w.Code != http.StatusOK { t.Fatalf("unexpected status code: %d", w.Code) } else if body := w.Body.String(); body != `{}`+"\n" { diff --git a/index.go b/index.go index ca5428e06..3dad6afd8 100644 --- a/index.go +++ b/index.go @@ -755,3 +755,26 @@ func (i *Index) openInputDefinition() error { } return nil } + +// InputBits Process the []Bit though the Frame import process +func (i *Index) InputBits(frame string, bits []*Bit) error { + var rowIDs, columnIDs []uint64 + timestamps := make([]*time.Time, len(bits)) + f := i.Frame(frame) + if f == nil { + return fmt.Errorf("Frame not found: %s", frame) + } + + for i, bit := range bits { + rowIDs = append(rowIDs, bit.RowID) + columnIDs = append(columnIDs, bit.ColumnID) + + // Convert timestamps to time.Time. + if bit.Timestamp > 0 { + t := time.Unix(0, bit.Timestamp) + timestamps[i] = &t + } + } + + return f.Import(rowIDs, columnIDs, timestamps) +} diff --git a/index_test.go b/index_test.go index 349dc0522..14e1ee617 100644 --- a/index_test.go +++ b/index_test.go @@ -459,3 +459,37 @@ func TestIndex_CreateFrameWhenOpenInputDefinition(t *testing.T) { } } + +func TestIndex_InputBits(t *testing.T) { + index := MustOpenIndex() + defer index.Close() + + // Set index time quantum. + if err := index.SetTimeQuantum(pilosa.TimeQuantum("YM")); err != nil { + t.Fatal(err) + } + + // Create frame. + if _, err := index.CreateFrameIfNotExists("f", pilosa.FrameOptions{}); err != nil { + t.Fatal(err) + } + + var bits []*pilosa.Bit + bits = append(bits, &pilosa.Bit{RowID: 0, ColumnID: 0}) + bits = append(bits, &pilosa.Bit{RowID: 0, ColumnID: 1}) + bits = append(bits, &pilosa.Bit{RowID: 2, ColumnID: 2, Timestamp: 1}) + + err := index.InputBits("f", bits) + if err != nil { + t.Fatal(err) + } + + f := index.Frame("f") + v := f.View(pilosa.ViewStandard) + fragment := v.Fragment(0) + + // Verify the Bits were set + if a := fragment.Row(0).Bits(); !reflect.DeepEqual(a, []uint64{0, 1}) { + t.Fatalf("unexpected bits: %+v", a) + } +} diff --git a/input_definition.go b/input_definition.go index 5c3154305..cc42bb4e4 100644 --- a/input_definition.go +++ b/input_definition.go @@ -15,12 +15,12 @@ package pilosa import ( + "fmt" "io/ioutil" "os" "path/filepath" "errors" - "fmt" "github.com/gogo/protobuf/proto" "github.com/pilosa/pilosa/internal" @@ -318,3 +318,51 @@ func (i *InputDefinition) ValidateAction(action *internal.InputDefinitionAction) } return nil } + +// HandleAction Process the input data with its action and return a bit to be imported later +// Note: if the Bit should not be set then nil is returned with no error +// From the JSON marshalling the possible types are: float64, boolean, string +// TODO handle Timestams +func HandleAction(a Action, value interface{}, colID uint64) (*Bit, error) { + var err error + var bit Bit + bit.ColumnID = colID + + switch a.ValueDestination { + case Mapping: + v, ok := value.(string) + if !ok { + return nil, fmt.Errorf("Mapping value must be a string %v", value) + } + bit.RowID, ok = a.ValueMap[v] + if !ok { + return nil, fmt.Errorf("Value %s does not exist in definition map", v) + } + case SingleRowBool: + switch value.(type) { + case bool: + if value.(bool) { + bit.RowID = *a.RowID + } else { // value is not True. + return nil, err + } + case float64: + if value.(float64) >= 1 { + bit.RowID = *a.RowID + } else { // value is not True. + return nil, err + } + default: + return nil, fmt.Errorf("single-row-boolean value %v must equate to a Bool", value) + } + case ValueToRow: + v, ok := value.(float64) + if !ok { + return nil, fmt.Errorf("value-to-row value must equate to an integer %v", value) + } + bit.RowID = uint64(v) + default: + return nil, fmt.Errorf("Unrecognized Value Destination: %s in Action", a.ValueDestination) + } + return &bit, err +} diff --git a/input_definition_test.go b/input_definition_test.go index 76b55c7f0..cc4614faf 100644 --- a/input_definition_test.go +++ b/input_definition_test.go @@ -18,9 +18,10 @@ import ( "encoding/json" "testing" + "strings" + "github.com/pilosa/pilosa" "github.com/pilosa/pilosa/internal" - "strings" ) func TestInputDefinition_Open(t *testing.T) { @@ -143,3 +144,92 @@ func TestInputDefinition_LoadDefinition(t *testing.T) { t.Fatalf("Expected frame required error, actual error: %s", err) } } + +func TestHandleAction(t *testing.T) { + var value interface{} + colID := uint64(0) + rowID := uint64(100) + action := pilosa.Action{ValueDestination: pilosa.SingleRowBool, RowID: &rowID} + + value = 1 + b, err := pilosa.HandleAction(action, value, colID) + if b != nil { + t.Fatalf("Expected integer type is not handled by single-row-boolean") + } else if !strings.Contains(err.Error(), "single-row-boolean value") { + t.Fatalf("Expected single-row-boolean value error, actual error: %s", err) + } + + value = "1" + b, err = pilosa.HandleAction(action, value, colID) + if b != nil { + t.Fatalf("Expected Ignore strings, only accept boolean") + } + + value = "t" + b, err = pilosa.HandleAction(action, value, colID) + if !strings.Contains(err.Error(), "must equate to a Bool") { + t.Fatalf("Expected Unrecognized Value Destination error, actual error: %s", err) + } + + value = float64(1.5) + b, err = pilosa.HandleAction(action, value, colID) + if b != nil { + if b.RowID != 100 { + t.Fatalf("Unexpected rowID %v", b.RowID) + } + } + value = float64(0) + b, err = pilosa.HandleAction(action, value, colID) + if b != nil { + t.Fatalf("Expected Ignore values that do not equate to True") + } + + value = false + b, err = pilosa.HandleAction(action, value, colID) + if b != nil { + t.Fatalf("Expected Ignore values that do not equate to True") + } + + value = true + b, err = pilosa.HandleAction(action, value, colID) + if b != nil { + if b.ColumnID != 0 { + t.Fatalf("Unexpected ColumnID %v", b.ColumnID) + } + } + + action.ValueDestination = pilosa.ValueToRow + rowID = 101 + value = float64(25.0) + b, err = pilosa.HandleAction(action, value, colID) + if b != nil { + if b.RowID != 25 { + t.Fatalf("Unexpected RowID %v", b.RowID) + } + } + value = "25" + b, err = pilosa.HandleAction(action, value, colID) + if b != nil { + t.Fatalf("Expected Ignore values that are not type float64") + } + + action.ValueDestination = pilosa.Mapping + value = "test" + b, err = pilosa.HandleAction(action, value, colID) + if b != nil { + t.Fatalf("Expected Ignore values that are not type string") + } + + value = 25 + b, err = pilosa.HandleAction(action, value, colID) + if b != nil { + t.Fatalf("Expected Ignore values that are not type string") + } + + action.ValueDestination = "test" + b, err = pilosa.HandleAction(action, value, colID) + if !strings.Contains(err.Error(), "Unrecognized Value Destination") { + t.Fatalf("Expected Unrecognized Value Destination error, actual error: %s", err) + } + +}