refs #5137: Updated GroupDocumentSource to 2.5.4

lib/pipeline/accumulators/AddToSetAccumulator.js

@@ -2,7 +2,7 @@
 
 /** 
  * Create an expression that finds the sum of n operands.
- * @class AddSoSetAccumulator
+ * @class AddToSetAccumulator
  * @namespace mungedb-aggregate.pipeline.accumulators
  * @module mungedb-aggregate
  * @constructor
@@ -15,11 +15,16 @@ var AddToSetAccumulator = module.exports = function AddToSetAccumulator(/* ctx *
 	base.call(this);
 }, klass = AddToSetAccumulator, Accumulator = require("./Accumulator"), base = Accumulator, proto = klass.prototype = Object.create(base.prototype, {constructor:{value:klass}});
 
-//DEPENDENCIES
+// NOTE: Skipping the create function, using the constructor instead
+
+// DEPENDENCIES
 var Value = require("../Value");
 
-proto.getFactory = function getFactory(){
-	return klass;	// using the ctor rather than a separate .create() method
+
+// MEMBER FUNCTIONS
+
+proto.getOpName = function getOpName(){
+	return "$addToSet";
 };
 
 proto.contains = function contains(value) {
@@ -46,7 +51,4 @@ proto.reset = function reset() {
     this.set = [];
 };
 
-// PROTOTYPE MEMBERS
-proto.getOpName = function getOpName(){
-	return "$addToSet";
-};
+

lib/pipeline/accumulators/AvgAccumulator.js

@@ -16,17 +16,12 @@ var AvgAccumulator = module.exports = function AvgAccumulator(){
 	base.call(this);
 }, klass = AvgAccumulator, Accumulator = require("./Accumulator"), base = Accumulator, proto = klass.prototype = Object.create(base.prototype, {constructor:{value:klass}});
 
-//DEPENDENCIES
-var Value = require("../Value");
-
-klass.create = function create() {
-	return new AvgAccumulator();
-};
+// NOTE: Skipping the create function, using the constructor instead
 
-proto.getFactory = function getFactory(){
-	return klass;	// using the ctor rather than a separate .create() method
-};
+// DEPENDENCIES
+var Value = require("../Value");
 
+// MEMBER FUNCTIONS
 proto.processInternal = function processInternal(input, merging) {
 	if (!merging) {
 		if (typeof input !== "number") {

lib/pipeline/accumulators/FirstAccumulator.js

@@ -14,15 +14,13 @@ var FirstAccumulator = module.exports = function FirstAccumulator(){
 	this._first = undefined;
 }, klass = FirstAccumulator, base = require("./Accumulator"), proto = klass.prototype = Object.create(base.prototype, {constructor:{value:klass}});
 
-// PROTOTYPE MEMBERS
+// NOTE: Skipping the create function, using the constructor instead
+
+// MEMBER FUNCTIONS
 proto.getOpName = function getOpName(){
 	return "$first";
 };
 
-proto.getFactory = function getFactory(){
-	return klass;	// using the ctor rather than a separate .create() method
-};
-
 proto.processInternal = function processInternal(input, merging) {
 	/* only remember the first value seen */
 	if (!this._haveFirst) {

lib/pipeline/accumulators/LastAccumulator.js

@@ -12,14 +12,13 @@ var LastAccumulator = module.exports = function LastAccumulator(){
 	this.value = undefined;
 }, klass = LastAccumulator, base = require("./Accumulator"), proto = klass.prototype = Object.create(base.prototype, {constructor:{value:klass}});
 
+// NOTE: Skipping the create function, using the constructor instead
+
+// MEMBER FUNCTIONS
 proto.processInternal = function processInternal(input, merging){
 	this.value = input;
 };
 
-proto.getFactory = function getFactory(){
-	return klass;	// using the ctor rather than a separate .create() method
-};
-
 proto.getValue = function getValue() {
 	return this.value;
 };

lib/pipeline/accumulators/MinMaxAccumulator.js

@@ -14,10 +14,12 @@ var MinMaxAccumulator = module.exports = function MinMaxAccumulator(sense){
 	if (this.sense !== 1 && this.sense !== -1) throw new Error("this should never happen");
 }, klass = MinMaxAccumulator, base = require("./Accumulator"), proto = klass.prototype = Object.create(base.prototype, {constructor:{value:klass}});
 
+// NOTE: Skipping the create function, using the constructor instead
+
 // DEPENDENCIES
 var Value = require("../Value");
 
-// PROTOTYPE MEMBERS
+// MEMBER FUNCTIONS
 proto.getOpName = function getOpName(){
 	if (this.sense == 1) return "$min";
 	return "$max";

lib/pipeline/accumulators/PushAccumulator.js

@@ -12,14 +12,13 @@ var PushAccumulator = module.exports = function PushAccumulator(){
 	base.call(this);
 }, klass = PushAccumulator, Accumulator = require("./Accumulator"), base = Accumulator, proto = klass.prototype = Object.create(base.prototype, {constructor:{value:klass}});
 
+// NOTE: Skipping the create function, using the constructor instead
+
+// MEMBER FUNCTIONS
 proto.getValue = function getValue(toBeMerged){
 	return this.values;
 };
 
-proto.getFactory = function getFactory(){
-	return klass;	// using the ctor rather than a separate .create() method
-};
-
 proto.getOpName = function getOpName(){
 	return "$push";
 };

lib/pipeline/accumulators/SumAccumulator.js

@@ -14,6 +14,9 @@ var SumAccumulator = module.exports = function SumAccumulator(){
 	base.call(this);
 }, klass = SumAccumulator, Accumulator = require("./Accumulator"), base = Accumulator, proto = klass.prototype = Object.create(base.prototype, {constructor:{value:klass}});
 
+// NOTE: Skipping the create function, using the constructor instead
+
+// MEMBER FUNCTIONS
 proto.processInternal = function processInternal(input, merging) {
 	if(typeof input === "number"){ // do nothing with non-numeric types
 		this.totalIsANumber = true;

lib/pipeline/documentSources/GroupDocumentSource.js

@@ -4,16 +4,20 @@ var DocumentSource = require("./DocumentSource"),
 	Document = require("../Document"),
 	Expression = require("../expressions/Expression"),
 	ConstantExpression = require("../expressions/ConstantExpression"),
-	FieldPathExpression = require("../expressions/FieldPathExpression");
-
+	FieldPathExpression = require("../expressions/FieldPathExpression"),
+	Variables = require("../expressions/Variables"),
+	VariablesIdGenerator = require("../expressions/VariablesIdGenerator"),
+	VariablesParseState = require("../expressions/VariablesParseState"),
+	async = require("async");
 
 /**
  * A class for grouping documents together
+ *
  * @class GroupDocumentSource
  * @namespace mungedb-aggregate.pipeline.documentSources
  * @module mungedb-aggregate
  * @constructor
- * @param [ctx] {ExpressionContext}
+ * @param [expCtx] {ExpressionContext}
  **/
 var GroupDocumentSource = module.exports = function GroupDocumentSource(expCtx) {
 	if (arguments.length > 1) throw new Error("up to one arg expected");
@@ -24,7 +28,7 @@ var GroupDocumentSource = module.exports = function GroupDocumentSource(expCtx)
 	this.groups = {}; // GroupsType Value -> Accumulators[]
 	this.groupsKeys = []; // This is to faciliate easier look up of groups
 	this.originalGroupsKeys = []; // This stores the original group key un-hashed/stringified/whatever
-
+	this._variables = null;
 	this.fieldNames = [];
 	this.accumulatorFactories = [];
 	this.expressions = [];
@@ -38,7 +42,7 @@ klass.groupOps = {
 	"$avg": Accumulators.Avg,
 	"$first": Accumulators.First,
 	"$last": Accumulators.Last,
-	"$max": Accumulators.MinMax.createMax,
+	"$max": Accumulators.MinMax.createMax, // $min and $max have special constructors because they share base features
 	"$min": Accumulators.MinMax.createMin,
 	"$push": Accumulators.Push,
 	"$sum": Accumulators.Sum
@@ -46,43 +50,142 @@ klass.groupOps = {
 
 klass.groupName = "$group";
 
+/**
+ * Factory for making GroupDocumentSources
+ *
+ * @method create
+ * @static
+ * @param [expCtx] {ExpressionContext}
+ **/
+klass.create = function create(expCtx) {
+	return new GroupDocumentSource(expCtx);
+};
+
+/**
+ * Factory for making GroupDocumentSources
+ *
+ * @method getSourceName
+ * @return {GroupDocumentSource}
+ **/
 proto.getSourceName = function getSourceName() {
 	return klass.groupName;
 };
 
 /**
- * Create an object that represents the document source.  The object
- * will have a single field whose name is the source's name.  This
- * will be used by the default implementation of addToJsonArray()
- * to add this object to a pipeline being represented in JSON.
+ * Gets the next document or DocumentSource.EOF if none
  *
- * @method	sourceToJson
- * @param	{Object} builder	JSONObjBuilder: a blank object builder to write to
- * @param	{Boolean}	explain	create explain output
+ * @method getNext
+ * @return {Object}
  **/
-proto.sourceToJson = function sourceToJson(builder, explain) {
-	var idExp = this.idExpression,
-		insides = {
-			_id: idExp ? idExp.toJSON() : {}
+proto.getNext = function getNext(callback) {
+	var self = this;
+	async.series([
+		function(next) {
+			if (!self.populated)
+				self.populate(function(err) {
+					return next(err);
+				});
+			else
+				return next();
 		},
-		aFac = this.accumulatorFactories,
-		aFacLen = aFac.length;
+		function(next) {
+			if(Object.keys(self.groups).length === 0) {
+				return next(null, DocumentSource.EOF);
+			}
+
+			//Note: Skipped the spilled logic
+
+			if(self.currentGroupsKeysIndex === self.groupsKeys.length) {
+				return next(null, DocumentSource.EOF);
+			}
+
+			var id = self.groupsKeys[self.currentGroupsKeysIndex],
+				accumulators = self.groups[id],
+				out = self.makeDocument(id, accumulators /*,mergeableOutput*/);
+
+			if(++self.currentGroupsKeysIndex === self.groupsKeys.length) {
+				self.dispose();
+			}
+
+			return next(null, out);
+		}
+	], function(err, results) {
+		callback(err, results[1]);
+	});
+};
+
+/**
+ * Sets this source as apparently empty
+ *
+ * @method dispose
+ **/
+proto.dispose = function dispose() {
+	//NOTE: Skipped 'freeing' our resources; at best we could remove some references to things, but our parent will probably forget us anyways!
 
-	for(var i=0; i < aFacLen; ++i) {
-		var acc = new aFac[i](/*pExpCtx*/);
-		acc.addOperand(this.expressions[i]);
+	// make us look done
+	this.currentGroupsKeysIndex = this.groupsKeys.length;
 
-		insides[this.fieldNames[i]] = acc.toJSON(true);
+	// free our source's resources
+	this.source.dispose();
+};
+
+/**
+ * Optimizes the expressions in the group
+ * @method optimize
+ **/
+proto.optimize = function optimize() {
+	var self = this;
+	self.idExpression = self.idExpression.optimize();
+	self.expressions.forEach(function(expression, i) {
+		self.expressions[i] = expression.optimize();
+	});
+};
+
+/**
+ * Create an object that represents the document source.  The object
+ * will have a single field whose name is the source's name.
+ *
+ * @method	serialize
+ * @param explain {Boolean} Create explain output
+ **/
+proto.serialize = function serialize(explain) {
+	var insides = {};
+
+	// add the _id
+	insides._id = this.idExpression.serialize(explain);
+
+	//add the remaining fields
+	var aFacs = this.accumulatorFactories,
+		aFacLen = aFacs.length;
+
+	for(var i=0; i < aFacLen; i++) {
+		var aFac = aFacs[i](),
+			serialExpression = this.expressions[i].serialize(explain), //Get the accumulator's expression
+			serialAccumulator = {}; //Where we'll put the expression
+		serialAccumulator[aFac.getOpName()] = serialExpression;
+		insides[this.fieldNames[i]] = serialAccumulator;
 	}
 
-	builder[this.getSourceName()] = insides;
+	var serialSource = {};
+	serialSource[this.getSourceName()] = insides;
+	return serialSource;
 };
 
-klass.createFromJson = function createFromJson(groupObj, ctx) {
-	if (!(groupObj instanceof Object && groupObj.constructor === Object)) throw new Error("a group's fields must be specified in an object");
+/**
+ * Creates a GroupDocumentSource from the given elem
+ *
+ * @method	createFromJson
+ * @param elem {Object} The group specification object; the right hand side of the $group
+ **/
+klass.createFromJson = function createFromJson(elem, expCtx) {
+	if (!(elem instanceof Object && elem.constructor === Object)) throw new Error("a group's fields must be specified in an object");
+
+	var group = GroupDocumentSource.create(expCtx),
+		idSet = false;
 
-	var idSet = false,
-		group = new GroupDocumentSource(ctx);
+	var groupObj = elem,
+		idGenerator = new VariablesIdGenerator(),
+		vps = new VariablesParseState(idGenerator);
 
 	for (var groupFieldName in groupObj) {
 		if (groupObj.hasOwnProperty(groupFieldName)) {
@@ -93,106 +196,168 @@ klass.createFromJson = function createFromJson(groupObj, ctx) {
 				if(idSet) throw new Error("15948 a group's _id may only be specified once");
 
 				if (groupField instanceof Object && groupField.constructor === Object) {
+					/*
+						Use the projection-like set of field paths to create the
+						group-by key.
+					*/
 					var objCtx = new Expression.ObjectCtx({isDocumentOk:true});
-					group.idExpression = Expression.parseObject(groupField, objCtx);
+					group.setIdExpression(Expression.parseObject(groupField, objCtx, vps));
 					idSet = true;
 
 				} else if (typeof groupField === "string") {
-					if (groupField[0] !== "$") {
-						group.idExpression = new ConstantExpression(groupField);
-					} else {
-						var pathString = Expression.removeFieldPrefix(groupField);
-						group.idExpression = new FieldPathExpression(pathString);
-					}
-					idSet = true;
-
-				} else {
-					var typeStr = group._getTypeStr(groupField);
-					switch (typeStr) {
-						case "number":
-						case "string":
-						case "boolean":
-						case "Object":
-						case "object": // null returns "object" Xp
-						case "Array":
-							group.idExpression = new ConstantExpression(groupField);
-							idSet = true;
-							break;
-						default:
-							throw new Error("a group's _id may not include fields of type " + typeStr  + "");
+					if (groupField[0] === "$") {
+						group.setIdExpression(FieldPathExpression.parse(groupField, vps));
+						idSet = true;
 					}
 				}
 
+				if (!idSet) {
+					// constant id - single group
+					group.setIdExpression(ConstantExpression.create(groupField));
+					idSet = true;
+				}
 
 			} else {
+				/*
+					Treat as a projection field with the additional ability to
+					add aggregation operators.
+				*/
 				if (groupFieldName.indexOf(".") !== -1) throw new Error("16414 the group aggregate field name '" + groupFieldName + "' cannot contain '.'");
 				if (groupFieldName[0] === "$") throw new Error("15950 the group aggregate field name '" + groupFieldName + "' cannot be an operator name");
 				if (group._getTypeStr(groupFieldName) === "Object") throw new Error("15951 the group aggregate field '" + groupFieldName + "' must be defined as an expression inside an object");
 
-				var subFieldCount = 0;
-				for (var subFieldName in groupField) {
-					if (groupField.hasOwnProperty(subFieldName)) {
-						var subField = groupField[subFieldName],
-							op = klass.groupOps[subFieldName];
-						if (!op) throw new Error("15952 unknown group operator '" + subFieldName + "'");
+				var subElementCount = 0;
+				for (var subElementName in groupField) {
+					if (groupField.hasOwnProperty(subElementName)) {
+						var subElement = groupField[subElementName],
+							op = klass.groupOps[subElementName];
+						if (!op) throw new Error("15952 unknown group operator '" + subElementName + "'");
 
 						var groupExpression,
-							subFieldTypeStr = group._getTypeStr(subField);
-						if (subFieldTypeStr === "Object") {
-							var subFieldObjCtx = new Expression.ObjectCtx({isDocumentOk:true});
-							groupExpression = Expression.parseObject(subField, subFieldObjCtx);
-						} else if (subFieldTypeStr === "Array") {
-							throw new Error("15953 aggregating group operators are unary (" + subFieldName + ")");
-						} else {
-							groupExpression = Expression.parseOperand(subField);
+							subElementTypeStr = group._getTypeStr(subElement);
+						if (subElementTypeStr === "Object") {
+							var subElementObjCtx = new Expression.ObjectCtx({isDocumentOk:true});
+							groupExpression = Expression.parseObject(subElement, subElementObjCtx, vps);
+						} else if (subElementTypeStr === "Array") {
+							throw new Error("15953 aggregating group operators are unary (" + subElementName + ")");
+						} else { /* assume its an atomic single operand */
+							groupExpression = Expression.parseOperand(subElement, vps);
 						}
-						group.addAccumulator(groupFieldName,op, groupExpression);
+						group.addAccumulator(groupFieldName, op, groupExpression);
 
-						++subFieldCount;
+						++subElementCount;
 					}
 				}
-				if (subFieldCount != 1) throw new Error("15954 the computed aggregate '" + groupFieldName + "' must specify exactly one operator");
+				if (subElementCount !== 1) throw new Error("15954 the computed aggregate '" + groupFieldName + "' must specify exactly one operator");
 			}
 		}
 	}
 
 	if (!idSet) throw new Error("15955 a group specification must include an _id");
 
+	group._variables = new Variables(idGenerator.getIdCount());
+
 	return group;
 };
 
-proto._getTypeStr = function _getTypeStr(obj) {
-	var typeofStr = typeof obj,
-		typeStr = (typeofStr == "object" && obj !== null) ? obj.constructor.name : typeofStr;
-	return typeStr;
-};
+/**
+ * Populates the GroupDocumentSource by grouping all of the input documents at once.
+ *
+ * @method populate
+ * @param callback {Function} Required. callback(err) when done populating.
+ * @async
+ **/
+proto.populate = function populate(callback) {
+	var numAccumulators = this.accumulatorFactories.length;
+	if(numAccumulators !== this.expressions.length) {
+		callback(new Error("Must have equal number of accumulators and expressions"));
+	}
 
-proto.advance = function advance() {
-	base.prototype.advance.call(this); // Check for interupts ????
-	if(!this.populated) this.populate();
+	var input,
+		self = this;
+	async.whilst(
+		function() {
+			return input !== DocumentSource.EOF;
+		},
+		function(cb) {
+			self.source.getNext(function(err, doc) {
+				if(err) return cb(err);
+				if(doc === DocumentSource.EOF) {
+					input = doc;
+					return cb(); //Need to stop now, no new input
+				}
 
-	//verify(this.currentGroupsKeysIndex < this.groupsKeys.length);
+				input = doc;
+				self._variables.setRoot(input);
 
-	++this.currentGroupsKeysIndex;
-	if (this.currentGroupsKeysIndex >= this.groupsKeys.length) {
-		this.currentDocument = null;
-		return false;
-	}
+				/* get the _id value */
+				var id = self.idExpression.evaluate(self._variables);
 
-	this.currentDocument = this.makeDocument(this.currentGroupsKeysIndex);
-	return true;
-};
+				if(undefined === id) id = null;
+
+				var groupKey = JSON.stringify(id),
+					group = self.groups[JSON.stringify(id)];
+
+				if(!group) {
+					self.groupsKeys.push(groupKey);
+					group = [];
+					self.groups[groupKey] = group;
+					// Add the accumulators
+					for(var afi = 0; afi<self.accumulatorFactories.length; afi++) {
+						group.push(self.accumulatorFactories[afi]());
+					}
+				}
+				//NOTE: Skipped memory usage stuff for case when group already existed
+
+				if(numAccumulators !== group.length) {
+					throw new Error('Group must have one of each accumulator');
+				}
+
+				//NOTE: passing the input to each accumulator
+				for(var gi=0; gi<group.length; gi++) {
+					group[gi].process(self.expressions[gi].evaluate(self._variables /*, doingMerge*/));
+				}
+
+				// We are done with the ROOT document so release it.
+				self._variables.clearRoot();
 
-proto.eof = function eof() {
-	if (!this.populated) this.populate();
-	return this.currentGroupsKeysIndex === this.groupsKeys.length;
+				//NOTE: Skipped the part about sorted files
+
+				return cb();
+			});
+		},
+		function(err) {
+			if(err) return callback(err);
+
+			self.populated = true;
+
+			return callback();
+		}
+	);
 };
 
-proto.getCurrent = function getCurrent() {
-	if (!this.populated) this.populate();
-	return this.currentDocument;
+/**
+ * Get the type of something. Handles objects specially to return their true type; i.e. their constructor
+ *
+ * @method populate
+ * @param obj {Object} The object to get the type of
+ * @return {String} The type of the object as a string
+ * @async
+ **/
+proto._getTypeStr = function _getTypeStr(obj) {
+	var typeofStr = typeof obj,
+		typeStr = (typeofStr == "object" && obj !== null) ? obj.constructor.name : typeofStr;
+	return typeStr;
 };
 
+/**
+ * Get the dependencies of the group
+ *
+ * @method getDependencies
+ * @param deps {Object} The
+ * @return {DocumentSource.getDepsReturn} An enum value specifying that these dependencies are exhaustive
+ * @async
+ **/
 proto.getDependencies = function getDependencies(deps) {
 	var self = this;
 	// add _id
@@ -205,67 +370,53 @@ proto.getDependencies = function getDependencies(deps) {
 	return DocumentSource.GetDepsReturn.EXHAUSTIVE;
 };
 
+/**
+ * Called internally only. Adds an accumulator for each matching group.
+ *
+ * @method addAccumulator
+ * @param fieldName {String} The name of the field where the accumulated value will be placed
+ * @param accumulatorFactory {Accumulator} The constructor for creating accumulators
+ * @param epxression {Expression} The expression to be evaluated on incoming documents before they are accumulated
+ **/
 proto.addAccumulator = function addAccumulator(fieldName, accumulatorFactory, expression) {
 	this.fieldNames.push(fieldName);
 	this.accumulatorFactories.push(accumulatorFactory);
 	this.expressions.push(expression);
 };
 
-proto.populate = function populate() {
-	for (var hasNext = !this.source.eof(); hasNext; hasNext = this.source.advance()) {
-		var group,
-			currentDocument = this.source.getCurrent(),
-			_id = this.idExpression.evaluate(currentDocument);
-
-		if (undefined === _id) _id = null;
+/**
+ * Makes a document with the given id and accumulators
+ *
+ * @method makeDocument
+ * @param fieldName {String} The name of the field where the accumulated value will be placed
+ * @param accums {Array} An array of accumulators
+ * @param epxression {Expression} The expression to be evaluated on incoming documents before they are accumulated
+ **/
+proto.makeDocument = function makeDocument(id, accums /*,mergeableOutput*/) {
+	var out = {};
 
-		var idHash = JSON.stringify(_id); //TODO: USE A REAL HASH.  I didn't have time to take collision into account.
+	/* add the _id field */
+	out._id = id;
 
-		if (idHash in this.groups) {
-			group = this.groups[idHash];
+	/* add the rest of the fields */
+	this.fieldNames.forEach(function(fieldName, i) {
+		var val = accums[i].getValue(/*mergeableOutput*/);
+		if(!val) {
+			out[fieldName] = null;
 		} else {
-			this.groups[idHash] = group = [];
-			this.groupsKeys[this.currentGroupsKeysIndex] = idHash;
-			this.originalGroupsKeys[this.currentGroupsKeysIndex] = (_id && typeof _id === 'object') ? Document.clone(_id) : _id;
-			++this.currentGroupsKeysIndex;
-			for (var ai = 0; ai < this.accumulatorFactories.length; ++ai) {
-				var accumulator = new this.accumulatorFactories[ai]();
-				accumulator.addOperand(this.expressions[ai]);
-				group.push(accumulator);
-			}
-		}
-
-
-		// tickle all the accumulators for the group we found
-		for (var gi = 0; gi < group.length; ++gi) {
-			group[gi].evaluate(currentDocument);
+			out[fieldName] = val;
 		}
+	});
 
-	}
-
-	this.currentGroupsKeysIndex = 0; // Start the group
-	if (this.groupsKeys.length > 0) {
-		this.currentDocument = this.makeDocument(this.currentGroupsKeysIndex);
-	}
-	this.populated = true;
-
+	return out;
 };
 
-proto.makeDocument = function makeDocument(groupKeyIndex) {
-	var groupKey = this.groupsKeys[groupKeyIndex],
-		originalGroupKey = this.originalGroupsKeys[groupKeyIndex],
-		group = this.groups[groupKey],
-		doc = {};
-
-	doc[Document.ID_PROPERTY_NAME] = originalGroupKey;
-
-	for (var i = 0; i < this.fieldNames.length; ++i) {
-		var fieldName = this.fieldNames[i],
-			item = group[i];
-		if (item !== "null" && item !== undefined) {
-			doc[fieldName] = item.getValue();
-		}
-	}
-
-	return doc;
+/**
+ * Sets the id expression for the group
+ *
+ * @method setIdExpression
+ * @param epxression {Expression} The expression to set
+ **/
+proto.setIdExpression = function setIdExpression(expression) {
+	this.idExpression = expression;
 };

lib/pipeline/expressions/Expression.js

@@ -222,6 +222,9 @@ klass.parseObject = function parseObject(obj, ctx){
 		exprObj; // the alt result
 	if (obj === undefined) return new ObjectExpression();
 	var fieldNames = Object.keys(obj);
+	if(fieldNames.length === 0) { //NOTE: Added this for mongo 2.5 port of document sources. Should reconsider when porting the expressions themselves
+		return new ObjectExpression();
+	}
 	for (var fc = 0, n = fieldNames.length; fc < n; ++fc) {
 		var fn = fieldNames[fc];
 		if (fn[0] === "$") {

test/lib/pipeline/documentSources/GroupDocumentSource.js

@@ -1,8 +1,10 @@
 "use strict";
 var assert = require("assert"),
+	DocumentSource = require("../../../../lib/pipeline/documentSources/DocumentSource"),
 	CursorDocumentSource = require("../../../../lib/pipeline/documentSources/CursorDocumentSource"),
 	Cursor = require("../../../../lib/Cursor"),
-	GroupDocumentSource = require("../../../../lib/pipeline/documentSources/GroupDocumentSource");
+	GroupDocumentSource = require("../../../../lib/pipeline/documentSources/GroupDocumentSource"),
+	async = require('async');
 
 
 /**
@@ -11,7 +13,7 @@ var assert = require("assert"),
 **/
 var checkJsonRepresentation = function checkJsonRepresentation(self, spec) {
     var rep = {};
-    self.sourceToJson(rep, true);
+    self.serialize(rep, true);
     assert.deepEqual(rep, {$group:spec});
 };
 
@@ -19,7 +21,7 @@ var checkJsonRepresentation = function checkJsonRepresentation(self, spec) {
 function assertExpectedResult(args) {
 	{// check for required args
 		if (args === undefined) throw new TypeError("missing arg: `args` is required");
-		if (args.spec && args.throw === undefined) args.throw = true; // Assume that spec only tests expect an error to be thrown 
+		if (args.spec && args.throw === undefined) args.throw = true; // Assume that spec only tests expect an error to be thrown
 		//if (args.spec === undefined) throw new Error("missing arg: `args.spec` is required");
 		if (args.expected !== undefined && args.docs === undefined) throw new Error("must provide docs with expected value");
 	}// check for required args
@@ -29,11 +31,34 @@ function assertExpectedResult(args) {
 		var gds = GroupDocumentSource.createFromJson(args.spec),
 			cwc = new CursorDocumentSource.CursorWithContext();
 		cwc._cursor = new Cursor( args.docs );
-		var cds = new CursorDocumentSource(cwc);
+		var next,
+			results = [],
+			cds = new CursorDocumentSource(cwc);
 		gds.setSource(cds);
-		var result = gds.getCurrent();
-		assert.deepEqual(result, args.expected);
-		checkJsonRepresentation(gds, args.spec);
+		async.whilst(
+			function() {
+				next !== DocumentSource.EOF;
+			},
+			function(done) {
+				gds.getNext(function(err, doc) {
+					if(err) return done(err);
+					next = doc;
+					if(next === DocumentSource.EOF) {
+						return done();
+					} else {
+						results.push(next);
+						return done();
+					}
+				});
+			},
+			function(err) {
+				assert.deepEqual(results, args.expected);
+				checkJsonRepresentation(gds, args.spec);
+				if(args.done) {
+					return args.done();
+				}
+			}
+		);
 	}else{
 		if(args.throw) {
 			assert.throws(function(){
@@ -72,6 +97,7 @@ module.exports = {
 
 			// $group _id is an empty object
 			"should not throw when _id is an empty object": function advanceTest(){
+				//NOTE: This is broken until expressions get #serialize methods
 				assertExpectedResult({spec:{_id:{}}, "throw":false});
 			},
 
@@ -79,7 +105,7 @@ module.exports = {
 			"should throw error when  _id is an invalid object expression": function testConstructor(){
 				assertExpectedResult({
 					spec:{_id:{$add:1, $and:1}},
-				});	
+				});
 			},
 
 
@@ -90,11 +116,13 @@ module.exports = {
 
 			// $group _id is the empty string
 			"should not throw when _id is an empty string": function advanceTest(){
+				//NOTE: This is broken until expressions get ported to 2.5; specifically, until they get a #create method
 				assertExpectedResult({spec:{_id:""}, "throw":false});
 			},
 
 			// $group _id is a string constant
 			"should not throw when _id is a string constant": function advanceTest(){
+				//NOTE: This is broken until expressions get ported to 2.5; specifically, until they get a #create method
 				assertExpectedResult({spec:{_id:"abc"}, "throw":false});
 			},
 
@@ -102,57 +130,61 @@ module.exports = {
 			"should throw when _id is an invalid field path": function advanceTest(){
 				assertExpectedResult({spec:{_id:"$a.."}});
 			},
-		
+
 			// $group _id is a numeric constant
 			"should not throw when _id is a numeric constant": function advanceTest(){
+				//NOTE: This is broken until expressions get ported to 2.5; specifically, until they get a #create method
 				assertExpectedResult({spec:{_id:2}, "throw":false});
 			},
 
 			// $group _id is an array constant
 			"should not throw when _id is an array constant": function advanceTest(){
+				//NOTE: This is broken until expressions get ported to 2.5; specifically, until they get a #create method
 				assertExpectedResult({spec:{_id:[1,2]}, "throw":false});
 			},
 
 			// $group _id is a regular expression (not supported)
 			"should throw when _id is a regex": function advanceTest(){
+				//NOTE: This is broken until expressions get ported to 2.5; specifically, until they get a #create method
 				assertExpectedResult({spec:{_id:/a/}});
 			},
 
 			// The name of an aggregate field is specified with a $ prefix
 			"should throw when aggregate field spec is specified with $ prefix": function advanceTest(){
+				//NOTE: This is broken until expressions get ported to 2.5; specifically, until they get a #create method
 				assertExpectedResult({spec:{_id:1, $foo:{$sum:1}}});
 			},
 
 			// An aggregate field spec that is not an object
 			"should throw when aggregate field spec is not an object": function advanceTest(){
+				//NOTE: This is broken until expressions get ported to 2.5; specifically, until they get a #create method
 				assertExpectedResult({spec:{_id:1, a:1}});
 			},
 
 			// An aggregate field spec that is not an object
 			"should throw when aggregate field spec is an empty object": function advanceTest(){
+				//NOTE: This is broken until expressions get ported to 2.5; specifically, until they get a #create method
 				assertExpectedResult({spec:{_id:1, a:{}}});
 			},
 
 			// An aggregate field spec with an invalid accumulator operator
 			"should throw when aggregate field spec is an invalid accumulator": function advanceTest(){
+				//NOTE: This is broken until expressions get ported to 2.5; specifically, until they get a #create method
 				assertExpectedResult({spec:{_id:1, a:{$bad:1}}});
 			},
 
 			// An aggregate field spec with an array argument
 			"should throw when aggregate field spec with an array as an argument": function advanceTest(){
+				//NOTE: This is broken until expressions get ported to 2.5; specifically, until they get a #create method
 				assertExpectedResult({spec:{_id:1, a:{$sum:[]}}});
 			},
 
 			// Multiple accumulator operators for a field
 			"should throw when aggregate field spec with multiple accumulators": function advanceTest(){
+				//NOTE: This is broken until expressions get ported to 2.5; specifically, until they get a #create method
 				assertExpectedResult({spec:{_id:1, a:{$sum:1, $push:1}}});
 			}
 
-			//Not Implementing, not way to support this in Javascript Objects
-			// Aggregation using duplicate field names is allowed currently
-
-
-
 		},
 
 		"#getSourceName()": {
@@ -163,67 +195,147 @@ module.exports = {
 			}
 		},
 
-		"#advance": {
+		"#getNext, #populate": {
+
+			// Aggregation using duplicate field names is allowed currently
+			// Note: Can't duplicate fields in javascript objects -- skipped
 
 			// $group _id is computed from an object expression
 			"should compute _id from an object expression": function testAdvance_ObjectExpression(){
+				//NOTE: This is broken until expressions get ported to 2.5; specifically, until they get a #create method
 				assertExpectedResult({
 					docs: [{a:6}],
 					spec: {_id:{z:"$a"}},
-					expected: {_id:{z:6}}
+					expected: [{_id:{z:6}}]
 				});
 			},
 
 			// $group _id is a field path expression
 			"should compute _id from a field path expression": function testAdvance_FieldPathExpression(){
+				//NOTE: This is broken until expressions get ported to 2.5; specifically, until they get a #create method
 				assertExpectedResult({
 					docs: [{a:5}],
 					spec: {_id:"$a"},
-					expected: {_id:5}
+					expected: [{_id:5}]
 				});
 			},
-			
+
 			// $group _id is a field path expression
 			"should compute _id from a Date": function testAdvance_Date(){
+				//NOTE: This is broken until expressions get ported to 2.5; specifically, until they get a #create method
 				var d = new Date();
 				assertExpectedResult({
 					docs: [{a:d}],
 					spec: {_id:"$a"},
-					expected: {_id:d}
+					expected: [{_id:d}]
 				});
 			},
 
 			// Aggregate the value of an object expression
 			"should aggregate the value of an object expression": function testAdvance_ObjectExpression(){
+				//NOTE: This is broken until expressions get ported to 2.5; specifically, until they get a #create method
 				assertExpectedResult({
 					docs: [{a:6}],
 					spec: {_id:0, z:{$first:{x:"$a"}}},
-					expected: {_id:0, z:{x:6}}
+					expected: [{_id:0, z:{x:6}}]
 				});
 			},
 
 			// Aggregate the value of an operator expression
 			"should aggregate the value of an operator expression": function testAdvance_OperatorExpression(){
+				//NOTE: This is broken until expressions get ported to 2.5; specifically, until they get a #create method
 				assertExpectedResult({
 					docs: [{a:6}],
 					spec: {_id:0, z:{$first:"$a"}},
-					expected: {_id:0, z:6}
+					expected: [{_id:0, z:6}]
 				});
 			},
 
 			// Aggregate the value of an operator expression
 			"should aggregate the value of an operator expression with a null id": function testAdvance_Null(){
+				//NOTE: This is broken until expressions get ported to 2.5; specifically, until they get a #create method
 				assertExpectedResult({
 					docs: [{a:6}],
 					spec: {_id:null, z:{$first:"$a"}},
-					expected: {_id:null, z:6}
+					expected: [{_id:null, z:6}]
 				});
-			}
+			},
+
+			// A $group performed on a single document
+			"should make one group with one values": function SingleDocument() {
+				assertExpectedResult({
+					docs: [{a:1}],
+					spec: {_id:0, a:{$sum:"$a"}},
+					expected: [{_id:0, a:1}]
+				});
+			},
+
+			// A $group performed on two values for a single key
+			"should make one group with two values": function TwoValuesSingleKey() {
+				assertExpectedResult({
+					docs: [{a:1}, {a:2}],
+					spec: {_id:"$_id", a:{$push:"$a"}},
+					expected: [{_id:0, a:[1,2]}]
+				});
+			},
 
+			// A $group performed on two values with one key each.
+			"should make two groups with one value": function TwoValuesTwoKeys() {
+				assertExpectedResult({
+					docs: [{_id:0,a:1}, {_id:1,a:2}],
+					spec: {_id:"$_id", a:{$push:"$a"}},
+					expected: [{_id:0, a:[1]}, {_id:1, a:[2]}]
+				});
+			},
+
+			// A $group performed on two values with two keys each.
+			"should make two groups with two values": function FourValuesTwoKeys() {
+				assertExpectedResult({
+					docs: [{_id:0,a:1}, {_id:1,a:2}, {_id:0,a:3}, {_id:1,a:4}],
+					spec: {_id:"$_id", a:{$push:"$a"}},
+					expected: [{_id:0, a:[1, 3]}, {_id:1, a:[2, 4]}]
+				});
+			},
+
+			// A $group performed on two values with two keys each and two accumulator operations.
+			"should make two groups with two values with two accumulators": function FourValuesTwoKeysTwoAccumulators() {
+				assertExpectedResult({
+					docs: [{_id:0,a:1}, {_id:1,a:2}, {_id:0,a:3}, {_id:1,a:4}],
+					spec: {_id:"$_id", list:{$push:"$a"}, sum:{$sum:{$divide:["$a", 2]}}},
+					expected: [{_id:0, list:[1, 3], sum:2}, {_id:1, list:[2, 4], sum:3}]
+				});
+			},
+
+			// Null and undefined _id values are grouped together.
+			"should group null and undefined _id's together": function GroupNullUndefinedIds() {
+				assertExpectedResult({
+					docs: [{a:null, b:100}, {b:10}],
+					spec: {_id:"$a", sum:{$sum:"$b"}},
+					expected: [{_id:null, sum:110}]
+				});
+			},
+
+			// A complex _id expression.
+			"should group based on a complex id": function ComplexId() {
+				assertExpectedResult({
+					docs: [{a:"de", b:"ad", c:"beef", d:""}, {a:"d", b:"eadbe", c:"", d:"ef"}],
+					spec: {_id:{$concat:["$a", "$b", "$c", "$d"]}},
+					expected: [{_id:'deadbeef'}]
+				});
+			},
+
+			// An undefined accumulator value is dropped.
+			"should ignore undefined values during accumulation":function UndefinedAccumulatorValue() {
+				assertExpectedResult({
+					docs: [{}],
+					spec: {_id:0, first:{$first:"$missing"}},
+					expected: [{_id:0, first:null}]
+				});
+			}
 		}
 
 	}
 
 };
 
-if (!module.parent)(new(require("mocha"))()).ui("exports").reporter("spec").addFile(__filename).run(process.exit);
+if (!module.parent)(new(require("mocha"))()).ui("exports").reporter("spec").addFile(__filename).grep(process.env.MOCHA_GREP || '').run(process.exit);