data-stream-sender.h

Go to the documentation of this file.

// Copyright 2012 Cloudera Inc.
//
// Licensed 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_RUNTIME_DATA_STREAM_SENDER_H
#define IMPALA_RUNTIME_DATA_STREAM_SENDER_H

#include <vector>
#include <string>

#include "exec/data-sink.h"
#include "common/global-types.h"
#include "common/object-pool.h"
#include "common/status.h"
#include "util/runtime-profile.h"
#include "gen-cpp/Results_types.h" // for TRowBatch

namespace impala {

class Expr;
class RowBatch;
class RowDescriptor;
class MemTracker;
class TDataStreamSink;
class TNetworkAddress;
class TPlanFragmentDestination;

//
class DataStreamSender : public DataSink {
 public:
  DataStreamSender(ObjectPool* pool, int sender_id,
    const RowDescriptor& row_desc, const TDataStreamSink& sink,
    const std::vector<TPlanFragmentDestination>& destinations,
    int per_channel_buffer_size);
  virtual ~DataStreamSender();

  virtual Status Prepare(RuntimeState* state);

  virtual Status Open(RuntimeState* state);

  virtual Status Send(RuntimeState* state, RowBatch* batch, bool eos);

  virtual void Close(RuntimeState* state);

  void SerializeBatch(RowBatch* src, TRowBatch* dest, int num_receivers = 1);

  int64_t GetNumDataBytesSent() const;

  virtual RuntimeProfile* profile() { return profile_; }

 private:
  class Channel;

  int sender_id_;
  RuntimeState* state_;
  ObjectPool* pool_;
  const RowDescriptor& row_desc_;
  bool broadcast_;  // if true, send all rows on all channels
  bool random_; // if true, round-robins row batches among channels
  int current_channel_idx_; // index of current channel to send to if random_ == true

  bool closed_;

  TRowBatch thrift_batch1_;
  TRowBatch thrift_batch2_;
  TRowBatch* current_thrift_batch_;  // the next one to fill in Send()

  std::vector<ExprContext*> partition_expr_ctxs_;  // compute per-row partition values
  std::vector<Channel*> channels_;

  RuntimeProfile* profile_; // Allocated from pool_
  RuntimeProfile::Counter* serialize_batch_timer_;
  RuntimeProfile::Counter* thrift_transmit_timer_;
  RuntimeProfile::Counter* bytes_sent_counter_;
  RuntimeProfile::Counter* uncompressed_bytes_counter_;
  boost::scoped_ptr<MemTracker> mem_tracker_;

  RuntimeProfile::Counter* network_throughput_;

  RuntimeProfile::Counter* overall_throughput_;

  PlanNodeId dest_node_id_;
};

}

#endif