buffered-tuple-stream.h

Go to the documentation of this file.

// Copyright 2013 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_BUFFERED_TUPLE_STREAM_H
#define IMPALA_RUNTIME_BUFFERED_TUPLE_STREAM_H

#include "common/status.h"
#include "runtime/buffered-block-mgr.h"

namespace impala {

class BufferedBlockMgr;
class RuntimeProfile;
class RuntimeState;
class RowBatch;
class RowDescriptor;
class SlotDescriptor;
class TupleRow;

//
//
//
//
//
//
//
//
class BufferedTupleStream {
 public:
  struct RowIdx {
    static const uint64_t BLOCK_MASK  = 0xFFFF;
    static const uint64_t BLOCK_SHIFT = 0;
    static const uint64_t OFFSET_MASK  = 0xFFFFFF0000;
    static const uint64_t OFFSET_SHIFT = 16;
    static const uint64_t IDX_MASK  = 0xFFFFFF0000000000;
    static const uint64_t IDX_SHIFT = 40;

    uint64_t block() const {
      return (data & BLOCK_MASK);
    };

    uint64_t offset() const {
      return (data & OFFSET_MASK) >> OFFSET_SHIFT;
    };

    uint64_t idx() const {
      return (data & IDX_MASK) >> IDX_SHIFT;
    }

    uint64_t set(uint64_t block, uint64_t offset, uint64_t idx) {
      DCHECK_LE(block, BLOCK_MASK)
          << "Cannot have more than 2^16 = 64K blocks in a tuple stream.";
      DCHECK_LE(offset, OFFSET_MASK >> OFFSET_SHIFT)
          << "Cannot have blocks larger than 2^24 = 16MB";
      DCHECK_LE(idx, IDX_MASK >> IDX_SHIFT)
          << "Cannot have more than 2^24 = 16M rows in a block.";
      data = block | (offset << OFFSET_SHIFT) | (idx << IDX_SHIFT);
      return data;
    }

    std::string DebugString() const;

    uint64_t data;
  };

  BufferedTupleStream(RuntimeState* state, const RowDescriptor& row_desc,
      BufferedBlockMgr* block_mgr, BufferedBlockMgr::Client* client,
      bool use_initial_small_buffers = true,
      bool delete_on_read = false, bool read_write = false);

  Status Init(RuntimeProfile* profile = NULL, bool pinned = true);

  Status SwitchToIoBuffers(bool* got_buffer);

  bool AddRow(TupleRow* row, uint8_t** dst = NULL);

  uint8_t* AllocateRow(int size);

  void GetTupleRow(const RowIdx& idx, TupleRow* row) const;

  Status PrepareForRead(bool* got_buffer = NULL);

  Status PinStream(bool already_reserved, bool* pinned);

  Status UnpinStream(bool all = false);

  Status GetNext(RowBatch* batch, bool* eos, std::vector<RowIdx>* indices = NULL);

  Status GetRows(boost::scoped_ptr<RowBatch>* batch, bool* got_rows);

  void Close();

  Status status() const { return status_; }

  int64_t num_rows() const { return num_rows_; }

  int64_t rows_returned() const { return rows_returned_; }

  int64_t byte_size() const { return total_byte_size_; }

  int64_t bytes_in_mem(bool ignore_current) const;

  bool is_pinned() const { return pinned_; }
  int blocks_pinned() const { return num_pinned_; }
  int blocks_unpinned() const { return blocks_.size() - num_pinned_ - num_small_blocks_; }
  bool has_read_block() const { return read_block_ != blocks_.end(); }
  bool has_write_block() const { return write_block_ != NULL; }
  bool using_small_buffers() const { return use_small_buffers_; }

  std::string DebugString() const;

 private:
  bool use_small_buffers_;

  const bool delete_on_read_;

  const bool read_write_;

  RuntimeState* const state_;

  const RowDescriptor& desc_;

  const bool nullable_tuple_;

  int fixed_tuple_row_size_;

  uint32_t null_indicators_read_block_;
  uint32_t null_indicators_write_block_;

  std::vector<std::pair<int, std::vector<SlotDescriptor*> > > string_slots_;

  BufferedBlockMgr* block_mgr_;
  BufferedBlockMgr::Client* block_mgr_client_;

  std::list<BufferedBlockMgr::Block*> blocks_;

  int64_t total_byte_size_;

  std::list<BufferedBlockMgr::Block*>::iterator read_block_;

  std::vector<uint8_t*> block_start_idx_;

  uint8_t* read_ptr_;

  uint32_t read_tuple_idx_;

  uint32_t write_tuple_idx_;

  int64_t read_bytes_;

  int64_t rows_returned_;

  int read_block_idx_;

  BufferedBlockMgr::Block* write_block_;

  int num_pinned_;

  int num_small_blocks_;

  bool closed_; // Used for debugging.
  Status status_;

  int64_t num_rows_;

  bool pinned_;

  RuntimeProfile::Counter* pin_timer_;
  RuntimeProfile::Counter* unpin_timer_;
  RuntimeProfile::Counter* get_new_block_timer_;

  template <bool HasNullableTuple>
  bool DeepCopyInternal(TupleRow* row, uint8_t** dst);

  bool DeepCopy(TupleRow* row, uint8_t** dst);

  Status NewBlockForWrite(int min_size, bool* got_block);

  Status NextBlockForRead();

  int ComputeRowSize(TupleRow* row) const;

  Status UnpinBlock(BufferedBlockMgr::Block* block);

  template <bool HasNullableTuple>
  Status GetNextInternal(RowBatch* batch, bool* eos, std::vector<RowIdx>* indices);

  int ComputeNumNullIndicatorBytes(int block_size) const;
};

}

#endif