Refactor IPC/File record batch read/write structure to reflect discussion in ARROW-384

Change-Id: Ic119c5647bdb6b1ff1fb31b3712527058c68e40c

Author: wesm

cpp/CMakeLists.txt

@@ -528,7 +528,6 @@ if(ARROW_BUILD_TESTS)
     ExternalProject_Add(gflags_ep
       GIT_REPOSITORY https://github.com/gflags/gflags.git
       GIT_TAG cce68f0c9c5d054017425e6e6fd54f696d36e8ee
-      # URL "https://github.com/gflags/gflags/archive/v${GFLAGS_VERSION}.tar.gz"
       BUILD_IN_SOURCE 1
       CMAKE_ARGS -DCMAKE_BUILD_TYPE=${CMAKE_BUILD_TYPE}
                  -DCMAKE_INSTALL_PREFIX=${GFLAGS_PREFIX}

cpp/src/arrow/io/memory.cc

@@ -258,8 +258,11 @@ Status BufferOutputStream::Reserve(int64_t nbytes) {
 // ----------------------------------------------------------------------
 // In-memory buffer reader
 
-BufferReader::BufferReader(const uint8_t* buffer, int buffer_size)
-    : buffer_(buffer), buffer_size_(buffer_size), position_(0) {}
+BufferReader::BufferReader(const std::shared_ptr<Buffer>& buffer)
+    : buffer_(buffer), data_(buffer->data()), size_(buffer->size()), position_(0) {}
+
+BufferReader::BufferReader(const uint8_t* data, int64_t size)
+    : buffer_(nullptr), data_(data), size_(size), position_(0) {}
 
 BufferReader::~BufferReader() {}
 
@@ -278,26 +281,32 @@ bool BufferReader::supports_zero_copy() const {
 }
 
 Status BufferReader::Read(int64_t nbytes, int64_t* bytes_read, uint8_t* buffer) {
-  memcpy(buffer, buffer_ + position_, nbytes);
-  *bytes_read = std::min(nbytes, buffer_size_ - position_);
+  memcpy(buffer, data_ + position_, nbytes);
+  *bytes_read = std::min(nbytes, size_ - position_);
   position_ += *bytes_read;
   return Status::OK();
 }
 
 Status BufferReader::Read(int64_t nbytes, std::shared_ptr<Buffer>* out) {
-  int64_t size = std::min(nbytes, buffer_size_ - position_);
-  *out = std::make_shared<Buffer>(buffer_ + position_, size);
+  int64_t size = std::min(nbytes, size_ - position_);
+
+  if (buffer_ != nullptr) {
+    *out = SliceBuffer(buffer_, position_, size);
+  } else {
+    *out = std::make_shared<Buffer>(data_ + position_, size);
+  }
+
   position_ += nbytes;
   return Status::OK();
 }
 
 Status BufferReader::GetSize(int64_t* size) {
-  *size = buffer_size_;
+  *size = size_;
   return Status::OK();
 }
 
 Status BufferReader::Seek(int64_t position) {
-  if (position < 0 || position >= buffer_size_) {
+  if (position < 0 || position >= size_) {
     return Status::IOError("position out of bounds");
   }
 

cpp/src/arrow/io/memory.h

@@ -99,7 +99,8 @@ class ARROW_EXPORT MemoryMappedFile : public ReadWriteFileInterface {
 
 class ARROW_EXPORT BufferReader : public ReadableFileInterface {
  public:
-  BufferReader(const uint8_t* buffer, int buffer_size);
+  BufferReader(const std::shared_ptr<Buffer>& buffer);
+  BufferReader(const uint8_t* data, int64_t size);
   ~BufferReader();
 
   Status Close() override;
@@ -116,8 +117,9 @@ class ARROW_EXPORT BufferReader : public ReadableFileInterface {
   bool supports_zero_copy() const override;
 
  private:
-  const uint8_t* buffer_;
-  int buffer_size_;
+  std::shared_ptr<Buffer> buffer_;
+  const uint8_t* data_;
+  int64_t size_;
   int64_t position_;
 };