Add query functions, refactor.

README.md

@@ -16,11 +16,11 @@ client, err := NewClient(config)
 Evaluate a segment's conditions and retrieve its members.
 
 ```golang
-gte := Gte{
+gte := segment.Gte{
     PartitionID: "sorted_set:ad43bf8e-0f0c-4102-be91-52bc84150af2:current_balances:flipside",
     Value:       10000000,
 }
-condition := Condition{
+condition := segment.Condition{
     Gte: gte,
 }
 
@@ -35,11 +35,11 @@ Identify the intersection of an array of members to a segment's evaluated condit
 intersectMembers := make([]string, 0)
 intersectMembers = append(intersectMembers, "a090b025a1489aa6c9204d7b85ac77d51b814402d5cbdec27335575bb46e4f20")
 
-gte := Gte{
+gte := segment.Gte{
     PartitionID: "sorted_set:ad43bf8e-0f0c-4102-be91-52bc84150af2:current_balances:flipside",
     Value:       10000000,
 }
-condition := Condition{
+condition := segment.Condition{
     Gte: gte,
 }
 
@@ -57,6 +57,59 @@ memberID := "a0969f676e0274c34fffb4261b59d3de48de0d5845ed9780ac43045cf954ed81"
 result, err := client.GetMemberPartitions(entityID, memberID)
 ```
 
+### ExecuteDynamicQuery
+
+Execute a query.
+
+```golang
+gte := segment.Gte{
+    PartitionID: "sorted_set:ad43bf8e-0f0c-4102-be91-52bc84150af2:current_balances:flipside",
+    Value:       10000000,
+}
+
+segments := make(map[string]segment.Condition)
+segments["large_balance_holder"] = segment.Condition{
+    Gte: gte,
+}
+
+aggregates := make([]dynamicquery.Aggregate, 0)
+aggregates = append(aggregates, dynamicquery.Aggregate{
+    Field:             "event_amount",
+    Label:             "total_amount",
+    DecimalAdjustment: 16,
+    Operation:         "sum",
+})
+
+groupBy := make([]dynamicquery.GroupBy, 0)
+groupBy = append(groupBy, dynamicquery.GroupBy{
+    Field:      "block_timestamp",
+    Timebucket: "1 day",
+    Label:      "metric_date",
+})
+
+inSegment := dynamicquery.InSegment{
+    Field: "event_to",
+    Value: "large_balance_holder",
+}
+filter := dynamicquery.Filter{
+    InSegment: inSegment,
+}
+
+query := dynamicquery.Query{
+    Table:      "udm_events_aion",
+    Schema:     "source",
+    Filter:     filter,
+    GroupBy:    groupBy,
+    Aggregates: aggregates,
+    Segments:   segments,
+}
+
+debug := false
+
+result, err := client.ExecuteDynamicQuery(query, debug)
+
+```
+
 ### Get Datasets
 
 Retreive available datasets and apply optional filters (`entityID`, `ownerID`).

client_test.go

@@ -4,19 +4,60 @@ import (
 	"fmt"
 	"os"
 	"testing"
+
+	"github.com/FlipsideCrypto/flip-rpc-client-go/dynamicquery"
+	"github.com/FlipsideCrypto/flip-rpc-client-go/segment"
 )
 
