Added new functions

Author: dancannon

datum.go

@@ -1,11 +1,12 @@
 package gorethink
 
 import (
-	"code.google.com/p/goprotobuf/proto"
 	"encoding/json"
 	"fmt"
-	p "github.com/dancannon/gorethink/ql2"
 	"reflect"
+
+	"code.google.com/p/goprotobuf/proto"
+	p "github.com/dancannon/gorethink/ql2"
 )
 
 // Converts a query term to a datum. If the term cannot be converted to a datum
@@ -165,7 +166,7 @@ func deconstructDatum(datum *p.Datum, opts map[string]interface{}) (interface{},
 func convertPseudotype(obj map[string]interface{}, opts map[string]interface{}) (interface{}, error) {
 	if reqlType, ok := obj["$reql_type$"]; ok {
 		if reqlType == "TIME" {
-			// load timeformat, set to native if the option was not set
+			// load timeFormat, set to native if the option was not set
 			timeFormat := "native"
 			if opt, ok := opts["time_format"]; ok {
 				if sopt, ok := opt.(string); ok {
@@ -182,6 +183,24 @@ func convertPseudotype(obj map[string]interface{}, opts map[string]interface{})
 			} else {
 				return nil, fmt.Errorf("Unknown time_format run option \"%s\".", reqlType)
 			}
+		} else if reqlType == "GROUPED_DATA" {
+			// load groupFormat, set to native if the option was not set
+			groupFormat := "native"
+			if opt, ok := opts["group_format"]; ok {
+				if sopt, ok := opt.(string); ok {
+					groupFormat = sopt
+				} else {
+					return nil, fmt.Errorf("Invalid group_format run option \"%s\".", opt)
+				}
+			}
+
+			if groupFormat == "native" {
+				return reqlGroupedDataToObj(obj)
+			} else if groupFormat == "raw" {
+				return obj, nil
+			} else {
+				return nil, fmt.Errorf("Unknown group_format run option \"%s\".", reqlType)
+			}
 		} else {
 			return obj, nil
 		}

example_query_select_test.go

@@ -2,8 +2,9 @@ package gorethink_test
 
 import (
 	"fmt"
-	r "github.com/dancannon/gorethink"
 	"log"
+
+	r "github.com/dancannon/gorethink"
 )
 
 func ExampleRqlTerm_Get() {
@@ -64,6 +65,7 @@ func ExampleRqlTerm_GetAll_compound() {
 	r.Db("test").Table("table").IndexCreateFunc("full_name", func(row r.RqlTerm) interface{} {
 		return []interface{}{row.Field("first_name"), row.Field("last_name")}
 	}).Run(sess)
+	r.Db("test").Table("table").IndexWait().Run(sess)
 
 	// Fetch the row from the database
 	row, err := r.Db("test").Table("table").GetAllByIndex("full_name", []interface{}{"John", "Smith"}).RunRow(sess)

query_aggregation.go

@@ -8,30 +8,42 @@ import (
 // These commands are used to compute smaller values from large sequences.
 
 // Produce a single value from a sequence through repeated application of a
-// reduction function.
-//
-// The reduce function gets invoked repeatedly not only for the input values but
-// also for results of previous reduce invocations. The type and format of the
-// object that is passed in to reduce must be the same with the one returned
-// from reduce.
-func (t RqlTerm) Reduce(f, base interface{}) RqlTerm {
-	return newRqlTermFromPrevVal(t, "Reduce", p.Term_REDUCE, []interface{}{funcWrap(f)}, map[string]interface{}{"base": base})
+// reduction function
+func (t RqlTerm) Reduce(f interface{}) RqlTerm {
+	return newRqlTermFromPrevVal(t, "Reduce", p.Term_REDUCE, []interface{}{funcWrap(f)}, map[string]interface{}{})
 }
 
-// Count the number of elements in the sequence.
-func (t RqlTerm) Count() RqlTerm {
-	return newRqlTermFromPrevVal(t, "Count", p.Term_COUNT, []interface{}{}, map[string]interface{}{})
+// Remove duplicate elements from the sequence.
+func (t RqlTerm) Distinct() RqlTerm {
+	return newRqlTermFromPrevVal(t, "Distinct", p.Term_DISTINCT, []interface{}{}, map[string]interface{}{})
 }
 
-// Count the number of elements in the sequence. CountFiltered uses the argument
-// passed to it to filter the results before counting.
-func (t RqlTerm) CountFiltered(f interface{}) RqlTerm {
-	return newRqlTermFromPrevVal(t, "Count", p.Term_COUNT, []interface{}{funcWrap(f)}, map[string]interface{}{})
+// Takes a stream and partitions it into multiple groups based on the
+// fields or functions provided. Commands chained after group will be
+//  called on each of these grouped sub-streams, producing grouped data.
+func (t RqlTerm) Group(fieldOrFunctions ...interface{}) RqlTerm {
+	for k, v := range fieldOrFunctions {
+		fieldOrFunctions[k] = funcWrap(v)
+	}
+
+	return newRqlTermFromPrevVal(t, "Group", p.Term_GROUP, fieldOrFunctions, map[string]interface{}{})
 }
 
-// Remove duplicate elements from the sequence.
-func (t RqlTerm) Distinct() RqlTerm {
-	return newRqlTermFromPrevVal(t, "Distinct", p.Term_DISTINCT, []interface{}{}, map[string]interface{}{})
+// Takes a stream and partitions it into multiple groups based on the
+// fields or functions provided. Commands chained after group will be
+// called on each of these grouped sub-streams, producing grouped data.
+func (t RqlTerm) GroupByIndex(index interface{}, fieldOrFunctions ...interface{}) RqlTerm {
+	for k, v := range fieldOrFunctions {
+		fieldOrFunctions[k] = funcWrap(v)
+	}
+
+	return newRqlTermFromPrevVal(t, "Group", p.Term_GROUP, fieldOrFunctions, map[string]interface{}{
+		"index": index,
+	})
+}
+
+func (t RqlTerm) Ungroup() RqlTerm {
+	return newRqlTermFromPrevVal(t, "Ungroup", p.Term_UNGROUP, []interface{}{}, map[string]interface{}{})
 }
 
 //Returns whether or not a sequence contains all the specified values, or if
@@ -48,23 +60,65 @@ func (t RqlTerm) Contains(args ...interface{}) RqlTerm {
 // Aggregators
 // These standard aggregator objects are to be used in conjunction with group_by.
 
-// Count the total size of the group.
-func Count() RqlTerm {
-	return Expr(map[string]interface{}{
-		"COUNT": true,
-	})
+// Count the number of elements in the sequence. With a single argument,
+// count the number of elements equal to it. If the argument is a function,
+// it is equivalent to calling filter before count.
+func (t RqlTerm) Count(filters ...interface{}) RqlTerm {
+	for k, v := range filters {
+		filters[k] = funcWrap(v)
+	}
+
+	return newRqlTermFromPrevVal(t, "Count", p.Term_COUNT, filters, map[string]interface{}{})
 }
 
-// Compute the sum of the given field in the group.
-func Sum(arg interface{}) RqlTerm {
-	return Expr(map[string]interface{}{
-		"SUM": arg,
-	})
+// Sums all the elements of a sequence. If called with a field name, sums all
+// the values of that field in the sequence, skipping elements of the sequence
+// that lack that field. If called with a function, calls that function on every
+// element of the sequence and sums the results, skipping elements of the
+// sequence where that function returns null or a non-existence error.
+func (t RqlTerm) Sum(fieldOrFunctions ...interface{}) RqlTerm {
+	for k, v := range fieldOrFunctions {
+		fieldOrFunctions[k] = funcWrap(v)
+	}
+
+	return newRqlTermFromPrevVal(t, "Sum", p.Term_SUM, fieldOrFunctions, map[string]interface{}{})
 }
 
-// Compute the average value of the given attribute for the group.
-func Avg(arg interface{}) RqlTerm {
-	return Expr(map[string]interface{}{
-		"AVG": arg,
-	})
+// Averages all the elements of a sequence. If called with a field name, averages
+// all the values of that field in the sequence, skipping elements of the sequence
+// that lack that field. If called with a function, calls that function on every
+// element of the sequence and averages the results, skipping elements of the
+// sequence where that function returns null or a non-existence error.
+func (t RqlTerm) Avg(fieldOrFunctions ...interface{}) RqlTerm {
+	for k, v := range fieldOrFunctions {
+		fieldOrFunctions[k] = funcWrap(v)
+	}
+
+	return newRqlTermFromPrevVal(t, "Sum", p.Term_SUM, fieldOrFunctions, map[string]interface{}{})
+}
+
+// Finds the minimum of a sequence. If called with a field name, finds the element
+// of that sequence with the smallest value in that field. If called with a function,
+// calls that function on every element of the sequence and returns the element
+// which produced the smallest value, ignoring any elements where the function
+// returns null or produces a non-existence error.
+func (t RqlTerm) Min(fieldOrFunctions ...interface{}) RqlTerm {
+	for k, v := range fieldOrFunctions {
+		fieldOrFunctions[k] = funcWrap(v)
+	}
+
+	return newRqlTermFromPrevVal(t, "Min", p.Term_MIN, fieldOrFunctions, map[string]interface{}{})
+}
+
+// Finds the maximum of a sequence. If called with a field name, finds the element
+// of that sequence with the largest value in that field. If called with a function,
+// calls that function on every element of the sequence and returns the element
+// which produced the largest value, ignoring any elements where the function
+// returns null or produces a non-existence error.
+func (t RqlTerm) Max(fieldOrFunctions ...interface{}) RqlTerm {
+	for k, v := range fieldOrFunctions {
+		fieldOrFunctions[k] = funcWrap(v)
+	}
+
+	return newRqlTermFromPrevVal(t, "Max", p.Term_MAX, fieldOrFunctions, map[string]interface{}{})
 }