1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
|
/**
* Copyright (c) 2011 10gen Inc.
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License, version 3,
* as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*
* As a special exception, the copyright holders give permission to link the
* code of portions of this program with the OpenSSL library under certain
* conditions as described in each individual source file and distribute
* linked combinations including the program with the OpenSSL library. You
* must comply with the GNU Affero General Public License in all respects for
* all of the code used other than as permitted herein. If you modify file(s)
* with this exception, you may extend this exception to your version of the
* file(s), but you are not obligated to do so. If you do not wish to do so,
* delete this exception statement from your version. If you delete this
* exception statement from all source files in the program, then also delete
* it in the license file.
*/
#pragma once
#include "mongo/pch.h"
#include <boost/optional.hpp>
#include <boost/unordered_map.hpp>
#include <deque>
#include "mongo/db/clientcursor.h"
#include "mongo/db/jsobj.h"
#include "mongo/db/matcher/matcher.h"
#include "mongo/db/pipeline/document.h"
#include "mongo/db/pipeline/dependencies.h"
#include "mongo/db/pipeline/expression_context.h"
#include "mongo/db/pipeline/expression.h"
#include "mongo/db/pipeline/value.h"
#include "mongo/db/sorter/sorter.h"
#include "mongo/s/shard.h"
#include "mongo/s/strategy.h"
#include "mongo/util/intrusive_counter.h"
namespace mongo {
class Accumulator;
class Document;
class Expression;
class ExpressionFieldPath;
class ExpressionObject;
class DocumentSourceLimit;
class PlanExecutor;
class DocumentSource : public IntrusiveCounterUnsigned {
public:
virtual ~DocumentSource() {}
/** Returns the next Document if there is one or boost::none if at EOF.
* Subclasses must call pExpCtx->checkForInterupt().
*/
virtual boost::optional<Document> getNext() = 0;
/**
* Inform the source that it is no longer needed and may release its resources. After
* dispose() is called the source must still be able to handle iteration requests, but may
* become eof().
* NOTE: For proper mutex yielding, dispose() must be called on any DocumentSource that will
* not be advanced until eof(), see SERVER-6123.
*/
virtual void dispose();
/**
Get the source's name.
@returns the std::string name of the source as a constant string;
this is static, and there's no need to worry about adopting it
*/
virtual const char *getSourceName() const;
/**
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().
@param pSource the underlying source to use
*/
virtual void setSource(DocumentSource *pSource);
/**
Attempt to coalesce this DocumentSource with its successor in the
document processing pipeline. If successful, the successor
DocumentSource should be removed from the pipeline and discarded.
If successful, this operation can be applied repeatedly, in an
attempt to coalesce several sources together.
The default implementation is to do nothing, and return false.
@param pNextSource the next source in the document processing chain.
@returns whether or not the attempt to coalesce was successful or not;
if the attempt was not successful, nothing has been changed
*/
virtual bool coalesce(const intrusive_ptr<DocumentSource> &pNextSource);
/**
Optimize the pipeline operation, if possible. This is a local
optimization that only looks within this DocumentSource. For best
results, first coalesce compatible sources using coalesce().
This is intended for any operations that include expressions, and
provides a hook for those to optimize those operations.
The default implementation is to do nothing.
*/
virtual void optimize();
enum GetDepsReturn {
NOT_SUPPORTED = 0x0, // The full object and all metadata may be required
SEE_NEXT = 0x1, // Later stages could need either fields or metadata
EXHAUSTIVE_FIELDS = 0x2, // Later stages won't need more fields from input
EXHAUSTIVE_META = 0x4, // Later stages won't need more metadata from input
EXHAUSTIVE_ALL = EXHAUSTIVE_FIELDS | EXHAUSTIVE_META, // Later stages won't need either
};
/**
* Get the dependencies this operation needs to do its job.
*/
virtual GetDepsReturn getDependencies(DepsTracker* deps) const {
return NOT_SUPPORTED;
}
/**
* In the default case, serializes the DocumentSource and adds it to the std::vector<Value>.
*
* A subclass may choose to overwrite this, rather than serialize,
* if it should output multiple stages (eg, $sort sometimes also outputs a $limit).
*/
virtual void serializeToArray(std::vector<Value>& array, bool explain = false) const;
/// Returns true if doesn't require an input source (most DocumentSources do).
virtual bool isValidInitialSource() const { return false; }
protected:
/**
Base constructor.
*/
DocumentSource(const intrusive_ptr<ExpressionContext> &pExpCtx);
/*
Most DocumentSources have an underlying source they get their data
from. This is a convenience for them.
The default implementation of setSource() sets this; if you don't
need a source, override that to verify(). The default is to
verify() if this has already been set.
*/
DocumentSource *pSource;
intrusive_ptr<ExpressionContext> pExpCtx;
private:
/**
* Create a Value that represents the document source.
*
* This is used by the default implementation of serializeToArray() to add this object
* to a pipeline being serialized. Returning a missing() Value results in no entry
* being added to the array for this stage (DocumentSource).
*/
virtual Value serialize(bool explain = false) const = 0;
};
/** This class marks DocumentSources that should be split between the merger and the shards.
* See Pipeline::Optimizations::Sharded::findSplitPoint() for details.
*/
class SplittableDocumentSource {
public:
/** returns a source to be run on the shards.
* if NULL, don't run on shards
*/
virtual intrusive_ptr<DocumentSource> getShardSource() = 0;
/** returns a source that combines results from shards.
* if NULL, don't run on merger
*/
virtual intrusive_ptr<DocumentSource> getMergeSource() = 0;
protected:
// It is invalid to delete through a SplittableDocumentSource-typed pointer.
virtual ~SplittableDocumentSource() {}
};
/** This class marks DocumentSources which need mongod-specific functionality.
* It causes a MongodInterface to be injected when in a mongod and prevents mongos from
* merging pipelines containing this stage.
*/
class DocumentSourceNeedsMongod {
public:
// Wraps mongod-specific functions to allow linking into mongos.
class MongodInterface {
public:
virtual ~MongodInterface() {};
/**
* Always returns a DBDirectClient.
* Callers must not cache the returned pointer outside the scope of a single function.
*/
virtual DBClientBase* directClient() = 0;
// Note that in some rare cases this could return a false negative but will never return
// a false positive. This method will be fixed in the future once it becomes possible to
// avoid false negatives.
virtual bool isSharded(const NamespaceString& ns) = 0;
virtual bool isCapped(const NamespaceString& ns) = 0;
// Add new methods as needed.
};
void injectMongodInterface(boost::shared_ptr<MongodInterface> mongod) {
_mongod = mongod;
}
protected:
// It is invalid to delete through a DocumentSourceNeedsMongod-typed pointer.
virtual ~DocumentSourceNeedsMongod() {}
// Gives subclasses access to a MongodInterface implementation
shared_ptr<MongodInterface> _mongod;
};
class DocumentSourceBsonArray :
public DocumentSource {
public:
// virtuals from DocumentSource
virtual boost::optional<Document> getNext();
virtual Value serialize(bool explain = false) const;
virtual void setSource(DocumentSource *pSource);
virtual bool isValidInitialSource() const { return true; }
/**
Create a document source based on a BSON array.
This is usually put at the beginning of a chain of document sources
in order to fetch data from the database.
CAUTION: the BSON is not read until the source is used. Any
elements that appear after these documents must not be read until
this source is exhausted.
@param array the BSON array to treat as a document source
@param pExpCtx the expression context for the pipeline
@returns the newly created document source
*/
static intrusive_ptr<DocumentSourceBsonArray> create(
const BSONObj& array,
const intrusive_ptr<ExpressionContext> &pExpCtx);
private:
DocumentSourceBsonArray(
const BSONObj& embeddedArray,
const intrusive_ptr<ExpressionContext> &pExpCtx);
BSONObj embeddedObject;
BSONObjIterator arrayIterator;
};
class DocumentSourceCommandShards :
public DocumentSource {
public:
// virtuals from DocumentSource
virtual boost::optional<Document> getNext();
virtual Value serialize(bool explain = false) const;
virtual void setSource(DocumentSource *pSource);
virtual bool isValidInitialSource() const { return true; }
/* convenient shorthand for a commonly used type */
typedef std::vector<Strategy::CommandResult> ShardOutput;
/** Returns the result arrays from shards using the 2.4 protocol.
* Call this instead of getNext() if you want access to the raw streams.
* This method should only be called at most once.
*/
std::vector<BSONArray> getArrays();
/**
Create a DocumentSource that wraps the output of many shards
@param shardOutput output from the individual shards
@param pExpCtx the expression context for the pipeline
@returns the newly created DocumentSource
*/
static intrusive_ptr<DocumentSourceCommandShards> create(
const ShardOutput& shardOutput,
const intrusive_ptr<ExpressionContext>& pExpCtx);
private:
DocumentSourceCommandShards(const ShardOutput& shardOutput,
const intrusive_ptr<ExpressionContext>& pExpCtx);
/**
Advance to the next document, setting pCurrent appropriately.
Adjusts pCurrent, pBsonSource, and iterator, as needed. On exit,
pCurrent is the Document to return, or NULL. If NULL, this
indicates there is nothing more to return.
*/
void getNextDocument();
bool unstarted;
bool hasCurrent;
bool newSource; // set to true for the first item of a new source
intrusive_ptr<DocumentSourceBsonArray> pBsonSource;
Document pCurrent;
ShardOutput::const_iterator iterator;
ShardOutput::const_iterator listEnd;
};
/**
* Constructs and returns Documents from the BSONObj objects produced by a supplied
* PlanExecutor.
*
* An object of this type may only be used by one thread, see SERVER-6123.
*/
class DocumentSourceCursor :
public DocumentSource {
public:
// virtuals from DocumentSource
virtual ~DocumentSourceCursor();
virtual boost::optional<Document> getNext();
virtual const char *getSourceName() const;
virtual Value serialize(bool explain = false) const;
virtual void setSource(DocumentSource *pSource);
virtual bool coalesce(const intrusive_ptr<DocumentSource>& nextSource);
virtual bool isValidInitialSource() const { return true; }
virtual void dispose();
/**
* Create a document source based on a passed-in PlanExecutor.
*
* This is usually put at the beginning of a chain of document sources
* in order to fetch data from the database.
*/
static intrusive_ptr<DocumentSourceCursor> create(
const std::string& ns,
const boost::shared_ptr<PlanExecutor>& exec,
const intrusive_ptr<ExpressionContext> &pExpCtx);
/*
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.
@param pBsonObj the query to record
*/
void setQuery(const BSONObj& query) { _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.
@param pBsonObj the sort to record
*/
void setSort(const BSONObj& sort) { _sort = sort; }
/**
* Informs this object of projection and dependency information.
*
* @param projection A projection specification describing the fields needed by the rest of
* the pipeline.
* @param deps The output of DepsTracker::toParsedDeps
*/
void setProjection(const BSONObj& projection, const boost::optional<ParsedDeps>& deps);
/// returns -1 for no limit
long long getLimit() const;
private:
DocumentSourceCursor(
const std::string& ns,
const boost::shared_ptr<PlanExecutor>& exec,
const intrusive_ptr<ExpressionContext> &pExpCtx);
void loadBatch();
std::deque<Document> _currentBatch;
// BSONObj members must outlive _projection and cursor.
BSONObj _query;
BSONObj _sort;
BSONObj _projection;
boost::optional<ParsedDeps> _dependencies;
intrusive_ptr<DocumentSourceLimit> _limit;
long long _docsAddedToBatches; // for _limit enforcement
const std::string _ns;
boost::shared_ptr<PlanExecutor> _exec; // PipelineRunner holds a weak_ptr to this.
};
class DocumentSourceGroup : public DocumentSource
, public SplittableDocumentSource {
public:
// virtuals from DocumentSource
virtual boost::optional<Document> getNext();
virtual const char *getSourceName() const;
virtual void optimize();
virtual GetDepsReturn getDependencies(DepsTracker* deps) const;
virtual void dispose();
virtual Value serialize(bool explain = false) const;
/**
Create a new grouping DocumentSource.
@param pExpCtx the expression context for the pipeline
@returns the DocumentSource
*/
static intrusive_ptr<DocumentSourceGroup> create(
const intrusive_ptr<ExpressionContext> &pExpCtx);
/**
Add an accumulator.
Accumulators become fields in the Documents that result from
grouping. Each unique group document must have it's own
accumulator; the accumulator factory is used to create that.
@param fieldName the name the accumulator result will have in the
result documents
@param pAccumulatorFactory used to create the accumulator for the
group field
*/
void addAccumulator(const std::string& fieldName,
intrusive_ptr<Accumulator> (*pAccumulatorFactory)(),
const intrusive_ptr<Expression> &pExpression);
/// Tell this source if it is doing a merge from shards. Defaults to false.
void setDoingMerge(bool doingMerge) { _doingMerge = doingMerge; }
/**
Create a grouping DocumentSource from BSON.
This is a convenience method that uses the above, and operates on
a BSONElement that has been deteremined to be an Object with an
element named $group.
@param pBsonElement the BSONELement that defines the group
@param pExpCtx the expression context
@returns the grouping DocumentSource
*/
static intrusive_ptr<DocumentSource> createFromBson(
BSONElement elem,
const intrusive_ptr<ExpressionContext> &pExpCtx);
// Virtuals for SplittableDocumentSource
virtual intrusive_ptr<DocumentSource> getShardSource();
virtual intrusive_ptr<DocumentSource> getMergeSource();
static const char groupName[];
private:
DocumentSourceGroup(const intrusive_ptr<ExpressionContext> &pExpCtx);
/// Spill groups map to disk and returns an iterator to the file.
shared_ptr<Sorter<Value, Value>::Iterator> spill();
// Only used by spill. Would be function-local if that were legal in C++03.
class SpillSTLComparator;
/*
Before returning anything, this source must fetch everything from
the underlying source and group it. populate() is used to do that
on the first call to any method on this source. The populated
boolean indicates that this has been done.
*/
void populate();
bool populated;
/**
* Parses the raw id expression into _idExpressions and possibly _idFieldNames.
*/
void parseIdExpression(BSONElement groupField, const VariablesParseState& vps);
/**
* Computes the internal representation of the group key.
*/
Value computeId(Variables* vars);
/**
* Converts the internal representation of the group key to the _id shape specified by the
* user.
*/
Value expandId(const Value& val);
typedef std::vector<intrusive_ptr<Accumulator> > Accumulators;
typedef boost::unordered_map<Value, Accumulators, Value::Hash> GroupsMap;
GroupsMap groups;
/*
The field names for the result documents and the accumulator
factories for the result documents. The Expressions are the
common expressions used by each instance of each accumulator
in order to find the right-hand side of what gets added to the
accumulator. Note that each of those is the same for each group,
so we can share them across all groups by adding them to the
accumulators after we use the factories to make a new set of
accumulators for each new group.
These three vectors parallel each other.
*/
std::vector<std::string> vFieldName;
std::vector<intrusive_ptr<Accumulator> (*)()> vpAccumulatorFactory;
std::vector<intrusive_ptr<Expression> > vpExpression;
Document makeDocument(const Value& id, const Accumulators& accums, bool mergeableOutput);
bool _doingMerge;
bool _spilled;
const bool _extSortAllowed;
const int _maxMemoryUsageBytes;
boost::scoped_ptr<Variables> _variables;
std::vector<std::string> _idFieldNames; // used when id is a document
std::vector<intrusive_ptr<Expression> > _idExpressions;
// only used when !_spilled
GroupsMap::iterator groupsIterator;
// only used when _spilled
scoped_ptr<Sorter<Value, Value>::Iterator> _sorterIterator;
std::pair<Value, Value> _firstPartOfNextGroup;
Value _currentId;
Accumulators _currentAccumulators;
};
class DocumentSourceMatch : public DocumentSource {
public:
// virtuals from DocumentSource
virtual boost::optional<Document> getNext();
virtual const char *getSourceName() const;
virtual bool coalesce(const intrusive_ptr<DocumentSource>& nextSource);
virtual Value serialize(bool explain = false) const;
virtual void setSource(DocumentSource* Source);
/**
Create a filter.
@param pBsonElement the raw BSON specification for the filter
@returns the filter
*/
static intrusive_ptr<DocumentSource> createFromBson(
BSONElement elem,
const intrusive_ptr<ExpressionContext> &pCtx);
/// Returns the query in Matcher syntax.
BSONObj getQuery() const;
static const char matchName[];
/** Returns the portion of the match that can safely be promoted to before a $redact.
* If this returns an empty BSONObj, no part of this match may safely be promoted.
*
* To be safe to promote, removing a field from a document to be matched must not cause
* that document to be accepted when it would otherwise be rejected. As an example,
* {name: {$ne: "bob smith"}} accepts documents without a name field, which means that
* running this filter before a redact that would remove the name field would leak
* information. On the other hand, {age: {$gt:5}} is ok because it doesn't accept documents
* that have had their age field removed.
*/
BSONObj redactSafePortion() const;
static bool isTextQuery(const BSONObj& query);
bool isTextQuery() const { return _isTextQuery; }
private:
DocumentSourceMatch(const BSONObj &query,
const intrusive_ptr<ExpressionContext> &pExpCtx);
scoped_ptr<Matcher> matcher;
bool _isTextQuery;
};
class DocumentSourceMergeCursors :
public DocumentSource {
public:
typedef std::vector<std::pair<ConnectionString, CursorId> > CursorIds;
// virtuals from DocumentSource
boost::optional<Document> getNext();
virtual void setSource(DocumentSource *pSource);
virtual const char *getSourceName() const;
virtual void dispose();
virtual Value serialize(bool explain = false) const;
virtual bool isValidInitialSource() const { return true; }
static intrusive_ptr<DocumentSource> createFromBson(
BSONElement elem,
const intrusive_ptr<ExpressionContext> &pExpCtx);
static intrusive_ptr<DocumentSource> create(
const CursorIds& cursorIds,
const intrusive_ptr<ExpressionContext> &pExpCtx);
static const char name[];
/** Returns non-owning pointers to cursors managed by this stage.
* Call this instead of getNext() if you want access to the raw streams.
* This method should only be called at most once.
*/
std::vector<DBClientCursor*> getCursors();
/**
* Returns the next object from the cursor, throwing an appropriate exception if the cursor
* reported an error. This is a better form of DBClientCursor::nextSafe.
*/
static Document nextSafeFrom(DBClientCursor* cursor);
private:
struct CursorAndConnection {
CursorAndConnection(ConnectionString host, NamespaceString ns, CursorId id);
ScopedDbConnection connection;
DBClientCursor cursor;
};
// using list to enable removing arbitrary elements
typedef std::list<boost::shared_ptr<CursorAndConnection> > Cursors;
DocumentSourceMergeCursors(
const CursorIds& cursorIds,
const intrusive_ptr<ExpressionContext> &pExpCtx);
// Converts _cursorIds into active _cursors.
void start();
// This is the description of cursors to merge.
const CursorIds _cursorIds;
// These are the actual cursors we are merging. Created lazily.
Cursors _cursors;
Cursors::iterator _currentCursor;
bool _unstarted;
};
class DocumentSourceOut : public DocumentSource
, public SplittableDocumentSource
, public DocumentSourceNeedsMongod {
public:
// virtuals from DocumentSource
virtual ~DocumentSourceOut();
virtual boost::optional<Document> getNext();
virtual const char *getSourceName() const;
virtual Value serialize(bool explain = false) const;
virtual GetDepsReturn getDependencies(DepsTracker* deps) const;
// Virtuals for SplittableDocumentSource
virtual intrusive_ptr<DocumentSource> getShardSource() { return NULL; }
virtual intrusive_ptr<DocumentSource> getMergeSource() { return this; }
const NamespaceString& getOutputNs() const { return _outputNs; }
/**
Create a document source for output and pass-through.
This can be put anywhere in a pipeline and will store content as
well as pass it on.
@param pBsonElement the raw BSON specification for the source
@param pExpCtx the expression context for the pipeline
@returns the newly created document source
*/
static intrusive_ptr<DocumentSource> createFromBson(
BSONElement elem,
const intrusive_ptr<ExpressionContext> &pExpCtx);
static const char outName[];
private:
DocumentSourceOut(const NamespaceString& outputNs,
const intrusive_ptr<ExpressionContext> &pExpCtx);
// Sets _tempsNs and prepares it to receive data.
void prepTempCollection();
void spill(DBClientBase* conn, const std::vector<BSONObj>& toInsert);
bool _done;
NamespaceString _tempNs; // output goes here as it is being processed.
const NamespaceString _outputNs; // output will go here after all data is processed.
};
class DocumentSourceProject : public DocumentSource {
public:
// virtuals from DocumentSource
virtual boost::optional<Document> getNext();
virtual const char *getSourceName() const;
virtual void optimize();
virtual Value serialize(bool explain = false) const;
virtual GetDepsReturn getDependencies(DepsTracker* deps) const;
/**
Create a new projection DocumentSource from BSON.
This is a convenience for directly handling BSON, and relies on the
above methods.
@param pBsonElement the BSONElement with an object named $project
@param pExpCtx the expression context for the pipeline
@returns the created projection
*/
static intrusive_ptr<DocumentSource> createFromBson(
BSONElement elem,
const intrusive_ptr<ExpressionContext> &pExpCtx);
static const char projectName[];
/** projection as specified by the user */
BSONObj getRaw() const { return _raw; }
private:
DocumentSourceProject(const intrusive_ptr<ExpressionContext>& pExpCtx,
const intrusive_ptr<ExpressionObject>& exprObj);
// configuration state
boost::scoped_ptr<Variables> _variables;
intrusive_ptr<ExpressionObject> pEO;
BSONObj _raw;
};
class DocumentSourceRedact :
public DocumentSource {
public:
virtual boost::optional<Document> getNext();
virtual const char* getSourceName() const;
virtual void optimize();
static const char redactName[];
static intrusive_ptr<DocumentSource> createFromBson(
BSONElement elem,
const intrusive_ptr<ExpressionContext>& expCtx);
virtual Value serialize(bool explain = false) const;
private:
DocumentSourceRedact(const intrusive_ptr<ExpressionContext>& expCtx,
const intrusive_ptr<Expression>& previsit);
// These both work over _variables
boost::optional<Document> redactObject(); // redacts CURRENT
Value redactValue(const Value& in);
Variables::Id _currentId;
boost::scoped_ptr<Variables> _variables;
intrusive_ptr<Expression> _expression;
};
class DocumentSourceSort : public DocumentSource
, public SplittableDocumentSource {
public:
// virtuals from DocumentSource
virtual boost::optional<Document> getNext();
virtual const char *getSourceName() const;
virtual void serializeToArray(std::vector<Value>& array, bool explain = false) const;
virtual bool coalesce(const intrusive_ptr<DocumentSource> &pNextSource);
virtual void dispose();
virtual GetDepsReturn getDependencies(DepsTracker* deps) const;
virtual intrusive_ptr<DocumentSource> getShardSource();
virtual intrusive_ptr<DocumentSource> getMergeSource();
/**
Add sort key field.
Adds a sort key field to the key being built up. A concatenated
key is built up by calling this repeatedly.
@param fieldPath the field path to the key component
@param ascending if true, use the key for an ascending sort,
otherwise, use it for descending
*/
void addKey(const std::string &fieldPath, bool ascending);
/// Write out a Document whose contents are the sort key.
Document serializeSortKey(bool explain) const;
/**
Create a sorting DocumentSource from BSON.
This is a convenience method that uses the above, and operates on
a BSONElement that has been deteremined to be an Object with an
element named $group.
@param pBsonElement the BSONELement that defines the group
@param pExpCtx the expression context for the pipeline
@returns the grouping DocumentSource
*/
static intrusive_ptr<DocumentSource> createFromBson(
BSONElement elem,
const intrusive_ptr<ExpressionContext> &pExpCtx);
/// Create a DocumentSourceSort with a given sort and (optional) limit
static intrusive_ptr<DocumentSourceSort> create(
const intrusive_ptr<ExpressionContext> &pExpCtx,
BSONObj sortOrder,
long long limit=-1);
/// returns -1 for no limit
long long getLimit() const;
intrusive_ptr<DocumentSourceLimit> getLimitSrc() const { return limitSrc; }
static const char sortName[];
private:
DocumentSourceSort(const intrusive_ptr<ExpressionContext> &pExpCtx);
virtual Value serialize(bool explain = false) const {
verify(false); // should call addToBsonArray instead
}
/*
Before returning anything, this source must fetch everything from
the underlying source and group it. populate() is used to do that
on the first call to any method on this source. The populated
boolean indicates that this has been done.
*/
void populate();
bool populated;
SortOptions makeSortOptions() const;
// These are used to merge pre-sorted results from a DocumentSourceMergeCursors or a
// DocumentSourceCommandShards depending on whether we have finished upgrading to 2.6 or
// not.
class IteratorFromCursor;
class IteratorFromBsonArray;
void populateFromCursors(const std::vector<DBClientCursor*>& cursors);
void populateFromBsonArrays(const std::vector<BSONArray>& arrays);
/* these two parallel each other */
typedef std::vector<intrusive_ptr<Expression> > SortKey;
SortKey vSortKey;
std::vector<char> vAscending; // used like std::vector<bool> but without specialization
/// Extracts the fields in vSortKey from the Document;
Value extractKey(const Document& d) const;
/// Compare two Values according to the specified sort key.
int compare(const Value& lhs, const Value& rhs) const;
typedef Sorter<Value, Document> MySorter;
// For MySorter
class Comparator {
public:
explicit Comparator(const DocumentSourceSort& source): _source(source) {}
int operator()(const MySorter::Data& lhs, const MySorter::Data& rhs) const {
return _source.compare(lhs.first, rhs.first);
}
private:
const DocumentSourceSort& _source;
};
intrusive_ptr<DocumentSourceLimit> limitSrc;
bool _done;
bool _mergingPresorted;
scoped_ptr<MySorter::Iterator> _output;
};
class DocumentSourceLimit : public DocumentSource
, public SplittableDocumentSource {
public:
// virtuals from DocumentSource
virtual boost::optional<Document> getNext();
virtual const char *getSourceName() const;
virtual bool coalesce(const intrusive_ptr<DocumentSource> &pNextSource);
virtual Value serialize(bool explain = false) const;
virtual GetDepsReturn getDependencies(DepsTracker* deps) const {
return SEE_NEXT; // This doesn't affect needed fields
}
/**
Create a new limiting DocumentSource.
@param pExpCtx the expression context for the pipeline
@returns the DocumentSource
*/
static intrusive_ptr<DocumentSourceLimit> create(
const intrusive_ptr<ExpressionContext> &pExpCtx,
long long limit);
// Virtuals for SplittableDocumentSource
// Need to run on rounter. Running on shard as well is an optimization.
virtual intrusive_ptr<DocumentSource> getShardSource() { return this; }
virtual intrusive_ptr<DocumentSource> getMergeSource() { return this; }
long long getLimit() const { return limit; }
void setLimit(long long newLimit) { limit = newLimit; }
/**
Create a limiting DocumentSource from BSON.
This is a convenience method that uses the above, and operates on
a BSONElement that has been deteremined to be an Object with an
element named $limit.
@param pBsonElement the BSONELement that defines the limit
@param pExpCtx the expression context
@returns the grouping DocumentSource
*/
static intrusive_ptr<DocumentSource> createFromBson(
BSONElement elem,
const intrusive_ptr<ExpressionContext> &pExpCtx);
static const char limitName[];
private:
DocumentSourceLimit(const intrusive_ptr<ExpressionContext> &pExpCtx,
long long limit);
long long limit;
long long count;
};
class DocumentSourceSkip : public DocumentSource
, public SplittableDocumentSource {
public:
// virtuals from DocumentSource
virtual boost::optional<Document> getNext();
virtual const char *getSourceName() const;
virtual bool coalesce(const intrusive_ptr<DocumentSource> &pNextSource);
virtual Value serialize(bool explain = false) const;
virtual GetDepsReturn getDependencies(DepsTracker* deps) const {
return SEE_NEXT; // This doesn't affect needed fields
}
/**
Create a new skipping DocumentSource.
@param pExpCtx the expression context
@returns the DocumentSource
*/
static intrusive_ptr<DocumentSourceSkip> create(
const intrusive_ptr<ExpressionContext> &pExpCtx);
// Virtuals for SplittableDocumentSource
// Need to run on rounter. Can't run on shards.
virtual intrusive_ptr<DocumentSource> getShardSource() { return NULL; }
virtual intrusive_ptr<DocumentSource> getMergeSource() { return this; }
long long getSkip() const { return _skip; }
void setSkip(long long newSkip) { _skip = newSkip; }
/**
Create a skipping DocumentSource from BSON.
This is a convenience method that uses the above, and operates on
a BSONElement that has been deteremined to be an Object with an
element named $skip.
@param pBsonElement the BSONELement that defines the skip
@param pExpCtx the expression context
@returns the grouping DocumentSource
*/
static intrusive_ptr<DocumentSource> createFromBson(
BSONElement elem,
const intrusive_ptr<ExpressionContext> &pExpCtx);
static const char skipName[];
private:
DocumentSourceSkip(const intrusive_ptr<ExpressionContext> &pExpCtx);
long long _skip;
bool _needToSkip;
};
class DocumentSourceUnwind :
public DocumentSource {
public:
// virtuals from DocumentSource
virtual boost::optional<Document> getNext();
virtual const char *getSourceName() const;
virtual Value serialize(bool explain = false) const;
virtual GetDepsReturn getDependencies(DepsTracker* deps) const;
/**
Create a new projection DocumentSource from BSON.
This is a convenience for directly handling BSON, and relies on the
above methods.
@param pBsonElement the BSONElement with an object named $project
@param pExpCtx the expression context for the pipeline
@returns the created projection
*/
static intrusive_ptr<DocumentSource> createFromBson(
BSONElement elem,
const intrusive_ptr<ExpressionContext> &pExpCtx);
static const char unwindName[];
private:
DocumentSourceUnwind(const intrusive_ptr<ExpressionContext> &pExpCtx);
/** Specify the field to unwind. */
void unwindPath(const FieldPath &fieldPath);
// Configuration state.
scoped_ptr<FieldPath> _unwindPath;
// Iteration state.
class Unwinder;
scoped_ptr<Unwinder> _unwinder;
};
class DocumentSourceGeoNear : public DocumentSource
, public SplittableDocumentSource
, public DocumentSourceNeedsMongod {
public:
// virtuals from DocumentSource
virtual boost::optional<Document> getNext();
virtual const char *getSourceName() const;
virtual void setSource(DocumentSource *pSource);
virtual bool coalesce(const intrusive_ptr<DocumentSource> &pNextSource);
virtual bool isValidInitialSource() const { return true; }
virtual Value serialize(bool explain = false) const;
// Virtuals for SplittableDocumentSource
virtual intrusive_ptr<DocumentSource> getShardSource();
virtual intrusive_ptr<DocumentSource> getMergeSource();
static intrusive_ptr<DocumentSource> createFromBson(
BSONElement elem,
const intrusive_ptr<ExpressionContext> &pCtx);
static char geoNearName[];
long long getLimit() { return limit; }
// this should only be used for testing
static intrusive_ptr<DocumentSourceGeoNear> create(
const intrusive_ptr<ExpressionContext> &pCtx);
private:
DocumentSourceGeoNear(const intrusive_ptr<ExpressionContext> &pExpCtx);
void parseOptions(BSONObj options);
BSONObj buildGeoNearCmd() const;
void runCommand();
// These fields describe the command to run.
// coords and distanceField are required, rest are optional
BSONObj coords; // "near" option, but near is a reserved keyword on windows
bool coordsIsArray;
scoped_ptr<FieldPath> distanceField; // Using scoped_ptr because FieldPath can't be empty
long long limit;
double maxDistance;
BSONObj query;
bool spherical;
double distanceMultiplier;
scoped_ptr<FieldPath> includeLocs;
bool uniqueDocs;
// these fields are used while processing the results
BSONObj cmdOutput;
boost::scoped_ptr<BSONObjIterator> resultsIterator; // iterator over cmdOutput["results"]
};
}
|