More refactoring

cpp/src/arrow/io/memory.h

@@ -17,12 +17,12 @@
 
 // Public API for different memory sharing / IO mechanisms
 
-#ifndef ARROW_IO_MEMORY_H
-#define ARROW_IO_MEMORY_H
+#pragma once
 
 #include <cstdint>
 #include <memory>
 
+#include "arrow/buffer.h"
 #include "arrow/io/interfaces.h"
 #include "arrow/memory_pool.h"
 #include "arrow/util/string_view.h"
@@ -74,7 +74,7 @@ class ARROW_EXPORT BufferOutputStream : public OutputStream {
 
   /// \brief Return reference to current in-progress buffer. May
   /// become invalid after any Write calls
-  const Buffer& buffer() const { return buffer_; }
+  const Buffer& buffer() const { return *buffer_; }
 
   /// \brief Reset position to 0 without modifying capacity or zeroing data
   void Clear();
@@ -177,5 +177,3 @@ class ARROW_EXPORT BufferReader : public RandomAccessFile {
 
 }  // namespace io
 }  // namespace arrow
-
-#endif  // ARROW_IO_MEMORY_H

cpp/src/parquet/CMakeLists.txt

@@ -285,6 +285,7 @@ add_parquet_test(bloom_filter-test)
 add_parquet_test(column_reader-test)
 add_parquet_test(column_scanner-test)
 add_parquet_test(column_writer-test)
+add_parquet_test(deprecated_io-test)
 add_parquet_test(file-serialize-test)
 add_parquet_test(properties-test)
 add_parquet_test(statistics-test)

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

