Refactoring, address code review comments. fix flake8 issues

Change-Id: I9ced59de48f169b6609dd27f8239ceb22fd5ebeb
NonVolatileComputing · Aug 19, 2017 · a9522c5 · a9522c5
1 parent ce5784d
commit a9522c5
Show file tree

Hide file tree

Showing 8 changed files with 238 additions and 192 deletions.
diff --git a/cpp/src/arrow/builder.h b/cpp/src/arrow/builder.h
@@ -244,7 +244,7 @@ class ARROW_EXPORT NumericBuilder : public PrimitiveBuilder<T> {
   using PrimitiveBuilder<T>::Reserve;
 
   /// Append a single scalar and increase the size if necessary.
-  Status Append(value_type val) {
+  Status Append(const value_type val) {
     RETURN_NOT_OK(ArrayBuilder::Reserve(1));
     UnsafeAppend(val);
     return Status::OK();
@@ -255,7 +255,7 @@ class ARROW_EXPORT NumericBuilder : public PrimitiveBuilder<T> {
   ///
   /// This method does not capacity-check; make sure to call Reserve
   /// beforehand.
-  void UnsafeAppend(value_type val) {
+  void UnsafeAppend(const value_type val) {
     BitUtil::SetBit(null_bitmap_data_, length_);
     raw_data_[length_++] = val;
   }
@@ -371,7 +371,7 @@ class ARROW_EXPORT AdaptiveUIntBuilder : public internal::AdaptiveIntBuilderBase
   using ArrayBuilder::Advance;
 
   /// Scalar append
-  Status Append(uint64_t val) {
+  Status Append(const uint64_t val) {
     RETURN_NOT_OK(Reserve(1));
     BitUtil::SetBit(null_bitmap_data_, length_);
 
@@ -430,7 +430,7 @@ class ARROW_EXPORT AdaptiveIntBuilder : public internal::AdaptiveIntBuilderBase
   using ArrayBuilder::Advance;
 
   /// Scalar append
-  Status Append(int64_t val) {
+  Status Append(const int64_t val) {
     RETURN_NOT_OK(Reserve(1));
     BitUtil::SetBit(null_bitmap_data_, length_);
 
@@ -511,7 +511,7 @@ class ARROW_EXPORT BooleanBuilder : public ArrayBuilder {
   std::shared_ptr<Buffer> data() const { return data_; }
 
   /// Scalar append
-  Status Append(bool val) {
+  Status Append(const bool val) {
     RETURN_NOT_OK(Reserve(1));
     BitUtil::SetBit(null_bitmap_data_, length_);
     if (val) {
@@ -523,7 +523,7 @@ class ARROW_EXPORT BooleanBuilder : public ArrayBuilder {
     return Status::OK();
   }
 
-  Status Append(uint8_t val) { return Append(val != 0); }
+  Status Append(const uint8_t val) { return Append(val != 0); }
 
   /// Vector append
   ///

diff --git a/cpp/src/arrow/python/arrow_to_python.cc b/cpp/src/arrow/python/arrow_to_python.cc
@@ -189,9 +189,8 @@ Status DeserializeTuple(std::shared_ptr<Array> array, int64_t start_idx, int64_t
   DESERIALIZE_SEQUENCE(PyTuple_New, PyTuple_SET_ITEM)
 }
 
-Status ReadSerializedPythonSequence(std::shared_ptr<io::RandomAccessFile> src,
-                                    std::shared_ptr<RecordBatch>* batch_out,
-                                    std::vector<std::shared_ptr<Tensor>>* tensors_out) {
+Status ReadSerializedObject(std::shared_ptr<io::RandomAccessFile> src,
+                            SerializedPyObject* out) {
   std::shared_ptr<ipc::RecordBatchStreamReader> reader;
   int64_t offset;
   int64_t bytes_read;
@@ -200,23 +199,22 @@ Status ReadSerializedPythonSequence(std::shared_ptr<io::RandomAccessFile> src,
   RETURN_NOT_OK(
       src->Read(sizeof(int32_t), &bytes_read, reinterpret_cast<uint8_t*>(&num_tensors)));
   RETURN_NOT_OK(ipc::RecordBatchStreamReader::Open(src, &reader));
-  RETURN_NOT_OK(reader->ReadNextRecordBatch(batch_out));
+  RETURN_NOT_OK(reader->ReadNextRecordBatch(&out->batch));
   RETURN_NOT_OK(src->Tell(&offset));
   offset += 4;  // Skip the end-of-stream message
   for (int i = 0; i < num_tensors; ++i) {
     std::shared_ptr<Tensor> tensor;
     RETURN_NOT_OK(ipc::ReadTensor(offset, src.get(), &tensor));
-    tensors_out->push_back(tensor);
+    out->tensors.push_back(tensor);
     RETURN_NOT_OK(src->Tell(&offset));
   }
   return Status::OK();
 }
 
-Status DeserializePythonSequence(std::shared_ptr<RecordBatch> batch,
-                                 std::vector<std::shared_ptr<Tensor>> tensors,
-                                 PyObject* base, PyObject** out) {
+Status DeserializeObject(const SerializedPyObject& obj, PyObject* base, PyObject** out) {
   PyAcquireGIL lock;
-  return DeserializeList(batch->column(0), 0, batch->num_rows(), base, tensors, out);
+  return DeserializeList(obj.batch->column(0), 0, obj.batch->num_rows(), base,
+                         obj.tensors, out);
 }
 
 }  // namespace py

diff --git a/cpp/src/arrow/python/arrow_to_python.h b/cpp/src/arrow/python/arrow_to_python.h
@@ -24,7 +24,9 @@
 #include <memory>
 #include <vector>
 
+#include "arrow/python/python_to_arrow.h"
 #include "arrow/status.h"
+#include "arrow/util/visibility.h"
 
 namespace arrow {
 
@@ -39,14 +41,24 @@ class RandomAccessFile;
 
 namespace py {
 
-Status ReadSerializedPythonSequence(std::shared_ptr<io::RandomAccessFile> src,
-                                    std::shared_ptr<RecordBatch>* batch_out,
-                                    std::vector<std::shared_ptr<Tensor>>* tensors_out);
+/// \brief Read serialized Python sequence from file interface using Arrow IPC
+/// \param[in] src a RandomAccessFile
+/// \param[out] out the reconstructed data
+/// \return Status
+ARROW_EXPORT
+Status ReadSerializedObject(std::shared_ptr<io::RandomAccessFile> src,
+                            SerializedPyObject* out);
 
-// This acquires the GIL
-Status DeserializePythonSequence(std::shared_ptr<RecordBatch> batch,
-                                 std::vector<std::shared_ptr<Tensor>> tensors,
-                                 PyObject* base, PyObject** out);
+/// \brief Reconstruct Python object from Arrow-serialized representation
+/// \param[in] object
+/// \param[in] base a Python object holding the underlying data that any NumPy
+/// arrays will reference, to avoid premature deallocation
+/// \param[out] out the returned object
+/// \return Status
+/// This acquires the GIL
+ARROW_EXPORT
+Status DeserializeObject(const SerializedPyObject& object, PyObject* base,
+                         PyObject** out);
 
 }  // namespace py
 }  // namespace arrow