scan-node.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_EXEC_SCAN_NODE_H_
#define IMPALA_EXEC_SCAN_NODE_H_

#include <string>
#include "exec/exec-node.h"
#include "util/runtime-profile.h"
#include "gen-cpp/ImpalaInternalService_types.h"

namespace impala {

class TScanRange;

//
//
//
//
//
//
//
//
//
//
//
//
//
//
class ScanNode : public ExecNode {
 public:
  ScanNode(ObjectPool* pool, const TPlanNode& tnode, const DescriptorTbl& descs)
    : ExecNode(pool, tnode, descs),
      scan_range_params_(NULL),
      active_scanner_thread_counter_(TUnit::UNIT, 0),
      active_hdfs_read_thread_counter_(TUnit::UNIT, 0) {}

  virtual Status Prepare(RuntimeState* state);

  void SetScanRanges(const std::vector<TScanRangeParams>& scan_range_params) {
    scan_range_params_ = &scan_range_params;
  }

  virtual bool IsScanNode() const { return true; }

  RuntimeProfile::Counter* bytes_read_counter() const { return bytes_read_counter_; }
  RuntimeProfile::Counter* rows_read_counter() const { return rows_read_counter_; }
  RuntimeProfile::Counter* read_timer() const { return read_timer_; }
  RuntimeProfile::Counter* total_throughput_counter() const {
    return total_throughput_counter_;
  }
  RuntimeProfile::Counter* per_read_thread_throughput_counter() const {
    return per_read_thread_throughput_counter_;
  }
  RuntimeProfile::Counter* materialize_tuple_timer() const {
    return materialize_tuple_timer_;
  }
  RuntimeProfile::Counter* scan_ranges_complete_counter() const {
    return scan_ranges_complete_counter_;
  }
  RuntimeProfile::ThreadCounters* scanner_thread_counters() const {
    return scanner_thread_counters_;
  }
  RuntimeProfile::Counter& active_scanner_thread_counter() {
    return active_scanner_thread_counter_;
  }
  RuntimeProfile::Counter* average_scanner_thread_concurrency() const {
    return average_scanner_thread_concurrency_;
  }

  static const std::string BYTES_READ_COUNTER;
  static const std::string ROWS_READ_COUNTER;
  static const std::string TOTAL_HDFS_READ_TIMER;
  static const std::string TOTAL_HBASE_READ_TIMER;
  static const std::string TOTAL_THROUGHPUT_COUNTER;
  static const std::string PER_READ_THREAD_THROUGHPUT_COUNTER;
  static const std::string NUM_DISKS_ACCESSED_COUNTER;
  static const std::string MATERIALIZE_TUPLE_TIMER;
  static const std::string SCAN_RANGES_COMPLETE_COUNTER;
  static const std::string SCANNER_THREAD_COUNTERS_PREFIX;
  static const std::string SCANNER_THREAD_TOTAL_WALLCLOCK_TIME;
  static const std::string AVERAGE_SCANNER_THREAD_CONCURRENCY;
  static const std::string AVERAGE_HDFS_READ_THREAD_CONCURRENCY;
  static const std::string NUM_SCANNER_THREADS_STARTED;

 protected:
  const std::vector<TScanRangeParams>* scan_range_params_;

  RuntimeProfile::Counter* bytes_read_counter_; // # bytes read from the scanner
  RuntimeProfile::TimeSeriesCounter* bytes_read_timeseries_counter_;
  RuntimeProfile::Counter* rows_read_counter_;
  RuntimeProfile::Counter* read_timer_; // total read time
  RuntimeProfile::Counter* total_throughput_counter_;
  RuntimeProfile::Counter* per_read_thread_throughput_counter_;
  RuntimeProfile::Counter* num_disks_accessed_counter_;
  RuntimeProfile::Counter* materialize_tuple_timer_;  // time writing tuple slots
  RuntimeProfile::Counter* scan_ranges_complete_counter_;
  RuntimeProfile::ThreadCounters* scanner_thread_counters_;

  RuntimeProfile::Counter active_scanner_thread_counter_;

  RuntimeProfile::Counter* average_scanner_thread_concurrency_;

  RuntimeProfile::Counter active_hdfs_read_thread_counter_;

  RuntimeProfile::Counter* average_hdfs_read_thread_concurrency_;

  RuntimeProfile::Counter* num_scanner_threads_started_counter_;

  std::vector<RuntimeProfile::Counter*> hdfs_read_thread_concurrency_bucket_;
};

}

#endif