Port more code, add basic wrapper implementation for legacy IO interfaces

cpp/src/arrow/io/memory.cc

@@ -62,6 +62,10 @@ Status BufferOutputStream::Reset(int64_t initial_capacity, MemoryPool* pool) {
   return Status::OK();
 }
 
+void BufferOutputStream::Clear() {
+  position_ = 0;
+}
+
 BufferOutputStream::~BufferOutputStream() {
   // This can fail, better to explicitly call close
   if (buffer_) {

cpp/src/arrow/io/memory.h

@@ -72,6 +72,13 @@ class ARROW_EXPORT BufferOutputStream : public OutputStream {
 
   int64_t capacity() const { return capacity_; }
 
+  /// \brief Return reference to current in-progress buffer. May
+  /// become invalid after any Write calls
+  const Buffer& buffer() const { return buffer_; }
+
+  /// \brief Reset position to 0 without modifying capacity or zeroing data
+  void Clear();
+
  private:
   BufferOutputStream();
 

cpp/src/parquet/arrow/reader.cc

@@ -50,7 +50,6 @@
 #include "parquet/properties.h"
 #include "parquet/schema.h"
 #include "parquet/types.h"
-#include "parquet/util/memory.h"
 #include "parquet/util/schema-util.h"
 
 using arrow::Array;
@@ -732,10 +731,8 @@ Status OpenFile(const std::shared_ptr<::arrow::io::RandomAccessFile>& file,
                 MemoryPool* allocator, const ReaderProperties& props,
                 const std::shared_ptr<FileMetaData>& metadata,
                 std::unique_ptr<FileReader>* reader) {
-  std::unique_ptr<RandomAccessSource> io_wrapper(new ArrowInputFile(file));
   std::unique_ptr<ParquetReader> pq_reader;
-  PARQUET_CATCH_NOT_OK(pq_reader =
-                           ParquetReader::Open(std::move(io_wrapper), props, metadata));
+  PARQUET_CATCH_NOT_OK(pq_reader = ParquetReader::Open(file, props, metadata));
   reader->reset(new FileReader(allocator, std::move(pq_reader)));
   return Status::OK();
 }
@@ -749,10 +746,9 @@ Status OpenFile(const std::shared_ptr<::arrow::io::RandomAccessFile>& file,
 Status OpenFile(const std::shared_ptr<::arrow::io::RandomAccessFile>& file,
                 ::arrow::MemoryPool* allocator, const ArrowReaderProperties& properties,
                 std::unique_ptr<FileReader>* reader) {
-  std::unique_ptr<RandomAccessSource> io_wrapper(new ArrowInputFile(file));
   std::unique_ptr<ParquetReader> pq_reader;
   PARQUET_CATCH_NOT_OK(
-      pq_reader = ParquetReader::Open(std::move(io_wrapper),
+      pq_reader = ParquetReader::Open(file,
                                       ::parquet::default_reader_properties(), nullptr));
   reader->reset(new FileReader(allocator, std::move(pq_reader), properties));
   return Status::OK();

cpp/src/parquet/column_reader.cc

@@ -104,9 +104,10 @@ ReaderProperties default_reader_properties() {
 // and the page metadata.
 class SerializedPageReader : public PageReader {
  public:
-  SerializedPageReader(std::unique_ptr<InputStream> stream, int64_t total_num_rows,
+  SerializedPageReader(const std::shared_ptr<ArrowInputStream>& stream,
+                       int64_t total_num_rows,
                        Compression::type codec, ::arrow::MemoryPool* pool)
-      : stream_(std::move(stream)),
+      : stream_(stream),
         decompression_buffer_(AllocateBuffer(pool, 0)),
         seen_num_rows_(0),
         total_num_rows_(total_num_rows) {
@@ -120,7 +121,7 @@ class SerializedPageReader : public PageReader {
   void set_max_page_header_size(uint32_t size) override { max_page_header_size_ = size; }
 
  private:
-  std::unique_ptr<InputStream> stream_;
+  std::shared_ptr<ArrowInputStream> stream_;
 
   format::PageHeader current_page_header_;
   std::shared_ptr<Page> current_page_;
@@ -153,15 +154,15 @@ std::shared_ptr<Page> SerializedPageReader::NextPage() {
     // We try to deserialize a larger buffer progressively
     // until a maximum allowed header limit
     while (true) {
-      buffer = stream_->Peek(allowed_page_size, &bytes_available);
-      if (bytes_available == 0) {
+      auto buffer = stream_->Peek(allowed_page_size);
+      if (buffer.size() == 0) {
         return std::shared_ptr<Page>(nullptr);
       }
 
       // This gets used, then set by DeserializeThriftMsg
-      header_size = static_cast<uint32_t>(bytes_available);
+      header_size = static_cast<uint32_t>(buffer.size());
       try {
-        DeserializeThriftMsg(buffer, &header_size, &current_page_header_);
+        DeserializeThriftMsg(buffer.data(), &header_size, &current_page_header_);
         break;
       } catch (std::exception& e) {
         // Failed to deserialize. Double the allowed page header size and try again
@@ -175,13 +176,14 @@ std::shared_ptr<Page> SerializedPageReader::NextPage() {
       }
     }
     // Advance the stream offset
-    stream_->Advance(header_size);
+    PARQUET_THROW_NOT_OK(stream_->Advance(header_size));
 
     int compressed_len = current_page_header_.compressed_page_size;
     int uncompressed_len = current_page_header_.uncompressed_page_size;
 
     // Read the compressed data page.
-    buffer = stream_->Read(compressed_len, &bytes_read);
+    std::shared_ptr<Buffer> page_buffer;
+    PARQUET_THROW_NOT_OK(stream_->Read(compressed_len, &page_buffer));
     if (bytes_read != compressed_len) {
       std::stringstream ss;
       ss << "Page was smaller (" << bytes_read << ") than expected (" << compressed_len
@@ -198,11 +200,9 @@ std::shared_ptr<Page> SerializedPageReader::NextPage() {
       PARQUET_THROW_NOT_OK(
           decompressor_->Decompress(compressed_len, buffer, uncompressed_len,
                                     decompression_buffer_->mutable_data()));
-      buffer = decompression_buffer_->data();
+      page_buffer = decompression_buffer_;
     }
 
-    auto page_buffer = std::make_shared<Buffer>(buffer, uncompressed_len);
-
     if (current_page_header_.type == format::PageType::DICTIONARY_PAGE) {
       const format::DictionaryPageHeader& dict_header =
           current_page_header_.dictionary_page_header;
@@ -257,12 +257,12 @@ std::shared_ptr<Page> SerializedPageReader::NextPage() {
   return std::shared_ptr<Page>(nullptr);
 }
 
-std::unique_ptr<PageReader> PageReader::Open(std::unique_ptr<InputStream> stream,
+std::unique_ptr<PageReader> PageReader::Open(const std::shared_ptr<ArrowInputStream>& stream,
                                              int64_t total_num_rows,
                                              Compression::type codec,
                                              ::arrow::MemoryPool* pool) {
   return std::unique_ptr<PageReader>(
-      new SerializedPageReader(std::move(stream), total_num_rows, codec, pool));
+      new SerializedPageReader(stream, total_num_rows, codec, pool));
 }
 
 // ----------------------------------------------------------------------

cpp/src/parquet/column_reader.h

@@ -88,7 +88,7 @@ class PARQUET_EXPORT PageReader {
   virtual ~PageReader() = default;
 
   static std::unique_ptr<PageReader> Open(
-      std::unique_ptr<InputStream> stream, int64_t total_num_rows,
+      const std::shared_ptr<ArrowInputStream>& stream, int64_t total_num_rows,
       Compression::type codec,
       ::arrow::MemoryPool* pool = ::arrow::default_memory_pool());
 

cpp/src/parquet/column_writer.cc

@@ -127,8 +127,8 @@ int LevelEncoder::Encode(int batch_size, const int16_t* levels) {
 // and the page metadata.
 class SerializedPageWriter : public PageWriter {
  public:
-  SerializedPageWriter(OutputStream* sink, Compression::type codec,
-                       ColumnChunkMetaDataBuilder* metadata,
+  SerializedPageWriter(const std::shared_ptr<ArrowOutputStream>& sink,
+                       Compression::type codec, ColumnChunkMetaDataBuilder* metadata,
                        ::arrow::MemoryPool* pool = ::arrow::default_memory_pool())
       : sink_(sink),
         metadata_(metadata),
@@ -142,6 +142,18 @@ class SerializedPageWriter : public PageWriter {
     thrift_serializer_.reset(new ThriftSerializer);
   }
 
+  int64_t WritePageGeneric(const format::PageHeader& header, const Buffer& body) {
+    int64_t start_pos = -1;
+    PARQUET_THROW_NOT_OK(sink_->Tell(&start_pos));
+    int64_t header_size = thrift_serializer_->Serialize(&page_header, sink_);
+    PARQUET_THROW_NOT_OK(sink_->Write(body.data(), body.size()));
+
+    total_uncompressed_size_ += uncompressed_size + header_size;
+    total_compressed_size_ += compressed_data->size() + header_size;
+
+    return start_pos;
+  }
+
   int64_t WriteDictionaryPage(const DictionaryPage& page) override {
     int64_t uncompressed_size = page.size();
     std::shared_ptr<Buffer> compressed_data = nullptr;
@@ -164,19 +176,14 @@ class SerializedPageWriter : public PageWriter {
     page_header.__set_uncompressed_page_size(static_cast<int32_t>(uncompressed_size));
     page_header.__set_compressed_page_size(static_cast<int32_t>(compressed_data->size()));
     page_header.__set_dictionary_page_header(dict_page_header);
-    // TODO(PARQUET-594) crc checksum
 
-    int64_t start_pos = sink_->Tell();
+    int64_t start_pos = WritePageGeneric(page_header, *compressed_data);
     if (dictionary_page_offset_ == 0) {
       dictionary_page_offset_ = start_pos;
     }
-    int64_t header_size = thrift_serializer_->Serialize(&page_header, sink_);
-    sink_->Write(compressed_data->data(), compressed_data->size());
-
-    total_uncompressed_size_ += uncompressed_size + header_size;
-    total_compressed_size_ += compressed_data->size() + header_size;
-
-    return sink_->Tell() - start_pos;
+    int64_t current_pos = -1;
+    PARQUET_THROW_NOT_OK(sink_->Tell(&current_pos));
+    return current_pos - start_pos;
   }
 
   void Close(bool has_dictionary, bool fallback) override {
@@ -228,21 +235,14 @@ class SerializedPageWriter : public PageWriter {
     page_header.__set_uncompressed_page_size(static_cast<int32_t>(uncompressed_size));
     page_header.__set_compressed_page_size(static_cast<int32_t>(compressed_data->size()));
     page_header.__set_data_page_header(data_page_header);
-    // TODO(PARQUET-594) crc checksum
 
-    int64_t start_pos = sink_->Tell();
+    int64_t start_pos = WritePageGeneric(page_header, *compressed_data);
     if (data_page_offset_ == 0) {
       data_page_offset_ = start_pos;
     }
-
-    int64_t header_size = thrift_serializer_->Serialize(&page_header, sink_);
-    sink_->Write(compressed_data->data(), compressed_data->size());
-
-    total_uncompressed_size_ += uncompressed_size + header_size;
-    total_compressed_size_ += compressed_data->size() + header_size;
-    num_values_ += page.num_values();
-
-    return sink_->Tell() - start_pos;
+    int64_t current_pos = -1;
+    PARQUET_THROW_NOT_OK(sink_->Tell(&current_pos));
+    return current_pos - start_pos;
   }
 
   bool has_compressor() override { return (compressor_ != nullptr); }
@@ -258,7 +258,7 @@ class SerializedPageWriter : public PageWriter {
   int64_t total_uncompressed_size() { return total_uncompressed_size_; }
 
  private:
-  OutputStream* sink_;
+  std::shared_ptr<ArrowOutputStream> sink_;
   ColumnChunkMetaDataBuilder* metadata_;
   ::arrow::MemoryPool* pool_;
   int64_t num_values_;
@@ -276,13 +276,15 @@ class SerializedPageWriter : public PageWriter {
 // This implementation of the PageWriter writes to the final sink on Close .
 class BufferedPageWriter : public PageWriter {
  public:
-  BufferedPageWriter(OutputStream* sink, Compression::type codec,
+  BufferedPageWriter(const std::shared_ptr<ArrowOutputStream>& sink,
+                     Compression::type codec,
                      ColumnChunkMetaDataBuilder* metadata,
                      ::arrow::MemoryPool* pool = ::arrow::default_memory_pool())
       : final_sink_(sink),
         metadata_(metadata),
-        in_memory_sink_(new InMemoryOutputStream(pool)),
-        pager_(new SerializedPageWriter(in_memory_sink_.get(), codec, metadata, pool)) {}
+        pager_(new SerializedPageWriter(in_memory_sink_.get(), codec, metadata, pool)) {
+    in_memory_sink_ = CreateOutputStream(pool);
+  }
 
   int64_t WriteDictionaryPage(const DictionaryPage& page) override {
     return pager_->WriteDictionaryPage(page);
@@ -299,8 +301,9 @@ class BufferedPageWriter : public PageWriter {
     metadata_->WriteTo(in_memory_sink_.get());
 
     // flush everything to the serialized sink
-    auto buffer = in_memory_sink_->GetBuffer();
-    final_sink_->Write(buffer->data(), buffer->size());
+    std::shared_ptr<Buffer> buffer;
+    PARQUET_THROW_NOT_OK(in_memory_sink_->Finish(&buffer));
+    PARQUET_THROW_NOT_OK(final_sink_->Write(buffer->data(), buffer->size()));
   }
 
   int64_t WriteDataPage(const CompressedDataPage& page) override {
@@ -314,13 +317,14 @@ class BufferedPageWriter : public PageWriter {
   bool has_compressor() override { return pager_->has_compressor(); }
 
  private:
-  OutputStream* final_sink_;
+  std::shared_ptr<ArrowOutputStream> final_sink_;
   ColumnChunkMetaDataBuilder* metadata_;
-  std::unique_ptr<InMemoryOutputStream> in_memory_sink_;
+  std::shared_ptr<ArrowOutputStream> in_memory_sink_;
   std::unique_ptr<SerializedPageWriter> pager_;
 };
 
-std::unique_ptr<PageWriter> PageWriter::Open(OutputStream* sink, Compression::type codec,
+std::unique_ptr<PageWriter> PageWriter::Open(const std::shared_ptr<ArrowOutputStream>& sink,
+                                             Compression::type codec,
                                              ColumnChunkMetaDataBuilder* metadata,
                                              ::arrow::MemoryPool* pool,
                                              bool buffered_row_group) {
@@ -361,7 +365,8 @@ class ColumnWriterImpl {
         total_compressed_bytes_(0),
         closed_(false),
         fallback_(false) {
-    definition_levels_sink_.reset(new InMemoryOutputStream(allocator_));
+    definition_levels_sink_ = CreateOutputStream(allocator_);
+    repetition_levels_sink_ = CreateOutputStream(allocator_);
     repetition_levels_sink_.reset(new InMemoryOutputStream(allocator_));
     definition_levels_rle_ =
         std::static_pointer_cast<ResizableBuffer>(AllocateBuffer(allocator_, 0));
@@ -404,15 +409,15 @@ class ColumnWriterImpl {
   // Write multiple definition levels
   void WriteDefinitionLevels(int64_t num_levels, const int16_t* levels) {
     DCHECK(!closed_);
-    definition_levels_sink_->Write(reinterpret_cast<const uint8_t*>(levels),
-                                   sizeof(int16_t) * num_levels);
+    PARQUET_THROW_NOT_OK(definition_levels_sink_->Write(
+        reinterpret_cast<const uint8_t*>(levels), sizeof(int16_t) * num_levels));
   }
 
   // Write multiple repetition levels
   void WriteRepetitionLevels(int64_t num_levels, const int16_t* levels) {
     DCHECK(!closed_);
-    repetition_levels_sink_->Write(reinterpret_cast<const uint8_t*>(levels),
-                                   sizeof(int16_t) * num_levels);
+    PARQUET_THROW_NOT_OK(repetition_levels_sink_->Write(
+        reinterpret_cast<const uint8_t*>(levels), sizeof(int16_t) * num_levels));
   }
 
   // RLE encode the src_buffer into dest_buffer and return the encoded size
@@ -462,8 +467,8 @@ class ColumnWriterImpl {
   // Flag to infer if dictionary encoding has fallen back to PLAIN
   bool fallback_;
 
-  std::unique_ptr<InMemoryOutputStream> definition_levels_sink_;
-  std::unique_ptr<InMemoryOutputStream> repetition_levels_sink_;
+  std::shared_ptr<ArrowOutputStream> definition_levels_sink_;
+  std::shared_ptr<ArrowOutputStream> repetition_levels_sink_;
 
   std::shared_ptr<ResizableBuffer> definition_levels_rle_;
   std::shared_ptr<ResizableBuffer> repetition_levels_rle_;
@@ -513,13 +518,13 @@ void ColumnWriterImpl::AddDataPage() {
 
   if (descr_->max_definition_level() > 0) {
     definition_levels_rle_size =
-        RleEncodeLevels(definition_levels_sink_->GetBufferRef(),
+        RleEncodeLevels(definition_levels_sink_->buffer(),
                         definition_levels_rle_.get(), descr_->max_definition_level());
   }
 
   if (descr_->max_repetition_level() > 0) {
     repetition_levels_rle_size =
-        RleEncodeLevels(repetition_levels_sink_->GetBufferRef(),
+        RleEncodeLevels(repetition_levels_sink_->buffer(),
                         repetition_levels_rle_.get(), descr_->max_repetition_level());
   }
 

cpp/src/parquet/column_writer.h

@@ -83,7 +83,8 @@ class PARQUET_EXPORT PageWriter {
   virtual ~PageWriter() {}
 
   static std::unique_ptr<PageWriter> Open(
-      OutputStream* sink, Compression::type codec, ColumnChunkMetaDataBuilder* metadata,
+      const std::shared_ptr<ArrowOutputStream>& sink,
+      Compression::type codec, ColumnChunkMetaDataBuilder* metadata,
       ::arrow::MemoryPool* pool = ::arrow::default_memory_pool(),
       bool buffered_row_group = false);
 

cpp/src/parquet/deprecated_io-test.cc

@@ -186,49 +186,4 @@ TEST_F(TestBufferedInputStream, BufferExactlyExhausted) {
   }
 }
 
-TEST(TestArrowInputFile, ReadAt) {
-  std::string data = "this is the data";
-
-  auto file = std::make_shared<::arrow::io::BufferReader>(data);
-  auto source = std::make_shared<ArrowInputFile>(file);
-
-  ASSERT_EQ(0, source->Tell());
-
-  uint8_t buffer[50];
-
-  ASSERT_NO_THROW(source->ReadAt(0, 4, buffer));
-  ASSERT_EQ(0, std::memcmp(buffer, "this", 4));
-
-  // Note: it's undefined (and possibly platform-dependent) whether ArrowInputFile
-  // updates the file position after ReadAt().
-}
-
-TEST(TestArrowInputFile, Read) {
-  std::string data = "this is the data";
-  auto data_buffer = reinterpret_cast<const uint8_t*>(data.c_str());
-
-  auto file = std::make_shared<::arrow::io::BufferReader>(data);
-  auto source = std::make_shared<ArrowInputFile>(file);
-
-  ASSERT_EQ(0, source->Tell());
-
-  std::shared_ptr<Buffer> pq_buffer, expected_buffer;
-
-  ASSERT_NO_THROW(pq_buffer = source->Read(4));
-  expected_buffer = std::make_shared<Buffer>(data_buffer, 4);
-  ASSERT_TRUE(expected_buffer->Equals(*pq_buffer.get()));
-
-  ASSERT_NO_THROW(pq_buffer = source->Read(7));
-  expected_buffer = std::make_shared<Buffer>(data_buffer + 4, 7);
-  ASSERT_TRUE(expected_buffer->Equals(*pq_buffer.get()));
-
-  ASSERT_EQ(11, source->Tell());
-
-  ASSERT_NO_THROW(pq_buffer = source->Read(8));
-  expected_buffer = std::make_shared<Buffer>(data_buffer + 11, 5);
-  ASSERT_TRUE(expected_buffer->Equals(*pq_buffer.get()));
-
-  ASSERT_EQ(16, source->Tell());
-}
-
 }  // namespace parquet

cpp/src/parquet/deprecated_io.cc

@@ -15,7 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
-#include "parquet/util/memory.h"
+#include "parquet/deprecated_io.h"
 
 #include <algorithm>
 #include <cstdint>
@@ -24,219 +24,96 @@
 #include <utility>
 
 #include "arrow/status.h"
-#include "arrow/util/bit-util.h"
 
-#include "arrow/util/compression.h"
-#include "arrow/util/logging.h"
 #include "parquet/exception.h"
 #include "parquet/types.h"
 
-using arrow::MemoryPool;
-using arrow::util::Codec;
-
 namespace parquet {
 
-// ----------------------------------------------------------------------
-// Arrow IO wrappers
+ParquetInputWrapper::ParquetInputWrapper(std::unique_ptr<RandomAccessSource> source,
+                    ::arrow::MemoryPool* pool)
+    : ParquetInputWrapper(source.get(), pool),
+      owned_source_(std::move(source)) {}
 
-void ArrowFileMethods::Close() {
-  // Closing the file is the responsibility of the owner of the handle
-  return;
-}
+ParquetInputWrapper::ParquetInputWrapper(RandomAccessSource* source,
+                                         ::arrow::MemoryPool* pool)
+    : source_(source), pool_(pool), closed_(false) {}
 
-// Return the current position in the output stream relative to the start
-int64_t ArrowFileMethods::Tell() {
-  int64_t position = 0;
-  PARQUET_THROW_NOT_OK(file_interface()->Tell(&position));
-  return position;
-}
-
-ArrowInputFile::ArrowInputFile(const std::shared_ptr<::arrow::io::RandomAccessFile>& file)
-    : file_(file) {}
-
-::arrow::io::FileInterface* ArrowInputFile::file_interface() { return file_.get(); }
-
-int64_t ArrowInputFile::Size() const {
-  int64_t size;
-  PARQUET_THROW_NOT_OK(file_->GetSize(&size));
-  return size;
-}
-
-// Returns bytes read
-int64_t ArrowInputFile::Read(int64_t nbytes, uint8_t* out) {
-  int64_t bytes_read = 0;
-  PARQUET_THROW_NOT_OK(file_->Read(nbytes, &bytes_read, out));
-  return bytes_read;
-}
-
-std::shared_ptr<Buffer> ArrowInputFile::Read(int64_t nbytes) {
-  std::shared_ptr<Buffer> out;
-  PARQUET_THROW_NOT_OK(file_->Read(nbytes, &out));
-  return out;
-}
-
-std::shared_ptr<Buffer> ArrowInputFile::ReadAt(int64_t position, int64_t nbytes) {
-  std::shared_ptr<Buffer> out;
-  PARQUET_THROW_NOT_OK(file_->ReadAt(position, nbytes, &out));
-  return out;
-}
-
-int64_t ArrowInputFile::ReadAt(int64_t position, int64_t nbytes, uint8_t* out) {
-  int64_t bytes_read = 0;
-  PARQUET_THROW_NOT_OK(file_->ReadAt(position, nbytes, &bytes_read, out));
-  return bytes_read;
-}
-
-ArrowOutputStream::ArrowOutputStream(
-    const std::shared_ptr<::arrow::io::OutputStream> file)
-    : file_(file) {}
-
-::arrow::io::FileInterface* ArrowOutputStream::file_interface() { return file_.get(); }
-
-// Copy bytes into the output stream
-void ArrowOutputStream::Write(const uint8_t* data, int64_t length) {
-  PARQUET_THROW_NOT_OK(file_->Write(data, length));
-}
-
-// ----------------------------------------------------------------------
-// InMemoryInputStream
-
-InMemoryInputStream::InMemoryInputStream(const std::shared_ptr<Buffer>& buffer)
-    : buffer_(buffer), offset_(0) {
-  len_ = buffer_->size();
-}
-
-InMemoryInputStream::InMemoryInputStream(RandomAccessSource* source, int64_t start,
-                                         int64_t num_bytes)
-    : offset_(0) {
-  buffer_ = source->ReadAt(start, num_bytes);
-  if (buffer_->size() < num_bytes) {
-    throw ParquetException("Unable to read column chunk data");
-  }
-  len_ = buffer_->size();
-}
-
-const uint8_t* InMemoryInputStream::Peek(int64_t num_to_peek, int64_t* num_bytes) {
-  *num_bytes = std::min(static_cast<int64_t>(num_to_peek), len_ - offset_);
-  return buffer_->data() + offset_;
-}
-
-const uint8_t* InMemoryInputStream::Read(int64_t num_to_read, int64_t* num_bytes) {
-  const uint8_t* result = Peek(num_to_read, num_bytes);
-  offset_ += *num_bytes;
-  return result;
-}
-
-void InMemoryInputStream::Advance(int64_t num_bytes) { offset_ += num_bytes; }
-
-// ----------------------------------------------------------------------
-// In-memory output stream
-
-InMemoryOutputStream::InMemoryOutputStream(MemoryPool* pool, int64_t initial_capacity)
-    : size_(0), capacity_(initial_capacity) {
-  if (initial_capacity == 0) {
-    initial_capacity = kInMemoryDefaultCapacity;
-  }
-  buffer_ = AllocateBuffer(pool, initial_capacity);
-}
-
-InMemoryOutputStream::~InMemoryOutputStream() {}
-
-uint8_t* InMemoryOutputStream::Head() { return buffer_->mutable_data() + size_; }
-
-void InMemoryOutputStream::Write(const uint8_t* data, int64_t length) {
-  if (size_ + length > capacity_) {
-    int64_t new_capacity = capacity_ * 2;
-    while (new_capacity < size_ + length) {
-      new_capacity *= 2;
-    }
-    PARQUET_THROW_NOT_OK(buffer_->Resize(new_capacity));
-    capacity_ = new_capacity;
-  }
-  // If length == 0, data may be null
-  if (length > 0) {
-    memcpy(Head(), data, length);
-    size_ += length;
+ParquetInputWrapper::~ParquetInputWrapper() {
+  if (!closed_) {
+    source_->Close();
+    closed_ = true;
   }
 }
 
-int64_t InMemoryOutputStream::Tell() { return size_; }
-
-std::shared_ptr<Buffer> InMemoryOutputStream::GetBuffer() {
-  PARQUET_THROW_NOT_OK(buffer_->Resize(size_));
-  std::shared_ptr<Buffer> result = buffer_;
-  buffer_ = nullptr;
-  return result;
+::arrow::Status ParquetInputWrapper::Close() {
+  PARQUET_CATCH_NOT_OK(source_->Close());
+  closed_ = true;
+  return ::arrow::Status::OK();
 }
 
-// ----------------------------------------------------------------------
-// BufferedInputStream
-
-BufferedInputStream::BufferedInputStream(MemoryPool* pool, int64_t buffer_size,
-                                         RandomAccessSource* source, int64_t start,
-                                         int64_t num_bytes)
-    : source_(source), stream_offset_(start), stream_end_(start + num_bytes) {
-  buffer_ = AllocateBuffer(pool, buffer_size);
-  buffer_offset_ = 0;
-  buffer_end_ = 0;
+::arrow::Status ParquetInputWrapper::Tell(int64_t* position) const {
+  PARQUET_CATCH_NOT_OK(*position = source_->Tell());
+  return ::arrow::Status::OK();
 }
 
-int64_t BufferedInputStream::remaining_in_buffer() const {
-  return buffer_end_ - buffer_offset_;
+bool ParquetInputWrapper::closed() const {
+  return closed_;
 }
 
-const uint8_t* BufferedInputStream::Peek(int64_t num_to_peek, int64_t* num_bytes) {
-  int64_t buffer_avail = buffer_end_ - buffer_offset_;
-  int64_t stream_avail = stream_end_ - stream_offset_;
-  // Do not try to peek more than the total remaining number of bytes.
-  *num_bytes = std::min(num_to_peek, buffer_avail + stream_avail);
-  // Increase the buffer size if needed
-  if (*num_bytes > buffer_->size() - buffer_offset_) {
-    // XXX Should adopt a shrinking heuristic if buffer_offset_ is close
-    // to buffer_end_.
-    PARQUET_THROW_NOT_OK(buffer_->Resize(*num_bytes + buffer_offset_));
-    DCHECK(buffer_->size() - buffer_offset_ >= *num_bytes);
-  }
-  // Read more data when buffer has insufficient left
-  if (*num_bytes > buffer_avail) {
-    // Read as much as possible to fill the buffer, but not past stream end
-    int64_t read_size = std::min(buffer_->size() - buffer_end_, stream_avail);
-    int64_t bytes_read =
-        source_->ReadAt(stream_offset_, read_size, buffer_->mutable_data() + buffer_end_);
-    stream_offset_ += bytes_read;
-    buffer_end_ += bytes_read;
-    if (bytes_read < read_size) {
-      throw ParquetException("Failed reading column data from source");
-    }
-  }
-  DCHECK(*num_bytes <= buffer_end_ - buffer_offset_);  // Enough bytes available
-  return buffer_->data() + buffer_offset_;
+::arrow::Status ParquetInputWrapper::Seek(int64_t position) {
+  PARQUET_CATCH_NOT_OK(source_->Seek(position));
+  return ::arrow::Status::OK();
 }
 
-const uint8_t* BufferedInputStream::Read(int64_t num_to_read, int64_t* num_bytes) {
-  const uint8_t* result = Peek(num_to_read, num_bytes);
-  buffer_offset_ += *num_bytes;
-  if (buffer_offset_ == buffer_end_) {
-    // Rewind pointer to reuse beginning of buffer
-    buffer_offset_ = 0;
-    buffer_end_ = 0;
-  }
-  return result;
+::arrow::Status ParquetInputWrapper::Read(int64_t nbytes, int64_t* bytes_read, void* out) {
+  PARQUET_CATCH_NOT_OK(*bytes_read = source_->Read(nbytes, out));
+  return ::arrow::Status::OK();
 }
 
-void BufferedInputStream::Advance(int64_t num_bytes) {
-  buffer_offset_ += num_bytes;
-  if (buffer_offset_ == buffer_end_) {
-    // Rewind pointer to reuse beginning of buffer
-    buffer_offset_ = 0;
-    buffer_end_ = 0;
+::arrow::Status ParquetInputWrapper::Read(int64_t nbytes, std::shared_ptr<Buffer>* out) {
+  PARQUET_CATCH_NOT_OK(*out = source_->Read(nbytes));
+  return ::arrow::Status::OK();
+}
+
+::arrow::Status ParquetInputWrapper::ReadAt(int64_t position, int64_t nbytes,
+                                            std::shared_ptr<Buffer>* out) {
+  PARQUET_CATCH_NOT_OK(*out = source_->ReadAt(position, nbytes));
+  return ::arrow::Status::OK();
+}
+
+ParquetOutputWrapper::ParquetOutputWrapper(std::unique_ptr<OutputStream> sink)
+    : ParquetOutputWrapper(sink.get()),
+      owned_sink_(std::move(sink)) {}
+
+ParquetOutputWrapper::ParquetOutputWrapper(OutputStream* sink)
+    : sink_(sink), closed_(false) {}
+
+ParquetOutputWrapper::~ParquetOutputWrapper() {
+  if (!closed_) {
+    sink_->Close();
+    closed_ = true;
   }
 }
 
-std::shared_ptr<ResizableBuffer> AllocateBuffer(MemoryPool* pool, int64_t size) {
-  std::shared_ptr<ResizableBuffer> result;
-  PARQUET_THROW_NOT_OK(arrow::AllocateResizableBuffer(pool, size, &result));
-  return result;
+::arrow::Status ParquetOutputWrapper::Close() {
+  PARQUET_CATCH_NOT_OK(sink_->Close());
+  closed_ = true;
+  return ::arrow::Status::OK();
+}
+
+::arrow::Status ParquetOutputWrapper::Tell(int64_t* position) const {
+  PARQUET_CATCH_NOT_OK(*position = source_->Tell());
+  return ::arrow::Status::OK();
+}
+
+bool ParquetOutputWrapper::closed() const {
+  return closed_;
+}
+
+::arrow::Status ParquetOutputWrapper::Write(const void* data, int64_t nbytes) {
+  PARQUET_CATCH_NOT_OK(sink_->Write(reinterpret_cast<const uint8_t*>(data), nbytes));
+  return ::arrow::Status::OK();
 }
 
 }  // namespace parquet

cpp/src/parquet/deprecated_io.h

@@ -15,10 +15,13 @@
 // specific language governing permissions and limitations
 // under the License.
 
-#ifndef PARQUET_UTIL_MEMORY_H
-#define PARQUET_UTIL_MEMORY_H
+// DEPRECATED IO INTERFACES: We have transitioned to using the Apache
+// Arrow file input and output abstract interfaces defined in
+// arrow/io/interfaces.h. These legacy interfaces are being preserved
+// through a wrapper layer for one to two releases
+
+#pragma
 
-#include <atomic>
 #include <cstdint>
 #include <cstdlib>
 #include <cstring>
@@ -37,8 +40,6 @@
 
 namespace parquet {
 
-static constexpr int64_t kInMemoryDefaultCapacity = 1024;
-
 // File input and output interfaces that translate arrow::Status to exceptions
 
 class PARQUET_EXPORT FileInterface {
@@ -79,179 +80,58 @@ class PARQUET_EXPORT OutputStream : virtual public FileInterface {
   virtual void Write(const uint8_t* data, int64_t length) = 0;
 };
 
-class PARQUET_EXPORT ArrowFileMethods : virtual public FileInterface {
- public:
-  // No-op. Closing the file is the responsibility of the owner of the handle
-  void Close() override;
-
-  int64_t Tell() override;
-
- protected:
-  virtual ::arrow::io::FileInterface* file_interface() = 0;
-};
-
-// Suppress C4250 warning caused by diamond inheritance
-#ifdef _MSC_VER
-#pragma warning(push)
-#pragma warning(disable : 4250)
-#endif
-
-/// This interface depends on the threadsafety of the underlying Arrow file interface
-class PARQUET_EXPORT ArrowInputFile : public ArrowFileMethods, public RandomAccessSource {
- public:
-  explicit ArrowInputFile(const std::shared_ptr<::arrow::io::RandomAccessFile>& file);
-
-  int64_t Size() const override;
-
-  // Returns bytes read
-  int64_t Read(int64_t nbytes, uint8_t* out) override;
-
-  std::shared_ptr<Buffer> Read(int64_t nbytes) override;
-
-  std::shared_ptr<Buffer> ReadAt(int64_t position, int64_t nbytes) override;
-
-  /// Returns bytes read
-  int64_t ReadAt(int64_t position, int64_t nbytes, uint8_t* out) override;
-
-  std::shared_ptr<::arrow::io::RandomAccessFile> file() const { return file_; }
-
-  // Diamond inheritance
-  using ArrowFileMethods::Close;
-  using ArrowFileMethods::Tell;
-
- private:
-  ::arrow::io::FileInterface* file_interface() override;
-  std::shared_ptr<::arrow::io::RandomAccessFile> file_;
-};
-
-class PARQUET_EXPORT ArrowOutputStream : public ArrowFileMethods, public OutputStream {
- public:
-  explicit ArrowOutputStream(const std::shared_ptr<::arrow::io::OutputStream> file);
-
-  // Copy bytes into the output stream
-  void Write(const uint8_t* data, int64_t length) override;
-
-  std::shared_ptr<::arrow::io::OutputStream> file() { return file_; }
-
-  // Diamond inheritance
-  using ArrowFileMethods::Close;
-  using ArrowFileMethods::Tell;
-
- private:
-  ::arrow::io::FileInterface* file_interface() override;
-  std::shared_ptr<::arrow::io::OutputStream> file_;
-};
-
-// Pop C4250 pragma
-#ifdef _MSC_VER
-#pragma warning(pop)
-#endif
-
-class PARQUET_EXPORT InMemoryOutputStream : public OutputStream {
- public:
-  explicit InMemoryOutputStream(
-      ::arrow::MemoryPool* pool = ::arrow::default_memory_pool(),
-      int64_t initial_capacity = kInMemoryDefaultCapacity);
-
-  virtual ~InMemoryOutputStream();
-
-  // Close is currently a no-op with the in-memory stream
-  virtual void Close() {}
-
-  virtual int64_t Tell();
-
-  virtual void Write(const uint8_t* data, int64_t length);
-
-  // Clears the stream
-  void Clear() { size_ = 0; }
-
-  // Get pointer to the underlying buffer
-  const Buffer& GetBufferRef() const { return *buffer_; }
-
-  // Return complete stream as Buffer
-  std::shared_ptr<Buffer> GetBuffer();
-
- private:
-  // Mutable pointer to the current write position in the stream
-  uint8_t* Head();
-
-  std::shared_ptr<ResizableBuffer> buffer_;
-  int64_t size_;
-  int64_t capacity_;
-
-  PARQUET_DISALLOW_COPY_AND_ASSIGN(InMemoryOutputStream);
-};
-
 // ----------------------------------------------------------------------
-// Streaming input interfaces
+// Wrapper classes
 
-// Interface for the column reader to get the bytes. The interface is a stream
-// interface, meaning the bytes in order and once a byte is read, it does not
-// need to be read again.
-class PARQUET_EXPORT InputStream {
+class ParquetInputWrapper : public ::arrow::io::RandomAccessFile {
  public:
-  // Returns the next 'num_to_peek' without advancing the current position.
-  // *num_bytes will contain the number of bytes returned which can only be
-  // less than num_to_peek at end of stream cases.
-  // Since the position is not advanced, calls to this function are idempotent.
-  // The buffer returned to the caller is still owned by the input stream and must
-  // stay valid until the next call to Peek() or Read().
-  virtual const uint8_t* Peek(int64_t num_to_peek, int64_t* num_bytes) = 0;
+  ParquetInputWrapper(std::unique_ptr<RandomAccessSource> source,
+                      ::arrow::MemoryPool* pool = ::arrow::default_memory_pool());
+  ParquetInputWrapper(RandomAccessSource* source,
+                      ::arrow::MemoryPool* pool = ::arrow::default_memory_pool());
 
-  // Identical to Peek(), except the current position in the stream is advanced by
-  // *num_bytes.
-  virtual const uint8_t* Read(int64_t num_to_read, int64_t* num_bytes) = 0;
+  ~ParquetInputWrapper() override;
 
-  // Advance the stream without reading
-  virtual void Advance(int64_t num_bytes) = 0;
+  // FileInterface
+  ::arrow::Status Close() override;
+  ::arrow::Status Tell(int64_t* position) const override;
+  bool closed() const override;
 
-  virtual ~InputStream() {}
+  // Seekable
+  ::arrow::Status Seek(int64_t position) override;
 
- protected:
-  InputStream() {}
-};
-
-// Implementation of an InputStream when all the bytes are in memory.
-class PARQUET_EXPORT InMemoryInputStream : public InputStream {
- public:
-  InMemoryInputStream(RandomAccessSource* source, int64_t start, int64_t end);
-  explicit InMemoryInputStream(const std::shared_ptr<Buffer>& buffer);
-  virtual const uint8_t* Peek(int64_t num_to_peek, int64_t* num_bytes);
-  virtual const uint8_t* Read(int64_t num_to_read, int64_t* num_bytes);
-
-  virtual void Advance(int64_t num_bytes);
+  // InputStream
+  ::arrow::Status Read(int64_t nbytes, int64_t* bytes_read, void* out) override;
+  ::arrow::Status Read(int64_t nbytes, std::shared_ptr<Buffer>* out) override;
+  ::arrow::Status ReadAt(int64_t position, int64_t nbytes,
+                         std::shared_ptr<Buffer>* out) override;
 
  private:
-  std::shared_ptr<Buffer> buffer_;
-  int64_t len_;
-  int64_t offset_;
-};
-
-// Implementation of an InputStream when only some of the bytes are in memory.
-class PARQUET_EXPORT BufferedInputStream : public InputStream {
- public:
-  BufferedInputStream(::arrow::MemoryPool* pool, int64_t buffer_size,
-                      RandomAccessSource* source, int64_t start, int64_t end);
-  virtual const uint8_t* Peek(int64_t num_to_peek, int64_t* num_bytes);
-  virtual const uint8_t* Read(int64_t num_to_read, int64_t* num_bytes);
-
-  virtual void Advance(int64_t num_bytes);
-
-  // Return the number of bytes remaining in buffer (i.e. without reading from source).
-  int64_t remaining_in_buffer() const;
-
- private:
-  std::shared_ptr<ResizableBuffer> buffer_;
+  std::unique_ptr<RandomAccessSource> owned_source_;
   RandomAccessSource* source_;
-  int64_t stream_offset_;
-  int64_t stream_end_;
-  int64_t buffer_offset_;
-  int64_t buffer_end_;
+  MemoryPool* pool_;
+  bool closed_;
 };
 
-std::shared_ptr<ResizableBuffer> PARQUET_EXPORT AllocateBuffer(
-    ::arrow::MemoryPool* pool = ::arrow::default_memory_pool(), int64_t size = 0);
+class ParquetOutputWrapper : public ::arrow::io::OutputStream {
+ public:
+  ParquetOutputWrapper(std::unique_ptr<OutputStream> sink);
+  ParquetOutputWrapper(OutputStream* sink);
+
+  ~ParquetOutputWrapper() override;
+
+  // FileInterface
+  ::arrow::Status Close() override;
+  ::arrow::Status Tell(int64_t* position) const override;
+  bool closed() const override;
+
+  // Writable
+  ::arrow::Status Write(const void* data, int64_t nbytes) override;
+
+ private:
+  std::unique_ptr<OutputStream> owned_sink_;
+  OutputStream* sink_;
+  bool closed_;
+};
 
 }  // namespace parquet
-
-#endif  // PARQUET_UTIL_MEMORY_H

cpp/src/parquet/encoding.cc

@@ -82,41 +82,47 @@ class PlainEncoder : public EncoderImpl, virtual public TypedEncoder<DType> {
   void Put(const T* buffer, int num_values) override;
 
  protected:
-  std::unique_ptr<InMemoryOutputStream> values_sink_;
+  std::shared_ptr<ArrowOutputStream> values_sink_;
 };
 
 template <typename DType>
 PlainEncoder<DType>::PlainEncoder(const ColumnDescriptor* descr,
                                   ::arrow::MemoryPool* pool)
     : EncoderImpl(descr, Encoding::PLAIN, pool) {
-  values_sink_.reset(new InMemoryOutputStream(pool));
+  values_sink_ = CreateOutputStream(pool);
 }
 template <typename DType>
 int64_t PlainEncoder<DType>::EstimatedDataEncodedSize() {
-  return values_sink_->Tell();
+  int64_t position = -1;
+  PARQUET_THROW_NOT_OK(values_sink_->Tell(&position));
+  return position;
 }
 
 template <typename DType>
 std::shared_ptr<Buffer> PlainEncoder<DType>::FlushValues() {
-  std::shared_ptr<Buffer> buffer = values_sink_->GetBuffer();
-  values_sink_.reset(new InMemoryOutputStream(this->pool_));
+  std::shared_ptr<Buffer> buffer;
+  PARQUET_THROW_NOT_OK(values_sink_->Finish(&buffer));
+  values_sink_ = CreateOutputStream(this->pool_);
   return buffer;
 }
 
 template <typename DType>
 void PlainEncoder<DType>::Put(const T* buffer, int num_values) {
-  values_sink_->Write(reinterpret_cast<const uint8_t*>(buffer), num_values * sizeof(T));
+  PARQUET_THROW_NOT_OK(values_sink_->Write(reinterpret_cast<const uint8_t*>(buffer),
+                                           num_values * sizeof(T)));
 }
 
 template <>
 inline void PlainEncoder<ByteArrayType>::Put(const ByteArray* src, int num_values) {
   for (int i = 0; i < num_values; ++i) {
     // Write the result to the output stream
-    values_sink_->Write(reinterpret_cast<const uint8_t*>(&src[i].len), sizeof(uint32_t));
+    PARQUET_THROW_NOT_OK(values_sink_->Write(
+        reinterpret_cast<const uint8_t*>(&src[i].len), sizeof(uint32_t)));
     if (src[i].len > 0) {
       DCHECK(nullptr != src[i].ptr) << "Value ptr cannot be NULL";
     }
-    values_sink_->Write(reinterpret_cast<const uint8_t*>(src[i].ptr), src[i].len);
+    PARQUET_THROW_NOT_OK(values_sink_->Write(
+        reinterpret_cast<const uint8_t*>(src[i].ptr), src[i].len));
   }
 }
 
@@ -127,8 +133,8 @@ inline void PlainEncoder<FLBAType>::Put(const FixedLenByteArray* src, int num_va
     if (descr_->type_length() > 0) {
       DCHECK(nullptr != src[i].ptr) << "Value ptr cannot be NULL";
     }
-    values_sink_->Write(reinterpret_cast<const uint8_t*>(src[i].ptr),
-                        descr_->type_length());
+    PARQUET_THROW_NOT_OK(values_sink_->Write(
+        reinterpret_cast<const uint8_t*>(src[i].ptr), descr_->type_length()));
   }
 }
 
@@ -163,7 +169,7 @@ class PlainBooleanEncoder : public EncoderImpl,
   int bits_available_;
   std::unique_ptr<::arrow::BitUtil::BitWriter> bit_writer_;
   std::shared_ptr<ResizableBuffer> bits_buffer_;
-  std::unique_ptr<InMemoryOutputStream> values_sink_;
+  std::shared_ptr<ArrowOutputStream> values_sink_;
 
   template <typename SequenceType>
   void PutImpl(const SequenceType& src, int num_values);
@@ -182,7 +188,8 @@ void PlainBooleanEncoder::PutImpl(const SequenceType& src, int num_values) {
 
     if (bits_available_ == 0) {
       bit_writer_->Flush();
-      values_sink_->Write(bit_writer_->buffer(), bit_writer_->bytes_written());
+      PARQUET_THROW_NOT_OK(values_sink_->Write(
+          bit_writer_->buffer(), bit_writer_->bytes_written()));
       bit_writer_->Clear();
     }
   }
@@ -201,7 +208,8 @@ void PlainBooleanEncoder::PutImpl(const SequenceType& src, int num_values) {
 
     if (bits_available_ == 0) {
       bit_writer_->Flush();
-      values_sink_->Write(bit_writer_->buffer(), bit_writer_->bytes_written());
+      PARQUET_THROW_NOT_OK(values_sink_->Write(
+          bit_writer_->buffer(), bit_writer_->bytes_written()));
       bit_writer_->Clear();
     }
   }
@@ -211,26 +219,30 @@ PlainBooleanEncoder::PlainBooleanEncoder(const ColumnDescriptor* descr,
                                          ::arrow::MemoryPool* pool)
     : EncoderImpl(descr, Encoding::PLAIN, pool),
       bits_available_(kInMemoryDefaultCapacity * 8),
-      bits_buffer_(AllocateBuffer(pool, kInMemoryDefaultCapacity)),
-      values_sink_(new InMemoryOutputStream(pool)) {
+      bits_buffer_(AllocateBuffer(pool, kInMemoryDefaultCapacity)) {
+  values_sink_ = CreateOutputStream(pool);
   bit_writer_.reset(new BitUtil::BitWriter(bits_buffer_->mutable_data(),
                                            static_cast<int>(bits_buffer_->size())));
 }
 
 int64_t PlainBooleanEncoder::EstimatedDataEncodedSize() {
-  return values_sink_->Tell() + bit_writer_->bytes_written();
+  int64_t position = -1;
+  PARQUET_THROW_NOT_OK(values_sink_->Tell(&position));
+  return position + bit_writer_->bytes_written();
 }
 
 std::shared_ptr<Buffer> PlainBooleanEncoder::FlushValues() {
   if (bits_available_ > 0) {
     bit_writer_->Flush();
-    values_sink_->Write(bit_writer_->buffer(), bit_writer_->bytes_written());
+    PARQUET_THROW_NOT_OK(values_sink_->Write(
+        bit_writer_->buffer(), bit_writer_->bytes_written()));
     bit_writer_->Clear();
     bits_available_ = static_cast<int>(bits_buffer_->size()) * 8;
   }
 
-  std::shared_ptr<Buffer> buffer = values_sink_->GetBuffer();
-  values_sink_.reset(new InMemoryOutputStream(this->pool_));
+  std::shared_ptr<Buffer> buffer;
+  PARQUET_THROW_NOT_OK(values_sink_->Finish(&buffer));
+  values_sink_ = CreateOutputStream(this->pool_);
   return buffer;
 }
 

cpp/src/parquet/file-deserialize-test.cc

@@ -105,12 +105,14 @@ class TestPageSerde : public ::testing::Test {
     ASSERT_NO_THROW(serializer.Serialize(&page_header_, out_stream_.get()));
   }
 
-  void ResetStream() { out_stream_.reset(new InMemoryOutputStream); }
+  void ResetStream() {
+    out_stream_ = CreateOutputStream();
+  }
 
   void EndStream() { out_buffer_ = out_stream_->GetBuffer(); }
 
  protected:
-  std::unique_ptr<InMemoryOutputStream> out_stream_;
+  std::shared_ptr<::arrow::io::BufferOutputStream> out_stream_;
   std::shared_ptr<Buffer> out_buffer_;
 
   std::unique_ptr<PageReader> page_reader_;
@@ -325,18 +327,19 @@ TEST_F(TestParquetFileReader, InvalidFooter) {
 }
 
 TEST_F(TestParquetFileReader, IncompleteMetadata) {
-  InMemoryOutputStream stream;
+  std::shared_ptr<ArrowOutputStream> stream = CreateOutputStream();
 
   const char* magic = "PAR1";
 
-  stream.Write(reinterpret_cast<const uint8_t*>(magic), strlen(magic));
+  ASSERT_OK(stream->Write(reinterpret_cast<const uint8_t*>(magic), strlen(magic)));
   std::vector<uint8_t> bytes(10);
-  stream.Write(bytes.data(), bytes.size());
+  ASSERT_OK(stream->Write(bytes.data(), bytes.size()));
   uint32_t metadata_len = 24;
-  stream.Write(reinterpret_cast<const uint8_t*>(&metadata_len), sizeof(uint32_t));
-  stream.Write(reinterpret_cast<const uint8_t*>(magic), strlen(magic));
+  ASSERT_OK(stream->Write(reinterpret_cast<const uint8_t*>(&metadata_len), sizeof(uint32_t)));
+  ASSERT_OK(stream->Write(reinterpret_cast<const uint8_t*>(magic), strlen(magic)));
 
-  auto buffer = stream.GetBuffer();
+  std::shared_ptr<Buffer> result;
+  ASSERT_OK(stream->Finish(&result));
   ASSERT_NO_FATAL_FAILURE(AssertInvalidFileThrows(buffer));
 }
 

cpp/src/parquet/file_reader.cc

@@ -79,7 +79,8 @@ const RowGroupMetaData* RowGroupReader::metadata() const { return contents_->met
 // RowGroupReader::Contents implementation for the Parquet file specification
 class SerializedRowGroup : public RowGroupReader::Contents {
  public:
-  SerializedRowGroup(RandomAccessSource* source, FileMetaData* file_metadata,
+  SerializedRowGroup(const std::shared_ptr<ArrowInputFile>& source,
+                     FileMetaData* file_metadata,
                      int row_group_number, const ReaderProperties& props)
       : source_(source), file_metadata_(file_metadata), properties_(props) {
     row_group_metadata_ = file_metadata->RowGroup(row_group_number);
@@ -100,7 +101,6 @@ class SerializedRowGroup : public RowGroupReader::Contents {
     }
 
     int64_t col_length = col->total_compressed_size();
-    std::unique_ptr<InputStream> stream;
 
     // PARQUET-816 workaround for old files created by older parquet-mr
     const ApplicationVersion& version = file_metadata_->writer_version();
@@ -113,14 +113,14 @@ class SerializedRowGroup : public RowGroupReader::Contents {
       col_length += padding;
     }
 
-    stream = properties_.GetStream(source_, col_start, col_length);
-
-    return PageReader::Open(std::move(stream), col->num_values(), col->compression(),
+    std::shared_ptr<ArrowInputStream> stream = properties_.GetStream(
+        source_, col_start, col_length);
+    return PageReader::Open(stream, col->num_values(), col->compression(),
                             properties_.memory_pool());
   }
 
  private:
-  RandomAccessSource* source_;
+  std::shared_ptr<ArrowInputFile> source_;
   FileMetaData* file_metadata_;
   std::unique_ptr<RowGroupMetaData> row_group_metadata_;
   ReaderProperties properties_;
@@ -134,9 +134,9 @@ class SerializedRowGroup : public RowGroupReader::Contents {
 // This class takes ownership of the provided data source
 class SerializedFile : public ParquetFileReader::Contents {
  public:
-  SerializedFile(std::unique_ptr<RandomAccessSource> source,
+  SerializedFile(const std::shared_ptr<ArrowInputFile>& source,
                  const ReaderProperties& props = default_reader_properties())
-      : source_(std::move(source)), properties_(props) {}
+      : source_(source), properties_(props) {}
 
   ~SerializedFile() override {
     try {
@@ -145,11 +145,12 @@ class SerializedFile : public ParquetFileReader::Contents {
     }
   }
 
-  void Close() override { source_->Close(); }
+  // TODO(wesm): Do files need to be closed here?
+  void Close() override {}
 
   std::shared_ptr<RowGroupReader> GetRowGroup(int i) override {
     std::unique_ptr<SerializedRowGroup> contents(
-        new SerializedRowGroup(source_.get(), file_metadata_.get(), i, properties_));
+        new SerializedRowGroup(source_, file_metadata_.get(), i, properties_));
     return std::make_shared<RowGroupReader>(std::move(contents));
   }
 
@@ -160,7 +161,8 @@ class SerializedFile : public ParquetFileReader::Contents {
   }
 
   void ParseMetaData() {
-    int64_t file_size = source_->Size();
+    int64_t file_size = -1;
+    PARQUET_THROW_NOT_OK(source_->GetSize(&file_size));
 
     if (file_size == 0) {
       throw ParquetException("Invalid Parquet file size is 0 bytes");
@@ -171,19 +173,19 @@ class SerializedFile : public ParquetFileReader::Contents {
       throw ParquetException(ss.str());
     }
 
-    uint8_t footer_buffer[kDefaultFooterReadSize];
+    std::shared_ptr<Buffer> footer_buffer;
     int64_t footer_read_size = std::min(file_size, kDefaultFooterReadSize);
-    int64_t bytes_read =
-        source_->ReadAt(file_size - footer_read_size, footer_read_size, footer_buffer);
+    PARQUET_THROW_NOT_OK(source_->ReadAt(file_size - footer_read_size, footer_read_size,
+                                         &footer_buffer));
 
     // Check if all bytes are read. Check if last 4 bytes read have the magic bits
-    if (bytes_read != footer_read_size ||
-        memcmp(footer_buffer + footer_read_size - 4, kParquetMagic, 4) != 0) {
+    if (footer_buffer->size() != footer_read_size ||
+        memcmp(footer_buffer->data() + footer_read_size - 4, kParquetMagic, 4) != 0) {
       throw ParquetException("Invalid parquet file. Corrupt footer.");
     }
 
     uint32_t metadata_len =
-        *reinterpret_cast<uint32_t*>(footer_buffer + footer_read_size - kFooterSize);
+      *reinterpret_cast<uint32_t*>(footer_buffer->data() + footer_read_size - kFooterSize);
     int64_t metadata_start = file_size - kFooterSize - metadata_len;
     if (kFooterSize + metadata_len > file_size) {
       throw ParquetException(
@@ -191,27 +193,24 @@ class SerializedFile : public ParquetFileReader::Contents {
           "file metadata size.");
     }
 
-    std::shared_ptr<ResizableBuffer> metadata_buffer =
-        AllocateBuffer(properties_.memory_pool(), metadata_len);
-
+    std::shared_ptr<Buffer> metadata_buffer;
     // Check if the footer_buffer contains the entire metadata
     if (footer_read_size >= (metadata_len + kFooterSize)) {
-      memcpy(metadata_buffer->mutable_data(),
-             footer_buffer + (footer_read_size - metadata_len - kFooterSize),
-             metadata_len);
+      metadata_buffer = SliceBuffer(footer_buffer,
+                                    footer_read_size - metadata_len - kFooterSize,
+                                    metadata_len);
     } else {
-      bytes_read =
-          source_->ReadAt(metadata_start, metadata_len, metadata_buffer->mutable_data());
-      if (bytes_read != metadata_len) {
+      PARQUET_THROW_NOT_OK(source_->ReadAt(metadata_start, metadata_len,
+                                           &metadata_buffer));
+      if (metadata_buffer->size() != metadata_len) {
         throw ParquetException("Invalid parquet file. Could not read metadata bytes.");
       }
     }
-
     file_metadata_ = FileMetaData::Make(metadata_buffer->data(), &metadata_len);
   }
 
  private:
-  std::unique_ptr<RandomAccessSource> source_;
+  std::unique_ptr<ArrowInputFile> source_;
   std::shared_ptr<FileMetaData> file_metadata_;
   ReaderProperties properties_;
 };
@@ -231,10 +230,10 @@ ParquetFileReader::~ParquetFileReader() {
 // Open the file. If no metadata is passed, it is parsed from the footer of
 // the file
 std::unique_ptr<ParquetFileReader::Contents> ParquetFileReader::Contents::Open(
-    std::unique_ptr<RandomAccessSource> source, const ReaderProperties& props,
+    const std::shared_ptr<RandomAccessSource>& source, const ReaderProperties& props,
     const std::shared_ptr<FileMetaData>& metadata) {
   std::unique_ptr<ParquetFileReader::Contents> result(
-      new SerializedFile(std::move(source), props));
+      new SerializedFile(source, props));
 
   // Access private methods here, but otherwise unavailable
   SerializedFile* file = static_cast<SerializedFile*>(result.get());
@@ -252,17 +251,17 @@ std::unique_ptr<ParquetFileReader::Contents> ParquetFileReader::Contents::Open(
 std::unique_ptr<ParquetFileReader> ParquetFileReader::Open(
     const std::shared_ptr<::arrow::io::RandomAccessFile>& source,
     const ReaderProperties& props, const std::shared_ptr<FileMetaData>& metadata) {
-  std::unique_ptr<RandomAccessSource> io_wrapper(new ArrowInputFile(source));
-  return Open(std::move(io_wrapper), props, metadata);
+  auto contents = SerializedFile::Open(source, props, metadata);
+  std::unique_ptr<ParquetFileReader> result(new ParquetFileReader());
+  result->Open(std::move(contents));
+  return result;
 }
 
 std::unique_ptr<ParquetFileReader> ParquetFileReader::Open(
     std::unique_ptr<RandomAccessSource> source, const ReaderProperties& props,
     const std::shared_ptr<FileMetaData>& metadata) {
-  auto contents = SerializedFile::Open(std::move(source), props, metadata);
-  std::unique_ptr<ParquetFileReader> result(new ParquetFileReader());
-  result->Open(std::move(contents));
-  return result;
+  auto wrapper = std::shared_ptr<ParquetInputWrapper>(std::move(source));
+  return Open(wrapper, props, metadata);
 }
 
 std::unique_ptr<ParquetFileReader> ParquetFileReader::OpenFile(

cpp/src/parquet/file_reader.h

@@ -73,7 +73,7 @@ class PARQUET_EXPORT ParquetFileReader {
   // An implementation of the Contents class is defined in the .cc file
   struct PARQUET_EXPORT Contents {
     static std::unique_ptr<Contents> Open(
-        std::unique_ptr<RandomAccessSource> source,
+        const std::shared_ptr<::arrow::io::RandomAccessFile>& source,
         const ReaderProperties& props = default_reader_properties(),
         const std::shared_ptr<FileMetaData>& metadata = NULLPTR);
 
@@ -92,6 +92,7 @@ class PARQUET_EXPORT ParquetFileReader {
   //
   // If you cannot provide exclusive access to your file resource, create a
   // subclass of RandomAccessSource that wraps the shared resource
+  ARROW_DEPRECATED("Use arrow::io::RandomAccessFile version");
   static std::unique_ptr<ParquetFileReader> Open(
       std::unique_ptr<RandomAccessSource> source,
       const ReaderProperties& props = default_reader_properties(),

cpp/src/parquet/file_writer.cc

@@ -75,7 +75,8 @@ inline void ThrowRowsMisMatchError(int col, int64_t prev, int64_t curr) {
 // RowGroupWriter::Contents implementation for the Parquet file specification
 class RowGroupSerializer : public RowGroupWriter::Contents {
  public:
-  RowGroupSerializer(OutputStream* sink, RowGroupMetaDataBuilder* metadata,
+  RowGroupSerializer(const std::shared_ptr<ArrowOutputStream>& sink,
+                     RowGroupMetaDataBuilder* metadata,
                      const WriterProperties* properties, bool buffered_row_group = false)
       : sink_(sink),
         metadata_(metadata),
@@ -182,7 +183,7 @@ class RowGroupSerializer : public RowGroupWriter::Contents {
   }
 
  private:
-  OutputStream* sink_;
+  std::shared_ptr<ArrowOutputStream> sink_;
   mutable RowGroupMetaDataBuilder* metadata_;
   const WriterProperties* properties_;
   int64_t total_bytes_written_;
@@ -237,7 +238,8 @@ class RowGroupSerializer : public RowGroupWriter::Contents {
 class FileSerializer : public ParquetFileWriter::Contents {
  public:
   static std::unique_ptr<ParquetFileWriter::Contents> Open(
-      const std::shared_ptr<OutputStream>& sink, const std::shared_ptr<GroupNode>& schema,
+      const std::shared_ptr<ArrowOutputStream>& sink,
+      const std::shared_ptr<GroupNode>& schema,
       const std::shared_ptr<WriterProperties>& properties,
       const std::shared_ptr<const KeyValueMetadata>& key_value_metadata) {
     std::unique_ptr<ParquetFileWriter::Contents> result(
@@ -260,8 +262,6 @@ class FileSerializer : public ParquetFileWriter::Contents {
       // Write magic bytes and metadata
       file_metadata_ = metadata_->Finish();
       WriteFileMetaData(*file_metadata_, sink_.get());
-
-      sink_->Close();
     }
   }
 
@@ -282,7 +282,7 @@ class FileSerializer : public ParquetFileWriter::Contents {
     num_row_groups_++;
     auto rg_metadata = metadata_->AppendRowGroup();
     std::unique_ptr<RowGroupWriter::Contents> contents(new RowGroupSerializer(
-        sink_.get(), rg_metadata, properties_.get(), buffered_row_group));
+        sink_, rg_metadata, properties_.get(), buffered_row_group));
     row_group_writer_.reset(new RowGroupWriter(std::move(contents)));
     return row_group_writer_.get();
   }
@@ -299,7 +299,7 @@ class FileSerializer : public ParquetFileWriter::Contents {
   }
 
  private:
-  FileSerializer(const std::shared_ptr<OutputStream>& sink,
+  FileSerializer(const std::shared_ptr<ArrowOutputStream>& sink,
                  const std::shared_ptr<GroupNode>& schema,
                  const std::shared_ptr<WriterProperties>& properties,
                  const std::shared_ptr<const KeyValueMetadata>& key_value_metadata)
@@ -313,7 +313,7 @@ class FileSerializer : public ParquetFileWriter::Contents {
     StartFile();
   }
 
-  std::shared_ptr<OutputStream> sink_;
+  std::shared_ptr<ArrowOutputStream> sink_;
   bool is_open_;
   const std::shared_ptr<WriterProperties> properties_;
   int num_row_groups_;
@@ -324,7 +324,7 @@ class FileSerializer : public ParquetFileWriter::Contents {
 
   void StartFile() {
     // Parquet files always start with PAR1
-    sink_->Write(PARQUET_MAGIC, 4);
+    PARQUET_THROW_NOT_OK(sink_->Write(PARQUET_MAGIC, 4));
   }
 };
 
@@ -345,8 +345,10 @@ std::unique_ptr<ParquetFileWriter> ParquetFileWriter::Open(
     const std::shared_ptr<GroupNode>& schema,
     const std::shared_ptr<WriterProperties>& properties,
     const std::shared_ptr<const KeyValueMetadata>& key_value_metadata) {
-  return Open(std::make_shared<ArrowOutputStream>(sink), schema, properties,
-              key_value_metadata);
+  auto contents = FileSerializer::Open(sink, schema, properties, key_value_metadata);
+  std::unique_ptr<ParquetFileWriter> result(new ParquetFileWriter());
+  result->Open(std::move(contents));
+  return result;
 }
 
 std::unique_ptr<ParquetFileWriter> ParquetFileWriter::Open(
@@ -354,13 +356,11 @@ std::unique_ptr<ParquetFileWriter> ParquetFileWriter::Open(
     const std::shared_ptr<schema::GroupNode>& schema,
     const std::shared_ptr<WriterProperties>& properties,
     const std::shared_ptr<const KeyValueMetadata>& key_value_metadata) {
-  auto contents = FileSerializer::Open(sink, schema, properties, key_value_metadata);
-  std::unique_ptr<ParquetFileWriter> result(new ParquetFileWriter());
-  result->Open(std::move(contents));
-  return result;
+  return Open(std::make_shared<ParquetOutputWrapper>(sink), schema, properties,
+              key_value_metadata);
 }
 
-void WriteFileMetaData(const FileMetaData& file_metadata, OutputStream* sink) {
+void WriteFileMetaData(const FileMetaData& file_metadata, ArrowOutputStream* sink) {
   // Write MetaData
   uint32_t metadata_len = static_cast<uint32_t>(sink->Tell());
 
@@ -368,8 +368,13 @@ void WriteFileMetaData(const FileMetaData& file_metadata, OutputStream* sink) {
   metadata_len = static_cast<uint32_t>(sink->Tell()) - metadata_len;
 
   // Write Footer
-  sink->Write(reinterpret_cast<uint8_t*>(&metadata_len), 4);
-  sink->Write(PARQUET_MAGIC, 4);
+  PARQUET_THROW_NOT_OK(sink->Write(reinterpret_cast<uint8_t*>(&metadata_len), 4));
+  PARQUET_THROW_NOT_OK(sink->Write(PARQUET_MAGIC, 4));
+}
+
+void WriteFileMetaData(const FileMetaData& file_metadata, OutputStream* sink) {
+  ParquetOutputWrapper wrapper(sink);
+  return WriteFileMetaData(file_metadata, &wrapper);
 }
 
 const SchemaDescriptor* ParquetFileWriter::schema() const { return contents_->schema(); }

cpp/src/parquet/file_writer.h

@@ -110,6 +110,9 @@ class PARQUET_EXPORT RowGroupWriter {
 PARQUET_EXPORT
 void WriteFileMetaData(const FileMetaData& file_metadata, OutputStream* sink);
 
+PARQUET_EXPORT
+void WriteFileMetaData(const FileMetaData& file_metadata, ArrowOutputStream* sink);
+
 class PARQUET_EXPORT ParquetFileWriter {
  public:
   // Forward declare a virtual class 'Contents' to aid dependency injection and more

cpp/src/parquet/platform.cc

@@ -0,0 +1,45 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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.
+
+#include "parquet/platform.h"
+
+#include <cstdint>
+#include <memory>
+
+#include "arrow/io/memory.h"
+
+#include "parquet/exception.h"
+
+namespace parquet {
+
+static constexpr int64_t kDefaultOutputStreamSize = 1024;
+
+std::shared_ptr<ArrowOutputStream> CreateOutputStream(
+    ::arrow::MemoryPool* pool = ::arrow::default_memory_pool()) {
+  std::shared_ptr<::arrow::io::BufferOutputStream> stream;
+  PARQUET_THROW_NOT_OK(::arrow::io::BufferOutputStream::Create(
+      kDefaultOutputStreamSize, pool, &stream));
+  return stream;
+}
+
+std::shared_ptr<ResizableBuffer> AllocateBuffer(MemoryPool* pool, int64_t size) {
+  std::shared_ptr<ResizableBuffer> result;
+  PARQUET_THROW_NOT_OK(arrow::AllocateResizableBuffer(pool, size, &result));
+  return result;
+}
+
+}  // namespace parquet

cpp/src/parquet/platform.h

@@ -84,5 +84,13 @@ using ResizableBuffer = ::arrow::ResizableBuffer;
 using ResizableBuffer = ::arrow::ResizableBuffer;
 using ArrowInputFile = ::arrow::io::RandomAccessFile;
 using ArrowInputStream = ::arrow::io::InputStream;
+using ArrowOutputStream = ::arrow::io::OutputStream;
+
+std::shared_ptr<ArrowOutputStream> CreateOutputStream(
+    ::arrow::MemoryPool* pool = ::arrow::default_memory_pool());
+
+PARQUET_EXPORT
+std::shared_ptr<ResizableBuffer> AllocateBuffer(
+    ::arrow::MemoryPool* pool = ::arrow::default_memory_pool(), int64_t size = 0);
 
 }  // namespace parquet

cpp/src/parquet/properties.h

@@ -49,17 +49,18 @@ class PARQUET_EXPORT ReaderProperties {
 
   std::unique_ptr<ArrowInputStream> GetStream(std::shared_ptr<ArrowInputFile> source,
                                              int64_t start, int64_t num_bytes) {
-    std::unique_ptr<ArrowInputStream> stream;
     if (buffered_stream_enabled_) {
       std::shared_ptr<::arrow::io::BufferedInputStream> stream;
-      PARQUET_THROW_NOT_OK(::arrow::io::BufferedInputStream::Create(buffer_size_, pool_,
-                                                                    source
-      stream.reset(
-          new BufferedInputStream(pool_, buffer_size_, source, start, num_bytes));
+      PARQUET_THROW_NOT_OK(source->Seek(start));
+      // TODO(wesm): ARROW-5428 set read extent
+      PARQUET_THROW_NOT_OK(::arrow::io::BufferedInputStream::Create(
+          buffer_size_, pool_, source, &stream));
+      return stream;
     } else {
-      stream.reset(new InMemoryInputStream(source, start, num_bytes));
+      std::shared_ptr<Buffer> data;
+      PARQUET_THROW_NOT_OK(source->ReadAt(start, num_bytes, &data));
+      return std::make_shared<::arrow::io::BufferReader>(data);
     }
-    return stream;
   }
 
   bool is_buffered_stream_enabled() const { return buffered_stream_enabled_; }

cpp/src/parquet/reader-test.cc

@@ -199,7 +199,7 @@ TEST_F(TestLocalFile, FileClosedOnDestruction) {
   bool close_called = false;
   {
     auto contents = ParquetFileReader::Contents::Open(
-        std::unique_ptr<RandomAccessSource>(new HelperFileClosed(handle, &close_called)));
+        std::make_shared<HelperFileClosed>(handle, &close_called));
     std::unique_ptr<ParquetFileReader> result(new ParquetFileReader());
     result->Open(std::move(contents));
   }
@@ -277,25 +277,25 @@ Column 1
   Uncompressed Size: 52, Compressed Size: 56
 )###";
   std::string valuesOutput = R"###(--- Values ---
-element                       b                             
-a                             1                             
-b                             1                             
-c                             1                             
-NULL                          
-d                             
-a                             
-b                             
-c                             
-d                             
-NULL                          
-e                             
-a                             
-b                             
-c                             
-d                             
-e                             
-NULL                          
-f                             
+element                       b
+a                             1
+b                             1
+c                             1
+NULL
+d
+a
+b
+c
+d
+NULL
+e
+a
+b
+c
+d
+e
+NULL
+f
 
 )###";
   std::string dumpOutput = R"###(--- Values ---