hdfs-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_HDFS_SCAN_NODE_H_
#define IMPALA_EXEC_HDFS_SCAN_NODE_H_

#include <vector>
#include <memory>
#include <stdint.h>

#include <boost/unordered_map.hpp>
#include <boost/unordered_set.hpp>
#include <boost/scoped_ptr.hpp>
#include <boost/thread/condition_variable.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/thread/thread.hpp>

#include "exec/scan-node.h"
#include "exec/scanner-context.h"
#include "runtime/descriptors.h"
#include "runtime/disk-io-mgr.h"
#include "runtime/string-buffer.h"
#include "util/progress-updater.h"
#include "util/spinlock.h"
#include "util/thread.h"

#include "gen-cpp/PlanNodes_types.h"

namespace impala {

class DescriptorTbl;
class HdfsScanner;
class RowBatch;
class Status;
class Tuple;
class TPlanNode;
class TScanRange;

struct HdfsFileDesc {
  hdfsFS fs;

  std::string filename;

  int64_t file_length;

  int64_t mtime;

  THdfsCompression::type file_compression;

  std::vector<DiskIoMgr::ScanRange*> splits;
  HdfsFileDesc(const std::string& filename)
      : filename(filename), file_length(0), mtime(0), file_compression(THdfsCompression::NONE) {
  }
};

struct ScanRangeMetadata {
  int64_t partition_id;

  ScanRangeMetadata(int64_t partition_id)
    : partition_id(partition_id) { }
};

class HdfsScanNode : public ScanNode {
 public:
  HdfsScanNode(ObjectPool* pool, const TPlanNode& tnode, const DescriptorTbl& descs);

  ~HdfsScanNode();

  virtual Status Prepare(RuntimeState* state);
  virtual Status Open(RuntimeState* state);
  virtual Status GetNext(RuntimeState* state, RowBatch* row_batch, bool* eos);
  virtual Status Reset(RuntimeState* state);
  virtual void Close(RuntimeState* state);

  int limit() const { return limit_; }

  const std::vector<SlotDescriptor*>& materialized_slots()
      const { return materialized_slots_; }

  int tuple_idx() const { return 0; }

  int num_partition_keys() const { return hdfs_table_->num_clustering_cols(); }

  int num_materialized_partition_keys() const { return partition_key_slots_.size(); }

  const TupleDescriptor* tuple_desc() { return tuple_desc_; }

  const HdfsTableDescriptor* hdfs_table() { return hdfs_table_; }

  RuntimeState* runtime_state() { return runtime_state_; }

  DiskIoMgr::RequestContext* reader_context() { return reader_context_; }

  RuntimeProfile::HighWaterMarkCounter* max_compressed_text_file_length() {
    return max_compressed_text_file_length_;
  }

  const static int SKIP_COLUMN = -1;

  Status GetConjunctCtxs(std::vector<ExprContext*>* ctxs);

  int GetMaterializedSlotIdx(const std::vector<int>& path) const {
    PathToSlotIdxMap::const_iterator result = path_to_materialized_slot_idx_.find(path);
    if (result == path_to_materialized_slot_idx_.end()) return SKIP_COLUMN;
    return result->second;
  }

  const bool* is_materialized_col() {
    return reinterpret_cast<const bool*>(&is_materialized_col_[0]);
  }

  void* GetCodegenFn(THdfsFileFormat::type);

  inline void IncNumScannersCodegenEnabled() {
    ++num_scanners_codegen_enabled_;
  }

  inline void IncNumScannersCodegenDisabled() {
    ++num_scanners_codegen_disabled_;
  }

  void AddMaterializedRowBatch(RowBatch* row_batch);

  DiskIoMgr::ScanRange* AllocateScanRange(
      hdfsFS fs, const char* file, int64_t len, int64_t offset, int64_t partition_id,
      int disk_id, bool try_cache, bool expected_local, int64_t mtime);

  Status AddDiskIoRanges(const std::vector<DiskIoMgr::ScanRange*>& ranges);

  Status AddDiskIoRanges(const HdfsFileDesc* file_desc);

  void MarkFileDescIssued(const HdfsFileDesc* file_desc);

  Tuple* InitTemplateTuple(RuntimeState* state,
                           const std::vector<ExprContext*>& value_ctxs);

  Tuple* InitEmptyTemplateTuple();

  void TransferToScanNodePool(MemPool* pool);

  HdfsFileDesc* GetFileDesc(const std::string& filename);

