parallelize arrow memcpy and memset operations

cpp/CMakeLists.txt

@@ -28,7 +28,7 @@ set(THIRDPARTY_DIR "${CMAKE_SOURCE_DIR}/thirdparty")
 
 set(GTEST_VERSION "1.7.0")
 set(GBENCHMARK_VERSION "1.0.0")
-set(FLATBUFFERS_VERSION "1.3.0")
+set(FLATBUFFERS_VERSION "1.6.0")
 
 find_package(ClangTools)
 if ("$ENV{CMAKE_EXPORT_COMPILE_COMMANDS}" STREQUAL "1" OR CLANG_TIDY_FOUND)

cpp/src/arrow/builder.cc

@@ -116,10 +116,24 @@ void ArrayBuilder::UnsafeAppendToBitmap(const uint8_t* valid_bytes, int32_t leng
 
 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) {
+
+  // Fill up the bytes until we have a byte alignment
+  int32_t pad_to_byte = 8 - (length_ % 8);
+  if (pad_to_byte == 8) { pad_to_byte = 0; }
+  for (int32_t i = 0; i < pad_to_byte; ++i) {
+    BitUtil::SetBit(null_bitmap_data_, i);
+  }
+
+  // Fast bitsetting
+  int32_t fast_length = (length - pad_to_byte) / 8;
+  memset(null_bitmap_data_ + ((length_ + pad_to_byte) / 8), 255,
+      static_cast<size_t>(fast_length));
+
+  // Trailing bytes
+  for (int32_t i = length_ + pad_to_byte + (fast_length * 8); i < new_length; ++i) {
     BitUtil::SetBit(null_bitmap_data_, i);
   }
+
   length_ = new_length;
 }
 

cpp/src/arrow/builder.h

@@ -130,6 +130,15 @@ class ARROW_EXPORT ArrayBuilder {
   DISALLOW_COPY_AND_ASSIGN(ArrayBuilder);
 };
 
+class ARROW_EXPORT NullArrayBuilder : public ArrayBuilder {
+ public:
+  explicit NullArrayBuilder(MemoryPool* pool, const TypePtr& type) : ArrayBuilder(pool, type) {}
+  virtual ~NullArrayBuilder() {};
+  Status Finish(std::shared_ptr<Array>* out) override {
+    return Status::OK();
+  }
+};
+
 }  // namespace arrow
 
 #endif  // ARROW_BUILDER_H_

cpp/src/arrow/io/CMakeLists.txt

@@ -80,9 +80,11 @@ if(ARROW_HDFS)
     ${ARROW_IO_TEST_LINK_LIBS})
 endif()
 
