Merge branch 'feature/mongo_2.6.5_documentSource' into feature/mongo_2.6.5_documentSource_Pipeline

.gitignore

@@ -1,2 +1,18 @@
-/node_modules/
+!.gitkeep
+
+# node
+/node_modules
+npm-debug.log
+
+# IDE files
+/.idea
+/.settings.xml
+/.settings
+/.c9revisions/
+
+# misc files
+*.swp
+.DS_Store
+
+# build files
 /reports/

lib/Cursor.js

@@ -1,30 +0,0 @@
-"use strict";
-
-/**
- * This class is a simplified implementation of the cursors used in MongoDB for reading from an Array of documents.
- * @param	{Array}	items	The array source of the data
- **/
-var klass = module.exports = function Cursor(items){
-	if (!(items instanceof Array)) throw new Error("arg `items` must be an Array");
-	this.cachedData = items.slice(0);	// keep a copy so array changes when using async doc srcs do not cause side effects
-	this.length = items.length;
-	this.offset = 0;
-}, base = Object, proto = klass.prototype = Object.create(base.prototype, {constructor:{value:klass}});
-
-proto.ok = function ok(){
-	return (this.offset < this.length) || this.hasOwnProperty("curr");
-};
-
-proto.advance = function advance(){
-	if (this.offset >= this.length){
-		delete this.curr;
-		return false;
-	}
-	this.curr = this.cachedData[this.offset++];
-	return this.curr;
-};
-
-proto.current = function current(){
-	if (!this.hasOwnProperty("curr")) this.advance();
-	return this.curr;
-};

lib/pipeline/Document.js