  void* GetFileMetadata(const std::string& filename);

  void SetFileMetadata(const std::string& filename, void* metadata);

  void RangeComplete(const THdfsFileFormat::type& file_type,
      const THdfsCompression::type& compression_type);
  void RangeComplete(const THdfsFileFormat::type& file_type,
      const std::vector<THdfsCompression::type>& compression_type);

  void ComputeSlotMaterializationOrder(std::vector<int>* order) const;

  typedef boost::unordered_map<int32_t, std::pair<int, int64_t> > PerVolumnStats;

  static void UpdateHdfsSplitStats(
      const std::vector<TScanRangeParams>& scan_range_params_list,
      PerVolumnStats* per_volume_stats);

  static void PrintHdfsSplitStats(const PerVolumnStats& per_volume_stats,
      std::stringstream* ss);

  static const std::string HDFS_SPLIT_STATS_DESC;

 private:
  friend class ScannerContext;

  boost::scoped_ptr<TPlanNode> thrift_plan_node_;

  RuntimeState* runtime_state_;

  const int tuple_id_;

  DiskIoMgr::RequestContext* reader_context_;

  const TupleDescriptor* tuple_desc_;

  const HdfsTableDescriptor* hdfs_table_;

  bool unknown_disk_id_warned_;

  boost::unordered_set<int64_t> partition_ids_;

  typedef std::map<std::string, HdfsFileDesc*> FileDescMap;
  FileDescMap file_descs_;

  typedef std::map<THdfsFileFormat::type, std::vector<HdfsFileDesc*> > FileFormatsMap;
  FileFormatsMap per_type_files_;

  bool initial_ranges_issued_;

  int64_t scanner_thread_bytes_required_;

  AtomicInt<int> num_unqueued_files_;

  typedef std::map<THdfsFileFormat::type, HdfsScanner*> ScannerMap;
  ScannerMap scanner_map_;

  typedef std::map<THdfsFileFormat::type, void*> CodegendFnMap;
  CodegendFnMap codegend_fn_map_;

  std::vector<ExprContext*> conjunct_ctxs_;

  typedef boost::unordered_map<std::vector<int>, int> PathToSlotIdxMap;
  PathToSlotIdxMap path_to_materialized_slot_idx_;

  //
  std::vector<char> is_materialized_col_;

  std::vector<SlotDescriptor*> materialized_slots_;

  std::vector<SlotDescriptor*> partition_key_slots_;

  ProgressUpdater progress_;

  boost::mutex metadata_lock_;
  std::map<std::string, void*> per_file_metadata_;

  ThreadGroup scanner_threads_;

  boost::scoped_ptr<RowBatchQueue> materialized_row_batches_;

  int max_materialized_row_batches_;

  AtomicInt<int> num_owned_io_buffers_;

  AtomicInt<int> num_skipped_tokens_;

  AtomicInt<int> num_scanners_codegen_enabled_;
  AtomicInt<int> num_scanners_codegen_disabled_;

  RuntimeProfile::HighWaterMarkCounter* max_compressed_text_file_length_;

  RuntimeProfile::Counter disks_accessed_bitmap_;

  RuntimeProfile::Counter* bytes_read_local_;

  RuntimeProfile::Counter* bytes_read_short_circuit_;

  RuntimeProfile::Counter* bytes_read_dn_cache_;

  RuntimeProfile::Counter* num_remote_ranges_;

  RuntimeProfile::Counter* unexpected_remote_bytes_;

  boost::mutex lock_;

  bool done_;

  bool all_ranges_started_;

  boost::scoped_ptr<MemPool> scan_node_pool_;

  Status status_;

  SpinLock file_type_counts_lock_;
  typedef std::map<
      std::pair<THdfsFileFormat::type, THdfsCompression::type>, int> FileTypeCountsMap;
  FileTypeCountsMap file_type_counts_;

  bool counters_running_;

  int32_t rm_callback_id_;

  void ThreadTokenAvailableCb(ThreadResourceMgr::ResourcePool* pool);

  HdfsScanner* CreateAndPrepareScanner(HdfsPartitionDescriptor* partition_desc,
      ScannerContext* context, Status* status);

  void ScannerThread();

  bool EnoughMemoryForScannerThread(bool new_thread);

  Status GetNextInternal(RuntimeState* state, RowBatch* row_batch, bool* eos);

  void SetDone();

  void StopAndFinalizeCounters();
};

}

#endif