merge action's processing

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
+}

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