scanner-context.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_SCANNER_CONTEXT_H
#define IMPALA_EXEC_SCANNER_CONTEXT_H

#include <boost/cstdint.hpp>
#include <boost/scoped_ptr.hpp>

#include "common/compiler-util.h"
#include "common/status.h"
#include "runtime/disk-io-mgr.h"
#include "runtime/row-batch.h"

namespace impala {

struct HdfsFileDesc;
class HdfsPartitionDescriptor;
class HdfsScanNode;
class MemPool;
class RowBatch;
class RuntimeState;
class StringBuffer;
class Tuple;
class TupleRow;

//
class ScannerContext {
 public:
  ScannerContext(RuntimeState*, HdfsScanNode*, HdfsPartitionDescriptor*,
      DiskIoMgr::ScanRange* scan_range);

  class Stream {
   public:
    bool GetBytes(int64_t requested_len, uint8_t** buffer, int64_t* out_len,
        Status* status, bool peek = false);

    Status GetBuffer(bool peek, uint8_t** buffer, int64_t* out_len);

    void set_contains_tuple_data(bool v) { contains_tuple_data_ = v; }

    typedef boost::function<int (int64_t)> ReadPastSizeCallback;
    void set_read_past_size_cb(ReadPastSizeCallback cb) { read_past_size_cb_ = cb; }

    int64_t bytes_left() { return scan_range_->len() - total_bytes_returned_; }

    bool eosr() const { return total_bytes_returned_ >= scan_range_->len() || eof(); }

    bool eof() const { return file_offset() == file_len_; }

    const char* filename() { return scan_range_->file(); }
    const DiskIoMgr::ScanRange* scan_range() { return scan_range_; }
    const HdfsFileDesc* file_desc() { return file_desc_; }

    int64_t file_offset() const { return scan_range_->offset() + total_bytes_returned_; }

    int64_t total_bytes_returned() { return total_bytes_returned_; }

    bool ReadBoolean(bool* boolean, Status*);

    bool ReadInt(int32_t* val, Status*, bool peek = false);

    bool ReadVLong(int64_t* val, Status*);

    bool ReadVInt(int32_t* val, Status*);

    bool ReadZLong(int64_t* val, Status*);

    bool SkipBytes(int64_t length, Status*);

    bool ReadBytes(int64_t length, uint8_t** buf, Status*, bool peek = false);

    bool ReadText(uint8_t** buf, int64_t* length, Status*);

    bool SkipText(Status*);

   private:
    friend class ScannerContext;
    ScannerContext* parent_;
    DiskIoMgr::ScanRange* scan_range_;
    const HdfsFileDesc* file_desc_;

    bool contains_tuple_data_;

    int64_t total_bytes_returned_;

    int64_t file_len_;

    ReadPastSizeCallback read_past_size_cb_;

    DiskIoMgr::BufferDescriptor* io_buffer_;

    uint8_t* io_buffer_pos_;

    int64_t io_buffer_bytes_left_;

    boost::scoped_ptr<MemPool> boundary_pool_;
    boost::scoped_ptr<StringBuffer> boundary_buffer_;
    uint8_t* boundary_buffer_pos_;
    int64_t boundary_buffer_bytes_left_;

    uint8_t** output_buffer_pos_;

    int64_t* output_buffer_bytes_left_;

    std::list<DiskIoMgr::BufferDescriptor*> completed_io_buffers_;

    Stream(ScannerContext* parent);

    Status GetBytesInternal(int64_t requested_len, uint8_t** buffer, bool peek,
                            int64_t* out_len);

    //
    //
    Status GetNextBuffer(int64_t read_past_size = 0);

    void ReleaseCompletedResources(RowBatch* batch, bool done);

    Status ReportIncompleteRead(int64_t length, int64_t bytes_read);
    Status ReportInvalidRead(int64_t length);
  };

  Stream* GetStream(int idx = 0) {
    DCHECK_GE(idx, 0);
    DCHECK_LT(idx, streams_.size());
    return streams_[idx];
  }

  //
  //
  void ReleaseCompletedResources(RowBatch* batch, bool done);

  Stream* AddStream(DiskIoMgr::ScanRange* range);

  bool cancelled() const;

  int num_completed_io_buffers() const { return num_completed_io_buffers_; }
  HdfsPartitionDescriptor* partition_descriptor() { return partition_desc_; }

 private:
  friend class Stream;

  RuntimeState* state_;
  HdfsScanNode* scan_node_;

  HdfsPartitionDescriptor* partition_desc_;

  std::vector<Stream*> streams_;

  int num_completed_io_buffers_;
};

}

#endif