Barely working direct HDFS-Parquet reads

Change-Id: I3f6329e159431df781959d6266b4e016e4f6fa2c

cpp/src/arrow/parquet/io.cc

@@ -55,9 +55,17 @@ void ParquetAllocator::Free(uint8_t* buffer, int64_t size) {
 // ----------------------------------------------------------------------
 // ParquetReadSource
 
-ParquetReadSource::ParquetReadSource(
-    const std::shared_ptr<ArrowROFile>& file, ParquetAllocator* allocator)
-    : file_(file), allocator_(allocator) {}
+ParquetReadSource::ParquetReadSource(ParquetAllocator* allocator)
+    : file_(nullptr), allocator_(allocator) {}
+
+Status ParquetReadSource::Open(const std::shared_ptr<io::RandomAccessFile>& file) {
+  int64_t file_size;
+  RETURN_NOT_OK(file->GetSize(&file_size));
+
+  file_ = file;
+  size_ = file_size;
+  return Status::OK();
+}
 
 void ParquetReadSource::Close() {
   PARQUET_THROW_NOT_OK(file_->Close());

cpp/src/arrow/parquet/io.h

@@ -59,8 +59,10 @@ 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(ParquetAllocator* allocator);
+
+  // We need to ask for the file size on opening the file, and this can fail
+  Status Open(const std::shared_ptr<io::RandomAccessFile>& file);
 
   void Close() override;
   int64_t Tell() const override;

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

@@ -23,6 +23,7 @@
 #include "gtest/gtest.h"
 
 #include "arrow/parquet/io.h"
+#include "arrow/test-util.h"
 #include "arrow/util/memory-pool.h"
 #include "arrow/util/status.h"
 
@@ -150,7 +151,9 @@ TEST(TestParquetReadSource, Basics) {
   ParquetAllocator allocator(default_memory_pool());
 
   auto file = std::make_shared<BufferReader>(data_buffer, data.size());
-  auto source = std::make_shared<ParquetReadSource>(file, &allocator);
+  auto source = std::make_shared<ParquetReadSource>(&allocator);
+
+  ASSERT_OK(source->Open(file));
 
   ASSERT_EQ(0, source->Tell());
   ASSERT_NO_THROW(source->Seek(5));

cpp/src/arrow/parquet/reader.cc

@@ -189,7 +189,8 @@ FileReader::~FileReader() {}
 // Static ctor
 Status OpenFile(const std::shared_ptr<io::RandomAccessFile>& file,
     ParquetAllocator* allocator, std::unique_ptr<FileReader>* reader) {
-  std::unique_ptr<ParquetRAS> source(new ParquetReadSource(file, allocator));
+  std::unique_ptr<ParquetReadSource> source(new ParquetReadSource(allocator));
+  RETURN_NOT_OK(source->Open(file));
 
   // TODO(wesm): reader properties
   std::unique_ptr<ParquetReader> pq_reader;

python/pyarrow/includes/parquet.pxd

@@ -99,8 +99,8 @@ cdef extern from "arrow/parquet/io.h" namespace "arrow::parquet" nogil:
         void set_pool(MemoryPool* pool)
 
     cdef cppclass ParquetReadSource:
-        ParquetReadSource(const shared_ptr[RandomAccessFile]& file,
-                          ParquetAllocator* allocator)
+        ParquetReadSource(ParquetAllocator* allocator)
+        Open(const shared_ptr[RandomAccessFile]& file)
 
 
 cdef extern from "arrow/parquet/reader.h" namespace "arrow::parquet" nogil: