Merge pull request #4 from bobrik/aggregations

Aggregations support

goes.go

@@ -348,3 +348,34 @@ func (r *Request) Url() string {
 
 	return u.String()
 }
+
+// Buckets returns list of buckets in aggregation
+func (a Aggregation) Buckets() []Bucket {
+	result := []Bucket{}
+	if buckets, ok := a["buckets"]; ok {
+		for _, bucket := range buckets.([]interface {}) {
+			result = append(result, bucket.(map[string]interface{}))
+		}
+	}
+
+	return result
+}
+
+// Key returns key for aggregation bucket
+func (b Bucket) Key() interface{} {
+	return b["key"]
+}
+
+// DocCount returns count of documents in this bucket
+func (b Bucket) DocCount() uint64 {
+	return uint64(b["doc_count"].(float64))
+}
+
+// Aggregation returns aggregation by name from bucket
+func (b Bucket) Aggregation(name string) Aggregation{
+	if agg, ok := b[name]; ok {
+		return agg.(map[string]interface{})
+	} else {
+		return Aggregation{}
+	}
+}

goes_test.go

@@ -705,3 +705,114 @@ func (s *GoesTestSuite) TestScroll(c *C) {
 	c.Assert(len(searchResults.ScrollId) > 0, Equals, true)
 	c.Assert(len(searchResults.Hits.Hits), Equals, 0)
 }
+
+
+func (s *GoesTestSuite) TestAggregations(c *C) {
+	indexName := "testaggs"
+	docType := "tweet"
+
+	tweets := []Document{
+		Document{
+			Id:          nil,
+			Index:       indexName,
+			Type:        docType,
+			BulkCommand: BULK_COMMAND_INDEX,
+			Fields: map[string]interface{}{
+				"user"    : "foo",
+				"message" : "some foo message",
+				"age"     : 25,
+			},
+		},
+
+		Document{
+			Id:          nil,
+			Index:       indexName,
+			Type:        docType,
+			BulkCommand: BULK_COMMAND_INDEX,
+			Fields: map[string]interface{}{
+				"user"    : "bar",
+				"message" : "some bar message",
+				"age"     : 30,
+			},
+		},
+
+		Document{
+			Id:          nil,
+			Index:       indexName,
+			Type:        docType,
+			BulkCommand: BULK_COMMAND_INDEX,
+			Fields: map[string]interface{}{
+				"user"    :    "foo",
+				"message" : "another foo message",
+			},
+		},
+	}
+
+	conn := NewConnection(ES_HOST, ES_PORT)
+
+	mapping := map[string]interface{}{
+		"settings": map[string]interface{}{
+			"index.number_of_shards":   1,
+			"index.number_of_replicas": 0,
+		},
+	}
+
+	defer conn.DeleteIndex(indexName)
+
+	_, err := conn.CreateIndex(indexName, mapping)
+	c.Assert(err, IsNil)
+
+	_, err = conn.BulkSend(indexName, tweets)
+	c.Assert(err, IsNil)
+
+	_, err = conn.RefreshIndex(indexName)
+	c.Assert(err, IsNil)
+
+	query := map[string]interface{}{
+		"aggs": map[string]interface{}{
+			"user": map[string]interface{}{
+				"terms": map[string]interface{}{
+					"field": "user",
+					"order": map[string]interface{}{
+						"_term": "asc",
+					},
+				},
+				"aggs": map[string]interface{}{
+					"age": map[string]interface{}{
+						"stats": map[string]interface{}{
+							"field": "age",
+						},
+					},
+				},
+			},
+			"age": map[string]interface{}{
+				"stats": map[string]interface{}{
+					"field": "age",
+				},
+			},
+		},
+	}
+
+	resp, err := conn.Search(query, []string{indexName}, []string{docType})
+
+	user, ok := resp.Aggregations["user"]
+	c.Assert(ok, Equals, true)
+
+	c.Assert(len(user.Buckets()), Equals, 2)
+	c.Assert(user.Buckets()[0].Key(), Equals, "bar")
+	c.Assert(user.Buckets()[1].Key(), Equals, "foo")
+
+	barAge := user.Buckets()[0].Aggregation("age")
+	c.Assert(barAge["count"], Equals, 1.0)
+	c.Assert(barAge["sum"], Equals, 30.0)
+
+	fooAge := user.Buckets()[1].Aggregation("age")
+	c.Assert(fooAge["count"], Equals, 1.0)
+	c.Assert(fooAge["sum"], Equals, 25.0)
+
+	age, ok := resp.Aggregations["age"]
+	c.Assert(ok, Equals, true)
+
+	c.Assert(age["count"], Equals, 2.0)
+	c.Assert(age["sum"], Equals, 25.0 + 30.0)
+}

structs.go

@@ -80,8 +80,16 @@ type Response struct {
 
 	// Scroll id for iteration
 	ScrollId string `json:"_scroll_id"`
+
+	Aggregations map[string]Aggregation `json:"aggregations,omitempty"`
 }
 
+// Represents an aggregation from response
+type Aggregation map[string]interface{}
+
+// Represents a bucket for aggregation
+type Bucket map[string]interface{}
+
 // Represents a document to send to elasticsearch
 type Document struct {
 	// XXX : interface as we can support nil values