Implement DocumentSourceExtensionExpandable & DocumentSourceExtensionOptimizable

XMLWordPrintableJSON

    • Type: Task
    • Resolution: Unresolved
    • Priority: Major - P3
    • None
    • Affects Version/s: None
    • Component/s: None
    • Query Integration
    • None
    • None
    • None
    • None
    • None
    • None
    • None

      As outlined in the technical design, we will need to introduce an explicit desugaring step that will be called from the aggregation execution path immediately after query stats registration.

      This requires that we implement specializations of DocumentSourceExtension, which each have distinct behaviour during expansion

      As part of this ticket:

      1. Add expand() to the DocumentSource base class:

      virtual list<intrusive_ptr<DocumentSource>> DocumentSource::expand() {
   return list{}; // return an empty list
}

      2. Implement DocumentSourceExtensionExpandable:

      class DocumentSourceExtensionExpandable : public DocumentSourceExtension {

   virtual list<intrusive_ptr<DocumentSource>> expand() {
       list<intrusive_ptr<DocumentSource>> expandedList;

       // call expand() on this object's parseNode, populating with a list of DocumentSources.
      auto nodes = parseNode->expand();
      while (!nodes.empty()) {
          auto currentNode = nodes.pop_front();
          if (currentNode.type == Ast) {
             // create a DocumentSourceExtensionOptimizable;
             expandedList.emplace_back(make_intrusive<DocumentSourceExtensionOptimizable>(currentNode.release());
          }
  else if (currentNode.type == Parse) {
          if (currentNode.vtable == host::ParseNode::VTABLE) {
              // we have a host::ParseNode, we have to parse it!!
              expandedList.emplace_back(createFromBson(currentNode.getBson());
          } else {
              // we have an extension provided ParseNode, recursively expand!
            // create a DocumentSourceExtensionExpandable; 
            auto expandableNode = make_intrusive<DocumentSourceExtensionExpandable>(currentNode.release();
            expandedList.emplace_back(expandableNode->expand());
    

            }
          }

      }
      return expandedList;
   }

}

      3. Implement DocumentSourceExtensionOptimizable

        DocumentSourceExtensionOptimizable does not override expand, since it does not itself expand (return empty list). However, DocumentSourceExtensionOptimizable must have a LogicalStage member, which DocumentSourceExtension does not. 

      4. Add unit tests for the above two classes.

      Alternative: It's possible that we may not want to add expand() to the generic DocumentSource interface. Our long term place is to introduce expansion in the LiteParsedPipeline phase, so it might be better to avoid adding expand() as a virtual method of the DocumentSource base class. 

      Instead, we could leverage a visitor style Expansion/Desugarer class (note, if we go with this approach. don't implement this here, it should go in SERVER-109780)

      class Desugarer {public:

Desugarer(list<intrusive_ptr<DocumentSource>& pipelineSources) : m_sources(pipelineSources) {}

void operator() {   try {
      auto itr = m_sources.begin();      while (itr != m_sources.end()) \{
          invariant(itr.get());
          itr = _desugar(itr, *itr.get()); // This is like a visit() function.
      }
   } catch (...) \{
     
   }
}

DocumentSourceContainer::iterator _desugar(DocumentSourceContainer::iterator, DocumentSource& stage) {
   invariant(itr->get() == &stage);   // Non-Desugar stages just advance to the next stage in the pipeline.   return std::next(itr);
}

// When we visit an Expandable stage, we can invoke the expand() function, which is only present on this class.
DocumentSourceContainer::iterator _desugar(DocumentSourceContainer::iterator, DocumentSourceExtensionExpandable& stage) {
    invariant(itr->get() == &stage);
   
    auto desugaredPipeline = stage.expand();    // replace desugar stage with its desugared pipeline    if (!desugaredPipeline.empty()) \{
       auto desugarIterator = itr;
       m_sources.insert(itr, desugaredPipeline.begin(), desugaredPipeline.end());
       itr = std::next(itr);
       m_sources.erase(desugarIterator);
    }    return itr; 
}private:
DocumentSourceContainer& m_sources;
};

       

            Assignee:
            Josh Siegel
            Reporter:
            Santiago Roche
            Votes:
            0 Vote for this issue
            Watchers:
            1 Start watching this issue

              Created:
              Updated: