SERVER-24638 Move command processing from Pipeline to AggregationRequest

1 files changed, 34 insertions, 52 deletions

diff --git a/src/mongo/db/pipeline/pipeline.h b/src/mongo/db/pipeline/pipeline.h
index f851edefd2d..84a8333ff8e 100644
--- a/src/mongo/db/pipeline/pipeline.h
+++ b/src/mongo/db/pipeline/pipeline.h
@@ -28,7 +28,8 @@
 
 #pragma once
 
-#include <deque>
+#include <list>
+#include <vector>
 
 #include <boost/intrusive_ptr.hpp>
 
@@ -42,33 +43,30 @@ class BSONObj;
 class BSONObjBuilder;
 class ClientBasic;
 class CollatorInterface;
-class Command;
 struct DepsTracker;
 class DocumentSource;
 struct ExpressionContext;
 class OperationContext;
-class Privilege;
 
-/** mongodb "commands" (sent via db.$cmd.findOne(...))
-    subclass to make a command.  define a singleton object for it.
-    */
+/**
+ * A Pipeline object represents a list of DocumentSources and is responsible for optimizing the
+ * pipeline.
+ */
 class Pipeline : public IntrusiveCounterUnsigned {
 public:
     typedef std::list<boost::intrusive_ptr<DocumentSource>> SourceContainer;
 
     /**
-     * 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
+     * Parses a Pipeline from a BSONElement representing a list of DocumentSources. Returns a non-OK
+     * status if it failed to parse.
      */
-    static boost::intrusive_ptr<Pipeline> parseCommand(
-        std::string& errmsg,
-        const BSONObj& cmdObj,
-        const boost::intrusive_ptr<ExpressionContext>& pCtx);
+    static StatusWith<boost::intrusive_ptr<Pipeline>> parse(
+        const std::vector<BSONObj>& rawPipeline,
+        const boost::intrusive_ptr<ExpressionContext>& expCtx);
 
-    // Helper to implement Command::checkAuthForCommand
+    /**
+     * Helper to implement Command::checkAuthForCommand.
+     */
     static Status checkAuthForCommand(ClientBasic* client,
                                       const std::string& dbname,
                                       const BSONObj& cmdObj);
@@ -86,7 +84,7 @@ public:
      * Sets the OperationContext of 'pCtx' to nullptr.
      *
      * The PipelineProxyStage is responsible for detaching the OperationContext and releasing any
-     * storage-engine state of the DocumentSourceCursor that may be present in 'sources'.
+     * storage-engine state of the DocumentSourceCursor that may be present in '_sources'.
      */
     void detachFromOperationContext();
 
@@ -94,7 +92,7 @@ public:
      * Sets the OperationContext of 'pCtx' to 'opCtx'.
      *
      * The PipelineProxyStage is responsible for reattaching the OperationContext and reacquiring
-     * any storage-engine state of the DocumentSourceCursor that may be present in 'sources'.
+     * any storage-engine state of the DocumentSourceCursor that may be present in '_sources'.
      */
     void reattachToOperationContext(OperationContext* opCtx);
 
@@ -140,22 +138,9 @@ public:
     std::vector<NamespaceString> getInvolvedCollections() const;
 
     /**
-      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
-    */
-    Document serialize() const;
-
-    /** Stitch together the source pointers (by calling setSource) for each source in sources.
-     *  Must be called after optimize and addInitialSource but before trying to get results.
+     * Serializes the pipeline into a form that can be parsed into an equivalent pipeline.
      */
-    void stitch();
+    std::vector<Value> serialize() const;
 
     /**
       Run the Pipeline on the given source.
@@ -164,17 +149,13 @@ public:
     */
     void run(BSONObjBuilder& result);
 
-    bool isExplain() const {
-        return explain;
-    }
-
     /// The initial source is special since it varies between mongos and mongod.
     void addInitialSource(boost::intrusive_ptr<DocumentSource> source);
 
     /// The source that represents the output. Returns a non-owning pointer.
     DocumentSource* output() {
-        invariant(!sources.empty());
-        return sources.back().get();
+        invariant(!_sources.empty());
+        return _sources.back().get();
     }
 
     /**
@@ -191,10 +172,6 @@ public:
      */
     DepsTracker getDependencies(const BSONObj& initialQuery) const;
 
-    /**
-      The aggregation command name.
-     */
-    static const char commandName[];
 
     /*
       PipelineD is a "sister" class that has additional functionality
@@ -220,17 +197,22 @@ private:
     friend class Optimizations::Local;
     friend class Optimizations::Sharded;
 
-    static const char pipelineName[];
-    static const char collationName[];
-    static const char explainName[];
-    static const char fromRouterName[];
-    static const char serverPipelineName[];
-    static const char mongosPipelineName[];
-
     Pipeline(const boost::intrusive_ptr<ExpressionContext>& pCtx);
 
-    SourceContainer sources;
-    bool explain;
+    /**
+     * Stitch together the source pointers by calling setSource() for each source in '_sources'.
+     * This function must be called any time the order of stages within the pipeline changes, e.g.
+     * in optimizePipeline().
+     */
+    void stitch();
+
+    /**
+     * Returns a non-OK status if any stage is in an invalid position. For example, if an $out stage
+     * is present but is not the last stage in the pipeline.
+     */
+    Status ensureAllStagesAreInLegalPositions() const;
+
+    SourceContainer _sources;
 
     boost::intrusive_ptr<ExpressionContext> pCtx;
 };