ARROW-605: [C++] Refactor IPC adapter code into generic ArrayLoader class. Add Date32Type

cpp/CMakeLists.txt

@@ -799,6 +799,7 @@ set(ARROW_SRCS
   src/arrow/builder.cc
   src/arrow/column.cc
   src/arrow/compare.cc
+  src/arrow/loader.cc
   src/arrow/memory_pool.cc
   src/arrow/pretty_print.cc
   src/arrow/schema.cc

cpp/src/arrow/CMakeLists.txt

@@ -19,10 +19,11 @@
 install(FILES
   api.h
   array.h
-  column.h
-  compare.h
   buffer.h
   builder.h
+  column.h
+  compare.h
+  loader.h
   memory_pool.h
   pretty_print.h
   schema.h

cpp/src/arrow/array.cc

@@ -165,6 +165,7 @@ template class NumericArray<Int32Type>;
 template class NumericArray<Int64Type>;
 template class NumericArray<TimestampType>;
 template class NumericArray<DateType>;
+template class NumericArray<Date32Type>;
 template class NumericArray<TimeType>;
 template class NumericArray<HalfFloatType>;
 template class NumericArray<FloatType>;
@@ -193,7 +194,7 @@ std::shared_ptr<Array> BooleanArray::Slice(int64_t offset, int64_t length) const
 
 Status ListArray::Validate() const {
   if (length_ < 0) { return Status::Invalid("Length was negative"); }
-  if (!value_offsets_) { return Status::Invalid("value_offsets_ was null"); }
+  if (length_ && !value_offsets_) { return Status::Invalid("value_offsets_ was null"); }
   if (value_offsets_->size() / static_cast<int>(sizeof(int32_t)) < length_) {
     std::stringstream ss;
     ss << "offset buffer size (bytes): " << value_offsets_->size()
@@ -425,20 +426,6 @@ std::shared_ptr<Array> UnionArray::Slice(int64_t offset, int64_t length) const {
 // ----------------------------------------------------------------------
 // DictionaryArray
 
-Status DictionaryArray::FromBuffer(const std::shared_ptr<DataType>& type, int64_t length,
-    const std::shared_ptr<Buffer>& indices, const std::shared_ptr<Buffer>& null_bitmap,
-    int64_t null_count, int64_t offset, std::shared_ptr<DictionaryArray>* out) {
-  DCHECK_EQ(type->type, Type::DICTIONARY);
-  const auto& dict_type = static_cast<const DictionaryType*>(type.get());
-
-  std::shared_ptr<Array> boxed_indices;
-  RETURN_NOT_OK(MakePrimitiveArray(dict_type->index_type(), length, indices, null_bitmap,
-      null_count, offset, &boxed_indices));
-
-  *out = std::make_shared<DictionaryArray>(type, boxed_indices);
-  return Status::OK();
-}
-
 DictionaryArray::DictionaryArray(
     const std::shared_ptr<DataType>& type, const std::shared_ptr<Array>& indices)
     : Array(type, indices->length(), indices->null_bitmap(), indices->null_count(),
@@ -469,40 +456,6 @@ std::shared_ptr<Array> DictionaryArray::Slice(int64_t offset, int64_t length) co
   return std::make_shared<DictionaryArray>(type_, sliced_indices);
 }
 
-// ----------------------------------------------------------------------
-
-#define MAKE_PRIMITIVE_ARRAY_CASE(ENUM, ArrayType)                                  \
-  case Type::ENUM:                                                                  \
-    out->reset(new ArrayType(type, length, data, null_bitmap, null_count, offset)); \
-    break;
-
-Status MakePrimitiveArray(const std::shared_ptr<DataType>& type, int64_t length,
-    const std::shared_ptr<Buffer>& data, const std::shared_ptr<Buffer>& null_bitmap,
-    int64_t null_count, int64_t offset, std::shared_ptr<Array>* out) {
-  switch (type->type) {
-    MAKE_PRIMITIVE_ARRAY_CASE(BOOL, BooleanArray);
-    MAKE_PRIMITIVE_ARRAY_CASE(UINT8, UInt8Array);
-    MAKE_PRIMITIVE_ARRAY_CASE(INT8, Int8Array);
-    MAKE_PRIMITIVE_ARRAY_CASE(UINT16, UInt16Array);
-    MAKE_PRIMITIVE_ARRAY_CASE(INT16, Int16Array);
-    MAKE_PRIMITIVE_ARRAY_CASE(UINT32, UInt32Array);
-    MAKE_PRIMITIVE_ARRAY_CASE(INT32, Int32Array);
-    MAKE_PRIMITIVE_ARRAY_CASE(UINT64, UInt64Array);
-    MAKE_PRIMITIVE_ARRAY_CASE(INT64, Int64Array);
-    MAKE_PRIMITIVE_ARRAY_CASE(FLOAT, FloatArray);
-    MAKE_PRIMITIVE_ARRAY_CASE(DOUBLE, DoubleArray);
-    MAKE_PRIMITIVE_ARRAY_CASE(TIME, Int64Array);
-    MAKE_PRIMITIVE_ARRAY_CASE(TIMESTAMP, TimestampArray);
-    default:
-      return Status::NotImplemented(type->ToString());
-  }
-#ifdef NDEBUG
-  return Status::OK();
-#else
-  return (*out)->Validate();
-#endif
-}
-
 // ----------------------------------------------------------------------
 // Default implementations of ArrayVisitor methods
 
@@ -527,6 +480,7 @@ ARRAY_VISITOR_DEFAULT(DoubleArray);
 ARRAY_VISITOR_DEFAULT(StringArray);
 ARRAY_VISITOR_DEFAULT(BinaryArray);
 ARRAY_VISITOR_DEFAULT(DateArray);
+ARRAY_VISITOR_DEFAULT(Date32Array);
 ARRAY_VISITOR_DEFAULT(TimeArray);
 ARRAY_VISITOR_DEFAULT(TimestampArray);
 ARRAY_VISITOR_DEFAULT(IntervalArray);

cpp/src/arrow/array.h

@@ -58,6 +58,7 @@ class ARROW_EXPORT ArrayVisitor {
   virtual Status Visit(const StringArray& array);
   virtual Status Visit(const BinaryArray& array);
   virtual Status Visit(const DateArray& array);
+  virtual Status Visit(const Date32Array& array);
   virtual Status Visit(const TimeArray& array);
   virtual Status Visit(const TimestampArray& array);
   virtual Status Visit(const IntervalArray& array);
@@ -485,12 +486,6 @@ class ARROW_EXPORT DictionaryArray : public Array {
   DictionaryArray(
       const std::shared_ptr<DataType>& type, const std::shared_ptr<Array>& indices);
 
-  // Alternate ctor; other attributes (like null count) are inherited from the
-  // passed indices array
-  static Status FromBuffer(const std::shared_ptr<DataType>& type, int64_t length,
-      const std::shared_ptr<Buffer>& indices, const std::shared_ptr<Buffer>& null_bitmap,
-      int64_t null_count, int64_t offset, std::shared_ptr<DictionaryArray>* out);
-
   Status Validate() const override;
 
   std::shared_ptr<Array> indices() const { return indices_; }
@@ -531,21 +526,13 @@ extern template class ARROW_EXPORT NumericArray<FloatType>;
 extern template class ARROW_EXPORT NumericArray<DoubleType>;
 extern template class ARROW_EXPORT NumericArray<TimestampType>;
 extern template class ARROW_EXPORT NumericArray<DateType>;
+extern template class ARROW_EXPORT NumericArray<Date32Type>;
 extern template class ARROW_EXPORT NumericArray<TimeType>;
 
 #if defined(__GNUC__) && !defined(__clang__)
 #pragma GCC diagnostic pop
 #endif
 
-// ----------------------------------------------------------------------
-// Helper functions
-
-// Create new arrays for logical types that are backed by primitive arrays.
-Status ARROW_EXPORT MakePrimitiveArray(const std::shared_ptr<DataType>& type,
-    int64_t length, const std::shared_ptr<Buffer>& data,
-    const std::shared_ptr<Buffer>& null_bitmap, int64_t null_count, int64_t offset,
-    std::shared_ptr<Array>* out);
-
 }  // namespace arrow
 
 #endif

cpp/src/arrow/builder.cc

@@ -238,6 +238,7 @@ template class PrimitiveBuilder<Int16Type>;
 template class PrimitiveBuilder<Int32Type>;
 template class PrimitiveBuilder<Int64Type>;
 template class PrimitiveBuilder<DateType>;
+template class PrimitiveBuilder<Date32Type>;
 template class PrimitiveBuilder<TimestampType>;
 template class PrimitiveBuilder<TimeType>;
 template class PrimitiveBuilder<HalfFloatType>;

cpp/src/arrow/builder.h

@@ -233,6 +233,7 @@ using Int64Builder = NumericBuilder<Int64Type>;
 using TimestampBuilder = NumericBuilder<TimestampType>;
 using TimeBuilder = NumericBuilder<TimeType>;
 using DateBuilder = NumericBuilder<DateType>;
+using Date32Builder = NumericBuilder<Date32Type>;
 
 using HalfFloatBuilder = NumericBuilder<HalfFloatType>;
 using FloatBuilder = NumericBuilder<FloatType>;

cpp/src/arrow/column.cc

@@ -97,6 +97,9 @@ Column::Column(const std::shared_ptr<Field>& field, const std::shared_ptr<Array>
   }
 }
 
+Column::Column(const std::string& name, const std::shared_ptr<Array>& data)
+    : Column(::arrow::field(name, data->type()), data) {}
+
 Column::Column(
     const std::shared_ptr<Field>& field, const std::shared_ptr<ChunkedArray>& data)
     : field_(field), data_(data) {}

cpp/src/arrow/column.h

@@ -69,6 +69,9 @@ class ARROW_EXPORT Column {
 
   Column(const std::shared_ptr<Field>& field, const std::shared_ptr<Array>& data);
 
+  /// Construct from name and array
+  Column(const std::string& name, const std::shared_ptr<Array>& data);
+
   int64_t length() const { return data_->length(); }
 
   int64_t null_count() const { return data_->null_count(); }

cpp/src/arrow/compare.cc

@@ -145,6 +145,10 @@ class RangeEqualsVisitor : public ArrayVisitor {
 
   Status Visit(const DateArray& left) override { return CompareValues<DateArray>(left); }
 
+  Status Visit(const Date32Array& left) override {
+    return CompareValues<Date32Array>(left);
+  }
+
   Status Visit(const TimeArray& left) override { return CompareValues<TimeArray>(left); }
 
   Status Visit(const TimestampArray& left) override {
@@ -381,6 +385,8 @@ class ArrayEqualsVisitor : public RangeEqualsVisitor {
 
   Status Visit(const DateArray& left) override { return ComparePrimitive(left); }
 
+  Status Visit(const Date32Array& left) override { return ComparePrimitive(left); }
+
   Status Visit(const TimeArray& left) override { return ComparePrimitive(left); }
 
   Status Visit(const TimestampArray& left) override { return ComparePrimitive(left); }
@@ -622,7 +628,7 @@ class TypeEqualsVisitor : public TypeVisitor {
 
   Status Visit(const TimestampType& left) override {
     const auto& right = static_cast<const TimestampType&>(right_);
-    result_ = left.unit == right.unit;
+    result_ = left.unit == right.unit && left.timezone == right.timezone;
     return Status::OK();
   }
 

cpp/src/arrow/io/memory.cc

@@ -28,6 +28,7 @@
 #include "arrow/buffer.h"
 #include "arrow/io/interfaces.h"
 #include "arrow/status.h"
+#include "arrow/util/logging.h"
 
 namespace arrow {
 namespace io {
@@ -43,9 +44,17 @@ BufferOutputStream::BufferOutputStream(const std::shared_ptr<ResizableBuffer>& b
       position_(0),
       mutable_data_(buffer->mutable_data()) {}
 
+Status BufferOutputStream::Create(int64_t initial_capacity, MemoryPool* pool,
+    std::shared_ptr<BufferOutputStream>* out) {
+  std::shared_ptr<ResizableBuffer> buffer;
+  RETURN_NOT_OK(AllocateResizableBuffer(pool, initial_capacity, &buffer));
+  *out = std::make_shared<BufferOutputStream>(buffer);
+  return Status::OK();
+}
+
 BufferOutputStream::~BufferOutputStream() {
   // This can fail, better to explicitly call close
-  Close();
+  if (buffer_) { Close(); }
 }
 
 Status BufferOutputStream::Close() {
@@ -56,12 +65,20 @@ Status BufferOutputStream::Close() {
   }
 }
 
+Status BufferOutputStream::Finish(std::shared_ptr<Buffer>* result) {
+  RETURN_NOT_OK(Close());
+  *result = buffer_;
+  buffer_ = nullptr;
+  return Status::OK();
+}
+
 Status BufferOutputStream::Tell(int64_t* position) {
   *position = position_;
   return Status::OK();
 }
 
 Status BufferOutputStream::Write(const uint8_t* data, int64_t nbytes) {
+  DCHECK(buffer_);
   RETURN_NOT_OK(Reserve(nbytes));
   std::memcpy(mutable_data_ + position_, data, nbytes);
   position_ += nbytes;

cpp/src/arrow/io/memory.h

@@ -43,13 +43,19 @@ class ARROW_EXPORT BufferOutputStream : public OutputStream {
  public:
   explicit BufferOutputStream(const std::shared_ptr<ResizableBuffer>& buffer);
 
+  static Status Create(int64_t initial_capacity, MemoryPool* pool,
+      std::shared_ptr<BufferOutputStream>* out);
+
   ~BufferOutputStream();
 
   // Implement the OutputStream interface
   Status Close() override;
   Status Tell(int64_t* position) override;
   Status Write(const uint8_t* data, int64_t nbytes) override;
 
+  /// Close the stream and return the buffer
+  Status Finish(std::shared_ptr<Buffer>* result);
+
  private:
   // Ensures there is sufficient space available to write nbytes
   Status Reserve(int64_t nbytes);