More progress toward compilation, port over parquet::BufferedInputStream unit tests

cpp/src/arrow/buffer.h

@@ -86,6 +86,8 @@ class ARROW_EXPORT Buffer {
     parent_ = parent;
   }
 
+  constexpr uint8_t operator[](std::size_t i) const { return data_[i]; }
+
   bool is_mutable() const { return is_mutable_; }
 
   /// \brief Construct a new std::string with a hexadecimal representation of the buffer.

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

@@ -37,6 +37,7 @@
 #include "arrow/io/buffered.h"
 #include "arrow/io/file.h"
 #include "arrow/io/interfaces.h"
+#include "arrow/io/memory.h"
 #include "arrow/io/test-common.h"
 #include "arrow/status.h"
 #include "arrow/testing/gtest_util.h"
@@ -453,5 +454,173 @@ TEST_F(TestBufferedInputStream, SetBufferSize) {
   ASSERT_OK(buffered_->SetBufferSize(5));
 }
 
+class TestBufferedInputStreamBound : public ::testing::Test {
+ public:
+  void SetUp() {
+    // Create a buffer larger than source size, to check that the
+    // stream end is respected
+    std::shared_ptr<ResizableBuffer> buf;
+    ASSERT_OK(AllocateResizableBuffer(default_memory_pool(), source_size_ + 10, &buf));
+    ASSERT_LT(source_size_, buf->size());
+    for (int i = 0; i < source_size_; i++) {
+      buf->mutable_data()[i] = static_cast<uint8_t>(i);
+    }
+    source_ = std::make_shared<BufferReader>(buf);
+    ASSERT_OK(source_->Advance(stream_offset_));
+    ASSERT_OK(BufferedInputStream::Create(chunk_size_, default_memory_pool(), source_,
+                                          &stream_, stream_size_));
+  }
+
+ protected:
+  int64_t source_size_ = 256;
+  int64_t stream_offset_ = 10;
+  int64_t stream_size_ = source_size_ - stream_offset_;
+  int64_t chunk_size_ = 50;
+  std::shared_ptr<InputStream> source_;
+  std::shared_ptr<BufferedInputStream> stream_;
+};
+
+TEST_F(TestBufferedInputStreamBound, Basics) {
+  std::shared_ptr<Buffer> buffer;
+  util::string_view view;
+
+  // Trigger buffering
+  ASSERT_OK(stream_->Read(1, &buffer));
+
+  // source is at offset 10
+  view = stream_->Peek(9);
+  ASSERT_EQ(9, view.size());
+  for (int i = 0; i < 9; i++) {
+    ASSERT_EQ(11 + i, view[i]) << i;
+  }
+
+  ASSERT_OK(stream_->Read(9, &buffer));
+  ASSERT_EQ(9, buffer->size());
+  for (int i = 0; i < 9; i++) {
+    ASSERT_EQ(11 + i, (*buffer)[i]) << i;
+  }
+
+  ASSERT_OK(stream_->Read(10, &buffer));
+  ASSERT_EQ(10, buffer->size());
+  for (int i = 0; i < 10; i++) {
+    ASSERT_EQ(20 + i, (*buffer)[i]) << i;
+  }
+  ASSERT_OK(stream_->Advance(5));
+  ASSERT_OK(stream_->Advance(5));
+
+  // source is at offset 40
+  // read across buffer boundary. buffer size is 50
+  ASSERT_OK(stream_->Read(20, &buffer));
+  ASSERT_EQ(20, buffer->size());
+  for (int i = 0; i < 20; i++) {
+    ASSERT_EQ(40 + i, (*buffer)[i]) << i;
+  }
+
+  // read more than original chunk size
+  ASSERT_OK(stream_->Read(60, &buffer));
+  ASSERT_EQ(60, buffer->size());
+  for (int i = 0; i < 60; i++) {
+    ASSERT_EQ(60 + i, (*buffer)[i]) << i;
+  }
+
+  ASSERT_OK(stream_->Advance(120));
+
+  // source is at offset 240
+  // read outside of source boundary. source size is 256
+  ASSERT_OK(stream_->Read(30, &buffer));
+
+  ASSERT_EQ(16, buffer->size());
+  for (int i = 0; i < 16; i++) {
+    ASSERT_EQ(240 + i, (*buffer)[i]) << i;
+  }
+  // Stream exhausted
+  ASSERT_OK(stream_->Read(1, &buffer));
+  ASSERT_EQ(0, buffer->size());
+}
+
+TEST_F(TestBufferedInputStreamBound, LargeFirstPeek) {
+  // Test a first peek larger than chunk size
+  std::shared_ptr<Buffer> buffer;
+  util::string_view view;
+  int64_t n = 70;
+  ASSERT_GT(n, chunk_size_);
+
+  // Trigger buffering
+  ASSERT_OK(stream_->Read(1, &buffer));
+
+  // source is at offset 11
+  view = stream_->Peek(n - 1);
+  ASSERT_EQ(n, static_cast<int>(view.size()));
+  for (int i = 0; i < n - 1; i++) {
+    ASSERT_EQ(11 + i, view[i]) << i;
+  }
+
+  view = stream_->Peek(n);
+  ASSERT_EQ(n, static_cast<int>(view.size()));
+  for (int i = 0; i < n; i++) {
+    ASSERT_EQ(10 + i, view[i]) << i;
+  }
+
+  ASSERT_OK(stream_->Read(n, &buffer));
+  ASSERT_EQ(n, buffer->size());
+  for (int i = 0; i < n; i++) {
+    ASSERT_EQ(10 + i, (*buffer)[i]) << i;
+  }
+  // source is at offset 10 + n
+  ASSERT_OK(stream_->Read(20, &buffer));
+  ASSERT_EQ(20, buffer->size());
+  for (int i = 0; i < 20; i++) {
+    ASSERT_EQ(10 + n + i, (*buffer)[i]) << i;
+  }
+}
+
+TEST_F(TestBufferedInputStreamBound, OneByteReads) {
+  std::shared_ptr<Buffer> buffer;
+  for (int i = 0; i < stream_size_; ++i) {
+    ASSERT_OK(stream_->Read(1, &buffer));
+    ASSERT_EQ(1, buffer->size());
+    ASSERT_EQ(10 + i, (*buffer)[0]) << i;
+  }
+  // Stream exhausted
+  ASSERT_OK(stream_->Read(1, &buffer));
+  ASSERT_EQ(0, buffer->size());
+}
+
+TEST_F(TestBufferedInputStreamBound, BufferExactlyExhausted) {
+  // Test exhausting the buffer exactly then issuing further reads (PARQUET-1571).
+  std::shared_ptr<Buffer> buffer;
+
+  // source is at offset 10
+  int64_t n = 10;
+  ASSERT_OK(stream_->Read(n, &buffer));
+  ASSERT_EQ(n, buffer->size());
+  for (int i = 0; i < n; i++) {
+    ASSERT_EQ(10 + i, (*buffer)[i]) << i;
+  }
+  // source is at offset 20
+  // Exhaust buffer exactly
+  n = stream_->bytes_buffered();
+  ASSERT_OK(stream_->Read(n, &buffer));
+  ASSERT_EQ(n, buffer->size());
+  for (int i = 0; i < n; i++) {
+    ASSERT_EQ(20 + i, (*buffer)[i]) << i;
+  }
+
+  // source is at offset 20 + n
+  // Read new buffer
+  ASSERT_OK(stream_->Read(10, &buffer));
+  ASSERT_EQ(10, buffer->size());
+  for (int i = 0; i < 10; i++) {
+    ASSERT_EQ(20 + n + i, (*buffer)[i]) << i;
+  }
+
+  // source is at offset 30 + n
+  ASSERT_OK(stream_->Read(10, &buffer));
+  ASSERT_EQ(10, buffer->size());
+  for (int i = 0; i < 10; i++) {
+    ASSERT_EQ(30 + n + i, (*buffer)[i]) << i;
+  }
+}
+
 }  // namespace io
 }  // namespace arrow

cpp/src/arrow/io/buffered.cc

