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
|
/**
* Copyright 2011 (c) 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/>.
*/
#pragma once
#include "pch.h"
#include "db/jsobj.h"
#include "util/timer.h"
#include "db/commands.h"
namespace mongo {
class BSONObj;
class BSONObjBuilder;
class DocumentSource;
class DocumentSourceProject;
class Expression;
class ExpressionContext;
class ExpressionNary;
struct OpDesc; // local private struct
/** mongodb "commands" (sent via db.$cmd.findOne(...))
subclass to make a command. define a singleton object for it.
*/
class Pipeline :
boost::noncopyable {
public:
virtual ~Pipeline();
/*
Create a pipeline from the command.
@param errmsg where to write errors, if there are any
@param cmdObj the command object sent from the client
@returns the pipeline, if created, otherwise a NULL reference
*/
static boost::shared_ptr<Pipeline> parseCommand(
string &errmsg, BSONObj &cmdObj,
const intrusive_ptr<ExpressionContext> &pCtx);
/*
Get the collection name from the command.
@returns the collection name
*/
string getCollectionName() const;
/*
Split the current Pipeline into a Pipeline for each shard, and
a Pipeline that combines the results within mongos.
This permanently alters this pipeline for the merging operation.
@returns the Spec for the pipeline command that should be sent
to the shards
*/
boost::shared_ptr<Pipeline> splitForSharded();
/*
Get Cursor creation modifiers.
If we have a $match or a $sort at the beginning of the pipeline,
these can be extracted and used to modify the cursor we'll use for
the initial collection scan.
If there is a Matcher query at the beginning of the pipeline,
get it, by adding its terms to the object under construction. If
not, this adds nothing to the object under construction.
If there is a sort at the beginning of the pipeline, get it, by
adding its terms to the object under construction. If not, this adds
nothing.
Optimization steps in parseCommand make sure that for any pairs
of adjacent matches and sorts, the match comes first. This ensures
that we sort a minimum of items, and doesn't change the result.
When getCursorMods() examines the pipeline, it looks for an initial
$match. If present, that is put into pQueryBuilder. If there is
a query, then the next stage is checked for a $sort, which will go
into pSortBuilder. If there is no initial $match, then a check is
made for an initial $sort, which will then still be put into
pSortBuilder.
As a side-effect, retrieving the Cursor modifications removes them
from the pipeline.
@param pQueryBuilder an initialized object builder
@param pSortBuilder an initialized object builder
*/
void getCursorMods(BSONObjBuilder *pQueryBuilder,
BSONObjBuilder *pSortBuilder);
/*
Write the Pipeline as a BSONObj command. This should be the
inverse of parseCommand().
This is only intended to be used by the shard command obtained
from splitForSharded(). Some pipeline operations in the merge
process do not have equivalent command forms, and using this on
the mongos Pipeline will cause assertions.
@param the builder to write the command to
*/
void toBson(BSONObjBuilder *pBuilder) const;
/*
Run the Pipeline on the given source.
@param result builder to write the result to
@param errmsg place to put error messages, if any
@param pSource the document source to use at the head of the chain
@returns true on success, false if an error occurs
*/
bool run(BSONObjBuilder &result, string &errmsg,
intrusive_ptr<DocumentSource> pSource);
/*
Debugging: should the processing pipeline be split within
mongod, simulating the real mongos/mongod split? This is determined
by setting the splitMongodPipeline field in an "aggregate"
command.
The split itself is handled by the caller, which is currently
pipeline_command.cpp.
@returns true if the pipeline is to be split
*/
bool getSplitMongodPipeline() const;
/*
The aggregation command name.
*/
static const char commandName[];
private:
static const char pipelineName[];
static const char fromRouterName[];
static const char splitMongodPipelineName[];
Pipeline(const intrusive_ptr<ExpressionContext> &pCtx);
string collectionName;
typedef vector<intrusive_ptr<DocumentSource> > SourceVector;
SourceVector sourceVector;
bool splitMongodPipeline;
intrusive_ptr<ExpressionContext> pCtx;
};
} // namespace mongo
/* ======================= INLINED IMPLEMENTATIONS ========================== */
namespace mongo {
inline string Pipeline::getCollectionName() const {
return collectionName;
}
inline bool Pipeline::getSplitMongodPipeline() const {
if (!DEBUG_BUILD)
return false;
return splitMongodPipeline;
}
} // namespace mongo
|
