hbase-table-scanner.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_HBASE_TABLE_SCANNER_H
#define IMPALA_EXEC_HBASE_TABLE_SCANNER_H

#include <boost/scoped_ptr.hpp>
#include <jni.h>
#include <string>
#include <sstream>
#include <vector>
#include "gen-cpp/PlanNodes_types.h"
#include "exec/scan-node.h"
#include "runtime/hbase-table-factory.h"
#include "runtime/hbase-table.h"

namespace impala {

class TupleDescriptor;
class Tuple;
class RuntimeState;
class MemPool;
class Status;
class HBaseScanNode;

//
//
//
//
class HBaseTableScanner {
 public:
  HBaseTableScanner(HBaseScanNode* scan_node, HBaseTableFactory* htable_factory,
                    RuntimeState* state);

  static Status Init();

  class ScanRange {
   public:
    ScanRange()
      : start_key_(),
        stop_key_() {
    }

    const std::string& start_key() const { return start_key_; }
    const std::string& stop_key() const {return stop_key_; }
    void set_start_key(const std::string& key) { start_key_ = key; }
    void set_stop_key(const std::string& key) { stop_key_ = key; }

    void DebugString(int indentation_level, std::stringstream* out);

   private:
    std::string start_key_;
    std::string stop_key_;
  };

  typedef std::vector<ScanRange> ScanRangeVector;

  Status StartScan(JNIEnv* env, const TupleDescriptor* tuple_desc,
                   const ScanRangeVector& scan_range_vector,
                   const std::vector<THBaseFilter>& filters);

  Status Next(JNIEnv* env, bool* has_next);

  Status GetRowKey(JNIEnv* env, void** key, int* key_length);

  Status GetRowKey(JNIEnv* env, const SlotDescriptor* slot_desc, Tuple* tuple);

  Status GetValue(JNIEnv* env, const std::string& family, const std::string& qualifier,
      void** value, int* value_length);

  Status GetValue(JNIEnv* env, const std::string& family, const std::string& qualifier,
      const SlotDescriptor* slot_desc, Tuple* tuple);

  void Close(JNIEnv* env);

  void set_num_requested_cells(int num_requested_cells) {
    num_requested_cells_ = num_requested_cells;
  }

 private:
  static const int DEFAULT_ROWS_CACHED = 1024;

  HBaseScanNode* scan_node_;
  RuntimeState* state_;

  static jclass scan_cl_;
  static jclass resultscanner_cl_;
  static jclass result_cl_;
  static jclass cell_cl_;
  static jclass hconstants_cl_;
  static jclass filter_list_cl_;
  static jclass filter_list_op_cl_;
  static jclass single_column_value_filter_cl_;
  static jclass compare_op_cl_;
  static jclass scanner_timeout_ex_cl_;

  static jmethodID scan_ctor_;
  static jmethodID scan_set_max_versions_id_;
  static jmethodID scan_set_caching_id_;
  static jmethodID scan_set_cache_blocks_id_;
  static jmethodID scan_add_column_id_;
  static jmethodID scan_set_filter_id_;
  static jmethodID scan_set_start_row_id_;
  static jmethodID scan_set_stop_row_id_;
  static jmethodID resultscanner_next_id_;
  static jmethodID resultscanner_close_id_;
  static jmethodID result_isempty_id_;
  static jmethodID result_raw_cells_id_;
  static jmethodID cell_get_row_array_;
  static jmethodID cell_get_family_array_;
  static jmethodID cell_get_qualifier_array_;
  static jmethodID cell_get_value_array_;
  static jmethodID cell_get_family_offset_id_;
  static jmethodID cell_get_family_length_id_;
  static jmethodID cell_get_qualifier_offset_id_;
  static jmethodID cell_get_qualifier_length_id_;
  static jmethodID cell_get_row_offset_id_;
  static jmethodID cell_get_row_length_id_;
  static jmethodID cell_get_value_offset_id_;
  static jmethodID cell_get_value_length_id_;
  static jmethodID filter_list_ctor_;
  static jmethodID filter_list_add_filter_id_;
  static jmethodID single_column_value_filter_ctor_;

  static jobject empty_row_;
  static jobject must_pass_all_op_;
  static jobjectArray compare_ops_;

  HBaseTableFactory* htable_factory_;

  const ScanRangeVector* scan_range_vector_;
  int current_scan_range_idx_;  // the index of the current scan range

  boost::scoped_ptr<HBaseTable> htable_;

  jobject scan_;           // Java type Scan
  jobject resultscanner_;  // Java type ResultScanner

  jobjectArray cells_;

  int cell_index_;

  int num_requested_cells_;

  int num_addl_requested_cols_;

  int num_cells_;

  bool all_cells_present_;

  boost::scoped_ptr<MemPool> value_pool_;

  int rows_cached_;

  bool cache_blocks_;

  RuntimeProfile::Counter* scan_setup_timer_;

  Status HandleResultScannerTimeout(JNIEnv* env, bool* timeout);

  int CompareStrings(const std::string& s, void* data, int length);

  Status CreateByteArray(JNIEnv* env, const std::string& s, jbyteArray* bytes);

  Status ScanSetup(JNIEnv* env, const TupleDescriptor* tuple_desc,
                   const std::vector<THBaseFilter>& filters);

  Status InitScanRange(JNIEnv* env, const ScanRange& scan_range);
  Status InitScanRange(JNIEnv* env, jbyteArray start_bytes, jbyteArray end_bytes);

  inline void GetRowKey(JNIEnv* env, jobject cell, void** data, int* length);

  inline void GetFamily(JNIEnv* env, jobject cell, void** data, int* length);

  inline void GetQualifier(JNIEnv* env, jobject cell, void** data, int* length);

  inline void GetValue(JNIEnv* env, jobject cell, void** data, int* length);

  inline Status GetCurrentValue(JNIEnv* env, const std::string& family,
      const std::string& qualifier, void** data, int* length, bool* is_null);

  inline void WriteTupleSlot(const SlotDescriptor* slot_desc, Tuple* tuple, void* data);
};

}  // namespace impala

#endif