diff --git a/cpp/CMakeLists.txt b/cpp/CMakeLists.txt
index 0bff7528578..83e4e6dec2b 100644
--- a/cpp/CMakeLists.txt
+++ b/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)
diff --git a/cpp/src/arrow/builder.cc b/cpp/src/arrow/builder.cc
index 151b257a3d8..ee4b1b9c0ae 100644
--- a/cpp/src/arrow/builder.cc
+++ b/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;
 }
 
diff --git a/cpp/src/arrow/builder.h b/cpp/src/arrow/builder.h
index cef17e5aaba..3babf4ba78e 100644
--- a/cpp/src/arrow/builder.h
+++ b/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_
diff --git a/cpp/src/arrow/io/CMakeLists.txt b/cpp/src/arrow/io/CMakeLists.txt
index 47bb0893863..2c562bc028d 100644
--- a/cpp/src/arrow/io/CMakeLists.txt
+++ b/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})
diff --git a/cpp/src/arrow/ipc/CMakeLists.txt b/cpp/src/arrow/ipc/CMakeLists.txt
index 6955bcb6c23..d0816c4a42b 100644
--- a/cpp/src/arrow/ipc/CMakeLists.txt
+++ b/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()
diff --git a/cpp/src/arrow/ipc/adapter.cc b/cpp/src/arrow/ipc/adapter.cc
index da718c08d54..4628ace3abf 100644
--- a/cpp/src/arrow/ipc/adapter.cc
+++ b/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");
diff --git a/cpp/src/arrow/ipc/ipc-metadata-test.cc b/cpp/src/arrow/ipc/ipc-metadata-test.cc
index 1dc39692332..8d101cf586a 100644
--- a/cpp/src/arrow/ipc/ipc-metadata-test.cc
+++ b/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() {}
diff --git a/cpp/src/arrow/ipc/metadata-internal.cc b/cpp/src/arrow/ipc/metadata-internal.cc
index 7102012c29a..19a810ac1c6 100644
--- a/cpp/src/arrow/ipc/metadata-internal.cc
+++ b/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;
diff --git a/cpp/src/arrow/type.cc b/cpp/src/arrow/type.cc
index 589bdadb77c..e82486d0ed2 100644
--- a/cpp/src/arrow/type.cc
+++ b/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;
 }
diff --git a/cpp/src/arrow/type.h b/cpp/src/arrow/type.h
index 5b4d7bc42bd..e2f47f11b07 100644
--- a/cpp/src/arrow/type.h
+++ b/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;
 };
diff --git a/cpp/src/arrow/types/primitive.cc b/cpp/src/arrow/types/primitive.cc
index 14667ee5b6e..2aa43ab0e8b 100644
--- a/cpp/src/arrow/types/primitive.cc
+++ b/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();
 }
diff --git a/cpp/src/arrow/types/union.cc b/cpp/src/arrow/types/union.cc
index cc2934b2e4a..2a5be998694 100644
--- a/cpp/src/arrow/types/union.cc
+++ b/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
diff --git a/cpp/src/arrow/types/union.h b/cpp/src/arrow/types/union.h
index 44f39cc6994..2e56ecf2e99 100644
--- a/cpp/src/arrow/types/union.h
+++ b/cpp/src/arrow/types/union.h
@@ -24,25 +24,45 @@
 
 #include "arrow/array.h"
 #include "arrow/type.h"