@@ -327,8 +327,7 @@ using ParquetWriter = TypedColumnWriter<ParquetDataType<T>>;
 void WriteTableToBuffer(const std::shared_ptr<Table>& table, int64_t row_group_size,
                         const std::shared_ptr<ArrowWriterProperties>& arrow_properties,
                         std::shared_ptr<Buffer>* out) {
-  auto sink = std::make_shared<InMemoryOutputStream>();
-
+  auto sink = CreateOutputStream();
   ASSERT_OK_NO_THROW(WriteTable(*table, ::arrow::default_memory_pool(), sink,
                                 row_group_size, default_writer_properties(),
                                 arrow_properties));
@@ -439,7 +438,7 @@ class TestParquetIO : public ::testing::Test {
 
   std::unique_ptr<ParquetFileWriter> MakeWriter(
       const std::shared_ptr<GroupNode>& schema) {
-    sink_ = std::make_shared<InMemoryOutputStream>();
+    sink_ = CreateOutputStream();
     return ParquetFileWriter::Open(sink_, schema);
   }
 
@@ -552,7 +551,11 @@ class TestParquetIO : public ::testing::Test {
     ASSERT_OK_NO_THROW(writer.Close());
   }
 
-  std::shared_ptr<InMemoryOutputStream> sink_;
+  void ResetStream() {
+    sink_ = CreateOutputStream();
+  }
+
+  std::shared_ptr<ArrowOutputStream> sink_;
 };
 
 // We have separate tests for UInt32Type as this is currently the only type
@@ -592,7 +595,8 @@ TYPED_TEST(TestParquetIO, SingleColumnTableRequiredWrite) {
   std::shared_ptr<Array> values;
   ASSERT_OK(NonNullArray<TypeParam>(SMALL_SIZE, &values));
   std::shared_ptr<Table> table = MakeSimpleTable(values, false);
-  this->sink_ = std::make_shared<InMemoryOutputStream>();
+
+  ResetSink();
   ASSERT_OK_NO_THROW(WriteTable(*table, ::arrow::default_memory_pool(), this->sink_,
                                 values->length(), default_writer_properties()));
 
@@ -747,7 +751,8 @@ TYPED_TEST(TestParquetIO, SingleColumnTableRequiredChunkedWrite) {
   std::shared_ptr<Array> values;
   ASSERT_OK(NonNullArray<TypeParam>(LARGE_SIZE, &values));
   std::shared_ptr<Table> table = MakeSimpleTable(values, false);
-  this->sink_ = std::make_shared<InMemoryOutputStream>();
+
+  ResetSink();
   ASSERT_OK_NO_THROW(WriteTable(*table, default_memory_pool(), this->sink_, 512,
                                 default_writer_properties()));
 
@@ -758,7 +763,8 @@ TYPED_TEST(TestParquetIO, SingleColumnTableRequiredChunkedWriteArrowIO) {
   std::shared_ptr<Array> values;
   ASSERT_OK(NonNullArray<TypeParam>(LARGE_SIZE, &values));
   std::shared_ptr<Table> table = MakeSimpleTable(values, false);
-  this->sink_ = std::make_shared<InMemoryOutputStream>();
+
+  ResetSink();
   auto buffer = AllocateBuffer();
 
   {
@@ -817,7 +823,7 @@ TYPED_TEST(TestParquetIO, SingleColumnTableOptionalChunkedWrite) {
 
   ASSERT_OK(NullableArray<TypeParam>(LARGE_SIZE, 100, kDefaultSeed, &values));
   std::shared_ptr<Table> table = MakeSimpleTable(values, true);
-  this->sink_ = std::make_shared<InMemoryOutputStream>();
+  ResetSink();
   ASSERT_OK_NO_THROW(WriteTable(*table, ::arrow::default_memory_pool(), this->sink_, 512,
                                 default_writer_properties()));
 
@@ -828,7 +834,7 @@ TYPED_TEST(TestParquetIO, FileMetaDataWrite) {
   std::shared_ptr<Array> values;
   ASSERT_OK(NonNullArray<TypeParam>(SMALL_SIZE, &values));
   std::shared_ptr<Table> table = MakeSimpleTable(values, false);
-  this->sink_ = std::make_shared<InMemoryOutputStream>();
+  ResetSink();
   ASSERT_OK_NO_THROW(WriteTable(*table, ::arrow::default_memory_pool(), this->sink_,
                                 values->length(), default_writer_properties()));
 
@@ -838,7 +844,7 @@ TYPED_TEST(TestParquetIO, FileMetaDataWrite) {
   ASSERT_EQ(1, metadata->num_columns());
   ASSERT_EQ(100, metadata->num_rows());
 
-  this->sink_ = std::make_shared<InMemoryOutputStream>();
+  ResetSink();
 
   ASSERT_OK_NO_THROW(::parquet::arrow::WriteFileMetaData(*metadata, this->sink_.get()));
 
@@ -887,7 +893,7 @@ TEST_F(TestInt96ParquetIO, ReadIntoTimestamp) {
 
   // We cannot write this column with Arrow, so we have to use the plain parquet-cpp API
   // to write an Int96 file.
-  this->sink_ = std::make_shared<InMemoryOutputStream>();
+  ResetSink();
   auto writer = ParquetFileWriter::Open(this->sink_, schema);
   RowGroupWriter* rg_writer = writer->AppendRowGroup();
   ColumnWriter* c_writer = rg_writer->NextColumn();
@@ -916,7 +922,7 @@ TEST_F(TestUInt32ParquetIO, Parquet_2_0_Compability) {
   std::shared_ptr<Table> table = MakeSimpleTable(values, true);
 
   // Parquet 2.0 roundtrip should yield an uint32_t column again
-  this->sink_ = std::make_shared<InMemoryOutputStream>();
+  ResetSink();
   std::shared_ptr<::parquet::WriterProperties> properties =
       ::parquet::WriterProperties::Builder()
           .version(ParquetVersion::PARQUET_2_0)
@@ -938,7 +944,7 @@ TEST_F(TestUInt32ParquetIO, Parquet_1_0_Compability) {
 
   // Parquet 1.0 returns an int64_t column as there is no way to tell a Parquet 1.0
   // reader that a column is unsigned.
-  this->sink_ = std::make_shared<InMemoryOutputStream>();
+  ResetSink();
   std::shared_ptr<::parquet::WriterProperties> properties =
       ::parquet::WriterProperties::Builder()
           .version(ParquetVersion::PARQUET_1_0)
@@ -994,7 +1000,7 @@ TEST_F(TestStringParquetIO, EmptyStringColumnRequiredWrite) {
   }
   ASSERT_OK(builder.Finish(&values));
   std::shared_ptr<Table> table = MakeSimpleTable(values, false);
-  this->sink_ = std::make_shared<InMemoryOutputStream>();
+  ResetSink();
   ASSERT_OK_NO_THROW(WriteTable(*table, ::arrow::default_memory_pool(), this->sink_,
                                 values->length(), default_writer_properties()));
 
@@ -1017,7 +1023,7 @@ TEST_F(TestNullParquetIO, NullColumn) {
   for (int32_t num_rows : {0, SMALL_SIZE}) {
     std::shared_ptr<Array> values = std::make_shared<::arrow::NullArray>(num_rows);
     std::shared_ptr<Table> table = MakeSimpleTable(values, true /* nullable */);
-    this->sink_ = std::make_shared<InMemoryOutputStream>();
+    ResetSink();
 
     const int64_t chunk_size = std::max(static_cast<int64_t>(1), table->num_rows());
     ASSERT_OK_NO_THROW(WriteTable(*table, ::arrow::default_memory_pool(), this->sink_,
@@ -1047,7 +1053,7 @@ TEST_F(TestNullParquetIO, NullListColumn) {
                                              default_memory_pool(), &list_array));
 
     std::shared_ptr<Table> table = MakeSimpleTable(list_array, false /* nullable */);
-    this->sink_ = std::make_shared<InMemoryOutputStream>();
+    ResetSink();
 
     const int64_t chunk_size = std::max(static_cast<int64_t>(1), table->num_rows());
     ASSERT_OK_NO_THROW(WriteTable(*table, ::arrow::default_memory_pool(), this->sink_,
@@ -1076,7 +1082,7 @@ TEST_F(TestNullParquetIO, NullDictionaryColumn) {
   std::shared_ptr<Array> dict_values =
       std::make_shared<::arrow::DictionaryArray>(dict_type, indices, dict);
   std::shared_ptr<Table> table = MakeSimpleTable(dict_values, true);
-  this->sink_ = std::make_shared<InMemoryOutputStream>();
+  ResetSink();
   ASSERT_OK_NO_THROW(WriteTable(*table, ::arrow::default_memory_pool(), this->sink_,
                                 dict_values->length(), default_writer_properties()));
 
@@ -1418,7 +1424,7 @@ TEST(TestArrowReadWrite, CoerceTimestampsLosePrecision) {
   auto t3 = Table::Make(s3, {c3});
   auto t4 = Table::Make(s4, {c4});
 
-  auto sink = std::make_shared<InMemoryOutputStream>();
+  auto sink = CreateOutputStream();
 
   // OK to write to millis
   auto coerce_millis =
@@ -1814,7 +1820,7 @@ std::shared_ptr<Table> InvalidTable() {
 
 TEST(TestArrowReadWrite, InvalidTable) {
   // ARROW-4774: Shouldn't segfault on writing an invalid table.
-  auto sink = std::make_shared<InMemoryOutputStream>();
+  auto sink = CreateOutputStream();
   auto invalid_table = InvalidTable();
 
   ASSERT_RAISES(Invalid, WriteTable(*invalid_table, ::arrow::default_memory_pool(), sink,
@@ -1942,7 +1948,7 @@ TEST(TestArrowWrite, CheckChunkSize) {
   std::shared_ptr<Table> table;
   ASSERT_NO_FATAL_FAILURE(MakeDoubleTable(num_columns, num_rows, 1, &table));
 
-  auto sink = std::make_shared<InMemoryOutputStream>();
+  auto sink = CreateOutputStream();
 
   ASSERT_RAISES(Invalid,
                 WriteTable(*table, ::arrow::default_memory_pool(), sink, chunk_size));
@@ -1962,7 +1968,7 @@ class TestNestedSchemaRead : public ::testing::TestWithParam<Repetition::type> {
   }
 
   void InitNewParquetFile(const std::shared_ptr<GroupNode>& schema, int num_rows) {
-    nested_parquet_ = std::make_shared<InMemoryOutputStream>();
+    nested_parquet_ = CreateOutputStream();
 
     writer_ = parquet::ParquetFileWriter::Open(nested_parquet_, schema,
                                                default_writer_properties());
@@ -2197,7 +2203,7 @@ class TestNestedSchemaRead : public ::testing::TestWithParam<Repetition::type> {
     std::shared_ptr<::arrow::Int32Array> expected_;
   };
 
-  std::shared_ptr<InMemoryOutputStream> nested_parquet_;
+  std::shared_ptr<ArrowOutputStream> nested_parquet_;
   std::unique_ptr<FileReader> reader_;
   std::unique_ptr<ParquetFileWriter> writer_;
   RowGroupWriter* row_group_writer_;

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

@@ -147,7 +147,7 @@ static void BM_WriteColumn(::benchmark::State& state) {
   std::shared_ptr<::arrow::Table> table = TableFromVector<ParquetType>(values, nullable);
 
   while (state.KeepRunning()) {
-    auto output = std::make_shared<InMemoryOutputStream>();
+    auto output = CreateOutputStream();
     EXIT_NOT_OK(
         WriteTable(*table, ::arrow::default_memory_pool(), output, BENCHMARK_SIZE));
   }
@@ -172,7 +172,7 @@ static void BM_ReadColumn(::benchmark::State& state) {
 
   std::vector<T> values(BENCHMARK_SIZE, static_cast<T>(128));
   std::shared_ptr<::arrow::Table> table = TableFromVector<ParquetType>(values, nullable);
-  auto output = std::make_shared<InMemoryOutputStream>();
+  auto output = CreateOutputStream();
   EXIT_NOT_OK(WriteTable(*table, ::arrow::default_memory_pool(), output, BENCHMARK_SIZE));
   std::shared_ptr<Buffer> buffer = output->GetBuffer();
 
@@ -201,7 +201,7 @@ BENCHMARK_TEMPLATE2(BM_ReadColumn, true, BooleanType);
 static void BM_ReadIndividualRowGroups(::benchmark::State& state) {
   std::vector<int64_t> values(BENCHMARK_SIZE, 128);
   std::shared_ptr<::arrow::Table> table = TableFromVector<Int64Type>(values, true);
-  auto output = std::make_shared<InMemoryOutputStream>();
+  auto output = CreateOutputStream();
   // This writes 10 RowGroups
   EXIT_NOT_OK(
       WriteTable(*table, ::arrow::default_memory_pool(), output, BENCHMARK_SIZE / 10));
@@ -233,7 +233,7 @@ BENCHMARK(BM_ReadIndividualRowGroups);
 static void BM_ReadMultipleRowGroups(::benchmark::State& state) {
   std::vector<int64_t> values(BENCHMARK_SIZE, 128);
   std::shared_ptr<::arrow::Table> table = TableFromVector<Int64Type>(values, true);
-  auto output = std::make_shared<InMemoryOutputStream>();
+  auto output = CreateOutputStream();
   // This writes 10 RowGroups
   EXIT_NOT_OK(
       WriteTable(*table, ::arrow::default_memory_pool(), output, BENCHMARK_SIZE / 10));

cpp/src/parquet/bloom_filter-test.cc

@@ -75,11 +75,13 @@ TEST(BasicTest, TestBloomFilter) {
   }
 
   // Serialize Bloom filter to memory output stream
-  InMemoryOutputStream sink;
-  bloom_filter.WriteTo(&sink);
+  auto sink = CreateOutputStream();
+  bloom_filter.WriteTo(sink.get());
 
   // Deserialize Bloom filter from memory
-  InMemoryInputStream source(sink.GetBuffer());
+  std::shared_ptr<Buffer> buffer;
+  ASSERT_OK(sink->Finish(&buffer));
+  ::arrow::io::BufferReader source(buffer);
 
   BlockSplitBloomFilter de_bloom = BlockSplitBloomFilter::Deserialize(&source);
 
@@ -172,7 +174,7 @@ TEST(CompatibilityTest, TestBloomFilter) {
   std::shared_ptr<Buffer> buffer(new Buffer(bitset.get(), size));
   PARQUET_THROW_NOT_OK(handle->Read(size, &buffer));
 
-  InMemoryInputStream source(buffer);
+  ::arrow::io::BufferReader source(buffer);
   BlockSplitBloomFilter bloom_filter1 = BlockSplitBloomFilter::Deserialize(&source);
 
   for (int i = 0; i < 4; i++) {
@@ -193,8 +195,8 @@ TEST(CompatibilityTest, TestBloomFilter) {
   }
 
   // Serialize Bloom filter to memory output stream
-  InMemoryOutputStream sink;
-  bloom_filter2.WriteTo(&sink);
+  auto sink = CreateOutputStream();
+  bloom_filter2.WriteTo(&sink.get());
   std::shared_ptr<Buffer> buffer1 = sink.GetBuffer();
 
   PARQUET_THROW_NOT_OK(handle->Seek(0));

cpp/src/parquet/bloom_filter.cc

@@ -70,18 +70,18 @@ void BlockSplitBloomFilter::Init(const uint8_t* bitset, uint32_t num_bytes) {
   this->hasher_.reset(new MurmurHash3());
 }
 
-BlockSplitBloomFilter BlockSplitBloomFilter::Deserialize(InputStream* input) {
+BlockSplitBloomFilter BlockSplitBloomFilter::Deserialize(ArrowInputStream* input) {
   int64_t bytes_available;
 
   const uint8_t* read_buffer = NULL;
-  read_buffer = input->Read(sizeof(uint32_t), &bytes_available);
+  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) {
     throw ParquetException("Failed to deserialize from input stream");
   }
   uint32_t len;
   memcpy(&len, read_buffer, sizeof(uint32_t));
 
-  read_buffer = input->Read(sizeof(uint32_t), &bytes_available);
+  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) {
     throw ParquetException("Failed to deserialize from input stream");
   }
@@ -91,7 +91,7 @@ BlockSplitBloomFilter BlockSplitBloomFilter::Deserialize(InputStream* input) {
     throw ParquetException("Unsupported hash strategy");
   }
 
-  read_buffer = input->Read(sizeof(uint32_t), &bytes_available);
+  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) {
     throw ParquetException("Failed to deserialize from input stream");
   }
@@ -102,17 +102,23 @@ BlockSplitBloomFilter BlockSplitBloomFilter::Deserialize(InputStream* input) {
   }
 
   BlockSplitBloomFilter bloom_filter;
-  bloom_filter.Init(input->Read(len, &bytes_available), len);
+
+  std::shared_ptr<Buffer> buffer;
+  PARQUET_THROW_NOT_OK(input->Read(len, &buffer));
+  bloom_filter.Init(buffer->data(), len);
   return bloom_filter;
 }
 
-void BlockSplitBloomFilter::WriteTo(OutputStream* sink) const {
+void BlockSplitBloomFilter::WriteTo(ArrowOutputStream* sink) const {
   DCHECK(sink != nullptr);
 
-  sink->Write(reinterpret_cast<const uint8_t*>(&num_bytes_), sizeof(num_bytes_));
-  sink->Write(reinterpret_cast<const uint8_t*>(&hash_strategy_), sizeof(hash_strategy_));
-  sink->Write(reinterpret_cast<const uint8_t*>(&algorithm_), sizeof(algorithm_));
-  sink->Write(data_->mutable_data(), 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(data_->mutable_data(), num_bytes_));
 }
 
 void BlockSplitBloomFilter::SetMask(uint32_t key, BlockMask& block_mask) const {

cpp/src/parquet/bloom_filter.h

@@ -208,7 +208,7 @@ class PARQUET_EXPORT BlockSplitBloomFilter : public BloomFilter {
   ///
   /// @param input_stream The input stream from which to construct the Bloom filter
   /// @return The BlockSplitBloomFilter.
-  static BlockSplitBloomFilter Deserialize(InputStream* input_stream);
+  static BlockSplitBloomFilter Deserialize(ArrowInputStream* input_stream);
 
  private:
   // Bytes in a tiny Bloom filter block.

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

@@ -33,7 +33,8 @@ using schema::PrimitiveNode;
 
 namespace benchmark {
 
-std::shared_ptr<Int64Writer> BuildWriter(int64_t output_size, OutputStream* dst,
+std::shared_ptr<Int64Writer> BuildWriter(int64_t output_size,
+                                         const std::shared_ptr<ArrowOutputStream>& dst,
                                          ColumnChunkMetaDataBuilder* metadata,
                                          ColumnDescriptor* schema,
                                          const WriterProperties* properties) {
@@ -82,9 +83,9 @@ static void BM_WriteInt64Column(::benchmark::State& state) {
       properties, schema.get(), reinterpret_cast<uint8_t*>(&thrift_metadata));
 
   while (state.KeepRunning()) {
-    InMemoryOutputStream stream;
+    auto stream = CreateOutputStream();
     std::shared_ptr<Int64Writer> writer = BuildWriter(
-        state.range(0), &stream, metadata.get(), schema.get(), properties.get());
+        state.range(0), stream, metadata.get(), schema.get(), properties.get());
     writer->WriteBatch(i8_values.length(), definition_levels.data(),
                        repetition_levels.data(), i8_values.raw_values());
     writer->Close();
@@ -118,9 +119,9 @@ BENCHMARK_TEMPLATE(BM_WriteInt64Column, Repetition::REPEATED, Compression::ZSTD)
 
 std::shared_ptr<Int64Reader> BuildReader(std::shared_ptr<Buffer>& buffer,
                                          int64_t num_values, ColumnDescriptor* schema) {
-  std::unique_ptr<InMemoryInputStream> source(new InMemoryInputStream(buffer));
+  auto source = std::make_shared<::arrow::io::BufferReader>(buffer);
   std::unique_ptr<PageReader> page_reader =
-      PageReader::Open(std::move(source), num_values, Compression::UNCOMPRESSED);
+      PageReader::Open(source, num_values, Compression::UNCOMPRESSED);
   return std::static_pointer_cast<Int64Reader>(
       ColumnReader::Make(schema, std::move(page_reader)));
 }
@@ -142,9 +143,9 @@ static void BM_ReadInt64Column(::benchmark::State& state) {
   auto metadata = ColumnChunkMetaDataBuilder::Make(
       properties, schema.get(), reinterpret_cast<uint8_t*>(&thrift_metadata));
 
-  InMemoryOutputStream stream;
+  auto stream = CreateOutputStream();
   std::shared_ptr<Int64Writer> writer = BuildWriter(
-      state.range(0), &stream, metadata.get(), schema.get(), properties.get());
+      state.range(0), stream, metadata.get(), schema.get(), properties.get());
   writer->WriteBatch(values.size(), definition_levels.data(), repetition_levels.data(),
                      values.data());
   writer->Close();

cpp/src/parquet/column_page.h

@@ -28,7 +28,6 @@
 
 #include "parquet/statistics.h"
 #include "parquet/types.h"
-#include "parquet/util/memory.h"
 
 namespace parquet {
 

cpp/src/parquet/column_reader.h

@@ -23,16 +23,10 @@
 #include <unordered_map>
 #include <utility>
 
-#include "arrow/buffer.h"
-#include "arrow/memory_pool.h"
-#include "arrow/util/bit-util.h"
-#include "arrow/util/macros.h"
-
 #include "parquet/encoding.h"
 #include "parquet/exception.h"
 #include "parquet/schema.h"
 #include "parquet/types.h"
-#include "parquet/util/memory.h"
 #include "parquet/platform.h"
 
 namespace arrow {
@@ -58,8 +52,6 @@ static constexpr uint32_t kDefaultMaxPageHeaderSize = 16 * 1024 * 1024;
 // 16 KB is the default expected page header size
 static constexpr uint32_t kDefaultPageHeaderSize = 16 * 1024;
 
-namespace BitUtil = ::arrow::BitUtil;
-
 class PARQUET_EXPORT LevelDecoder {
  public:
   LevelDecoder();

cpp/src/parquet/column_writer-test.cc

@@ -77,8 +77,9 @@ class TestPrimitiveWriter : public PrimitiveTypedTest<TestType> {
 
   void BuildReader(int64_t num_rows,
                    Compression::type compression = Compression::UNCOMPRESSED) {
-    auto buffer = sink_->GetBuffer();
-    std::unique_ptr<InMemoryInputStream> source(new InMemoryInputStream(buffer));
+    std::shared_ptr<Buffer> buffer;
+    ASSERT_OK(sink_->Finish(&buffer));
+    auto source = std::make_shared<::arrow::io::BufferReader>(buffer);
     std::unique_ptr<PageReader> page_reader =
         PageReader::Open(std::move(source), num_rows, compression);
     reader_ = std::static_pointer_cast<TypedColumnReader<TestType>>(
@@ -89,7 +90,7 @@ class TestPrimitiveWriter : public PrimitiveTypedTest<TestType> {
       int64_t output_size = SMALL_SIZE,
       const ColumnProperties& column_properties = ColumnProperties(),
       const ParquetVersion::type version = ParquetVersion::PARQUET_1_0) {
-    sink_.reset(new InMemoryOutputStream());
+    sink_ = CreateOutputStream();
     WriterProperties::Builder wp_builder;
     wp_builder.version(version);
     if (column_properties.encoding() == Encoding::PLAIN_DICTIONARY ||
@@ -105,7 +106,7 @@ class TestPrimitiveWriter : public PrimitiveTypedTest<TestType> {
 
     metadata_ = ColumnChunkMetaDataBuilder::Make(writer_properties_, this->descr_);
     std::unique_ptr<PageWriter> pager =
-        PageWriter::Open(sink_.get(), column_properties.compression(), metadata_.get());
+        PageWriter::Open(sink_, column_properties.compression(), metadata_.get());
     std::shared_ptr<ColumnWriter> writer =
         ColumnWriter::Make(metadata_.get(), std::move(pager), writer_properties_.get());
     return std::static_pointer_cast<TypedColumnWriter<TestType>>(writer);
@@ -280,7 +281,7 @@ class TestPrimitiveWriter : public PrimitiveTypedTest<TestType> {
 
  private:
   std::unique_ptr<ColumnChunkMetaDataBuilder> metadata_;
-  std::unique_ptr<InMemoryOutputStream> sink_;
+  std::shared_ptr<ArrowOutputStream> sink_;
   std::shared_ptr<WriterProperties> writer_properties_;
   std::vector<std::vector<uint8_t>> data_buffer_;
 };
@@ -669,12 +670,12 @@ TEST(TestColumnWriter, RepeatedListsUpdateSpacedBug) {
   SchemaDescriptor schema;
   schema.Init(root);
 
-  InMemoryOutputStream sink;
+  auto sink = CreateOutputStream();
   auto props = WriterProperties::Builder().build();
 
   auto metadata = ColumnChunkMetaDataBuilder::Make(props, schema.Column(0));
   std::unique_ptr<PageWriter> pager =
-      PageWriter::Open(&sink, Compression::UNCOMPRESSED, metadata.get());
+      PageWriter::Open(sink, Compression::UNCOMPRESSED, metadata.get());
   std::shared_ptr<ColumnWriter> writer =
       ColumnWriter::Make(metadata.get(), std::move(pager), props.get());
   auto typed_writer = std::static_pointer_cast<TypedColumnWriter<Int32Type>>(writer);

cpp/src/parquet/column_writer.cc

@@ -367,7 +367,6 @@ class ColumnWriterImpl {
         fallback_(false) {
     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));
     repetition_levels_rle_ =

cpp/src/parquet/encoding.cc

@@ -24,19 +24,17 @@
 #include <utility>
 #include <vector>
 
-#include "arrow/status.h"
 #include "arrow/util/bit-stream-utils.h"
 #include "arrow/util/bit-util.h"
 #include "arrow/util/hashing.h"
 #include "arrow/util/logging.h"
-#include "arrow/util/macros.h"
 #include "arrow/util/rle-encoding.h"
 #include "arrow/util/string_view.h"
 
 #include "parquet/exception.h"
+#include "parquet/platform.h"
 #include "parquet/schema.h"
 #include "parquet/types.h"
-#include "parquet/util/memory.h"
 
 namespace parquet {
 

cpp/src/parquet/encoding.h

@@ -22,14 +22,10 @@
 #include <memory>
 #include <vector>
 
-#include "arrow/buffer.h"
-#include "arrow/memory_pool.h"
 #include "arrow/util/bit-util.h"
-#include "arrow/util/macros.h"
 
 #include "parquet/exception.h"
 #include "parquet/types.h"
-#include "parquet/util/memory.h"
 #include "parquet/platform.h"
 
 namespace parquet {

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

@@ -70,9 +70,9 @@ class TestPageSerde : public ::testing::Test {
   void InitSerializedPageReader(int64_t num_rows,
                                 Compression::type codec = Compression::UNCOMPRESSED) {
     EndStream();
-    std::unique_ptr<InputStream> stream;
-    stream.reset(new InMemoryInputStream(out_buffer_));
-    page_reader_ = PageReader::Open(std::move(stream), num_rows, codec);
+
+    auto stream = std::make_shared<::arrow::io::BufferReader>(out_buffer_);
+    page_reader_ = PageReader::Open(stream, num_rows, codec);
   }
 
   void WriteDataPageHeader(int max_serialized_len = 1024, int32_t uncompressed_size = 0,

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

@@ -53,7 +53,7 @@ class TestSerialize : public PrimitiveTypedTest<TestType> {
   int rows_per_batch_;
 
   void FileSerializeTest(Compression::type codec_type) {
-    std::shared_ptr<InMemoryOutputStream> sink(new InMemoryOutputStream());
+    auto sink = CreateOutputStream();
     auto gnode = std::static_pointer_cast<GroupNode>(this->node_);
 
     WriterProperties::Builder prop_builder;
@@ -140,7 +140,7 @@ class TestSerialize : public PrimitiveTypedTest<TestType> {
   }
 
   void UnequalNumRows(int64_t max_rows, const std::vector<int64_t> rows_per_column) {
-    std::shared_ptr<InMemoryOutputStream> sink(new InMemoryOutputStream());
+    auto sink = CreateOutputStream();
     auto gnode = std::static_pointer_cast<GroupNode>(this->node_);
 
     std::shared_ptr<WriterProperties> props = WriterProperties::Builder().build();
@@ -164,7 +164,7 @@ class TestSerialize : public PrimitiveTypedTest<TestType> {
 
   void UnequalNumRowsBuffered(int64_t max_rows,
                               const std::vector<int64_t> rows_per_column) {
-    std::shared_ptr<InMemoryOutputStream> sink(new InMemoryOutputStream());
+    auto sink = CreateOutputStream();
     auto gnode = std::static_pointer_cast<GroupNode>(this->node_);
 
     std::shared_ptr<WriterProperties> props = WriterProperties::Builder().build();
@@ -193,7 +193,7 @@ class TestSerialize : public PrimitiveTypedTest<TestType> {
     const int kNumRows = 100;
     this->GenerateData(kNumRows);
 
-    std::shared_ptr<InMemoryOutputStream> sink(new InMemoryOutputStream());
+    auto sink = CreateOutputStream();
     auto gnode = std::static_pointer_cast<GroupNode>(this->node_);
     std::shared_ptr<WriterProperties> props = WriterProperties::Builder().build();
     auto file_writer = ParquetFileWriter::Open(sink, gnode, props);
@@ -227,7 +227,7 @@ class TestSerialize : public PrimitiveTypedTest<TestType> {
   }
 
   void ZeroRowsRowGroup() {
-    std::shared_ptr<InMemoryOutputStream> sink(new InMemoryOutputStream());
+    auto sink = CreateOutputStream();
     auto gnode = std::static_pointer_cast<GroupNode>(this->node_);
 
     std::shared_ptr<WriterProperties> props = WriterProperties::Builder().build();

cpp/src/parquet/file_reader.h

@@ -23,9 +23,6 @@
 #include <string>
 #include <vector>
 
-#include "arrow/io/interfaces.h"
-#include "arrow/util/macros.h"
-
 #include "parquet/metadata.h"  // IWYU pragma:: keep
 #include "parquet/properties.h"
 #include "parquet/platform.h"

cpp/src/parquet/file_writer.h

@@ -22,8 +22,6 @@
 #include <memory>
 #include <ostream>
 
-#include "arrow/util/macros.h"
-
 #include "parquet/exception.h"
 #include "parquet/metadata.h"
 #include "parquet/properties.h"

cpp/src/parquet/parquet.thrift

@@ -17,7 +17,7 @@
  * under the License.
  */
 
-cpp_include "parquet/util/windows_compatibility.h"
+cpp_include "parquet/windows_compatibility.h"
 
 /**
  * File format description for the parquet file format
@@ -647,7 +647,7 @@ struct ColumnMetaData {
   /** total byte size of all uncompressed pages in this column chunk (including the headers) **/
   6: required i64 total_uncompressed_size
 
-  /** total byte size of all compressed, and potentially encrypted, pages 
+  /** total byte size of all compressed, and potentially encrypted, pages
    *  in this column chunk (including the headers) **/
   7: required i64 total_compressed_size
 
@@ -678,7 +678,7 @@ struct EncryptionWithFooterKey {
 struct EncryptionWithColumnKey {
   /** Column path in schema **/
   1: required list<string> path_in_schema
-  
+
   /** Retrieval metadata of column encryption key **/
   2: optional binary key_metadata
 }
@@ -717,7 +717,7 @@ struct ColumnChunk {
 
   /** Crypto metadata of encrypted columns **/
   8: optional ColumnCryptoMetaData crypto_metadata
-  
+
   /** Encrypted column metadata for this chunk **/
   9: optional binary encrypted_column_metadata
 }
@@ -743,10 +743,10 @@ struct RowGroup {
    * in this row group **/
   5: optional i64 file_offset
 
-  /** Total byte size of all compressed (and potentially encrypted) column data 
+  /** Total byte size of all compressed (and potentially encrypted) column data
    *  in this row group **/
   6: optional i64 total_compressed_size
-  
+
   /** Row group ordinal in the file **/
   7: optional i16 ordinal
 }
@@ -874,7 +874,7 @@ struct AesGcmV1 {
 
   /** Unique file identifier part of AAD suffix **/
   2: optional binary aad_file_unique
-  
+
   /** In files encrypted with AAD prefix without storing it,
    * readers must supply the prefix **/
   3: optional bool supply_aad_prefix
@@ -886,7 +886,7 @@ struct AesGcmCtrV1 {
 
   /** Unique file identifier part of AAD suffix **/
   2: optional binary aad_file_unique
-  
+
   /** In files encrypted with AAD prefix without storing it,
    * readers must supply the prefix **/
   3: optional bool supply_aad_prefix
@@ -941,32 +941,30 @@ struct FileMetaData {
    */
   7: optional list<ColumnOrder> column_orders;
 
-  /** 
+  /**
    * Encryption algorithm. This field is set only in encrypted files
    * with plaintext footer. Files with encrypted footer store algorithm id
    * in FileCryptoMetaData structure.
    */
   8: optional EncryptionAlgorithm encryption_algorithm
 
-  /** 
-   * Retrieval metadata of key used for signing the footer. 
-   * Used only in encrypted files with plaintext footer. 
-   */ 
+  /**
+   * Retrieval metadata of key used for signing the footer.
+   * Used only in encrypted files with plaintext footer.
+   */
   9: optional binary footer_signing_key_metadata
 }
 
 /** Crypto metadata for files with encrypted footer **/
 struct FileCryptoMetaData {
-  /** 
+  /**
    * Encryption algorithm. This field is only used for files
    * with encrypted footer. Files with plaintext footer store algorithm id
    * inside footer (FileMetaData structure).
    */
   1: required EncryptionAlgorithm encryption_algorithm
-    
-  /** Retrieval metadata of key used for encryption of footer, 
+
+  /** Retrieval metadata of key used for encryption of footer,
    *  and (possibly) columns **/
   2: optional binary key_metadata
 }
-
-

cpp/src/parquet/platform.h

@@ -15,9 +15,13 @@
 // specific language governing permissions and limitations
 // under the License.
 
-#pragma
+#pragma once
 
-#include "arrow/util/macros.h"
+#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
 
 #if defined(_WIN32) || defined(__CYGWIN__)
 

cpp/src/parquet/properties.cc

@@ -0,0 +1,40 @@
+// 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/properties.h"
+
+#include "arrow/io/buffered.h"
+#include "arrow/io/memory.h"
+
+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);
+}
+}

cpp/src/parquet/properties.h

@@ -48,20 +48,7 @@ class PARQUET_EXPORT ReaderProperties {
   ::arrow::MemoryPool* memory_pool() const { return pool_; }
 
   std::unique_ptr<ArrowInputStream> 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);
-    }
-  }
+                                              int64_t start, int64_t num_bytes);
 
   bool is_buffered_stream_enabled() const { return buffered_stream_enabled_; }
 

cpp/src/parquet/statistics-test.cc

@@ -333,7 +333,7 @@ class TestStatistics : public PrimitiveTypedTest<TestType> {
     auto expected_stats = TypedStats::Make(this->schema_.Column(0));
     expected_stats->Update(this->values_ptr_, num_values - null_count, null_count);
 
-    auto sink = std::make_shared<InMemoryOutputStream>();
+    auto sink = CreateOutputStream();
     auto gnode = std::static_pointer_cast<GroupNode>(this->node_);
     std::shared_ptr<WriterProperties> writer_properties =
         WriterProperties::Builder().enable_statistics("column")->build();
@@ -363,7 +363,8 @@ class TestStatistics : public PrimitiveTypedTest<TestType> {
     row_group_writer->Close();
     file_writer->Close();
 
-    auto buffer = sink->GetBuffer();
+    std::shared_ptr<Buffer> buffer;
+    ASSERT_OK(sink->Finish(&buffer));
     auto source = std::make_shared<::arrow::io::BufferReader>(buffer);
     auto file_reader = ParquetFileReader::Open(source);
     auto rg_reader = file_reader->RowGroup(0);
@@ -616,7 +617,7 @@ class TestStatisticsSortOrder : public ::testing::Test {
     schema_ = std::static_pointer_cast<GroupNode>(
         GroupNode::Make("Schema", Repetition::REQUIRED, fields_));
 
-    parquet_sink_ = std::make_shared<InMemoryOutputStream>();
+    parquet_sink_ = CreateOutputStream();
   }
 
   void SetValues();
@@ -645,7 +646,8 @@ class TestStatisticsSortOrder : public ::testing::Test {
   }
 
   void VerifyParquetStats() {
-    auto pbuffer = parquet_sink_->GetBuffer();
+    std::shared_ptr<Buffer> pbuffer;
+    ASSERT_OK(parquet_sink_->Finish(&pbuffer));
 
     // Create a ParquetReader instance
     std::unique_ptr<parquet::ParquetFileReader> parquet_reader =
@@ -668,7 +670,7 @@ class TestStatisticsSortOrder : public ::testing::Test {
   std::vector<uint8_t> values_buf_;
   std::vector<schema::NodePtr> fields_;
   std::shared_ptr<schema::GroupNode> schema_;
-  std::shared_ptr<InMemoryOutputStream> parquet_sink_;
+  std::shared_ptr<ArrowOutputStream> parquet_sink_;
   std::vector<EncodedStatistics> stats_;
 };
 

cpp/src/parquet/statistics.cc

@@ -24,8 +24,8 @@
 
 #include "parquet/encoding.h"
 #include "parquet/exception.h"
+#include "parquet/platform.h"
 #include "parquet/statistics.h"
-#include "parquet/util/memory.h"
 
 using arrow::default_memory_pool;
 using arrow::MemoryPool;

cpp/src/parquet/statistics.h

@@ -25,7 +25,6 @@
 #include "parquet/platform.h"
 #include "parquet/schema.h"
 #include "parquet/types.h"
-#include "parquet/util/memory.h"
 
 namespace parquet {
 

cpp/src/parquet/test-util.h

@@ -282,7 +282,7 @@ class DataPageBuilder {
   typedef typename Type::c_type T;
 
   // This class writes data and metadata to the passed inputs
-  explicit DataPageBuilder(InMemoryOutputStream* sink)
+  explicit DataPageBuilder(ArrowOutputStream* sink)
       : sink_(sink),
         num_values_(0),
         encoding_(Encoding::PLAIN),
@@ -314,7 +314,7 @@ class DataPageBuilder {
                     Encoding::type encoding = Encoding::PLAIN) {
     std::shared_ptr<Buffer> values_sink = EncodeValues<Type>(
         encoding, false, values.data(), static_cast<int>(values.size()), d);
-    sink_->Write(values_sink->data(), values_sink->size());
+    PARQUET_THROW_NOT_OK(sink_->Write(values_sink->data(), values_sink->size()));
 
     num_values_ = std::max(static_cast<int32_t>(values.size()), num_values_);
     encoding_ = encoding;
@@ -330,7 +330,7 @@ class DataPageBuilder {
   Encoding::type def_level_encoding() const { return definition_level_encoding_; }
 
  private:
-  InMemoryOutputStream* sink_;
+  ArrowOutputStream* sink_;
 
   int32_t num_values_;
   Encoding::type encoding_;
@@ -361,8 +361,9 @@ class DataPageBuilder {
     encoder.Encode(static_cast<int>(levels.size()), levels.data());
 
     int32_t rle_bytes = encoder.len();
-    sink_->Write(reinterpret_cast<const uint8_t*>(&rle_bytes), sizeof(int32_t));
-    sink_->Write(encode_buffer.data(), rle_bytes);
+    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));
   }
 };
 
@@ -378,7 +379,7 @@ void DataPageBuilder<BooleanType>::AppendValues(const ColumnDescriptor* d,
   dynamic_cast<BooleanEncoder*>(encoder.get())
       ->Put(values, static_cast<int>(values.size()));
   std::shared_ptr<Buffer> buffer = encoder->FlushValues();
-  sink_->Write(buffer->data(), buffer->size());
+  PARQUET_THROW_NOT_OK(sink_->Write(buffer->data(), buffer->size()));
 
   num_values_ = std::max(static_cast<int32_t>(values.size()), num_values_);
   encoding_ = encoding;
@@ -393,8 +394,8 @@ static std::shared_ptr<DataPageV1> MakeDataPage(
     const std::vector<int16_t>& rep_levels, int16_t max_rep_level) {
   int num_values = 0;
 
-  InMemoryOutputStream page_stream;
-  test::DataPageBuilder<Type> page_builder(&page_stream);
+  auto page_stream = CreateOutputStream();
+  test::DataPageBuilder<Type> page_builder(page_stream.get());
 
   if (!rep_levels.empty()) {
     page_builder.AppendRepLevels(rep_levels, max_rep_level);
@@ -483,7 +484,6 @@ static std::shared_ptr<DictionaryPage> MakeDictPage(
     const ColumnDescriptor* d, const std::vector<typename Type::c_type>& values,
     const std::vector<int>& values_per_page, Encoding::type encoding,
     std::vector<std::shared_ptr<Buffer>>& rle_indices) {
-  InMemoryOutputStream page_stream;
   test::DictionaryPageBuilder<Type> page_builder(d);
   int num_pages = static_cast<int>(values_per_page.size());
   int value_start = 0;

cpp/src/parquet/thrift.h

@@ -42,8 +42,8 @@
 
 #include "arrow/util/logging.h"
 #include "parquet/exception.h"
+#include "parquet/platform.h"
 #include "parquet/statistics.h"
-#include "parquet/util/memory.h"
 
 #include "parquet/parquet_types.h"  // IYWU pragma: export
 
@@ -186,11 +186,11 @@ class ThriftSerializer {
   }
 
   template <class T>
-  int64_t Serialize(const T* obj, OutputStream* out) {
+  int64_t Serialize(const T* obj, ArrowOutputStream* out) {
     uint8_t* out_buffer;
     uint32_t out_length;
     SerializeToBuffer(obj, &out_length, &out_buffer);
-    out->Write(out_buffer, out_length);
+    PARQUET_THROW_NOT_OK(out->Write(out_buffer, out_length));
     return static_cast<int64_t>(out_length);
   }
 

cpp/src/parquet/types.h

@@ -26,9 +26,16 @@
 #include <sstream>
 #include <string>
 
-#include "arrow/util/macros.h"
 #include "parquet/platform.h"
 
+namespace arrow {
+namespace util {
+
+class Codec;
+
+}  // namespace util
+}  // namespace arrow
+
 namespace parquet {
 
 // ----------------------------------------------------------------------
@@ -429,14 +436,6 @@ struct Compression {
   enum type { UNCOMPRESSED, SNAPPY, GZIP, LZO, BROTLI, LZ4, ZSTD };
 };
 
-namespace arrow {
-namespace util {
-
-class Codec;
-
-}  // namespace util
-}  // namespace arrow
-
 PARQUET_EXPORT
 std::unique_ptr<::arrow::util::Codec> GetCodecFromArrow(Compression::type codec);
 

cpp/src/parquet/util/CMakeLists.txt

@@ -17,5 +17,3 @@
 
 # Headers: util
 arrow_install_all_headers("parquet/util")
-
-add_parquet_test(memory-test)