Merge branch 'devel' of https://github.com/triAGENS/ArangoDB into devel

arangod/Aql/ExecutionBlock.cpp

@@ -1489,7 +1489,7 @@ int EnumerateListBlock::initialize () {
   auto en = reinterpret_cast<EnumerateListNode const*>(_exeNode);
 
   // get the inVariable register id . . .
-  // staticAnalysis has been run, so getPlanNode()->_varOverview is set up
+  // planRegisters() has been run, so getPlanNode()->_varOverview is set up
   auto it = getPlanNode()->getVarOverview()->varInfo.find(en->_inVariable->id);
 
   if (it == getPlanNode()->getVarOverview()->varInfo.end()){
@@ -1765,7 +1765,8 @@ int CalculationBlock::initialize () {
     return res;
   }
 
-  // We know that staticAnalysis has been run, so getPlanNode()->_varOverview is set up
+  // We know that planRegisters has been run, so
+  // getPlanNode()->_varOverview is set up
   auto en = static_cast<CalculationNode const*>(getPlanNode());
 
   std::unordered_set<Variable*> inVars = _expression->variables();
@@ -1867,7 +1868,8 @@ int SubqueryBlock::initialize () {
     return res;
   }
 
-  // We know that staticAnalysis has been run, so getPlanNode()->_varOverview is set up
+  // We know that planRegisters() has been run, so
+  // getPlanNode()->_varOverview is set up
 
   auto en = static_cast<SubqueryNode const*>(getPlanNode());
 
@@ -1930,7 +1932,8 @@ int FilterBlock::initialize () {
     return res;
   }
 
-  // We know that staticAnalysis has been run, so getPlanNode()->_varOverview is set up
+  // We know that planRegisters() has been run, so
+  // getPlanNode()->_varOverview is set up
 
   auto en = static_cast<FilterNode const*>(getPlanNode());
 
@@ -2117,7 +2120,8 @@ int AggregateBlock::initialize () {
   _variableNames.clear();
 
   for (auto p : en->_aggregateVariables){
-    // We know that staticAnalysis has been run, so getPlanNode()->_varOverview is set up
+    // We know that planRegisters() has been run, so
+    // getPlanNode()->_varOverview is set up
     auto itOut = en->getVarOverview()->varInfo.find(p.first->id);
     TRI_ASSERT(itOut != en->getVarOverview()->varInfo.end());
 
@@ -2349,8 +2353,9 @@ int SortBlock::initialize () {
 
   _sortRegisters.clear();
 
-  for( auto p: en->_elements){
-    // We know that staticAnalysis has been run, so getPlanNode()->_varOverview is set up
+  for (auto p: en->_elements){
+    // We know that planRegisters() has been run, so
+    // getPlanNode()->_varOverview is set up
     auto it = en->getVarOverview()->varInfo.find(p.first->id);
     TRI_ASSERT(it != en->getVarOverview()->varInfo.end());
     _sortRegisters.push_back(make_pair(it->second.registerId, p.second));
@@ -3233,7 +3238,8 @@ int GatherBlock::initialize () {
     _sortRegisters.clear();
 
     for( auto p: en->_elements){
-      // We know that staticAnalysis has been run, so getPlanNode()->_varOverview is set up
+      // We know that planRegisters has been run, so
+      // getPlanNode()->_varOverview is set up
       auto it = en->getVarOverview()->varInfo.find(p.first->id);
       TRI_ASSERT(it != en->getVarOverview()->varInfo.end());
       _sortRegisters.push_back(make_pair(it->second.registerId, p.second));

arangod/Aql/ExecutionEngine.cpp

@@ -461,7 +461,7 @@ std::cout << "REGISTERING QUERY ON COORDINATOR WITH ID: " << id << "\n";
       collection->setCurrentShard(shardId);
 
       plan.findVarUsage();
-      plan.staticAnalysis();
+      plan.planRegisters();
 
       // create a JSON representation of the plan
       triagens::basics::Json result(triagens::basics::Json::Array);
@@ -693,7 +693,7 @@ ExecutionEngine* ExecutionEngine::instanciateFromPlan (QueryRegistry* queryRegis
     if (! plan->varUsageComputed()) {
       plan->findVarUsage();
     }
-    plan->staticAnalysis();
+    plan->planRegisters();
 
     ExecutionBlock* root = nullptr;
 

arangod/Aql/ExecutionNode.cpp

@@ -209,7 +209,7 @@ ExecutionNode::ExecutionNode (ExecutionPlan* plan,
     _id(JsonHelper::checkAndGetNumericValue<size_t>(json.json(), "id")),
     _estimatedCost(0.0), 
     _estimatedCostSet(false),
-    _varUsageValid(false),
+    _varUsageValid(true),
     _plan(plan),
     _depth(JsonHelper::checkAndGetNumericValue<size_t>(json.json(), "depth")) {
 
@@ -532,9 +532,9 @@ triagens::basics::Json ExecutionNode::toJsonHelperGeneric (triagens::basics::Jso
 /// @brief static analysis debugger
 ////////////////////////////////////////////////////////////////////////////////
 
-struct StaticAnalysisDebugger : public WalkerWorker<ExecutionNode> {
-  StaticAnalysisDebugger () : indent(0) {};
-  ~StaticAnalysisDebugger () {};
+struct RegisterPlanningDebugger : public WalkerWorker<ExecutionNode> {
+  RegisterPlanningDebugger () : indent(0) {};
+  ~RegisterPlanningDebugger () {};
 
   int indent;
 
@@ -571,10 +571,10 @@ struct StaticAnalysisDebugger : public WalkerWorker<ExecutionNode> {
 };
 
 ////////////////////////////////////////////////////////////////////////////////
-/// @brief staticAnalysis
+/// @brief planRegisters
 ////////////////////////////////////////////////////////////////////////////////
 
-void ExecutionNode::staticAnalysis (ExecutionNode* super) {
+void ExecutionNode::planRegisters (ExecutionNode* super) {
   // The super is only for the case of subqueries.
   shared_ptr<VarOverview> v;
   if (super == nullptr) {
@@ -588,14 +588,14 @@ void ExecutionNode::staticAnalysis (ExecutionNode* super) {
   // Now handle the subqueries:
   for (auto s : v->subQueryNodes) {
     auto sq = static_cast<SubqueryNode*>(s);
-    sq->getSubquery()->staticAnalysis(s);
+    sq->getSubquery()->planRegisters(s);
   }
   v->reset();
 
   // Just for debugging:
   /*
   std::cout << std::endl;
-  StaticAnalysisDebugger debugger;
+  RegisterPlanningDebugger debugger;
   walk(&debugger);
   std::cout << std::endl;
   */
@@ -1808,6 +1808,7 @@ void ReturnNode::toJsonHelper (triagens::basics::Json& nodes,
 ////////////////////////////////////////////////////////////////////////////////
 /// @brief clone ExecutionNode recursively
 ////////////////////////////////////////////////////////////////////////////////
+
 ExecutionNode* ReturnNode::clone (ExecutionPlan* plan,
                                   bool withDependencies,
                                   bool withProperties) const {

arangod/Aql/ExecutionNode.h

@@ -566,7 +566,7 @@ namespace triagens {
 /// @brief static analysis
 ////////////////////////////////////////////////////////////////////////////////
 
-        void staticAnalysis (ExecutionNode* super = nullptr);
+        void planRegisters (ExecutionNode* super = nullptr);
 
 ////////////////////////////////////////////////////////////////////////////////
 /// @brief get varOverview
@@ -694,13 +694,13 @@ namespace triagens {
         ExecutionPlan* _plan;
 
 ////////////////////////////////////////////////////////////////////////////////
-/// @brief info about variables, filled in by staticAnalysis
+/// @brief info about variables, filled in by planRegisters
 ////////////////////////////////////////////////////////////////////////////////
 
         std::shared_ptr<VarOverview> _varOverview;
 
 ////////////////////////////////////////////////////////////////////////////////
-/// @brief depth of the current frame, will be filled in by staticAnalysis
+/// @brief depth of the current frame, will be filled in by planRegisters
 ////////////////////////////////////////////////////////////////////////////////
 
         int _depth;

arangod/Aql/ExecutionPlan.h

@@ -234,8 +234,8 @@ namespace triagens {
 /// @brief static analysis
 ////////////////////////////////////////////////////////////////////////////////
 
-        void staticAnalysis () {
-          _root->staticAnalysis();
+        void planRegisters () {
+          _root->planRegisters();
         }
 
 ////////////////////////////////////////////////////////////////////////////////

arangod/Aql/RestAqlHandler.cpp

@@ -414,6 +414,10 @@ void RestAqlHandler::deleteQuery (std::string const& idString) {
 ///   "pos":   The number of the row in "items" to take, usually 0.
 /// For the "shutdown" operation no additional arguments are required and
 /// an empty JSON object in the body is OK.
+/// All operations allow to set the HTTP header "Shard-ID:". If this is
+/// set, then the root block of the stored query must be a ScatterBlock
+/// and the shard ID is given as an additional argument to the ScatterBlock's
+/// special API.
 ////////////////////////////////////////////////////////////////////////////////
 
 void RestAqlHandler::useQuery (std::string const& operation,
@@ -654,6 +658,13 @@ bool RestAqlHandler::findQuery (std::string const& idString,
 void RestAqlHandler::handleUseQuery (std::string const& operation,
                                      Query* query,
                                      Json const& queryJson) {
+  bool found;
+  std::string shardId;
+  char const* shardIdCharP = _request->header("shard-id", found);
+  if (shardIdCharP != nullptr) {
+    shardId = shardIdCharP;
+  }
+
   Json answerBody(Json::Array, 2);
 
   if (operation == "getSome") {
@@ -661,7 +672,17 @@ void RestAqlHandler::handleUseQuery (std::string const& operation,
                                                          "atLeast", 1);
     auto atMost = JsonHelper::getNumericValue<uint64_t>(queryJson.json(),
                                "atMost", ExecutionBlock::DefaultBatchSize);
-    std::unique_ptr<AqlItemBlock> items(query->engine()->getSome(atLeast, atMost));
+    std::unique_ptr<AqlItemBlock> items;
+    if (shardId.empty()) {
+      items.reset(query->engine()->getSome(atLeast, atMost));
+    }
+    else {
+      auto scatter = static_cast<ScatterBlock*>(query->engine()->root());
+      if (scatter->getPlanNode()->getType() != ExecutionNode::SCATTER) {
+        THROW_ARANGO_EXCEPTION(TRI_ERROR_INTERNAL);
+      }
+      items.reset(scatter->getSomeForShard(atLeast, atMost, shardId));
+    }
     if (items.get() == nullptr) {
       answerBody("exhausted", Json(true))
                 ("error", Json(false));