IMPALA-2905: Handle coordinator fragment lifecycle like all others

The plan-root fragment instance that runs on the coordinator should be
handled like all others: started via RPC and run asynchronously. Without
this, the fragment requires special-case code throughout the
coordinator, and does not show up in system metrics etc.

This patch adds a new sink type, PlanRootSink, to the root fragment
instance so that the coordinator can pull row batches that are pushed by
the root instance. The coordinator signals completion to the fragment
instance via closing the consumer side of the sink, whereupon the
instance is free to complete.

Since the root instance now runs asynchronously wrt to the coordinator,
we add several coordination methods to allow the coordinator to wait for
a point in the instance's execution to be hit - e.g. to wait until the
instance has been opened.

Done in this patch:

* Add PlanRootSink
* Add coordination to PFE to allow coordinator to observe lifecycle
* Make FragmentMgr a singleton
* Removed dead code from Coordinator::Wait() and elsewhere.
* Moved result output exprs out of QES and into PlanRootSink.
* Remove special-case limit-based teardown of coordinator fragment, and
  supporting functions in PlanFragmentExecutor.
* Simplified lifecycle of PlanFragmentExecutor by separating Open() into
  Open() and Exec(), the latter of which drives the sink by reading
  rows from the plan tree.
* Add child profile to PlanFragmentExecutor to measure time spent in
  each lifecycle phase.
* Removed dependency between InitExecProfiles() and starting root
  fragment.
* Removed mostly dead-code handling of LIMIT 0 queries.
* Ensured that SET returns a result set in all cases.
* Fix test_get_log() HS2 test. Errors are only guaranteed to be visible
  after fetch calls return EOS, but test was assuming this would happen
  after first fetch.

Change-Id: Ibb0064ec2f085fa3a5598ea80894fb489a01e4df
Reviewed-on: http://gerrit.cloudera.org:8080/4402
Tested-by: Internal Jenkins
Reviewed-by: Henry Robinson <he...@cloudera.com>


Project: http://git-wip-us.apache.org/repos/asf/incubator-impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-impala/commit/9f61397f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-impala/tree/9f61397f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-impala/diff/9f61397f

Branch: refs/heads/master
Commit: 9f61397fc4d638aa78b37db2cd5b9c35b6deed94
Parents: 05b91a9
Author: Henry Robinson <he...@cloudera.com>
Authored: Wed Oct 5 11:48:01 2016 -0700
Committer: Henry Robinson <he...@cloudera.com>
Committed: Sun Oct 16 15:55:29 2016 +0000

----------------------------------------------------------------------
 be/src/exec/CMakeLists.txt                      |   1 +
 be/src/exec/data-sink.cc                        |   4 +
 be/src/exec/plan-root-sink.cc                   | 171 ++++++
 be/src/exec/plan-root-sink.h                    | 133 +++++
 be/src/runtime/coordinator.cc                   | 560 +++++++------------
 be/src/runtime/coordinator.h                    | 142 ++---
 be/src/runtime/exec-env.cc                      |  19 +-
 be/src/runtime/exec-env.h                       |   2 +
 be/src/runtime/plan-fragment-executor.cc        | 264 ++++-----
 be/src/runtime/plan-fragment-executor.h         | 149 ++---
 be/src/scheduling/query-schedule.cc             |  32 +-
 be/src/scheduling/query-schedule.h              |  27 +-
 be/src/scheduling/simple-scheduler.cc           |   5 -
 be/src/service/fragment-exec-state.cc           |   6 +-
 be/src/service/fragment-exec-state.h            |   8 +-
 be/src/service/fragment-mgr.cc                  |   8 +-
 be/src/service/impala-beeswax-server.cc         |  52 +-
 be/src/service/impala-hs2-server.cc             |  36 +-
 be/src/service/impala-internal-service.h        |  15 +-
 be/src/service/impala-server.cc                 |   5 +-
 be/src/service/impala-server.h                  |  47 --
 be/src/service/query-exec-state.cc              | 107 +---
 be/src/service/query-exec-state.h               |  11 +-
 be/src/service/query-result-set.h               |  64 +++
 be/src/testutil/in-process-servers.cc           |   4 +
 common/thrift/DataSinks.thrift                  |   7 +-
 .../org/apache/impala/analysis/QueryStmt.java   |   6 +
 .../org/apache/impala/planner/PlanRootSink.java |  39 ++
 .../java/org/apache/impala/planner/Planner.java |   2 +
 .../apache/impala/planner/PlannerContext.java   |   1 +
 .../queries/PlannerTest/aggregation.test        | 104 ++++
 .../queries/PlannerTest/analytic-fns.test       | 118 ++++
 .../PlannerTest/complex-types-file-formats.test |  14 +
 .../queries/PlannerTest/conjunct-ordering.test  |  30 +
 .../queries/PlannerTest/constant.test           |   4 +
 .../queries/PlannerTest/data-source-tables.test |  10 +
 .../PlannerTest/disable-preaggregations.test    |   4 +
 .../queries/PlannerTest/distinct-estimate.test  |   8 +
 .../queries/PlannerTest/distinct.test           |  54 ++
 .../queries/PlannerTest/empty.test              |  58 ++
 .../queries/PlannerTest/hbase.test              | 118 ++++
 .../queries/PlannerTest/hdfs.test               | 226 ++++++++
 .../queries/PlannerTest/implicit-joins.test     |  28 +
 .../queries/PlannerTest/inline-view-limit.test  |  58 ++
 .../queries/PlannerTest/inline-view.test        | 116 ++++
 .../queries/PlannerTest/join-order.test         |  72 +++
 .../queries/PlannerTest/joins.test              | 178 ++++++
 .../queries/PlannerTest/kudu-selectivity.test   |  16 +
 .../queries/PlannerTest/kudu.test               |  32 ++
 .../PlannerTest/mem-limit-broadcast-join.test   |   2 +
 .../queries/PlannerTest/nested-collections.test | 144 +++++
 .../queries/PlannerTest/nested-loop-join.test   |  12 +
 .../queries/PlannerTest/order.test              | 104 ++++
 .../queries/PlannerTest/outer-joins.test        |  54 ++
 .../PlannerTest/partition-key-scans.test        |  38 ++
 .../PlannerTest/predicate-propagation.test      |  90 +++
 .../PlannerTest/runtime-filter-propagation.test |  86 +++
 .../queries/PlannerTest/small-query-opt.test    |  42 ++
 .../queries/PlannerTest/subquery-rewrite.test   | 144 +++++
 .../queries/PlannerTest/topn.test               |  48 ++
 .../queries/PlannerTest/tpcds-all.test          | 146 +++++
 .../queries/PlannerTest/tpch-all.test           | 132 +++++
 .../queries/PlannerTest/tpch-kudu.test          |  44 ++
 .../queries/PlannerTest/tpch-nested.test        |  88 +++
 .../queries/PlannerTest/tpch-views.test         |  44 ++
 .../queries/PlannerTest/union.test              | 198 +++++++
 .../queries/PlannerTest/values.test             |  16 +
 .../queries/PlannerTest/views.test              |  48 ++
 .../queries/PlannerTest/with-clause.test        |  58 ++
 tests/custom_cluster/test_client_ssl.py         |   1 +
 tests/failure/test_failpoints.py                |   2 +-
 tests/hs2/test_hs2.py                           |  16 +-
 tests/hs2/test_json_endpoints.py                |   4 +
 tests/shell/util.py                             |   5 +-
 74 files changed, 3831 insertions(+), 910 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/9f61397f/be/src/exec/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/be/src/exec/CMakeLists.txt b/be/src/exec/CMakeLists.txt