+#include "arrow/types/primitive.h"
 
 namespace arrow {
 
 class Buffer;
 
-class UnionArray : public Array {
+class ARROW_EXPORT UnionArray : public Array {
+ public:
+  UnionArray(const TypePtr& type, int32_t length, std::vector<ArrayPtr>& children,
+       std::shared_ptr<Buffer> types, std::shared_ptr<Buffer> offset_buf,
+       int32_t null_count = 0, std::shared_ptr<Buffer> null_bitmap = nullptr)
+       : Array(type, length, null_count, null_bitmap), types_(types) {
+    type_ = type;
+    children_ = children;
+    offset_buf_ = offset_buf;
+  }
+
+  const std::shared_ptr<Buffer>& types() const { return types_; }
+
+  const std::vector<ArrayPtr>& children() const { return children_; }
+
+  const std::shared_ptr<Buffer>& offset_buf() const { return offset_buf_; }
+
+  Status Validate() const override;
+  
+  Status Accept(ArrayVisitor* visitor) const override;
+
+  bool Equals(const std::shared_ptr<Array>& arr) const override;
+  bool RangeEquals(int32_t start_idx, int32_t end_idx, int32_t other_start_idx,
+      const std::shared_ptr<Array>& arr) const override;
+
+  ArrayPtr  child(int32_t index) const { return children_[index]; }
  protected:
   // The data are types encoded as int16
-  Buffer* types_;
+  std::shared_ptr<Buffer> types_;
   std::vector<std::shared_ptr<Array>> children_;
+  std::shared_ptr<Buffer> offset_buf_;
 };
 
-class DenseUnionArray : public UnionArray {
- protected:
-  Buffer* offset_buf_;
-};
-
-class SparseUnionArray : public UnionArray {};
-
 }  // namespace arrow
 
 #endif  // ARROW_TYPES_UNION_H
diff --git a/cpp/src/arrow/util/memory-util.h b/cpp/src/arrow/util/memory-util.h
new file mode 100644
index 00000000000..6c00f91e685
--- /dev/null
+++ b/cpp/src/arrow/util/memory-util.h
@@ -0,0 +1,152 @@
+// 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 ARROW_UTIL_MEM_UTIL_H
+#define ARROW_UTIL_MEM_UTIL_H
+#include <cassert>
+#include <vector>
+#include <thread>
+
+#include <string.h>
+#include <unistd.h>
+#include <sys/time.h>
+#include <stdlib.h>
+#include <stdio.h>
+
+#define NUMTHREADS 8
+#define MB  (1<<20)
+#define KB  (1<<10)
+
+namespace arrow {
+
+static std::vector<std::thread> threadpool(NUMTHREADS);
+
+static inline int memcopy_aligned(uint8_t *dst, const uint8_t *src,
+                                  uint64_t nbytes, uint64_t blocksz,
+                                  bool timeit) {
+  int rv = 0;
+  struct timeval tv1, tv2;
+  double elapsed = 0;
+  const uint64_t numthreads = NUMTHREADS;
+  const char *srcbp = (char *)(((uint64_t)src + blocksz-1) & ~(blocksz-1));
+  char *srcep = (char *)(((uint64_t)src + nbytes) & ~(blocksz-1));
+
+  const uint64_t numblocks = (((uint64_t)srcep - (uint64_t)srcbp)) / blocksz;
+  // Now we divide these blocks between available threads. Remainder is pushed
+  // to the suffix-handling thread.
+  // uint64_t remainder = numblocks % numthreads;
+  // Update the end pointer
+  srcep = srcep - (numblocks % numthreads)*blocksz;
+  const uint64_t chunksz = ((uint64_t)srcep - (uint64_t)srcbp) / numthreads;//B
+  //assert(srcep >= srcbp);
+  const uint64_t prefix = (uint64_t)srcbp - (uint64_t)src; // Bytes
+  const uint64_t suffix = (uint64_t)(src+nbytes) - (uint64_t)srcep; // Bytes
+  char *dstep = (char *)((uint64_t)dst + prefix + numthreads*chunksz);
+  // Now data == | prefix | k*numthreads*blocksz | suffix |
+  // chunksz = k*blocksz => data == | prefix | numthreads*chunksz | suffix |
+  // Each thread gets a "chunk" of k blocks, except prefix and suffix threads.
+
+  if (timeit) {
+    gettimeofday(&tv1, NULL);
+  }
+  // Start memcpy threads and then copy the prefix and suffix while threads run.
+  for (int i = 0; i < numthreads; i++) {
+    threadpool[i] = std::thread(
+        memcpy, dst+prefix+i*chunksz, srcbp + i*chunksz, chunksz);
+  }
+//  threads.push_back(std::thread(memcpy, dstep, srcep, suffix));
+//  threadpool[NUMTHREADS-1] = std::thread(memcpy, dstep, srcep, suffix);
+  memcpy(dst, src, prefix);
+  memcpy(dstep, srcep, suffix);
+
+  for (auto &t: threadpool) {
+      t.join(); // Join all the memcpy threads.
+  }
+  if (timeit) {
+    gettimeofday(&tv2, NULL);
+    elapsed = ((tv2.tv_sec - tv1.tv_sec)*1000000 + (tv2.tv_usec - tv1.tv_usec))/1000000.0;
+    printf("copied %llu bytes in time = %8.4f MBps=%8.4f\n",
+           nbytes, elapsed, nbytes/((1<<20)*elapsed));
+  }
+  return rv;
+}
+
+static inline int memset_aligned(uint8_t *dst, int val, uint64_t nbytes,
+                                 uint64_t blocksz, bool timeit) {
+  int rv = 0;
+  struct timeval tv1, tv2;
+  double elapsed = 0;
+  const uint64_t numthreads = NUMTHREADS;
+  const char *dstbp = (char *)(((uint64_t)dst + blocksz-1) & ~(blocksz-1));
+  char *dstep = (char *)(((uint64_t)dst + nbytes) & ~(blocksz-1));
+  const uint64_t chunksz = ((uint64_t)dstep - (uint64_t)dstbp) / numthreads;//B
+  //assert(dstep >= dstbp);
+  const uint64_t numblocks = (((uint64_t)dstep - (uint64_t)dstbp)) / blocksz;
+  // Now we divide these blocks between available threads. Remainder is pushed
+  // to the suffix-handling thread.
+  // uint64_t remainder = numblocks % numthreads;
+  // Update the end pointer
+  dstep = dstep - (numblocks % numthreads)*blocksz;
+  const uint64_t prefix = (uint64_t)dstbp - (uint64_t)dst; // Bytes
+  const uint64_t suffix = (uint64_t)(dst+nbytes) - (uint64_t)dstep; // Bytes
+  std::vector<std::thread> threads;
+  if (timeit) {
+    gettimeofday(&tv1, NULL);
+  }
+//  threads.push_back(std::thread(memset, dst, val, prefix));
+  // Start all threads first. Handle leftovers while threads are running.
+  for (int i = 0; i < numthreads; i++) {
+//    threads.push_back(std::thread(
+    threadpool[i] = std::thread(memset, dst+prefix+i*chunksz, val, chunksz);
+  }
+//  threads.push_back(std::thread(memset, dstep, val, suffix));
+  memset(dst, val, prefix);
+  memset(dstep, val, suffix);
+
+  // Join the memcpy threads.
+  for (auto &t : threadpool) {
+    t.join();
+  }
+  if (timeit) {
+    gettimeofday(&tv2, NULL);
+    elapsed =
+        ((tv2.tv_sec - tv1.tv_sec) * 1000000 + (tv2.tv_usec - tv1.tv_usec))
+            / 1000000.0;
+    printf("copied %llu bytes in time = %8.4f MBps=%8.4f\n", nbytes, elapsed,
+        nbytes / ((1 << 20) * elapsed));
+  }
+  return rv;
+}
+
+inline int memset_block_aligned(uint8_t *dst, int val, uint64_t nbytes){
+  return memset_aligned(dst, val, nbytes, 64, false);
+}
+
+inline int memset_page_aligned(uint8_t *dst, int val, uint64_t nbytes) {
+  return memset_aligned(dst, val, nbytes, getpagesize(), false);
+}
+
+inline int memcopy_block_aligned(uint8_t *dst, uint8_t *src, uint64_t nbytes) {
+  return memcopy_aligned(dst, src, nbytes, 64, false);
+}
+inline int memcopy_page_aligned(uint8_t *dst, uint8_t *src, uint64_t nbytes) {
+  return memcopy_aligned(dst, src, nbytes, getpagesize(), false);
+}
+
+}  // namespace arrow
+
+#endif  // ARROW_UTIL_MEM_UTIL_H