-func makeCondition() Condition {
-	gte := Gte{
+func makeCondition() segment.Condition {
+	gte := segment.Gte{
 		PartitionID: "sorted_set:ad43bf8e-0f0c-4102-be91-52bc84150af2:current_balances:flipside",
-		Value:       10000000,
+		Value:       100000000,
 	}
-	c := Condition{
+	c := segment.Condition{
 		Gte: gte,
 	}
 	return c
 }
 
+func makeQuery(condition segment.Condition) dynamicquery.Query {
+	segments := make(map[string]segment.Condition)
+	segments["large_balance_holder"] = condition
+
+	aggregates := make([]dynamicquery.Aggregate, 0)
+	aggregates = append(aggregates, dynamicquery.Aggregate{
+		Field:             "event_amount",
+		Label:             "total_amount",
+		DecimalAdjustment: 16,
+		Operation:         "sum",
+	})
+
+	groupBy := make([]dynamicquery.GroupBy, 0)
+	groupBy = append(groupBy, dynamicquery.GroupBy{
+		Field:      "block_timestamp",
+		Timebucket: "1 day",
+		Label:      "metric_date",
+	})
+
+	inSegment := dynamicquery.InSegment{
+		Field: "event_to",
+		Value: "large_balance_holder",
+	}
+	filter := dynamicquery.Filter{
+		InSegment: inSegment,
+	}
+
+	query := dynamicquery.Query{
+		Table:      "udm_events_aion",
+		Schema:     "source",
+		Filter:     filter,
+		GroupBy:    groupBy,
+		Aggregates: aggregates,
+		Segments:   segments,
+	}
+	return query
+}
+
 func TestClient_GetSegmentMembers(t *testing.T) {
 	client := getClient(t)
 
@@ -52,6 +93,22 @@ func TestClient_IntersectMembersToSegment(t *testing.T) {
 	fmt.Println("")
 }
 
+func TestClient_ExecuteDynamicQuery(t *testing.T) {
+	client := getClient(t)
+
+	c, err := client.ExecuteDynamicQuery(makeQuery(makeCondition()), false)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if c == nil {
+		t.Fatal("count is nil")
+	}
+
+	fmt.Fprintln(os.Stdout, "ExecuteDynamicQuery")
+	fmt.Fprintln(os.Stdout, *c)
+	fmt.Println("")
+}
+
 func TestClient_GetMemberPartitions(t *testing.T) {
 	client := getClient(t)
 	entityID := "ad43bf8e-0f0c-4102-be91-52bc84150af2"

dynamic_query.go

@@ -0,0 +1,35 @@
+package flip
+
+import (
+	"github.com/FlipsideCrypto/flip-rpc-client-go/dynamicquery"
+	"github.com/mitchellh/mapstructure"
+	"github.com/pkg/errors"
+)
+
+// ExecuteDynamicQueryResponse returns the RPC response
+type ExecuteDynamicQueryResponse struct {
+	Results     []interface{} `mapstructure:"results"`
+	ResultCount int           `mapstructure:"result_count"`
+	CompiledSQL string        `mapstructure:"compiled_sql"`
+}
+
+// ExecuteDynamicQuery returns the query results
+func (c Client) ExecuteDynamicQuery(query dynamicquery.Query, debug bool) (*ExecuteDynamicQueryResponse, error) {
+	var input = make(map[string]interface{})
+	input["query"] = query
+	input["debug"] = debug
+
+	var response ExecuteDynamicQueryResponse
+
+	result, err := c.CallRPC("RPCService.ExecuteDynamicQuery", input)
+	if err != nil {
+		return &response, err
+	}
+
+	err = mapstructure.Decode(result, &response)
+	if err != nil {
+		return &response, errors.Wrap(err, "error decoding into `ExecuteDynamicQueryResponse`")
+	}
+
+	return &response, nil
+}

dynamicquery/aggregate.go

@@ -0,0 +1,9 @@
+package dynamicquery
+
+// Aggregate builds aggregate operations
+type Aggregate struct {
+	Field             string `json:"field"`
+	Label             string `json:"label"`
+	Operation         string `json:"operation"`
+	DecimalAdjustment int    `json:"decimal_adjustment"`
+}

dynamicquery/filter.go

@@ -0,0 +1,63 @@
+package dynamicquery
+
+// Filter --
+type Filter struct {
+	In           In           `json:"in"`
+	NotIn        NotIn        `json:"not_in"`
+	InSegment    InSegment    `json:"in_segment"`
+	NotInSegment NotInSegment `json:"not_in_segment"`
+	Gte          Gte          `json:"gte"`
+	Lte          Lte          `json:"lte"`
+	Gt           Gt           `json:"gt"`
+	Lt           Lt           `json:"lt"`
+	And          []*Filter    `json:"and"`
+	Or           []*Filter    `json:"or"`
+}
+
+// Gte greater than or equal to filter
+type Gte struct {
+	Field string      `json:"field"`
+	Value interface{} `json:"value"`
+}
+
+// Lte less than or equal to filter
+type Lte struct {
+	Field string      `json:"field"`
+	Value interface{} `json:"value"`
+}
+
+// Gt greater than filter
+type Gt struct {
+	Field string      `json:"field"`
+	Value interface{} `json:"value"`
+}
+
+// Lt less than filter
+type Lt struct {
+	Field string      `json:"field"`
+	Value interface{} `json:"value"`
+}
+
+// In in filter
+type In struct {
+	Field string   `json:"field"`
+	Value []string `json:"value"`
+}
+
+// NotIn not in filter
+type NotIn struct {
+	Field string   `json:"field"`
+	Value []string `json:"value"`
+}
+
+// InSegment in filter
+type InSegment struct {
+	Field string `json:"field"`
+	Value string `json:"value"`
+}
+
+// NotInSegment not in filter
+type NotInSegment struct {
+	Field string `json:"field"`
+	Value string `json:"value"`
+}

dynamicquery/group_by.go

@@ -0,0 +1,8 @@
+package dynamicquery
+
+// GroupBy a set of fields
+type GroupBy struct {
+	Field      string `json:"field"`
+	Label      string `json:"label"`
+	Timebucket string `json:"timebucket"`
+}

dynamicquery/order_by.go

@@ -0,0 +1,7 @@
+package dynamicquery
+
+// OrderBy a set of fields
+type OrderBy struct {
+	Label     string `json:"label"`
+	Direction string `json:"direction"`
+}

dynamicquery/query.go

@@ -0,0 +1,15 @@
+package dynamicquery
+
+import "github.com/FlipsideCrypto/flip-rpc-client-go/segment"
+
+// Query user inputs to generate a SQL query
+type Query struct {
+	Table      string                       `json:"table"`
+	Schema     string                       `json:"schema"`
+	Aggregates []Aggregate                  `json:"aggregates"`
+	GroupBy    []GroupBy                    `json:"group_by"`
+	Filter     Filter                       `json:"filter"`
+	OrderBy    []OrderBy                    `json:"order_by"`
+	Limit      int                          `json:"limit"`
+	Segments   map[string]segment.Condition `json:"segments"`
+}

segment.go

@@ -1,47 +1,11 @@
 package flip
 
 import (
+	"github.com/FlipsideCrypto/flip-rpc-client-go/segment"
 	"github.com/mitchellh/mapstructure"
 	"github.com/pkg/errors"
 )
 
-// Gte = greater than or equal to
-type Gte struct {
-	PartitionID string  `json:"partition_id"`
-	Value       float64 `json:"value"`
-}
-
-// Lte = less than or equal to
-type Lte struct {
-	PartitionID string  `json:"partition_id"`
-	Value       float64 `json:"value"`
-}
-
-// Lt = less than
-type Lt struct {
-	PartitionID string  `json:"partition_id"`
-	Value       float64 `json:"value"`
-}
-
-// Gt = greater than
-type Gt struct {
-	PartitionID string  `json:"partition_id"`
-	Value       float64 `json:"value"`
-}
-
-// Condition is set of logic
-type Condition struct {
-	PartitionID string  `json:"partition_id"`
-	Value       float64 `json:"value"`
-
-	Or  []*Condition `json:"or"`
-	And []*Condition `json:"and"`
-	Gt  Gt           `json:"gt"`
-	Gte Gte          `json:"gte"`
-	Lt  Lt           `json:"lt"`
-	Lte Lte          `json:"lte"`
-}
-
 // GetSegmentMembersResponse returns the RPC response
 type GetSegmentMembersResponse struct {
 	Members     []string `mapstructure:"members"`
@@ -49,8 +13,8 @@ type GetSegmentMembersResponse struct {
 }
 
 // GetSegmentMembers returns the members belonging to the result set of a condition.
-func (c Client) GetSegmentMembers(condition Condition) (*GetSegmentMembersResponse, error) {
-	var input = make(map[string]Condition)
+func (c Client) GetSegmentMembers(condition segment.Condition) (*GetSegmentMembersResponse, error) {
+	var input = make(map[string]segment.Condition)
 	input["segment"] = condition
 
 	var segmentMembers GetSegmentMembersResponse
@@ -76,7 +40,7 @@ type IntersectMembersToSegmentResponse struct {
 }
 
 // IntersectMembersToSegment returns the intersection between a set of inputs against a segment formed by conditions
-func (c Client) IntersectMembersToSegment(members []string, condition Condition) (*IntersectMembersToSegmentResponse, error) {
+func (c Client) IntersectMembersToSegment(members []string, condition segment.Condition) (*IntersectMembersToSegmentResponse, error) {
 	var input = make(map[string]interface{})
 	input["segment"] = condition
 	input["members"] = members

segment/condition.go

@@ -0,0 +1,37 @@
+package segment
+
+// Gte = greater than or equal to
+type Gte struct {
+	PartitionID string  `json:"partition_id"`
+	Value       float64 `json:"value"`
+}
+
+// Lte = less than or equal to
+type Lte struct {
+	PartitionID string  `json:"partition_id"`
+	Value       float64 `json:"value"`
+}
+
+// Lt = less than
+type Lt struct {
+	PartitionID string  `json:"partition_id"`
+	Value       float64 `json:"value"`
+}
+
+// Gt = greater than
+type Gt struct {
+	PartitionID string  `json:"partition_id"`
+	Value       float64 `json:"value"`
+}
+
+// Condition is set of logic
+type Condition struct {
+	PartitionID string       `json:"partition_id"`
+	Value       float64      `json:"value"`
+	Or          []*Condition `json:"or"`
+	And         []*Condition `json:"and"`
+	Gt          Gt           `json:"gt"`
+	Gte         Gte          `json:"gte"`
+	Lt          Lt           `json:"lt"`
+	Lte         Lte          `json:"lte"`
+}