ARROW-203: Python: Basic filename based Parquet read/write

cpp/src/arrow/column.h

@@ -67,6 +67,8 @@ class Column {
 
   int64_t null_count() const { return data_->null_count(); }
 
+  const std::shared_ptr<Field>& field() const { return field_; }
+
   // @returns: the column's name in the passed metadata
   const std::string& name() const { return field_->name; }
 

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

@@ -69,11 +69,26 @@ std::shared_ptr<PrimitiveArray> NullableArray(
   return std::static_pointer_cast<PrimitiveArray>(builder.Finish());
 }
 
+std::shared_ptr<Column> MakeColumn(const std::string& name,
+    const std::shared_ptr<PrimitiveArray>& array, bool nullable) {
+  auto field = std::make_shared<Field>(name, array->type(), nullable);
+  return std::make_shared<Column>(field, array);
+}
+
+std::shared_ptr<Table> MakeSimpleTable(
+    const std::shared_ptr<PrimitiveArray>& values, bool nullable) {
+  std::shared_ptr<Column> column = MakeColumn("col", values, nullable);
+  std::vector<std::shared_ptr<Column>> columns({column});
+  std::vector<std::shared_ptr<Field>> fields({column->field()});
+  auto schema = std::make_shared<Schema>(fields);
+  return std::make_shared<Table>("table", schema, columns);
+}
+
 class TestParquetIO : public ::testing::Test {
  public:
   virtual void SetUp() {}
 
-  std::shared_ptr<GroupNode> Schema(
+  std::shared_ptr<GroupNode> MakeSchema(
       ParquetType::type parquet_type, Repetition::type repetition) {
     auto pnode = PrimitiveNode::Make("column1", repetition, parquet_type);
     NodePtr node_ =
@@ -102,9 +117,17 @@ class TestParquetIO : public ::testing::Test {
     ASSERT_NE(nullptr, out->get());
   }
 
+  void ReadTableFromFile(
+      std::unique_ptr<ParquetFileReader> file_reader, std::shared_ptr<Table>* out) {
+    arrow::parquet::FileReader reader(default_memory_pool(), std::move(file_reader));
+    ASSERT_NO_THROW(ASSERT_OK(reader.ReadFlatTable(out)));
+    ASSERT_NE(nullptr, out->get());
+  }
+
   std::unique_ptr<ParquetFileReader> Int64File(
       std::vector<int64_t>& values, int num_chunks) {
-    std::shared_ptr<GroupNode> schema = Schema(ParquetType::INT64, Repetition::REQUIRED);
+    std::shared_ptr<GroupNode> schema =
+        MakeSchema(ParquetType::INT64, Repetition::REQUIRED);
     std::unique_ptr<ParquetFileWriter> file_writer = MakeWriter(schema);
     size_t chunk_size = values.size() / num_chunks;
     for (int i = 0; i < num_chunks; i++) {
@@ -120,7 +143,6 @@ class TestParquetIO : public ::testing::Test {
     return ReaderFromSink();
   }
 
- private:
   std::shared_ptr<InMemoryOutputStream> sink_;
 };
 
@@ -137,6 +159,23 @@ TEST_F(TestParquetIO, SingleColumnInt64Read) {
   }
 }
 
+TEST_F(TestParquetIO, SingleColumnInt64TableRead) {
+  std::vector<int64_t> values(100, 128);
+  std::unique_ptr<ParquetFileReader> file_reader = Int64File(values, 1);
+
+  std::shared_ptr<Table> out;
+  ReadTableFromFile(std::move(file_reader), &out);
+  ASSERT_EQ(1, out->num_columns());
+  ASSERT_EQ(100, out->num_rows());
+
+  std::shared_ptr<ChunkedArray> chunked_array = out->column(0)->data();
+  ASSERT_EQ(1, chunked_array->num_chunks());
+  Int64Array* out_array = static_cast<Int64Array*>(chunked_array->chunk(0).get());
+  for (size_t i = 0; i < values.size(); i++) {
+    EXPECT_EQ(values[i], out_array->raw_data()[i]);
+  }
+}
+
 TEST_F(TestParquetIO, SingleColumnInt64ChunkedRead) {
   std::vector<int64_t> values(100, 128);
   std::unique_ptr<ParquetFileReader> file_reader = Int64File(values, 4);
@@ -150,10 +189,28 @@ TEST_F(TestParquetIO, SingleColumnInt64ChunkedRead) {
   }
 }
 
+TEST_F(TestParquetIO, SingleColumnInt64ChunkedTableRead) {
+  std::vector<int64_t> values(100, 128);
+  std::unique_ptr<ParquetFileReader> file_reader = Int64File(values, 4);
+
+  std::shared_ptr<Table> out;
+  ReadTableFromFile(std::move(file_reader), &out);
+  ASSERT_EQ(1, out->num_columns());
+  ASSERT_EQ(100, out->num_rows());
+
+  std::shared_ptr<ChunkedArray> chunked_array = out->column(0)->data();
+  ASSERT_EQ(1, chunked_array->num_chunks());
+  Int64Array* out_array = static_cast<Int64Array*>(chunked_array->chunk(0).get());
+  for (size_t i = 0; i < values.size(); i++) {
+    EXPECT_EQ(values[i], out_array->raw_data()[i]);
+  }
+}
+
 TEST_F(TestParquetIO, SingleColumnInt64Write) {
   std::shared_ptr<PrimitiveArray> values = NonNullArray<Int64Type>(100, 128);
 
-  std::shared_ptr<GroupNode> schema = Schema(ParquetType::INT64, Repetition::REQUIRED);
+  std::shared_ptr<GroupNode> schema =
+      MakeSchema(ParquetType::INT64, Repetition::REQUIRED);
   FileWriter writer(default_memory_pool(), MakeWriter(schema));
   ASSERT_NO_THROW(ASSERT_OK(writer.NewRowGroup(values->length())));
   ASSERT_NO_THROW(ASSERT_OK(writer.WriteFlatColumnChunk(values.get())));
@@ -164,11 +221,29 @@ TEST_F(TestParquetIO, SingleColumnInt64Write) {
   ASSERT_TRUE(values->Equals(out));
 }
 
+TEST_F(TestParquetIO, SingleColumnTableInt64Write) {
+  std::shared_ptr<PrimitiveArray> values = NonNullArray<Int64Type>(100, 128);
+  std::shared_ptr<Table> table = MakeSimpleTable(values, false);
+  sink_ = std::make_shared<InMemoryOutputStream>();
+  ASSERT_NO_THROW(ASSERT_OK(
+      WriteFlatTable(table.get(), default_memory_pool(), sink_, values->length())));
+
+  std::shared_ptr<Table> out;
+  ReadTableFromFile(ReaderFromSink(), &out);
+  ASSERT_EQ(1, out->num_columns());
+  ASSERT_EQ(100, out->num_rows());
+
+  std::shared_ptr<ChunkedArray> chunked_array = out->column(0)->data();
+  ASSERT_EQ(1, chunked_array->num_chunks());
+  ASSERT_TRUE(values->Equals(chunked_array->chunk(0)));
+}
+
 TEST_F(TestParquetIO, SingleColumnDoubleReadWrite) {
   // This also tests max_definition_level = 1
   std::shared_ptr<PrimitiveArray> values = NullableArray<DoubleType>(100, 128, 10);
 
-  std::shared_ptr<GroupNode> schema = Schema(ParquetType::DOUBLE, Repetition::OPTIONAL);
+  std::shared_ptr<GroupNode> schema =
+      MakeSchema(ParquetType::DOUBLE, Repetition::OPTIONAL);
   FileWriter writer(default_memory_pool(), MakeWriter(schema));
   ASSERT_NO_THROW(ASSERT_OK(writer.NewRowGroup(values->length())));
   ASSERT_NO_THROW(ASSERT_OK(writer.WriteFlatColumnChunk(values.get())));
@@ -179,11 +254,30 @@ TEST_F(TestParquetIO, SingleColumnDoubleReadWrite) {
   ASSERT_TRUE(values->Equals(out));
 }
 
+TEST_F(TestParquetIO, SingleColumnTableDoubleReadWrite) {
+  // This also tests max_definition_level = 1
+  std::shared_ptr<PrimitiveArray> values = NullableArray<DoubleType>(100, 128, 10);
+  std::shared_ptr<Table> table = MakeSimpleTable(values, true);
+  sink_ = std::make_shared<InMemoryOutputStream>();
+  ASSERT_NO_THROW(ASSERT_OK(
+      WriteFlatTable(table.get(), default_memory_pool(), sink_, values->length())));
+
+  std::shared_ptr<Table> out;
+  ReadTableFromFile(ReaderFromSink(), &out);
+  ASSERT_EQ(1, out->num_columns());
+  ASSERT_EQ(100, out->num_rows());
+
+  std::shared_ptr<ChunkedArray> chunked_array = out->column(0)->data();
+  ASSERT_EQ(1, chunked_array->num_chunks());
+  ASSERT_TRUE(values->Equals(chunked_array->chunk(0)));
+}
+
 TEST_F(TestParquetIO, SingleColumnInt64ChunkedWrite) {
   std::shared_ptr<PrimitiveArray> values = NonNullArray<Int64Type>(100, 128);
   std::shared_ptr<PrimitiveArray> values_chunk = NonNullArray<Int64Type>(25, 128);
 
-  std::shared_ptr<GroupNode> schema = Schema(ParquetType::INT64, Repetition::REQUIRED);
+  std::shared_ptr<GroupNode> schema =
+      MakeSchema(ParquetType::INT64, Repetition::REQUIRED);
   FileWriter writer(default_memory_pool(), MakeWriter(schema));
   for (int i = 0; i < 4; i++) {
     ASSERT_NO_THROW(ASSERT_OK(writer.NewRowGroup(values_chunk->length())));
@@ -196,13 +290,31 @@ TEST_F(TestParquetIO, SingleColumnInt64ChunkedWrite) {
   ASSERT_TRUE(values->Equals(out));
 }
 
+TEST_F(TestParquetIO, SingleColumnTableInt64ChunkedWrite) {
+  std::shared_ptr<PrimitiveArray> values = NonNullArray<Int64Type>(1000, 128);
+  std::shared_ptr<Table> table = MakeSimpleTable(values, false);
+  sink_ = std::make_shared<InMemoryOutputStream>();
+  ASSERT_NO_THROW(
+      ASSERT_OK(WriteFlatTable(table.get(), default_memory_pool(), sink_, 512)));
+
+  std::shared_ptr<Table> out;
+  ReadTableFromFile(ReaderFromSink(), &out);
+  ASSERT_EQ(1, out->num_columns());
+  ASSERT_EQ(1000, out->num_rows());
+
+  std::shared_ptr<ChunkedArray> chunked_array = out->column(0)->data();
+  ASSERT_EQ(1, chunked_array->num_chunks());
+  ASSERT_TRUE(values->Equals(chunked_array->chunk(0)));
+}
+
 TEST_F(TestParquetIO, SingleColumnDoubleChunkedWrite) {
   std::shared_ptr<PrimitiveArray> values = NullableArray<DoubleType>(100, 128, 10);
   std::shared_ptr<PrimitiveArray> values_chunk_nulls =
       NullableArray<DoubleType>(25, 128, 10);
   std::shared_ptr<PrimitiveArray> values_chunk = NullableArray<DoubleType>(25, 128, 0);
 
-  std::shared_ptr<GroupNode> schema = Schema(ParquetType::DOUBLE, Repetition::OPTIONAL);
+  std::shared_ptr<GroupNode> schema =
+      MakeSchema(ParquetType::DOUBLE, Repetition::OPTIONAL);
   FileWriter writer(default_memory_pool(), MakeWriter(schema));
   ASSERT_NO_THROW(ASSERT_OK(writer.NewRowGroup(values_chunk_nulls->length())));
   ASSERT_NO_THROW(ASSERT_OK(writer.WriteFlatColumnChunk(values_chunk_nulls.get())));
@@ -217,6 +329,24 @@ TEST_F(TestParquetIO, SingleColumnDoubleChunkedWrite) {
   ASSERT_TRUE(values->Equals(out));
 }
 
+TEST_F(TestParquetIO, SingleColumnTableDoubleChunkedWrite) {
+  // This also tests max_definition_level = 1
+  std::shared_ptr<PrimitiveArray> values = NullableArray<DoubleType>(1000, 128, 100);
+  std::shared_ptr<Table> table = MakeSimpleTable(values, true);
+  sink_ = std::make_shared<InMemoryOutputStream>();
+  ASSERT_NO_THROW(
+      ASSERT_OK(WriteFlatTable(table.get(), default_memory_pool(), sink_, 512)));
+
+  std::shared_ptr<Table> out;
+  ReadTableFromFile(ReaderFromSink(), &out);
+  ASSERT_EQ(1, out->num_columns());
+  ASSERT_EQ(1000, out->num_rows());
+
+  std::shared_ptr<ChunkedArray> chunked_array = out->column(0)->data();
+  ASSERT_EQ(1, chunked_array->num_chunks());
+  ASSERT_TRUE(values->Equals(chunked_array->chunk(0)));
+}
+
 }  // namespace parquet
 
 }  // namespace arrow

cpp/src/arrow/parquet/reader.cc

@@ -18,10 +18,14 @@
 #include "arrow/parquet/reader.h"
 
 #include <queue>
+#include <string>
+#include <vector>
 
+#include "arrow/column.h"
 #include "arrow/parquet/schema.h"
 #include "arrow/parquet/utils.h"
 #include "arrow/schema.h"
+#include "arrow/table.h"
 #include "arrow/types/primitive.h"
 #include "arrow/util/status.h"
 
@@ -40,6 +44,7 @@ class FileReader::Impl {
   bool CheckForFlatColumn(const ::parquet::ColumnDescriptor* descr);
   Status GetFlatColumn(int i, std::unique_ptr<FlatColumnReader>* out);
   Status ReadFlatColumn(int i, std::shared_ptr<Array>* out);
+  Status ReadFlatTable(std::shared_ptr<Table>* out);
 
  private:
   MemoryPool* pool_;
@@ -103,6 +108,23 @@ Status FileReader::Impl::ReadFlatColumn(int i, std::shared_ptr<Array>* out) {
   return flat_column_reader->NextBatch(reader_->num_rows(), out);
 }
 
+Status FileReader::Impl::ReadFlatTable(std::shared_ptr<Table>* table) {
+  const std::string name = reader_->descr()->schema()->name();
+  std::shared_ptr<Schema> schema;
+  RETURN_NOT_OK(FromParquetSchema(reader_->descr(), &schema));
+
+  std::vector<std::shared_ptr<Column>> columns;
+  for (int i = 0; i < reader_->num_columns(); i++) {
+    std::shared_ptr<Array> array;
+    RETURN_NOT_OK(ReadFlatColumn(i, &array));
+    auto column = std::make_shared<Column>(schema->field(i), array);
+    columns.push_back(column);
+  }
+
+  *table = std::make_shared<Table>(name, schema, columns);
+  return Status::OK();
+}
+
 FileReader::FileReader(
     MemoryPool* pool, std::unique_ptr<::parquet::ParquetFileReader> reader)
     : impl_(new FileReader::Impl(pool, std::move(reader))) {}
@@ -117,6 +139,10 @@ Status FileReader::ReadFlatColumn(int i, std::shared_ptr<Array>* out) {
   return impl_->ReadFlatColumn(i, out);
 }
 
+Status FileReader::ReadFlatTable(std::shared_ptr<Table>* out) {
+  return impl_->ReadFlatTable(out);
+}
+
 FlatColumnReader::Impl::Impl(MemoryPool* pool, const ::parquet::ColumnDescriptor* descr,
     ::parquet::ParquetFileReader* reader, int column_index)
     : pool_(pool),

cpp/src/arrow/parquet/reader.h

@@ -29,6 +29,7 @@ class Array;
 class MemoryPool;
 class RowBatch;
 class Status;
+class Table;
 
 namespace parquet {
 
@@ -90,6 +91,8 @@ class FileReader {
   Status GetFlatColumn(int i, std::unique_ptr<FlatColumnReader>* out);
   // Read column as a whole into an Array.
   Status ReadFlatColumn(int i, std::shared_ptr<Array>* out);
+  // Read a table of flat columns into a Table.
+  Status ReadFlatTable(std::shared_ptr<Table>* out);
 
   virtual ~FileReader();
 

cpp/src/arrow/parquet/writer.cc

@@ -17,11 +17,21 @@
 
 #include "arrow/parquet/writer.h"
 
+#include <algorithm>
+#include <vector>
+
 #include "arrow/array.h"
+#include "arrow/column.h"
+#include "arrow/table.h"
+#include "arrow/types/construct.h"
 #include "arrow/types/primitive.h"
+#include "arrow/parquet/schema.h"
 #include "arrow/parquet/utils.h"
 #include "arrow/util/status.h"
 
+using parquet::ParquetFileWriter;
+using parquet::schema::GroupNode;
+
 namespace arrow {
 
 namespace parquet {
@@ -143,6 +153,82 @@ Status FileWriter::Close() {
 
 FileWriter::~FileWriter() {}
 
+// Create a slice of a PrimitiveArray.
+//
+// This method is specially crafted for WriteFlatTable and assumes the following:
+//  * chunk_size is a multiple of 512
+Status TemporaryArraySlice(int64_t chunk, int64_t chunk_size, PrimitiveArray* array,
+    std::shared_ptr<PrimitiveArray>* out) {
+  // The last chunk may be smaller than the chunk_size
+  int64_t size = std::min(chunk_size, array->length() - chunk * chunk_size);
+  int64_t buffer_offset = chunk * chunk_size * array->type()->value_size();
+  int64_t value_size = size * array->type()->value_size();
+  auto chunk_buffer = std::make_shared<Buffer>(array->data(), buffer_offset, value_size);
+  std::shared_ptr<Buffer> null_bitmap;
+  int32_t null_count = 0;
+  if (array->null_count() > 0) {
+    int64_t null_offset = (chunk * chunk_size) / 8;
+    int64_t null_size = util::ceil_byte(size) / 8;
+    null_bitmap = std::make_shared<Buffer>(array->null_bitmap(), null_offset, null_size);
+    for (int64_t k = 0; k < size; k++) {
+      if (!util::get_bit(null_bitmap->data(), k)) { null_count++; }
+    }
+  }
+  std::shared_ptr<Array> out_array;
+  RETURN_NOT_OK(MakePrimitiveArray(
+      array->type(), size, chunk_buffer, null_count, null_bitmap, &out_array));
+  *out = std::static_pointer_cast<PrimitiveArray>(out_array);
+  return Status::OK();
+}
+
+Status WriteFlatTable(const Table* table, MemoryPool* pool,
+    std::shared_ptr<::parquet::OutputStream> sink, int64_t chunk_size) {
+  // Ensure alignment of sliced PrimitiveArray, esp. the null bitmap
+  // TODO: Support other chunksizes than multiples of 512
+  if (((chunk_size & 511) != 0) && (chunk_size != table->num_rows())) {
+    return Status::NotImplemented(
+        "Only chunk sizes that are a multiple of 512 are supported");
+  }
+
+  std::shared_ptr<::parquet::SchemaDescriptor> parquet_schema;
+  RETURN_NOT_OK(ToParquetSchema(table->schema().get(), &parquet_schema));
+  auto schema_node = std::static_pointer_cast<GroupNode>(parquet_schema->schema());
+  std::unique_ptr<ParquetFileWriter> parquet_writer =
+      ParquetFileWriter::Open(sink, schema_node);
+  FileWriter writer(pool, std::move(parquet_writer));
+
+  // TODO: Support writing chunked arrays.
+  for (int i = 0; i < table->num_columns(); i++) {
+    if (table->column(i)->data()->num_chunks() != 1) {
+      return Status::NotImplemented("No support for writing chunked arrays yet.");
+    }
+  }
+
+  // Cast to PrimitiveArray instances as we work with them.
+  std::vector<std::shared_ptr<PrimitiveArray>> arrays(table->num_columns());
+  for (int i = 0; i < table->num_columns(); i++) {
+    // num_chunks == 1 as per above loop
+    std::shared_ptr<Array> array = table->column(i)->data()->chunk(0);
+    auto primitive_array = std::dynamic_pointer_cast<PrimitiveArray>(array);
+    if (!primitive_array) {
+      return Status::NotImplemented("Table must consist of PrimitiveArray instances");
+    }
+    arrays[i] = primitive_array;
+  }
+
+  for (int chunk = 0; chunk * chunk_size < table->num_rows(); chunk++) {
+    int64_t size = std::min(chunk_size, table->num_rows() - chunk * chunk_size);
+    RETURN_NOT_OK(writer.NewRowGroup(size));
+    for (int i = 0; i < table->num_columns(); i++) {
+      std::shared_ptr<PrimitiveArray> array;
+      RETURN_NOT_OK(TemporaryArraySlice(chunk, chunk_size, arrays[i].get(), &array));
+      RETURN_NOT_OK(writer.WriteFlatColumnChunk(array.get()));
+    }
+  }
+
+  return writer.Close();
+}
+
 }  // namespace parquet
 
 }  // namespace arrow

cpp/src/arrow/parquet/writer.h

@@ -29,6 +29,7 @@ class MemoryPool;
 class PrimitiveArray;
 class RowBatch;
 class Status;
+class Table;
 
 namespace parquet {
 
@@ -52,6 +53,14 @@ class FileWriter {
   std::unique_ptr<Impl> impl_;
 };
 
+/**
+ * Write a flat Table to Parquet.
+ *
+ * The table shall only consist of nullable, non-repeated columns of primitive type.
+ */
+Status WriteFlatTable(const Table* table, MemoryPool* pool,
+    std::shared_ptr<::parquet::OutputStream> sink, int64_t chunk_size);
+
 }  // namespace parquet
 
 }  // namespace arrow

python/CMakeLists.txt

@@ -344,6 +344,8 @@ find_package(Arrow REQUIRED)
 include_directories(SYSTEM ${ARROW_INCLUDE_DIR})
 ADD_THIRDPARTY_LIB(arrow
   SHARED_LIB ${ARROW_SHARED_LIB})
+ADD_THIRDPARTY_LIB(arrow_parquet
+    SHARED_LIB ${ARROW_PARQUET_SHARED_LIB})
 
 ############################################################
 # Linker setup
@@ -414,6 +416,7 @@ set(PYARROW_SRCS
   src/pyarrow/common.cc
   src/pyarrow/config.cc
   src/pyarrow/helpers.cc
+  src/pyarrow/parquet.cc
   src/pyarrow/status.cc
 
   src/pyarrow/adapters/builtin.cc
@@ -422,6 +425,7 @@ set(PYARROW_SRCS
 
 set(LINK_LIBS
   arrow
+  arrow_parquet
 )
 
 SET(CMAKE_INSTALL_RPATH_USE_LINK_PATH TRUE)
@@ -442,6 +446,7 @@ set(CYTHON_EXTENSIONS
   array
   config
   error
+  parquet
   scalar
   schema
   table

python/cmake_modules/FindArrow.cmake

@@ -42,19 +42,27 @@ find_library(ARROW_LIB_PATH NAMES arrow
   ${ARROW_SEARCH_LIB_PATH}
   NO_DEFAULT_PATH)
 
-if (ARROW_INCLUDE_DIR AND ARROW_LIB_PATH)
+find_library(ARROW_PARQUET_LIB_PATH NAMES arrow_parquet
+  PATHS
+  ${ARROW_SEARCH_LIB_PATH}
+  NO_DEFAULT_PATH)
+
+if (ARROW_INCLUDE_DIR AND ARROW_LIB_PATH AND ARROW_PARQUET_LIB_PATH)
   set(ARROW_FOUND TRUE)
   set(ARROW_LIB_NAME libarrow)
+  set(ARROW_PARQUET_LIB_NAME libarrow_parquet)
   set(ARROW_LIBS ${ARROW_SEARCH_LIB_PATH})
   set(ARROW_STATIC_LIB ${ARROW_SEARCH_LIB_PATH}/${ARROW_LIB_NAME}.a)
   set(ARROW_SHARED_LIB ${ARROW_LIBS}/${ARROW_LIB_NAME}${CMAKE_SHARED_LIBRARY_SUFFIX})
+  set(ARROW_PARQUET_STATIC_LIB ${ARROW_SEARCH_LIB_PATH}/${ARROW_PARQUET_LIB_NAME}.a)
+  set(ARROW_PARQUET_SHARED_LIB ${ARROW_LIBS}/${ARROW_PARQUET_LIB_NAME}${CMAKE_SHARED_LIBRARY_SUFFIX})
 else ()
   set(ARROW_FOUND FALSE)
 endif ()
 
 if (ARROW_FOUND)
   if (NOT Arrow_FIND_QUIETLY)
-    message(STATUS "Found the Arrow library: ${ARROW_LIB_PATH}")
+    message(STATUS "Found the Arrow library: ${ARROW_LIB_PATH}, ${ARROW_PARQUET_LIB_PATH}")
   endif ()
 else ()
   if (NOT Arrow_FIND_QUIETLY)
@@ -74,4 +82,6 @@ mark_as_advanced(
   ARROW_LIBS
   ARROW_STATIC_LIB
   ARROW_SHARED_LIB
+  ARROW_PARQUET_STATIC_LIB
+  ARROW_PARQUET_SHARED_LIB
 )

python/pyarrow/array.pyx

@@ -68,6 +68,14 @@ cdef class Array:
         values = array_format(self, window=10)
         return '{0}\n{1}'.format(type_format, values)
 
+    def __richcmp__(Array self, Array other, int op):
+        if op == cpython.Py_EQ:
+            return self.ap.Equals(other.sp_array)
+        elif op == cpython.Py_NE:
+            return not self.ap.Equals(other.sp_array)
+        else:
+            raise TypeError('Invalid comparison')
+
     def __len__(self):
         if self.sp_array.get():
             return self.sp_array.get().length()

python/pyarrow/error.pxd

@@ -15,6 +15,8 @@
 # specific language governing permissions and limitations
 # under the License.
 
+from pyarrow.includes.libarrow cimport CStatus
 from pyarrow.includes.pyarrow cimport *
 
+cdef check_cstatus(const CStatus& status)
 cdef check_status(const Status& status)

python/pyarrow/error.pyx

@@ -21,6 +21,13 @@ from pyarrow.compat import frombytes
 class ArrowException(Exception):
     pass
 
+cdef check_cstatus(const CStatus& status):
+    if status.ok():
+        return
+
+    cdef c_string c_message = status.ToString()
+    raise ArrowException(frombytes(c_message))
+
 cdef check_status(const Status& status):
     if status.ok():
         return

python/pyarrow/includes/common.pxd

@@ -34,6 +34,11 @@ cdef extern from "<Python.h>":
 
 cdef extern from "<memory>" namespace "std" nogil:
 
+    cdef cppclass unique_ptr[T]:
+        unique_ptr()
+        unique_ptr(T*)
+        T* get()
+
     cdef cppclass shared_ptr[T]:
         shared_ptr()
         shared_ptr(T*)

python/pyarrow/includes/libarrow.pxd

@@ -72,6 +72,8 @@ cdef extern from "arrow/api.h" namespace "arrow" nogil:
     cdef cppclass MemoryPool" arrow::MemoryPool":
         int64_t bytes_allocated()
 
+    cdef MemoryPool* default_memory_pool()
+
     cdef cppclass CListType" arrow::ListType"(CDataType):
         CListType(const shared_ptr[CDataType]& value_type)
 
@@ -103,6 +105,7 @@ cdef extern from "arrow/api.h" namespace "arrow" nogil:
         int32_t null_count()
         Type type_enum()
 
+        c_bool Equals(const shared_ptr[CArray]& arr)
         c_bool IsNull(int i)
 
     cdef cppclass CBooleanArray" arrow::BooleanArray"(CArray):

python/pyarrow/includes/parquet.pxd

@@ -18,6 +18,26 @@
 # distutils: language = c++
 
 from pyarrow.includes.common cimport *
+from pyarrow.includes.libarrow cimport CSchema, CStatus, CTable, MemoryPool
+
+
+cdef extern from "parquet/api/schema.h" namespace "parquet::schema" nogil:
+  cdef cppclass Node:
+    pass
+
+  cdef cppclass GroupNode(Node):
+    pass
+
+  cdef cppclass PrimitiveNode(Node):
+    pass
+
+cdef extern from "parquet/api/schema.h" namespace "parquet" nogil:
+  cdef cppclass SchemaDescriptor:
+    shared_ptr[Node] schema()
+    GroupNode* group()
+
+  cdef cppclass ColumnDescriptor:
+    pass
 
 cdef extern from "parquet/api/reader.h" namespace "parquet" nogil:
     cdef cppclass ColumnReader:
@@ -48,4 +68,30 @@ cdef extern from "parquet/api/reader.h" namespace "parquet" nogil:
         pass
 
     cdef cppclass ParquetFileReader:
+        # TODO: Some default arguments are missing
+        @staticmethod
+        unique_ptr[ParquetFileReader] OpenFile(const c_string& path)
+
+cdef extern from "parquet/api/writer.h" namespace "parquet" nogil:
+    cdef cppclass OutputStream:
         pass
+
+    cdef cppclass LocalFileOutputStream(OutputStream):
+        LocalFileOutputStream(const c_string& path)
+        void Close()
+
+
+cdef extern from "arrow/parquet/reader.h" namespace "arrow::parquet" nogil:
+    cdef cppclass FileReader:
+        FileReader(MemoryPool* pool, unique_ptr[ParquetFileReader] reader)
+        CStatus ReadFlatTable(shared_ptr[CTable]* out);
+
+
+cdef extern from "arrow/parquet/schema.h" namespace "arrow::parquet" nogil:
+    CStatus FromParquetSchema(const SchemaDescriptor* parquet_schema, shared_ptr[CSchema]* out)
+    CStatus ToParquetSchema(const CSchema* arrow_schema, shared_ptr[SchemaDescriptor]* out)
+
+
+cdef extern from "arrow/parquet/writer.h" namespace "arrow::parquet" nogil:
+    cdef CStatus WriteFlatTable(const CTable* table, MemoryPool* pool, shared_ptr[OutputStream] sink, int64_t chunk_size)
+

python/pyarrow/parquet.pyx

@@ -19,5 +19,44 @@
 # distutils: language = c++
 # cython: embedsignature = True
 
-from pyarrow.compat import frombytes, tobytes
+from pyarrow.includes.libarrow cimport *
+cimport pyarrow.includes.pyarrow as pyarrow
 from pyarrow.includes.parquet cimport *
+
+from pyarrow.error cimport check_cstatus
+from pyarrow.table cimport Table
+
+def read_table(filename, columns=None):
+    """
+    Read a Table from Parquet format
+    Returns
+    -------
+    table: pyarrow.Table
+    """
+    cdef unique_ptr[FileReader] reader
+    cdef Table table = Table()
+    cdef shared_ptr[CTable] ctable
+
+    # Must be in one expression to avoid calling std::move which is not possible
+    # in Cython (due to missing rvalue support)
+    reader = unique_ptr[FileReader](new FileReader(default_memory_pool(),
+        ParquetFileReader.OpenFile(filename)))
+    check_cstatus(reader.get().ReadFlatTable(&ctable))
+    table.init(ctable)
+    return table
+
+def write_table(table, filename, chunk_size=None):
+    """
+    Write a Table to Parquet format
+    """
+    cdef Table table_ = table
+    cdef CTable* ctable_ = table_.table
+    cdef shared_ptr[OutputStream] sink = shared_ptr[OutputStream](new LocalFileOutputStream(filename))
+    cdef int64_t chunk_size_ = 0
+    if chunk_size is None:
+        chunk_size_ = ctable_.num_rows()
+    else:
+        chunk_size_ = chunk_size
+
+    check_cstatus(WriteFlatTable(ctable_, default_memory_pool(), sink, chunk_size_))
+

python/pyarrow/tests/test_parquet.py

@@ -0,0 +1,50 @@
+# 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.
+
+from pyarrow.compat import unittest
+import pyarrow as arrow
+import pyarrow.parquet
+
+A = arrow
+
+from shutil import rmtree
+from tempfile import mkdtemp
+
+import os.path
+
+
+class TestParquetIO(unittest.TestCase):
+
+  def setUp(self):
+    self.temp_directory = mkdtemp()
+
+  def tearDown(self):
+    rmtree(self.temp_directory)
+
+
+  def test_single_int64_column(self):
+    filename = os.path.join(self.temp_directory, 'single_int64_column.parquet')
+    data = [A.from_pylist(range(5))]
+    table = A.Table.from_arrays(('a', 'b'), data, 'table_name')
+    A.parquet.write_table(table, filename)
+    table_read = pyarrow.parquet.read_table(filename)
+    for col_written, col_read in zip(table.itercolumns(), table_read.itercolumns()):
+        assert col_written.name == col_read.name
+        assert col_read.data.num_chunks == 1
+        data_written = col_written.data.chunk(0)
+        data_read = col_read.data.chunk(0)
+        assert data_written == data_read

python/setup.py

@@ -214,7 +214,7 @@ class build_ext(_build_ext):
             return name + suffix
 
     def get_cmake_cython_names(self):
-        return ['array', 'config', 'error', 'scalar', 'schema', 'table']
+        return ['array', 'config', 'error', 'parquet', 'scalar', 'schema', 'table']
 
     def get_names(self):
         return self._found_names

python/src/pyarrow/parquet.cc

@@ -0,0 +1,30 @@
+// 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 "pyarrow/parquet.h"
+
+#include <arrow/parquet/reader.h>
+#include <arrow/parquet/schema.h>
+#include <arrow/parquet/writer.h>
+
+using namespace arrow;
+
+namespace pyarrow {
+
+
+} // namespace pyarrow
+

python/src/pyarrow/parquet.h

@@ -0,0 +1,30 @@
+// 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.
+
+#ifndef PYARROW_PARQUET_H
+#define PYARROW_PARQUET_H
+
+#include <arrow/parquet/reader.h>
+#include <arrow/parquet/writer.h>
+
+namespace pyarrow {
+
+
+} // namespace pyarrow
+
+#endif // PYARROW_PARQUET_H
+