@@ -151,7 +151,7 @@ klass.cloneDeep = function cloneDeep(doc) {	//there are casese this is actually
 	for (var key in doc) {
 		if (doc.hasOwnProperty(key)) {
 			var val = doc[key];
-			obj[key] = val instanceof Object && val.constructor === Object ? Document.clone(val) : val;
+			obj[key] = val instanceof Object && val.constructor === Object ? Document.cloneDeep(val) : val;
 		}
 	}
 	return obj;

lib/pipeline/documentSources/CursorDocumentSource.js

@@ -1,14 +1,13 @@
 "use strict";
 
-var DocumentSource = require('./DocumentSource'),
+var async = require('async'),
+	Value = require('../Value'),
+	Runner = require('../../query/Runner'),
+	DocumentSource = require('./DocumentSource'),
 	LimitDocumentSource = require('./LimitDocumentSource');
 
-// Mimicking max memory size from mongo/db/query/new_find.cpp
-// Need to actually decide some size for this?
-var MAX_BATCH_DOCS = 150;
-
 /**
- * Constructs and returns Documents from the objects produced by a supplied Cursor.
+ * Constructs and returns Documents from the BSONObj objects produced by a supplied Runner.
  * An object of this type may only be used by one thread, see SERVER-6123.
  *
  * This is usually put at the beginning of a chain of document sources
@@ -20,46 +19,36 @@ var MAX_BATCH_DOCS = 150;
  * @constructor
  * @param	{CursorDocumentSource.CursorWithContext}	cursorWithContext the cursor to use to fetch data
  **/
-var CursorDocumentSource = module.exports = CursorDocumentSource = function CursorDocumentSource(cursorWithContext, expCtx){
+var CursorDocumentSource = module.exports = CursorDocumentSource = function CursorDocumentSource(namespace, runner, expCtx){
 	base.call(this, expCtx);
 
-	this.current = null;
+	this._docsAddedToBatches = 0;
+	this._ns = namespace;
+	this._runner = runner;
 
-//	this.ns = null;
-//	/*
-//	The bson dependencies must outlive the Cursor wrapped by this
-//	source.  Therefore, bson dependencies must appear before pCursor
-//	in order cause its destructor to be called *after* pCursor's.
-//	*/
-//	this.query = null;
-//	this.sort = null;
+}, klass = CursorDocumentSource, base = DocumentSource, proto = klass.prototype = Object.create(base.prototype, {constructor:{value:klass}});
 
-	this._projection = null;
+klass.MaxDocumentsToReturnToClientAtOnce = 150; //DEVIATION: we are using documents instead of bytes
 
-	this._cursorWithContext = cursorWithContext;
-	this._curIdx = 0;
-	this._currentBatch = [];
-	this._limit = undefined;
-	this._docsAddedToBatches = 0;
+proto._currentBatch = [];
+proto._currentBatchIndex = 0;
 
-	if (!this._cursorWithContext || !this._cursorWithContext._cursor) throw new Error("CursorDocumentSource requires a valid cursorWithContext");
+// BSONObj members must outlive _projection and cursor.
+proto._query = undefined;
+proto._sort = undefined;
+proto._projection = undefined;
+proto._dependencies = undefined;
+proto._limit = undefined;
+proto._docsAddedToBatches = undefined; // for _limit enforcement
 
-}, klass = CursorDocumentSource, base = require('./DocumentSource'), proto = klass.prototype = Object.create(base.prototype, {constructor:{value:klass}});
+proto._ns = undefined;
+proto._runner = undefined; // PipelineRunner holds a weak_ptr to this.
 
 
-klass.CursorWithContext = (function (){
-	/**
-	 * Holds a Cursor and all associated state required to access the cursor.
-	 * @class CursorWithContext
-	 * @namespace mungedb-aggregate.pipeline.documentSources.CursorDocumentSource
-	 * @module mungedb-aggregate
-	 * @constructor
-	 **/
-	var klass = function CursorWithContext(ns){
-		this._cursor = null;
-	};
-	return klass;
-})();
+
+proto.isValidInitialSource = function(){
+	return true;
+};
 
 /**
  * Release the Cursor and the read lock it requires, but without changing the other data.
@@ -69,189 +58,207 @@ klass.CursorWithContext = (function (){
  * @method	dispose
  **/
 proto.dispose = function dispose() {
-	this._cursorWithContext = null;
+	if (this._runner) this._runner.reset();
 	this._currentBatch = [];
-	this._curIdx = 0;
 };
 
+/**
+ * Get the source's name.
+ * @method	getSourceName
+ * @returns	{String}	the string name of the source as a constant string; this is static, and there's no need to worry about adopting it
+ **/
 proto.getSourceName = function getSourceName() {
 	return "$cursor";
 };
 
+/**
+ * Returns the next Document if there is one
+ *
+ * @method	getNext
+ **/
 proto.getNext = function getNext(callback) {
-	if (!callback) throw new Error(this.getSourceName() + ' #getNext() requires callback');
-
-	if (this._currentBatch.length <= this._curIdx) {
-		this.loadBatch();
-
-		if (this._currentBatch.length <= this._curIdx) {
-			callback(null, DocumentSource.EOF);
-			return DocumentSource.EOF;
-		}
+	if (this.expCtx && this.expCtx.checkForInterrupt && this.expCtx.checkForInterrupt()){
+		return callback(new Error('Interrupted'));
 	}
-
-	// Don't unshift. It's expensiver.
-	var out = this._currentBatch[this._curIdx];
-	this._curIdx++;
-
-	callback(null, out);
-	return out;
+	
+	var self = this;
+	if (self._currentBatchIndex >= self._currentBatch.length) {
+		self._currentBatchIndex = 0;
+		self._currentBatch = [];
+		return self.loadBatch(function(err){
+			if (err) return callback(err);
+			if (self._currentBatch.length === 0)
+				return callback(null, null);
+			
+			return callback(null, self._currentBatch[self._currentBatchIndex++]);
+		});
+	}
+	return callback(null, self._currentBatch[self._currentBatchIndex++]);
 };
 
+/**
+ * Attempt to coalesce this DocumentSource with any $limits that it encounters
+ *
+ * @method	coalesce
+ * @param	{DocumentSource}	nextSource	the next source in the document processing chain.
+ * @returns	{Boolean}	whether or not the attempt to coalesce was successful or not; if the attempt was not successful, nothing has been changed
+ **/
 proto.coalesce = function coalesce(nextSource) {
-	if (this._limit) {
+	// Note: Currently we assume the $limit is logically after any $sort or
+	// $match. If we ever pull in $match or $sort using this method, we
+	// will need to keep track of the order of the sub-stages.
+
+	if (!this._limit) {
+		if (nextSource instanceof LimitDocumentSource) {
+			this._limit = nextSource;
+			return this._limit;
+		}
+		return false;// false if next is not a $limit
+	}
+	else {
 		return this._limit.coalesce(nextSource);
-	} else if (nextSource instanceof LimitDocumentSource) {
-		this._limit = nextSource;
-		return this._limit;
-	} else {
-		return false;
 	}
+
+	return false;
 };
 
-///**
-// * Record the namespace.  Required for explain.
-// *
-// * @method	setNamespace
-// * @param	{String}	ns	the namespace
-// **/
-//proto.setNamespace = function setNamespace(ns) {}
-//
-///**
-// * Record the query that was specified for the cursor this wraps, if any.
-// * This should be captured after any optimizations are applied to
-// * the pipeline so that it reflects what is really used.
-// * This gets used for explain output.
-// *
-// * @method	setQuery
-// * @param	{Object}	pBsonObj	the query to record
-// **/
+
+/**
+ * Record the query that was specified for the cursor this wraps, if
+ * any.
+ * 
+ * This should be captured after any optimizations are applied to
+ * the pipeline so that it reflects what is really used.
+ * 
+ * This gets used for explain output.
+ *
+ * @method	setQuery
+ * @param	{Object}	pBsonObj	the query to record
+ **/
 proto.setQuery = function setQuery(query) {
 	this._query = query;
 };
 
-///**
-// * Record the sort that was specified for the cursor this wraps, if any.
-// * This should be captured after any optimizations are applied to
-// * the pipeline so that it reflects what is really used.
-// * This gets used for explain output.
-// *
-// * @method	setSort
-// * @param	{Object}	pBsonObj	the query to record
-// **/
-//proto.setSort = function setSort(pBsonObj) {};
+/**
+ * Record the sort that was specified for the cursor this wraps, if
+ * any.
+ * 
+ * This should be captured after any optimizations are applied to
+ * the pipeline so that it reflects what is really used.
+ * 
+ * This gets used for explain output.
+ *
+ * @method	setSort
+ * @param	{Object}	pBsonObj	the query to record
+ **/
+proto.setSort = function setSort(sort) {
+	this._sort = sort;
+};
 
 /**
- * setProjection method
+ * Informs this object of projection and dependency information.
  *
  * @method	setProjection
  * @param	{Object}	projection
  **/
 proto.setProjection = function setProjection(projection, deps) {
-
-	if (this._projection){
-		throw new Error("projection is already set");
-	}
-
-
-	//dont think we need this yet
-
-//	this._projection = new Projection();
-//	this._projection.init(projection);
-//
-//	this.cursor().fields = this._projection;
-
-	this._projection = projection;  //just for testing
+	this._projection = projection;
 	this._dependencies = deps;
 };
 
-//----------------virtuals from DocumentSource--------------
-
 /**
- * Set the underlying source this source should use to get Documents
- * from.
- * It is an error to set the source more than once.  This is to
- * prevent changing sources once the original source has been started;
- * this could break the state maintained by the DocumentSource.
- * This pointer is not reference counted because that has led to
- * some circular references.  As a result, this doesn't keep
- * sources alive, and is only intended to be used temporarily for
- * the lifetime of a Pipeline::run().
  *
  * @method setSource
  * @param source   {DocumentSource}  the underlying source to use
  * @param callback  {Function}        a `mungedb-aggregate`-specific extension to the API to half-way support reading from async sources
  **/
 proto.setSource = function setSource(theSource) {
-	if (theSource) throw new Error("CursorDocumentSource doesn't take a source"); //TODO: This needs to put back without the if once async is fully and properly supported
+	throw new Error('this doesnt take a source');
 };
 
 proto.serialize = function serialize(explain) {
-	if (!explain)
-		return null;
 
-	if (!this._cursorWithContext)
-		throw new Error("code 17135; Cursor deleted.");
+	// we never parse a documentSourceCursor, so we only serialize for explain
+	if (!explain)
+		return {};
 
-	// A stab at what mongo wants
-	return {
+	var out = {};
+	out[this.getSourceName()] = {
 		query: this._query,
 		sort: this._sort ? this._sort : null,
-		limit: this._limit ? this._limit : null,
+		limit: this._limit ? this._limit.getLimit() : null,
 		fields: this._projection ? this._projection : null,
-		indexonly: false,
-		cursorType: this._cursorWithContext ? "cursor" : null
+		plan: this._runner.getInfo(explain)
 	};
+	return out;
 };
 
-// LimitDocumentSource has the setLimit function which trickles down to any documentsource
+/**
+ * returns -1 for no limit
+ * 
+ * @method getLimit
+**/
 proto.getLimit = function getLimit() {
 	return this._limit ? this._limit.getLimit() : -1;
 };
 
-//----------------private--------------
-
-//proto.chunkMgr = function chunkMgr(){};
-
-//proto.canUseCoveredIndex = function canUseCoveredIndex(){};
-
-//proto.yieldSometimes = function yieldSometimes(){};
-
-proto.loadBatch = function loadBatch() {
-	var nDocs = 0,
-		cursor = this._cursorWithContext ? this._cursorWithContext._cursor : null;
-
-	if (!cursor)
-		return this.dispose();
-
-	for(;cursor.ok(); cursor.advance()) {
-		if (!cursor.ok())
-			break;
-
-		// these methods do not exist
-		// if (!cursor.currentMatches() || cursor.currentIsDup())
-		// continue;
-
-		var next = cursor.current();
-		this._currentBatch.push(this._projection ? base.documentFromJsonWithDeps(next, this._dependencies) : next);
-
-		if (this._limit) {
-			this._docsAddedToBatches++;
-			if (this._docsAddedToBatches == this._limit.getLimit())
-				break;
-
-			if (this._docsAddedToBatches >= this._limit.getLimit()) {
-				throw new Error("added documents to the batch over limit size");
+/**
+ * Load a batch of documents from the Runner into the internal array
+ * 
+ * @method loadBatch
+**/
+proto.loadBatch = function loadBatch(callback) {
+	if (!this._runner) {
+		this.dispose();
+		return callback;
+	}
+	
+	this._runner.restoreState();
+
+	var self = this,
+		whileBreak = false,		// since we are in an async loop instead of a normal while loop, need to mimic the
+		whileReturn = false;	// functionality.  These flags are similar to saying 'break' or 'return' from inside the loop
+	return async.whilst(
+		function test(){
+			return !whileBreak && !whileReturn;
+		},
+		function(next) {
+			return self._runner.getNext(function(err, obj, state){
+				if (err) return next(err);
+				if (state === Runner.RunnerState.RUNNER_ADVANCED) {
+					if (self._dependencies) {
+						self._currentBatch.push(self._dependencies.extractFields(obj));
+					} else {
+						self._currentBatch.push(obj);
+					}
+
+					if (self._limit) {
+						if (++self._docsAddedToBatches === self._limit.getLimit()) {
+							whileBreak = true;
+							return next();
+						}
+						//this was originally a 'verify' in the mongo code
+						if (self._docsAddedToBatches > self._limit.getLimit()){
+							return next(new Error('documents collected past the end of the limit'));
+						}
+					}
+
+					if (self._currentBatch >= klass.MaxDocumentsToReturnToClientAtOnce) {
+						// End self batch and prepare Runner for yielding.
+						self._runner.saveState();
+						whileReturn = true;
+					}
+				} else {
+					whileBreak = true;
+				}
+				return next();
+			});
+		},
+		function(err){
+			if (!whileReturn){
+				self._runner.reset();
 			}
+			callback(err);
 		}
-
-		// Mongo uses number of bytes, but that doesn't make sense here. Yield when nDocs is over a threshold
-		if (nDocs > MAX_BATCH_DOCS) {
-			this._curIdx++; // advance the deque
-			nDocs++;
-			return;
-		}
-	}
-
-	this._cursorWithContext = undefined;	//NOTE: Trying to emulate erasing the cursor; not exactly how mongo does it
+	);
 };

lib/pipeline/documentSources/DocumentSource.js

@@ -37,22 +37,6 @@ var DocumentSource = module.exports = function DocumentSource(expCtx){
 
 }, klass = DocumentSource, base = Object, proto = klass.prototype = Object.create(base.prototype, {constructor:{value:klass}});
 
-/**
- * Use EOF as boost::none for document sources to signal the end of their document stream.
- **/
-klass.EOF = (function() {
-	/**
-	 * Represents a non-value in a document stream
-	 * @class EOF
-	 * @namespace mungedb-aggregate.pipeline.documentSources.DocumentSource
-	 * @module mungedb-aggregate
-	 * @constructor
-	 **/
-	var klass = function EOF(){
-	};
-	return klass;
-})();
-
 /*
 class DocumentSource :
 public IntrusiveCounterUnsigned,
@@ -83,7 +67,7 @@ proto.getPipelineStep = function getPipelineStep() {
 };
 
 /**
- * Returns the next Document if there is one or DocumentSource.EOF if at EOF.
+ * Returns the next Document if there is one or null if at EOF.
  *
  * some implementations do the equivalent of verify(!eof()) so check eof() first
  * @method	getNext
@@ -206,7 +190,7 @@ proto.serializeToArray = function serializeToArray(array, explain) {
  * @method GET_NEXT_PASS_THROUGH
  * @param callback {Function}
  * @param callback.err {Error} An error or falsey
- * @param callback.doc {Object} The source's next object or DocumentSource.EOF
+ * @param callback.doc {Object} The source's next object or null
  **/
 klass.GET_NEXT_PASS_THROUGH = function GET_NEXT_PASS_THROUGH(callback) {
 	if (!callback) throw new Error(this.getSourceName() + ' #getNext() requires callback');

lib/pipeline/documentSources/GroupDocumentSource.js

@@ -21,22 +21,29 @@ var DocumentSource = require("./DocumentSource"),
  **/
 var GroupDocumentSource = module.exports = function GroupDocumentSource(expCtx) {
 	if (arguments.length > 1) throw new Error("up to one arg expected");
+	expCtx = !expCtx ? {} : expCtx;
 	base.call(this, expCtx);
 
 	this.populated = false;
-	this.idExpression = null;
+	this.doingMerge = false;
+	this.spilled = false;
+	this.extSortAllowed = expCtx.extSortAllowed && !expCtx.inRouter;
+
+	this.accumulatorFactories = [];
+	this.currentAccumulators = [];
 	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.originalGroupsKeys = [];
+	this.variables = null;
 	this.fieldNames = [];
-	this.accumulatorFactories = [];
+	this.idFieldNames = [];
 	this.expressions = [];
-	this.currentDocument = null;
+	this.idExpressions = [];
 	this.currentGroupsKeysIndex = 0;
 
 }, klass = GroupDocumentSource, base = DocumentSource, proto = klass.prototype = Object.create(base.prototype, {constructor:{value:klass}});
 
+// TODO: Do we need this?
 klass.groupOps = {
 	"$addToSet": Accumulators.AddToSet,
 	"$avg": Accumulators.Avg,
@@ -72,12 +79,16 @@ proto.getSourceName = function getSourceName() {
 };
 
 /**
- * Gets the next document or DocumentSource.EOF if none
+ * Gets the next document or null if none
  *
  * @method getNext
  * @return {Object}
  **/
 proto.getNext = function getNext(callback) {
+	if (!callback) throw new Error(this.getSourceName() + ' #getNext() requires callback.');
+	if (this.expCtx.checkForInterrupt && this.expCtx.checkForInterrupt() === false)
+		return callback(new Error("Interrupted"));
+
 	var self = this;
 	async.series([
 		function(next) {
@@ -89,25 +100,25 @@ proto.getNext = function getNext(callback) {
 				return next();
 		},
 		function(next) {
-			if(Object.keys(self.groups).length === 0) {
-				return next(null, DocumentSource.EOF);
-			}
-
-			//Note: Skipped the spilled logic
+			// NOTE: Skipped the spilled functionality
+			if (self.spilled) {
+				throw new Error("Spilled is not implemented.");
+			} else {
+				if(self.currentGroupsKeysIndex === self.groupsKeys.length) {
+					return next(null, null);
+				}
 
-			if(self.currentGroupsKeysIndex === self.groupsKeys.length) {
-				return next(null, DocumentSource.EOF);
-			}
+				var id = self.originalGroupsKeys[self.currentGroupsKeysIndex],
+					stringifiedId = self.groupsKeys[self.currentGroupsKeysIndex],
+					accumulators = self.groups[stringifiedId],
+					out = self.makeDocument(id, accumulators, self.expCtx.inShard);
 
-			var id = self.groupsKeys[self.currentGroupsKeysIndex],
-				accumulators = self.groups[id],
-				out = self.makeDocument(id, accumulators /*,mergeableOutput*/);
+				if(++self.currentGroupsKeysIndex === self.groupsKeys.length) {
+					self.dispose();
+				}
 
-			if(++self.currentGroupsKeysIndex === self.groupsKeys.length) {
-				self.dispose();
+				return next(null, out);
 			}
-
-			return next(null, out);
 		}
 	], function(err, results) {
 		callback(err, results[1]);
@@ -134,8 +145,14 @@ proto.dispose = function dispose() {
  * @method optimize
  **/
 proto.optimize = function optimize() {
+	// TODO if all _idExpressions are ExpressionConstants after optimization, then we know there
+	// will only be one group. We should take advantage of that to avoid going through the hash
+	// table.
 	var self = this;
-	self.idExpression = self.idExpression.optimize();
+	self.idExpressions.forEach(function(expression, i) {
+		self.idExpressions[i] = expression.optimize();
+	});
+
 	self.expressions.forEach(function(expression, i) {
 		self.expressions[i] = expression.optimize();
 	});
@@ -149,25 +166,38 @@ proto.optimize = function optimize() {
  * @param explain {Boolean} Create explain output
  **/
 proto.serialize = function serialize(explain) {
-	var insides = {};
+	var self = this,
+		insides = {};
 
 	// add the _id
-	insides._id = this.idExpression.serialize(explain);
+	if (self.idFieldNames.length === 0) {
+		if (self.idExpressions.length !== 1) throw new Error("Should only have one _id field");
+		insides._id = self.idExpressions[0].serialize(explain);
+	} else {
+		if (self.idExpressions.length !== self.idFieldNames.length)
+			throw new Error("Should have the same number of idExpressions and idFieldNames.");
+
+		var md = {};
+		self.idExpressions.forEach(function(expression, i) {
+			md[self.idFieldNames[i]] = expression.serialize(explain);
+		});
+		insides._id = md;
+	}
 
 	//add the remaining fields
-	var aFacs = this.accumulatorFactories,
+	var aFacs = self.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
+		var aFac = new aFacs[i](),
+			serialExpression = self.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;
+		insides[self.fieldNames[i]] = serialAccumulator;
 	}
 
 	var serialSource = {};
-	serialSource[this.getSourceName()] = insides;
+	serialSource[self.getSourceName()] = insides;
 	return serialSource;
 };
 
@@ -192,31 +222,13 @@ klass.createFromJson = function createFromJson(elem, expCtx) {
 			var groupField = groupObj[groupFieldName];
 
 			if (groupFieldName === "_id") {
-
 				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.setIdExpression(Expression.parseObject(groupField, objCtx, vps));
-					idSet = true;
-
-				} else if (typeof groupField === "string") {
-					if (groupField[0] === "$") {
-						group.setIdExpression(FieldPathExpression.parse(groupField, vps));
-						idSet = true;
-					}
-				}
-
-				if (!idSet) {
-					// constant id - single group
-					group.setIdExpression(ConstantExpression.create(groupField));
-					idSet = true;
-				}
+				group.parseIdExpression(groupField, vps);
+				idSet = true;
 
+			} else if (groupFieldName === '$doingMerge' && groupField) {
+				throw new Error("17030 $doingMerge should be true if present");
 			} else {
 				/*
 					Treat as a projection field with the additional ability to
@@ -255,7 +267,7 @@ klass.createFromJson = function createFromJson(elem, expCtx) {
 
 	if (!idSet) throw new Error("15955 a group specification must include an _id");
 
-	group._variables = new Variables(idGenerator.getIdCount());
+	group.variables = new Variables(idGenerator.getIdCount());
 
 	return group;
 };
@@ -269,6 +281,7 @@ klass.createFromJson = function createFromJson(elem, expCtx) {
  **/
 proto.populate = function populate(callback) {
 	var numAccumulators = this.accumulatorFactories.length;
+	// NOTE: this is not in mongo, does it belong here?
 	if(numAccumulators !== this.expressions.length) {
 		callback(new Error("Must have equal number of accumulators and expressions"));
 	}
@@ -277,34 +290,35 @@ proto.populate = function populate(callback) {
 		self = this;
 	async.whilst(
 		function() {
-			return input !== DocumentSource.EOF;
+			return input !== null;
 		},
 		function(cb) {
 			self.source.getNext(function(err, doc) {
 				if(err) return cb(err);
-				if(doc === DocumentSource.EOF) {
+				if(doc === null) {
 					input = doc;
 					return cb(); //Need to stop now, no new input
 				}
 
 				input = doc;
-				self._variables.setRoot(input);
+				self.variables.setRoot(input);
 
 				/* get the _id value */
-				var id = self.idExpression.evaluate(self._variables);
+				var id = self.computeId(self.variables);
 
 				if(undefined === id) id = null;
 
 				var groupKey = JSON.stringify(id),
-					group = self.groups[JSON.stringify(id)];
+					group = self.groups[groupKey];
 
 				if(!group) {
+					self.originalGroupsKeys.push(id);
 					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]());
+						group.push(new self.accumulatorFactories[afi]());
 					}
 				}
 				//NOTE: Skipped memory usage stuff for case when group already existed
@@ -315,11 +329,11 @@ proto.populate = function populate(callback) {
 
 				//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*/));
+					group[gi].process(self.expressions[gi].evaluate(self.variables, self.doingMerge));
 				}
 
 				// We are done with the ROOT document so release it.
-				self._variables.clearRoot();
+				self.variables.clearRoot();
 
 				//NOTE: Skipped the part about sorted files
 
@@ -336,20 +350,6 @@ proto.populate = function populate(callback) {
 	);
 };
 
-/**
- * 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
  *
@@ -361,7 +361,9 @@ proto._getTypeStr = function _getTypeStr(obj) {
 proto.getDependencies = function getDependencies(deps) {
 	var self = this;
 	// add _id
-	this.idExpression.addDependencies(deps);
+	this.idExpressions.forEach(function(expression, i) {
+		expression.addDependencies(deps);
+	});
 	// add the rest
 	this.fieldNames.forEach(function (field, i) {
 		self.expressions[i].addDependencies(deps);
@@ -392,16 +394,16 @@ proto.addAccumulator = function addAccumulator(fieldName, accumulatorFactory, ex
  * @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*/) {
+proto.makeDocument = function makeDocument(id, accums, mergeableOutput) {
 	var out = {};
 
 	/* add the _id field */
-	out._id = id;
+	out._id = this.expandId(id);
 
 	/* add the rest of the fields */
 	this.fieldNames.forEach(function(fieldName, i) {
-		var val = accums[i].getValue(/*mergeableOutput*/);
-		if(!val) {
+		var val = accums[i].getValue(mergeableOutput);
+		if (!val) {
 			out[fieldName] = null;
 		} else {
 			out[fieldName] = val;
@@ -412,11 +414,96 @@ proto.makeDocument = function makeDocument(id, accums /*,mergeableOutput*/) {
 };
 
 /**
- * Sets the id expression for the group
+ * Computes the internal representation of the group key.
  *
- * @method setIdExpression
- * @param epxression {Expression} The expression to set
+ * @method computeId
+ * @param vars a VariablesParseState
+ * @return vals
+ */
+proto.computeId = function computeId(vars) {
+	var self = this;
+	// If only one expression return result directly
+	if (self.idExpressions.length === 1)
+		return self.idExpressions[0].evaluate(vars); // NOTE: self will probably need to be async soon
+
+	// Multiple expressions get results wrapped in an array
+	var vals = [];
+	self.idExpressions.forEach(function(expression, i) {
+		vals.push(expression.evaluate(vars));
+	});
+
+	return vals;
+};
+
+/**
+ * Converts the internal representation of the group key to the _id shape specified by the
+ * user.
+ *
+ * @method expandId
+ * @param val
+ * @return document representing an id
+ */
+proto.expandId = function expandId(val) {
+	var self = this;
+	// _id doesn't get wrapped in a document
+	if (self.idFieldNames.length === 0)
+		return val;
+
+	var doc = {};
+
+	// _id is a single-field document containing val
+	if (self.idFieldNames.length === 1) {
+		doc[self.idFieldNames[0]] = val;
+		return doc;
+	}
+
+	// _id is a multi-field document containing the elements of val
+	val.forEach(function(v, i) {
+		doc[self.idFieldNames[i]] = v;
+	});
+
+	return doc;
+};
+
+/**
+ * Parses the raw id expression into _idExpressions and possibly _idFieldNames.
+ *
+ * @method parseIdExpression
+ * @param groupField {Object} The object with the spec
+ */
+proto.parseIdExpression = function parseIdExpression(groupField, vps) {
+	var self = this;
+	if (self._getTypeStr(groupField) === 'Object' && Object.keys(groupField).length !== 0) {
+		// {_id: {}} is treated as grouping on a constant, not an expression
+
+		var idKeyObj = groupField;
+		if (Object.keys(idKeyObj)[0][0] == '$') {
+			var objCtx = new Expression.ObjectCtx({});
+			self.idExpressions.push(Expression.parseObject(idKeyObj, objCtx, vps));
+		} else {
+			Object.keys(idKeyObj).forEach(function(key, i) {
+				var field = {}; //idKeyObj[key];
+				field[key] = idKeyObj[key];
+				self.idFieldNames.push(key);
+				self.idExpressions.push(Expression.parseOperand(field[key], vps));
+			});
+		}
+	} else if (self._getTypeStr(groupField) === 'string' && groupField[0] === '$') {
+		self.idExpressions.push(FieldPathExpression.parse(groupField, vps));
+	} else {
+		self.idExpressions.push(ConstantExpression.create(groupField));
+	}
+};
+
+/**
+ * Get the type of something. Handles objects specially to return their true type; i.e. their constructor
+ *
+ * @method _getTypeStr
+ * @param obj {Object} The object to get the type of
+ * @return {String} The type of the object as a string
  **/
-proto.setIdExpression = function setIdExpression(expression) {
-	this.idExpression = expression;
+proto._getTypeStr = function _getTypeStr(obj) {
+	var typeofStr = typeof obj,
+		typeStr = (typeofStr == "object" && obj !== null) ? obj.constructor.name : typeofStr;
+	return typeStr;
 };

lib/pipeline/documentSources/LimitDocumentSource.js

@@ -57,8 +57,8 @@ proto.getNext = function getNext(callback) {
 
 	if (++this.count > this.limit) {
 		this.source.dispose();
-		callback(null, DocumentSource.EOF);
-		return DocumentSource.EOF;
+		callback(null, null);
+		return null;
 	}
 
 	return this.source.getNext(callback);

lib/pipeline/documentSources/MatchDocumentSource.js

@@ -50,9 +50,9 @@ proto.getNext = function getNext(callback) {
 			return self.matcher.matches(doc);
 		},
 		makeReturn = function makeReturn(doc) {
-			if(doc !== DocumentSource.EOF && test(doc)) { // Passes the match criteria
+			if(doc !== null && test(doc)) { // Passes the match criteria
 				return doc;
-			} else if(doc === DocumentSource.EOF){ // Got EOF
+			} else if(doc === null){ // Got EOF
 				return doc;
 			}
 			return undefined; // Didn't match, but not EOF
@@ -61,14 +61,14 @@ proto.getNext = function getNext(callback) {
 		function(cb) {
 			self.source.getNext(function(err, doc) {
 				if(err) return callback(err);
-				if (makeReturn(doc)) {
+				if (makeReturn(doc) !== undefined) {
 					next = doc;
 				}
 				return cb();
 			});
 		},
 		function() {
-			var foundDoc = (next === DocumentSource.EOF || next !== undefined);
+			var foundDoc = (next === null || next !== undefined);
 			return foundDoc; //keep going until doc is found
 		},
 		function(err) {

lib/pipeline/documentSources/ProjectDocumentSource.js

@@ -46,9 +46,9 @@ proto.getNext = function getNext(callback) {
 		if (err)
 			return callback(null, err);
 
-		if (input === DocumentSource.EOF) {
+		if (input === null) {
 			out = input;
-			return callback(null, DocumentSource.EOF);
+			return callback(null, null);
 		}
 
 		/* create the result document */
@@ -139,4 +139,4 @@ proto.toJSON = function toJSON(){
 	var obj = {};
 	this.sourceToJson(obj);
 	return obj;
-};
+};

lib/pipeline/documentSources/RedactDocumentSource.js

@@ -37,17 +37,17 @@ proto.getNext = function getNext(callback) {
 		doc;
 	async.whilst(
 		function() {
-			return doc !== DocumentSource.EOF;
+			return doc !== null;
 		},
 		function(cb) {
 			self.source.getNext(function(err, input) {
 				doc = input;
-				if (input === DocumentSource.EOF)
+				if (input === null)
 					return cb();
 				self._variables.setRoot(input);
 				self._variables.setValue(self._currentId, input);
 				var result = self.redactObject();
-				if (result !== DocumentSource.EOF)
+				if (result !== null)
 					return cb(result); //Using the err argument to pass the result document; this lets us break out without having EOF
 				return cb();
 			});
@@ -55,7 +55,7 @@ proto.getNext = function getNext(callback) {
 		function(doc) {
 			if (doc)
 				return callback(null, doc);
-			return callback(null, DocumentSource.EOF);
+			return callback(null, null);
 		}
 	);
 	return doc;
@@ -79,7 +79,7 @@ proto.redactValue = function redactValue(input) {
 	} else if (input instanceof Object && input.constructor === Object) {
 		this._variables.setValue(this._currentId, input);
 		var result = this.redactObject();
-		if (result !== DocumentSource.EOF)
+		if (result !== null)
 			return result;
 		return null;
 	} else {
@@ -96,7 +96,7 @@ proto.redactObject = function redactObject() {
 	if (expressionResult === KEEP_VAL) {
 		return this._variables.getDocument(this._currentId);
 	} else if (expressionResult === PRUNE_VAL) {
-		return DocumentSource.EOF;
+		return null;
 	} else if (expressionResult === DESCEND_VAL) {
 		var input = this._variables.getDocument(this._currentId);
 		var out = {};

lib/pipeline/documentSources/SkipDocumentSource.js

@@ -89,7 +89,7 @@ proto.getNext = function getNext(callback) {
 				});
 			},
 			function() {
-				return self.count < self.skip || next === DocumentSource.EOF;
+				return self.count < self.skip || next === null;
 			},
 			function (err) {
 				if (err) { return callback(err); }

lib/pipeline/documentSources/UnwindDocumentSource.js

@@ -1,6 +1,11 @@
 "use strict";
 
-var async = require("async");
+var async = require('async'),
+	DocumentSource = require('./DocumentSource'),
+	Expression = require('../expressions/Expression'),
+	FieldPath = require('../FieldPath'),
+	Value = require('../Value'),
+	Document = require('../Document');
 
 /**
  * A document source unwinder
@@ -11,61 +16,51 @@ var async = require("async");
  * @param [ctx] {ExpressionContext}
  **/
 var UnwindDocumentSource = module.exports = function UnwindDocumentSource(ctx){
-	if (arguments.length > 1) throw new Error("up to one arg expected");
-	base.call(this, ctx);
-
-	// Configuration state.
-	this._unwindPath = null;
+	if (arguments.length > 1) {
+		throw new Error('Up to one argument expected.');
+	}
 
-	// Iteration state.
-	this._unwinder = null;
+	base.call(this, ctx);
 
+	this._unwindPath = null; // Configuration state.
+	this._unwinder = null; // Iteration state.
 }, klass = UnwindDocumentSource, base = require('./DocumentSource'), proto = klass.prototype = Object.create(base.prototype, {constructor:{value:klass}});
 
-var DocumentSource = base,
-	FieldPath = require('../FieldPath'),
-	Document = require('../Document'),
-	Expression = require('../expressions/Expression');
+klass.unwindName = '$unwind';
 
-klass.Unwinder = (function(){
+klass.Unwinder = (function() {
 	/**
-	 * Helper class to unwind arrays within a series of documents.
-	 * @param	{String}	unwindPath is the field path to the array to unwind.
-	 **/
-	var klass = function Unwinder(unwindPath){
-		// Path to the array to unwind.
-		this._unwindPath = unwindPath;
-		// The souce document to unwind.
-		this._document = null;
-		// Document indexes of the field path components.
-		this._unwindPathFieldIndexes = [];
-		// Iterator over the array within _document to unwind.
-		this._unwindArrayIterator = null;
-		// The last value returned from _unwindArrayIterator.
-		//this._unwindArrayIteratorCurrent = undefined; //dont define this yet
-	}, base = Object, proto = klass.prototype = Object.create(base.prototype, {constructor:{value:klass}});
+	 * Construct a new Unwinder instance. Used as a parent class for UnwindDocumentSource.
+	 *
+	 * @param unwindPath
+	 * @constructor
+	 */
+	var klass = function Unwinder(unwindPath) {
+		this._unwindPath = new FieldPath(unwindPath);
 
-	/**
-	 * Reset the unwinder to unwind a new document.
-	 * @param	{Object}	document
-	 **/
-	proto.resetDocument = function resetDocument(document){
-		if (!document) throw new Error("document is required!");
+		this._inputArray = undefined;
+		this._document = undefined;
+		this._index = undefined;
+	}, base = Object, proto = klass.prototype = Object.create(base.prototype, {constructor: {value: klass}});
 
-		// Reset document specific attributes.
+	proto.resetDocument = function resetDocument(document) {
+		if (!document) throw new Error('Document is required!');
+
+		this._inputArray = [];
 		this._document = document;
-		this._unwindPathFieldIndexes.length = 0;
-		this._unwindArrayIterator = null;
-		delete this._unwindArrayIteratorCurrent;
+		this._index = 0;
 
-		var pathValue = this.extractUnwindValue(); // sets _unwindPathFieldIndexes
-		if (!pathValue || pathValue.length === 0) return;  // The path does not exist.
+		var pathValue = Document.getNestedField(this._document, this._unwindPath);
 
-		if (!(pathValue instanceof Array)) throw new Error(UnwindDocumentSource.unwindName + ":  value at end of field path must be an array; code 15978");
+		if (!pathValue || pathValue.length === 0) {
+			return;
+		}
 
-		// Start the iterator used to unwind the array.
-		this._unwindArrayIterator = pathValue.slice(0);
-		this._unwindArrayIteratorCurrent = this._unwindArrayIterator.splice(0,1)[0];
+		if (!(pathValue instanceof Array)) {
+			throw new Error(UnwindDocumentSource.unwindName + ':  value at end of field path must be an array; code 15978');
+		}
+
+		this._inputArray = pathValue;
 	};
 
 	/**
@@ -75,199 +70,144 @@ klass.Unwinder = (function(){
 	 * than the original mongo implementation, but should get updated to follow the current API.
 	 **/
 	proto.getNext = function getNext() {
-		if (this.eof())
-			return DocumentSource.EOF;
-
-		var output = this.getCurrent();
-		this.advance();
-		return output;
-	};
-
-	/**
-	 * eof
-	 * @returns	{Boolean}	true if done unwinding the last document passed to resetDocument().
-	 **/
-	proto.eof = function eof(){
-		return !this.hasOwnProperty("_unwindArrayIteratorCurrent");
-	};
-
-	/**
-	 * Try to advance to the next document unwound from the document passed to resetDocument().
-	 * @returns	{Boolean} true if advanced to a new unwound document, but false if done advancing.
-	 **/
-	proto.advance = function advance(){
-		if (!this._unwindArrayIterator) {
-			// resetDocument() has not been called or the supplied document had no results to
-			// unwind.
-			delete this._unwindArrayIteratorCurrent;
-		} else if (!this._unwindArrayIterator.length) {
-			// There are no more results to unwind.
-			delete this._unwindArrayIteratorCurrent;
-		} else {
-			this._unwindArrayIteratorCurrent = this._unwindArrayIterator.splice(0, 1)[0];
-		}
-	};
-
-	/**
-	 * Get the current document unwound from the document provided to resetDocument(), using
-	 * the current value in the array located at the provided unwindPath.  But return
-	 * intrusive_ptr<Document>() if resetDocument() has not been called or the results to unwind
-	 * have been exhausted.
-	 *
-	 * @returns	{Object}
-	 **/
-	proto.getCurrent = function getCurrent(){
-		if (!this.hasOwnProperty("_unwindArrayIteratorCurrent")) {
+		if (this._inputArray === undefined || this._index === this._inputArray.length) {
 			return null;
 		}
 
-		// Clone all the documents along the field path so that the end values are not shared across
-		// documents that have come out of this pipeline operator.  This is a partial deep clone.
-		// Because the value at the end will be replaced, everything along the path leading to that
-		// will be replaced in order not to share that change with any other clones (or the
-		// original).
-
-		var clone = Document.clone(this._document);
-		var current = clone;
-		var n = this._unwindPathFieldIndexes.length;
-		if (!n) throw new Error("unwindFieldPathIndexes are empty");
-		for (var i = 0; i < n; ++i) {
-			var fi = this._unwindPathFieldIndexes[i];
-			var fp = current[fi];
-			if (i + 1 < n) {
-				// For every object in the path but the last, clone it and continue on down.
-				var next = Document.clone(fp);
-				current[fi] = next;
-				current = next;
-			} else {
-				// In the last nested document, subsitute the current unwound value.
-				current[fi] = this._unwindArrayIteratorCurrent;
-			}
-		}
-
-		return clone;
-	};
-
-	/**
-	 * Get the value at the unwind path, otherwise an empty pointer if no such value
-	 * exists.  The _unwindPathFieldIndexes attribute will be set as the field path is traversed
-	 * to find the value to unwind.
-	 *
-	 * @returns	{Object}
-	 **/
-	proto.extractUnwindValue = function extractUnwindValue() {
-		var current = this._document;
-		var pathValue;
-		var pathLength = this._unwindPath.getPathLength();
-		for (var i = 0; i < pathLength; ++i) {
-
-			var idx = this._unwindPath.getFieldName(i);
-
-			if (!current.hasOwnProperty(idx)) return null; // The target field is missing.
-
-			// Record the indexes of the fields down the field path in order to quickly replace them
-			// as the documents along the field path are cloned.
-			this._unwindPathFieldIndexes.push(idx);
-
-			pathValue = current[idx];
+		this._document = Document.cloneDeep(this._document);
+		Document.setNestedField(this._document, this._unwindPath, this._inputArray[this._index++]);
 
-			if (i < pathLength - 1) {
-				if (typeof pathValue !== 'object') return null; // The next field in the path cannot exist (inside a non object).
-				current = pathValue; // Move down the object tree.
-			}
-		}
-
-		return pathValue;
+		return this._document;
 	};
 
 	return klass;
 })();
 
 /**
- * Specify the field to unwind.
-**/
-proto.unwindPath = function unwindPath(fieldPath){
-	// Can't set more than one unwind path.
-	if (this._unwindPath) throw new Error(this.getSourceName() + " can't unwind more than one path; code 15979");
-
-	// Record the unwind path.
-	this._unwindPath = new FieldPath(fieldPath);
-	this._unwinder = new klass.Unwinder(this._unwindPath);
-};
-
-klass.unwindName = "$unwind";
-
-proto.getSourceName = function getSourceName(){
+ * Get the document source name.
+ *
+ * @returns {string}
+ */
+proto.getSourceName = function getSourceName() {
 	return klass.unwindName;
 };
 
 /**
- * Get the fields this operation needs to do its job.
- * Deps should be in "a.b.c" notation
+ * Get the next source.
  *
- * @method	getDependencies
- * @param	{Object} deps	set (unique array) of strings
- * @returns	DocumentSource.GetDepsReturn
-**/
-proto.getDependencies = function getDependencies(deps) {
-	if (!this._unwindPath) throw new Error("unwind path does not exist!");
-	deps[this._unwindPath.getPath(false)] = 1;
-	return DocumentSource.GetDepsReturn.SEE_NEXT;
-};
-
+ * @param callback
+ * @returns {*}
+ */
 proto.getNext = function getNext(callback) {
-	if (!callback) throw new Error(this.getSourceName() + ' #getNext() requires callback');
+	if (!callback) {
+		throw new Error(this.getSourceName() + ' #getNext() requires callback.');
+	}
+
+	if (this.expCtx.checkForInterrupt && this.expCtx.checkForInterrupt() === false) {
+		return callback(new Error('Interrupted'));
+	}
 
 	var self = this,
 		out = this._unwinder.getNext(),
 		exhausted = false;
 
 	async.until(
-		function() {
-			if(out === DocumentSource.EOF && exhausted) return true;	// Really is EOF, not just an empty unwinder
-			else if(out !== DocumentSource.EOF) return true; // Return whatever we got that wasn't EOF
+		function () {
+			if (out !== null || exhausted) {
+				return true;
+			}
+
 			return false;
 		},
-		function(cb) {
-			self.source.getNext(function(err, doc) {
-				if(err) return cb(err);
-				out = doc;
-				if(out === DocumentSource.EOF) { // Our source is out of documents, we're done
+		function (cb) {
+			self.source.getNext(function (err, doc) {
+				if (err) {
+					return cb(err);
+				}
+
+				if (doc === null) {
 					exhausted = true;
-					return cb();
 				} else {
 					self._unwinder.resetDocument(doc);
 					out = self._unwinder.getNext();
-					return cb();
 				}
+
+				return cb();
 			});
 		},
 		function(err) {
-			if(err) return callback(err);
+			if (err) {
+				return callback(err);
+			}
+
 			return callback(null, out);
 		}
 	);
 
-	return out; //For sync mode
+	return out;
 };
 
+/**
+ * Serialize the data.
+ *
+ * @param explain
+ * @returns {{}}
+ */
 proto.serialize = function serialize(explain) {
-	if (!this._unwindPath) throw new Error("unwind path does not exist!");
+	if (!this._unwindPath) {
+		throw new Error('unwind path does not exist!');
+	}
+
 	var doc = {};
+
 	doc[this.getSourceName()] = this._unwindPath.getPath(true);
+
 	return doc;
 };
 
+/**
+ * Get the fields this operation needs to do its job.
+ *
+ * @param deps
+ * @returns {DocumentSource.GetDepsReturn.SEE_NEXT|*}
+ */
+proto.getDependencies = function getDependencies(deps) {
+	if (!this._unwindPath) {
+		throw new Error('unwind path does not exist!');
+	}
+
+	deps[this._unwindPath.getPath(false)] = 1;
+
+	return DocumentSource.GetDepsReturn.SEE_NEXT;
+};
+
+/**
+ * Unwind path.
+ *
+ * @param fieldPath
+ */
+proto.unwindPath = function unwindPath(fieldPath) {
+	if (this._unwindPath) {
+		throw new Error(this.getSourceName() + ' can\'t unwind more than one path; code 15979');
+	}
+
+	// Record the unwind path.
+	this._unwindPath = new FieldPath(fieldPath);
+	this._unwinder = new klass.Unwinder(fieldPath);
+};
+
 /**
  * Creates a new UnwindDocumentSource with the input path as the path to unwind
  * @param {String} JsonElement this thing is *called* Json, but it expects a string
 **/
 klass.createFromJson = function createFromJson(jsonElement, ctx) {
-	// The value of $unwind should just be a field path.
-	if (jsonElement.constructor !== String) throw new Error("the " + klass.unwindName + " field path must be specified as a string; code 15981");
+	if (jsonElement.constructor !== String) {
+		throw new Error('the ' + klass.unwindName + ' field path must be specified as a string; code 15981');
+	}
+
+	var pathString = Expression.removeFieldPrefix(jsonElement),
+		unwind = new UnwindDocumentSource(ctx);
 
-	var pathString = Expression.removeFieldPrefix(jsonElement);
-	var unwind = new UnwindDocumentSource(ctx);
 	unwind.unwindPath(pathString);
 
 	return unwind;

lib/pipeline/expressions/CompareExpression.js

@@ -9,8 +9,8 @@
  */
 var CompareExpression = module.exports = function CompareExpression(cmpOp) {
 	if (!(arguments.length === 1 && typeof cmpOp === "string")) throw new Error(klass.name + ": args expected: cmpOp");
-    this.cmpOp = cmpOp;
-    base.call(this);
+	this.cmpOp = cmpOp;
+	base.call(this);
 }, klass = CompareExpression, base = require("./FixedArityExpressionT")(CompareExpression, 2), proto = klass.prototype = Object.create(base.prototype, {constructor:{value:klass}});
 
 
@@ -82,7 +82,7 @@ proto.evaluateInternal = function evaluateInternal(vars) {
 		right = this.operands[1].evaluateInternal(vars),
 		cmp = Value.compare(left, right);
 
-    // Make cmp one of 1, 0, or -1.
+	// Make cmp one of 1, 0, or -1.
 	if (cmp === 0) {
 		//leave as 0
 	} else if (cmp < 0) {

lib/pipeline/expressions/CondExpression.js

@@ -9,11 +9,11 @@
  */
 var CondExpression = module.exports = function CondExpression() {
 	if (arguments.length !== 0) throw new Error(klass.name + ": expected args: NONE");
-    base.call(this);
+	base.call(this);
 }, klass = CondExpression, base = require("./FixedArityExpressionT")(CondExpression, 3), proto = klass.prototype = Object.create(base.prototype, {constructor:{value:klass}});
 
 var Value = require("../Value"),
-    Expression = require("./Expression");
+	Expression = require("./Expression");
 
 proto.evaluateInternal = function evaluateInternal(vars) {
 	var cond = this.operands[0].evaluateInternal(vars);
@@ -22,12 +22,12 @@ proto.evaluateInternal = function evaluateInternal(vars) {
 };
 
 klass.parse = function parse(expr, vps) {
-    if (Value.getType(expr) !== "Object") {
+	if (Value.getType(expr) !== "Object") {
 		return base.parse(expr, vps);
 	}
 	// verify(str::equals(expr.fieldName(), "$cond")); //NOTE: DEVIATION FROM MONGO: we do not have fieldName any more and not sure this is even possible anyway
 
-    var ret = new CondExpression();
+	var ret = new CondExpression();
 	ret.operands.length = 3;
 
 	var args = expr;
@@ -44,11 +44,11 @@ klass.parse = function parse(expr, vps) {
 		}
 	}
 
-    if (!ret.operands[0]) throw new Error("Missing 'if' parameter to $cond; uassert code 17080");
-    if (!ret.operands[1]) throw new Error("Missing 'then' parameter to $cond; uassert code 17081");
-    if (!ret.operands[2]) throw new Error("Missing 'else' parameter to $cond; uassert code 17082");
+	if (!ret.operands[0]) throw new Error("Missing 'if' parameter to $cond; uassert code 17080");
+	if (!ret.operands[1]) throw new Error("Missing 'then' parameter to $cond; uassert code 17081");
+	if (!ret.operands[2]) throw new Error("Missing 'else' parameter to $cond; uassert code 17082");
 
-    return ret;
+	return ret;
 };
 
 Expression.registerExpression("$cond", CondExpression.parse);

lib/pipeline/expressions/ConstantExpression.js

@@ -8,9 +8,9 @@
  * @constructor
  */
 var ConstantExpression = module.exports = function ConstantExpression(value){
-    if (arguments.length !== 1) throw new Error(klass.name + ": args expected: value");
-    this.value = value;
-    base.call(this);
+	if (arguments.length !== 1) throw new Error(klass.name + ": args expected: value");
+	this.value = value;
+	base.call(this);
 }, klass = ConstantExpression, base = require("./FixedArityExpressionT")(ConstantExpression, 1), proto = klass.prototype = Object.create(base.prototype, {constructor:{value:klass}});
 
 var Expression = require("./Expression");
@@ -43,7 +43,7 @@ proto.evaluateInternal = function evaluateInternal(vars) {
 
 /// Helper function to easily wrap constants with $const.
 function serializeConstant(val) {
-    return {$const: val};
+	return {$const: val};
 }
 
 proto.serialize = function serialize(explain) {
@@ -59,5 +59,5 @@ proto.getOpName = function getOpName() {
 };
 
 proto.getValue = function getValue() {
-    return this.value;
+	return this.value;
 };

lib/pipeline/expressions/DayOfMonthExpression.js

@@ -8,8 +8,8 @@
  * @constructor
  */
 var DayOfMonthExpression = module.exports = function DayOfMonthExpression() {
-    if (arguments.length !== 0) throw new Error(klass.name + ": no args expected");
-    base.call(this);
+	if (arguments.length !== 0) throw new Error(klass.name + ": no args expected");
+	base.call(this);
 }, klass = DayOfMonthExpression, base = require("./FixedArityExpressionT")(klass, 1), proto = klass.prototype = Object.create(base.prototype, {constructor: {value: klass}});
 
 var Expression = require("./Expression"),

lib/pipeline/expressions/DivideExpression.js

@@ -10,8 +10,8 @@
  * @constructor
  **/
 var DivideExpression = module.exports = function DivideExpression(){
-    if (arguments.length !== 0) throw new Error(klass.name + ": no args expected");
-    base.call(this);
+	if (arguments.length !== 0) throw new Error(klass.name + ": no args expected");
+	base.call(this);
 }, klass = DivideExpression, base = require("./FixedArityExpressionT")(DivideExpression, 2), proto = klass.prototype = Object.create(base.prototype, {constructor:{value:klass}});
 
 var Value = require("../Value"),
@@ -26,16 +26,16 @@ proto.evaluateInternal = function evaluateInternal(vars) {
 		rhs = this.operands[1].evaluateInternal(vars);
 
 	if (typeof lhs === "number" && typeof rhs === "number") {
-        var numer = lhs,
-            denom = rhs;
-        if (denom === 0) throw new Error("can't $divide by zero; uassert code 16608");
+		var numer = lhs,
+			denom = rhs;
+		if (denom === 0) throw new Error("can't $divide by zero; uassert code 16608");
 
-        return numer / denom;
-    } else if (lhs === undefined || lhs === null || rhs === undefined || rhs === null) {
-        return null;
-    } else{
-        throw new Error("User assertion: 16609: $divide only supports numeric types, not " + Value.getType(lhs) + " and " + Value.getType(rhs));
-    }
+		return numer / denom;
+	} else if (lhs === undefined || lhs === null || rhs === undefined || rhs === null) {
+		return null;
+	} else{
+		throw new Error("User assertion: 16609: $divide only supports numeric types, not " + Value.getType(lhs) + " and " + Value.getType(rhs));
+	}
 };
 
 Expression.registerExpression("$divide", base.parse);

lib/pipeline/expressions/Expression.js

@@ -2,12 +2,6 @@
 
 /**
  * A base class for all pipeline expressions; Performs common expressions within an Op.
- *
- * NOTE: An object expression can take any of the following forms:
- *
- *      f0: {f1: ..., f2: ..., f3: ...}
- *      f0: {$operator:[operand1, operand2, ...]}
- *
  * @class Expression
  * @namespace mungedb-aggregate.pipeline.expressions
  * @module mungedb-aggregate
@@ -80,12 +74,12 @@ var ObjectCtx = Expression.ObjectCtx = (function() {
  */
 klass.parseObject = function parseObject(obj, ctx, vps) {
 	if (!(ctx instanceof ObjectCtx)) throw new Error("ctx must be ObjectCtx");
-	/*
-	  An object expression can take any of the following forms:
-
-	  f0: {f1: ..., f2: ..., f3: ...}
-	  f0: {$operator:[operand1, operand2, ...]}
-	*/
+	/**
+	 * An object expression can take any of the following forms:
+	 *
+	 * f0: {f1: ..., f2: ..., f3: ...}
+	 * f0: {$operator:[operand1, operand2, ...]}
+	 */
 
 	var expression, // the result
 		expressionObject, // the alt result
@@ -223,7 +217,7 @@ klass.parseOperand = function parseOperand(exprElement, vps) {
 	var t = typeof(exprElement);
 	if (t === "string" && exprElement[0] === "$") {
 		//if we got here, this is a field path expression
-	    return FieldPathExpression.parse(exprElement, vps);
+		return FieldPathExpression.parse(exprElement, vps);
 	} else if (t === "object" && exprElement && exprElement.constructor === Object) {
 		var oCtx = new ObjectCtx({
 			isDocumentOk: true

lib/pipeline/expressions/FieldPathExpression.js

@@ -1,8 +1,8 @@
 "use strict";
 
 var Expression = require("./Expression"),
-    Variables = require("./Variables"),
-    FieldPath = require("../FieldPath");
+	Variables = require("./Variables"),
+	FieldPath = require("../FieldPath");
 
 /**
  * Create a field path expression.
@@ -18,9 +18,9 @@ var Expression = require("./Expression"),
  * @param {String} theFieldPath the field path string, without any leading document indicator
  */
 var FieldPathExpression = module.exports = function FieldPathExpression(theFieldPath, variable) {
-    if (arguments.length != 2) throw new Error(klass.name + ": expected args: theFieldPath[, variable]");
-    this._fieldPath = new FieldPath(theFieldPath);
-    this._variable = variable;
+	if (arguments.length != 2) throw new Error(klass.name + ": expected args: theFieldPath[, variable]");
+	this._fieldPath = new FieldPath(theFieldPath);
+	this._variable = variable;
 }, klass = FieldPathExpression, base = Expression, proto = klass.prototype = Object.create(base.prototype, {constructor:{value:klass}});
 
 /**
@@ -35,9 +35,9 @@ var FieldPathExpression = module.exports = function FieldPathExpression(theField
  * @param fieldPath the field path string, without any leading document
  * indicator
  * @returns the newly created field path expression
- **/
+ */
 klass.create = function create(fieldPath) {
-    return new FieldPathExpression("CURRENT." + fieldPath, Variables.ROOT_ID);
+	return new FieldPathExpression("CURRENT." + fieldPath, Variables.ROOT_ID);
 };
 
 // this is the new version that supports every syntax
@@ -48,52 +48,52 @@ klass.create = function create(fieldPath) {
  * @returns a new FieldPathExpression
  */
 klass.parse = function parse(raw, vps) {
-    if (raw[0] !== "$") throw new Error("FieldPath: '" + raw + "' doesn't start with a $; uassert code 16873");
-    if (raw.length < 2) throw new Error("'$' by itself is not a valid FieldPath; uassert code 16872"); // need at least "$" and either "$" or a field name
-    if (raw[1] === "$") {
-        var fieldPath = raw.substr(2), // strip off $$
-            dotIndex = fieldPath.indexOf("."),
-            varName = fieldPath.substr(0, dotIndex !== -1 ? dotIndex : fieldPath.length);
-        Variables.uassertValidNameForUserRead(varName);
-        return new FieldPathExpression(fieldPath, vps.getVariable(varName));
-    } else {
-        return new FieldPathExpression("CURRENT." + raw.substr(1), // strip the "$" prefix
-            vps.getVariable("CURRENT"));
-    }
+	if (raw[0] !== "$") throw new Error("FieldPath: '" + raw + "' doesn't start with a $; uassert code 16873");
+	if (raw.length < 2) throw new Error("'$' by itself is not a valid FieldPath; uassert code 16872"); // need at least "$" and either "$" or a field name
+	if (raw[1] === "$") {
+		var fieldPath = raw.substr(2), // strip off $$
+			dotIndex = fieldPath.indexOf("."),
+			varName = fieldPath.substr(0, dotIndex !== -1 ? dotIndex : fieldPath.length);
+		Variables.uassertValidNameForUserRead(varName);
+		return new FieldPathExpression(fieldPath, vps.getVariable(varName));
+	} else {
+		return new FieldPathExpression("CURRENT." + raw.substr(1), // strip the "$" prefix
+			vps.getVariable("CURRENT"));
+	}
 };
 
 proto.optimize = function optimize() {
-    // nothing can be done for these
-    return this;
+	// nothing can be done for these
+	return this;
 };
 
 proto.addDependencies = function addDependencies(deps) {
-    if (this._variable === Variables.ROOT_ID) {
-        if (this._fieldPath.fieldNames.length === 1) {
-            deps.needWholeDocument = true; // need full doc if just "$$ROOT"
-        } else {
-            deps.fields[this._fieldPath.tail().getPath(false)] = 1;
-        }
-    }
+	if (this._variable === Variables.ROOT_ID) {
+		if (this._fieldPath.fieldNames.length === 1) {
+			deps.needWholeDocument = true; // need full doc if just "$$ROOT"
+		} else {
+			deps.fields[this._fieldPath.tail().getPath(false)] = 1;
+		}
+	}
 };
 
 /**
  * Helper for evaluatePath to handle Array case
  */
 proto._evaluatePathArray = function _evaluatePathArray(index, input) {
-    if (!(input instanceof Array)) throw new Error("must be array; dassert");
-
-    // Check for remaining path in each element of array
-    var result = [];
-    for (var i = 0, l = input.length; i < l; i++) {
-        if (!(input[i] instanceof Object))
-            continue;
-
-        var nested = this._evaluatePath(index, input[i]);
-        if (nested !== undefined)
-            result.push(nested);
-    }
-    return result;
+	if (!(input instanceof Array)) throw new Error("must be array; dassert");
+
+	// Check for remaining path in each element of array
+	var result = [];
+	for (var i = 0, l = input.length; i < l; i++) {
+		if (!(input[i] instanceof Object))
+			continue;
+
+		var nested = this._evaluatePath(index, input[i]);
+		if (nested !== undefined)
+			result.push(nested);
+	}
+	return result;
 };
 
 /**
@@ -110,52 +110,52 @@ proto._evaluatePathArray = function _evaluatePathArray(index, input) {
  * @returns the field found; could be an array
  */
 proto._evaluatePath = function _evaluatePath(index, input) {
-    // Note this function is very hot so it is important that is is well optimized.
-    // In particular, all return paths should support RVO.
-
-    // if we've hit the end of the path, stop
-    if (index == this._fieldPath.fieldNames.length - 1)
-        return input[this._fieldPath.fieldNames[index]];
-
-    // Try to dive deeper
-    var val = input[this._fieldPath.fieldNames[index]];
-    if (val instanceof Object && val.constructor === Object) {
-        return this._evaluatePath(index + 1, val);
-    } else if (val instanceof Array) {
-        return this._evaluatePathArray(index + 1, val);
-    } else {
-        return undefined;
-    }
+	// Note this function is very hot so it is important that is is well optimized.
+	// In particular, all return paths should support RVO.
+
+	// if we've hit the end of the path, stop
+	if (index == this._fieldPath.fieldNames.length - 1)
+		return input[this._fieldPath.fieldNames[index]];
+
+	// Try to dive deeper
+	var val = input[this._fieldPath.fieldNames[index]];
+	if (val instanceof Object && val.constructor === Object) {
+		return this._evaluatePath(index + 1, val);
+	} else if (val instanceof Array) {
+		return this._evaluatePathArray(index + 1, val);
+	} else {
+		return undefined;
+	}
 };
 
 proto.evaluateInternal = function evaluateInternal(vars) {
-    if (this._fieldPath.fieldNames.length === 1) // get the whole variable
-        return vars.getValue(this._variable);
-
-    if (this._variable === Variables.ROOT_ID) {
-        // ROOT is always a document so use optimized code path
-        return this._evaluatePath(1, vars.getRoot());
-    }
-
-    var val = vars.getValue(this._variable);
-    if (val instanceof Object && val.constructor === Object) {
-        return this._evaluatePath(1, val);
-    } else if(val instanceof Array) {
-        return this._evaluatePathArray(1,val);
-    } else {
-        return undefined;
-    }
+	if (this._fieldPath.fieldNames.length === 1) // get the whole variable
+		return vars.getValue(this._variable);
+
+	if (this._variable === Variables.ROOT_ID) {
+		// ROOT is always a document so use optimized code path
+		return this._evaluatePath(1, vars.getRoot());
+	}
+
+	var val = vars.getValue(this._variable);
+	if (val instanceof Object && val.constructor === Object) {
+		return this._evaluatePath(1, val);
+	} else if(val instanceof Array) {
+		return this._evaluatePathArray(1,val);
+	} else {
+		return undefined;
+	}
 };
 
 proto.serialize = function serialize(){
-    if(this._fieldPath.fieldNames[0] === "CURRENT" && this._fieldPath.fieldNames.length > 1) {
-        // use short form for "$$CURRENT.foo" but not just "$$CURRENT"
-        return "$" + this._fieldPath.tail().getPath(false);
-    } else {
-        return "$$" + this._fieldPath.getPath(false);
-    }
+	if(this._fieldPath.fieldNames[0] === "CURRENT" && this._fieldPath.fieldNames.length > 1) {
+		// use short form for "$$CURRENT.foo" but not just "$$CURRENT"
+		return "$" + this._fieldPath.tail().getPath(false);
+	} else {
+		return "$$" + this._fieldPath.getPath(false);
+	}
 };
 
 proto.getFieldPath = function getFieldPath(){
-    return this._fieldPath;
+	return this._fieldPath;
 };

lib/pipeline/expressions/index.js

@@ -4,6 +4,7 @@ module.exports = {
 	AndExpression: require("./AndExpression.js"),
 	CoerceToBoolExpression: require("./CoerceToBoolExpression.js"),
 	CompareExpression: require("./CompareExpression.js"),
+	ConcatExpression: require("./ConcatExpression.js"),
 	CondExpression: require("./CondExpression.js"),
 	ConstantExpression: require("./ConstantExpression.js"),
 	DayOfMonthExpression: require("./DayOfMonthExpression.js"),
@@ -30,6 +31,7 @@ module.exports = {
 	SubtractExpression: require("./SubtractExpression.js"),
 	ToLowerExpression: require("./ToLowerExpression.js"),
 	ToUpperExpression: require("./ToUpperExpression.js"),
+	VariadicExpressionT: require("./VariadicExpressionT.js"),
 	WeekExpression: require("./WeekExpression.js"),
 	YearExpression: require("./YearExpression.js")
 };

lib/query/ArrayRunner.js

@@ -0,0 +1,89 @@
+"use strict";
+
+var Runner = require('./Runner');
+
+/**
+ * This class is an array runner used to run a pipeline against a static array of data
+ * @param	{Array}	items	The array source of the data
+ **/
+var klass = module.exports = function ArrayRunner(array){
+	base.call(this);
+	
+	if (!array || array.constructor !== Array ) throw new Error('Array runner requires an array');
+	this._array = array;
+	this._position = 0;
+	this._state = Runner.RunnerState.RUNNER_ADVANCED;
+}, base = Runner, proto = klass.prototype = Object.create(base.prototype, {constructor:{value:klass}});
+
+/**
+ * Get the next result from the array.
+ * 
+ * @method getNext
+ * @param [callback] {Function}
+ */
+proto.getNext = function getNext(callback) {
+	var obj, err;
+	try {
+		if (this._state === Runner.RunnerState.RUNNER_ADVANCED) {
+			if (this._position < this._array.length){
+				obj = this._array[this._position++];
+			} else {
+				this._state = Runner.RunnerState.RUNNER_EOF;
+			}
+		}
+	} catch (ex) {
+		err = ex;
+		this._state = Runner.RunnerState.RUNNER_ERROR;
+	}
+	
+	return callback(err, obj, this._state);
+};
+
+/**
+ * Save any state required to yield.
+ * 
+ * @method saveState
+ */
+proto.saveState = function saveState() {
+	//nothing to do here
+};
+
+/**
+ * Restore saved state, possibly after a yield.  Return true if the runner is OK, false if
+ * it was killed.
+ * 
+ * @method restoreState
+ */
+proto.restoreState = function restoreState() {
+	//nothing to do here
+};
+
+/**
+ * Returns a description of the Runner
+ * 
+ * @method getInfo
+ * @param [explain]
+ * @param [planInfo]
+ */
+proto.getInfo = function getInfo(explain) {
+	if (explain){
+		return {
+			type: this.constructor.name,
+			nDocs: this._array.length,
+			position: this._position,
+			state: this._state
+		};
+	}
+	return undefined;
+};
+
+/**
+ * dispose of the Runner.
+ * 
+ * @method reset
+ */
+proto.reset = function reset(){
+	this._array = [];
+	this._position = 0;
+	this._state = Runner.RunnerState.RUNNER_DEAD;
+};

lib/query/Runner.js

@@ -0,0 +1,222 @@
+"use strict";
+
+/**
+ * This class is an implementation of the base class for runners used in MongoDB
+ * 
+ * Note that a lot of stuff here is not used by our code yet.  Check the existing implementations
+ * for what we currently use
+ * 
+ **/
+var klass = module.exports = function Runner(){
+	
+}, base = Object, proto = klass.prototype = Object.create(base.prototype, {constructor:{value:klass}});
+
+klass.RunnerState = {
+	// We successfully populated the out parameter.
+	RUNNER_ADVANCED: "RUNNER_ADVANCED",
+
+	// We're EOF.  We won't return any more results (edge case exception: capped+tailable).
+	RUNNER_EOF: "RUNNER_EOF",
+
+	// We were killed or had an error.
+	RUNNER_DEAD: "RUNNER_DEAD",
+
+	// getNext was asked for data it cannot provide, or the underlying PlanStage had an
+	// unrecoverable error.
+	// If the underlying PlanStage has any information on the error, it will be available in
+	// the objOut parameter. Call WorkingSetCommon::toStatusString() to retrieve the error
+	// details from the output BSON object.
+	RUNNER_ERROR: "RUNNER_ERROR"
+};
+
+klass.YieldPolicy = {
+	// Any call to getNext() may yield.  In particular, the runner may be killed during any
+	// call to getNext().  If this occurs, getNext() will return RUNNER_DEAD.
+	//
+	// If you are enabling autoyield, you must register the Runner with ClientCursor via
+	// ClientCursor::registerRunner and deregister via ClientCursor::deregisterRunnerwhen
+	// done.  Registered runners are informed about DiskLoc deletions and Namespace
+	// invalidations and other important events.
+	//
+	// Exception: This is not required if the Runner is cached inside of a ClientCursor.
+	// This is only done if the Runner is cached and can be referred to by a cursor id.
+	// This is not a popular thing to do.
+	YIELD_AUTO: "YIELD_AUTO",
+
+	// Owner must yield manually if yields are requested.  How to yield yourself:
+	//
+	// 0. Let's say you have Runner* runner.
+	//
+	// 1. Register your runner with ClientCursor.  Registered runners are informed about
+	// DiskLoc deletions and Namespace invalidation and other important events.  Do this by
+	// calling ClientCursor::registerRunner(runner).  This could be done once when you get
+	// your runner, or per-yield.
+	//
+	// 2. Call runner->saveState() before you yield.
+	//
+	// 3. Call RunnerYieldPolicy::staticYield(runner->ns(), NULL) to yield.  Any state that
+	// may change between yields must be checked by you.  (For example, DiskLocs may not be
+	// valid across yielding, indices may be dropped, etc.)
+	//
+	// 4. Call runner->restoreState() before using the runner again.
+	//
+	// 5. Your runner's next call to getNext may return RUNNER_DEAD.
+	//
+	// 6. When you're done with your runner, deregister it from ClientCursor via
+	// ClientCursor::deregister(runner).
+	YIELD_MANUAL: "YIELD_MANUAL"
+};
+
+
+/**
+ * Set the yielding policy of the underlying runner.  See the RunnerYieldPolicy enum above.
+ * 
+ * @method setYieldPolicy
+ * @param [policy]
+ */
+proto.setYieldPolicy = function setYieldPolicy(policy) {
+	throw new Error('Not implemented');
+};
+
+/**
+ * Get the next result from the query.
+ *
+ * If objOut is not NULL, only results that have a BSONObj are returned.  The BSONObj may
+ * point to on-disk data (isOwned will be false) and must be copied by the caller before
+ * yielding.
+ *
+ * If dlOut is not NULL, only results that have a valid DiskLoc are returned.
+ *
+ * If both objOut and dlOut are not NULL, only results with both a valid BSONObj and DiskLoc
+ * will be returned.  The BSONObj is the object located at the DiskLoc provided.
+ *
+ * If the underlying query machinery produces a result that does not have the data requested
+ * by the user, it will be silently dropped.
+ *
+ * If the caller is running a query, they probably only care about the object.
+ * If the caller is an internal client, they may only care about DiskLocs (index scan), or
+ * about object + DiskLocs (collection scan).
+ *
+ * Some notes on objOut and ownership:
+ *
+ * objOut may be an owned object in certain cases: invalidation of the underlying DiskLoc,
+ * the object is created from covered index key data, the object is projected or otherwise
+ * the result of a computation.
+ *
+ * objOut will also be owned when the underlying PlanStage has provided error details in the
+ * event of a RUNNER_ERROR. Call WorkingSetCommon::toStatusString() to convert the object
+ * to a loggable format.
+ *
+ * objOut will be unowned if it's the result of a fetch or a collection scan.
+ * 
+ * @method getNext
+ * @param [callback] {Function}
+ */
+proto.getNext = function getNext(callback) {
+	throw new Error('Not implemented');
+};
+
+
+/**
+ * Will the next call to getNext() return EOF?  It's useful to know if the runner is done
+ * without having to take responsibility for a result.
+ * 
+ * @method isEOF
+ */
+proto.isEOF = function isEOF(){
+	throw new Error('Not implemented');
+};
+
+/**
+ * Inform the runner about changes to DiskLoc(s) that occur while the runner is yielded.
+ * The runner must take any actions required to continue operating correctly, including
+ * broadcasting the invalidation request to the PlanStage tree being run.
+ *
+ * Called from CollectionCursorCache::invalidateDocument.
+ *
+ * See db/invalidation_type.h for InvalidationType.
+ * 
+ * @method invalidate
+ * @param [dl]
+ * @param [type]
+ */
+proto.invalidate = function invalidate(dl, type) {
+	throw new Error('Not implemented');
+};
+
+/**
+ * Mark the Runner as no longer valid.  Can happen when a runner yields and the underlying
+ * database is dropped/indexes removed/etc.  All future to calls to getNext return
+ * RUNNER_DEAD. Every other call is a NOOP.
+ *
+ * The runner must guarantee as a postcondition that future calls to collection() will
+ * return NULL.
+ * 
+ * @method kill
+ */
+proto.kill = function kill() {
+	throw new Error('Not implemented');
+};
+
+/**
+ * Save any state required to yield.
+ * 
+ * @method saveState
+ */
+proto.saveState = function saveState() {
+	throw new Error('Not implemented');
+};
+
+/**
+ * Restore saved state, possibly after a yield.  Return true if the runner is OK, false if
+ * it was killed.
+ * 
+ * @method restoreState
+ */
+proto.restoreState = function restoreState() {
+	throw new Error('Not implemented');
+};
+
+/**
+ * Return the NS that the query is running over.
+ * 
+ * @method ns
+ */
+proto.ns = function ns() {
+	throw new Error('Not implemented');
+};
+
+/**
+ * Return the Collection that the query is running over.
+ * 
+ * @method collection
+ */
+proto.collection = function collection() {
+	throw new Error('Not implemented');
+};
+
+/**
+ * Returns OK, allocating and filling '*explain' or '*planInfo' with a description of the
+ * chosen plan, depending on which is non-NULL (one of the two should be NULL). Caller
+ * takes onwership of either '*explain' and '*planInfo'. Otherwise, returns false
+ * a detailed error status.
+ *
+ * If 'explain' is NULL, then this out-parameter is ignored. Similarly, if 'staticInfo'
+ * is NULL, then no static debug information is produced.
+ * 
+ * @method getInfo
+ * @param [explain]
+ * @param [planInfo]
+ */
+proto.getInfo = function getInfo(explain, planInfo) {
+	throw new Error('Not implemented');
+};
+
+/**
+ * dispose of the Runner.
+ * 
+ * @method reset
+ */
+proto.reset = function reset(){
+	throw new Error('Not implemented');
+};

lib/query/index.js

@@ -0,0 +1,5 @@
+"use strict";
+module.exports = {
+	Runner: require("./Runner.js"),
+	ArrayRunner: require("./ArrayRunner.js")
+};

test/lib/Cursor.js

@@ -1,93 +0,0 @@
-"use strict";
-var assert = require("assert"),
-	Cursor = require("../../lib/Cursor");
-
-module.exports = {
-
-	"Cursor": {
-
-		"constructor(data)": {
-			"should throw an exception if it does not get a valid array or stream": function(){
-				assert.throws(function(){
-					var c = new Cursor();
-				});
-				assert.throws(function(){
-					var c = new Cursor(5);
-				});
-			}
-		},
-
-		"#ok": {
-			"should return true if there is still data in the array": function(){
-				var c = new Cursor([1,2,3,4,5]);
-				assert.equal(c.ok(), true);
-			},
-			"should return false if there is no data left in the array": function(){
-				var c = new Cursor([]);
-				assert.equal(c.ok(), false);
-			},
-			"should return true if there is no data left in the array, but there is still a current value": function(){
-				var c = new Cursor([1,2]);
-				c.advance();
-				c.advance();
-				assert.equal(c.ok(), true);
-				c.advance();
-				assert.equal(c.ok(), false);
-			}
-//			,
-//			"should return true if there is still data in the stream": function(){
-//				
-//			},
-//			"should return false if there is no data left in the stream": function(){
-//				
-//			}
-
-		},
-		
-		"#advance": {
-			"should return true if there is still data in the array": function(){
-				var c = new Cursor([1,2,3,4,5]);
-				assert.equal(c.advance(), true);
-			},
-			"should return false if there is no data left in the array": function(){
-				var c = new Cursor([1]);
-				c.advance();
-				assert.equal(c.advance(), false);
-			},
-			"should update the current object to the next item in the array": function(){
-				var c = new Cursor([1,"2"]);
-				c.advance();
-				assert.strictEqual(c.current(), 1);
-				c.advance();
-				assert.strictEqual(c.current(), "2");
-				c.advance();
-				assert.strictEqual(c.current(), undefined);
-			}
-//,			"should return true if there is still data in the stream": function(){
-//				
-//			},
-//			"should return false if there is no data left in the stream": function(){
-//				
-//			},
-//			"should update the current object to the next item in the stream": function(){
-//				
-//			}
-		},
-		
-		"#current": {
-			"should return the first value if the cursor has not been advanced yet": function(){
-				var c = new Cursor([1,2,3,4,5]);
-				assert.equal(c.current(), 1);
-			},
-			"should return the first value if the cursor has been advanced once": function(){
-				var c = new Cursor([1,2,3,4,5]);
-				c.advance();
-				assert.equal(c.current(), 1);
-			}
-		}
-
-	}
-
-};
-
-if (!module.parent)(new(require("mocha"))()).ui("exports").reporter("spec").addFile(__filename).run();

test/lib/pipeline/documentSources/CursorDocumentSource.js

@@ -5,14 +5,11 @@ var assert = require("assert"),
 	CursorDocumentSource = require("../../../../lib/pipeline/documentSources/CursorDocumentSource"),
 	LimitDocumentSource = require("../../../../lib/pipeline/documentSources/LimitDocumentSource"),
 	SkipDocumentSource = require("../../../../lib/pipeline/documentSources/SkipDocumentSource"),
-	Cursor = require("../../../../lib/Cursor");
-
-var getCursor = function(values) {
-	if (!values)
-		values = [1,2,3,4,5];
-	var cwc = new CursorDocumentSource.CursorWithContext();
-	cwc._cursor = new Cursor( values );
-	return new CursorDocumentSource(cwc);
+	ArrayRunner = require("../../../../lib/query/ArrayRunner");
+
+var getCursorDocumentSource = function(values) {
+	values = values || [1,2,3,4,5];
+	return new CursorDocumentSource(null, new ArrayRunner(values), null);
 };
 
 
@@ -21,24 +18,15 @@ module.exports = {
 	"CursorDocumentSource": {
 
 		"constructor(data)": {
-			"should fail if CursorWithContext is not provided": function(){
-				assert.throws(function(){
-					var cds = new CursorDocumentSource();
-				});
-			},
 			"should get a accept a CursorWithContext and set it internally": function(){
-				var cwc = new CursorDocumentSource.CursorWithContext();
-				cwc._cursor = new Cursor( [] );
-
-				var cds = new CursorDocumentSource(cwc);
-
-				assert.ok(cds._cursorWithContext);
+				var cds = getCursorDocumentSource([]);
+				assert.ok(cds._runner);
 			}
 		},
 
 		"#coalesce": {
 			"should be able to coalesce a limit into itself": function (){
-				var cds = getCursor(),
+				var cds = getCursorDocumentSource(),
 					lds = LimitDocumentSource.createFromJson(2);
 
 				assert.equal(cds.coalesce(lds) instanceof LimitDocumentSource, true);
@@ -46,7 +34,7 @@ module.exports = {
 			},
 
 			"should keep original limit if coalesced to a larger limit": function() {
-				var cds = getCursor();
+				var cds = getCursorDocumentSource();
 				cds.coalesce(LimitDocumentSource.createFromJson(2));
 				cds.coalesce(LimitDocumentSource.createFromJson(3));
 				assert.equal(cds.getLimit(), 2);
@@ -54,7 +42,7 @@ module.exports = {
 
 
 			"cursor only returns $limit number when coalesced": function(next) {
-				var cds = getCursor(),
+				var cds = getCursorDocumentSource(),
 					lds = LimitDocumentSource.createFromJson(2);
 
 
@@ -69,40 +57,29 @@ module.exports = {
 						});
 					},
 					function() {
-						return docs[i++] !== DocumentSource.EOF;
+						return docs[i++] !== null;
 					},
 					function(err) {
-						assert.deepEqual([1, 2, DocumentSource.EOF], docs);
+						if (err) throw err;
+						assert.deepEqual([1, 2, null], docs);
 						next();
 					}
 				);
 			},
 
 			"should leave non-limit alone": function () {
-				var cwc = new CursorDocumentSource.CursorWithContext();
-				cwc._cursor = new Cursor( [] );
-
 				var sds = new SkipDocumentSource(),
-					cds = new CursorDocumentSource(cwc);
+					cds = getCursorDocumentSource([]);
 
 				assert.equal(cds.coalesce(sds), false);
 			}
 		},
 
 		"#getNext": {
-			"should throw an error if no callback is given": function() {
-				var cwc = new CursorDocumentSource.CursorWithContext();
-				cwc._cursor = new Cursor( [1,2,3,4] );
-				var cds = new CursorDocumentSource(cwc);
-				assert.throws(cds.getNext.bind(cds));
-			},
-
 			"should return the current cursor value async": function(next){
 				var expected = JSON.stringify([1,2]);
-				var cwc = new CursorDocumentSource.CursorWithContext();
-				cwc._cursor = new Cursor( [1,2,3,4] );
 
-				var cds = new CursorDocumentSource(cwc);
+				var cds = getCursorDocumentSource([1,2,3,4]);
 				async.series([
 						cds.getNext.bind(cds),
 						cds.getNext.bind(cds),
@@ -111,18 +88,16 @@ module.exports = {
 						cds.getNext.bind(cds),
 					],
 					function(err,res) {
-						assert.deepEqual([1,2,3,4,DocumentSource.EOF], res);
+						assert.deepEqual([1,2,3,4,null], res);
 						next();
 					}
 				);
 			},
 			"should return values past the batch limit": function(next){
-				var cwc = new CursorDocumentSource.CursorWithContext(),
-					n = 0,
+				var n = 0,
 					arr = Array.apply(0, new Array(200)).map(function() { return n++; });
-				cwc._cursor = new Cursor( arr );
 
-				var cds = new CursorDocumentSource(cwc);
+				var cds = getCursorDocumentSource(arr);
 				async.each(arr,
 					function(a,next) {
 						cds.getNext(function(err,val) {
@@ -135,25 +110,24 @@ module.exports = {
 					}
 				);
 				cds.getNext(function(err,val) {
-					assert.equal(val, DocumentSource.EOF);
+					assert.equal(val, null);
 					next();
 				});
 			},
 		},
 		"#dispose": {
 			"should empty the current cursor": function(next){
-				var cwc = new CursorDocumentSource.CursorWithContext();
-				cwc._cursor = new Cursor( [1,2,3] );
-
-				var cds = new CursorDocumentSource(cwc);
+				var cds = getCursorDocumentSource();
 				async.series([
 						cds.getNext.bind(cds),
 						cds.getNext.bind(cds),
-						cds.getNext.bind(cds),
-						cds.getNext.bind(cds),
+						function(next){
+							cds.dispose();
+							return cds.getNext(next);
+						}
 					],
 					function(err,res) {
-						assert.deepEqual([1,2,3,DocumentSource.EOF], res);
+						assert.deepEqual([1,2,null], res);
 						next();
 					}
 				);
@@ -163,46 +137,10 @@ module.exports = {
 		"#setProjection": {
 
 			"should set a projection": function() {
-				var cwc = new CursorDocumentSource.CursorWithContext();
-				cwc._cursor = new Cursor( [1,2,3] );
-
-				var cds = new CursorDocumentSource(cwc);
+				var cds = getCursorDocumentSource();
 				cds.setProjection({a:1}, {a:true});
 				assert.deepEqual(cds._projection, {a:1});
 				assert.deepEqual(cds._dependencies, {a:true});
-			},
-
-			"should throw an error if projection is already set": function (){
-				var cwc = new CursorDocumentSource.CursorWithContext();
-				cwc._cursor = new Cursor( [1,2,3] );
-
-				var cds = new CursorDocumentSource(cwc);
-				cds.setProjection({a:1}, {});
-				assert.throws(function() {
-					cds.setProjection({a:1}, {});
-				});
-			},
-
-			"should project properly": function(next) {
-				var cwc = new CursorDocumentSource.CursorWithContext();
-				cwc._cursor = new Cursor( [{a:1},{a:2,b:3},{c:4,d:5}] );
-
-				var cds = new CursorDocumentSource(cwc);
-				cds.setProjection({a:1}, {a:true});
-				assert.deepEqual(cds._projection, {a:1});
-				assert.deepEqual(cds._dependencies, {a:true});
-
-				async.series([
-						cds.getNext.bind(cds),
-						cds.getNext.bind(cds),
-						cds.getNext.bind(cds),
-						cds.getNext.bind(cds),
-					],
-					function(err,res) {
-						assert.deepEqual([{a:1},{a:2},{},DocumentSource.EOF], res);
-						next();
-					}
-				);
 			}
 
 		}
@@ -211,4 +149,4 @@ module.exports = {
 
 };
 
-if (!module.parent)(new(require("mocha"))()).ui("exports").reporter("spec").addFile(__filename).run();
+if (!module.parent)(new(require("mocha"))()).ui("exports").reporter("spec").addFile(__filename).grep(process.env.MOCHA_GREP || '').run(process.exit);

test/lib/pipeline/documentSources/GeoNearDocumentSource.js

@@ -3,13 +3,17 @@ var assert = require("assert"),
 	DocumentSource = require("../../../../lib/pipeline/documentSources/DocumentSource"),
 	GeoNearDocumentSource = require("../../../../lib/pipeline/documentSources/GeoNearDocumentSource"),
 	CursorDocumentSource = require("../../../../lib/pipeline/documentSources/CursorDocumentSource"),
-	Cursor = require("../../../../lib/Cursor"),
+	ArrayRunner = require("../../../../lib/query/ArrayRunner"),
 	FieldPath = require("../../../../lib/pipeline/FieldPath");
 
 var createGeoNear = function(ctx) {
 	var ds = new GeoNearDocumentSource(ctx);
 	return ds;
 };
+var addSource = function addSource(ds, data) {
+	var cds = new CursorDocumentSource(null, new ArrayRunner(data), null);
+	ds.setSource(cds);
+};
 
 module.exports = {
 
@@ -55,14 +59,11 @@ module.exports = {
 		"#setSource()":{
 
 			"check that setting source of GeoNearDocumentSource throws error":function() {
-				var cwc = new CursorDocumentSource.CursorWithContext();
 				var input = [{}];
-				cwc._cursor = new Cursor( input );
-				var cds = new CursorDocumentSource(cwc);
 				var gnds = createGeoNear();
 
 				assert.throws(function(){
-					gnds.setSource(cds);
+					addSource(gnds, input);
 				});
 			}
 
@@ -95,4 +96,4 @@ module.exports = {
 	}
 };
 
-if (!module.parent)(new(require("mocha"))()).ui("exports").reporter("spec").addFile(__filename).grep(process.env.MOCHA_GREP || '').run(process.exit);
+if (!module.parent)(new(require("mocha"))()).ui("exports").reporter("spec").addFile(__filename).grep(process.env.MOCHA_GREP || '').run(process.exit);

test/lib/pipeline/documentSources/GroupDocumentSource.js

@@ -2,21 +2,13 @@
 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"),
-	async = require('async');
+	ArrayRunner = require("../../../../lib/query/ArrayRunner"),
+	async = require('async'),
+	utils = require("../expressions/utils"),
+	expressions = require("../../../../lib/pipeline/expressions");
 
 
-/**
- * Tests if the given spec is the same as what the DocumentSource resolves to as JSON.
- * MUST CALL WITH A DocumentSource AS THIS (e.g. checkJsonRepresentation.call(this, spec) where this is a DocumentSource and spec is the JSON used to create the source).
- **/
-var checkJsonRepresentation = function checkJsonRepresentation(self, spec) {
-	var rep = {};
-	self.serialize(rep, true);
-	assert.deepEqual(rep, {$group: spec});
-};
-
 /// An assertion for `ObjectExpression` instances based on Mongo's `ExpectedResultBase` class
 function assertExpectedResult(args) {
 	{// check for required args
@@ -29,21 +21,20 @@ function assertExpectedResult(args) {
 	// run implementation
 	if(args.expected && args.docs){
 		var gds = GroupDocumentSource.createFromJson(args.spec),
-			cwc = new CursorDocumentSource.CursorWithContext();
-		cwc._cursor = new Cursor( args.docs );
-		var next,
+			next,
 			results = [],
-			cds = new CursorDocumentSource(cwc);
+			cds = new CursorDocumentSource(null, new ArrayRunner(args.docs), null);
+			debugger;
 		gds.setSource(cds);
 		async.whilst(
 			function() {
-				next !== DocumentSource.EOF;
+				return next !== null;
 			},
 			function(done) {
 				gds.getNext(function(err, doc) {
 					if(err) return done(err);
 					next = doc;
-					if(next === DocumentSource.EOF) {
+					if(next === null) {
 						return done();
 					} else {
 						results.push(next);
@@ -52,8 +43,7 @@ function assertExpectedResult(args) {
 				});
 			},
 			function(err) {
-				assert.deepEqual(results, args.expected);
-				checkJsonRepresentation(gds, args.spec);
+				assert.equal(JSON.stringify(results), JSON.stringify(args.expected));
 				if(args.done) {
 					return args.done();
 				}
@@ -67,7 +57,6 @@ function assertExpectedResult(args) {
 		} else {
 			assert.doesNotThrow(function(){
 				var gds = GroupDocumentSource.createFromJson(args.spec);
-				checkJsonRepresentation(gds, args.spec);
 			});
 		}
 	}
@@ -97,7 +86,6 @@ 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});
 			},
 
@@ -116,13 +104,11 @@ 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});
 			},
 
@@ -133,55 +119,46 @@ module.exports = {
 
 			// $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/}});
+			"should not throw when _id is a regex": function advanceTest(){
+				assertExpectedResult({spec:{_id:/a/}, "throw":false});
 			},
 
 			// 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}}});
 			}
 
@@ -202,7 +179,6 @@ module.exports = {
 
 			// $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"}},
@@ -212,7 +188,6 @@ module.exports = {
 
 			// $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"},
@@ -222,7 +197,6 @@ module.exports = {
 
 			// $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}],
@@ -233,7 +207,6 @@ module.exports = {
 
 			// 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"}}},
@@ -243,7 +216,6 @@ module.exports = {
 
 			// 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"}},
@@ -253,7 +225,6 @@ module.exports = {
 
 			// 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"}},
@@ -274,7 +245,7 @@ module.exports = {
 			"should make one group with two values": function TwoValuesSingleKey() {
 				assertExpectedResult({
 					docs: [{a:1}, {a:2}],
-					spec: {_id:"$_id", a:{$push:"$a"}},
+					spec: {_id:0, a:{$push:"$a"}},
 					expected: [{_id:0, a:[1,2]}]
 				});
 			},

test/lib/pipeline/documentSources/LimitDocumentSource.js

@@ -1,8 +1,14 @@
 "use strict";
 var assert = require("assert"),
 	DocumentSource = require("../../../../lib/pipeline/documentSources/DocumentSource"),
-	LimitDocumentSource = require("../../../../lib/pipeline/documentSources/LimitDocumentSource");
+	LimitDocumentSource = require("../../../../lib/pipeline/documentSources/LimitDocumentSource"),
+	CursorDocumentSource = require("../../../../lib/pipeline/documentSources/CursorDocumentSource"),
+	ArrayRunner = require("../../../../lib/query/ArrayRunner");
 
+var addSource = function addSource(ds, data) {
+	var cds = new CursorDocumentSource(null, new ArrayRunner(data), null);
+	ds.setSource(cds);
+};
 
 module.exports = {
 
@@ -73,7 +79,7 @@ module.exports = {
 			"should return the current document source": function currSource(next){
 				var lds = new LimitDocumentSource({"$limit":[{"a":1},{"a":2}]});
 				lds.limit = 1;
-				lds.source = {getNext:function(cb){cb(null,{ item:1 });}};
+				addSource(lds, [{item:1}]);
 				lds.getNext(function(err,val) {
 					assert.deepEqual(val, { item:1 });
 					return next();
@@ -84,19 +90,10 @@ module.exports = {
 			"should return EOF for no sources remaining": function noMoar(next){
 				var lds = new LimitDocumentSource({"$match":[{"a":1},{"a":1}]});
 				lds.limit = 1;
-				lds.source = {
-					calls: 0,
-					getNext:function(cb) {
-						if (lds.source.calls)
-							return cb(null,DocumentSource.EOF);
-						lds.source.calls++;
-						return cb(null,{item:1});
-					},
-					dispose:function() { return true; }
-				};
+				addSource(lds, [{item:1}]);
 				lds.getNext(function(){});
 				lds.getNext(function(err,val) {
-					assert.strictEqual(val, DocumentSource.EOF);
+					assert.strictEqual(val, null);
 					return next();
 				});
 			},
@@ -104,18 +101,10 @@ module.exports = {
 			"should return EOF if we hit our limit": function noMoar(next){
 				var lds = new LimitDocumentSource();
 				lds.limit = 1;
-				lds.source = {
-					calls: 0,
-					getNext:function(cb) {
-						if (lds.source.calls)
-							return cb(null,DocumentSource.EOF);
-						return cb(null,{item:1});
-					},
-					dispose:function() { return true; }
-				};
+				addSource(lds, [{item:1},{item:2}]);
 				lds.getNext(function(){});
 				lds.getNext(function (err,val) {
-					assert.strictEqual(val, DocumentSource.EOF);
+					assert.strictEqual(val, null);
 					return next();
 				});
 			}

test/lib/pipeline/documentSources/MatchDocumentSource.js

@@ -2,13 +2,18 @@
 var assert = require("assert"),
 	async = require("async"),
 	DocumentSource = require("../../../../lib/pipeline/documentSources/DocumentSource"),
-	MatchDocumentSource = require("../../../../lib/pipeline/documentSources/MatchDocumentSource");
+	MatchDocumentSource = require("../../../../lib/pipeline/documentSources/MatchDocumentSource"),
+	CursorDocumentSource = require("../../../../lib/pipeline/documentSources/CursorDocumentSource"),
+	ArrayRunner = require("../../../../lib/query/ArrayRunner");
 
 var testRedactSafe = function testRedactSafe(input, safePortion) {
 	var match = MatchDocumentSource.createFromJson(input);
 	assert.deepEqual(match.redactSafePortion(), safePortion);
 };
-
+var addSource = function addSource(match, data) {
+	var cds = new CursorDocumentSource(null, new ArrayRunner(data), null);
+	match.setSource(cds);
+};
 
 module.exports = {
 
@@ -67,7 +72,7 @@ module.exports = {
 
 			"should return the current document source": function currSource(next){
 				var mds = new MatchDocumentSource({item: 1});
-				mds.source = {getNext:function(cb){cb(null,{ item:1 });}};
+				addSource(mds, [{ item:1 }]);
 				mds.getNext(function(err,val) {
 					assert.deepEqual(val, { item:1 });
 					next();
@@ -77,16 +82,9 @@ module.exports = {
 			"should return matched sources remaining": function (next){
 				var mds = new MatchDocumentSource({ item: {$lt: 5} }),
 					items = [ 1,2,3,4,5,6,7,8,9 ];
-				mds.source = {
-					calls: 0,
-					getNext:function(cb) {
-						if (this.calls >= items.length)
-							return cb(null,DocumentSource.EOF);
-						return cb(null,{item: items[this.calls++]});
-					},
-					dispose:function() { return true; }
-				};
+				addSource(mds, items.map(function(i){return {item:i};}));
 
+				debugger;
 				async.series([
 						mds.getNext.bind(mds),
 						mds.getNext.bind(mds),
@@ -95,7 +93,7 @@ module.exports = {
 						mds.getNext.bind(mds),
 					],
 					function(err,res) {
-						assert.deepEqual([{item:1},{item:2},{item:3},{item:4},DocumentSource.EOF], res);
+						assert.deepEqual([{item:1},{item:2},{item:3},{item:4},null], res);
 						next();
 					}
 				);
@@ -104,15 +102,7 @@ module.exports = {
 			"should not return matched out documents for sources remaining": function (next){
 				var mds = new MatchDocumentSource({ item: {$gt: 5} }),
 					items = [ 1,2,3,4,5,6,7,8,9 ];
-				mds.source = {
-					calls: 0,
-					getNext:function(cb) {
-						if (this.calls >= items.length)
-							return cb(null,DocumentSource.EOF);
-						return cb(null,{item: items[this.calls++]});
-					},
-					dispose:function() { return true; }
-				};
+				addSource(mds, items.map(function(i){return {item:i};}));
 
 				async.series([
 						mds.getNext.bind(mds),
@@ -122,7 +112,7 @@ module.exports = {
 						mds.getNext.bind(mds),
 					],
 					function(err,res) {
-						assert.deepEqual([{item:6},{item:7},{item:8},{item:9},DocumentSource.EOF], res);
+						assert.deepEqual([{item:6},{item:7},{item:8},{item:9},null], res);
 						next();
 					}
 				);
@@ -131,21 +121,13 @@ module.exports = {
 			"should return EOF for no sources remaining": function (next){
 				var mds = new MatchDocumentSource({ item: {$gt: 5} }),
 					items = [ ];
-				mds.source = {
-					calls: 0,
-					getNext:function(cb) {
-						if (this.calls >= items.length)
-							return cb(null,DocumentSource.EOF);
-						return cb(null,{item: items[this.calls++]});
-					},
-					dispose:function() { return true; }
-				};
+				addSource(mds, items.map(function(i){return {item:i};}));
 
 				async.series([
 						mds.getNext.bind(mds),
 					],
 					function(err,res) {
-						assert.deepEqual([DocumentSource.EOF], res);
+						assert.deepEqual([null], res);
 						next();
 					}
 				);

test/lib/pipeline/documentSources/OutDocumentSource.js

@@ -4,12 +4,16 @@ var assert = require("assert"),
 	DocumentSource = require("../../../../lib/pipeline/documentSources/DocumentSource"),
 	OutDocumentSource = require("../../../../lib/pipeline/documentSources/OutDocumentSource"),
 	CursorDocumentSource = require("../../../../lib/pipeline/documentSources/CursorDocumentSource"),
-	Cursor = require("../../../../lib/Cursor");
+	ArrayRunner = require("../../../../lib/query/ArrayRunner");
 
 var createOut = function(ctx) {
 	var ds = new OutDocumentSource(ctx);
 	return ds;
 };
+var addSource = function addSource(ds, data) {
+	var cds = new CursorDocumentSource(null, new ArrayRunner(data), null);
+	ds.setSource(cds);
+};
 
 module.exports = {
 
@@ -43,12 +47,9 @@ module.exports = {
 
 			"should act as passthrough (for now)": function(next) {
 				var ods = OutDocumentSource.createFromJson("test"),
-					cwc = new CursorDocumentSource.CursorWithContext(),
 					l = [{_id:0,a:[{b:1},{b:2}]}, {_id:1,a:[{b:1},{b:1}]} ];
 
-				cwc._cursor = new Cursor( l );
-				var cds = new CursorDocumentSource(cwc);
-				ods.setSource(cds);
+				addSource(ods, l);
 
 				var docs = [], i = 0;
 				async.doWhilst(
@@ -59,10 +60,10 @@ module.exports = {
 						});
 					},
 					function() {
-						return docs[i++] !== DocumentSource.EOF;
+						return docs[i++] !== null;
 					},
 					function(err) {
-						assert.deepEqual([{_id:0,a:[{b:1},{b:2}]}, {_id:1,a:[{b:1},{b:1}]}, DocumentSource.EOF], docs);
+						assert.deepEqual([{_id:0,a:[{b:1},{b:2}]}, {_id:1,a:[{b:1},{b:1}]}, null], docs);
 						next();
 					}
 				);
@@ -83,13 +84,10 @@ module.exports = {
 		"#serialize()":{
 
 			"serialize":function() {
-				var cwc = new CursorDocumentSource.CursorWithContext();
 				var input = [{_id: 0, a: 1}, {_id: 1, a: 2}];
-				cwc._cursor = new Cursor( input );
-				var cds = new CursorDocumentSource(cwc);
 				var title = "CognitiveScientists";
 				var ods = OutDocumentSource.createFromJson(title);
-				ods.setSource(cds);
+				addSource(ods, input);
 				var srcNm = ods.getSourceName();
 				var serialize = {};
 				serialize[srcNm] = title;

test/lib/pipeline/documentSources/ProjectDocumentSource.js

@@ -1,10 +1,11 @@
 "use strict";
 var assert = require("assert"),
 	async = require("async"),
+	DepsTracker = require("../../../../lib/pipeline/DepsTracker"),
 	DocumentSource = require("../../../../lib/pipeline/documentSources/DocumentSource"),
 	ProjectDocumentSource = require("../../../../lib/pipeline/documentSources/ProjectDocumentSource"),
 	CursorDocumentSource = require("../../../../lib/pipeline/documentSources/CursorDocumentSource"),
-	Cursor = require("../../../../lib/Cursor"),
+	ArrayRunner = require("../../../../lib/query/ArrayRunner"),
 	TestBase = require("./TestBase"),
 	And = require("../../../../lib/pipeline/expressions/AndExpression");
 
@@ -37,268 +38,236 @@ var createProject = function createProject(projection) {
 //TESTS
 module.exports = {
 
-	"ProjectDocumentSource": {
+	"constructor()": {
 
-		"mongo tests": {
-			"Inclusion": function() {
-				var test = new TestBase();
-				//client.insert( ns, fromjson( "{_id:0,a:1,b:1,c:{d:1}}" ) );
-				test.createSource();
-				test.createProject();
-			}
+		"should not throw Error when constructing without args": function testConstructor() {
+			assert.doesNotThrow(function() {
+				new ProjectDocumentSource();
+			});
 		},
 
-		"constructor()": {
-
-			"should not throw Error when constructing without args": function testConstructor() {
-				assert.doesNotThrow(function() {
-					new ProjectDocumentSource();
-				});
-			},
-
-			"should throw Error when constructing with more than 1 arg": function testConstructor() {
-				assert.throws(function() {
-					new ProjectDocumentSource("a", "b", "c");
-				});
-			}
+		"should throw Error when constructing with more than 1 arg": function testConstructor() {
+			assert.throws(function() {
+				new ProjectDocumentSource("a", "b", "c");
+			});
+		}
 
-		},
+	},
 
-		"#getSourceName()": {
+	"#getSourceName()": {
 
-			"should return the correct source name; $project": function testSourceName() {
-				var pds = new ProjectDocumentSource();
-				assert.strictEqual(pds.getSourceName(), "$project");
-			}
+		"should return the correct source name; $project": function testSourceName() {
+			var pds = new ProjectDocumentSource();
+			assert.strictEqual(pds.getSourceName(), "$project");
+		}
 
+	},
+
+	"#getNext()": {
+
+		"should return EOF": function testEOF(next) {
+			var pds = createProject({});
+			pds.setSource({
+				getNext: function getNext(cb) {
+					return cb(null, null);
+				}
+			});
+			pds.getNext(function(err, doc) {
+				assert.equal(null, doc);
+				next();
+			});
 		},
 
-		"#getNext()": {
-
-			"should return EOF": function testEOF(next) {
-				var pds = createProject({});
-				pds.setSource({
-					getNext: function getNext(cb) {
-						return cb(null, DocumentSource.EOF);
-					}
-				});
-				pds.getNext(function(err, doc) {
-					assert.equal(DocumentSource.EOF, doc);
-					next();
-				});
-			},
-
-			"iterator state accessors consistently report the source is exhausted": function assertExhausted() {
-				var cwc = new CursorDocumentSource.CursorWithContext();
-				var input = [{}];
-				cwc._cursor = new Cursor( input );
-				var cds = new CursorDocumentSource(cwc);
-				var pds = createProject();
-				pds.setSource(cds);
-				pds.getNext(function(err, actual) {
-					pds.getNext(function(err, actual1) {
-						assert.equal(DocumentSource.EOF, actual1);
-						pds.getNext(function(err, actual2) {
-							assert.equal(DocumentSource.EOF, actual2);
-							pds.getNext(function(err, actual3) {
-								assert.equal(DocumentSource.EOF, actual3);
-							});
+		"iterator state accessors consistently report the source is exhausted": function assertExhausted() {
+			var input = [{}];
+			var cds = new CursorDocumentSource(null, new ArrayRunner(input), null);
+			var pds = createProject();
+			pds.setSource(cds);
+			pds.getNext(function(err, actual) {
+				pds.getNext(function(err, actual1) {
+					assert.equal(null, actual1);
+					pds.getNext(function(err, actual2) {
+						assert.equal(null, actual2);
+						pds.getNext(function(err, actual3) {
+							assert.equal(null, actual3);
 						});
 					});
 				});
-			},
-
-			"callback is required": function requireCallback() {
-				var pds = createProject();
-				assert.throws(pds.getNext.bind(pds));
-			},
-
-			"should not return EOF when a document is still in cursor": function testNotEOFTrueIfDocPresent() {
-				var cwc = new CursorDocumentSource.CursorWithContext();
-				var input = [{_id: 0, a: 1}, {_id: 1, a: 2}];
-					cwc._cursor = new Cursor( input );
-				var cds = new CursorDocumentSource(cwc);
-				var pds = createProject();
-				pds.setSource(cds);
-				pds.getNext(function(err,actual) {
-					// first go round
-					assert.notEqual(actual, DocumentSource.EOF);
-				});
-			},
+			});
+		},
+
+		"callback is required": function requireCallback() {
+			var pds = createProject();
+			assert.throws(pds.getNext.bind(pds));
+		},
 
-			"can retrieve second document from source": function testAdvanceFirst() {
-				var cwc = new CursorDocumentSource.CursorWithContext();
-				var input = [{_id: 0, a: 1}, {_id: 1, a: 2}];
-				cwc._cursor = new Cursor( input );
-				var cds = new CursorDocumentSource(cwc);
-				var pds = createProject();
-				pds.setSource(cds);
+		"should not return EOF when a document is still in cursor": function testNotEOFTrueIfDocPresent() {
+			var input = [{_id: 0, a: 1}, {_id: 1, a: 2}];
+			var cds = new CursorDocumentSource(null, new ArrayRunner(input), null);
+			var pds = createProject();
+			pds.setSource(cds);
+			pds.getNext(function(err,actual) {
+				// first go round
+				assert.notEqual(actual, null);
+			});
+		},
 
-				pds.getNext(function(err,val) {
-					// eh, ignored
-					pds.getNext(function(err,val) {
-						assert.equal(2, val.a);
-					});
-				});
-			},
-
-			"should get the first document out of a cursor": function getCurrentCalledFirst() {
-				var cwc = new CursorDocumentSource.CursorWithContext();
-				var input = [{_id: 0, a: 1}];
-				cwc._cursor = new Cursor( input );
-				var cds = new CursorDocumentSource(cwc);
-				var pds = createProject();
-				pds.setSource(cds);
-				pds.getNext(function(err, actual) {
-					assert.equal(1, actual.a);
-				});
-			},
-
-			"The a and c.d fields are included but the b field is not": function testFullProject1(next) {
-				var cwc = new CursorDocumentSource.CursorWithContext();
-				var input = [{
-					_id:1,
-					a: 1,
-					b: 1,
-					c: {
-						d: 1
-					}
-				}];
-				cwc._cursor = new Cursor(input);
-				var cds = new CursorDocumentSource(cwc);
-				var pds = createProject({
-						_id: 0,
-						a: true,
-						c: {
-							d: true
-						}
-					}),
-					expected = {a:1, c:{ d: 1 }};
-				pds.setSource(cds);
+		"can retrieve second document from source": function testAdvanceFirst() {
+			var input = [{_id: 0, a: 1}, {_id: 1, a: 2}];
+			var cds = new CursorDocumentSource(null, new ArrayRunner(input), null);
+			var pds = createProject();
+			pds.setSource(cds);
 
+			pds.getNext(function(err,val) {
+				// eh, ignored
 				pds.getNext(function(err,val) {
-					assert.deepEqual(expected, val);
-					next();
+					assert.equal(2, val.a);
 				});
-			},
-
-			"Two documents": function testTwoDocumentsProject(next) {
-				var cwc = new CursorDocumentSource.CursorWithContext();
-				var input = [{
-					a: 1,
-					b: 2
-				}, {
-					a: 3,
-					b: 4
-				}],
-				expected = [
-					{a:1},
-					{a:3},
-					DocumentSource.EOF
-				];
-				cwc._cursor = new Cursor(input);
-				var cds = new CursorDocumentSource(cwc);
-				var pds = createProject({
+			});
+		},
+
+		"should get the first document out of a cursor": function getCurrentCalledFirst() {
+			var input = [{_id: 0, a: 1}];
+			var cds = new CursorDocumentSource(null, new ArrayRunner(input), null);
+			var pds = createProject();
+			pds.setSource(cds);
+			pds.getNext(function(err, actual) {
+				assert.equal(1, actual.a);
+			});
+		},
+
+		"The a and c.d fields are included but the b field is not": function testFullProject1(next) {
+			var input = [{
+				_id:0,
+				a: 1,
+				b: 1,
+				c: {
+					d: 1
+				}
+			}];
+			var cds = new CursorDocumentSource(null, new ArrayRunner(input), null);
+			var pds = createProject({
 					a: true,
 					c: {
 						d: true
 					}
-				});
-				pds.setSource(cds);
-
-				async.series([
-						pds.getNext.bind(pds),
-						pds.getNext.bind(pds),
-						pds.getNext.bind(pds),
-					],
-					function(err,res) {
-						assert.deepEqual(expected, res);
-						next();
-					}
-				);
-			}
+				}),
+				expected = {_id: 0, a:1, c:{ d: 1 }};
+			pds.setSource(cds);
+
+			pds.getNext(function(err,val) {
+				assert.deepEqual(expected, val);
+				next();
+			});
 		},
 
-		"#optimize()": {
+		"Two documents": function testTwoDocumentsProject(next) {
+			var input = [{
+				a: 1,
+				b: 2
+			}, {
+				a: 3,
+				b: 4
+			}],
+			expected = [
+				{a:1},
+				{a:3},
+				null
+			];
+			var cds = new CursorDocumentSource(null, new ArrayRunner(input), null);
+			var pds = createProject({
+				a: true,
+				c: {
+					d: true
+				}
+			});
+			pds.setSource(cds);
+
+			async.series([
+					pds.getNext.bind(pds),
+					pds.getNext.bind(pds),
+					pds.getNext.bind(pds),
+				],
+				function(err,res) {
+					assert.deepEqual(expected, res);
+					next();
+				}
+			);
+		}
+	},
 
-			"Optimize the projection": function optimizeProject() {
-				var pds = createProject({
-					a: {
-						$and: [true]
-					}
-				});
-				pds.optimize();
-				checkJsonRepresentation(pds, {
-					$project: {
-						a: {
-							$const: true
-						}
-					}
-				});
-			}
+	"#optimize()": {
 
-		},
+		"Optimize the projection": function optimizeProject() {
+			var pds = createProject({
+				a: {
+					$and: [{$const:true}]
+				}
+			});
 
-		"#createFromJson()": {
+			pds.optimize();
+			checkJsonRepresentation(pds, {$project:{a:{$const:true}}});
+		}
 
-			"should error if called with non-object": function testNonObjectPassed() {
-				//String as arg
-				assert.throws(function() {
-					var pds = createProject("not an object");
-				});
-				//Date as arg
-				assert.throws(function() {
-					var pds = createProject(new Date());
-				});
-				//Array as arg
-				assert.throws(function() {
-					var pds = createProject([]);
-				});
-				//Empty args
-				assert.throws(function() {
-					var pds = ProjectDocumentSource.createFromJson();
-				});
-				//Top level operator
-				assert.throws(function() {
-					var pds = createProject({
-						$add: []
-					});
+	},
+
+	"#createFromJson()": {
+
+		"should error if called with non-object": function testNonObjectPassed() {
+			//String as arg
+			assert.throws(function() {
+				var pds = createProject("not an object");
+			});
+			//Date as arg
+			assert.throws(function() {
+				var pds = createProject(new Date());
+			});
+			//Array as arg
+			assert.throws(function() {
+				var pds = createProject([]);
+			});
+			//Empty args
+			assert.throws(function() {
+				var pds = ProjectDocumentSource.createFromJson();
+			});
+			//Top level operator
+			assert.throws(function() {
+				var pds = createProject({
+					$add: []
 				});
-				//Invalid spec
-				assert.throws(function() {
-					var pds = createProject({
-						a: {
-							$invalidOperator: 1
-						}
-					});
+			});
+			//Invalid spec
+			assert.throws(function() {
+				var pds = createProject({
+					a: {
+						$invalidOperator: 1
+					}
 				});
+			});
 
-			}
-
-		},
-
-		"#getDependencies()": {
-
-			"should properly detect dependencies in project": function testGetDependencies() {
-				var cwc = new CursorDocumentSource.CursorWithContext();
-				var input = {
-					a: true,
-					x: '$b',
-					y: {
-						$and: ['$c', '$d']
-					}
-				};
-				var pds = createProject(input);
-				var dependencies = {};
-				assert.equal(DocumentSource.GetDepsReturn.EXHAUSTIVE, pds.getDependencies(dependencies));
-				assert.equal(5, Object.keys(dependencies).length);
-				assert.ok(dependencies._id);
-				assert.ok(dependencies.a);
-				assert.ok(dependencies.b);
-				assert.ok(dependencies.c);
-				assert.ok(dependencies.d);
-			}
+		}
 
+	},
+
+	"#getDependencies()": {
+
+		"should properly detect dependencies in project": function testGetDependencies() {
+			var input = {
+				a: true,
+				x: '$b',
+				y: {
+					$and: ['$c', '$d']
+				}
+			};
+			var pds = createProject(input);
+			var dependencies = new DepsTracker();
+			assert.equal(DocumentSource.GetDepsReturn.EXHAUSTIVE, pds.getDependencies(dependencies));
+			assert.equal(5, Object.keys(dependencies.fields).length);
+			assert.ok(dependencies.fields._id);
+			assert.ok(dependencies.fields.a);
+			assert.ok(dependencies.fields.b);
+			assert.ok(dependencies.fields.c);
+			assert.ok(dependencies.fields.d);
 		}
 
 	}

test/lib/pipeline/documentSources/RedactDocumentSource.js

@@ -4,7 +4,7 @@ var assert = require("assert"),
 	DocumentSource = require("../../../../lib/pipeline/documentSources/DocumentSource"),
 	RedactDocumentSource = require("../../../../lib/pipeline/documentSources/RedactDocumentSource"),
 	CursorDocumentSource = require("../../../../lib/pipeline/documentSources/CursorDocumentSource"),
-	Cursor = require("../../../../lib/Cursor"),
+	ArrayRunner = require("../../../../lib/query/ArrayRunner"),
 	Expressions = require("../../../../lib/pipeline/expressions");
 
 var exampleRedact = {$cond:{
@@ -15,9 +15,7 @@ var exampleRedact = {$cond:{
 
 var createCursorDocumentSource = function createCursorDocumentSource (input) {
 	if (!input || input.constructor !== Array) throw new Error('invalid');
-	var cwc = new CursorDocumentSource.CursorWithContext();
-	cwc._cursor = new Cursor(input);
-	return new CursorDocumentSource(cwc);
+	return new CursorDocumentSource(null, new ArrayRunner(input), null);
 };
 
 var createRedactDocumentSource = function createRedactDocumentSource (src, expression) {
@@ -55,29 +53,27 @@ module.exports = {
 				var rds = RedactDocumentSource.createFromJson(exampleRedact);
 				rds.setSource({
 					getNext: function getNext(cb) {
-						return cb(null, DocumentSource.EOF);
+						return cb(null, null);
 					}
 				});
 				rds.getNext(function(err, doc) {
-					assert.equal(DocumentSource.EOF, doc);
+					assert.equal(null, doc);
 					next();
 				});
 			},
 
 			"iterator state accessors consistently report the source is exhausted": function assertExhausted() {
-				var cwc = new CursorDocumentSource.CursorWithContext();
 				var input = [{}];
-				cwc._cursor = new Cursor( input );
-				var cds = new CursorDocumentSource(cwc);
+				var cds = createCursorDocumentSource(input);
 				var rds = RedactDocumentSource.createFromJson(exampleRedact);
 				rds.setSource(cds);
 				rds.getNext(function(err, actual) {
 					rds.getNext(function(err, actual1) {
-						assert.equal(DocumentSource.EOF, actual1);
+						assert.equal(null, actual1);
 						rds.getNext(function(err, actual2) {
-							assert.equal(DocumentSource.EOF, actual2);
+							assert.equal(null, actual2);
 							rds.getNext(function(err, actual3) {
-								assert.equal(DocumentSource.EOF, actual3);
+								assert.equal(null, actual3);
 							});
 						});
 					});
@@ -235,4 +231,4 @@ module.exports = {
 
 };
 
-if (!module.parent)(new(require("mocha"))()).ui("exports").reporter("spec").addFile(__filename).grep(process.env.MOCHA_GREP || '').run(process.exit);
+if (!module.parent)(new(require("mocha"))()).ui("exports").reporter("spec").addFile(__filename).grep(process.env.MOCHA_GREP || '').run(process.exit);

test/lib/pipeline/documentSources/SkipDocumentSource.js

@@ -1,10 +1,15 @@
 "use strict";
 var assert = require("assert"),
 	async = require("async"),
-	Cursor = require("../../../../lib/Cursor"),
 	DocumentSource = require("../../../../lib/pipeline/documentSources/DocumentSource"),
+	SkipDocumentSource = require("../../../../lib/pipeline/documentSources/SkipDocumentSource"),
 	CursorDocumentSource = require("../../../../lib/pipeline/documentSources/CursorDocumentSource"),
-	SkipDocumentSource = require("../../../../lib/pipeline/documentSources/SkipDocumentSource");
+	ArrayRunner = require("../../../../lib/query/ArrayRunner");
+
+var addSource = function addSource(ds, data) {
+	var cds = new CursorDocumentSource(null, new ArrayRunner(data), null);
+	ds.setSource(cds);
+};
 
 
 module.exports = {
@@ -84,19 +89,15 @@ module.exports = {
 
 				var expected = [
 					{val:4},
-					DocumentSource.EOF
+					null
 				];
-
-				var cwc = new CursorDocumentSource.CursorWithContext();
 				var input = [
 					{val:1},
 					{val:2},
 					{val:3},
 					{val:4},
 				];
-				cwc._cursor = new Cursor( input );
-				var cds = new CursorDocumentSource(cwc);
-				sds.setSource(cds);
+				addSource(sds, input);
 
 				async.series([
 						sds.getNext.bind(sds),
@@ -108,20 +109,17 @@ module.exports = {
 					}
 				);
 				sds.getNext(function(err, actual) {
-					assert.equal(actual, DocumentSource.EOF);
+					assert.equal(actual, null);
 				});
 			},
 			"should return documents if skip count is not hit and there are more documents": function hitSkip(next){
 				var sds = SkipDocumentSource.createFromJson(1);
 
-				var cwc = new CursorDocumentSource.CursorWithContext();
 				var input = [{val:1},{val:2},{val:3}];
-				cwc._cursor = new Cursor( input );
-				var cds = new CursorDocumentSource(cwc);
-				sds.setSource(cds);
+				addSource(sds, input);
 
 				sds.getNext(function(err,actual) {
-					assert.notEqual(actual, DocumentSource.EOF);
+					assert.notEqual(actual, null);
 					assert.deepEqual(actual, {val:2});
 					next();
 				});
@@ -130,11 +128,8 @@ module.exports = {
 			"should return the current document source": function currSource(){
 				var sds = SkipDocumentSource.createFromJson(1);
 
-				var cwc = new CursorDocumentSource.CursorWithContext();
 				var input = [{val:1},{val:2},{val:3}];
-				cwc._cursor = new Cursor( input );
-				var cds = new CursorDocumentSource(cwc);
-				sds.setSource(cds);
+				addSource(sds, input);
 
 				sds.getNext(function(err, actual) {
 					assert.deepEqual(actual, { val:2 });
@@ -147,17 +142,11 @@ module.exports = {
 
 				var expected = [
 					{item:4},
-					DocumentSource.EOF
+					null
 				];
-
-				var i = 1;
-				sds.source = {
-					getNext:function(cb){
-						if (i>=5)
-							return cb(null,DocumentSource.EOF);
-						return cb(null, { item:i++ });
-					}
-				};
+				
+				var input = [{item:1},{item:2},{item:3},{item:4}];
+				addSource(sds, input);
 
 				async.series([
 						sds.getNext.bind(sds),

test/lib/pipeline/documentSources/UnwindDocumentSource.js

@@ -4,7 +4,7 @@ var assert = require("assert"),
 	DocumentSource = require("../../../../lib/pipeline/documentSources/DocumentSource"),
 	UnwindDocumentSource = require("../../../../lib/pipeline/documentSources/UnwindDocumentSource"),
 	CursorDocumentSource = require("../../../../lib/pipeline/documentSources/CursorDocumentSource"),
-	Cursor = require("../../../../lib/Cursor");
+	ArrayRunner = require("../../../../lib/query/ArrayRunner");
 
 
 //HELPERS
@@ -35,10 +35,7 @@ var createUnwind = function createUnwind(unwind) {
 };
 
 var addSource = function addSource(unwind, data) {
-	var cwc = new CursorDocumentSource.CursorWithContext();
-	cwc._cursor = new Cursor(data);
-	var cds = new CursorDocumentSource(cwc);
-	var pds = new UnwindDocumentSource();
+	var cds = new CursorDocumentSource(null, new ArrayRunner(data), null);
 	unwind.setSource(cds);
 };
 
@@ -53,7 +50,7 @@ var checkResults = function checkResults(data, expectedResults, path, next) {
 
 	expectedResults = expectedResults || [];
 
-	expectedResults.push(DocumentSource.EOF);
+	expectedResults.push(null);
 
 	//Load the results from the DocumentSourceUnwind
 	var docs = [], i = 0;
@@ -65,7 +62,7 @@ var checkResults = function checkResults(data, expectedResults, path, next) {
 			});
 		},
 		function() {
-			return docs[i++] !== DocumentSource.EOF;
+			return docs[i++] !== null;
 		},
 		function(err) {
 			assert.deepEqual(expectedResults, docs);
@@ -111,7 +108,7 @@ module.exports = {
 				var pds = createUnwind();
 				addSource(pds, []);
 				pds.getNext(function(err,doc) {
-					assert.strictEqual(doc, DocumentSource.EOF);
+					assert.strictEqual(doc, null);
 					next();
 				});
 			},
@@ -120,7 +117,7 @@ module.exports = {
 				var pds = createUnwind();
 				addSource(pds, [{_id:0, a:[1]}]);
 				pds.getNext(function(err,doc) {
-					assert.notStrictEqual(doc, DocumentSource.EOF);
+					assert.notStrictEqual(doc, null);
 					next();
 				});
 			},
@@ -129,7 +126,7 @@ module.exports = {
 				var pds = createUnwind();
 				addSource(pds, [{_id:0, a:[1,2]}]);
 				pds.getNext(function(err,doc) {
-					assert.notStrictEqual(doc, DocumentSource.EOF);
+					assert.notStrictEqual(doc, null);
 					assert.strictEqual(doc.a, 1);
 					pds.getNext(function(err,doc) {
 						assert.strictEqual(doc.a, 2);
@@ -151,10 +148,10 @@ module.exports = {
 						});
 					},
 					function() {
-						return docs[i++] !== DocumentSource.EOF;
+						return docs[i++] !== null;
 					},
 					function(err) {
-						assert.deepEqual([{_id:0, a:1},{_id:0, a:2},DocumentSource.EOF], docs);
+						assert.deepEqual([{_id:0, a:1},{_id:0, a:2},null], docs);
 						next();
 					}
 				);

test/lib/pipeline/expressions/AddExpression_test.js

@@ -49,8 +49,8 @@ var TestBase = function TestBase(overrides) {
 		}, base = ExpectedResultBase, proto = klass.prototype = Object.create(base.prototype);
 		proto.run = function() {
 			base.prototype.run.call(this);
-            // Now add the operands in the reverse direction.
-            this._reverse = true;
+			// Now add the operands in the reverse direction.
+			this._reverse = true;
 			base.prototype.run.call(this);
 		};
 		proto.populateOperands = function(expr) {
@@ -122,7 +122,7 @@ exports.AddExpression = {
 		"w/ 1 operand": {
 
 			"should pass through a single int": function testInt() {
-        		/** Single int argument. */
+				/** Single int argument. */
 				new SingleOperandBase({
 					operand: 1,
 				}).run();

test/lib/pipeline/expressions/AllElementsTrueExpression_test.js

@@ -40,7 +40,7 @@ var ExpectedResultBase = (function() {
 			var asserters = spec.error,
 				n = asserters.length;
 			for (var i = 0; i < n; ++i) {
-                // var obj2 = {<asserters[i]>: args}; //NOTE: DEVIATION FROM MONGO: see parseExpression below
+				// var obj2 = {<asserters[i]>: args}; //NOTE: DEVIATION FROM MONGO: see parseExpression below
 				var idGenerator2 = new VariablesIdGenerator(),
 					vps2 = new VariablesParseState(idGenerator2);
 				assert.throws(function() {
@@ -82,7 +82,7 @@ exports.AllElementsTrueExpression = {
 	"#evaluate()": {
 
 		"should return false if just false": function JustFalse() {
-            new ExpectedResultBase({
+			new ExpectedResultBase({
 				getSpec: {
 					input: [[false]],
 					expected: {

test/lib/pipeline/expressions/CoerceToBoolExpression_test.js

@@ -64,7 +64,7 @@ exports.CoerceToBoolExpression = {
 		"should be able to output in to JSON Object": function testAddToBsonObj() {
 			/** Output to BSONObj. */
 			var expr = CoerceToBoolExpression.create(FieldPathExpression.create("foo"));
-            // serialized as $and because CoerceToBool isn't an ExpressionNary
+			// serialized as $and because CoerceToBool isn't an ExpressionNary
 			assert.deepEqual({field:{$and:["$foo"]}}, {field:expr.serialize(false)});
 		},
 

test/lib/pipeline/expressions/CompareExpression_test.js

@@ -329,7 +329,7 @@ exports.CompareExpression = {
 			}).run();
 		},
 
-        /** Incompatible types can be compared. */
+		/** Incompatible types can be compared. */
 		"IncompatibleTypes": function IncompatibleTypes() {
 			var specElement = {$ne:["a",1]},
 				idGenerator = new VariablesIdGenerator(),

test/lib/pipeline/expressions/utils.js

@@ -31,10 +31,10 @@ var utils = module.exports = {
 
 	//SKIPPED: toJson
 
-    /**
-     * Convert Expression to BSON.
-     * @method expressionToJson
-     */
+	/**
+	 * Convert Expression to BSON.
+	 * @method expressionToJson
+	 */
 	expressionToJson: function expressionToJson(expr) {
 		return expr.serialize(false);
 	},

test/lib/query/ArrayRunner.js

@@ -0,0 +1,88 @@
+"use strict";
+var assert = require("assert"),
+	Runner = require("../../../lib/query/Runner"),
+	ArrayRunner = require("../../../lib/query/ArrayRunner");
+
+module.exports = {
+
+	"ArrayRunner": {
+		"#constructor": {
+			"should accept an array of data": function(){
+				assert.doesNotThrow(function(){
+					var ar = new ArrayRunner([1,2,3]);
+				});
+			},
+			"should fail if not given an array": function(){
+				assert.throws(function(){
+					var ar = new ArrayRunner();
+				});
+				assert.throws(function(){
+					var ar = new ArrayRunner(123);
+				});
+			}
+		},
+		"#getNext": {
+			"should return the next item in the array": function(done){
+				var ar = new ArrayRunner([1,2,3]);
+				
+				ar.getNext(function(err, out, state){
+					assert.strictEqual(state, Runner.RunnerState.RUNNER_ADVANCED);
+					assert.strictEqual(out, 1);
+					ar.getNext(function(err, out, state){
+						assert.strictEqual(state, Runner.RunnerState.RUNNER_ADVANCED);
+						assert.strictEqual(out, 2);
+						ar.getNext(function(err, out, state){
+							assert.strictEqual(state, Runner.RunnerState.RUNNER_ADVANCED);
+							assert.strictEqual(out, 3);
+							done();
+						});
+					});
+				});
+			},
+			"should return EOF if there is nothing left in the array": function(done){
+				var ar = new ArrayRunner([1]);
+				
+				ar.getNext(function(err, out, state){
+					assert.strictEqual(state, Runner.RunnerState.RUNNER_ADVANCED);
+					assert.strictEqual(out, 1);
+					ar.getNext(function(err, out, state){
+						assert.strictEqual(state, Runner.RunnerState.RUNNER_EOF);
+						assert.strictEqual(out, undefined);
+						done();
+					});
+				});
+			}
+		},
+		"#getInfo": {
+			"should return nothing if explain flag is not set": function(){
+				var ar = new ArrayRunner([1,2,3]);
+				assert.strictEqual(ar.getInfo(), undefined);
+			},
+			"should return information about the runner if explain flag is set": function(){
+				var ar = new ArrayRunner([1,2,3]);
+				assert.deepEqual(ar.getInfo(true), {
+					"type":"ArrayRunner",
+					"nDocs":3,
+					"position":0,
+					"state": Runner.RunnerState.RUNNER_ADVANCED
+				});
+			}
+		},
+		"#reset": {
+			"should clear out the runner": function(){
+				var ar = new ArrayRunner([1,2,3]);
+				ar.reset();
+				
+				assert.deepEqual(ar.getInfo(true), {
+					"type":"ArrayRunner",
+					"nDocs":0,
+					"position":0,
+					"state": Runner.RunnerState.RUNNER_DEAD
+				});				
+			}
+		}
+	}
+
+};
+
+if (!module.parent)(new(require("mocha"))()).ui("exports").reporter("spec").addFile(__filename).run();