@@ -221,8 +221,12 @@ std::shared_ptr<OutputStream> BufferedOutputStream::raw() const { return impl_->
 
 class BufferedInputStream::Impl : public BufferedBase {
  public:
-  Impl(std::shared_ptr<InputStream> raw, MemoryPool* pool)
-      : BufferedBase(pool), raw_(std::move(raw)), bytes_buffered_(0) {}
+  Impl(std::shared_ptr<InputStream> raw, MemoryPool* pool, int64_t raw_total_bytes_bound)
+      : BufferedBase(pool),
+        raw_(std::move(raw)),
+        raw_read_total_(0),
+        raw_read_bound_(raw_total_bytes_bound),
+        bytes_buffered_(0) {}
 
   ~Impl() { DCHECK_OK(Close()); }
 
@@ -282,8 +286,14 @@ class BufferedInputStream::Impl : public BufferedBase {
       if (!buffer_) {
         RETURN_NOT_OK(ResetBuffer());
       }
-      RETURN_NOT_OK(raw_->Read(buffer_size_, &bytes_buffered_, buffer_data_));
+
+      int64_t bytes_to_buffer = buffer_size_;
+      if (raw_read_bound_ >= 0) {
+        bytes_to_buffer = std::min(buffer_size_, raw_read_bound_ - raw_read_total_);
+      }
+      RETURN_NOT_OK(raw_->Read(bytes_to_buffer, &bytes_buffered_, buffer_data_));
       buffer_pos_ = 0;
+      raw_read_total_ += bytes_buffered_;
 
       // Do not make assumptions about the raw stream position
       raw_pos_ = -1;
@@ -305,11 +315,20 @@ class BufferedInputStream::Impl : public BufferedBase {
       RETURN_NOT_OK(BufferIfNeeded());
     }
 
+    *bytes_read = 0;
+
     if (nbytes > bytes_buffered_) {
       // Copy buffered bytes into out, then read rest
       memcpy(out, buffer_data_ + buffer_pos_, bytes_buffered_);
-      RETURN_NOT_OK(raw_->Read(nbytes - bytes_buffered_, bytes_read,
+
+      int64_t bytes_to_read = nbytes - bytes_buffered_;
+      if (raw_read_bound_ >= 0) {
+        bytes_to_read = std::min(bytes_to_read, raw_read_bound_ - raw_read_total_);
+      }
+      RETURN_NOT_OK(raw_->Read(bytes_to_read, bytes_read,
                                reinterpret_cast<uint8_t*>(out) + bytes_buffered_));
+      raw_read_total_ += *bytes_read;
+
       // Do not make assumptions about the raw stream position
       raw_pos_ = -1;
       *bytes_read += bytes_buffered_;
@@ -344,6 +363,8 @@ class BufferedInputStream::Impl : public BufferedBase {
 
  private:
   std::shared_ptr<InputStream> raw_;
+  int64_t raw_read_total_;
+  int64_t raw_read_bound_;
 
   // Number of remaining bytes in the buffer, to be reduced on each read from
   // the buffer
@@ -351,17 +372,19 @@ class BufferedInputStream::Impl : public BufferedBase {
 };
 
 BufferedInputStream::BufferedInputStream(std::shared_ptr<InputStream> raw,
-                                         MemoryPool* pool) {
-  impl_.reset(new Impl(std::move(raw), pool));
+                                         MemoryPool* pool,
+                                         int64_t raw_total_bytes_bound) {
+  impl_.reset(new Impl(std::move(raw), pool, raw_total_bytes_bound));
 }
 
 BufferedInputStream::~BufferedInputStream() { DCHECK_OK(impl_->Close()); }
 
 Status BufferedInputStream::Create(int64_t buffer_size, MemoryPool* pool,
                                    std::shared_ptr<InputStream> raw,
-                                   std::shared_ptr<BufferedInputStream>* out) {
-  auto result =
-      std::shared_ptr<BufferedInputStream>(new BufferedInputStream(std::move(raw), pool));
+                                   std::shared_ptr<BufferedInputStream>* out,
+                                   int64_t raw_total_bytes_bound) {
+  auto result = std::shared_ptr<BufferedInputStream>(
+      new BufferedInputStream(std::move(raw), pool, raw_total_bytes_bound));
   RETURN_NOT_OK(result->SetBufferSize(buffer_size));
   *out = std::move(result);
   return Status::OK();

cpp/src/arrow/io/buffered.h

@@ -99,9 +99,13 @@ class ARROW_EXPORT BufferedInputStream : public InputStream {
   /// \param[in] pool a MemoryPool to use for allocations
   /// \param[in] raw a raw InputStream
   /// \param[out] out the created BufferedInputStream
+  /// \param[in] raw_read_bound a bound on the maximum number of bytes
+  /// to read from the raw input stream. The default -1 indicates that
+  /// it is unbounded
   static Status Create(int64_t buffer_size, MemoryPool* pool,
                        std::shared_ptr<InputStream> raw,
-                       std::shared_ptr<BufferedInputStream>* out);
+                       std::shared_ptr<BufferedInputStream>* out,
+                       int64_t raw_read_bound = -1);
 
   /// \brief Resize internal read buffer; calls to Read(...) will read at least
   /// \param[in] new_buffer_size the new read buffer size
@@ -138,7 +142,8 @@ class ARROW_EXPORT BufferedInputStream : public InputStream {
   Status Read(int64_t nbytes, std::shared_ptr<Buffer>* out) override;
 
  private:
-  explicit BufferedInputStream(std::shared_ptr<InputStream> raw, MemoryPool* pool);
+  explicit BufferedInputStream(std::shared_ptr<InputStream> raw, MemoryPool* pool,
+                               int64_t raw_total_bytes_bound);
 
   class ARROW_NO_EXPORT Impl;
   std::unique_ptr<Impl> impl_;

cpp/src/arrow/io/memory.cc

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

cpp/src/parquet/CMakeLists.txt

@@ -169,7 +169,9 @@ set(PARQUET_SRCS
     murmur3.cc
     parquet_constants.cpp
     parquet_types.cpp
+    platform.cc
     printer.cc
+    properties.cc
     schema.cc
     statistics.cc
     types.cc)

cpp/src/parquet/api/io.h

@@ -18,7 +18,7 @@
 #ifndef PARQUET_API_IO_H
 #define PARQUET_API_IO_H
 
-#include "parquet/exception.h"
 #include "parquet/deprecated_io.h"
+#include "parquet/exception.h"
 
 #endif  // PARQUET_API_IO_H

cpp/src/parquet/arrow/arrow-reader-writer-test.cc

@@ -331,7 +331,7 @@ void WriteTableToBuffer(const std::shared_ptr<Table>& table, int64_t row_group_s
   ASSERT_OK_NO_THROW(WriteTable(*table, ::arrow::default_memory_pool(), sink,
                                 row_group_size, default_writer_properties(),
                                 arrow_properties));
-  *out = sink->GetBuffer();
+  ASSERT_OK_NO_THROW(sink->Finish(out));
 }
 
 void AssertChunkedEqual(const ChunkedArray& expected, const ChunkedArray& actual) {
@@ -443,7 +443,8 @@ class TestParquetIO : public ::testing::Test {
   }
 
   void ReaderFromSink(std::unique_ptr<FileReader>* out) {
-    std::shared_ptr<Buffer> buffer = sink_->GetBuffer();
+    std::shared_ptr<Buffer> buffer;
+    ASSERT_OK_NO_THROW(sink_->Finish(&buffer));
     ASSERT_OK_NO_THROW(OpenFile(std::make_shared<BufferReader>(buffer),
                                 ::arrow::default_memory_pool(),
                                 ::parquet::default_reader_properties(), nullptr, out));
@@ -551,9 +552,7 @@ class TestParquetIO : public ::testing::Test {
     ASSERT_OK_NO_THROW(writer.Close());
   }
 
-  void ResetStream() {
-    sink_ = CreateOutputStream();
-  }
+  void ResetStream() { sink_ = CreateOutputStream(); }
 
   std::shared_ptr<ArrowOutputStream> sink_;
 };
@@ -1961,7 +1960,8 @@ class TestNestedSchemaRead : public ::testing::TestWithParam<Repetition::type> {
   std::shared_ptr<::arrow::Int32Array> values_array_ = nullptr;
 
   void InitReader() {
-    std::shared_ptr<Buffer> buffer = nested_parquet_->GetBuffer();
+    std::shared_ptr<Buffer> buffer;
+    ASSERT_OK_NO_THROW(nested_parquet_->Finish(&buffer));
     ASSERT_OK_NO_THROW(
         OpenFile(std::make_shared<BufferReader>(buffer), ::arrow::default_memory_pool(),
                  ::parquet::default_reader_properties(), nullptr, &reader_));

cpp/src/parquet/arrow/reader-writer-benchmark.cc

@@ -25,7 +25,7 @@
 #include "parquet/column_writer.h"
 #include "parquet/file_reader.h"
 #include "parquet/file_writer.h"
-#include "parquet/util/memory.h"
+#include "parquet/platform.h"
 
 #include "arrow/api.h"
 
@@ -174,7 +174,9 @@ static void BM_ReadColumn(::benchmark::State& state) {
   std::shared_ptr<::arrow::Table> table = TableFromVector<ParquetType>(values, nullable);
   auto output = CreateOutputStream();
   EXIT_NOT_OK(WriteTable(*table, ::arrow::default_memory_pool(), output, BENCHMARK_SIZE));
-  std::shared_ptr<Buffer> buffer = output->GetBuffer();
+
+  std::shared_ptr<Buffer> buffer;
+  PARQUET_THROW_NOT_OK(output->Finish(&buffer));
 
   while (state.KeepRunning()) {
     auto reader =
@@ -205,7 +207,9 @@ static void BM_ReadIndividualRowGroups(::benchmark::State& state) {
   // This writes 10 RowGroups
   EXIT_NOT_OK(
       WriteTable(*table, ::arrow::default_memory_pool(), output, BENCHMARK_SIZE / 10));
-  std::shared_ptr<Buffer> buffer = output->GetBuffer();
+
+  std::shared_ptr<Buffer> buffer;
+  PARQUET_THROW_NOT_OK(output->Finish(&buffer));
 
   while (state.KeepRunning()) {
     auto reader =
@@ -237,7 +241,8 @@ static void BM_ReadMultipleRowGroups(::benchmark::State& state) {
   // This writes 10 RowGroups
   EXIT_NOT_OK(
       WriteTable(*table, ::arrow::default_memory_pool(), output, BENCHMARK_SIZE / 10));
-  std::shared_ptr<Buffer> buffer = output->GetBuffer();
+  std::shared_ptr<Buffer> buffer;
+  PARQUET_THROW_NOT_OK(output->Finish(&buffer));
 
   while (state.KeepRunning()) {
     auto reader =

cpp/src/parquet/arrow/reader.cc

@@ -747,9 +747,8 @@ Status OpenFile(const std::shared_ptr<::arrow::io::RandomAccessFile>& file,
                 ::arrow::MemoryPool* allocator, const ArrowReaderProperties& properties,
                 std::unique_ptr<FileReader>* reader) {
   std::unique_ptr<ParquetReader> pq_reader;
-  PARQUET_CATCH_NOT_OK(
-      pq_reader = ParquetReader::Open(file,
-                                      ::parquet::default_reader_properties(), nullptr));
+  PARQUET_CATCH_NOT_OK(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/arrow/record_reader.h

@@ -22,9 +22,7 @@
 #include <memory>
 #include <vector>
 
-#include "arrow/memory_pool.h"
-
-#include "parquet/util/memory.h"
+#include "parquet/platform.h"
 
 namespace arrow {
 

cpp/src/parquet/arrow/schema.h

@@ -23,8 +23,8 @@
 #include <vector>
 
 #include "parquet/metadata.h"
-#include "parquet/schema.h"
 #include "parquet/platform.h"
+#include "parquet/schema.h"
 
 namespace arrow {
 

cpp/src/parquet/arrow/writer.cc

@@ -37,10 +37,11 @@
 
 #include "parquet/arrow/schema.h"
 #include "parquet/column_writer.h"
+#include "parquet/deprecated_io.h"
 #include "parquet/exception.h"
 #include "parquet/file_writer.h"
+#include "parquet/platform.h"
 #include "parquet/schema.h"
-#include "parquet/util/memory.h"
 
 using arrow::Array;
 using arrow::BinaryArray;
@@ -1105,14 +1106,14 @@ FileWriter::FileWriter(MemoryPool* pool, std::unique_ptr<ParquetFileWriter> writ
       schema_(schema) {}
 
 Status FileWriter::Open(const ::arrow::Schema& schema, ::arrow::MemoryPool* pool,
-                        const std::shared_ptr<OutputStream>& sink,
+                        const std::shared_ptr<::arrow::io::OutputStream>& sink,
                         const std::shared_ptr<WriterProperties>& properties,
                         std::unique_ptr<FileWriter>* writer) {
   return Open(schema, pool, sink, properties, default_arrow_writer_properties(), writer);
 }
 
 Status FileWriter::Open(const ::arrow::Schema& schema, ::arrow::MemoryPool* pool,
-                        const std::shared_ptr<OutputStream>& sink,
+                        const std::shared_ptr<::arrow::io::OutputStream>& sink,
                         const std::shared_ptr<WriterProperties>& properties,
                         const std::shared_ptr<ArrowWriterProperties>& arrow_properties,
                         std::unique_ptr<FileWriter>* writer) {
@@ -1131,32 +1132,10 @@ Status FileWriter::Open(const ::arrow::Schema& schema, ::arrow::MemoryPool* pool
   return Status::OK();
 }
 
-Status FileWriter::Open(const ::arrow::Schema& schema, ::arrow::MemoryPool* pool,
-                        const std::shared_ptr<::arrow::io::OutputStream>& sink,
-                        const std::shared_ptr<WriterProperties>& properties,
-                        std::unique_ptr<FileWriter>* writer) {
-  auto wrapper = std::make_shared<ArrowOutputStream>(sink);
-  return Open(schema, pool, wrapper, properties, writer);
-}
-
-Status FileWriter::Open(const ::arrow::Schema& schema, ::arrow::MemoryPool* pool,
-                        const std::shared_ptr<::arrow::io::OutputStream>& sink,
-                        const std::shared_ptr<WriterProperties>& properties,
-                        const std::shared_ptr<ArrowWriterProperties>& arrow_properties,
-                        std::unique_ptr<FileWriter>* writer) {
-  auto wrapper = std::make_shared<ArrowOutputStream>(sink);
-  return Open(schema, pool, wrapper, properties, arrow_properties, writer);
-}
-
-Status WriteFileMetaData(const FileMetaData& file_metadata, OutputStream* sink) {
-  PARQUET_CATCH_NOT_OK(::parquet::WriteFileMetaData(file_metadata, sink));
-  return Status::OK();
-}
-
 Status WriteFileMetaData(const FileMetaData& file_metadata,
                          const std::shared_ptr<::arrow::io::OutputStream>& sink) {
-  ArrowOutputStream wrapper(sink);
-  return ::parquet::arrow::WriteFileMetaData(file_metadata, &wrapper);
+  PARQUET_CATCH_NOT_OK(::parquet::WriteFileMetaData(file_metadata, sink.get()));
+  return Status::OK();
 }
 
 Status FileWriter::WriteTable(const Table& table, int64_t chunk_size) {
@@ -1197,8 +1176,8 @@ Status FileWriter::WriteTable(const Table& table, int64_t chunk_size) {
 }
 
 Status WriteTable(const ::arrow::Table& table, ::arrow::MemoryPool* pool,
-                  const std::shared_ptr<OutputStream>& sink, int64_t chunk_size,
-                  const std::shared_ptr<WriterProperties>& properties,
+                  const std::shared_ptr<::arrow::io::OutputStream>& sink,
+                  int64_t chunk_size, const std::shared_ptr<WriterProperties>& properties,
                   const std::shared_ptr<ArrowWriterProperties>& arrow_properties) {
   std::unique_ptr<FileWriter> writer;
   RETURN_NOT_OK(FileWriter::Open(*table.schema(), pool, sink, properties,
@@ -1207,13 +1186,5 @@ Status WriteTable(const ::arrow::Table& table, ::arrow::MemoryPool* pool,
   return writer->Close();
 }
 
-Status WriteTable(const ::arrow::Table& table, ::arrow::MemoryPool* pool,
-                  const std::shared_ptr<::arrow::io::OutputStream>& sink,
-                  int64_t chunk_size, const std::shared_ptr<WriterProperties>& properties,
-                  const std::shared_ptr<ArrowWriterProperties>& arrow_properties) {
-  auto wrapper = std::make_shared<ArrowOutputStream>(sink);
-  return WriteTable(table, pool, wrapper, chunk_size, properties, arrow_properties);
-}
-
 }  // namespace arrow
 }  // namespace parquet

cpp/src/parquet/arrow/writer.h

@@ -21,9 +21,9 @@
 #include <cstdint>
 #include <memory>
 
+#include "parquet/platform.h"
 #include "parquet/properties.h"
 #include "parquet/types.h"
-#include "parquet/platform.h"
 
 #include "arrow/type.h"
 
@@ -141,18 +141,6 @@ class PARQUET_EXPORT FileWriter {
              const std::shared_ptr<ArrowWriterProperties>& arrow_properties =
                  default_arrow_writer_properties());
 
-  static ::arrow::Status Open(const ::arrow::Schema& schema, ::arrow::MemoryPool* pool,
-                              const std::shared_ptr<OutputStream>& sink,
-                              const std::shared_ptr<WriterProperties>& properties,
-                              std::unique_ptr<FileWriter>* writer);
-
-  static ::arrow::Status Open(
-      const ::arrow::Schema& schema, ::arrow::MemoryPool* pool,
-      const std::shared_ptr<OutputStream>& sink,
-      const std::shared_ptr<WriterProperties>& properties,
-      const std::shared_ptr<ArrowWriterProperties>& arrow_properties,
-      std::unique_ptr<FileWriter>* writer);
-
   static ::arrow::Status Open(const ::arrow::Schema& schema, ::arrow::MemoryPool* pool,
                               const std::shared_ptr<::arrow::io::OutputStream>& sink,
                               const std::shared_ptr<WriterProperties>& properties,
@@ -189,27 +177,16 @@ class PARQUET_EXPORT FileWriter {
   std::shared_ptr<::arrow::Schema> schema_;
 };
 
-/// \brief Write Parquet file metadata only to indicated OutputStream
-PARQUET_EXPORT
-::arrow::Status WriteFileMetaData(const FileMetaData& file_metadata, OutputStream* sink);
-
 /// \brief Write Parquet file metadata only to indicated Arrow OutputStream
 PARQUET_EXPORT
 ::arrow::Status WriteFileMetaData(const FileMetaData& file_metadata,
-                                  const std::shared_ptr<::arrow::io::OutputStream>& sink);
+                                  ::arrow::io::OutputStream* sink);
 
 /**
  * Write a Table to Parquet.
  *
  * The table shall only consist of columns of primitive type or of primitive lists.
  */
-::arrow::Status PARQUET_EXPORT WriteTable(
-    const ::arrow::Table& table, ::arrow::MemoryPool* pool,
-    const std::shared_ptr<OutputStream>& sink, int64_t chunk_size,
-    const std::shared_ptr<WriterProperties>& properties = default_writer_properties(),
-    const std::shared_ptr<ArrowWriterProperties>& arrow_properties =
-        default_arrow_writer_properties());
-
 ::arrow::Status PARQUET_EXPORT WriteTable(
     const ::arrow::Table& table, ::arrow::MemoryPool* pool,
     const std::shared_ptr<::arrow::io::OutputStream>& sink, int64_t chunk_size,

cpp/src/parquet/bloom_filter-test.cc

@@ -31,9 +31,9 @@
 #include "parquet/bloom_filter.h"
 #include "parquet/exception.h"
 #include "parquet/murmur3.h"
+#include "parquet/platform.h"
 #include "parquet/test-util.h"
 #include "parquet/types.h"
-#include "parquet/util/memory.h"
 
 namespace parquet {
 namespace test {
@@ -197,7 +197,8 @@ TEST(CompatibilityTest, TestBloomFilter) {
   // Serialize Bloom filter to memory output stream
   auto sink = CreateOutputStream();
   bloom_filter2.WriteTo(&sink.get());
-  std::shared_ptr<Buffer> buffer1 = sink.GetBuffer();
+  std::shared_ptr<Buffer> buffer1;
+  PARQUET_THROW_NOT_OK(sink->Finish(&buffer1));
 
   PARQUET_THROW_NOT_OK(handle->Seek(0));
   PARQUET_THROW_NOT_OK(handle->GetSize(&size));

cpp/src/parquet/bloom_filter.cc

@@ -71,32 +71,26 @@ void BlockSplitBloomFilter::Init(const uint8_t* bitset, uint32_t num_bytes) {
 }
 
 BlockSplitBloomFilter BlockSplitBloomFilter::Deserialize(ArrowInputStream* input) {
-  int64_t bytes_available;
+  uint32_t len, hash, algorithm;
+  int64_t bytes_available = -1;
 
-  const uint8_t* read_buffer = NULL;
-  PARQUET_THROW_NOT_OK(input->Read(sizeof(uint32_t), &bytes_available, read_buffer));
-  if (static_cast<uint32_t>(bytes_available) != sizeof(uint32_t) || !read_buffer) {
+  PARQUET_THROW_NOT_OK(input->Read(sizeof(uint32_t), &bytes_available, &len));
+  if (static_cast<uint32_t>(bytes_available) != sizeof(uint32_t)) {
     throw ParquetException("Failed to deserialize from input stream");
   }
-  uint32_t len;
-  memcpy(&len, read_buffer, sizeof(uint32_t));
 
-  PARQUET_THROW_NOT_OK(input->Read(sizeof(uint32_t), &bytes_available, read_buffer));
-  if (static_cast<uint32_t>(bytes_available) != sizeof(uint32_t) || !read_buffer) {
+  PARQUET_THROW_NOT_OK(input->Read(sizeof(uint32_t), &bytes_available, &hash));
+  if (static_cast<uint32_t>(bytes_available) != sizeof(uint32_t)) {
     throw ParquetException("Failed to deserialize from input stream");
   }
-  uint32_t hash;
-  memcpy(&hash, read_buffer, sizeof(uint32_t));
   if (static_cast<HashStrategy>(hash) != HashStrategy::MURMUR3_X64_128) {
     throw ParquetException("Unsupported hash strategy");
   }
 
-  PARQUET_THROW_NOT_OK(input->Read(sizeof(uint32_t), &bytes_available, read_buffer));
-  if (static_cast<uint32_t>(bytes_available) != sizeof(uint32_t) || !read_buffer) {
+  PARQUET_THROW_NOT_OK(input->Read(sizeof(uint32_t), &bytes_available, &algorithm));
+  if (static_cast<uint32_t>(bytes_available) != sizeof(uint32_t)) {
     throw ParquetException("Failed to deserialize from input stream");
   }
-  uint32_t algorithm;
-  memcpy(&algorithm, read_buffer, sizeof(uint32_t));
   if (static_cast<Algorithm>(algorithm) != BloomFilter::Algorithm::BLOCK) {
     throw ParquetException("Unsupported Bloom filter algorithm");
   }
@@ -112,12 +106,12 @@ BlockSplitBloomFilter BlockSplitBloomFilter::Deserialize(ArrowInputStream* input
 void BlockSplitBloomFilter::WriteTo(ArrowOutputStream* sink) const {
   DCHECK(sink != nullptr);
 
-  PARQUET_THROW_NOT_OK(sink->Write(reinterpret_cast<const uint8_t*>(&num_bytes_),
-                                   sizeof(num_bytes_)));
+  PARQUET_THROW_NOT_OK(
+      sink->Write(reinterpret_cast<const uint8_t*>(&num_bytes_), sizeof(num_bytes_)));
   PARQUET_THROW_NOT_OK(sink->Write(reinterpret_cast<const uint8_t*>(&hash_strategy_),
                                    sizeof(hash_strategy_)));
-  PARQUET_THROW_NOT_OK(sink->Write(reinterpret_cast<const uint8_t*>(&algorithm_),
-                                   sizeof(algorithm_)));
+  PARQUET_THROW_NOT_OK(
+      sink->Write(reinterpret_cast<const uint8_t*>(&algorithm_), sizeof(algorithm_)));
   PARQUET_THROW_NOT_OK(sink->Write(data_->mutable_data(), num_bytes_));
 }
 

cpp/src/parquet/bloom_filter.h

@@ -25,9 +25,8 @@
 #include "arrow/util/bit-util.h"
 #include "arrow/util/logging.h"
 #include "parquet/hasher.h"
-#include "parquet/types.h"
-#include "parquet/util/memory.h"
 #include "parquet/platform.h"
+#include "parquet/types.h"
 
 namespace arrow {
 
@@ -61,7 +60,7 @@ class PARQUET_EXPORT BloomFilter {
   /// include bitset length, hash strategy, algorithm, and bitset.
   ///
   /// @param sink the output stream to write
-  virtual void WriteTo(OutputStream* sink) const = 0;
+  virtual void WriteTo(ArrowOutputStream* sink) const = 0;
 
   /// Get the number of bytes of bitset
   virtual uint32_t GetBitsetSize() const = 0;
@@ -190,7 +189,7 @@ class PARQUET_EXPORT BlockSplitBloomFilter : public BloomFilter {
 
   bool FindHash(uint64_t hash) const override;
   void InsertHash(uint64_t hash) override;
-  void WriteTo(OutputStream* sink) const override;
+  void WriteTo(ArrowOutputStream* sink) const override;
   uint32_t GetBitsetSize() const override { return num_bytes_; }
 
   uint64_t Hash(int64_t value) const override { return hasher_->Hash(value); }

cpp/src/parquet/column-io-benchmark.cc

@@ -24,8 +24,8 @@
 #include "parquet/column_writer.h"
 #include "parquet/file_reader.h"
 #include "parquet/metadata.h"
+#include "parquet/platform.h"
 #include "parquet/thrift.h"
-#include "parquet/util/memory.h"
 
 namespace parquet {
 
@@ -144,13 +144,14 @@ static void BM_ReadInt64Column(::benchmark::State& state) {
       properties, schema.get(), reinterpret_cast<uint8_t*>(&thrift_metadata));
 
   auto stream = CreateOutputStream();
-  std::shared_ptr<Int64Writer> writer = BuildWriter(
-      state.range(0), stream, metadata.get(), schema.get(), properties.get());
+  std::shared_ptr<Int64Writer> writer =
+      BuildWriter(state.range(0), stream, metadata.get(), schema.get(), properties.get());
   writer->WriteBatch(values.size(), definition_levels.data(), repetition_levels.data(),
                      values.data());
   writer->Close();
 
-  std::shared_ptr<Buffer> src = stream.GetBuffer();
+  std::shared_ptr<Buffer> src;
+  PARQUET_THROW_NOT_OK(stream->Finish(&src));
   std::vector<int64_t> values_out(state.range(1));
   std::vector<int16_t> definition_levels_out(state.range(1));
   std::vector<int16_t> repetition_levels_out(state.range(1));

cpp/src/parquet/column_reader.cc

@@ -39,6 +39,8 @@ using arrow::MemoryPool;
 
 namespace parquet {
 
+namespace BitUtil = ::arrow::BitUtil;
+
 LevelDecoder::LevelDecoder() : num_values_remaining_(0) {}
 
 LevelDecoder::~LevelDecoder() {}
@@ -105,8 +107,8 @@ ReaderProperties default_reader_properties() {
 class SerializedPageReader : public PageReader {
  public:
   SerializedPageReader(const std::shared_ptr<ArrowInputStream>& stream,
-                       int64_t total_num_rows,
-                       Compression::type codec, ::arrow::MemoryPool* pool)
+                       int64_t total_num_rows, Compression::type codec,
+                       ::arrow::MemoryPool* pool)
       : stream_(stream),
         decompression_buffer_(AllocateBuffer(pool, 0)),
         seen_num_rows_(0),
@@ -145,7 +147,6 @@ std::shared_ptr<Page> SerializedPageReader::NextPage() {
   // finding a page that we do know what to do with
   while (seen_num_rows_ < total_num_rows_) {
     int64_t bytes_read = 0;
-    int64_t bytes_available = 0;
     uint32_t header_size = 0;
     const uint8_t* buffer;
     uint32_t allowed_page_size = kDefaultPageHeaderSize;
@@ -162,7 +163,8 @@ std::shared_ptr<Page> SerializedPageReader::NextPage() {
       // This gets used, then set by DeserializeThriftMsg
       header_size = static_cast<uint32_t>(buffer.size());
       try {
-        DeserializeThriftMsg(buffer.data(), &header_size, &current_page_header_);
+        DeserializeThriftMsg(reinterpret_cast<const uint8_t*>(buffer.data()),
+                             &header_size, &current_page_header_);
         break;
       } catch (std::exception& e) {
         // Failed to deserialize. Double the allowed page header size and try again
@@ -257,10 +259,9 @@ std::shared_ptr<Page> SerializedPageReader::NextPage() {
   return std::shared_ptr<Page>(nullptr);
 }
 
-std::unique_ptr<PageReader> PageReader::Open(const std::shared_ptr<ArrowInputStream>& stream,
-                                             int64_t total_num_rows,
-                                             Compression::type codec,
-                                             ::arrow::MemoryPool* pool) {
+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(stream, total_num_rows, codec, pool));
 }

cpp/src/parquet/column_reader.h

@@ -25,9 +25,9 @@
 
 #include "parquet/encoding.h"
 #include "parquet/exception.h"
+#include "parquet/platform.h"
 #include "parquet/schema.h"
 #include "parquet/types.h"
-#include "parquet/platform.h"
 
 namespace arrow {
 

cpp/src/parquet/column_scanner.h

@@ -25,15 +25,11 @@
 #include <string>
 #include <vector>
 
-#include "arrow/buffer.h"
-#include "arrow/memory_pool.h"
-
 #include "parquet/column_reader.h"
 #include "parquet/exception.h"
+#include "parquet/platform.h"
 #include "parquet/schema.h"
 #include "parquet/types.h"
-#include "parquet/util/memory.h"
-#include "parquet/platform.h"
 
 namespace parquet {
 

cpp/src/parquet/column_writer-test.cc

@@ -24,12 +24,12 @@
 #include "parquet/column_reader.h"
 #include "parquet/column_writer.h"
 #include "parquet/metadata.h"
+#include "parquet/platform.h"
 #include "parquet/properties.h"
 #include "parquet/statistics.h"
 #include "parquet/test-util.h"
 #include "parquet/thrift.h"
 #include "parquet/types.h"
-#include "parquet/util/memory.h"
 
 namespace parquet {
 
@@ -281,7 +281,7 @@ class TestPrimitiveWriter : public PrimitiveTypedTest<TestType> {
 
  private:
   std::unique_ptr<ColumnChunkMetaDataBuilder> metadata_;
-  std::shared_ptr<ArrowOutputStream> sink_;
+  std::shared_ptr<::arrow::io::BufferedOutputStream> sink_;
   std::shared_ptr<WriterProperties> writer_properties_;
   std::vector<std::vector<uint8_t>> data_buffer_;
 };

cpp/src/parquet/column_writer.cc

@@ -31,11 +31,11 @@
 #include "arrow/util/rle-encoding.h"
 
 #include "parquet/metadata.h"
+#include "parquet/platform.h"
 #include "parquet/properties.h"
 #include "parquet/statistics.h"
 #include "parquet/thrift.h"
 #include "parquet/types.h"
-#include "parquet/util/memory.h"
 
 namespace parquet {
 
@@ -142,14 +142,15 @@ class SerializedPageWriter : public PageWriter {
     thrift_serializer_.reset(new ThriftSerializer);
   }
 
-  int64_t WritePageGeneric(const format::PageHeader& header, const Buffer& body) {
+  int64_t WritePageGeneric(const format::PageHeader& header, int64_t uncompressed_size,
+                           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_);
+    int64_t header_size = thrift_serializer_->Serialize(&header, sink_.get());
     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;
+    total_compressed_size_ += body.size() + header_size;
 
     return start_pos;
   }
@@ -177,7 +178,8 @@ class SerializedPageWriter : public PageWriter {
     page_header.__set_compressed_page_size(static_cast<int32_t>(compressed_data->size()));
     page_header.__set_dictionary_page_header(dict_page_header);
 
-    int64_t start_pos = WritePageGeneric(page_header, *compressed_data);
+    int64_t start_pos =
+        WritePageGeneric(page_header, uncompressed_size, *compressed_data);
     if (dictionary_page_offset_ == 0) {
       dictionary_page_offset_ = start_pos;
     }
@@ -193,7 +195,7 @@ class SerializedPageWriter : public PageWriter {
                       fallback);
 
     // Write metadata at end of column chunk
-    metadata_->WriteTo(sink_);
+    metadata_->WriteTo(sink_.get());
   }
 
   /**
@@ -236,7 +238,8 @@ class SerializedPageWriter : public PageWriter {
     page_header.__set_compressed_page_size(static_cast<int32_t>(compressed_data->size()));
     page_header.__set_data_page_header(data_page_header);
 
-    int64_t start_pos = WritePageGeneric(page_header, *compressed_data);
+    int64_t start_pos =
+        WritePageGeneric(page_header, uncompressed_size, *compressed_data);
     if (data_page_offset_ == 0) {
       data_page_offset_ = start_pos;
     }
@@ -277,13 +280,12 @@ class SerializedPageWriter : public PageWriter {
 class BufferedPageWriter : public PageWriter {
  public:
   BufferedPageWriter(const std::shared_ptr<ArrowOutputStream>& sink,
-                     Compression::type codec,
-                     ColumnChunkMetaDataBuilder* metadata,
+                     Compression::type codec, ColumnChunkMetaDataBuilder* metadata,
                      ::arrow::MemoryPool* pool = ::arrow::default_memory_pool())
-      : final_sink_(sink),
-        metadata_(metadata),
-        pager_(new SerializedPageWriter(in_memory_sink_.get(), codec, metadata, pool)) {
+      : final_sink_(sink), metadata_(metadata) {
     in_memory_sink_ = CreateOutputStream(pool);
+    pager_ = std::unique_ptr<SerializedPageWriter>(
+        new SerializedPageWriter(in_memory_sink_, codec, metadata, pool));
   }
 
   int64_t WriteDictionaryPage(const DictionaryPage& page) override {
@@ -292,9 +294,11 @@ class BufferedPageWriter : public PageWriter {
 
   void Close(bool has_dictionary, bool fallback) override {
     // index_page_offset = -1 since they are not supported
+    int64_t final_position = -1;
+    PARQUET_THROW_NOT_OK(final_sink_->Tell(&final_position));
     metadata_->Finish(
-        pager_->num_values(), pager_->dictionary_page_offset() + final_sink_->Tell(), -1,
-        pager_->data_page_offset() + final_sink_->Tell(), pager_->total_compressed_size(),
+        pager_->num_values(), pager_->dictionary_page_offset() + final_position, -1,
+        pager_->data_page_offset() + final_position, pager_->total_compressed_size(),
         pager_->total_uncompressed_size(), has_dictionary, fallback);
 
     // Write metadata at end of column chunk
@@ -319,15 +323,14 @@ class BufferedPageWriter : public PageWriter {
  private:
   std::shared_ptr<ArrowOutputStream> final_sink_;
   ColumnChunkMetaDataBuilder* metadata_;
-  std::shared_ptr<ArrowOutputStream> in_memory_sink_;
+  std::shared_ptr<::arrow::io::BufferOutputStream> in_memory_sink_;
   std::unique_ptr<SerializedPageWriter> pager_;
 };
 
-std::unique_ptr<PageWriter> PageWriter::Open(const std::shared_ptr<ArrowOutputStream>& sink,
-                                             Compression::type codec,
-                                             ColumnChunkMetaDataBuilder* metadata,
-                                             ::arrow::MemoryPool* pool,
-                                             bool buffered_row_group) {
+std::unique_ptr<PageWriter> PageWriter::Open(
+    const std::shared_ptr<ArrowOutputStream>& sink, Compression::type codec,
+    ColumnChunkMetaDataBuilder* metadata, ::arrow::MemoryPool* pool,
+    bool buffered_row_group) {
   if (buffered_row_group) {
     return std::unique_ptr<PageWriter>(
         new BufferedPageWriter(sink, codec, metadata, pool));
@@ -466,8 +469,8 @@ class ColumnWriterImpl {
   // Flag to infer if dictionary encoding has fallen back to PLAIN
   bool fallback_;
 
-  std::shared_ptr<ArrowOutputStream> definition_levels_sink_;
-  std::shared_ptr<ArrowOutputStream> repetition_levels_sink_;
+  std::shared_ptr<::arrow::io::BufferOutputStream> definition_levels_sink_;
+  std::shared_ptr<::arrow::io::BufferOutputStream> repetition_levels_sink_;
 
   std::shared_ptr<ResizableBuffer> definition_levels_rle_;
   std::shared_ptr<ResizableBuffer> repetition_levels_rle_;
@@ -517,14 +520,14 @@ void ColumnWriterImpl::AddDataPage() {
 
   if (descr_->max_definition_level() > 0) {
     definition_levels_rle_size =
-        RleEncodeLevels(definition_levels_sink_->buffer(),
-                        definition_levels_rle_.get(), descr_->max_definition_level());
+        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_->buffer(),
-                        repetition_levels_rle_.get(), descr_->max_repetition_level());
+        RleEncodeLevels(repetition_levels_sink_->buffer(), repetition_levels_rle_.get(),
+                        descr_->max_repetition_level());
   }
 
   int64_t uncompressed_size =

cpp/src/parquet/column_writer.h

@@ -26,10 +26,9 @@
 #include "parquet/column_page.h"
 #include "parquet/encoding.h"
 #include "parquet/exception.h"
+#include "parquet/platform.h"
 #include "parquet/schema.h"
 #include "parquet/types.h"
-#include "parquet/util/memory.h"
-#include "parquet/platform.h"
 
 namespace arrow {
 
@@ -83,8 +82,8 @@ class PARQUET_EXPORT PageWriter {
   virtual ~PageWriter() {}
 
   static std::unique_ptr<PageWriter> Open(
-      const std::shared_ptr<ArrowOutputStream>& 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

@@ -24,166 +24,10 @@
 #include <gtest/gtest.h>
 
 #include "parquet/exception.h"
+#include "parquet/platform.h"
 #include "parquet/test-util.h"
-#include "parquet/util/memory.h"
 
 using arrow::default_memory_pool;
 using arrow::MemoryPool;
 
-namespace parquet {
-
-class TestBufferedInputStream : public ::testing::Test {
- public:
-  void SetUp() {
-    // Create a buffer larger than source size, to check that the stream end is upholded.
-    std::shared_ptr<ResizableBuffer> buf =
-        AllocateBuffer(default_memory_pool(), source_size_ + 10);
-    ASSERT_LT(source_size_, buf->size());
-    for (int i = 0; i < source_size_; i++) {
-      buf->mutable_data()[i] = static_cast<uint8_t>(i);
-    }
-    source_ = std::make_shared<ArrowInputFile>(
-        std::make_shared<::arrow::io::BufferReader>(buf));
-    stream_.reset(new BufferedInputStream(default_memory_pool(), chunk_size_,
-                                          source_.get(), stream_offset_, stream_size_));
-  }
-
- protected:
-  int64_t source_size_ = 256;
-  int64_t stream_offset_ = 10;
-  int64_t stream_size_ = source_size_ - stream_offset_;
-  int64_t chunk_size_ = 50;
-  std::shared_ptr<RandomAccessSource> source_;
-  std::unique_ptr<BufferedInputStream> stream_;
-};
-
-TEST_F(TestBufferedInputStream, Basics) {
-  const uint8_t* output;
-  int64_t bytes_read;
-
-  // source is at offset 10
-  output = stream_->Peek(10, &bytes_read);
-  ASSERT_EQ(10, bytes_read);
-  for (int i = 0; i < 10; i++) {
-    ASSERT_EQ(10 + i, output[i]) << i;
-  }
-  output = stream_->Read(10, &bytes_read);
-  ASSERT_EQ(10, bytes_read);
-  for (int i = 0; i < 10; i++) {
-    ASSERT_EQ(10 + i, output[i]) << i;
-  }
-  output = stream_->Read(10, &bytes_read);
-  ASSERT_EQ(10, bytes_read);
-  for (int i = 0; i < 10; i++) {
-    ASSERT_EQ(20 + i, output[i]) << i;
-  }
-  stream_->Advance(5);
-  stream_->Advance(5);
-  // source is at offset 40
-  // read across buffer boundary. buffer size is 50
-  output = stream_->Read(20, &bytes_read);
-  ASSERT_EQ(20, bytes_read);
-  for (int i = 0; i < 20; i++) {
-    ASSERT_EQ(40 + i, output[i]) << i;
-  }
-  // read more than original chunk size
-  output = stream_->Read(60, &bytes_read);
-  ASSERT_EQ(60, bytes_read);
-  for (int i = 0; i < 60; i++) {
-    ASSERT_EQ(60 + i, output[i]) << i;
-  }
-
-  stream_->Advance(120);
-  // source is at offset 240
-  // read outside of source boundary. source size is 256
-  output = stream_->Read(30, &bytes_read);
-  ASSERT_EQ(16, bytes_read);
-  for (int i = 0; i < 16; i++) {
-    ASSERT_EQ(240 + i, output[i]) << i;
-  }
-  // Stream exhausted
-  output = stream_->Read(1, &bytes_read);
-  ASSERT_EQ(bytes_read, 0);
-}
-
-TEST_F(TestBufferedInputStream, LargeFirstPeek) {
-  // Test a first peek larger than chunk size
-  const uint8_t* output;
-  int64_t bytes_read;
-  int64_t n = 70;
-  ASSERT_GT(n, chunk_size_);
-
-  // source is at offset 10
-  output = stream_->Peek(n, &bytes_read);
-  ASSERT_EQ(n, bytes_read);
-  for (int i = 0; i < n; i++) {
-    ASSERT_EQ(10 + i, output[i]) << i;
-  }
-  output = stream_->Peek(n, &bytes_read);
-  ASSERT_EQ(n, bytes_read);
-  for (int i = 0; i < n; i++) {
-    ASSERT_EQ(10 + i, output[i]) << i;
-  }
-  output = stream_->Read(n, &bytes_read);
-  ASSERT_EQ(n, bytes_read);
-  for (int i = 0; i < n; i++) {
-    ASSERT_EQ(10 + i, output[i]) << i;
-  }
-  // source is at offset 10 + n
-  output = stream_->Read(20, &bytes_read);
-  ASSERT_EQ(20, bytes_read);
-  for (int i = 0; i < 20; i++) {
-    ASSERT_EQ(10 + n + i, output[i]) << i;
-  }
-}
-
-TEST_F(TestBufferedInputStream, OneByteReads) {
-  const uint8_t* output;
-  int64_t bytes_read;
-
-  for (int i = 0; i < stream_size_; ++i) {
-    output = stream_->Read(1, &bytes_read);
-    ASSERT_EQ(bytes_read, 1);
-    ASSERT_EQ(10 + i, output[0]) << i;
-  }
-  // Stream exhausted
-  output = stream_->Read(1, &bytes_read);
-  ASSERT_EQ(bytes_read, 0);
-}
-
-TEST_F(TestBufferedInputStream, BufferExactlyExhausted) {
-  // Test exhausting the buffer exactly then issuing further reads (PARQUET-1571).
-  const uint8_t* output;
-  int64_t bytes_read;
-
-  // source is at offset 10
-  int64_t n = 10;
-  output = stream_->Read(n, &bytes_read);
-  ASSERT_EQ(n, bytes_read);
-  for (int i = 0; i < n; i++) {
-    ASSERT_EQ(10 + i, output[i]) << i;
-  }
-  // source is at offset 20
-  // Exhaust buffer exactly
-  n = stream_->remaining_in_buffer();
-  output = stream_->Read(n, &bytes_read);
-  ASSERT_EQ(n, bytes_read);
-  for (int i = 0; i < n; i++) {
-    ASSERT_EQ(20 + i, output[i]) << i;
-  }
-  // source is at offset 20 + n
-  // Read new buffer
-  output = stream_->Read(10, &bytes_read);
-  ASSERT_EQ(10, bytes_read);
-  for (int i = 0; i < 10; i++) {
-    ASSERT_EQ(20 + n + i, output[i]) << i;
-  }
-  // source is at offset 30 + n
-  output = stream_->Read(10, &bytes_read);
-  ASSERT_EQ(10, bytes_read);
-  for (int i = 0; i < 10; i++) {
-    ASSERT_EQ(30 + n + i, output[i]) << i;
-  }
-}
-
-}  // namespace parquet
+namespace parquet {}  // namespace parquet

cpp/src/parquet/deprecated_io.cc

@@ -30,14 +30,13 @@
 
 namespace parquet {
 
-ParquetInputWrapper::ParquetInputWrapper(std::unique_ptr<RandomAccessSource> source,
-                    ::arrow::MemoryPool* pool)
-    : ParquetInputWrapper(source.get(), pool),
-      owned_source_(std::move(source)) {}
+ParquetInputWrapper::ParquetInputWrapper(std::unique_ptr<RandomAccessSource> source)
+    : ParquetInputWrapper(source.get()) {
+  owned_source_ = std::move(source);
+}
 
-ParquetInputWrapper::ParquetInputWrapper(RandomAccessSource* source,
-                                         ::arrow::MemoryPool* pool)
-    : source_(source), pool_(pool), closed_(false) {}
+ParquetInputWrapper::ParquetInputWrapper(RandomAccessSource* source)
+    : source_(source), closed_(false) {}
 
 ParquetInputWrapper::~ParquetInputWrapper() {
   if (!closed_) {
@@ -57,17 +56,16 @@ ParquetInputWrapper::~ParquetInputWrapper() {
   return ::arrow::Status::OK();
 }
 
-bool ParquetInputWrapper::closed() const {
-  return closed_;
-}
+bool ParquetInputWrapper::closed() const { return closed_; }
 
 ::arrow::Status ParquetInputWrapper::Seek(int64_t position) {
-  PARQUET_CATCH_NOT_OK(source_->Seek(position));
-  return ::arrow::Status::OK();
+  return ::arrow::Status::NotImplemented("Seek");
 }
 
-::arrow::Status ParquetInputWrapper::Read(int64_t nbytes, int64_t* bytes_read, void* out) {
-  PARQUET_CATCH_NOT_OK(*bytes_read = source_->Read(nbytes, out));
+::arrow::Status ParquetInputWrapper::Read(int64_t nbytes, int64_t* bytes_read,
+                                          void* out) {
+  PARQUET_CATCH_NOT_OK(*bytes_read =
+                           source_->Read(nbytes, reinterpret_cast<uint8_t*>(out)));
   return ::arrow::Status::OK();
 }
 
@@ -82,11 +80,23 @@ bool ParquetInputWrapper::closed() const {
   return ::arrow::Status::OK();
 }
 
-ParquetOutputWrapper::ParquetOutputWrapper(std::unique_ptr<OutputStream> sink)
-    : ParquetOutputWrapper(sink.get()),
-      owned_sink_(std::move(sink)) {}
+::arrow::Status ParquetInputWrapper::GetSize(int64_t* size) {
+  PARQUET_CATCH_NOT_OK(*size = source_->Tell());
+  return ::arrow::Status::OK();
+}
 
-ParquetOutputWrapper::ParquetOutputWrapper(OutputStream* sink)
+ParquetOutputWrapper::ParquetOutputWrapper(std::unique_ptr<::parquet::OutputStream> sink)
+    : ParquetOutputWrapper(sink.get()) {
+  owned_sink_ = std::move(sink);
+}
+
+ParquetOutputWrapper::ParquetOutputWrapper(
+    const std::shared_ptr<::parquet::OutputStream>& sink)
+    : ParquetOutputWrapper(sink.get()) {
+  shared_sink_ = sink;
+}
+
+ParquetOutputWrapper::ParquetOutputWrapper(::parquet::OutputStream* sink)
     : sink_(sink), closed_(false) {}
 
 ParquetOutputWrapper::~ParquetOutputWrapper() {
@@ -103,13 +113,11 @@ ParquetOutputWrapper::~ParquetOutputWrapper() {
 }
 
 ::arrow::Status ParquetOutputWrapper::Tell(int64_t* position) const {
-  PARQUET_CATCH_NOT_OK(*position = source_->Tell());
+  PARQUET_CATCH_NOT_OK(*position = sink_->Tell());
   return ::arrow::Status::OK();
 }
 
-bool ParquetOutputWrapper::closed() const {
-  return closed_;
-}
+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));

cpp/src/parquet/deprecated_io.h

@@ -20,7 +20,7 @@
 // arrow/io/interfaces.h. These legacy interfaces are being preserved
 // through a wrapper layer for one to two releases
 
-#pragma
+#pragma once
 
 #include <cstdint>
 #include <cstdlib>
@@ -85,10 +85,8 @@ class PARQUET_EXPORT OutputStream : virtual public FileInterface {
 
 class ParquetInputWrapper : public ::arrow::io::RandomAccessFile {
  public:
-  ParquetInputWrapper(std::unique_ptr<RandomAccessSource> source,
-                      ::arrow::MemoryPool* pool = ::arrow::default_memory_pool());
-  ParquetInputWrapper(RandomAccessSource* source,
-                      ::arrow::MemoryPool* pool = ::arrow::default_memory_pool());
+  explicit ParquetInputWrapper(std::unique_ptr<RandomAccessSource> source);
+  explicit ParquetInputWrapper(RandomAccessSource* source);
 
   ~ParquetInputWrapper() override;
 
@@ -100,23 +98,24 @@ class ParquetInputWrapper : public ::arrow::io::RandomAccessFile {
   // Seekable
   ::arrow::Status Seek(int64_t position) override;
 
-  // InputStream
+  // InputStream / RandomAccessFile
   ::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;
+  ::arrow::Status GetSize(int64_t* size) override;
 
  private:
   std::unique_ptr<RandomAccessSource> owned_source_;
   RandomAccessSource* source_;
-  MemoryPool* pool_;
   bool closed_;
 };
 
 class ParquetOutputWrapper : public ::arrow::io::OutputStream {
  public:
-  ParquetOutputWrapper(std::unique_ptr<OutputStream> sink);
-  ParquetOutputWrapper(OutputStream* sink);
+  explicit ParquetOutputWrapper(const std::shared_ptr<::parquet::OutputStream>& sink);
+  explicit ParquetOutputWrapper(std::unique_ptr<::parquet::OutputStream> sink);
+  explicit ParquetOutputWrapper(::parquet::OutputStream* sink);
 
   ~ParquetOutputWrapper() override;
 
@@ -129,8 +128,9 @@ class ParquetOutputWrapper : public ::arrow::io::OutputStream {
   ::arrow::Status Write(const void* data, int64_t nbytes) override;
 
  private:
-  std::unique_ptr<OutputStream> owned_sink_;
-  OutputStream* sink_;
+  std::unique_ptr<::parquet::OutputStream> owned_sink_;
+  std::shared_ptr<::parquet::OutputStream> shared_sink_;
+  ::parquet::OutputStream* sink_;
   bool closed_;
 };
 

cpp/src/parquet/encoding-benchmark.cc

@@ -26,8 +26,8 @@
 #include "arrow/type.h"
 
 #include "parquet/encoding.h"
+#include "parquet/platform.h"
 #include "parquet/schema.h"
-#include "parquet/util/memory.h"
 
 #include <random>
 

cpp/src/parquet/encoding-test.cc

@@ -30,10 +30,10 @@
 #include "arrow/util/bit-util.h"
 
 #include "parquet/encoding.h"
+#include "parquet/platform.h"
 #include "parquet/schema.h"
 #include "parquet/test-util.h"
 #include "parquet/types.h"
-#include "parquet/util/memory.h"
 
 using arrow::default_memory_pool;
 using arrow::MemoryPool;

cpp/src/parquet/encoding.cc

@@ -38,6 +38,8 @@
 
 namespace parquet {
 
+constexpr int64_t kInMemoryDefaultCapacity = 1024;
+
 namespace BitUtil = ::arrow::BitUtil;
 
 class EncoderImpl : virtual public Encoder {
@@ -80,7 +82,7 @@ class PlainEncoder : public EncoderImpl, virtual public TypedEncoder<DType> {
   void Put(const T* buffer, int num_values) override;
 
  protected:
-  std::shared_ptr<ArrowOutputStream> values_sink_;
+  std::shared_ptr<::arrow::io::BufferOutputStream> values_sink_;
 };
 
 template <typename DType>
@@ -119,8 +121,8 @@ inline void PlainEncoder<ByteArrayType>::Put(const ByteArray* src, int num_value
     if (src[i].len > 0) {
       DCHECK(nullptr != src[i].ptr) << "Value ptr cannot be NULL";
     }
-    PARQUET_THROW_NOT_OK(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));
   }
 }
 
@@ -131,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";
     }
-    PARQUET_THROW_NOT_OK(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()));
   }
 }
 
@@ -167,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::shared_ptr<ArrowOutputStream> values_sink_;
+  std::shared_ptr<::arrow::io::BufferOutputStream> values_sink_;
 
   template <typename SequenceType>
   void PutImpl(const SequenceType& src, int num_values);
@@ -186,8 +188,8 @@ void PlainBooleanEncoder::PutImpl(const SequenceType& src, int num_values) {
 
     if (bits_available_ == 0) {
       bit_writer_->Flush();
-      PARQUET_THROW_NOT_OK(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();
     }
   }
@@ -206,8 +208,8 @@ void PlainBooleanEncoder::PutImpl(const SequenceType& src, int num_values) {
 
     if (bits_available_ == 0) {
       bit_writer_->Flush();
-      PARQUET_THROW_NOT_OK(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();
     }
   }
@@ -232,8 +234,8 @@ int64_t PlainBooleanEncoder::EstimatedDataEncodedSize() {
 std::shared_ptr<Buffer> PlainBooleanEncoder::FlushValues() {
   if (bits_available_ > 0) {
     bit_writer_->Flush();
-    PARQUET_THROW_NOT_OK(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;
   }

cpp/src/parquet/encoding.h

@@ -25,8 +25,8 @@
 #include "arrow/util/bit-util.h"
 
 #include "parquet/exception.h"
-#include "parquet/types.h"
 #include "parquet/platform.h"
+#include "parquet/types.h"
 
 namespace parquet {
 

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

@@ -25,10 +25,10 @@
 #include "parquet/column_reader.h"
 #include "parquet/exception.h"
 #include "parquet/file_reader.h"
+#include "parquet/platform.h"
 #include "parquet/test-util.h"
 #include "parquet/thrift.h"
 #include "parquet/types.h"
-#include "parquet/util/memory.h"
 
 #include "arrow/io/memory.h"
 #include "arrow/status.h"
@@ -105,11 +105,9 @@ class TestPageSerde : public ::testing::Test {
     ASSERT_NO_THROW(serializer.Serialize(&page_header_, out_stream_.get()));
   }
 
-  void ResetStream() {
-    out_stream_ = CreateOutputStream();
-  }
+  void ResetStream() { out_stream_ = CreateOutputStream(); }
 
-  void EndStream() { out_buffer_ = out_stream_->GetBuffer(); }
+  void EndStream() { PARQUET_THROW_NOT_OK(out_stream_->Finish(&out_buffer_)); }
 
  protected:
   std::shared_ptr<::arrow::io::BufferOutputStream> out_stream_;
@@ -192,11 +190,14 @@ TEST_F(TestPageSerde, TestLargePageHeaders) {
 
   int max_header_size = 512 * 1024;  // 512 KB
   ASSERT_NO_FATAL_FAILURE(WriteDataPageHeader(max_header_size));
-  ASSERT_GE(max_header_size, out_stream_->Tell());
+
+  int64_t position = -1;
+  ASSERT_OK(out_stream_->Tell(&position));
+  ASSERT_GE(max_header_size, position);
 
   // check header size is between 256 KB to 16 MB
-  ASSERT_LE(stats_size, out_stream_->Tell());
-  ASSERT_GE(kDefaultMaxPageHeaderSize, out_stream_->Tell());
+  ASSERT_LE(stats_size, position);
+  ASSERT_GE(kDefaultMaxPageHeaderSize, position);
 
   InitSerializedPageReader(num_rows);
   std::shared_ptr<Page> current_page = page_reader_->NextPage();
@@ -212,10 +213,12 @@ TEST_F(TestPageSerde, TestFailLargePageHeaders) {
   // Serialize the Page header
   int max_header_size = 512 * 1024;  // 512 KB
   ASSERT_NO_FATAL_FAILURE(WriteDataPageHeader(max_header_size));
-  ASSERT_GE(max_header_size, out_stream_->Tell());
+  int64_t position = -1;
+  ASSERT_OK(out_stream_->Tell(&position));
+  ASSERT_GE(max_header_size, position);
 
   int smaller_max_size = 128 * 1024;
-  ASSERT_LE(smaller_max_size, out_stream_->Tell());
+  ASSERT_LE(smaller_max_size, position);
   InitSerializedPageReader(num_rows);
 
   // Set the max page header size to 128 KB, which is less than the current
@@ -260,7 +263,7 @@ TEST_F(TestPageSerde, Compression) {
 
       ASSERT_NO_FATAL_FAILURE(
           WriteDataPageHeader(1024, data_size, static_cast<int32_t>(actual_size)));
-      out_stream_->Write(buffer.data(), actual_size);
+      ASSERT_OK(out_stream_->Write(buffer.data(), actual_size));
     }
 
     InitSerializedPageReader(num_rows * num_pages, codec_type);
@@ -284,7 +287,7 @@ TEST_F(TestPageSerde, LZONotSupported) {
   int data_size = 1024;
   std::vector<uint8_t> faux_data(data_size);
   ASSERT_NO_FATAL_FAILURE(WriteDataPageHeader(1024, data_size, data_size));
-  out_stream_->Write(faux_data.data(), data_size);
+  ASSERT_OK(out_stream_->Write(faux_data.data(), data_size));
   ASSERT_THROW(InitSerializedPageReader(data_size, Compression::LZO), ParquetException);
 }
 
@@ -297,9 +300,8 @@ class TestParquetFileReader : public ::testing::Test {
     reader_.reset(new ParquetFileReader());
 
     auto reader = std::make_shared<BufferReader>(buffer);
-    auto wrapper = std::unique_ptr<ArrowInputFile>(new ArrowInputFile(reader));
 
-    ASSERT_THROW(reader_->Open(ParquetFileReader::Contents::Open(std::move(wrapper))),
+    ASSERT_THROW(reader_->Open(ParquetFileReader::Contents::Open(reader)),
                  ParquetException);
   }
 
@@ -327,7 +329,7 @@ TEST_F(TestParquetFileReader, InvalidFooter) {
 }
 
 TEST_F(TestParquetFileReader, IncompleteMetadata) {
-  std::shared_ptr<ArrowOutputStream> stream = CreateOutputStream();
+  auto stream = CreateOutputStream();
 
   const char* magic = "PAR1";
 
@@ -335,12 +337,13 @@ TEST_F(TestParquetFileReader, IncompleteMetadata) {
   std::vector<uint8_t> bytes(10);
   ASSERT_OK(stream->Write(bytes.data(), bytes.size()));
   uint32_t metadata_len = 24;
-  ASSERT_OK(stream->Write(reinterpret_cast<const uint8_t*>(&metadata_len), sizeof(uint32_t)));
+  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)));
 
   std::shared_ptr<Buffer> result;
   ASSERT_OK(stream->Finish(&result));
-  ASSERT_NO_FATAL_FAILURE(AssertInvalidFileThrows(buffer));
+  ASSERT_NO_FATAL_FAILURE(AssertInvalidFileThrows(result));
 }
 
 }  // namespace parquet

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

@@ -21,9 +21,9 @@
 #include "parquet/column_writer.h"
 #include "parquet/file_reader.h"
 #include "parquet/file_writer.h"
+#include "parquet/platform.h"
 #include "parquet/test-util.h"
 #include "parquet/types.h"
-#include "parquet/util/memory.h"
 
 namespace parquet {
 
@@ -104,7 +104,9 @@ class TestSerialize : public PrimitiveTypedTest<TestType> {
     }
     file_writer->Close();
 
-    auto buffer = sink->GetBuffer();
+    std::shared_ptr<Buffer> buffer;
+    PARQUET_THROW_NOT_OK(sink->Finish(&buffer));
+
     int num_rows_ = num_rowgroups_ * rows_per_rowgroup_;
 
     auto source = std::make_shared<::arrow::io::BufferReader>(buffer);

cpp/src/parquet/file_reader.cc

@@ -31,12 +31,13 @@
 
 #include "parquet/column_reader.h"
 #include "parquet/column_scanner.h"
+#include "parquet/deprecated_io.h"
 #include "parquet/exception.h"
 #include "parquet/metadata.h"
+#include "parquet/platform.h"
 #include "parquet/properties.h"
 #include "parquet/schema.h"
 #include "parquet/types.h"
-#include "parquet/util/memory.h"
 
 namespace parquet {
 
@@ -80,8 +81,8 @@ const RowGroupMetaData* RowGroupReader::metadata() const { return contents_->met
 class SerializedRowGroup : public RowGroupReader::Contents {
  public:
   SerializedRowGroup(const std::shared_ptr<ArrowInputFile>& source,
-                     FileMetaData* file_metadata,
-                     int row_group_number, const ReaderProperties& props)
+                     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);
   }
@@ -108,13 +109,15 @@ class SerializedRowGroup : public RowGroupReader::Contents {
       // The Parquet MR writer had a bug in 1.2.8 and below where it didn't include the
       // dictionary page header size in total_compressed_size and total_uncompressed_size
       // (see IMPALA-694). We add padding to compensate.
-      int64_t bytes_remaining = source_->Size() - (col_start + col_length);
+      int64_t size = -1;
+      PARQUET_THROW_NOT_OK(source_->GetSize(&size));
+      int64_t bytes_remaining = size - (col_start + col_length);
       int64_t padding = std::min<int64_t>(kMaxDictHeaderSize, bytes_remaining);
       col_length += padding;
     }
 
-    std::shared_ptr<ArrowInputStream> stream = properties_.GetStream(
-        source_, col_start, col_length);
+    std::shared_ptr<ArrowInputStream> stream =
+        properties_.GetStream(source_, col_start, col_length);
     return PageReader::Open(stream, col->num_values(), col->compression(),
                             properties_.memory_pool());
   }
@@ -138,14 +141,6 @@ class SerializedFile : public ParquetFileReader::Contents {
                  const ReaderProperties& props = default_reader_properties())
       : source_(source), properties_(props) {}
 
-  ~SerializedFile() override {
-    try {
-      Close();
-    } catch (...) {
-    }
-  }
-
-  // TODO(wesm): Do files need to be closed here?
   void Close() override {}
 
   std::shared_ptr<RowGroupReader> GetRowGroup(int i) override {
@@ -175,8 +170,8 @@ class SerializedFile : public ParquetFileReader::Contents {
 
     std::shared_ptr<Buffer> footer_buffer;
     int64_t footer_read_size = std::min(file_size, kDefaultFooterReadSize);
-    PARQUET_THROW_NOT_OK(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 (footer_buffer->size() != footer_read_size ||
@@ -184,8 +179,9 @@ class SerializedFile : public ParquetFileReader::Contents {
       throw ParquetException("Invalid parquet file. Corrupt footer.");
     }
 
-    uint32_t metadata_len =
-      *reinterpret_cast<uint32_t*>(footer_buffer->data() + footer_read_size - kFooterSize);
+    uint32_t metadata_len = *reinterpret_cast<const uint32_t*>(
+        reinterpret_cast<const uint8_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(
@@ -196,12 +192,11 @@ class SerializedFile : public ParquetFileReader::Contents {
     std::shared_ptr<Buffer> metadata_buffer;
     // Check if the footer_buffer contains the entire metadata
     if (footer_read_size >= (metadata_len + kFooterSize)) {
-      metadata_buffer = SliceBuffer(footer_buffer,
-                                    footer_read_size - metadata_len - kFooterSize,
-                                    metadata_len);
+      metadata_buffer = SliceBuffer(
+          footer_buffer, footer_read_size - metadata_len - kFooterSize, metadata_len);
     } else {
-      PARQUET_THROW_NOT_OK(source_->ReadAt(metadata_start, metadata_len,
-                                           &metadata_buffer));
+      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.");
       }
@@ -210,7 +205,7 @@ class SerializedFile : public ParquetFileReader::Contents {
   }
 
  private:
-  std::unique_ptr<ArrowInputFile> source_;
+  std::shared_ptr<ArrowInputFile> source_;
   std::shared_ptr<FileMetaData> file_metadata_;
   ReaderProperties properties_;
 };
@@ -230,10 +225,9 @@ 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(
-    const std::shared_ptr<RandomAccessSource>& source, const ReaderProperties& props,
+    const std::shared_ptr<ArrowInputFile>& source, const ReaderProperties& props,
     const std::shared_ptr<FileMetaData>& metadata) {
-  std::unique_ptr<ParquetFileReader::Contents> result(
-      new SerializedFile(source, props));
+  std::unique_ptr<ParquetFileReader::Contents> result(new SerializedFile(source, props));
 
   // Access private methods here, but otherwise unavailable
   SerializedFile* file = static_cast<SerializedFile*>(result.get());
@@ -260,7 +254,7 @@ std::unique_ptr<ParquetFileReader> ParquetFileReader::Open(
 std::unique_ptr<ParquetFileReader> ParquetFileReader::Open(
     std::unique_ptr<RandomAccessSource> source, const ReaderProperties& props,
     const std::shared_ptr<FileMetaData>& metadata) {
-  auto wrapper = std::shared_ptr<ParquetInputWrapper>(std::move(source));
+  auto wrapper = std::make_shared<ParquetInputWrapper>(std::move(source));
   return Open(wrapper, props, metadata);
 }
 

cpp/src/parquet/file_reader.h

@@ -24,8 +24,8 @@
 #include <vector>
 
 #include "parquet/metadata.h"  // IWYU pragma:: keep
-#include "parquet/properties.h"
 #include "parquet/platform.h"
+#include "parquet/properties.h"
 
 namespace parquet {
 
@@ -74,7 +74,7 @@ class PARQUET_EXPORT ParquetFileReader {
         const ReaderProperties& props = default_reader_properties(),
         const std::shared_ptr<FileMetaData>& metadata = NULLPTR);
 
-    virtual ~Contents() {}
+    virtual ~Contents() = default;
     // Perform any cleanup associated with the file contents
     virtual void Close() = 0;
     virtual std::shared_ptr<RowGroupReader> GetRowGroup(int i) = 0;
@@ -89,7 +89,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");
+  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

@@ -21,8 +21,9 @@
 #include <vector>
 
 #include "parquet/column_writer.h"
+#include "parquet/deprecated_io.h"
+#include "parquet/platform.h"
 #include "parquet/schema.h"
-#include "parquet/util/memory.h"
 
 using arrow::MemoryPool;
 
@@ -361,11 +362,15 @@ std::unique_ptr<ParquetFileWriter> ParquetFileWriter::Open(
 }
 
 void WriteFileMetaData(const FileMetaData& file_metadata, ArrowOutputStream* sink) {
+  int64_t position = -1;
+  PARQUET_THROW_NOT_OK(sink->Tell(&position));
+
   // Write MetaData
-  uint32_t metadata_len = static_cast<uint32_t>(sink->Tell());
+  uint32_t metadata_len = static_cast<uint32_t>(position);
 
   file_metadata.WriteTo(sink);
-  metadata_len = static_cast<uint32_t>(sink->Tell()) - metadata_len;
+  PARQUET_THROW_NOT_OK(sink->Tell(&position));
+  metadata_len = static_cast<uint32_t>(position) - metadata_len;
 
   // Write Footer
   PARQUET_THROW_NOT_OK(sink->Write(reinterpret_cast<uint8_t*>(&metadata_len), 4));

cpp/src/parquet/file_writer.h

@@ -24,9 +24,9 @@
 
 #include "parquet/exception.h"
 #include "parquet/metadata.h"
+#include "parquet/platform.h"
 #include "parquet/properties.h"
 #include "parquet/schema.h"
-#include "parquet/platform.h"
 
 namespace arrow {
 
@@ -105,11 +105,13 @@ class PARQUET_EXPORT RowGroupWriter {
   std::unique_ptr<Contents> contents_;
 };
 
+ARROW_DEPRECATED("Use version with arrow::io::OutputStream*")
 PARQUET_EXPORT
 void WriteFileMetaData(const FileMetaData& file_metadata, OutputStream* sink);
 
 PARQUET_EXPORT
-void WriteFileMetaData(const FileMetaData& file_metadata, ArrowOutputStream* sink);
+void WriteFileMetaData(const FileMetaData& file_metadata,
+                       ::arrow::io::OutputStream* sink);
 
 class PARQUET_EXPORT ParquetFileWriter {
  public:

cpp/src/parquet/metadata.cc

@@ -24,6 +24,7 @@
 
 #include "parquet/exception.h"
 #include "parquet/metadata.h"
+#include "parquet/platform.h"
 #include "parquet/schema-internal.h"
 #include "parquet/schema.h"
 #include "parquet/statistics.h"
@@ -33,8 +34,6 @@
 
 namespace parquet {
 
-class OutputStream;
-
 const ApplicationVersion& ApplicationVersion::PARQUET_251_FIXED_VERSION() {
   static ApplicationVersion version("parquet-mr", 1, 8, 0);
   return version;
@@ -365,7 +364,7 @@ class FileMetaData::FileMetaDataImpl {
 
   const ApplicationVersion& writer_version() const { return writer_version_; }
 
-  void WriteTo(OutputStream* dst) const {
+  void WriteTo(::arrow::io::OutputStream* dst) const {
     ThriftSerializer serializer;
     serializer.Serialize(metadata_.get(), dst);
   }
@@ -493,7 +492,9 @@ std::shared_ptr<const KeyValueMetadata> FileMetaData::key_value_metadata() const
 
 void FileMetaData::set_file_path(const std::string& path) { impl_->set_file_path(path); }
 
-void FileMetaData::WriteTo(OutputStream* dst) const { return impl_->WriteTo(dst); }
+void FileMetaData::WriteTo(::arrow::io::OutputStream* dst) const {
+  return impl_->WriteTo(dst);
+}
 
 ApplicationVersion::ApplicationVersion(const std::string& application, int major,
                                        int minor, int patch)
@@ -666,7 +667,7 @@ class ColumnChunkMetaDataBuilder::ColumnChunkMetaDataBuilderImpl {
     column_chunk_->meta_data.__set_encodings(thrift_encodings);
   }
 
-  void WriteTo(OutputStream* sink) {
+  void WriteTo(::arrow::io::OutputStream* sink) {
     ThriftSerializer serializer;
     serializer.Serialize(column_chunk_, sink);
   }
@@ -731,7 +732,9 @@ void ColumnChunkMetaDataBuilder::Finish(int64_t num_values,
                 compressed_size, uncompressed_size, has_dictionary, dictionary_fallback);
 }
 
-void ColumnChunkMetaDataBuilder::WriteTo(OutputStream* sink) { impl_->WriteTo(sink); }
+void ColumnChunkMetaDataBuilder::WriteTo(::arrow::io::OutputStream* sink) {
+  impl_->WriteTo(sink);
+}
 
 const ColumnDescriptor* ColumnChunkMetaDataBuilder::descr() const {
   return impl_->descr();

cpp/src/parquet/metadata.h

@@ -26,15 +26,14 @@
 #include "arrow/util/key_value_metadata.h"
 #include "arrow/util/macros.h"
 
+#include "parquet/platform.h"
 #include "parquet/properties.h"
 #include "parquet/types.h"
-#include "parquet/platform.h"
 
 namespace parquet {
 
 class ColumnDescriptor;
 class EncodedStatistics;
-class OutputStream;
 class Statistics;
 class SchemaDescriptor;
 
@@ -185,7 +184,7 @@ class PARQUET_EXPORT FileMetaData {
 
   const ApplicationVersion& writer_version() const;
 
-  void WriteTo(OutputStream* dst) const;
+  void WriteTo(::arrow::io::OutputStream* dst) const;
 
   // Return const-pointer to make it clear that this object is not to be copied
   const SchemaDescriptor* schema() const;
@@ -235,7 +234,7 @@ class PARQUET_EXPORT ColumnChunkMetaDataBuilder {
   const void* contents() const;
 
   // For writing metadata at end of column chunk
-  void WriteTo(OutputStream* sink);
+  void WriteTo(::arrow::io::OutputStream* sink);
 
  private:
   explicit ColumnChunkMetaDataBuilder(const std::shared_ptr<WriterProperties>& props,

cpp/src/parquet/murmur3.h

@@ -25,8 +25,8 @@
 #include <cstdint>
 
 #include "parquet/hasher.h"
-#include "parquet/types.h"
 #include "parquet/platform.h"
+#include "parquet/types.h"
 
 namespace parquet {
 

cpp/src/parquet/platform.cc

@@ -26,13 +26,10 @@
 
 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> CreateOutputStream(MemoryPool* pool) {
   std::shared_ptr<::arrow::io::BufferOutputStream> stream;
-  PARQUET_THROW_NOT_OK(::arrow::io::BufferOutputStream::Create(
-      kDefaultOutputStreamSize, pool, &stream));
+  PARQUET_THROW_NOT_OK(
+      ::arrow::io::BufferOutputStream::Create(kDefaultOutputStreamSize, pool, &stream));
   return stream;
 }
 

cpp/src/parquet/platform.h

@@ -17,11 +17,14 @@
 
 #pragma once
 
-#include "arrow/buffer.h"  // IWYU pragma: export
+#include <memory>
+
+#include "arrow/buffer.h"         // IWYU pragma: export
 #include "arrow/io/interfaces.h"  // IWYU pragma: export
-#include "arrow/memory_pool.h"  // IWYU pragma: export
-#include "arrow/status.h"  // IWYU pragma: export
-#include "arrow/util/macros.h"  // IWYU pragma: export
+#include "arrow/io/memory.h"      // IWYU pragma: export
+#include "arrow/memory_pool.h"    // IWYU pragma: export
+#include "arrow/status.h"         // IWYU pragma: export
+#include "arrow/util/macros.h"    // IWYU pragma: export
 
 #if defined(_WIN32) || defined(__CYGWIN__)
 
@@ -83,6 +86,7 @@
 namespace parquet {
 
 using Buffer = ::arrow::Buffer;
+using MemoryPool = ::arrow::MemoryPool;
 using MutableBuffer = ::arrow::MutableBuffer;
 using ResizableBuffer = ::arrow::ResizableBuffer;
 using ResizableBuffer = ::arrow::ResizableBuffer;
@@ -90,7 +94,9 @@ using ArrowInputFile = ::arrow::io::RandomAccessFile;
 using ArrowInputStream = ::arrow::io::InputStream;
 using ArrowOutputStream = ::arrow::io::OutputStream;
 
-std::shared_ptr<ArrowOutputStream> CreateOutputStream(
+constexpr int64_t kDefaultOutputStreamSize = 1024;
+
+std::shared_ptr<::arrow::io::BufferOutputStream> CreateOutputStream(
     ::arrow::MemoryPool* pool = ::arrow::default_memory_pool());
 
 PARQUET_EXPORT

cpp/src/parquet/properties.cc

@@ -15,6 +15,8 @@
 // specific language governing permissions and limitations
 // under the License.
 
+#include <utility>
+
 #include "parquet/properties.h"
 
 #include "arrow/io/buffered.h"
@@ -22,19 +24,20 @@
 
 namespace parquet {
 
-std::unique_ptr<ArrowInputStream> ReaderProperties::GetStream(
-    std::shared_ptr<ArrowInputFile> source,
-    int64_t start, int64_t num_bytes) {
-if (buffered_stream_enabled_) {
-  std::shared_ptr<::arrow::io::BufferedInputStream> stream;
-  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 {
-  std::shared_ptr<Buffer> data;
-  PARQUET_THROW_NOT_OK(source->ReadAt(start, num_bytes, &data));
-  return std::make_shared<::arrow::io::BufferReader>(data);
-}
+std::shared_ptr<ArrowInputStream> ReaderProperties::GetStream(
+    std::shared_ptr<ArrowInputFile> source, int64_t start, int64_t num_bytes) {
+  if (buffered_stream_enabled_) {
+    std::shared_ptr<::arrow::io::BufferedInputStream> stream;
+    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 std::move(stream);
+  } else {
+    std::shared_ptr<Buffer> data;
+    PARQUET_THROW_NOT_OK(source->ReadAt(start, num_bytes, &data));
+    return std::make_shared<::arrow::io::BufferReader>(data);
+  }
 }
+
+}  // namespace parquet

cpp/src/parquet/properties.h

@@ -24,9 +24,9 @@
 
 #include "parquet/exception.h"
 #include "parquet/parquet_version.h"
+#include "parquet/platform.h"
 #include "parquet/schema.h"
 #include "parquet/types.h"
-#include "parquet/platform.h"
 
 namespace parquet {
 
@@ -47,7 +47,7 @@ class PARQUET_EXPORT ReaderProperties {
 
   ::arrow::MemoryPool* memory_pool() const { return pool_; }
 
-  std::unique_ptr<ArrowInputStream> GetStream(std::shared_ptr<ArrowInputFile> source,
+  std::shared_ptr<ArrowInputStream> GetStream(std::shared_ptr<ArrowInputFile> source,
                                               int64_t start, int64_t num_bytes);
 
   bool is_buffered_stream_enabled() const { return buffered_stream_enabled_; }

cpp/src/parquet/reader-test.cc

@@ -29,9 +29,9 @@
 #include "parquet/column_scanner.h"
 #include "parquet/file_reader.h"
 #include "parquet/metadata.h"
+#include "parquet/platform.h"
 #include "parquet/printer.h"
 #include "parquet/test-util.h"
-#include "parquet/util/memory.h"
 
 namespace parquet {
 
@@ -183,29 +183,6 @@ class TestLocalFile : public ::testing::Test {
   std::shared_ptr<::arrow::io::ReadableFile> handle;
 };
 
-class HelperFileClosed : public ArrowInputFile {
- public:
-  explicit HelperFileClosed(const std::shared_ptr<::arrow::io::RandomAccessFile>& file,
-                            bool* close_called)
-      : ArrowInputFile(file), close_called_(close_called) {}
-
-  void Close() override { *close_called_ = true; }
-
- private:
-  bool* close_called_;
-};
-
-TEST_F(TestLocalFile, FileClosedOnDestruction) {
-  bool close_called = false;
-  {
-    auto contents = ParquetFileReader::Contents::Open(
-        std::make_shared<HelperFileClosed>(handle, &close_called));
-    std::unique_ptr<ParquetFileReader> result(new ParquetFileReader());
-    result->Open(std::move(contents));
-  }
-  ASSERT_TRUE(close_called);
-}
-
 TEST_F(TestLocalFile, OpenWithMetadata) {
   // PARQUET-808
   std::stringstream ss;

cpp/src/parquet/statistics-test.cc

@@ -28,12 +28,12 @@
 #include "parquet/column_writer.h"
 #include "parquet/file_reader.h"
 #include "parquet/file_writer.h"
+#include "parquet/platform.h"
 #include "parquet/schema.h"
 #include "parquet/statistics.h"
 #include "parquet/test-util.h"
 #include "parquet/thrift.h"
 #include "parquet/types.h"
-#include "parquet/util/memory.h"
 
 using arrow::default_memory_pool;
 using arrow::MemoryPool;
@@ -841,7 +841,9 @@ TEST_F(TestStatisticsSortOrderFLBA, UnknownSortOrder) {
   this->SetUpSchema();
   this->WriteParquet();
 
-  auto pbuffer = parquet_sink_->GetBuffer();
+  std::shared_ptr<Buffer> pbuffer;
+  PARQUET_THROW_NOT_OK(parquet_sink_->Finish(&pbuffer));
+
   // Create a ParquetReader instance
   std::unique_ptr<parquet::ParquetFileReader> parquet_reader =
       parquet::ParquetFileReader::Open(

cpp/src/parquet/test-util.h

@@ -38,7 +38,7 @@
 #include "parquet/column_reader.h"
 #include "parquet/column_writer.h"
 #include "parquet/encoding.h"
-#include "parquet/util/memory.h"
+#include "parquet/platform.h"
 
 namespace parquet {
 
@@ -361,8 +361,8 @@ class DataPageBuilder {
     encoder.Encode(static_cast<int>(levels.size()), levels.data());
 
     int32_t rle_bytes = encoder.len();
-    PARQUET_THROW_NOT_OK(sink_->Write(reinterpret_cast<const uint8_t*>(&rle_bytes),
-                                      sizeof(int32_t)));
+    PARQUET_THROW_NOT_OK(
+        sink_->Write(reinterpret_cast<const uint8_t*>(&rle_bytes), sizeof(int32_t)));
     PARQUET_THROW_NOT_OK(sink_->Write(encode_buffer.data(), rle_bytes));
   }
 };
@@ -408,11 +408,12 @@ static std::shared_ptr<DataPageV1> MakeDataPage(
     page_builder.AppendValues(d, values, encoding);
     num_values = page_builder.num_values();
   } else {  // DICTIONARY PAGES
-    page_stream.Write(indices, indices_size);
+    PARQUET_THROW_NOT_OK(page_stream->Write(indices, indices_size));
     num_values = std::max(page_builder.num_values(), num_vals);
   }
 
-  auto buffer = page_stream.GetBuffer();
+  std::shared_ptr<Buffer> buffer;
+  PARQUET_THROW_NOT_OK(page_stream->Finish(&buffer));
 
   return std::make_shared<DataPageV1>(buffer, num_values, encoding,
                                       page_builder.def_level_encoding(),