thoughts/node_modules/mongoose/lib/aggregate.js
node_modules/mongoose/lib/aggregate.jsRaw
1/*!
2 * Module dependencies
3 */
4
5var util = require('util');
6var utils = require('./utils');
7var PromiseProvider = require('./promise_provider');
8var Query = require('./query');
9var read = Query.prototype.read;
10
11/**
12 * Aggregate constructor used for building aggregation pipelines.
13 *
14 * ####Example:
15 *
16 * new Aggregate();
17 * new Aggregate({ $project: { a: 1, b: 1 } });
18 * new Aggregate({ $project: { a: 1, b: 1 } }, { $skip: 5 });
19 * new Aggregate([{ $project: { a: 1, b: 1 } }, { $skip: 5 }]);
20 *
21 * Returned when calling Model.aggregate().
22 *
23 * ####Example:
24 *
25 * Model
26 * .aggregate({ $match: { age: { $gte: 21 }}})
27 * .unwind('tags')
28 * .exec(callback)
29 *
30 * ####Note:
31 *
32 * - The documents returned are plain javascript objects, not mongoose documents (since any shape of document can be returned).
33 * - Requires MongoDB >= 2.1
34 * - Mongoose does **not** cast pipeline stages. `new Aggregate({ $match: { _id: '00000000000000000000000a' } });` will not work unless `_id` is a string in the database. Use `new Aggregate({ $match: { _id: mongoose.Types.ObjectId('00000000000000000000000a') } });` instead.
35 *
36 * @see MongoDB http://docs.mongodb.org/manual/applications/aggregation/
37 * @see driver http://mongodb.github.com/node-mongodb-native/api-generated/collection.html#aggregate
38 * @param {Object|Array} [ops] aggregation operator(s) or operator array
39 * @api public
40 */
41
42function Aggregate() {
43 this._pipeline = [];
44 this._model = undefined;
45 this.options = undefined;
46
47 if (arguments.length === 1 && util.isArray(arguments[0])) {
48 this.append.apply(this, arguments[0]);
49 } else {
50 this.append.apply(this, arguments);
51 }
52}
53
54/**
55 * Binds this aggregate to a model.
56 *
57 * @param {Model} model the model to which the aggregate is to be bound
58 * @return {Aggregate}
59 * @api public
60 */
61
62Aggregate.prototype.model = function(model) {
63 this._model = model;
64 return this;
65};
66
67/**
68 * Appends new operators to this aggregate pipeline
69 *
70 * ####Examples:
71 *
72 * aggregate.append({ $project: { field: 1 }}, { $limit: 2 });
73 *
74 * // or pass an array
75 * var pipeline = [{ $match: { daw: 'Logic Audio X' }} ];
76 * aggregate.append(pipeline);
77 *
78 * @param {Object} ops operator(s) to append
79 * @return {Aggregate}
80 * @api public
81 */
82
83Aggregate.prototype.append = function() {
84 var args = (arguments.length === 1 && util.isArray(arguments[0]))
85 ? arguments[0]
86 : utils.args(arguments);
87
88 if (!args.every(isOperator)) {
89 throw new Error('Arguments must be aggregate pipeline operators');
90 }
91
92 this._pipeline = this._pipeline.concat(args);
93
94 return this;
95};
96
97/**
98 * Appends a new $project operator to this aggregate pipeline.
99 *
100 * Mongoose query [selection syntax](#query_Query-select) is also supported.
101 *
102 * ####Examples:
103 *
104 * // include a, include b, exclude _id
105 * aggregate.project("a b -_id");
106 *
107 * // or you may use object notation, useful when
108 * // you have keys already prefixed with a "-"
109 * aggregate.project({a: 1, b: 1, _id: 0});
110 *
111 * // reshaping documents
112 * aggregate.project({
113 * newField: '$b.nested'
114 * , plusTen: { $add: ['$val', 10]}
115 * , sub: {
116 * name: '$a'
117 * }
118 * })
119 *
120 * // etc
121 * aggregate.project({ salary_k: { $divide: [ "$salary", 1000 ] } });
122 *
123 * @param {Object|String} arg field specification
124 * @see projection http://docs.mongodb.org/manual/reference/aggregation/project/
125 * @return {Aggregate}
126 * @api public
127 */
128
129Aggregate.prototype.project = function(arg) {
130 var fields = {};
131
132 if (typeof arg === 'object' && !util.isArray(arg)) {
133 Object.keys(arg).forEach(function(field) {
134 fields[field] = arg[field];
135 });
136 } else if (arguments.length === 1 && typeof arg === 'string') {
137 arg.split(/\s+/).forEach(function(field) {
138 if (!field) {
139 return;
140 }
141 var include = field[0] === '-' ? 0 : 1;
142 if (include === 0) {
143 field = field.substring(1);
144 }
145 fields[field] = include;
146 });
147 } else {
148 throw new Error('Invalid project() argument. Must be string or object');
149 }
150
151 return this.append({$project: fields});
152};
153
154/**
155 * Appends a new custom $group operator to this aggregate pipeline.
156 *
157 * ####Examples:
158 *
159 * aggregate.group({ _id: "$department" });
160 *
161 * @see $group http://docs.mongodb.org/manual/reference/aggregation/group/
162 * @method group
163 * @memberOf Aggregate
164 * @param {Object} arg $group operator contents
165 * @return {Aggregate}
166 * @api public
167 */
168
169/**
170 * Appends a new custom $match operator to this aggregate pipeline.
171 *
172 * ####Examples:
173 *
174 * aggregate.match({ department: { $in: [ "sales", "engineering" } } });
175 *
176 * @see $match http://docs.mongodb.org/manual/reference/aggregation/match/
177 * @method match
178 * @memberOf Aggregate
179 * @param {Object} arg $match operator contents
180 * @return {Aggregate}
181 * @api public
182 */
183
184/**
185 * Appends a new $skip operator to this aggregate pipeline.
186 *
187 * ####Examples:
188 *
189 * aggregate.skip(10);
190 *
191 * @see $skip http://docs.mongodb.org/manual/reference/aggregation/skip/
192 * @method skip
193 * @memberOf Aggregate
194 * @param {Number} num number of records to skip before next stage
195 * @return {Aggregate}
196 * @api public
197 */
198
199/**
200 * Appends a new $limit operator to this aggregate pipeline.
201 *
202 * ####Examples:
203 *
204 * aggregate.limit(10);
205 *
206 * @see $limit http://docs.mongodb.org/manual/reference/aggregation/limit/
207 * @method limit
208 * @memberOf Aggregate
209 * @param {Number} num maximum number of records to pass to the next stage
210 * @return {Aggregate}
211 * @api public
212 */
213
214/**
215 * Appends a new $geoNear operator to this aggregate pipeline.
216 *
217 * ####NOTE:
218 *
219 * **MUST** be used as the first operator in the pipeline.
220 *
221 * ####Examples:
222 *
223 * aggregate.near({
224 * near: [40.724, -73.997],
225 * distanceField: "dist.calculated", // required
226 * maxDistance: 0.008,
227 * query: { type: "public" },
228 * includeLocs: "dist.location",
229 * uniqueDocs: true,
230 * num: 5
231 * });
232 *
233 * @see $geoNear http://docs.mongodb.org/manual/reference/aggregation/geoNear/
234 * @method near
235 * @memberOf Aggregate
236 * @param {Object} parameters
237 * @return {Aggregate}
238 * @api public
239 */
240
241Aggregate.prototype.near = function(arg) {
242 var op = {};
243 op.$geoNear = arg;
244 return this.append(op);
245};
246
247/*!
248 * define methods
249 */
250
251'group match skip limit out'.split(' ').forEach(function($operator) {
252 Aggregate.prototype[$operator] = function(arg) {
253 var op = {};
254 op['$' + $operator] = arg;
255 return this.append(op);
256 };
257});
258
259/**
260 * Appends new custom $unwind operator(s) to this aggregate pipeline.
261 *
262 * Note that the `$unwind` operator requires the path name to start with '$'.
263 * Mongoose will prepend '$' if the specified field doesn't start '$'.
264 *
265 * ####Examples:
266 *
267 * aggregate.unwind("tags");
268 * aggregate.unwind("a", "b", "c");
269 *
270 * @see $unwind http://docs.mongodb.org/manual/reference/aggregation/unwind/
271 * @param {String} fields the field(s) to unwind
272 * @return {Aggregate}
273 * @api public
274 */
275
276Aggregate.prototype.unwind = function() {
277 var args = utils.args(arguments);
278
279 var res = [];
280 for (var i = 0; i < args.length; ++i) {
281 var arg = args[i];
282 if (arg && typeof arg === 'object') {
283 res.push({ $unwind: arg });
284 } else if (typeof arg === 'string') {
285 res.push({
286 $unwind: (arg && arg.charAt(0) === '$') ? arg : '$' + arg
287 });
288 } else {
289 throw new Error('Invalid arg "' + arg + '" to unwind(), ' +
290 'must be string or object');
291 }
292 }
293
294 return this.append.apply(this, res);
295};
296
297/**
298 * Appends new custom $lookup operator(s) to this aggregate pipeline.
299 *
300 * ####Examples:
301 *
302 * aggregate.lookup({ from: 'users', localField: 'userId', foreignField: '_id', as: 'users' });
303 *
304 * @see $lookup https://docs.mongodb.org/manual/reference/operator/aggregation/lookup/#pipe._S_lookup
305 * @param {Object} options to $lookup as described in the above link
306 * @return {Aggregate}
307 * @api public
308 */
309
310Aggregate.prototype.lookup = function(options) {
311 return this.append({$lookup: options});
312};
313
314/**
315 * Appends new custom $sample operator(s) to this aggregate pipeline.
316 *
317 * ####Examples:
318 *
319 * aggregate.sample(3); // Add a pipeline that picks 3 random documents
320 *
321 * @see $sample https://docs.mongodb.org/manual/reference/operator/aggregation/sample/#pipe._S_sample
322 * @param {Number} size number of random documents to pick
323 * @return {Aggregate}
324 * @api public
325 */
326
327Aggregate.prototype.sample = function(size) {
328 return this.append({$sample: {size: size}});
329};
330
331/**
332 * Appends a new $sort operator to this aggregate pipeline.
333 *
334 * If an object is passed, values allowed are `asc`, `desc`, `ascending`, `descending`, `1`, and `-1`.
335 *
336 * If a string is passed, it must be a space delimited list of path names. The sort order of each path is ascending unless the path name is prefixed with `-` which will be treated as descending.
337 *
338 * ####Examples:
339 *
340 * // these are equivalent
341 * aggregate.sort({ field: 'asc', test: -1 });
342 * aggregate.sort('field -test');
343 *
344 * @see $sort http://docs.mongodb.org/manual/reference/aggregation/sort/
345 * @param {Object|String} arg
346 * @return {Aggregate} this
347 * @api public
348 */
349
350Aggregate.prototype.sort = function(arg) {
351 // TODO refactor to reuse the query builder logic
352
353 var sort = {};
354
355 if (arg.constructor.name === 'Object') {
356 var desc = ['desc', 'descending', -1];
357 Object.keys(arg).forEach(function(field) {
358 sort[field] = desc.indexOf(arg[field]) === -1 ? 1 : -1;
359 });
360 } else if (arguments.length === 1 && typeof arg === 'string') {
361 arg.split(/\s+/).forEach(function(field) {
362 if (!field) {
363 return;
364 }
365 var ascend = field[0] === '-' ? -1 : 1;
366 if (ascend === -1) {
367 field = field.substring(1);
368 }
369 sort[field] = ascend;
370 });
371 } else {
372 throw new TypeError('Invalid sort() argument. Must be a string or object.');
373 }
374
375 return this.append({$sort: sort});
376};
377
378/**
379 * Sets the readPreference option for the aggregation query.
380 *
381 * ####Example:
382 *
383 * Model.aggregate(..).read('primaryPreferred').exec(callback)
384 *
385 * @param {String} pref one of the listed preference options or their aliases
386 * @param {Array} [tags] optional tags for this query
387 * @see mongodb http://docs.mongodb.org/manual/applications/replication/#read-preference
388 * @see driver http://mongodb.github.com/node-mongodb-native/driver-articles/anintroductionto1_1and2_2.html#read-preferences
389 */
390
391Aggregate.prototype.read = function(pref, tags) {
392 if (!this.options) {
393 this.options = {};
394 }
395 read.call(this, pref, tags);
396 return this;
397};
398
399/**
400 * Execute the aggregation with explain
401 *
402 * ####Example:
403 *
404 * Model.aggregate(..).explain(callback)
405 *
406 * @param {Function} callback
407 * @return {Promise}
408 */
409
410Aggregate.prototype.explain = function(callback) {
411 var _this = this;
412 var Promise = PromiseProvider.get();
413 return new Promise.ES6(function(resolve, reject) {
414 if (!_this._pipeline.length) {
415 var err = new Error('Aggregate has empty pipeline');
416 if (callback) {
417 callback(err);
418 }
419 reject(err);
420 return;
421 }
422
423 prepareDiscriminatorPipeline(_this);
424
425 _this._model
426 .collection
427 .aggregate(_this._pipeline, _this.options || {})
428 .explain(function(error, result) {
429 if (error) {
430 if (callback) {
431 callback(error);
432 }
433 reject(error);
434 return;
435 }
436
437 if (callback) {
438 callback(null, result);
439 }
440 resolve(result);
441 });
442 });
443};
444
445/**
446 * Sets the allowDiskUse option for the aggregation query (ignored for < 2.6.0)
447 *
448 * ####Example:
449 *
450 * Model.aggregate(..).allowDiskUse(true).exec(callback)
451 *
452 * @param {Boolean} value Should tell server it can use hard drive to store data during aggregation.
453 * @param {Array} [tags] optional tags for this query
454 * @see mongodb http://docs.mongodb.org/manual/reference/command/aggregate/
455 */
456
457Aggregate.prototype.allowDiskUse = function(value) {
458 if (!this.options) {
459 this.options = {};
460 }
461 this.options.allowDiskUse = value;
462 return this;
463};
464
465/**
466 * Sets the cursor option option for the aggregation query (ignored for < 2.6.0).
467 * Note the different syntax below: .exec() returns a cursor object, and no callback
468 * is necessary.
469 *
470 * ####Example:
471 *
472 * var cursor = Model.aggregate(..).cursor({ batchSize: 1000 }).exec();
473 * cursor.each(function(error, doc) {
474 * // use doc
475 * });
476 *
477 * @param {Object} options set the cursor batch size
478 * @see mongodb http://mongodb.github.io/node-mongodb-native/2.0/api/AggregationCursor.html
479 */
480
481Aggregate.prototype.cursor = function(options) {
482 if (!this.options) {
483 this.options = {};
484 }
485 this.options.cursor = options || {};
486 return this;
487};
488
489/**
490 * Adds a [cursor flag](http://mongodb.github.io/node-mongodb-native/2.1/api/Cursor.html#addCursorFlag)
491 *
492 * ####Example:
493 *
494 * var cursor = Model.aggregate(..).cursor({ batchSize: 1000 }).exec();
495 * cursor.each(function(error, doc) {
496 * // use doc
497 * });
498 *
499 * @param {String} flag
500 * @param {Boolean} value
501 * @see mongodb http://mongodb.github.io/node-mongodb-native/2.1/api/Cursor.html#addCursorFlag
502 */
503
504Aggregate.prototype.addCursorFlag = function(flag, value) {
505 if (!this.options) {
506 this.options = {};
507 }
508 this.options[flag] = value;
509 return this;
510};
511
512/**
513 * Executes the aggregate pipeline on the currently bound Model.
514 *
515 * ####Example:
516 *
517 * aggregate.exec(callback);
518 *
519 * // Because a promise is returned, the `callback` is optional.
520 * var promise = aggregate.exec();
521 * promise.then(..);
522 *
523 * @see Promise #promise_Promise
524 * @param {Function} [callback]
525 * @return {Promise}
526 * @api public
527 */
528
529Aggregate.prototype.exec = function(callback) {
530 if (!this._model) {
531 throw new Error('Aggregate not bound to any Model');
532 }
533 var _this = this;
534 var Promise = PromiseProvider.get();
535 var options = utils.clone(this.options);
536
537 if (options && options.cursor) {
538 if (options.cursor.async) {
539 delete options.cursor.async;
540 return new Promise.ES6(function(resolve) {
541 if (!_this._model.collection.buffer) {
542 process.nextTick(function() {
543 var cursor = _this._model.collection.
544 aggregate(_this._pipeline, options || {});
545 resolve(cursor);
546 callback && callback(null, cursor);
547 });
548 return;
549 }
550 _this._model.collection.emitter.once('queue', function() {
551 var cursor = _this._model.collection.
552 aggregate(_this._pipeline, options || {});
553 resolve(cursor);
554 callback && callback(null, cursor);
555 });
556 });
557 }
558 return this._model.collection.
559 aggregate(this._pipeline, this.options || {});
560 }
561
562 return new Promise.ES6(function(resolve, reject) {
563 if (!_this._pipeline.length) {
564 var err = new Error('Aggregate has empty pipeline');
565 if (callback) {
566 callback(err);
567 }
568 reject(err);
569 return;
570 }
571
572 prepareDiscriminatorPipeline(_this);
573
574 _this._model
575 .collection
576 .aggregate(_this._pipeline, _this.options || {}, function(error, result) {
577 if (error) {
578 if (callback) {
579 callback(error);
580 }
581 reject(error);
582 return;
583 }
584
585 if (callback) {
586 callback(null, result);
587 }
588 resolve(result);
589 });
590 });
591};
592
593/**
594 * Provides promise for aggregate.
595 *
596 * ####Example:
597 *
598 * Model.aggregate(..).then(successCallback, errorCallback);
599 *
600 * @see Promise #promise_Promise
601 * @param {Function} [resolve] successCallback
602 * @param {Function} [reject] errorCallback
603 * @return {Promise}
604 */
605Aggregate.prototype.then = function(resolve, reject) {
606 var _this = this;
607 var Promise = PromiseProvider.get();
608 var promise = new Promise.ES6(function(success, error) {
609 _this.exec(function(err, val) {
610 if (err) error(err);
611 else success(val);
612 });
613 });
614 return promise.then(resolve, reject);
615};
616
617/*!
618 * Helpers
619 */
620
621/**
622 * Checks whether an object is likely a pipeline operator
623 *
624 * @param {Object} obj object to check
625 * @return {Boolean}
626 * @api private
627 */
628
629function isOperator(obj) {
630 var k;
631
632 if (typeof obj !== 'object') {
633 return false;
634 }
635
636 k = Object.keys(obj);
637
638 return k.length === 1 && k
639 .some(function(key) {
640 return key[0] === '$';
641 });
642}
643
644/*!
645 * Adds the appropriate `$match` pipeline step to the top of an aggregate's
646 * pipeline, should it's model is a non-root discriminator type. This is
647 * analogous to the `prepareDiscriminatorCriteria` function in `lib/query.js`.
648 *
649 * @param {Aggregate} aggregate Aggregate to prepare
650 */
651
652function prepareDiscriminatorPipeline(aggregate) {
653 var schema = aggregate._model.schema,
654 discriminatorMapping = schema && schema.discriminatorMapping;
655
656 if (discriminatorMapping && !discriminatorMapping.isRoot) {
657 var originalPipeline = aggregate._pipeline,
658 discriminatorKey = discriminatorMapping.key,
659 discriminatorValue = discriminatorMapping.value;
660
661 // If the first pipeline stage is a match and it doesn't specify a `__t`
662 // key, add the discriminator key to it. This allows for potential
663 // aggregation query optimizations not to be disturbed by this feature.
664 if (originalPipeline[0] && originalPipeline[0].$match && !originalPipeline[0].$match[discriminatorKey]) {
665 originalPipeline[0].$match[discriminatorKey] = discriminatorValue;
666 // `originalPipeline` is a ref, so there's no need for
667 // aggregate._pipeline = originalPipeline
668 } else if (originalPipeline[0] && originalPipeline[0].$geoNear) {
669 originalPipeline[0].$geoNear.query =
670 originalPipeline[0].$geoNear.query || {};
671 originalPipeline[0].$geoNear.query[discriminatorKey] = discriminatorValue;
672 } else {
673 var match = {};
674 match[discriminatorKey] = discriminatorValue;
675 aggregate._pipeline = [{$match: match}].concat(originalPipeline);
676 }
677 }
678}
679
680
681/*!
682 * Exports
683 */
684
685module.exports = Aggregate;