ARROW-82: Initial IPC support for ListArray

cpp/CMakeLists.txt

@@ -562,10 +562,10 @@ endif()
 if (${CLANG_TIDY_FOUND})
   # runs clang-tidy and attempts to fix any warning automatically
   add_custom_target(clang-tidy ${BUILD_SUPPORT_DIR}/run-clang-tidy.sh ${CLANG_TIDY_BIN} ${CMAKE_BINARY_DIR}/compile_commands.json 1
-  `find ${CMAKE_CURRENT_SOURCE_DIR}/src -name \\*.cc | sed -e '/_generated/g'`)
+  `find ${CMAKE_CURRENT_SOURCE_DIR}/src -name \\*.cc |grep -v ipc-adapter-test.cc | sed -e '/_generated/g'`)
   # runs clang-tidy and exits with a non-zero exit code if any errors are found.
   add_custom_target(check-clang-tidy ${BUILD_SUPPORT_DIR}/run-clang-tidy.sh ${CLANG_TIDY_BIN} ${CMAKE_BINARY_DIR}/compile_commands.json 
-  0 `find ${CMAKE_CURRENT_SOURCE_DIR}/src -name \\*.cc | sed -e '/_generated/g'`)
+  0 `find ${CMAKE_CURRENT_SOURCE_DIR}/src -name \\*.cc |grep -v ipc-adapter-test.cc | sed -e '/_generated/g'`)
 
 endif()
 

cpp/src/arrow/array.cc

@@ -20,6 +20,7 @@
 #include <cstdint>
 
 #include "arrow/util/buffer.h"
