refs #5124: Updated port of PipelineD js

lib/pipeline/PipelineD.js

@@ -20,6 +20,7 @@ var DocumentSource = require('./documentSources/DocumentSource'),
  * Create a Cursor wrapped in a DocumentSourceCursor, which is suitable to be the first source for a pipeline to begin with.
  * This source will feed the execution of the pipeline.
  *
+ * //NOTE: Not doing anything here, as we don't use any of these cursor source features
  * //NOTE: DEVIATION FROM THE MONGO: We don't have special optimized cursors; You could support something similar by overriding `Pipeline#run` to call `DocumentSource#coalesce` on the `inputSource` if you really need it.
  *
  * This method looks for early pipeline stages that can be folded into
@@ -32,48 +33,53 @@ var DocumentSource = require('./documentSources/DocumentSource'),
  * @param ctx       {Object}    Context for expressions
  * @returns	{CursorDocumentSource}	the cursor that was created
 **/
-klass.prepareCursorSource = function prepareCursorSource(pipeline, /*dbName,*/ expCtx){
+klass.prepareCursorSource = function prepareCursorSource(pipeline, expCtx){
 
-	var sources = pipeline.sourceVector;
+	var sources = pipeline.sources;
 
-	//NOTE: SKIPPED: look for initial match
-	//NOTE: SKIPPED: create a query object
+	// NOTE: SKIPPED: look for initial match
+	// NOTE: SKIPPED: create a query object
 
-	//Look for an initial simple project; we'll avoid constructing Values for fields that won't make it through the projection
+	// Look for an initial simple project; we'll avoid constructing Values for fields that won't make it through the projection
 	var projection = {};
-	var deps = [];
+	var deps = {};
 	var status = DocumentSource.GetDepsReturn.SEE_NEXT;
-	for (var i=0; i < sources.length && status != DocumentSource.GetDepsReturn.EXHAUSTIVE; i++) {
+	for (var i=0; i < sources.length && status !== DocumentSource.GetDepsReturn.EXHAUSTIVE; i++) {
 		status = sources[i].getDependencies(deps);
+		if(Object.keys(deps).length === 0) {
+			status = DocumentSource.GetDepsReturn.NOT_SUPPORTED;
+		}
 	}
-	if (status == DocumentSource.GetDepsReturn.EXHAUSTIVE) {
+	if (status === DocumentSource.GetDepsReturn.EXHAUSTIVE) {
 		projection = DocumentSource.depsToProjection(deps);
 	}
 
-	//NOTE: SKIPPED: Look for an initial sort
-	//NOTE: SKIPPED: Create the sort object
+	// NOTE: SKIPPED: Look for an initial sort
+	// NOTE: SKIPPED: Create the sort object
 
-//	//get the full "namespace" name
-//	var fullName = dbName + "." + pipeline.collectionName;
+	//get the full "namespace" name
+	// var fullName = dbName + "." + pipeline.collectionName;
 
-	//NOTE: SKIPPED: if(DEV) log messages
+	// NOTE: SKIPPED: if(DEV) log messages
 
-	//Create the necessary context to use a Cursor
-	//NOTE: SKIPPED: pSortedCursor bit
-	//NOTE: SKIPPED: pUnsortedCursor bit
+	// Create the necessary context to use a Cursor
+	// NOTE: SKIPPED: pSortedCursor bit
+	// NOTE: SKIPPED: pUnsortedCursor bit
 	var cursorWithContext = new CursorDocumentSource.CursorWithContext(/*fullName*/);
 
 	// Now add the Cursor to cursorWithContext
-	cursorWithContext._cursor = new Cursor( pipeline.collectionName );
+	cursorWithContext._cursor = new Cursor( expCtx.ns );	//NOTE: collectionName will likely be an array of documents in munge
 
 	// wrap the cursor with a DocumentSource and return that
 	var source = new CursorDocumentSource( cursorWithContext, expCtx );
 
-//	source.namespace = fullName;
-
-	//NOTE: SKIPPED: Note the query and sort
+	// NOTE: SKIPPED: Note the query and sort
 
 	if (Object.keys(projection).length) source.setProjection(projection);
 
-	return source;
+	while(sources.length > 0 && source.coalesce(sources[0])) { //Note: Attempting to coalesce into the cursor source
+		sources.shift();
+	}	
+
+	pipeline.addInitialSource(source);
 };

test/lib/pipeline/PipelineD.js

@@ -56,9 +56,9 @@ module.exports = {
 				proto.getDependencies = function(deps){
 					if (!deps.testDep){
 						deps.testDep = 1;
-						return DocumentSource.GetDepsReturn.SEE_NEXT;
+						return DocumentSource.GetDepsReturn.EXHAUSTIVE;
 					}
-					return DocumentSource.GetDepsReturn.EXHAUSTIVE;
+					return DocumentSource.GetDepsReturn.SEE_NEXT;
 				};
 
 				klass.createFromJson = function(options, ctx){
@@ -72,16 +72,16 @@ module.exports = {
 
 		"prepareCursorSource": {
 
-			"should return a CursorDocumentSource": function () {
+			"should place a CursorDocumentSource in pipeline": function () {
 				var p = Pipeline.parseCommand({pipeline:[{$test:{coalesce:false}}, {$test:{coalesce:false}}], aggregate:[]}),
-					cs = PipelineD.prepareCursorSource(p, [1,2,3,4,5]);
-				assert.equal(cs.constructor, CursorDocumentSource);
+					cs = PipelineD.prepareCursorSource(p, {ns:[1,2,3,4,5]});
+				assert.equal(p.sources[0].constructor, CursorDocumentSource);
 			},
 
 			"should get projection from all sources": function () {
 				var p = Pipeline.parseCommand({pipeline:[{$test:{coalesce:false}}, {$test:{coalesce:false}}], aggregate:[]}),
-					cs = PipelineD.prepareCursorSource(p, [1,2,3,4,5]);
-				assert.deepEqual(cs._projection, {"_id":0,"testDep":1});
+					cs = PipelineD.prepareCursorSource(p, {ns:[1,2,3,4,5]});
+				assert.deepEqual(p.sources[0]._projection, {"_id":0,"testDep":1});
 			},
 
 			"should get projection's deps": function () {
@@ -104,8 +104,8 @@ module.exports = {
 					]
 				};
 				var p = Pipeline.parseCommand(cmdObj),
-					cs = PipelineD.prepareCursorSource(p, [1,2,3,4,5]);
-				assert.equal(JSON.stringify(cs._projection), JSON.stringify({_id: 1, 'a.b.c': 1, d: 1, 'e.f.g': 1}));
+					cs = PipelineD.prepareCursorSource(p, {ns:[1,2,3,4,5]});
+				assert.equal(JSON.stringify(p.sources[0]._projection), JSON.stringify({'a.b.c': 1, d: 1, 'e.f.g': 1, _id: 1}));
 			},
 
 			"should get group's deps": function(){
@@ -130,8 +130,8 @@ module.exports = {
 					]
 				};
 				var p = Pipeline.parseCommand(cmdObj),
-					cs = PipelineD.prepareCursorSource(p, [1,2,3,4,5]);
-				assert.equal(JSON.stringify(cs._projection), JSON.stringify({ _id: 0, a: 1, b: 1, 'x.y.z': 1 }));
+					cs = PipelineD.prepareCursorSource(p, {ns:[1,2,3,4,5]});
+				assert.equal(JSON.stringify(p.sources[0]._projection), JSON.stringify({ _id: 0, a: 1, b: 1, 'x.y.z': 1 }));
 			}
 		}
 	}