ARROW-387: [C++] Verify zero-copy Buffer slices from BufferReader ret…

…ain reference to parent Buffer

This is stacked on top of the patch for ARROW-294, will rebase.

Author: Wes McKinney <wes.mckinney@twosigma.com>

Closes #266 from wesm/ARROW-387 and squashes the following commits:

061ef8b [Wes McKinney] Verify BufferReader passes on ownership of parent buffer to zero-copy slices
42a83a4 [Wes McKinney] Remove duplicated includes
3928ab0 [Wes McKinney] Base MemoryMappedFile implementation on common OSFile interface. Add test case for ARROW-340.

cpp/src/arrow/io/interfaces.cc

@@ -44,5 +44,10 @@ Status ReadableFileInterface::ReadAt(
   return Read(nbytes, out);
 }
 
+Status Writeable::Write(const std::string& data) {
+  return Write(reinterpret_cast<const uint8_t*>(data.c_str()),
+      static_cast<int64_t>(data.size()));
+}
+
 }  // namespace io
 }  // namespace arrow

cpp/src/arrow/io/interfaces.h

@@ -20,6 +20,7 @@
 
 #include <cstdint>
 #include <memory>
+#include <string>
 
 #include "arrow/util/macros.h"
 #include "arrow/util/visibility.h"
@@ -67,9 +68,11 @@ class Seekable {
   virtual Status Seek(int64_t position) = 0;
 };
 
-class Writeable {
+class ARROW_EXPORT Writeable {
  public:
   virtual Status Write(const uint8_t* data, int64_t nbytes) = 0;
+
+  Status Write(const std::string& data);
 };
 
 class Readable {

cpp/src/arrow/io/io-memory-test.cc

@@ -48,12 +48,33 @@ TEST_F(TestBufferOutputStream, CloseResizes) {
   const int64_t nbytes = static_cast<int64_t>(data.size());
   const int K = 100;
   for (int i = 0; i < K; ++i) {
-    EXPECT_OK(stream_->Write(reinterpret_cast<const uint8_t*>(data.c_str()), nbytes));
+    EXPECT_OK(stream_->Write(data));
   }
 
   ASSERT_OK(stream_->Close());
   ASSERT_EQ(K * nbytes, buffer_->size());
 }
 
+TEST(TestBufferReader, RetainParentReference) {
+  // ARROW-387
+  std::string data = "data123456";
+
+  std::shared_ptr<Buffer> slice1;
+  std::shared_ptr<Buffer> slice2;
+  {
+    auto buffer = std::make_shared<PoolBuffer>();
+    ASSERT_OK(buffer->Resize(static_cast<int64_t>(data.size())));
+    std::memcpy(buffer->mutable_data(), data.c_str(), data.size());
+    BufferReader reader(buffer);
+    ASSERT_OK(reader.Read(4, &slice1));
+    ASSERT_OK(reader.Read(6, &slice2));
+  }
+
+  ASSERT_TRUE(slice1->parent() != nullptr);
+
+  ASSERT_EQ(0, std::memcmp(slice1->data(), data.c_str(), 4));
+  ASSERT_EQ(0, std::memcmp(slice2->data(), data.c_str() + 4, 6));
+}
+
 }  // namespace io
 }  // namespace arrow