-add_library(arrow_io SHARED
+add_library(arrow_io STATIC
   ${ARROW_IO_SRCS}
 )
+set_property(TARGET arrow_io PROPERTY POSITION_INDEPENDENT_CODE 1)
+
 target_link_libraries(arrow_io
   LINK_PUBLIC ${ARROW_IO_LINK_LIBS}
   LINK_PRIVATE ${ARROW_IO_PRIVATE_LINK_LIBS})

cpp/src/arrow/ipc/CMakeLists.txt

@@ -41,9 +41,10 @@ set(ARROW_IPC_SRCS
 )
 
 # TODO(wesm): SHARED and STATIC targets
-add_library(arrow_ipc SHARED
+add_library(arrow_ipc STATIC
   ${ARROW_IPC_SRCS}
 )
+set_property(TARGET arrow_ipc PROPERTY POSITION_INDEPENDENT_CODE 1)
 if(FLATBUFFERS_VENDORED)
   add_dependencies(arrow_ipc flatbuffers_ep)
 endif()

cpp/src/arrow/ipc/adapter.cc

@@ -37,6 +37,7 @@
 #include "arrow/types/primitive.h"
 #include "arrow/types/string.h"
 #include "arrow/types/struct.h"
+#include "arrow/types/union.h"
 #include "arrow/util/bit-util.h"
 #include "arrow/util/buffer.h"
 #include "arrow/util/logging.h"
@@ -115,6 +116,13 @@ Status VisitArray(const Array* arr, std::vector<flatbuf::FieldNode>* field_nodes
       RETURN_NOT_OK(
           VisitArray(field.get(), field_nodes, buffers, max_recursion_depth - 1));
     }
+  } else if (arr->type_enum() == Type::UNION) {
+    const auto union_arr = static_cast<const UnionArray*>(arr);
+    buffers->push_back(union_arr->types());
+    buffers->push_back(union_arr->offset_buf());
+    for (auto& child_arr : union_arr->children()) {
+      RETURN_NOT_OK(VisitArray(child_arr.get(), field_nodes, buffers, max_recursion_depth - 1));
+    }
   } else {
     return Status::NotImplemented("Unrecognized type");
   }
@@ -363,6 +371,22 @@ class RecordBatchReader::RecordBatchReaderImpl {
       out->reset(new StructArray(
           type, field_meta.length, fields, field_meta.null_count, null_bitmap));
       return Status::OK();
+    } else if (type->type == Type::UNION) {
+      std::shared_ptr<Buffer> types;
+      RETURN_NOT_OK(GetBuffer(buffer_index_++, &types));
+      std::shared_ptr<Buffer> offset_buf;
+      RETURN_NOT_OK(GetBuffer(buffer_index_++, &offset_buf));
+      auto union_type = std::dynamic_pointer_cast<UnionType>(type);
+      const int num_children = union_type->num_children();
+      std::vector<ArrayPtr> results;
+      for (int child_idx = 0; child_idx < num_children; ++child_idx) {
+        std::shared_ptr<Array> result;
+        RETURN_NOT_OK(NextArray(union_type->child(child_idx).get(), max_recursion_depth - 1, &result));
+        results.push_back(result);
+      }
+      out->reset(new UnionArray(
+        type, field_meta.length, results, types, offset_buf, field_meta.null_count, null_bitmap));
+      return Status::OK();
     }
 
     return Status::NotImplemented("Non-primitive types not complete yet");

cpp/src/arrow/ipc/ipc-metadata-test.cc

@@ -26,6 +26,7 @@
 #include "arrow/schema.h"
 #include "arrow/test-util.h"
 #include "arrow/type.h"
+#include "arrow/types/union.h"
 #include "arrow/util/status.h"
 
 namespace arrow {
@@ -97,6 +98,22 @@ TEST_F(TestSchemaMessage, NestedFields) {
   CheckRoundtrip(&schema);
 }
 
+TEST_F(TestSchemaMessage, UnionType) {
+  auto f0 = std::make_shared<Field>("f0", TypePtr(new Int32Type()));
+  auto f1 = std::make_shared<Field>("f1", TypePtr(new Int64Type()));
+  std::vector<uint8_t> type_ids = {}; // TODO(pcm): Implement typeIds
+  auto ud = TypePtr(new UnionType(std::vector<std::shared_ptr<Field>>({f0, f1}),
+                                  type_ids, UnionMode::DENSE));
+  auto fd = std::make_shared<Field>("f", ud);
+  Schema schema_dense({fd});
+  CheckRoundtrip(&schema_dense);
+  auto us = TypePtr(new UnionType(std::vector<std::shared_ptr<Field>>({f0, f1}),
+                                  type_ids, UnionMode::SPARSE));
+  auto fs = std::make_shared<Field>("f", us);
+  Schema schema_sparse({fs});
+  CheckRoundtrip(&schema_sparse);
+}
+
 class TestFileFooter : public ::testing::Test {
  public:
   void SetUp() {}

cpp/src/arrow/ipc/metadata-internal.cc

@@ -28,6 +28,7 @@
 #include "arrow/ipc/Message_generated.h"
 #include "arrow/schema.h"
 #include "arrow/type.h"
+#include "arrow/types/union.h"
 #include "arrow/util/buffer.h"
 #include "arrow/util/status.h"
 
@@ -119,8 +120,20 @@ static Status TypeFromFlatbuffer(flatbuf::Type type, const void* type_data,
     case flatbuf::Type_Struct_:
       *out = std::make_shared<StructType>(children);
       return Status::OK();
-    case flatbuf::Type_Union:
-      return Status::NotImplemented("Type is not implemented");
+    case flatbuf::Type_Union: {
+        std::vector<uint8_t> type_ids = {}; // TODO(pcm): Implement typeIds
+        auto union_data = static_cast<const flatbuf::Union*>(type_data);
+        UnionMode mode;
+        if (union_data->mode() == flatbuf::UnionMode_Sparse) {
+          mode = UnionMode::SPARSE;
+        } else if (union_data->mode() == flatbuf::UnionMode_Dense) {
+          mode = UnionMode::DENSE;
+        } else {
+          return Status::Invalid("Unrecognized UnionMode");
+        }
+        *out = std::make_shared<UnionType>(children, type_ids, mode);
+      }
+      return Status::OK();
     default:
       return Status::Invalid("Unrecognized type");
   }
@@ -158,6 +171,18 @@ static Status StructToFlatbuffer(FBB& fbb, const std::shared_ptr<DataType>& type
   return Status::OK();
 }
 
+static Status UnionToFlatbuffer(FBB& fbb, const std::shared_ptr<DataType>& type,
+    std::vector<FieldOffset>* out_children, Offset* offset) {
+  auto union_type = std::dynamic_pointer_cast<UnionType>(type);
+  FieldOffset field;
+  for (int i = 0; i < union_type->num_children(); ++i) {
+    RETURN_NOT_OK(FieldToFlatbuffer(fbb, union_type->child(i), &field));
+    out_children->push_back(field);
+  }
+  *offset = flatbuf::CreateUnion(fbb).Union();
+  return Status::OK();
+}
+
 #define INT_TO_FB_CASE(BIT_WIDTH, IS_SIGNED)            \
   *out_type = flatbuf::Type_Int;                        \
   *offset = IntToFlatbuffer(fbb, BIT_WIDTH, IS_SIGNED); \
@@ -208,6 +233,9 @@ static Status TypeToFlatbuffer(FBB& fbb, const std::shared_ptr<DataType>& type,
     case Type::STRUCT:
       *out_type = flatbuf::Type_Struct_;
       return StructToFlatbuffer(fbb, type, children, offset);
+    case Type::UNION:
+      *out_type = flatbuf::Type_Union;
+      return UnionToFlatbuffer(fbb, type, children, offset);
     default:
       *out_type = flatbuf::Type_NONE;  // Make clang-tidy happy
       std::stringstream ss;

cpp/src/arrow/type.cc

@@ -105,6 +105,16 @@ std::string UnionType::ToString() const {
   return s.str();
 }
 
+bool UnionType::Equals(const DataType* other) const {
+  if (!DataType::Equals(other)) {
+    return false;
+  }
+  const UnionType *union_type = dynamic_cast<const UnionType*>(other);
+  return union_type && type_id == union_type->type_id
+             && std::equal(type_ids.begin(), type_ids.end(),
+                           union_type->type_ids.begin());
+}
+
 int NullType::bit_width() const {
   return 0;
 }

cpp/src/arrow/type.h

@@ -375,6 +375,11 @@ struct ARROW_EXPORT UnionType : public DataType {
   static std::string name() { return "union"; }
   Status Accept(TypeVisitor* visitor) const override;
 
+  bool Equals(const DataType* other) const override;
+  bool Equals(const std::shared_ptr<DataType>& other) const {
+    return Equals(other.get());
+  }
+
   UnionMode mode;
   std::vector<uint8_t> type_ids;
 };

cpp/src/arrow/types/primitive.cc

@@ -23,6 +23,7 @@
 #include "arrow/util/bit-util.h"
 #include "arrow/util/buffer.h"
 #include "arrow/util/logging.h"
+#include "arrow/util/memory-util.h"
 
 namespace arrow {
 
@@ -98,7 +99,7 @@ Status PrimitiveBuilder<T>::Init(int32_t capacity) {
   int64_t nbytes = TypeTraits<T>::bytes_required(capacity);
   RETURN_NOT_OK(data_->Resize(nbytes));
   // TODO(emkornfield) valgrind complains without this
-  memset(data_->mutable_data(), 0, nbytes);
+  // memset(data_->mutable_data(), 0, nbytes);
 
   raw_data_ = reinterpret_cast<value_type*>(data_->mutable_data());
   return Status::OK();
@@ -117,7 +118,12 @@ Status PrimitiveBuilder<T>::Resize(int32_t capacity) {
     const int64_t new_bytes = TypeTraits<T>::bytes_required(capacity);
     RETURN_NOT_OK(data_->Resize(new_bytes));
     raw_data_ = reinterpret_cast<value_type*>(data_->mutable_data());
-    memset(data_->mutable_data() + old_bytes, 0, new_bytes - old_bytes);
+    if ((new_bytes - old_bytes) >= 32*KB) {
+      memset_page_aligned(
+          data_->mutable_data() + old_bytes, 0, new_bytes - old_bytes);
+    } else {
+      memset(data_->mutable_data() + old_bytes, 0, new_bytes - old_bytes);
+    }
   }
   return Status::OK();
 }
@@ -128,7 +134,13 @@ Status PrimitiveBuilder<T>::Append(
   RETURN_NOT_OK(Reserve(length));
 
   if (length > 0) {
-    memcpy(raw_data_ + length_, values, TypeTraits<T>::bytes_required(length));
+    size_t numbytes = TypeTraits<T>::bytes_required(length);
+    if (numbytes >= MB) {
+        memcopy_block_aligned((uint8_t *)(raw_data_ + length_),
+                              (uint8_t *)values, numbytes);
+    } else {
+      memcpy(raw_data_ + length_, values, numbytes);
+    }
   }
 
   // length_ is update by these
@@ -172,7 +184,7 @@ Status BooleanBuilder::Init(int32_t capacity) {
   int64_t nbytes = BitUtil::BytesForBits(capacity);
   RETURN_NOT_OK(data_->Resize(nbytes));
   // TODO(emkornfield) valgrind complains without this
-  memset(data_->mutable_data(), 0, nbytes);
+  //memset(data_->mutable_data(), 0, nbytes);
 
   raw_data_ = reinterpret_cast<uint8_t*>(data_->mutable_data());
   return Status::OK();
@@ -191,7 +203,12 @@ Status BooleanBuilder::Resize(int32_t capacity) {
 
     RETURN_NOT_OK(data_->Resize(new_bytes));
     raw_data_ = reinterpret_cast<uint8_t*>(data_->mutable_data());
-    memset(data_->mutable_data() + old_bytes, 0, new_bytes - old_bytes);
+    if ((new_bytes - old_bytes) >= 32*KB) {
+      memset_page_aligned(
+          data_->mutable_data() + old_bytes, 0, new_bytes - old_bytes);
+    } else {
+      memset(data_->mutable_data() + old_bytes, 0, new_bytes - old_bytes);
+    }
   }
   return Status::OK();
 }

cpp/src/arrow/types/union.cc

@@ -23,5 +23,50 @@
 #include <vector>
 
 #include "arrow/type.h"
+#include "arrow/util/status.h"
 
-namespace arrow {}  // namespace arrow
+namespace arrow {
+
+bool UnionArray::Equals(const std::shared_ptr<Array>& arr) const {
+  if (this == arr.get()) { return true; }
+  if (!arr) { return false; }
+  if (this->type_enum() != arr->type_enum()) { return false; }
+  if (null_count_ != arr->null_count()) { return false; }
+  return RangeEquals(0, length_, 0, arr);
+}
+
+bool UnionArray::RangeEquals(int32_t start_idx, int32_t end_idx, int32_t other_start_idx,
+      const std::shared_ptr<Array>& arr) const {
+  if (this == arr.get()) { return true; }
+  if (Type::UNION != arr->type_enum()) { return false; }
+  const auto other = static_cast<UnionArray*>(arr.get());
+
+  // TODO(pcm): Handle sparse case here
+
+  int32_t i = start_idx;
+  int32_t o_i = other_start_idx;
+  for (size_t c = 0; c < other->children().size(); ++c) {
+    for (int32_t e = 0; e < other->children()[c]->length(); ++e) {
+      if (!children()[c]->RangeEquals(e, e + 1, e, other->children()[c])) { // FIXME(pcm): fix this
+        return false;
+      }
+      i += 1;
+      o_i += 1;
+      if (i >= end_idx) {
+        return true;
+      }
+    }
+  }
+  return false; // to make the compiler happy
+}
+
+Status UnionArray::Validate() const {
+  // TODO(pcm): what to do here?
+  return Status::OK();
+}
+
+Status UnionArray::Accept(ArrayVisitor* visitor) const {
+  return visitor->Visit(*this);
+}
+
+}  // namespace arrow