Implement alternate ctor to construct parquet::FileReader from an arrow::io::RandomAccessFile

Change-Id: I7982556541a60ca03b3064a333b207fd45e323c3

cpp/src/arrow/parquet/io.cc

@@ -56,8 +56,10 @@ void ParquetAllocator::Free(uint8_t* buffer, int64_t size) {
 // ParquetReadSource
 
 ParquetReadSource::ParquetReadSource(
-    const std::shared_ptr<ArrowROFile>& file, ParquetAllocator* allocator)
-    : file_(file), allocator_(allocator) {}
+    const std::shared_ptr<ArrowROFile>& file, MemoryPool* pool)
+    : file_(file) {
+  allocator_.set_pool(pool);
+}
 
 void ParquetReadSource::Close() {
   PARQUET_THROW_NOT_OK(file_->Close());
@@ -82,7 +84,7 @@ int64_t ParquetReadSource::Read(int64_t nbytes, uint8_t* out) {
 std::shared_ptr<::parquet::Buffer> ParquetReadSource::Read(int64_t nbytes) {
   // TODO(wesm): This code is duplicated from parquet/util/input.cc; suggests
   // that there should be more code sharing amongst file-like sources
-  auto result = std::make_shared<::parquet::OwnedMutableBuffer>(0, allocator_);
+  auto result = std::make_shared<::parquet::OwnedMutableBuffer>(0, &allocator_);
   result->Resize(nbytes);
 
   int64_t bytes_read = Read(nbytes, result->mutable_data());

cpp/src/arrow/parquet/io.h

@@ -49,6 +49,8 @@ class ARROW_EXPORT ParquetAllocator : public ::parquet::MemoryAllocator {
   uint8_t* Malloc(int64_t size) override;
   void Free(uint8_t* buffer, int64_t size) override;
 
+  void set_pool(MemoryPool* pool) { pool_ = pool; }
+
   MemoryPool* pool() { return pool_; }
 
  private:
@@ -57,8 +59,7 @@ class ARROW_EXPORT ParquetAllocator : public ::parquet::MemoryAllocator {
 
 class ARROW_EXPORT ParquetReadSource : public ::parquet::RandomAccessSource {
  public:
-  ParquetReadSource(
-      const std::shared_ptr<io::RandomAccessFile>& file, ParquetAllocator* allocator);
+  ParquetReadSource(const std::shared_ptr<io::RandomAccessFile>& file, MemoryPool* pool);
 
   void Close() override;
   int64_t Tell() const override;
@@ -71,7 +72,7 @@ class ARROW_EXPORT ParquetReadSource : public ::parquet::RandomAccessSource {
   std::shared_ptr<io::RandomAccessFile> file_;
 
   // The allocator is required for creating managed buffers
-  ParquetAllocator* allocator_;
+  ParquetAllocator allocator_;
 };
 
 }  // namespace parquet

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

@@ -147,9 +147,8 @@ TEST(TestParquetReadSource, Basics) {
   std::string data = "this is the data";
   auto data_buffer = reinterpret_cast<const uint8_t*>(data.c_str());
 
-  ParquetAllocator allocator;
   auto file = std::make_shared<BufferReader>(data_buffer, data.size());
-  auto source = std::make_shared<ParquetReadSource>(file, &allocator);
+  auto source = std::make_shared<ParquetReadSource>(file, default_memory_pool());
 
   ASSERT_EQ(0, source->Tell());
   ASSERT_NO_THROW(source->Seek(5));

cpp/src/arrow/parquet/reader.cc

@@ -23,6 +23,7 @@
 #include <vector>
 
 #include "arrow/column.h"
+#include "arrow/parquet/io.h"
 #include "arrow/parquet/schema.h"
 #include "arrow/parquet/utils.h"
 #include "arrow/schema.h"
@@ -35,6 +36,10 @@ using parquet::ColumnReader;
 using parquet::Repetition;
 using parquet::TypedColumnReader;
 
+// Help reduce verbosity
+using ParquetRAS = parquet::RandomAccessSource;
+using ParquetReader = parquet::ParquetFileReader;
+
 namespace arrow {
 namespace parquet {
 
@@ -181,6 +186,18 @@ FileReader::FileReader(
 
 FileReader::~FileReader() {}
 
+// Static ctor
+Status Open(MemoryPool* pool, const std::shared_ptr<io::RandomAccessFile>& file,
+    std::unique_ptr<FileReader>* reader) {
+  std::unique_ptr<ParquetRAS> source(new ParquetReadSource(file, pool));
+
+  // TODO(wesm): reader properties
+  std::unique_ptr<ParquetReader> pq_reader;
+  PARQUET_CATCH_NOT_OK(pq_reader = ParquetReader::Open(std::move(source)));
+  reader->reset(new FileReader(pool, std::move(pq_reader)));
+  return Status::OK();
+}
+
 Status FileReader::GetFlatColumn(int i, std::unique_ptr<FlatColumnReader>* out) {
   return impl_->GetFlatColumn(i, out);
 }

cpp/src/arrow/parquet/reader.h

@@ -23,6 +23,8 @@
 #include "parquet/api/reader.h"
 #include "parquet/api/schema.h"
 
+#include "arrow/io/interfaces.h"
+#include "arrow/parquet/io.h"
 #include "arrow/util/visibility.h"
 
 namespace arrow {
@@ -81,6 +83,11 @@ class FlatColumnReader;
 // arrays
 class ARROW_EXPORT FileReader {
  public:
+  // Helper function to create a file reader from an implementation of an Arrow
+  // readable file
+  static Status Open(MemoryPool* pool, const std::shared_ptr<io::RandomAccessFile>& file,
+      std::unique_ptr<FileReader>* reader);
+
   FileReader(MemoryPool* pool, std::unique_ptr<::parquet::ParquetFileReader> reader);
 
   // Since the distribution of columns amongst a Parquet file's row groups may
@@ -99,7 +106,7 @@ class ARROW_EXPORT FileReader {
   virtual ~FileReader();
 
  private:
-  class Impl;
+  class PARQUET_NO_EXPORT Impl;
   std::unique_ptr<Impl> impl_;
 };
 

cpp/src/arrow/parquet/writer.cc

@@ -35,7 +35,6 @@ using parquet::ParquetVersion;
 using parquet::schema::GroupNode;
 
 namespace arrow {
-
 namespace parquet {
 
 class FileWriter::Impl {

cpp/src/arrow/parquet/writer.h

@@ -55,7 +55,7 @@ class ARROW_EXPORT FileWriter {
   MemoryPool* memory_pool() const;
 
  private:
-  class Impl;
+  class PARQUET_NO_EXPORT Impl;
   std::unique_ptr<Impl> impl_;
 };