+#include "arrow/util/status.h"
 
 namespace arrow {
 
@@ -47,6 +48,10 @@ bool Array::EqualsExact(const Array& other) const {
   return true;
 }
 
+Status Array::Validate() const {
+  return Status::OK();
+}
+
 bool NullArray::Equals(const std::shared_ptr<Array>& arr) const {
   if (this == arr.get()) { return true; }
   if (Type::NA != arr->type_enum()) { return false; }

cpp/src/arrow/array.h

@@ -28,6 +28,7 @@
 namespace arrow {
 
 class Buffer;
+class Status;
 
 // Immutable data array with some logical type and some length. Any memory is
 // owned by the respective Buffer instance (or its parents).
@@ -39,7 +40,7 @@ class Array {
   Array(const std::shared_ptr<DataType>& type, int32_t length, int32_t null_count = 0,
       const std::shared_ptr<Buffer>& null_bitmap = nullptr);
 
-  virtual ~Array() {}
+  virtual ~Array() = default;
 
   // Determine if a slot is null. For inner loops. Does *not* boundscheck
   bool IsNull(int i) const {
@@ -58,6 +59,9 @@ class Array {
 
   bool EqualsExact(const Array& arr) const;
   virtual bool Equals(const std::shared_ptr<Array>& arr) const = 0;
+  // Determines if the array is internally consistent.  Defaults to always
+  // returning Status::OK.  This can be an expensive check.
+  virtual Status Validate() const;
 
  protected:
   std::shared_ptr<DataType> type_;

cpp/src/arrow/builder.cc

@@ -25,6 +25,25 @@
 
 namespace arrow {
 
+Status ArrayBuilder::AppendToBitmap(bool is_null) {
+  if (length_ == capacity_) {
+    // If the capacity was not already a multiple of 2, do so here
+    // TODO(emkornfield) doubling isn't great default allocation practice
+    // see https://github.com/facebook/folly/blob/master/folly/docs/FBVector.md
+    // fo discussion
+    RETURN_NOT_OK(Resize(util::next_power2(capacity_ + 1)));
+  }
+  UnsafeAppendToBitmap(is_null);
+  return Status::OK();
+}
+
+Status ArrayBuilder::AppendToBitmap(const uint8_t* valid_bytes, int32_t length) {
+  Reserve(length);
+
+  UnsafeAppendToBitmap(valid_bytes, length);
+  return Status::OK();
+}
+
 Status ArrayBuilder::Init(int32_t capacity) {
   capacity_ = capacity;
   int32_t to_alloc = util::ceil_byte(capacity) / 8;
@@ -36,6 +55,7 @@ Status ArrayBuilder::Init(int32_t capacity) {
 }
 
 Status ArrayBuilder::Resize(int32_t new_bits) {
+  if (!null_bitmap_) { return Init(new_bits); }
   int32_t new_bytes = util::ceil_byte(new_bits) / 8;
   int32_t old_bytes = null_bitmap_->size();
   RETURN_NOT_OK(null_bitmap_->Resize(new_bytes));
@@ -56,10 +76,46 @@ Status ArrayBuilder::Advance(int32_t elements) {
 
 Status ArrayBuilder::Reserve(int32_t elements) {
   if (length_ + elements > capacity_) {
+    // TODO(emkornfield) power of 2 growth is potentially suboptimal
     int32_t new_capacity = util::next_power2(length_ + elements);
     return Resize(new_capacity);
   }
   return Status::OK();
 }
 
+Status ArrayBuilder::SetNotNull(int32_t length) {
+  RETURN_NOT_OK(Reserve(length));
+  UnsafeSetNotNull(length);
+  return Status::OK();
+}
+
+void ArrayBuilder::UnsafeAppendToBitmap(bool is_null) {
+  if (is_null) {
+    ++null_count_;
+  } else {
+    util::set_bit(null_bitmap_data_, length_);
+  }
+  ++length_;
+}
+
+void ArrayBuilder::UnsafeAppendToBitmap(const uint8_t* valid_bytes, int32_t length) {
+  if (valid_bytes == nullptr) {
+    UnsafeSetNotNull(length);
+    return;
+  }
+  for (int32_t i = 0; i < length; ++i) {
+    // TODO(emkornfield) Optimize for large values of length?
+    AppendToBitmap(valid_bytes[i] == 0);
+  }
+}
+
+void ArrayBuilder::UnsafeSetNotNull(int32_t length) {
+  const int32_t new_length = length + length_;
+  // TODO(emkornfield) Optimize for large values of length?
+  for (int32_t i = length_; i < new_length; ++i) {
+    util::set_bit(null_bitmap_data_, i);
+  }
+  length_ = new_length;
+}
+
 }  // namespace arrow

cpp/src/arrow/builder.h

@@ -34,7 +34,10 @@ class PoolBuffer;
 
 static constexpr int32_t MIN_BUILDER_CAPACITY = 1 << 5;
 
-// Base class for all data array builders
+// Base class for all data array builders.
+// This class provides a facilities for incrementally building the null bitmap
+// (see Append methods) and as a side effect the current number of slots and
+// the null count.
 class ArrayBuilder {
  public:
   explicit ArrayBuilder(MemoryPool* pool, const TypePtr& type)
@@ -46,7 +49,7 @@ class ArrayBuilder {
         length_(0),
         capacity_(0) {}
 
-  virtual ~ArrayBuilder() {}
+  virtual ~ArrayBuilder() = default;
 
   // For nested types. Since the objects are owned by this class instance, we
   // skip shared pointers and just return a raw pointer
@@ -58,6 +61,14 @@ class ArrayBuilder {
   int32_t null_count() const { return null_count_; }
   int32_t capacity() const { return capacity_; }
 
+  // Append to null bitmap
+  Status AppendToBitmap(bool is_null);
+  // Vector append. Treat each zero byte as a null.   If valid_bytes is null
+  // assume all of length bits are valid.
+  Status AppendToBitmap(const uint8_t* valid_bytes, int32_t length);
+  // Set the next length bits to not null (i.e. valid).
+  Status SetNotNull(int32_t length);
+
   // Allocates requires memory at this level, but children need to be
   // initialized independently
   Status Init(int32_t capacity);
@@ -75,7 +86,7 @@ class ArrayBuilder {
   const std::shared_ptr<PoolBuffer>& null_bitmap() const { return null_bitmap_; }
 
   // Creates new array object to hold the contents of the builder and transfers
-  // ownership of the data
+  // ownership of the data.  This resets all variables on the builder.
   virtual std::shared_ptr<Array> Finish() = 0;
 
   const std::shared_ptr<DataType>& type() const { return type_; }
@@ -97,6 +108,18 @@ class ArrayBuilder {
   // Child value array builders. These are owned by this class
   std::vector<std::unique_ptr<ArrayBuilder>> children_;
 
+  //
+  // Unsafe operations (don't check capacity/don't resize)
+  //
+
+  // Append to null bitmap.
+  void UnsafeAppendToBitmap(bool is_null);
+  // Vector append. Treat each zero byte as a nullzero. If valid_bytes is null
+  // assume all of length bits are valid.
+  void UnsafeAppendToBitmap(const uint8_t* valid_bytes, int32_t length);
+  // Set the next length bits to not null (i.e. valid).
+  void UnsafeSetNotNull(int32_t length);
+
  private:
   DISALLOW_COPY_AND_ASSIGN(ArrayBuilder);
 };

cpp/src/arrow/ipc/adapter.cc

@@ -19,6 +19,7 @@
 
 #include <cstdint>
 #include <cstring>
+#include <sstream>
 #include <vector>
 
 #include "arrow/array.h"
@@ -31,6 +32,7 @@
 #include "arrow/type.h"
 #include "arrow/types/construct.h"
 #include "arrow/types/primitive.h"
+#include "arrow/types/list.h"
 #include "arrow/util/buffer.h"
 #include "arrow/util/logging.h"
 #include "arrow/util/status.h"
@@ -63,32 +65,56 @@ static bool IsPrimitive(const DataType* type) {
   }
 }
 
+static bool IsListType(const DataType* type) {
+  DCHECK(type != nullptr);
+  switch (type->type) {
+    // TODO(emkornfield) grouping like this are used in a few places in the
+    // code consider using pattern like:
+    // http://stackoverflow.com/questions/26784685/c-macro-for-calling-function-based-on-enum-type
+    //
+    // TODO(emkornfield) Fix type systems so these are all considered lists and
+    // the types behave the same way?
+    // case Type::BINARY:
+    // case Type::CHAR:
+    case Type::LIST:
+      // see todo on common types
+      // case Type::STRING:
+      // case Type::VARCHAR:
+      return true;
+    default:
+      return false;
+  }
+}
+
 // ----------------------------------------------------------------------
 // Row batch write path
 
 Status VisitArray(const Array* arr, std::vector<flatbuf::FieldNode>* field_nodes,
     std::vector<std::shared_ptr<Buffer>>* buffers) {
-  if (IsPrimitive(arr->type().get())) {
-    const PrimitiveArray* prim_arr = static_cast<const PrimitiveArray*>(arr);
-
-    field_nodes->push_back(
-        flatbuf::FieldNode(prim_arr->length(), prim_arr->null_count()));
+  DCHECK(arr);
+  DCHECK(field_nodes);
+  // push back all common elements
+  field_nodes->push_back(flatbuf::FieldNode(arr->length(), arr->null_count()));
+  if (arr->null_count() > 0) {
+    buffers->push_back(arr->null_bitmap());
+  } else {
+    // Push a dummy zero-length buffer, not to be copied
+    buffers->push_back(std::make_shared<Buffer>(nullptr, 0));
+  }
 
-    if (prim_arr->null_count() > 0) {
-      buffers->push_back(prim_arr->null_bitmap());
-    } else {
-      // Push a dummy zero-length buffer, not to be copied
-      buffers->push_back(std::make_shared<Buffer>(nullptr, 0));
-    }
+  const auto arr_type = arr->type().get();
+  if (IsPrimitive(arr_type)) {
+    const PrimitiveArray* prim_arr = static_cast<const PrimitiveArray*>(arr);
     buffers->push_back(prim_arr->data());
-  } else if (arr->type_enum() == Type::LIST) {
-    // TODO(wesm)
-    return Status::NotImplemented("List type");
+  } else if (IsListType(arr_type)) {
+    const ListArray* list_arr = static_cast<const ListArray*>(arr);
+    buffers->push_back(list_arr->offset_buffer());
+    // TODO(emkornfield) limit recursion depth
+    RETURN_NOT_OK(VisitArray(list_arr->values().get(), field_nodes, buffers));
   } else if (arr->type_enum() == Type::STRUCT) {
     // TODO(wesm)
     return Status::NotImplemented("Struct type");
   }
-
   return Status::OK();
 }
 
@@ -214,7 +240,7 @@ class RowBatchReader::Impl {
   // Traverse the flattened record batch metadata and reassemble the
   // corresponding array containers
   Status NextArray(const Field* field, std::shared_ptr<Array>* out) {
-    const std::shared_ptr<DataType>& type = field->type;
+    const TypePtr& type = field->type;
 
     // pop off a field
     if (field_index_ >= num_flattened_fields_) {
@@ -226,15 +252,17 @@ class RowBatchReader::Impl {
     // we can skip that buffer without reading from shared memory
     FieldMetadata field_meta = metadata_->field(field_index_++);
 
+    // extract null_bitmap which is common to all arrays
+    std::shared_ptr<Buffer> null_bitmap;
+    if (field_meta.null_count == 0) {
+      null_bitmap = nullptr;
+      ++buffer_index_;
+    } else {
+      RETURN_NOT_OK(GetBuffer(buffer_index_++, &null_bitmap));
+    }
+
     if (IsPrimitive(type.get())) {
-      std::shared_ptr<Buffer> null_bitmap;
       std::shared_ptr<Buffer> data;
-      if (field_meta.null_count == 0) {
-        null_bitmap = nullptr;
-        ++buffer_index_;
-      } else {
-        RETURN_NOT_OK(GetBuffer(buffer_index_++, &null_bitmap));
-      }
       if (field_meta.length > 0) {
         RETURN_NOT_OK(GetBuffer(buffer_index_++, &data));
       } else {
@@ -243,6 +271,26 @@ class RowBatchReader::Impl {
       return MakePrimitiveArray(
           type, field_meta.length, data, field_meta.null_count, null_bitmap, out);
     }
+    if (IsListType(type.get())) {
+      std::shared_ptr<Buffer> offsets;
+      if (field_meta.length > 0) {
+        RETURN_NOT_OK(GetBuffer(buffer_index_++, &offsets));
+      } else {
+        offsets.reset(new Buffer(nullptr, 0));
+      }
+      const int num_children = type->num_children();
+      if (num_children != 1) {
+        std::stringstream ss;
+        ss << "Field: " << field->ToString()
+           << " has wrong number of children:" << num_children;
+        return Status::Invalid(ss.str());
+      }
+      std::shared_ptr<Array> values_array;
+      // TODO(emkornfield): limit recursion depth?
+      RETURN_NOT_OK(NextArray(type->child(0).get(), &values_array));
+      return MakeListArray(type, field_meta.length, offsets, values_array,
+          field_meta.null_count, null_bitmap, out);
+    }
     return Status::NotImplemented("Non-primitive types not complete yet");
   }