index 571198f..fce5c81 100644
--- a/be/src/exec/CMakeLists.txt
+++ b/be/src/exec/CMakeLists.txt
@@ -77,6 +77,7 @@ add_library(Exec
   partitioned-hash-join-builder-ir.cc
   partitioned-hash-join-node.cc
   partitioned-hash-join-node-ir.cc
+  plan-root-sink.cc
   kudu-scanner.cc
   kudu-scan-node.cc
   kudu-table-sink.cc

http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/9f61397f/be/src/exec/data-sink.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/data-sink.cc b/be/src/exec/data-sink.cc
index b6ec0ee..c95b854 100644
--- a/be/src/exec/data-sink.cc
+++ b/be/src/exec/data-sink.cc
@@ -26,6 +26,7 @@
 #include "exec/hdfs-table-sink.h"
 #include "exec/kudu-table-sink.h"
 #include "exec/kudu-util.h"
+#include "exec/plan-root-sink.h"
 #include "exprs/expr.h"
 #include "gen-cpp/ImpalaInternalService_constants.h"
 #include "gen-cpp/ImpalaInternalService_types.h"
@@ -94,6 +95,9 @@ Status DataSink::CreateDataSink(ObjectPool* pool,
       }
 
       break;
+    case TDataSinkType::PLAN_ROOT_SINK:
+      sink->reset(new PlanRootSink(row_desc, output_exprs, thrift_sink));
+      break;
     default:
       stringstream error_msg;
       map<int, const char*>::const_iterator i =

http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/9f61397f/be/src/exec/plan-root-sink.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/plan-root-sink.cc b/be/src/exec/plan-root-sink.cc
new file mode 100644
index 0000000..bd73953
--- /dev/null
+++ b/be/src/exec/plan-root-sink.cc
@@ -0,0 +1,171 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "exec/plan-root-sink.h"
+
+#include "exprs/expr-context.h"
+#include "exprs/expr.h"
+#include "runtime/row-batch.h"
+#include "runtime/tuple-row.h"
+#include "service/query-result-set.h"
+
+#include <memory>
+#include <boost/thread/mutex.hpp>
+
+using namespace std;
+using boost::unique_lock;
+using boost::mutex;
+
+namespace impala {
+
+const string PlanRootSink::NAME = "PLAN_ROOT_SINK";
+
+PlanRootSink::PlanRootSink(const RowDescriptor& row_desc,
+    const std::vector<TExpr>& output_exprs, const TDataSink& thrift_sink)
+  : DataSink(row_desc), thrift_output_exprs_(output_exprs) {}
+
+Status PlanRootSink::Prepare(RuntimeState* state, MemTracker* mem_tracker) {
+  RETURN_IF_ERROR(DataSink::Prepare(state, mem_tracker));
+  RETURN_IF_ERROR(
+      Expr::CreateExprTrees(state->obj_pool(), thrift_output_exprs_, 
&output_expr_ctxs_));
+  RETURN_IF_ERROR(
+      Expr::Prepare(output_expr_ctxs_, state, row_desc_, 
expr_mem_tracker_.get()));
+
+  return Status::OK();
+}
+
+Status PlanRootSink::Open(RuntimeState* state) {
+  RETURN_IF_ERROR(Expr::Open(output_expr_ctxs_, state));
+  return Status::OK();
+}
+
+namespace {
+
+/// Validates that all collection-typed slots in the given batch are set to 
NULL.
+/// See SubplanNode for details on when collection-typed slots are set to NULL.
+/// TODO: This validation will become obsolete when we can return collection 
values.
+/// We will then need a different mechanism to assert the correct behavior of 
the
+/// SubplanNode with respect to setting collection-slots to NULL.
+void ValidateCollectionSlots(const RowDescriptor& row_desc, RowBatch* batch) {
+#ifndef NDEBUG
+  if (!row_desc.HasVarlenSlots()) return;
+  for (int i = 0; i < batch->num_rows(); ++i) {
+    TupleRow* row = batch->GetRow(i);
+    for (int j = 0; j < row_desc.tuple_descriptors().size(); ++j) {
+      const TupleDescriptor* tuple_desc = row_desc.tuple_descriptors()[j];
+      if (tuple_desc->collection_slots().empty()) continue;
+      for (int k = 0; k < tuple_desc->collection_slots().size(); ++k) {
+        const SlotDescriptor* slot_desc = tuple_desc->collection_slots()[k];
+        int tuple_idx = row_desc.GetTupleIdx(slot_desc->parent()->id());
+        const Tuple* tuple = row->GetTuple(tuple_idx);
+        if (tuple == NULL) continue;
+        DCHECK(tuple->IsNull(slot_desc->null_indicator_offset()));
+      }
+    }
+  }
+#endif
+}
+}
+
+Status PlanRootSink::Send(RuntimeState* state, RowBatch* batch) {
+  ValidateCollectionSlots(row_desc_, batch);
+  int current_batch_row = 0;
+  do {
+    unique_lock<mutex> l(lock_);
+    while (results_ == nullptr && !consumer_done_) sender_cv_.wait(l);
+    if (consumer_done_ || batch == nullptr) {
+      eos_ = true;
+      return Status::OK();
+    }
+
+    // Otherwise the consumer is ready. Fill out the rows.
+    DCHECK(results_ != nullptr);
+    // List of expr values to hold evaluated rows from the query
+    vector<void*> result_row;
+    result_row.resize(output_expr_ctxs_.size());
+
+    // List of scales for floating point values in result_row
+    vector<int> scales;
+    scales.resize(result_row.size());
+
+    int num_to_fetch = batch->num_rows() - current_batch_row;
+    if (num_rows_requested_ > 0) num_to_fetch = min(num_to_fetch, 
num_rows_requested_);
+    for (int i = 0; i < num_to_fetch; ++i) {
+      TupleRow* row = batch->GetRow(current_batch_row);
+      GetRowValue(row, &result_row, &scales);
+      RETURN_IF_ERROR(results_->AddOneRow(result_row, scales));
+      ++current_batch_row;
+    }
+    // Signal the consumer.
+    results_ = nullptr;
+    ExprContext::FreeLocalAllocations(output_expr_ctxs_);
+    consumer_cv_.notify_all();
+  } while (current_batch_row < batch->num_rows());
+  return Status::OK();
+}
+
+Status PlanRootSink::FlushFinal(RuntimeState* state) {
+  unique_lock<mutex> l(lock_);
+  sender_done_ = true;
+  eos_ = true;
+  consumer_cv_.notify_all();
+  return Status::OK();
+}
+
+void PlanRootSink::Close(RuntimeState* state) {
+  unique_lock<mutex> l(lock_);
+  // No guarantee that FlushFinal() has been called, so need to mark 
sender_done_ here as
+  // well.
+  sender_done_ = true;
+  consumer_cv_.notify_all();
+  // Wait for consumer to be done, in case sender tries to tear-down this sink 
while the
+  // sender is still reading from it.
+  while (!consumer_done_) sender_cv_.wait(l);
+  Expr::Close(output_expr_ctxs_, state);
+  DataSink::Close(state);
+}
+
+void PlanRootSink::CloseConsumer() {
+  unique_lock<mutex> l(lock_);
+  consumer_done_ = true;
+  sender_cv_.notify_all();
+}
+
+Status PlanRootSink::GetNext(
+    RuntimeState* state, QueryResultSet* results, int num_results, bool* eos) {
+  unique_lock<mutex> l(lock_);
+  DCHECK(!consumer_done_);
+
+  results_ = results;
+  num_rows_requested_ = num_results;
+  sender_cv_.notify_all();
+
+  while (!eos_ && results_ != nullptr && !sender_done_) consumer_cv_.wait(l);
+  *eos = eos_;
+  RETURN_IF_ERROR(state->CheckQueryState());
+  return Status::OK();
+}
+
+void PlanRootSink::GetRowValue(
+    TupleRow* row, vector<void*>* result, vector<int>* scales) {
+  DCHECK(result->size() >= output_expr_ctxs_.size());
+  for (int i = 0; i < output_expr_ctxs_.size(); ++i) {
+    (*result)[i] = output_expr_ctxs_[i]->GetValue(row);
+    (*scales)[i] = output_expr_ctxs_[i]->root()->output_scale();
+  }
+}
+}

http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/9f61397f/be/src/exec/plan-root-sink.h
----------------------------------------------------------------------
diff --git a/be/src/exec/plan-root-sink.h b/be/src/exec/plan-root-sink.h
new file mode 100644
index 0000000..cc7c045
--- /dev/null
+++ b/be/src/exec/plan-root-sink.h
@@ -0,0 +1,133 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#ifndef IMPALA_EXEC_PLAN_ROOT_SINK_H
+#define IMPALA_EXEC_PLAN_ROOT_SINK_H
+
+#include "exec/data-sink.h"
+
+#include <boost/thread/condition_variable.hpp>
+
+namespace impala {
+
+class TupleRow;
+class RowBatch;
+class QueryResultSet;
+class ExprContext;
+
+/// Sink which manages the handoff between a 'sender' (a fragment instance) 
that produces
+/// batches by calling Send(), and a 'consumer' (e.g. the coordinator) which 
consumes rows
+/// formed by computing a set of output expressions over the input batches, by 
calling
+/// GetNext(). Send() and GetNext() are called concurrently.
+///
+/// The consumer calls GetNext() with a QueryResultSet and a requested fetch
+/// size. GetNext() shares these fields with Send(), and then signals Send() 
to begin
+/// populating the result set. GetNext() returns when either the sender has 
sent all of
+/// its rows, or the requested fetch size has been satisfied.
+///
+/// Send() fills in as many rows as are requested from the current batch. When 
the batch
+/// is exhausted - which may take several calls to GetNext() - control is 
returned to the
+/// sender to produce another row batch.
+///
+/// Consumers must call CloseConsumer() when finished to allow the fragment to 
shut
+/// down. Senders must call Close() to signal to the consumer that no more 
batches will be
+/// produced.
+///
+/// The sink is thread safe up to a single producer and single consumer.
+///
+/// TODO: The consumer drives the sender in lock-step with GetNext() calls, 
forcing a
+/// context-switch on every invocation. Measure the impact of this, and 
consider moving to
+/// a fully asynchronous implementation with a queue to manage buffering 
between sender
+/// and consumer. See IMPALA-4268.
+class PlanRootSink : public DataSink {
+ public:
+  PlanRootSink(const RowDescriptor& row_desc, const std::vector<TExpr>& 
output_exprs,
+      const TDataSink& thrift_sink);
+
+  virtual std::string GetName() { return NAME; }
+
+  virtual Status Prepare(RuntimeState* state, MemTracker* tracker);
+
+  virtual Status Open(RuntimeState* state);
+
+  /// Sends a new batch. Ownership of 'batch' remains with the sender. Blocks 
until the
+  /// consumer has consumed 'batch' by calling GetNext().
+  virtual Status Send(RuntimeState* state, RowBatch* batch);
+
+  /// Sets eos and notifies consumer.
+  virtual Status FlushFinal(RuntimeState* state);
+
+  /// To be called by sender only. Signals to the consumer that no more 
batches will be
+  /// produced, then blocks until the consumer calls CloseConsumer().
+  virtual void Close(RuntimeState* state);
+
+  /// Populates 'result_set' with up to 'num_rows' rows produced by the 
fragment instance
+  /// that calls Send(). *eos is set to 'true' when there are no more rows to 
consume.
+  Status GetNext(
+      RuntimeState* state, QueryResultSet* result_set, int num_rows, bool* 
eos);
+
+  /// Signals to the producer that the sink will no longer be used. It's an 
error to call
+  /// GetNext() after this returns. May be called more than once; only the 
first call has
+  /// any effect.
+  void CloseConsumer();
+
+  static const std::string NAME;
+
+ private:
+  /// Protects all members, including the condition variables.
+  boost::mutex lock_;
+
+  /// Waited on by the sender only. Signalled when the consumer has written 
results_ and
+  /// num_rows_requested_, and so the sender may begin satisfying that request 
for rows
+  /// from its current batch. Also signalled when CloseConsumer() is called, 
to unblock
+  /// the sender.
+  boost::condition_variable sender_cv_;
+
+  /// Waited on by the consumer only. Signalled when the sender has finished 
serving a
+  /// request for rows. Also signalled by Close() and FlushFinal() to signal 
to the
+  /// consumer that no more rows are coming.
+  boost::condition_variable consumer_cv_;
+
+  /// Signals to producer that the consumer is done, and the sink may be torn 
down.
+  bool consumer_done_ = false;
+
+  /// Signals to consumer that the sender is done, and that there are no more 
row batches
+  /// to consume.
+  bool sender_done_ = false;
+
+  /// The current result set passed to GetNext(), to fill in Send(). Not owned 
by this
+  /// sink. Reset to nullptr after Send() completes the request to signal to 
the consumer
+  /// that it can return.
+  QueryResultSet* results_ = nullptr;
+
+  /// Set by GetNext() to indicate to Send() how many rows it should write to 
results_.
+  int num_rows_requested_ = 0;
+
+  /// Set to true in Send() and FlushFinal() when the Sink() has finished 
producing rows.
+  bool eos_ = false;
+
+  /// Output expressions to map plan row batches onto result set rows.
+  std::vector<TExpr> thrift_output_exprs_;
+  std::vector<ExprContext*> output_expr_ctxs_;
+
+  /// Writes a single row into 'result' and 'scales' by evaluating 
output_expr_ctxs_ over
+  /// 'row'.
+  void GetRowValue(TupleRow* row, std::vector<void*>* result, 
std::vector<int>* scales);
+};
+}
+
+#endif

http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/9f61397f/be/src/runtime/coordinator.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/coordinator.cc b/be/src/runtime/coordinator.cc
index df4ad7b..4214d4d 100644
--- a/be/src/runtime/coordinator.cc
+++ b/be/src/runtime/coordinator.cc
@@ -39,22 +39,28 @@
 #include <errno.h>
 
 #include "common/logging.h"
-#include "exprs/expr.h"
 #include "exec/data-sink.h"
+#include "exec/plan-root-sink.h"
+#include "exec/scan-node.h"
+#include "gen-cpp/Frontend_types.h"
+#include "gen-cpp/ImpalaInternalService.h"
+#include "gen-cpp/ImpalaInternalService_constants.h"
+#include "gen-cpp/ImpalaInternalService_types.h"
+#include "gen-cpp/Partitions_types.h"
+#include "gen-cpp/PlanNodes_types.h"
+#include "runtime/backend-client.h"
 #include "runtime/client-cache.h"
-#include "runtime/data-stream-sender.h"
 #include "runtime/data-stream-mgr.h"
+#include "runtime/data-stream-sender.h"
 #include "runtime/exec-env.h"
 #include "runtime/hdfs-fs-cache.h"
 #include "runtime/mem-tracker.h"
+#include "runtime/parallel-executor.h"
 #include "runtime/plan-fragment-executor.h"
 #include "runtime/row-batch.h"
-#include "runtime/backend-client.h"
-#include "runtime/parallel-executor.h"
 #include "runtime/tuple-row.h"
 #include "scheduling/scheduler.h"
-#include "exec/data-sink.h"
-#include "exec/scan-node.h"
+#include "service/fragment-exec-state.h"
 #include "util/bloom-filter.h"
 #include "util/container-util.h"
 #include "util/counting-barrier.h"
@@ -67,12 +73,6 @@
 #include "util/summary-util.h"
 #include "util/table-printer.h"
 #include "util/uid-util.h"
-#include "gen-cpp/ImpalaInternalService.h"
-#include "gen-cpp/ImpalaInternalService_types.h"
-#include "gen-cpp/Frontend_types.h"
-#include "gen-cpp/PlanNodes_types.h"
-#include "gen-cpp/Partitions_types.h"
-#include "gen-cpp/ImpalaInternalService_constants.h"
 
 #include "common/names.h"
 
@@ -240,8 +240,8 @@ class Coordinator::FragmentInstanceState {
   mutex lock_;
 
   /// If the status indicates an error status, execution of this fragment has 
either been
-  /// aborted by the remote impalad (which then reported the error) or 
cancellation has
-  /// been initiated; either way, execution must not be cancelled
+  /// aborted by the executing impalad (which then reported the error) or 
cancellation has
+  /// been initiated; either way, execution must not be cancelled.
   Status status_;
 
   /// If true, ExecPlanFragment() rpc has been sent.
@@ -377,15 +377,13 @@ Coordinator::Coordinator(const QuerySchedule& schedule, 
ExecEnv* exec_env,
     exec_env_(exec_env),
     has_called_wait_(false),
     returned_all_results_(false),
-    executor_(NULL), // Set in Prepare()
     query_mem_tracker_(), // Set in Exec()
     num_remaining_fragment_instances_(0),
     obj_pool_(new ObjectPool()),
     query_events_(events),
     filter_routing_table_complete_(false),
-    filter_mode_(schedule.query_options().runtime_filter_mode),
-    torn_down_(false) {
-}
+    filter_mode_(schedule_.query_options().runtime_filter_mode),
+    torn_down_(false) {}
 
 Coordinator::~Coordinator() {
   DCHECK(torn_down_) << "TearDown() must be called before Coordinator is 
destroyed";
@@ -448,14 +446,15 @@ static void ProcessQueryOptions(
       << "because nodes cannot be cancelled in Close()";
 }
 
-Status Coordinator::Exec(vector<ExprContext*>* output_expr_ctxs) {
+Status Coordinator::Exec() {
   const TQueryExecRequest& request = schedule_.request();
   DCHECK(request.fragments.size() > 0 || request.mt_plan_exec_info.size() > 0);
+
   needs_finalization_ = request.__isset.finalize_params;
   if (needs_finalization_) finalize_params_ = request.finalize_params;
 
   VLOG_QUERY << "Exec() query_id=" << schedule_.query_id()
-      << " stmt=" << request.query_ctx.request.stmt;
+             << " stmt=" << request.query_ctx.request.stmt;
   stmt_type_ = request.stmt_type;
   query_id_ = schedule_.query_id();
   desc_tbl_ = request.desc_tbl;
@@ -468,18 +467,6 @@ Status Coordinator::Exec(vector<ExprContext*>* 
output_expr_ctxs) {
 
   SCOPED_TIMER(query_profile_->total_time_counter());
 
-  // After the coordinator fragment is started, it may call UpdateFilter() 
asynchronously,
-  // which waits on this barrier for completion.
-  // TODO: remove special treatment of coord fragment
-  int num_remote_instances = schedule_.GetNumRemoteFInstances();
-  if (num_remote_instances > 0) {
-    exec_complete_barrier_.reset(new CountingBarrier(num_remote_instances));
-  }
-  num_remaining_fragment_instances_ = num_remote_instances;
-
-  // TODO: move initial setup into a separate function; right now part of it
-  // (InitExecProfile()) depends on the coordinator fragment having been 
started
-
   // initialize progress updater
   const string& str = Substitute("Query $0", PrintId(query_id_));
   progress_.Init(str, schedule_.num_scan_ranges());
@@ -489,64 +476,51 @@ Status Coordinator::Exec(vector<ExprContext*>* 
output_expr_ctxs) {
   // execution at Impala daemons where it hasn't even started
   lock_guard<mutex> l(lock_);
 
-  bool has_coordinator_fragment = schedule_.GetCoordFragment() != NULL;
-  if (has_coordinator_fragment) {
-    // Start this before starting any more plan
-    // fragments, otherwise they start sending data before the local exchange 
node had a
-    // chance to register with the stream mgr.
-    // TODO: This is no longer necessary (see IMPALA-1599). Consider starting 
all
-    // fragments in the same way with no coordinator special case.
-    RETURN_IF_ERROR(PrepareCoordFragment(output_expr_ctxs));
-  } else {
-    // The coordinator instance may require a query mem tracker even if there 
is no
-    // coordinator fragment. For example, result-caching tracks memory via the 
query mem
-    // tracker.
-    // If there is a fragment, the fragment executor created above initializes 
the query
-    // mem tracker. If not, the query mem tracker is created here.
-    int64_t query_limit = -1;
-    if (query_ctx_.request.query_options.__isset.mem_limit &&
-        query_ctx_.request.query_options.mem_limit > 0) {
-      query_limit = query_ctx_.request.query_options.mem_limit;
-    }
-    MemTracker* pool_tracker = MemTracker::GetRequestPoolMemTracker(
-        schedule_.request_pool(), exec_env_->process_mem_tracker());
-    query_mem_tracker_ =
-        MemTracker::GetQueryMemTracker(query_id_, query_limit, pool_tracker);
-
-    executor_.reset(NULL);
-  }
+  // The coordinator may require a query mem tracker for result-caching, which 
tracks
+  // memory via the query mem tracker.
+  int64_t query_limit = -1;
+  if (query_ctx_.request.query_options.__isset.mem_limit
+      && query_ctx_.request.query_options.mem_limit > 0) {
+    query_limit = query_ctx_.request.query_options.mem_limit;
+  }
+  MemTracker* pool_tracker = MemTracker::GetRequestPoolMemTracker(
+      schedule_.request_pool(), exec_env_->process_mem_tracker());
+  query_mem_tracker_ =
+      MemTracker::GetQueryMemTracker(query_id_, query_limit, pool_tracker);
+  DCHECK(query_mem_tracker() != nullptr);
   filter_mem_tracker_.reset(
       new MemTracker(-1, "Runtime Filter (Coordinator)", query_mem_tracker(), 
false));
 
-  // initialize execution profile structures
+  // Initialize the execution profile structures.
   bool is_mt_execution = request.query_ctx.request.query_options.mt_dop > 0;
   if (is_mt_execution) {
     MtInitExecProfiles();
     MtInitExecSummary();
+    MtStartFInstances();
   } else {
     InitExecProfile(request);
+    StartFragments();
   }
 
-  if (num_remote_instances > 0) {
-    // pre-size fragment_instance_states_ in order to directly address by 
instance idx
-    // when creating FragmentInstanceStates (instead of push_back())
-    int num_fragment_instances = schedule_.GetTotalFInstances();
-    DCHECK_GT(num_fragment_instances, 0);
-    fragment_instance_states_.resize(num_fragment_instances);
+  RETURN_IF_ERROR(FinishInstanceStartup());
 
-    if (is_mt_execution) {
-      MtStartRemoteFInstances();
-    } else {
-      StartRemoteFragments();
-    }
-    RETURN_IF_ERROR(FinishRemoteInstanceStartup());
+  // Grab executor and wait until Prepare() has finished so that runtime state 
etc. will
+  // be set up.
+  if (schedule_.GetCoordFragment() != nullptr) {
+    // Coordinator fragment instance has same ID as query.
+    shared_ptr<FragmentMgr::FragmentExecState> root_fragment_instance =
+        
ExecEnv::GetInstance()->fragment_mgr()->GetFragmentExecState(query_id_);
+    DCHECK(root_fragment_instance.get() != nullptr);
+    executor_ = root_fragment_instance->executor();
 
-    // If we have a coordinator fragment and remote fragments (the common 
case), release
-    // the thread token on the coordinator fragment. This fragment spends most 
of the time
-    // waiting and doing very little work. Holding on to the token causes 
underutilization
-    // of the machine. If there are 12 queries on this node, that's 12 tokens 
reserved for
-    // no reason.
-    if (has_coordinator_fragment) executor_->ReleaseThreadToken();
+    // When WaitForPrepare() returns OK(), the executor's root sink will be 
set up. At
+    // that point, the coordinator must be sure to call 
root_sink()->CloseConsumer(); the
+    // fragment instance's executor will not complete until that point.
+    // TODO: Consider moving this to Wait().
+    Status prepare_status = executor_->WaitForPrepare();
+    root_sink_ = executor_->root_sink();
+    RETURN_IF_ERROR(prepare_status);
+    DCHECK(root_sink_ != nullptr);
   }
 
   PrintFragmentInstanceInfo();
@@ -610,60 +584,14 @@ void Coordinator::UpdateFilterRoutingTable(const 
vector<TPlanNode>& plan_nodes,
   }
 }
 
-Status Coordinator::PrepareCoordFragment(vector<ExprContext*>* 
output_expr_ctxs) {
-  const TQueryExecRequest& request = schedule_.request();
-  bool is_mt_execution = request.query_ctx.request.query_options.mt_dop > 0;
-  if (!is_mt_execution && filter_mode_ != TRuntimeFilterMode::OFF) {
-    UpdateFilterRoutingTable(schedule_.GetCoordFragment()->plan.nodes, 1, 0);
-    if (schedule_.GetNumFragmentInstances() == 0) 
MarkFilterRoutingTableComplete();
-  }
+void Coordinator::StartFragments() {
+  int num_fragment_instances = schedule_.GetNumFragmentInstances();
+  DCHECK_GT(num_fragment_instances, 0);
 
-  // create rpc params and FragmentInstanceState for the coordinator fragment
-  TExecPlanFragmentParams rpc_params;
-  FragmentInstanceState* coord_state = nullptr;
-  if (is_mt_execution) {
-    const FInstanceExecParams& coord_params = 
schedule_.GetCoordInstanceExecParams();
-    MtSetExecPlanFragmentParams(coord_params, &rpc_params);
-    coord_state = obj_pool()->Add(
-        new FragmentInstanceState(coord_params, obj_pool()));
-  } else {
-    const TPlanFragment& coord_fragment = *schedule_.GetCoordFragment();
-    SetExecPlanFragmentParams(
-        coord_fragment, schedule_.exec_params()[0], 0, &rpc_params);
-    coord_state = obj_pool()->Add(
-        new FragmentInstanceState(
-          coord_fragment.idx, schedule_.exec_params()[0], 0, obj_pool()));
-    // apparently this was never called for the coordinator fragment
-    // TODO: fix this
-    //exec_state->ComputeTotalSplitSize(
-        //rpc_params.fragment_instance_ctx.per_node_scan_ranges);
-  }
-  // register state before calling Prepare(), in case it fails
-  DCHECK_EQ(GetInstanceIdx(coord_state->fragment_instance_id()), 0);
-  fragment_instance_states_.push_back(coord_state);
-  DCHECK(coord_state != nullptr);
-  DCHECK_EQ(fragment_instance_states_.size(), 1);
-  executor_.reset(new PlanFragmentExecutor(
-      exec_env_, PlanFragmentExecutor::ReportStatusCallback()));
-  RETURN_IF_ERROR(executor_->Prepare(rpc_params));
-  coord_state->set_profile(executor_->profile());
-
-  // Prepare output_expr_ctxs before optimizing the LLVM module. The other 
exprs of this
-  // coordinator fragment have been prepared in executor_->Prepare().
-  DCHECK(output_expr_ctxs != NULL);
-  RETURN_IF_ERROR(Expr::CreateExprTrees(
-      runtime_state()->obj_pool(), schedule_.GetCoordFragment()->output_exprs,
-      output_expr_ctxs));
-  MemTracker* output_expr_tracker = runtime_state()->obj_pool()->Add(new 
MemTracker(
-      -1, "Output exprs", runtime_state()->instance_mem_tracker(), false));
-  RETURN_IF_ERROR(Expr::Prepare(
-      *output_expr_ctxs, runtime_state(), row_desc(), output_expr_tracker));
+  fragment_instance_states_.resize(num_fragment_instances);
+  exec_complete_barrier_.reset(new CountingBarrier(num_fragment_instances));
+  num_remaining_fragment_instances_ = num_fragment_instances;
 
-  return Status::OK();
-}
-
-void Coordinator::StartRemoteFragments() {
-  int num_fragment_instances = schedule_.GetNumFragmentInstances();
   DebugOptions debug_options;
   ProcessQueryOptions(schedule_.query_options(), &debug_options);
   const TQueryExecRequest& request = schedule_.request();
@@ -671,19 +599,14 @@ void Coordinator::StartRemoteFragments() {
   VLOG_QUERY << "starting " << num_fragment_instances << " fragment instances 
for query "
              << query_id_;
   query_events_->MarkEvent(
-      Substitute("Ready to start $0 remote fragment instances", 
num_fragment_instances));
+      Substitute("Ready to start $0 fragments", num_fragment_instances));
 
-  bool has_coordinator_fragment =
-      request.fragments[0].partition.type == TPartitionType::UNPARTITIONED;
-  int instance_state_idx = has_coordinator_fragment ? 1 : 0;
-  int first_remote_fragment_idx = has_coordinator_fragment ? 1 : 0;
+  int instance_state_idx = 0;
   if (filter_mode_ != TRuntimeFilterMode::OFF) {
-    // Populate the runtime filter routing table. This should happen before 
starting
-    // the remote fragments.
-    // This code anticipates the indices of the instance states created later 
on in
-    // ExecRemoteFragment()
-    for (int fragment_idx = first_remote_fragment_idx;
-         fragment_idx < request.fragments.size(); ++fragment_idx) {
+    // Populate the runtime filter routing table. This should happen before 
starting the
+    // fragment instances. This code anticipates the indices of the instance 
states
+    // created later on in ExecRemoteFragment()
+    for (int fragment_idx = 0; fragment_idx < request.fragments.size(); 
++fragment_idx) {
       const FragmentExecParams& params = schedule_.exec_params()[fragment_idx];
       int num_hosts = params.hosts.size();
       DCHECK_GT(num_hosts, 0);
@@ -697,8 +620,7 @@ void Coordinator::StartRemoteFragments() {
   int num_instances = 0;
   // Start one fragment instance per fragment per host (number of hosts 
running each
   // fragment may not be constant).
-  for (int fragment_idx = first_remote_fragment_idx;
-       fragment_idx < request.fragments.size(); ++fragment_idx) {
+  for (int fragment_idx = 0; fragment_idx < request.fragments.size(); 
++fragment_idx) {
     const FragmentExecParams& params = schedule_.exec_params()[fragment_idx];
     int num_hosts = params.hosts.size();
     DCHECK_GT(num_hosts, 0);
@@ -719,11 +641,17 @@ void Coordinator::StartRemoteFragments() {
   }
   exec_complete_barrier_->Wait();
   query_events_->MarkEvent(
-      Substitute("All $0 remote fragments instances started", num_instances));
+      Substitute("All $0 fragments instances started", num_instances));
 }
 
-void Coordinator::MtStartRemoteFInstances() {
+void Coordinator::MtStartFInstances() {
   int num_fragment_instances = schedule_.GetNumFragmentInstances();
+  DCHECK_GT(num_fragment_instances, 0);
+
+  fragment_instance_states_.resize(num_fragment_instances);
+  exec_complete_barrier_.reset(new CountingBarrier(num_fragment_instances));
+  num_remaining_fragment_instances_ = num_fragment_instances;
+
   DebugOptions debug_options;
   ProcessQueryOptions(schedule_.query_options(), &debug_options);
   const TQueryExecRequest& request = schedule_.request();
@@ -731,7 +659,7 @@ void Coordinator::MtStartRemoteFInstances() {
   VLOG_QUERY << "starting " << num_fragment_instances << " fragment instances 
for query "
              << query_id_;
   query_events_->MarkEvent(
-      Substitute("Ready to start $0 remote fragment instances", 
num_fragment_instances));
+      Substitute("Ready to start $0 fragment instances", 
num_fragment_instances));
 
   // TODO: populate the runtime filter routing table
   // this requires local aggregation of filters prior to sending
@@ -739,7 +667,6 @@ void Coordinator::MtStartRemoteFInstances() {
 
   int num_instances = 0;
   for (const MtFragmentExecParams& fragment_params: 
schedule_.mt_fragment_exec_params()) {
-    if (fragment_params.is_coord_fragment) continue;
     for (int i = 0; i < fragment_params.instance_exec_params.size();
         ++i, ++num_instances) {
       const FInstanceExecParams& instance_params =
@@ -760,10 +687,10 @@ void Coordinator::MtStartRemoteFInstances() {
   VLOG_QUERY << "started " << num_fragment_instances << " fragment instances 
for query "
       << query_id_;
   query_events_->MarkEvent(
-      Substitute("All $0 remote fragment instances started", num_instances));
+      Substitute("All $0 fragment instances started", num_instances));
 }
 
-Status Coordinator::FinishRemoteInstanceStartup() {
+Status Coordinator::FinishInstanceStartup() {
   Status status = Status::OK();
   const TMetricDef& def =
       MakeTMetricDef("fragment-latencies", TMetricKind::HISTOGRAM, 
TUnit::TIME_MS);
@@ -868,7 +795,7 @@ Status Coordinator::UpdateStatus(const Status& status, 
const TUniqueId& instance
   {
     lock_guard<mutex> l(lock_);
 
-    // The query is done and we are just waiting for remote fragments to clean 
up.
+    // The query is done and we are just waiting for fragment instances to 
clean up.
     // Ignore their cancelled updates.
     if (returned_all_results_ && status.IsCancelled()) return query_status_;
 
@@ -1187,7 +1114,8 @@ Status Coordinator::WaitForAllInstances() {
   if (query_status_.ok()) {
     VLOG_QUERY << "All fragment instances finished successfully.";
   } else {
-    VLOG_QUERY << "All fragment instances finished due to one or more errors.";
+    VLOG_QUERY << "All fragment instances finished due to one or more errors. "
+               << query_status_.GetDetail();
   }
 
   return query_status_;
@@ -1198,143 +1126,84 @@ Status Coordinator::Wait() {
   SCOPED_TIMER(query_profile_->total_time_counter());
   if (has_called_wait_) return Status::OK();
   has_called_wait_ = true;
-  Status return_status = Status::OK();
-  if (executor_.get() != NULL) {
-    // Open() may block
-    return_status = UpdateStatus(executor_->Open(),
-        runtime_state()->fragment_instance_id(), FLAGS_hostname);
-
-    if (return_status.ok()) {
-      // If the coordinator fragment has a sink, it will have finished 
executing at this
-      // point.  It's safe therefore to copy the set of files to move and 
updated
-      // partitions into the query-wide set.
-      RuntimeState* state = runtime_state();
-      DCHECK(state != NULL);
-
-      // No other instances should have updated these structures if the 
coordinator has a
-      // fragment. (Instances have a sink only if the coordinator does not)
-      DCHECK_EQ(files_to_move_.size(), 0);
-      DCHECK_EQ(per_partition_status_.size(), 0);
-
-      // Because there are no other updates, safe to copy the maps rather than 
merge them.
-      files_to_move_ = *state->hdfs_files_to_move();
-      per_partition_status_ = *state->per_partition_status();
-    }
-  } else {
-    // Query finalization can only happen when all instances have reported
-    // relevant state. They only have relevant state to report in the parallel
-    // INSERT case, otherwise all the relevant state is from the coordinator
-    // fragment which will be available after Open() returns.
-    // Ignore the returned status if finalization is required., since 
FinalizeQuery() will
-    // pick it up and needs to execute regardless.
-    Status status = WaitForAllInstances();
-    if (!needs_finalization_ && !status.ok()) return status;
-  }
 
-  // Query finalization is required only for HDFS table sinks
-  if (needs_finalization_) {
-    RETURN_IF_ERROR(FinalizeQuery());
+  if (stmt_type_ == TStmtType::QUERY) {
+    DCHECK(executor_ != nullptr);
+    return UpdateStatus(executor_->WaitForOpen(), 
runtime_state()->fragment_instance_id(),
+        FLAGS_hostname);
   }
 
-  if (stmt_type_ == TStmtType::DML) {
-    query_profile_->AddInfoString("Insert Stats",
-        DataSink::OutputInsertStats(per_partition_status_, "\n"));
-    // For DML queries, when Wait is done, the query is complete.  Report 
aggregate
-    // query profiles at this point.
-    // TODO: make sure ReportQuerySummary gets called on error
-    ReportQuerySummary();
-  }
+  DCHECK_EQ(stmt_type_, TStmtType::DML);
+  // Query finalization can only happen when all backends have reported
+  // relevant state. They only have relevant state to report in the parallel
+  // INSERT case, otherwise all the relevant state is from the coordinator
+  // fragment which will be available after Open() returns.
+  // Ignore the returned status if finalization is required., since 
FinalizeQuery() will
+  // pick it up and needs to execute regardless.
+  Status status = WaitForAllInstances();
+  if (!needs_finalization_ && !status.ok()) return status;
 
-  if (filter_routing_table_.size() > 0) {
-    query_profile_->AddInfoString("Final filter table", FilterDebugString());
-  }
+  // Query finalization is required only for HDFS table sinks
+  if (needs_finalization_) RETURN_IF_ERROR(FinalizeQuery());
 
-  return return_status;
+  query_profile_->AddInfoString(
+      "Insert Stats", DataSink::OutputInsertStats(per_partition_status_, 
"\n"));
+  // For DML queries, when Wait is done, the query is complete.  Report 
aggregate
+  // query profiles at this point.
+  // TODO: make sure ReportQuerySummary gets called on error
+  ReportQuerySummary();
+
+  return status;
 }
 
-Status Coordinator::GetNext(RowBatch** batch, RuntimeState* state) {
+Status Coordinator::GetNext(QueryResultSet* results, int max_rows, bool* eos) {
   VLOG_ROW << "GetNext() query_id=" << query_id_;
   DCHECK(has_called_wait_);
   SCOPED_TIMER(query_profile_->total_time_counter());
 
-  if (executor_.get() == NULL) {
-    // If there is no local fragment, we produce no output, and execution will
-    // have finished after Wait.
-    *batch = NULL;
-    return GetStatus();
-  }
-
-  // do not acquire lock_ here, otherwise we could block and prevent an async
-  // Cancel() from proceeding
-  Status status = executor_->GetNext(batch);
+  DCHECK(root_sink_ != nullptr)
+      << "GetNext() called without result sink. Perhaps Prepare() failed and 
was not "
+      << "checked?";
+  Status status = root_sink_->GetNext(runtime_state(), results, max_rows, eos);
 
   // if there was an error, we need to return the query's error status rather 
than
   // the status we just got back from the local executor (which may well be 
CANCELLED
   // in that case).  Coordinator fragment failed in this case so we log the 
query_id.
-  RETURN_IF_ERROR(UpdateStatus(status, runtime_state()->fragment_instance_id(),
-      FLAGS_hostname));
+  RETURN_IF_ERROR(
+      UpdateStatus(status, runtime_state()->fragment_instance_id(), 
FLAGS_hostname));
 
-  if (*batch == NULL) {
+  if (*eos) {
     returned_all_results_ = true;
-    if (executor_->ReachedLimit()) {
-      // We've reached the query limit, cancel the remote fragments.  The
-      // Exchange node on our fragment is no longer receiving rows so the
-      // remote fragments must be explicitly cancelled.
-      CancelRemoteFragments();
-      RuntimeState* state = runtime_state();
-      if (state != NULL) {
-        // Cancel the streams receiving batches.  The exchange nodes that would
-        // normally read from the streams are done.
-        state->stream_mgr()->Cancel(state->fragment_instance_id());
-      }
-    }
-
-    // Don't return final NULL until all instances have completed.
-    // GetNext must wait for all instances to complete before
-    // ultimately signalling the end of execution via a NULL
-    // batch. After NULL is returned, the coordinator may tear down
-    // query state, and perform post-query finalization which might
-    // depend on the reports from all instances.
+    // Trigger tear-down of coordinator fragment by closing the consumer. Must 
do before
+    // WaitForAllInstances().
+    root_sink_->CloseConsumer();
+    root_sink_ = nullptr;
+
+    // Don't return final NULL until all instances have completed.  GetNext 
must wait for
+    // all instances to complete before ultimately signalling the end of 
execution via a
+    // NULL batch. After NULL is returned, the coordinator may tear down query 
state, and
+    // perform post-query finalization which might depend on the reports from 
all
+    // instances.
+    //
+    // TODO: Waiting should happen in TearDown() (and then we wouldn't need to 
call
+    // CloseConsumer() here). See IMPALA-4275 for details.
     RETURN_IF_ERROR(WaitForAllInstances());
     if (query_status_.ok()) {
       // If the query completed successfully, report aggregate query profiles.
       ReportQuerySummary();
     }
-  } else {
-#ifndef NDEBUG
-    ValidateCollectionSlots(*batch);
-#endif
   }
 
   return Status::OK();
 }
 
-void Coordinator::ValidateCollectionSlots(RowBatch* batch) {
-  const RowDescriptor& row_desc = executor_->row_desc();
-  if (!row_desc.HasVarlenSlots()) return;
-  for (int i = 0; i < batch->num_rows(); ++i) {
-    TupleRow* row = batch->GetRow(i);
-    for (int j = 0; j < row_desc.tuple_descriptors().size(); ++j) {
-      const TupleDescriptor* tuple_desc = row_desc.tuple_descriptors()[j];
-      if (tuple_desc->collection_slots().empty()) continue;
-      for (int k = 0; k < tuple_desc->collection_slots().size(); ++k) {
-        const SlotDescriptor* slot_desc = tuple_desc->collection_slots()[k];
-        int tuple_idx = row_desc.GetTupleIdx(slot_desc->parent()->id());
-        const Tuple* tuple = row->GetTuple(tuple_idx);
-        if (tuple == NULL) continue;
-        DCHECK(tuple->IsNull(slot_desc->null_indicator_offset()));
-      }
-    }
-  }
-}
-
 void Coordinator::PrintFragmentInstanceInfo() {
   for (FragmentInstanceState* state: fragment_instance_states_) {
     SummaryStats& acc = 
fragment_profiles_[state->fragment_idx()].bytes_assigned;
     acc(state->total_split_size());
   }
 
-  for (int id = (executor_.get() == NULL ? 0 : 1); id < 
fragment_profiles_.size(); ++id) {
+  for (int id = (executor_ == NULL ? 0 : 1); id < fragment_profiles_.size(); 
++id) {
     SummaryStats& acc = fragment_profiles_[id].bytes_assigned;
     double min = accumulators::min(acc);
     double max = accumulators::max(acc);
@@ -1360,6 +1229,7 @@ void Coordinator::PrintFragmentInstanceInfo() {
 
 void Coordinator::InitExecProfile(const TQueryExecRequest& request) {
   // Initialize the structure to collect execution summary of every plan node.
+  fragment_profiles_.resize(request.fragments.size());
   exec_summary_.__isset.nodes = true;
   for (int i = 0; i < request.fragments.size(); ++i) {
     if (!request.fragments[i].__isset.plan) continue;
@@ -1394,46 +1264,30 @@ void Coordinator::InitExecProfile(const 
TQueryExecRequest& request) {
     }
   }
 
-  if (executor_.get() != NULL) {
-    // register coordinator's fragment profile now, before those of the 
backends,
-    // so it shows up at the top
-    query_profile_->AddChild(executor_->profile());
-    executor_->profile()->set_name(Substitute("Coordinator Fragment $0",
-        request.fragments[0].display_name));
-    CollectScanNodeCounters(executor_->profile(), &coordinator_counters_);
-  }
-
   // Initialize the runtime profile structure. This adds the per fragment 
average
   // profiles followed by the per fragment instance profiles.
-  bool has_coordinator_fragment =
-      request.fragments[0].partition.type == TPartitionType::UNPARTITIONED;
-  fragment_profiles_.resize(request.fragments.size());
   for (int i = 0; i < request.fragments.size(); ++i) {
-    fragment_profiles_[i].num_instances = 0;
-
-    // Special case fragment idx 0 if there is a coordinator. There is only one
-    // instance of this profile so the average is just the coordinator profile.
-    if (i == 0 && has_coordinator_fragment) {
-      fragment_profiles_[i].averaged_profile = executor_->profile();
-      fragment_profiles_[i].num_instances = 1;
-      continue;
-    }
-    fragment_profiles_[i].averaged_profile =
-        obj_pool()->Add(new RuntimeProfile(obj_pool(),
-            Substitute("Averaged Fragment $0", 
request.fragments[i].display_name), true));
-    // Insert the avg profiles in ascending fragment number order. If
-    // there is a coordinator fragment, it's been placed in
-    // fragment_profiles_[0].averaged_profile, ensuring that this code
-    // will put the first averaged profile immediately after it. If
-    // there is no coordinator fragment, the first averaged profile
-    // will be inserted as the first child of query_profile_, and then
-    // all other averaged fragments will follow.
-    query_profile_->AddChild(fragment_profiles_[i].averaged_profile, true,
-        (i > 0) ? fragment_profiles_[i-1].averaged_profile : NULL);
-
+    // Insert the avg profiles in ascending fragment number order. If there is 
a
+    // coordinator fragment, it's been placed in 
fragment_profiles_[0].averaged_profile,
+    // ensuring that this code will put the first averaged profile immediately 
after
+    // it. If there is no coordinator fragment, the first averaged profile 
will be
+    // inserted as the first child of query_profile_, and then all other 
averaged
+    // fragments will follow.
+    bool is_coordinator_fragment = (i == 0 && schedule_.GetCoordFragment() != 
nullptr);
+    string profile_name =
+        Substitute(is_coordinator_fragment ? "Coordinator Fragment $0" : 
"Fragment $0",
+            request.fragments[i].display_name);
     fragment_profiles_[i].root_profile =
-        obj_pool()->Add(new RuntimeProfile(obj_pool(),
-            Substitute("Fragment $0", request.fragments[i].display_name)));
+        obj_pool()->Add(new RuntimeProfile(obj_pool(), profile_name));
+    if (is_coordinator_fragment) {
+      fragment_profiles_[i].averaged_profile = nullptr;
+    } else {
+      fragment_profiles_[i].averaged_profile = obj_pool()->Add(new 
RuntimeProfile(
+          obj_pool(),
+          Substitute("Averaged Fragment $0", 
request.fragments[i].display_name), true));
+      query_profile_->AddChild(fragment_profiles_[i].averaged_profile, true,
+          (i > 0) ? fragment_profiles_[i - 1].averaged_profile : NULL);
+    }
     // Note: we don't start the wall timer here for the fragment
     // profile; it's uninteresting and misleading.
     query_profile_->AddChild(fragment_profiles_[i].root_profile);
@@ -1490,40 +1344,23 @@ void Coordinator::MtInitExecProfiles() {
   schedule_.GetTPlanFragments(&fragments);
   fragment_profiles_.resize(fragments.size());
 
-  // start with coordinator fragment, if there is one
   const TPlanFragment* coord_fragment = schedule_.GetCoordFragment();
-  if (coord_fragment != NULL) {
-    DCHECK(executor_.get() != NULL);
-    PerFragmentProfileData* data = &fragment_profiles_[coord_fragment->idx];
-    data->num_instances = 1;
-    // TODO: fix this; this is not an averaged profile; we should follow the 
exact
-    // same structure we have for all other profiles (average + root + single
-    // instance profile)
-    data->averaged_profile = executor_->profile();
-
-    // register coordinator's fragment profile in the query profile now, 
before those
-    // of the fragment instances, so it shows up at the top
-    query_profile_->AddChild(executor_->profile());
-    executor_->profile()->set_name(Substitute("Coordinator Fragment $0",
-        coord_fragment->display_name));
-    CollectScanNodeCounters(executor_->profile(), &coordinator_counters_);
-  }
 
   // Initialize the runtime profile structure. This adds the per fragment 
average
   // profiles followed by the per fragment instance profiles.
   for (const TPlanFragment* fragment: fragments) {
-    if (fragment == coord_fragment) continue;
+    string profile_name =
+        (fragment == coord_fragment) ? "Coordinator Fragment $0" : "Fragment 
$0";
     PerFragmentProfileData* data = &fragment_profiles_[fragment->idx];
     data->num_instances =
         
schedule_.GetFragmentExecParams(fragment->idx).instance_exec_params.size();
-
-    data->averaged_profile =
-        obj_pool()->Add(new RuntimeProfile(obj_pool(),
-          Substitute("Averaged Fragment $0", fragment->display_name), true));
-    query_profile_->AddChild(data->averaged_profile, true);
-    data->root_profile =
-        obj_pool()->Add(new RuntimeProfile(obj_pool(),
-          Substitute("Fragment $0", fragment->display_name)));
+    if (fragment != coord_fragment) {
+      data->averaged_profile = obj_pool()->Add(new RuntimeProfile(
+          obj_pool(), Substitute("Averaged Fragment $0", 
fragment->display_name), true));
+      query_profile_->AddChild(data->averaged_profile, true);
+    }
+    data->root_profile = obj_pool()->Add(
+        new RuntimeProfile(obj_pool(), Substitute(profile_name, 
fragment->display_name)));
     // Note: we don't start the wall timer here for the fragment profile;
     // it's uninteresting and misleading.
     query_profile_->AddChild(data->root_profile);
@@ -1637,8 +1474,6 @@ void Coordinator::ExecRemoteFragment(const 
FragmentExecParams& fragment_exec_par
             << " instance_id=" << PrintId(exec_state->fragment_instance_id())
             << " host=" << exec_state->impalad_address();
 
-  // Guard against concurrent UpdateExecStatus() that may arrive after RPC 
returns.
-  lock_guard<mutex> l(*exec_state->lock());
   int64_t start = MonotonicMillis();
 
   Status client_connect_status;
@@ -1685,32 +1520,27 @@ void Coordinator::Cancel(const Status* cause) {
   // if the query status indicates an error, cancellation has already been 
initiated
   if (!query_status_.ok()) return;
   // prevent others from cancelling a second time
+
+  // TODO: This should default to OK(), not CANCELLED if there is no cause (or 
callers
+  // should explicitly pass Status::OK()). Fragment instances may be cancelled 
at the end
+  // of a successful query. Need to clean up relationship between 
query_status_ here and
+  // in QueryExecState. See IMPALA-4279.
   query_status_ = (cause != NULL && !cause->ok()) ? *cause : Status::CANCELLED;
   CancelInternal();
 }
 
 void Coordinator::CancelInternal() {
   VLOG_QUERY << "Cancel() query_id=" << query_id_;
-  DCHECK(!query_status_.ok());
-
-  // cancel local fragment
-  if (executor_.get() != NULL) executor_->Cancel();
-
-  CancelRemoteFragments();
+  CancelFragmentInstances();
 
   // Report the summary with whatever progress the query made before being 
cancelled.
   ReportQuerySummary();
 }
 
-void Coordinator::CancelRemoteFragments() {
+void Coordinator::CancelFragmentInstances() {
+  int num_cancelled = 0;
   for (FragmentInstanceState* exec_state: fragment_instance_states_) {
     DCHECK(exec_state != nullptr);
-    if (exec_state->fragment_idx() == 0) continue;  // the coord fragment
-
-    // If a fragment failed before we finished issuing all remote fragments,
-    // this function will have been called before we finished populating
-    // fragment_instance_states_. Skip any such uninitialized exec states.
-    if (exec_state == NULL) continue;
 
     // lock each exec_state individually to synchronize correctly with
     // UpdateFragmentExecStatus() (which doesn't get the global lock_
@@ -1735,7 +1565,7 @@ void Coordinator::CancelRemoteFragments() {
     ImpalaBackendConnection backend_client(
         exec_env_->impalad_client_cache(), exec_state->impalad_address(), 
&status);
     if (!status.ok()) continue;
-
+    ++num_cancelled;
     TCancelPlanFragmentParams params;
     params.protocol_version = ImpalaInternalServiceVersion::V1;
     params.__set_fragment_instance_id(exec_state->fragment_instance_id());
@@ -1765,8 +1595,11 @@ void Coordinator::CancelRemoteFragments() {
       exec_state->status()->AddDetail(join(res.status.error_msgs, "; "));
     }
   }
+  VLOG_QUERY << Substitute(
+      "CancelFragmentInstances() query_id=$0, tried to cancel $1 fragment 
instances",
+      PrintId(query_id_), num_cancelled);
 
-  // notify that we completed with an error
+  // Notify that we completed with an error.
   instance_completion_cv_.notify_all();
 }
 
@@ -1799,11 +1632,11 @@ Status Coordinator::UpdateFragmentExecStatus(const 
TReportExecStatusParams& para
       // We can't update this backend's profile if ReportQuerySummary() is 
running,
       // because it depends on all profiles not changing during its execution 
(when it
       // calls SortChildren()). ReportQuerySummary() only gets called after
-      // WaitForAllInstances() returns or at the end of 
CancelRemoteFragments().
+      // WaitForAllInstances() returns or at the end of 
CancelFragmentInstances().
       // WaitForAllInstances() only returns after all backends have completed 
(in which
       // case we wouldn't be in this function), or when there's an error, in 
which case
-      // CancelRemoteFragments() is called. CancelRemoteFragments sets all 
exec_state's
-      // statuses to cancelled.
+      // CancelFragmentInstances() is called. CancelFragmentInstances sets all
+      // exec_state's statuses to cancelled.
       // TODO: We're losing this profile information. Call ReportQuerySummary 
only after
       // all backends have completed.
       exec_state->profile()->Update(cumulative_profile);
@@ -1899,18 +1732,12 @@ Status Coordinator::UpdateFragmentExecStatus(const 
TReportExecStatusParams& para
   return Status::OK();
 }
 
-const RowDescriptor& Coordinator::row_desc() const {
-  DCHECK(executor_.get() != NULL);
-  return executor_->row_desc();
-}
-
 RuntimeState* Coordinator::runtime_state() {
-  return executor_.get() == NULL ? NULL : executor_->runtime_state();
+  return executor_ == NULL ? NULL : executor_->runtime_state();
 }
 
 MemTracker* Coordinator::query_mem_tracker() {
-  return executor_.get() == NULL ? query_mem_tracker_.get() :
-      executor_->runtime_state()->query_mem_tracker();
+  return query_mem_tracker_.get();
 }
 
 bool Coordinator::PrepareCatalogUpdate(TUpdateCatalogRequest* catalog_update) {
@@ -1941,7 +1768,9 @@ void 
Coordinator::UpdateAverageProfile(FragmentInstanceState* instance_state) {
   PerFragmentProfileData* data = &fragment_profiles_[fragment_idx];
 
   // No locks are taken since UpdateAverage() and AddChild() take their own 
locks
-  data->averaged_profile->UpdateAverage(instance_state->profile());
+  if (data->averaged_profile != nullptr) {
+    data->averaged_profile->UpdateAverage(instance_state->profile());
+  }
   data->root_profile->AddChild(instance_state->profile());
 }
 
@@ -2000,18 +1829,18 @@ void Coordinator::UpdateExecSummary(const 
FragmentInstanceState& instance_state)
 
 // This function appends summary information to the query_profile_ before
 // outputting it to VLOG.  It adds:
-//   1. Averaged remote fragment profiles (TODO: add outliers)
-//   2. Summary of remote fragment durations (min, max, mean, stddev)
-//   3. Summary of remote fragment rates (min, max, mean, stddev)
+//   1. Averaged fragment instance profiles (TODO: add outliers)
+//   2. Summary of fragment instance durations (min, max, mean, stddev)
+//   3. Summary of fragment instance rates (min, max, mean, stddev)
 // TODO: add histogram/percentile
 void Coordinator::ReportQuerySummary() {
-  // In this case, the query did not even get to start on all the remote nodes,
-  // some of the state that is used below might be uninitialized.  In this 
case,
+  // In this case, the query did not even get to start all fragment instances.
+  // Some of the state that is used below might be uninitialized.  In this 
case,
   // the query has made so little progress, reporting a summary is not very 
useful.
   if (!has_called_wait_) return;
 
   if (!fragment_instance_states_.empty()) {
-    // Average all remote fragments for each fragment.
+    // Average all fragment instances for each fragment.
     for (FragmentInstanceState* state: fragment_instance_states_) {
       // TODO: make profiles uniform across all fragments so we don't have
       // to keep special-casing the coord fragment
@@ -2028,7 +1857,7 @@ void Coordinator::ReportQuerySummary() {
 
     InstanceComparator comparator;
     // Per fragment instances have been collected, output summaries
-    for (int i = (executor_.get() != NULL ? 1 : 0); i < 
fragment_profiles_.size(); ++i) {
+    for (int i = (executor_ != NULL ? 1 : 0); i < fragment_profiles_.size(); 
++i) {
       fragment_profiles_[i].root_profile->SortChildren(comparator);
       SummaryStats& completion_times = fragment_profiles_[i].completion_times;
       SummaryStats& rates = fragment_profiles_[i].rates;
@@ -2088,16 +1917,6 @@ void Coordinator::ReportQuerySummary() {
 
 string Coordinator::GetErrorLog() {
   ErrorLogMap merged;
-  {
-    lock_guard<mutex> l(lock_);
-    // TODO-MT: use FragmentInstanceState::error_log_ instead
-    // as part of getting rid of the special-casing of the coordinator instance
-    if (executor_.get() != NULL && executor_->runtime_state() != NULL) {
-      ErrorLogMap runtime_error_log;
-      executor_->runtime_state()->GetErrors(&runtime_error_log);
-      MergeErrorMaps(&merged, runtime_error_log);
-    }
-  }
   for (FragmentInstanceState* state: fragment_instance_states_) {
     lock_guard<mutex> l(*state->lock());
     if (state->error_log()->size() > 0)  MergeErrorMaps(&merged, 
*state->error_log());
@@ -2285,11 +2104,20 @@ void DistributeFilters(shared_ptr<TPublishFilterParams> 
params,
 void Coordinator::TearDown() {
   DCHECK(!torn_down_) << "Coordinator::TearDown() may not be called twice";
   torn_down_ = true;
-  lock_guard<SpinLock> l(filter_lock_);
-  for (auto& filter: filter_routing_table_) {
-    FilterState* state = &filter.second;
-    state->Disable(filter_mem_tracker_.get());
+  if (filter_routing_table_.size() > 0) {
+    query_profile_->AddInfoString("Final filter table", FilterDebugString());
   }
+
+  {
+    lock_guard<SpinLock> l(filter_lock_);
+    for (auto& filter : filter_routing_table_) {
+      FilterState* state = &filter.second;
+      state->Disable(filter_mem_tracker_.get());
+    }
+  }
+
+  // Need to protect against failed Prepare(), where root_sink() would not be 
set.
+  if (root_sink_ != nullptr) root_sink_->CloseConsumer();
 }
 
 void Coordinator::UpdateFilter(const TUpdateFilterParams& params) {

http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/9f61397f/be/src/runtime/coordinator.h
----------------------------------------------------------------------
diff --git a/be/src/runtime/coordinator.h b/be/src/runtime/coordinator.h
index bb67377..f73cf42 100644
--- a/be/src/runtime/coordinator.h
+++ b/be/src/runtime/coordinator.h
@@ -34,16 +34,18 @@
 #include <boost/thread/mutex.hpp>
 #include <boost/thread/condition_variable.hpp>
 
+#include "common/global-types.h"
 #include "common/hdfs.h"
 #include "common/status.h"
-#include "common/global-types.h"
-#include "util/progress-updater.h"
-#include "util/histogram-metric.h"
-#include "util/runtime-profile.h"
+#include "gen-cpp/Frontend_types.h"
+#include "gen-cpp/Types_types.h"
 #include "runtime/runtime-state.h"
 #include "scheduling/simple-scheduler.h"
-#include "gen-cpp/Types_types.h"
-#include "gen-cpp/Frontend_types.h"
+#include "service/fragment-exec-state.h"
+#include "service/fragment-mgr.h"
+#include "util/histogram-metric.h"
+#include "util/progress-updater.h"
+#include "util/runtime-profile.h"
 
 namespace impala {
 
@@ -67,28 +69,33 @@ class TRuntimeProfileTree;
 class RuntimeProfile;
 class TablePrinter;
 class TPlanFragment;
+class QueryResultSet;
 
 struct DebugOptions;
 
-/// Query coordinator: handles execution of plan fragments on remote nodes, 
given
-/// a TQueryExecRequest. As part of that, it handles all interactions with the
-/// executing backends; it is also responsible for implementing all client 
requests
-/// regarding the query, including cancellation.
-/// The coordinator fragment is executed locally in the calling thread, all 
other
-/// fragments are sent to remote nodes. The coordinator also monitors
-/// the execution status of the remote fragments and aborts the entire query 
if an error
-/// occurs, either in any of the remote fragments or in the local fragment.
+/// Query coordinator: handles execution of fragment instances on remote 
nodes, given a
+/// TQueryExecRequest. As part of that, it handles all interactions with the 
executing
+/// backends; it is also responsible for implementing all client requests 
regarding the
+/// query, including cancellation.
+///
+/// The coordinator monitors the execution status of fragment instances and 
aborts the
+/// entire query if an error is reported by any of them.
+///
+/// Queries that have results have those results fetched by calling GetNext(). 
Results
+/// rows are produced by a fragment instance that always executes on the same 
machine as
+/// the coordinator.
+///
 /// Once a query has finished executing and all results have been returned 
either to the
 /// caller of GetNext() or a data sink, execution_completed() will return 
true. If the
-/// query is aborted, execution_completed should also be set to true.
-/// Coordinator is thread-safe, with the exception of GetNext().
+/// query is aborted, execution_completed should also be set to true. 
Coordinator is
+/// thread-safe, with the exception of GetNext().
 //
 /// A typical sequence of calls for a single query (calls under the same 
numbered
 /// item can happen concurrently):
 /// 1. client: Exec()
 /// 2. client: Wait()/client: Cancel()/backend: UpdateFragmentExecStatus()
 /// 3. client: GetNext()*/client: Cancel()/backend: UpdateFragmentExecStatus()
-//
+///
 /// The implementation ensures that setting an overall error status and 
initiating
 /// cancellation of local and all remote fragments is atomic.
 ///
@@ -104,14 +111,10 @@ class Coordinator {
       RuntimeProfile::EventSequence* events);
   ~Coordinator();
 
-  /// Initiate asynchronous execution of a query with the given schedule. 
Returns as soon
-  /// as all plan fragments have started executing at their respective 
backends.
-  /// 'schedule' must contain at least a coordinator plan fragment (ie, can't
-  /// be for a query like 'SELECT 1').
-  /// Populates and prepares output_expr_ctxs from the coordinator's fragment 
if there is
-  /// one, and LLVM optimizes them together with the fragment's other exprs.
+  /// Initiate asynchronous execution of a query with the given schedule. When 
it returns,
+  /// all fragment instances have started executing at their respective 
backends.
   /// A call to Exec() must precede all other member function calls.
-  Status Exec(std::vector<ExprContext*>* output_expr_ctxs);
+  Status Exec();
 
   /// Blocks until result rows are ready to be retrieved via GetNext(), or, if 
the
   /// query doesn't return rows, until the query finishes or is cancelled.
@@ -120,25 +123,25 @@ class Coordinator {
   /// Wait() calls concurrently.
   Status Wait();
 
-  /// Returns tuples from the coordinator fragment. Any returned tuples are 
valid until
-  /// the next GetNext() call. If *batch is NULL, execution has completed and 
GetNext()
-  /// must not be called again.
-  /// GetNext() will not set *batch=NULL until all fragment instances have
-  /// either completed or have failed.
-  /// It is safe to call GetNext() even in the case where there is no 
coordinator fragment
-  /// (distributed INSERT).
-  /// '*batch' is owned by the underlying PlanFragmentExecutor and must not be 
deleted.
-  /// *state is owned by the caller, and must not be deleted.
-  /// Returns an error status if an error was encountered either locally or by
-  /// any of the remote fragments or if the query was cancelled.
-  /// GetNext() is not thread-safe: multiple threads must not make concurrent
-  /// GetNext() calls (but may call any of the other member functions 
concurrently
-  /// with GetNext()).
-  Status GetNext(RowBatch** batch, RuntimeState* state);
+  /// Fills 'results' with up to 'max_rows' rows. May return fewer than 
'max_rows'
+  /// rows, but will not return more.
+  ///
+  /// If *eos is true, execution has completed and GetNext() must not be called
+  /// again.
+  ///
+  /// GetNext() will not set *eos=true until all fragment instances have 
either completed
+  /// or have failed.
+  ///
+  /// Returns an error status if an error was encountered either locally or by 
any of the
+  /// remote fragments or if the query was cancelled.
+  ///
+  /// GetNext() is not thread-safe: multiple threads must not make concurrent 
GetNext()
+  /// calls (but may call any of the other member functions concurrently with 
GetNext()).
+  Status GetNext(QueryResultSet* results, int max_rows, bool* eos);
 
   /// Cancel execution of query. This includes the execution of the local plan 
fragment,
-  /// if any, as well as all plan fragments on remote nodes. Sets 
query_status_ to
-  /// the given cause if non-NULL. Otherwise, sets query_status_ to 
Status::CANCELLED.
+  /// if any, as well as all plan fragments on remote nodes. Sets 
query_status_ to the
+  /// given cause if non-NULL. Otherwise, sets query_status_ to 
Status::CANCELLED.
   /// Idempotent.
   void Cancel(const Status* cause = NULL);
 
@@ -151,12 +154,18 @@ class Coordinator {
   /// to CancelInternal().
   Status UpdateFragmentExecStatus(const TReportExecStatusParams& params);
 
-  /// only valid *after* calling Exec(), and may return NULL if there is no 
executor
+  /// Only valid *after* calling Exec(). Return nullptr if the running query 
does not
+  /// produce any rows.
+  ///
+  /// TODO: The only dependency on this is QueryExecState, used to track 
memory for the
+  /// result cache. Remove this dependency, possibly by moving result caching 
inside this
+  /// class.
   RuntimeState* runtime_state();
-  const RowDescriptor& row_desc() const;
 
   /// Only valid after Exec(). Returns runtime_state()->query_mem_tracker() if 
there
   /// is a coordinator fragment, or query_mem_tracker_ (initialized in Exec()) 
otherwise.
+  ///
+  /// TODO: Remove, see runtime_state().
   MemTracker* query_mem_tracker();
 
   /// Get cumulative profile aggregated over all fragments of the query.
@@ -275,8 +284,18 @@ class Coordinator {
   /// Once this is set to true, errors from remote fragments are ignored.
   bool returned_all_results_;
 
-  /// execution state of coordinator fragment
-  boost::scoped_ptr<PlanFragmentExecutor> executor_;
+  /// Non-null if and only if the query produces results for the client; i.e. 
is of
+  /// TStmtType::QUERY. Coordinator uses these to pull results from plan tree 
and return
+  /// them to the client in GetNext(), and also to access the fragment 
instance's runtime
+  /// state.
+  ///
+  /// Result rows are materialized by this fragment instance in its own 
thread. They are
+  /// materialized into a QueryResultSet provided to the coordinator during 
GetNext().
+  ///
+  /// Not owned by this class, created during fragment instance start-up by
+  /// FragmentExecState and set here in Exec().
+  PlanFragmentExecutor* executor_ = nullptr;
+  PlanRootSink* root_sink_ = nullptr;
 
   /// Query mem tracker for this coordinator initialized in Exec(). Only valid 
if there
   /// is no coordinator fragment (i.e. executor_ == NULL). If executor_ is not 
NULL,
@@ -383,7 +402,7 @@ class Coordinator {
   RuntimeProfile::Counter* finalization_timer_;
 
   /// Barrier that is released when all calls to ExecRemoteFragment() have
-  /// returned, successfully or not. Initialised during StartRemoteFragments().
+  /// returned, successfully or not. Initialised during Exec().
   boost::scoped_ptr<CountingBarrier> exec_complete_barrier_;
 
   /// Represents a runtime filter target.
@@ -465,10 +484,9 @@ class Coordinator {
   /// Runs cancel logic. Assumes that lock_ is held.
   void CancelInternal();
 
-  /// Cancels remote fragments. Assumes that lock_ is held.  This can be 
called when
-  /// the query is not being cancelled in the case where the query limit is
-  /// reached.
-  void CancelRemoteFragments();
+  /// Cancels all fragment instances. Assumes that lock_ is held. This may be 
called when
+  /// the query is not being cancelled in the case where the query limit is 
reached.
+  void CancelFragmentInstances();
 
   /// Acquires lock_ and updates query_status_ with 'status' if it's not 
already
   /// an error status, and returns the current query_status_.
@@ -531,30 +549,18 @@ class Coordinator {
   void PopulatePathPermissionCache(hdfsFS fs, const std::string& path_str,
       PermissionCache* permissions_cache);
 
-  /// Validates that all collection-typed slots in the given batch are set to 
NULL.
-  /// See SubplanNode for details on when collection-typed slots are set to 
NULL.
-  /// TODO: This validation will become obsolete when we can return collection 
values.
-  /// We will then need a different mechanism to assert the correct behavior 
of the
-  /// SubplanNode with respect to setting collection-slots to NULL.
-  void ValidateCollectionSlots(RowBatch* batch);
-
-  /// Prepare coordinator fragment for execution (update filter routing table,
-  /// prepare executor, set up output exprs) and create its 
FragmentInstanceState.
-  Status PrepareCoordFragment(std::vector<ExprContext*>* output_expr_ctxs);
-
-  /// Starts all remote fragments contained in the schedule by issuing RPCs in 
parallel,
+  /// Starts all fragment instances contained in the schedule by issuing RPCs 
in parallel,
   /// and then waiting for all of the RPCs to complete.
-  void StartRemoteFragments();
+  void StartFragments();
 
-  /// Starts all remote fragment instances contained in the schedule by 
issuing RPCs in
-  /// parallel and then waiting for all of the RPCs to complete. Also sets up 
and
-  /// registers the state for all non-coordinator fragment instance.
-  void MtStartRemoteFInstances();
+  /// Starts all fragment instances contained in the schedule by issuing RPCs 
in parallel
+  /// and then waiting for all of the RPCs to complete.
+  void MtStartFInstances();
 
   /// Calls CancelInternal() and returns an error if there was any error 
starting the
   /// fragments.
   /// Also updates query_profile_ with the startup latency histogram.
-  Status FinishRemoteInstanceStartup();
+  Status FinishInstanceStartup();
 
   /// Build the filter routing table by iterating over all plan nodes and 
collecting the
   /// filters that they either produce or consume. The source and target 
fragment

http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/9f61397f/be/src/runtime/exec-env.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/exec-env.cc b/be/src/runtime/exec-env.cc
index 1b3fa14..db0d242 100644
--- a/be/src/runtime/exec-env.cc
+++ b/be/src/runtime/exec-env.cc
@@ -24,6 +24,8 @@
 #include <gutil/strings/substitute.h>
 
 #include "common/logging.h"
+#include "gen-cpp/CatalogService.h"
+#include "gen-cpp/ImpalaInternalService.h"
 #include "runtime/backend-client.h"
 #include "runtime/client-cache.h"
 #include "runtime/coordinator.h"
@@ -36,25 +38,24 @@
 #include "runtime/thread-resource-mgr.h"
 #include "runtime/tmp-file-mgr.h"
 #include "scheduling/request-pool-service.h"
-#include "service/frontend.h"
 #include "scheduling/simple-scheduler.h"
+#include "service/fragment-mgr.h"
+#include "service/frontend.h"
 #include "statestore/statestore-subscriber.h"
 #include "util/debug-util.h"
+#include "util/debug-util.h"
 #include "util/default-path-handlers.h"
 #include "util/hdfs-bulk-ops.h"
 #include "util/mem-info.h"
+#include "util/mem-info.h"
+#include "util/memory-metrics.h"
+#include "util/memory-metrics.h"
 #include "util/metrics.h"
 #include "util/network-util.h"
 #include "util/parse-util.h"
-#include "util/memory-metrics.h"
-#include "util/webserver.h"
-#include "util/mem-info.h"
-#include "util/debug-util.h"
-#include "util/memory-metrics.h"
 #include "util/pretty-printer.h"
 #include "util/thread-pool.h"
-#include "gen-cpp/ImpalaInternalService.h"
-#include "gen-cpp/CatalogService.h"
+#include "util/webserver.h"
 
 #include "common/names.h"
 
@@ -145,6 +146,7 @@ ExecEnv::ExecEnv()
     fragment_exec_thread_pool_(new 
CallableThreadPool("coordinator-fragment-rpc",
         "worker", FLAGS_coordinator_rpc_threads, 
numeric_limits<int32_t>::max())),
     async_rpc_pool_(new CallableThreadPool("rpc-pool", "async-rpc-sender", 8, 
10000)),
+    fragment_mgr_(new FragmentMgr()),
     enable_webserver_(FLAGS_enable_webserver),
     is_fe_tests_(false),
     backend_address_(MakeNetworkAddress(FLAGS_hostname, FLAGS_be_port)) {
@@ -197,6 +199,7 @@ ExecEnv::ExecEnv(const string& hostname, int backend_port, 
int subscriber_port,
     fragment_exec_thread_pool_(new 
CallableThreadPool("coordinator-fragment-rpc",
         "worker", FLAGS_coordinator_rpc_threads, 
numeric_limits<int32_t>::max())),
     async_rpc_pool_(new CallableThreadPool("rpc-pool", "async-rpc-sender", 8, 
10000)),
+    fragment_mgr_(new FragmentMgr()),
     enable_webserver_(FLAGS_enable_webserver && webserver_port > 0),
     is_fe_tests_(false),
     backend_address_(MakeNetworkAddress(FLAGS_hostname, FLAGS_be_port)) {

http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/9f61397f/be/src/runtime/exec-env.h
----------------------------------------------------------------------
diff --git a/be/src/runtime/exec-env.h b/be/src/runtime/exec-env.h
index 303876f..718c6d0 100644
--- a/be/src/runtime/exec-env.h
+++ b/be/src/runtime/exec-env.h
@@ -96,6 +96,7 @@ class ExecEnv {
   Frontend* frontend() { return frontend_.get(); };
   RequestPoolService* request_pool_service() { return 
request_pool_service_.get(); }
   CallableThreadPool* rpc_pool() { return async_rpc_pool_.get(); }
+  FragmentMgr* fragment_mgr() { return fragment_mgr_.get(); }
 
   void set_enable_webserver(bool enable) { enable_webserver_ = enable; }
 
@@ -137,6 +138,7 @@ class ExecEnv {
   boost::scoped_ptr<Frontend> frontend_;
   boost::scoped_ptr<CallableThreadPool> fragment_exec_thread_pool_;
   boost::scoped_ptr<CallableThreadPool> async_rpc_pool_;
+  boost::scoped_ptr<FragmentMgr> fragment_mgr_;
 
   /// Not owned by this class
   ImpalaServer* impala_server_;

Reply via email to