From cbd492d76b1a3872f18e3e65bf9332cf2e15fffc Mon Sep 17 00:00:00 2001
From: Wes McKinney <wesm@apache.org>
Date: Sun, 7 Jun 2020 19:57:31 -0500
Subject: [PATCH 1/8] New Take implementation and support infrastructure

---
 cpp/src/arrow/compute/api_vector.h            |   7 +-
 cpp/src/arrow/compute/benchmark_util.h        |  15 +-
 cpp/src/arrow/compute/kernel.cc               |  82 +-
 cpp/src/arrow/compute/kernel.h                |  13 +
 .../kernels/vector_selection_benchmark.cc     |  12 +-
 cpp/src/arrow/compute/kernels/vector_take.cc  | 791 +++++++++++++++++-
 .../arrow/compute/kernels/vector_take_test.cc | 156 +++-
 cpp/src/arrow/type_traits.h                   |   1 -
 cpp/src/arrow/util/bit_block_counter.cc       |  12 +
 cpp/src/arrow/util/bit_block_counter.h        |  42 +
 cpp/src/arrow/util/bit_block_counter_test.cc  |  23 +
 cpp/src/arrow/util/bitmap_reader.h            |  14 +
 cpp/src/arrow/util/int_util.cc                |  90 ++
 cpp/src/arrow/util/int_util.h                 |  10 +
 cpp/src/arrow/util/int_util_test.cc           | 101 +++
 cpp/src/arrow/util/macros.h                   |   2 +
 python/pyarrow/_compute.pyx                   |  11 +
 python/pyarrow/compute.py                     |  11 +-
 python/pyarrow/includes/libarrow.pxd          |   7 +-
 python/pyarrow/tests/test_pandas.py           |   1 +
 20 files changed, 1314 insertions(+), 87 deletions(-)

diff --git a/cpp/src/arrow/compute/api_vector.h b/cpp/src/arrow/compute/api_vector.h
index 84086802f18..8aefbcf1ccf 100644
--- a/cpp/src/arrow/compute/api_vector.h
+++ b/cpp/src/arrow/compute/api_vector.h
@@ -65,7 +65,12 @@ Result<Datum> Filter(const Datum& values, const Datum& filter,
                      ExecContext* ctx = NULLPTR);
 
 struct ARROW_EXPORT TakeOptions : public FunctionOptions {
-  static TakeOptions Defaults() { return TakeOptions{}; }
+  explicit TakeOptions(bool boundscheck = true) : boundscheck(boundscheck) {}
+
+  bool boundscheck = true;
+  static TakeOptions Boundscheck() { return TakeOptions(true); }
+  static TakeOptions NoBoundscheck() { return TakeOptions(false); }
+  static TakeOptions Defaults() { return Boundscheck(); }
 };
 
 /// \brief Take from an array of values at indices in another array
diff --git a/cpp/src/arrow/compute/benchmark_util.h b/cpp/src/arrow/compute/benchmark_util.h
index 77149f91aab..702f1c7056a 100644
--- a/cpp/src/arrow/compute/benchmark_util.h
+++ b/cpp/src/arrow/compute/benchmark_util.h
@@ -56,9 +56,10 @@ void BenchmarkSetArgsWithSizes(benchmark::internal::Benchmark* bench,
                                const std::vector<int64_t>& sizes = kMemorySizes) {
   bench->Unit(benchmark::kMicrosecond);
 
+  // 0 is treated as "no nulls"
   for (const auto size : sizes) {
     for (const auto inverse_null_proportion :
-         std::vector<ArgsType>({10000, 100, 10, 2, 1})) {
+           std::vector<ArgsType>({10000, 100, 50, 10, 2, 1, 0})) {
       bench->Args({static_cast<ArgsType>(size), inverse_null_proportion});
     }
   }
@@ -80,12 +81,16 @@ struct RegressionArgs {
   const int64_t size;
 
   // proportion of nulls in generated arrays
-  const double null_proportion;
+  double null_proportion;
 
   explicit RegressionArgs(benchmark::State& state)
-      : size(state.range(0)),
-        null_proportion(std::min(1., 1. / static_cast<double>(state.range(1)))),
-        state_(state) {}
+      : size(state.range(0)), null_proportion(), state_(state) {
+    if (state.range(1) == 0) {
+      this->null_proportion = 0.0;
+    } else {
+      this->null_proportion = std::min(1., 1. / static_cast<double>(state.range(1)));
+    }
+  }
 
   ~RegressionArgs() {
     state_.counters["size"] = static_cast<double>(size);
diff --git a/cpp/src/arrow/compute/kernel.cc b/cpp/src/arrow/compute/kernel.cc
index 1a6f1bd1031..4747e63c159 100644
--- a/cpp/src/arrow/compute/kernel.cc
+++ b/cpp/src/arrow/compute/kernel.cc
@@ -26,6 +26,7 @@
 #include "arrow/compute/exec.h"
 #include "arrow/compute/util_internal.h"
 #include "arrow/result.h"
+#include "arrow/type_traits.h"
 #include "arrow/util/bit_util.h"
 #include "arrow/util/checked_cast.h"
 #include "arrow/util/hash_util.h"
@@ -96,7 +97,6 @@ class SameTypeIdMatcher : public TypeMatcher {
     if (this == &other) {
       return true;
     }
-
     auto casted = dynamic_cast<const SameTypeIdMatcher*>(&other);
     if (casted == nullptr) {
       return false;
@@ -150,6 +150,86 @@ std::shared_ptr<TypeMatcher> TimestampUnit(TimeUnit::type unit) {
   return std::make_shared<TimestampUnitMatcher>(unit);
 }
 
+class IntegerMatcher : public TypeMatcher {
+ public:
+  IntegerMatcher() {}
+
+  bool Matches(const DataType& type) const override { return is_integer(type.id()); }
+
+  bool Equals(const TypeMatcher& other) const override {
+    if (this == &other) {
+      return true;
+    }
+    auto casted = dynamic_cast<const IntegerMatcher*>(&other);
+    return casted != nullptr;
+  }
+
+  std::string ToString() const override { return "integer"; }
+};
+
+std::shared_ptr<TypeMatcher> Integer() { return std::make_shared<IntegerMatcher>(); }
+
+class PrimitiveMatcher : public TypeMatcher {
+ public:
+  PrimitiveMatcher() {}
+
+  bool Matches(const DataType& type) const override { return is_primitive(type.id()); }
+
+  bool Equals(const TypeMatcher& other) const override {
+    if (this == &other) {
+      return true;
+    }
+    auto casted = dynamic_cast<const PrimitiveMatcher*>(&other);
+    return casted != nullptr;
+  }
+
+  std::string ToString() const override { return "primitive"; }
+};
+
+std::shared_ptr<TypeMatcher> Primitive() { return std::make_shared<PrimitiveMatcher>(); }
+
+class BinaryLikeMatcher : public TypeMatcher {
+ public:
+  BinaryLikeMatcher() {}
+
+  bool Matches(const DataType& type) const override { return is_binary_like(type.id()); }
+
+  bool Equals(const TypeMatcher& other) const override {
+    if (this == &other) {
+      return true;
+    }
+    auto casted = dynamic_cast<const BinaryLikeMatcher*>(&other);
+    return casted != nullptr;
+  }
+  std::string ToString() const override { return "binary-like"; }
+};
+
+std::shared_ptr<TypeMatcher> BinaryLike() {
+  return std::make_shared<BinaryLikeMatcher>();
+}
+
+class LargeBinaryLikeMatcher : public TypeMatcher {
+ public:
+  LargeBinaryLikeMatcher() {}
+
+  bool Matches(const DataType& type) const override {
+    return is_large_binary_like(type.id());
+  }
+
+  bool Equals(const TypeMatcher& other) const override {
+    if (this == &other) {
+      return true;
+    }
+    auto casted = dynamic_cast<const LargeBinaryLikeMatcher*>(&other);
+    return casted != nullptr;
+  }
+  std::string ToString() const override { return "large-binary-like"; }
+};
+
+std::shared_ptr<TypeMatcher> LargeBinaryLike() {
+  return std::make_shared<LargeBinaryLikeMatcher>();
+}
+
 }  // namespace match
 
 // ----------------------------------------------------------------------
diff --git a/cpp/src/arrow/compute/kernel.h b/cpp/src/arrow/compute/kernel.h
index bdd2e9d1079..059a7c2001a 100644
--- a/cpp/src/arrow/compute/kernel.h
+++ b/cpp/src/arrow/compute/kernel.h
@@ -150,6 +150,19 @@ ARROW_EXPORT std::shared_ptr<TypeMatcher> SameTypeId(Type::type type_id);
 /// zones can be different.
 ARROW_EXPORT std::shared_ptr<TypeMatcher> TimestampUnit(TimeUnit::type unit);
 
+// \brief Match any integer type
+ARROW_EXPORT std::shared_ptr<TypeMatcher> Integer();
+
+// Match types using 32-bit varbinary representation
+ARROW_EXPORT std::shared_ptr<TypeMatcher> BinaryLike();
+
+// Match types using 64-bit varbinary representation
+ARROW_EXPORT std::shared_ptr<TypeMatcher> LargeBinaryLike();
+
+// \brief Match any primitive type (boolean or any type representable as a C
+// Type)
+ARROW_EXPORT std::shared_ptr<TypeMatcher> Primitive();
+
 }  // namespace match
 
 /// \brief An object used for type- and shape-checking arguments to be passed
diff --git a/cpp/src/arrow/compute/kernels/vector_selection_benchmark.cc b/cpp/src/arrow/compute/kernels/vector_selection_benchmark.cc
index 6031a907dee..822363ac554 100644
--- a/cpp/src/arrow/compute/kernels/vector_selection_benchmark.cc
+++ b/cpp/src/arrow/compute/kernels/vector_selection_benchmark.cc
@@ -90,7 +90,7 @@ struct TakeBenchmark {
         args(state),
         rand(kSeed),
         indices_have_nulls(indices_have_nulls),
-        monotonic_indices(false) {}
+        monotonic_indices(monotonic_indices) {}
 
   void Int64() {
     const int64_t array_size = args.size / sizeof(int64_t);
@@ -123,10 +123,10 @@ struct TakeBenchmark {
   }
 
   void Bench(const std::shared_ptr<Array>& values) {
-    bool indices_null_proportion = indices_have_nulls ? args.null_proportion : 0;
+    double indices_null_proportion = indices_have_nulls ? args.null_proportion : 0;
     auto indices =
-        rand.Int32(static_cast<int32_t>(values->length()), 0,
-                   static_cast<int32_t>(values->length() - 1), indices_null_proportion);
+        rand.Int32(values->length(), 0, static_cast<int32_t>(values->length() - 1),
+                   indices_null_proportion);
 
     if (monotonic_indices) {
       auto arg_sorter = *SortToIndices(*indices);
@@ -277,12 +277,12 @@ void TakeSetArgs(benchmark::internal::Benchmark* bench) {
 
 BENCHMARK(TakeInt64RandomIndicesNoNulls)->Apply(TakeSetArgs);
 BENCHMARK(TakeInt64RandomIndicesWithNulls)->Apply(TakeSetArgs);
+BENCHMARK(TakeInt64MonotonicIndices)->Apply(TakeSetArgs);
 BENCHMARK(TakeFSLInt64RandomIndicesNoNulls)->Apply(TakeSetArgs);
 BENCHMARK(TakeFSLInt64RandomIndicesWithNulls)->Apply(TakeSetArgs);
+BENCHMARK(TakeFSLInt64MonotonicIndices)->Apply(TakeSetArgs);
 BENCHMARK(TakeStringRandomIndicesNoNulls)->Apply(TakeSetArgs);
 BENCHMARK(TakeStringRandomIndicesWithNulls)->Apply(TakeSetArgs);
-BENCHMARK(TakeInt64MonotonicIndices)->Apply(TakeSetArgs);
-BENCHMARK(TakeFSLInt64MonotonicIndices)->Apply(TakeSetArgs);
 BENCHMARK(TakeStringMonotonicIndices)->Apply(TakeSetArgs);
 
 }  // namespace compute
diff --git a/cpp/src/arrow/compute/kernels/vector_take.cc b/cpp/src/arrow/compute/kernels/vector_take.cc
index 1658f965212..8aae6811a2a 100644
--- a/cpp/src/arrow/compute/kernels/vector_take.cc
+++ b/cpp/src/arrow/compute/kernels/vector_take.cc
@@ -15,15 +15,32 @@
 // specific language governing permissions and limitations
 // under the License.
 
+#include <algorithm>
+#include <limits>
+#include <type_traits>
+
 #include "arrow/array/array_base.h"
+#include "arrow/array/builder_primitive.h"
 #include "arrow/array/concatenate.h"
+#include "arrow/array/data.h"
+#include "arrow/buffer_builder.h"
 #include "arrow/compute/api_vector.h"
 #include "arrow/compute/kernels/common.h"
-#include "arrow/compute/kernels/vector_selection_internal.h"
 #include "arrow/record_batch.h"
 #include "arrow/result.h"
+#include "arrow/util/bit_block_counter.h"
+#include "arrow/util/bitmap_reader.h"
+#include "arrow/util/int_util.h"
 
 namespace arrow {
+
+using internal::BitBlockCount;
+using internal::BitmapReader;
+using internal::GetArrayView;
+using internal::IndexBoundscheck;
+using internal::OptionalBitBlockCounter;
+using internal::OptionalBitIndexer;
+
 namespace compute {
 namespace internal {
 
@@ -38,44 +55,715 @@ std::unique_ptr<KernelState> InitTake(KernelContext*, const KernelInitArgs& args
   return std::unique_ptr<KernelState>(new TakeState{*take_options});
 }
 
-template <typename ValueType, typename IndexType>
-struct TakeFunctor {
-  using ValueArrayType = typename TypeTraits<ValueType>::ArrayType;
-  using IndexArrayType = typename TypeTraits<IndexType>::ArrayType;
-  using IS = ArrayIndexSequence<IndexType>;
+namespace {}  // namespace
+
+// ----------------------------------------------------------------------
+// Implement optimized take for primitive types from boolean to 1/2/4/8-byte
+// C-type based types. Use common implementation for every byte width and only
+// generate code for unsigned integer indices, since after boundschecking to
+// check for negative numbers the indices we can safely reinterpret_cast signed
+// integers as unsigned.
+
+struct PrimitiveTakeArgs {
+  const uint8_t* values;
+  const uint8_t* values_bitmap = nullptr;
+  int values_bit_width;
+  int64_t values_length;
+  int64_t values_offset;
+  int64_t values_null_count;
+  const uint8_t* indices;
+  const uint8_t* indices_bitmap = nullptr;
+  int indices_bit_width;
+  int64_t indices_length;
+  int64_t indices_offset;
+  int64_t indices_null_count;
+};
+
+// Reduce code size by dealing with the unboxing of the kernel inputs once
+// rather than duplicating compiled code to do all these in each kernel.
+PrimitiveTakeArgs GetPrimitiveTakeArgs(const ExecBatch& batch) {
+  PrimitiveTakeArgs args;
+
+  const ArrayData& arg0 = *batch[0].array();
+  const ArrayData& arg1 = *batch[1].array();
+
+  // Values
+  args.values_bit_width = static_cast<const FixedWidthType&>(*arg0.type).bit_width();
+  args.values = arg0.buffers[1]->data();
+  if (args.values_bit_width > 1) {
+    args.values += arg0.offset * args.values_bit_width / 8;
+  }
+  args.values_length = arg0.length;
+  args.values_offset = arg0.offset;
+  args.values_null_count = arg0.GetNullCount();
+  if (arg0.buffers[0]) {
+    args.values_bitmap = arg0.buffers[0]->data();
+  }
+
+  // Indices
+  args.indices_bit_width = static_cast<const FixedWidthType&>(*arg1.type).bit_width();
+  args.indices = arg1.buffers[1]->data() + arg1.offset * args.indices_bit_width / 8;
+  args.indices_length = arg1.length;
+  args.indices_offset = arg1.offset;
+  args.indices_null_count = arg1.GetNullCount();
+  if (arg1.buffers[0]) {
+    args.indices_bitmap = arg1.buffers[0]->data();
+  }
+
+  return args;
+}
+
+/// \brief The Take implementation for primitive (fixed-width) types does not
+/// use the logical Arrow type but rather then physical C type. This way we
+/// only generate one take function for each byte width.
+///
+/// This function assumes that the indices have been boundschecked.
+template <typename IndexCType, typename ValueCType>
+struct PrimitiveTakeImpl {
+  static void Exec(const PrimitiveTakeArgs& args, Datum* out_datum) {
+    auto values = reinterpret_cast<const ValueCType*>(args.values);
+    auto values_bitmap = args.values_bitmap;
+    auto values_offset = args.values_offset;
+
+    auto indices = reinterpret_cast<const IndexCType*>(args.indices);
+    auto indices_bitmap = args.indices_bitmap;
+    auto indices_offset = args.indices_offset;
+
+    ArrayData* out_arr = out_datum->mutable_array();
+    auto out = out_arr->GetMutableValues<ValueCType>(1);
+    auto out_bitmap = out_arr->buffers[0]->mutable_data();
+    auto out_offset = out_arr->offset;
+
+    // If either the values or indices have nulls, we preemptively zero out the
+    // out validity bitmap so that we don't have to use ClearBit in each
+    // iteration for nulls.
+    if (args.values_null_count > 0 || args.indices_null_count > 0) {
+      BitUtil::SetBitsTo(out_bitmap, out_offset, args.indices_length, false);
+    }
+
+    OptionalBitBlockCounter indices_bit_counter(indices_bitmap, indices_offset,
+                                                args.indices_length);
+    int64_t position = 0;
+    int64_t valid_count = 0;
+    while (true) {
+      BitBlockCount block = indices_bit_counter.NextBlock();
+      if (block.length == 0) {
+        // All indices processed.
+        break;
+      }
+      if (args.values_null_count == 0) {
+        // Values are never null, so things are easier
+        valid_count += block.popcount;
+        if (block.popcount == block.length) {
+          // Fastest path: neither values nor index nulls
+          BitUtil::SetBitsTo(out_bitmap, out_offset + position, block.length, true);
+          for (int64_t i = 0; i < block.length; ++i) {
+            out[position] = values[indices[position]];
+            ++position;
+          }
+        } else if (block.popcount > 0) {
+          // Slow path: some indices but not all are null
+          for (int64_t i = 0; i < block.length; ++i) {
+            if (BitUtil::GetBit(indices_bitmap, indices_offset + position)) {
+              // index is not null
+              BitUtil::SetBit(out_bitmap, out_offset + position);
+              out[position] = values[indices[position]];
+            }
+            ++position;
+          }
+        }
+      } else {
+        // Values have nulls, so we must do random access into the values bitmap
+        if (block.popcount == block.length) {
+          // Faster path: indices are not null but values may be
+          for (int64_t i = 0; i < block.length; ++i) {
+            if (BitUtil::GetBit(values_bitmap, values_offset + indices[position])) {
+              // value is not null
+              out[position] = values[indices[position]];
+              BitUtil::SetBit(out_bitmap, out_offset + position);
+              ++valid_count;
+            }
+            ++position;
+          }
+        } else if (block.popcount > 0) {
+          // Slow path: some but not all indices are null. Since we are doing
+          // random access in general we have to check the value nullness one by
+          // one.
+          for (int64_t i = 0; i < block.length; ++i) {
+            if (BitUtil::GetBit(indices_bitmap, indices_offset + position)) {
+              // index is not null
+              if (BitUtil::GetBit(values_bitmap, values_offset + indices[position])) {
+                // value is not null
+                out[position] = values[indices[position]];
+                BitUtil::SetBit(out_bitmap, out_offset + position);
+                ++valid_count;
+              }
+            }
+            ++position;
+          }
+        }
+      }
+    }
+    out_arr->null_count = out_arr->length - valid_count;
+  }
+};
+
+template <typename IndexCType>
+struct BooleanTakeImpl {
+  static void Exec(const PrimitiveTakeArgs& args, Datum* out_datum) {
+    auto values = args.values;
+    auto values_bitmap = args.values_bitmap;
+    auto values_offset = args.values_offset;
+
+    auto indices = reinterpret_cast<const IndexCType*>(args.indices);
+    auto indices_bitmap = args.indices_bitmap;
+    auto indices_offset = args.indices_offset;
+
+    ArrayData* out_arr = out_datum->mutable_array();
+    auto out = out_arr->buffers[1]->mutable_data();
+    auto out_bitmap = out_arr->buffers[0]->mutable_data();
+    auto out_offset = out_arr->offset;
 
-  static void Exec(KernelContext* ctx, const ExecBatch& batch, Datum* out) {
-    ValueArrayType values(batch[0].array());
-    IndexArrayType indices(batch[1].array());
-    std::shared_ptr<Array> result;
-    KERNEL_RETURN_IF_ERROR(ctx, Select(ctx, values, IS(indices), &result));
-    out->value = result->data();
+    // If either the values or indices have nulls, we preemptively zero out the
+    // out validity bitmap so that we don't have to use ClearBit in each
+    // iteration for nulls.
+    if (args.values_null_count > 0 || args.indices_null_count > 0) {
+      BitUtil::SetBitsTo(out_bitmap, out_offset, args.indices_length, false);
+    }
+
+    auto PlaceDataBit = [&](int64_t loc, IndexCType index) {
+      BitUtil::SetBitTo(out, out_offset + loc,
+                        BitUtil::GetBit(values, values_offset + index));
+    };
+
+    OptionalBitBlockCounter indices_bit_counter(indices_bitmap, indices_offset,
+                                                args.indices_length);
+    int64_t position = 0;
+    int64_t valid_count = 0;
+    while (true) {
+      BitBlockCount block = indices_bit_counter.NextBlock();
+      if (block.length == 0) {
+        // All indices processed.
+        break;
+      }
+      if (args.values_null_count == 0) {
+        // Values are never null, so things are easier
+        valid_count += block.popcount;
+        if (block.popcount == block.length) {
+          // Fastest path: neither values nor index nulls
+          BitUtil::SetBitsTo(out_bitmap, out_offset + position, block.length, true);
+          for (int64_t i = 0; i < block.length; ++i) {
+            PlaceDataBit(position, indices[position]);
+            ++position;
+          }
+        } else if (block.popcount > 0) {
+          // Slow path: some but not all indices are null
+          for (int64_t i = 0; i < block.length; ++i) {
+            if (BitUtil::GetBit(indices_bitmap, indices_offset + position)) {
+              // index is not null
+              BitUtil::SetBit(out_bitmap, out_offset + position);
+              PlaceDataBit(position, indices[position]);
+            }
+            ++position;
+          }
+        }
+      } else {
+        // Values have nulls, so we must do random access into the values bitmap
+        if (block.popcount == block.length) {
+          // Faster path: indices are not null but values may be
+          for (int64_t i = 0; i < block.length; ++i) {
+            if (BitUtil::GetBit(values_bitmap, values_offset + indices[position])) {
+              // value is not null
+              BitUtil::SetBit(out_bitmap, out_offset + position);
+              PlaceDataBit(position, indices[position]);
+              ++valid_count;
+            }
+            ++position;
+          }
+        } else if (block.popcount > 0) {
+          // Slow path: some but not all indices are null. Since we are doing
+          // random access in general we have to check the value nullness one by
+          // one.
+          for (int64_t i = 0; i < block.length; ++i) {
+            if (BitUtil::GetBit(indices_bitmap, indices_offset + position)) {
+              // index is not null
+              if (BitUtil::GetBit(values_bitmap, values_offset + indices[position])) {
+                // value is not null
+                PlaceDataBit(position, indices[position]);
+                BitUtil::SetBit(out_bitmap, out_offset + position);
+                ++valid_count;
+              }
+            }
+            ++position;
+          }
+        }
+      }
+    }
+    out_arr->null_count = out_arr->length - valid_count;
   }
 };
 
-struct TakeKernelVisitor {
-  TakeKernelVisitor(const DataType& value_type, const DataType& index_type)
-      : value_type(value_type), index_type(index_type) {}
+template <template <typename...> class TakeImpl, typename... Args>
+void TakeIndexDispatch(const PrimitiveTakeArgs& args, Datum* out) {
+  // With the simplifying assumption that boundschecking has taken place
+  // already at a higher level, we can now assume that the index values are all
+  // non-negative. Thus, we can interpret signed integers as unsigned and avoid
+  // having to generate double the amount of binary code to handle each integer
+  // width.
+  switch (args.indices_bit_width) {
+    case 8:
+      return TakeImpl<uint8_t, Args...>::Exec(args, out);
+    case 16:
+      return TakeImpl<uint16_t, Args...>::Exec(args, out);
+    case 32:
+      return TakeImpl<uint32_t, Args...>::Exec(args, out);
+    case 64:
+      return TakeImpl<uint64_t, Args...>::Exec(args, out);
+    default:
+      DCHECK(false) << "Invalid indices byte width";
+      break;
+  }
+}
+
+Status PreallocateTake(KernelContext* ctx, int64_t length, int bit_width, Datum* out) {
+  // Preallocate memory
+  ArrayData* out_arr = out->mutable_array();
+  out_arr->length = length;
+  out_arr->buffers.resize(2);
 
-  template <typename Type>
-  Status Visit(const Type&) {
-    this->result = codegen::Integer<TakeFunctor, Type>(index_type.id());
+  ARROW_ASSIGN_OR_RAISE(out_arr->buffers[0], ctx->AllocateBitmap(length));
+  if (bit_width == 1) {
+    ARROW_ASSIGN_OR_RAISE(out_arr->buffers[1], ctx->AllocateBitmap(length));
+  } else {
+    ARROW_ASSIGN_OR_RAISE(out_arr->buffers[1], ctx->Allocate(length * bit_width / 8));
+  }
+  return Status::OK();
+}
+
+static void PrimitiveTakeExec(KernelContext* ctx, const ExecBatch& batch, Datum* out) {
+  const auto& state = checked_cast<const TakeState&>(*ctx->state());
+  if (state.options.boundscheck) {
+    KERNEL_RETURN_IF_ERROR(ctx, IndexBoundscheck(*batch[1].array(), batch[0].length()));
+  }
+  PrimitiveTakeArgs args = GetPrimitiveTakeArgs(batch);
+  KERNEL_RETURN_IF_ERROR(
+      ctx, PreallocateTake(ctx, args.indices_length, args.values_bit_width, out));
+  switch (args.values_bit_width) {
+    case 1:
+      return TakeIndexDispatch<BooleanTakeImpl>(args, out);
+    case 8:
+      return TakeIndexDispatch<PrimitiveTakeImpl, int8_t>(args, out);
+    case 16:
+      return TakeIndexDispatch<PrimitiveTakeImpl, int16_t>(args, out);
+    case 32:
+      return TakeIndexDispatch<PrimitiveTakeImpl, int32_t>(args, out);
+    case 64:
+      return TakeIndexDispatch<PrimitiveTakeImpl, int64_t>(args, out);
+    default:
+      DCHECK(false) << "Invalid values byte width";
+      break;
+  }
+}
+
+static void NullTakeExec(KernelContext* ctx, const ExecBatch& batch, Datum* out) {
+  const auto& state = checked_cast<const TakeState&>(*ctx->state());
+  if (state.options.boundscheck) {
+    KERNEL_RETURN_IF_ERROR(ctx, IndexBoundscheck(*batch[1].array(), batch[0].length()));
+  }
+  out->value = std::make_shared<NullArray>(batch.length)->data();
+}
+
+static void DictionaryTakeExec(KernelContext* ctx, const ExecBatch& batch, Datum* out) {
+  const auto& state = checked_cast<const TakeState&>(*ctx->state());
+  DictionaryArray values(batch[0].array());
+  Result<Datum> result =
+      Take(Datum(values.indices()), batch[1], state.options, ctx->exec_context());
+  if (!result.ok()) {
+    ctx->SetStatus(result.status());
+    return;
+  }
+  DictionaryArray taken_values(values.type(), (*result).make_array(),
+                               values.dictionary());
+  out->value = taken_values.data();
+}
+
+static void ExtensionTakeExec(KernelContext* ctx, const ExecBatch& batch, Datum* out) {
+  const auto& state = checked_cast<const TakeState&>(*ctx->state());
+  ExtensionArray values(batch[0].array());
+  Result<Datum> result =
+      Take(Datum(values.storage()), batch[1], state.options, ctx->exec_context());
+  if (!result.ok()) {
+    ctx->SetStatus(result.status());
+    return;
+  }
+
+  ExtensionArray taken_values(values.type(), (*result).make_array());
+  out->value = taken_values.data();
+}
+
+// ----------------------------------------------------------------------
+
+// Use CRTP here
+template <typename Impl, typename Type>
+struct GenericTakeImpl {
+  using ValuesArrayType = typename TypeTraits<Type>::ArrayType;
+
+  KernelContext* ctx;
+  std::shared_ptr<ArrayData> values;
+  std::shared_ptr<ArrayData> indices;
+  ArrayData* out;
+  TypedBufferBuilder<bool> validity_builder;
+
+  GenericTakeImpl(KernelContext* ctx, const ExecBatch& batch, Datum* out)
+      : ctx(ctx),
+        values(batch[0].array()),
+        indices(batch[1].array()),
+        out(out->mutable_array()),
+        validity_builder(ctx->memory_pool()) {}
+
+  Status FinishCommon() {
+    out->buffers.resize(values->buffers.size());
+    out->length = validity_builder.length();
+    out->null_count = validity_builder.false_count();
+    return validity_builder.Finish(&out->buffers[0]);
+  }
+
+  template <typename IndexCType, typename ValidVisitor, typename NullVisitor>
+  Status VisitIndices(const ArrayData& indices, ValidVisitor&& visit_valid,
+                      NullVisitor&& visit_null) {
+    const auto indices_values = indices.GetValues<IndexCType>(1);
+    const uint8_t* bitmap = nullptr;
+    if (indices.buffers[0]) {
+      bitmap = indices.buffers[0]->data();
+    }
+    OptionalBitIndexer values_is_valid(values->buffers[0], values->offset);
+
+    OptionalBitBlockCounter bit_counter(bitmap, indices.offset, indices.length);
+    int64_t position = 0;
+    while (position < indices.length) {
+      BitBlockCount block = bit_counter.NextBlock();
+      if (block.popcount == block.length) {
+        // Fast path, no null indices
+        for (int64_t i = 0; i < block.length; ++i) {
+          if (values_is_valid[indices_values[position]]) {
+            validity_builder.UnsafeAppend(true);
+            RETURN_NOT_OK(visit_valid(indices_values[position]));
+          } else {
+            validity_builder.UnsafeAppend(false);
+            RETURN_NOT_OK(visit_null());
+          }
+          ++position;
+        }
+      } else {
+        for (int64_t i = 0; i < block.length; ++i) {
+          if (BitUtil::GetBit(bitmap, indices.offset + position) &&
+              values_is_valid[indices_values[position]]) {
+            validity_builder.UnsafeAppend(true);
+            RETURN_NOT_OK(visit_valid(indices_values[position]));
+          } else {
+            validity_builder.UnsafeAppend(false);
+            RETURN_NOT_OK(visit_null());
+          }
+          ++position;
+        }
+      }
+    }
     return Status::OK();
   }
 
-  Status Create() { return VisitTypeInline(value_type, this); }
+  virtual Status Init() { return Status::OK(); }
 
-  const DataType& value_type;
-  const DataType& index_type;
-  ArrayKernelExec result;
+  // Implementation specific finish logic
+  virtual Status Finish() = 0;
+
+  virtual Status Exec() {
+    RETURN_NOT_OK(this->validity_builder.Reserve(indices->length));
+    RETURN_NOT_OK(Init());
+    int index_width =
+        static_cast<const FixedWidthType&>(*this->indices->type).bit_width() / 8;
+    switch (index_width) {
+      case 1:
+        RETURN_NOT_OK(static_cast<Impl*>(this)->template ProcessIndices<uint8_t>());
+        break;
+      case 2:
+        RETURN_NOT_OK(static_cast<Impl*>(this)->template ProcessIndices<uint16_t>());
+        break;
+      case 4:
+        RETURN_NOT_OK(static_cast<Impl*>(this)->template ProcessIndices<uint32_t>());
+        break;
+      case 8:
+        RETURN_NOT_OK(static_cast<Impl*>(this)->template ProcessIndices<uint64_t>());
+        break;
+      default:
+        DCHECK(false) << "Invalid index width";
+        break;
+    }
+    RETURN_NOT_OK(this->FinishCommon());
+    return Finish();
+  }
 };
 
-Status GetTakeKernel(const DataType& value_type, const DataType& index_type,
-                     ArrayKernelExec* exec) {
-  TakeKernelVisitor visitor(value_type, index_type);
-  RETURN_NOT_OK(visitor.Create());
-  *exec = visitor.result;
-  return Status::OK();
+#define LIFT_BASE_MEMBERS()                               \
+  using ValuesArrayType = typename Base::ValuesArrayType; \
+  using Base::ctx;                                        \
+  using Base::values;                                     \
+  using Base::indices;                                    \
+  using Base::out;                                        \
+  using Base::validity_builder
+
+static inline Status VisitNoop() { return Status::OK(); }
+
+// A take implementation for 32-bit and 64-bit variable binary types. Common
+// generated kernels are shared between Binary/String and
+// LargeBinary/LargeString
+template <typename Type>
+struct VarBinaryTakeImpl : public GenericTakeImpl<VarBinaryTakeImpl<Type>, Type> {
+  using offset_type = typename Type::offset_type;
+
+  using Base = GenericTakeImpl<VarBinaryTakeImpl<Type>, Type>;
+  LIFT_BASE_MEMBERS();
+
+  std::shared_ptr<ArrayData> values_as_binary;
+  TypedBufferBuilder<offset_type> offset_builder;
+  TypedBufferBuilder<uint8_t> data_builder;
+
+  static constexpr int64_t kOffsetLimit = std::numeric_limits<offset_type>::max() - 1;
+
+  VarBinaryTakeImpl(KernelContext* ctx, const ExecBatch& batch, Datum* out)
+      : Base(ctx, batch, out),
+        offset_builder(ctx->memory_pool()),
+        data_builder(ctx->memory_pool()) {}
+
+  template <typename IndexCType>
+  Status ProcessIndices() {
+    ValuesArrayType typed_values(this->values_as_binary);
+
+    // Allocate at least 32K at a time to avoid having to call Reserve for
+    // every value for lots of small strings
+    static constexpr int64_t kAllocateChunksize = 1 << 15;
+    RETURN_NOT_OK(data_builder.Reserve(kAllocateChunksize));
+    int64_t space_available = data_builder.capacity();
+
+    offset_type offset = 0;
+    RETURN_NOT_OK(this->template VisitIndices<IndexCType>(
+        *indices,
+        [&](IndexCType index) {
+          offset_builder.UnsafeAppend(offset);
+          auto val = typed_values.GetView(index);
+          offset_type value_size = static_cast<offset_type>(val.size());
+          if (ARROW_PREDICT_FALSE(static_cast<int64_t>(offset) +
+                                  static_cast<int64_t>(value_size)) > kOffsetLimit) {
+            return Status::Invalid("Take operation overflowed binary array capacity");
+          }
+          offset += value_size;
+          if (ARROW_PREDICT_FALSE(value_size > space_available)) {
+            RETURN_NOT_OK(data_builder.Reserve(value_size + kAllocateChunksize));
+            space_available = data_builder.capacity() - data_builder.length();
+          }
+          data_builder.UnsafeAppend(reinterpret_cast<const uint8_t*>(val.data()),
+                                    value_size);
+          space_available -= value_size;
+          return Status::OK();
+        },
+        [&]() {
+          offset_builder.UnsafeAppend(offset);
+          return Status::OK();
+        }));
+    offset_builder.UnsafeAppend(offset);
+    return Status::OK();
+  }
+
+  Status Init() override {
+    ARROW_ASSIGN_OR_RAISE(this->values_as_binary,
+                          GetArrayView(this->values, TypeTraits<Type>::type_singleton()));
+    return offset_builder.Reserve(indices->length + 1);
+  }
+
+  Status Finish() override {
+    RETURN_NOT_OK(offset_builder.Finish(&out->buffers[1]));
+    return data_builder.Finish(&out->buffers[2]);
+  }
+};
+
+struct FSBTakeImpl : public GenericTakeImpl<FSBTakeImpl, FixedSizeBinaryType> {
+  using Base = GenericTakeImpl<FSBTakeImpl, FixedSizeBinaryType>;
+  LIFT_BASE_MEMBERS();
+
+  TypedBufferBuilder<uint8_t> data_builder;
+
+  FSBTakeImpl(KernelContext* ctx, const ExecBatch& batch, Datum* out)
+      : Base(ctx, batch, out), data_builder(ctx->memory_pool()) {}
+
+  template <typename IndexCType>
+  Status ProcessIndices() {
+    FixedSizeBinaryArray typed_values(this->values);
+    int32_t value_size = typed_values.byte_width();
+
+    RETURN_NOT_OK(data_builder.Reserve(value_size * indices->length));
+    RETURN_NOT_OK(this->template VisitIndices<IndexCType>(
+        *indices,
+        [&](IndexCType index) {
+          auto val = typed_values.GetView(index);
+          data_builder.UnsafeAppend(reinterpret_cast<const uint8_t*>(val.data()),
+                                    value_size);
+          return Status::OK();
+        },
+        [&]() {
+          data_builder.UnsafeAppend(value_size, static_cast<uint8_t>(0x00));
+          return Status::OK();
+        }));
+    return Status::OK();
+  }
+
+  Status Finish() override { return data_builder.Finish(&out->buffers[1]); }
+};
+
+template <typename Type>
+struct ListTakeImpl : public GenericTakeImpl<ListTakeImpl<Type>, Type> {
+  using offset_type = typename Type::offset_type;
+
+  using Base = GenericTakeImpl<ListTakeImpl<Type>, Type>;
+  LIFT_BASE_MEMBERS();
+
+  TypedBufferBuilder<offset_type> offset_builder;
+  typename TypeTraits<Type>::OffsetBuilderType child_index_builder;
+
+  ListTakeImpl(KernelContext* ctx, const ExecBatch& batch, Datum* out)
+      : Base(ctx, batch, out),
+        offset_builder(ctx->memory_pool()),
+        child_index_builder(ctx->memory_pool()) {}
+
+  template <typename IndexCType>
+  Status ProcessIndices() {
+    ValuesArrayType typed_values(this->values);
+
+    offset_type offset = 0;
+    auto PushValidIndex = [&](IndexCType index) {
+      offset_builder.UnsafeAppend(offset);
+      offset_type value_offset = typed_values.value_offset(index);
+      offset_type value_length = typed_values.value_length(index);
+      offset += value_length;
+      RETURN_NOT_OK(child_index_builder.Reserve(value_length));
+      for (offset_type j = value_offset; j < value_offset + value_length; ++j) {
+        child_index_builder.UnsafeAppend(j);
+      }
+      return Status::OK();
+    };
+
+    auto PushNullIndex = [&]() {
+      offset_builder.UnsafeAppend(offset);
+      return Status::OK();
+    };
+
+    RETURN_NOT_OK(this->template VisitIndices<IndexCType>(
+        *indices, std::move(PushValidIndex), std::move(PushNullIndex)));
+    offset_builder.UnsafeAppend(offset);
+    return Status::OK();
+  }
+
+  Status Init() override {
+    RETURN_NOT_OK(offset_builder.Reserve(indices->length + 1));
+    return Status::OK();
+  }
+
+  Status Finish() override {
+    std::shared_ptr<Array> child_indices;
+    RETURN_NOT_OK(child_index_builder.Finish(&child_indices));
+
+    ValuesArrayType typed_values(this->values);
+
+    // No need to boundscheck the child values indices
+    ARROW_ASSIGN_OR_RAISE(std::shared_ptr<Array> taken_child,
+                          Take(*typed_values.values(), *child_indices,
+                               TakeOptions::NoBoundscheck(), ctx->exec_context()));
+    RETURN_NOT_OK(offset_builder.Finish(&out->buffers[1]));
+    out->child_data = {taken_child->data()};
+    return Status::OK();
+  }
+};
+
+struct FSLTakeImpl : public GenericTakeImpl<FSLTakeImpl, FixedSizeListType> {
+  Int64Builder child_index_builder;
+
+  using Base = GenericTakeImpl<FSLTakeImpl, FixedSizeListType>;
+  LIFT_BASE_MEMBERS();
+
+  FSLTakeImpl(KernelContext* ctx, const ExecBatch& batch, Datum* out)
+      : Base(ctx, batch, out), child_index_builder(ctx->memory_pool()) {}
+
+  template <typename IndexCType>
+  Status ProcessIndices() {
+    ValuesArrayType typed_values(this->values);
+    int32_t list_size = typed_values.list_type()->list_size();
+
+    /// We must take list_size elements even for null elements of
+    /// indices.
+    RETURN_NOT_OK(child_index_builder.Reserve(indices->length * list_size));
+    return this->template VisitIndices<IndexCType>(
+        *indices,
+        [&](IndexCType index) {
+          int64_t offset = index * list_size;
+          for (int64_t j = offset; j < offset + list_size; ++j) {
+            child_index_builder.UnsafeAppend(j);
+          }
+          return Status::OK();
+        },
+        [&]() { return child_index_builder.AppendNulls(list_size); });
+  }
+
+  Status Finish() override {
+    std::shared_ptr<Array> child_indices;
+    RETURN_NOT_OK(child_index_builder.Finish(&child_indices));
+
+    ValuesArrayType typed_values(this->values);
+
+    // No need to boundscheck the child values indices
+    ARROW_ASSIGN_OR_RAISE(std::shared_ptr<Array> taken_child,
+                          Take(*typed_values.values(), *child_indices,
+                               TakeOptions::NoBoundscheck(), ctx->exec_context()));
+    out->child_data = {taken_child->data()};
+    return Status::OK();
+  }
+};
+
+struct StructTakeImpl : public GenericTakeImpl<StructTakeImpl, StructType> {
+  using Base = GenericTakeImpl<StructTakeImpl, StructType>;
+  LIFT_BASE_MEMBERS();
+
+  StructTakeImpl(KernelContext* ctx, const ExecBatch& batch, Datum* out)
+      : Base(ctx, batch, out) {}
+
+  template <typename IndexCType>
+  Status ProcessIndices() {
+    StructArray typed_values(values);
+    return this->template VisitIndices<IndexCType>(
+        *indices, [&](IndexCType index) { return Status::OK(); },
+        /*visit_null=*/VisitNoop);
+  }
+
+  Status Finish() override {
+    StructArray typed_values(values);
+
+    // Select from children without boundschecking
+    out->child_data.resize(values->type->num_fields());
+    for (int field_index = 0; field_index < values->type->num_fields(); ++field_index) {
+      ARROW_ASSIGN_OR_RAISE(Datum taken_field,
+                            Take(Datum(typed_values.field(field_index)), Datum(indices),
+                                 TakeOptions::NoBoundscheck(), ctx->exec_context()));
+      out->child_data[field_index] = taken_field.array();
+    }
+    return Status::OK();
+  }
+};
+
+template <typename Impl>
+static void GenericTakeExec(KernelContext* ctx, const ExecBatch& batch, Datum* out) {
+  const auto& state = checked_cast<const TakeState&>(*ctx->state());
+  if (state.options.boundscheck) {
+    KERNEL_RETURN_IF_ERROR(ctx, IndexBoundscheck(*batch[1].array(), batch[0].length()));
+  }
+  Impl kernel(ctx, batch, out);
+  KERNEL_RETURN_IF_ERROR(ctx, kernel.Exec());
 }
 
 // Shorthand naming of these functions
@@ -234,32 +922,49 @@ class TakeMetaFunction : public MetaFunction {
   }
 };
 
+static InputType kTakeIndexType(match::Integer(), ValueDescr::ARRAY);
+
 void RegisterVectorTake(FunctionRegistry* registry) {
   VectorKernel base;
   base.init = InitTake;
   base.can_execute_chunkwise = false;
 
   auto array_take = std::make_shared<VectorFunction>("array_take", Arity::Binary());
-  OutputType out_ty(FirstType);
 
-  auto AddKernel = [&](InputType value_ty, const DataType& example_value_ty,
-                       const std::shared_ptr<DataType>& index_ty) {
+  auto AddTakeKernel = [&](InputType value_ty, ArrayKernelExec exec) {
     base.signature =
-        KernelSignature::Make({value_ty, InputType::Array(index_ty)}, out_ty);
-    DCHECK_OK(GetTakeKernel(example_value_ty, *index_ty, &base.exec));
+        KernelSignature::Make({value_ty, kTakeIndexType}, OutputType(FirstType));
+    base.exec = exec;
     DCHECK_OK(array_take->AddKernel(base));
   };
 
-  for (const auto& value_ty : PrimitiveTypes()) {
-    for (const auto& index_ty : IntTypes()) {
-      AddKernel(InputType::Array(value_ty), *value_ty, index_ty);
-    }
-  }
-  for (const auto& value_ty : ExampleParametricTypes()) {
-    for (const auto& index_ty : IntTypes()) {
-      AddKernel(InputType::Array(value_ty->id()), *value_ty, index_ty);
-    }
-  }
+  // Single kernel entry point for all primitive types. We dispatch to take
+  // implementations inside the kernel for now. The primitive take
+  // implementation writes into preallocated memory while the other
+  // implementations handle their own memory allocation.
+  AddTakeKernel(InputType(match::Primitive(), ValueDescr::ARRAY), PrimitiveTakeExec);
+
+  // Take implementations for Binary, String, LargeBinary, LargeString, and
+  // FixedSizeBinary
+  AddTakeKernel(InputType(match::BinaryLike(), ValueDescr::ARRAY),
+                GenericTakeExec<VarBinaryTakeImpl<BinaryType>>);
+  AddTakeKernel(InputType(match::LargeBinaryLike(), ValueDescr::ARRAY),
+                GenericTakeExec<VarBinaryTakeImpl<LargeBinaryType>>);
+  AddTakeKernel(InputType::Array(Type::FIXED_SIZE_BINARY), GenericTakeExec<FSBTakeImpl>);
+
+  AddTakeKernel(InputType::Array(null()), NullTakeExec);
+  AddTakeKernel(InputType::Array(Type::DECIMAL), GenericTakeExec<FSBTakeImpl>);
+  AddTakeKernel(InputType::Array(Type::DICTIONARY), DictionaryTakeExec);
+  AddTakeKernel(InputType::Array(Type::EXTENSION), ExtensionTakeExec);
+  AddTakeKernel(InputType::Array(Type::LIST), GenericTakeExec<ListTakeImpl<ListType>>);
+  AddTakeKernel(InputType::Array(Type::LARGE_LIST),
+                GenericTakeExec<ListTakeImpl<LargeListType>>);
+  AddTakeKernel(InputType::Array(Type::FIXED_SIZE_LIST), GenericTakeExec<FSLTakeImpl>);
+  AddTakeKernel(InputType::Array(Type::STRUCT), GenericTakeExec<StructTakeImpl>);
+
+  // TODO: Reuse ListType kernel for MAP
+  AddTakeKernel(InputType::Array(Type::MAP), GenericTakeExec<ListTakeImpl<MapType>>);
+
   DCHECK_OK(registry->AddFunction(std::move(array_take)));
 
   // Add take metafunction
diff --git a/cpp/src/arrow/compute/kernels/vector_take_test.cc b/cpp/src/arrow/compute/kernels/vector_take_test.cc
index 10c855697c1..c30b87ec060 100644
--- a/cpp/src/arrow/compute/kernels/vector_take_test.cc
+++ b/cpp/src/arrow/compute/kernels/vector_take_test.cc
@@ -41,7 +41,7 @@ void AssertTakeArrays(const std::shared_ptr<Array>& values,
                       const std::shared_ptr<Array>& expected) {
   ASSERT_OK_AND_ASSIGN(std::shared_ptr<Array> actual, Take(*values, *indices));
   ASSERT_OK(actual->ValidateFull());
-  AssertArraysEqual(*expected, *actual);
+  AssertArraysEqual(*expected, *actual, /*verbose=*/true);
 }
 
 Status TakeJSON(const std::shared_ptr<DataType>& type, const std::string& values,
@@ -58,7 +58,7 @@ void CheckTake(const std::shared_ptr<DataType>& type, const std::string& values,
   for (auto index_type : {int8(), uint32()}) {
     ASSERT_OK(TakeJSON(type, values, index_type, indices, &actual));
     ASSERT_OK(actual->ValidateFull());
-    AssertArraysEqual(*ArrayFromJSON(type, expected), *actual);
+    AssertArraysEqual(*ArrayFromJSON(type, expected), *actual, /*verbose=*/true);
   }
 }
 
@@ -72,6 +72,63 @@ void AssertTakeBoolean(const std::string& values, const std::string& indices,
   CheckTake(boolean(), values, indices, expected);
 }
 
+template <typename ValuesType, typename IndexType>
+void ValidateTakeImpl(const std::shared_ptr<Array>& values,
+                      const std::shared_ptr<Array>& indices,
+                      const std::shared_ptr<Array>& result) {
+  using ValuesArrayType = typename TypeTraits<ValuesType>::ArrayType;
+  using IndexArrayType = typename TypeTraits<IndexType>::ArrayType;
+  auto typed_values = checked_pointer_cast<ValuesArrayType>(values);
+  auto typed_result = checked_pointer_cast<ValuesArrayType>(result);
+  auto typed_indices = checked_pointer_cast<IndexArrayType>(indices);
+  for (int64_t i = 0; i < indices->length(); ++i) {
+    if (typed_indices->IsNull(i) || typed_values->IsNull(typed_indices->Value(i))) {
+      ASSERT_TRUE(result->IsNull(i));
+      continue;
+    }
+    ASSERT_EQ(typed_result->GetView(i), typed_values->GetView(typed_indices->Value(i)))
+        << i;
+  }
+}
+
+template <typename ValuesType>
+void ValidateTake(const std::shared_ptr<Array>& values,
+                  const std::shared_ptr<Array>& indices) {
+  ASSERT_OK_AND_ASSIGN(Datum out, Take(values, indices));
+  auto taken = out.make_array();
+  ASSERT_OK(taken->ValidateFull());
+  ASSERT_EQ(indices->length(), taken->length());
+  switch (indices->type_id()) {
+    case Type::INT8:
+      ValidateTakeImpl<ValuesType, Int8Type>(values, indices, taken);
+      break;
+    case Type::INT16:
+      ValidateTakeImpl<ValuesType, Int16Type>(values, indices, taken);
+      break;
+    case Type::INT32:
+      ValidateTakeImpl<ValuesType, Int32Type>(values, indices, taken);
+      break;
+    case Type::INT64:
+      ValidateTakeImpl<ValuesType, Int64Type>(values, indices, taken);
+      break;
+    case Type::UINT8:
+      ValidateTakeImpl<ValuesType, UInt8Type>(values, indices, taken);
+      break;
+    case Type::UINT16:
+      ValidateTakeImpl<ValuesType, UInt16Type>(values, indices, taken);
+      break;
+    case Type::UINT32:
+      ValidateTakeImpl<ValuesType, UInt32Type>(values, indices, taken);
+      break;
+    case Type::UINT64:
+      ValidateTakeImpl<ValuesType, UInt64Type>(values, indices, taken);
+      break;
+    default:
+      FAIL() << "Invalid index type";
+      break;
+  }
+}
+
 template <typename ArrowType>
 class TestTakeKernel : public ::testing::Test {};
 
@@ -115,25 +172,6 @@ class TestTakeKernelWithNumeric : public TestTakeKernel<ArrowType> {
   std::shared_ptr<DataType> type_singleton() {
     return TypeTraits<ArrowType>::type_singleton();
   }
-
-  void ValidateTake(const std::shared_ptr<Array>& values,
-                    const std::shared_ptr<Array>& indices_boxed) {
-    ASSERT_OK_AND_ASSIGN(Datum out, Take(values, indices_boxed));
-    auto taken = out.make_array();
-    ASSERT_OK(taken->ValidateFull());
-    ASSERT_EQ(indices_boxed->length(), taken->length());
-
-    ASSERT_EQ(indices_boxed->type_id(), Type::INT32);
-    auto indices = checked_pointer_cast<Int32Array>(indices_boxed);
-    for (int64_t i = 0; i < indices->length(); ++i) {
-      if (indices->IsNull(i)) {
-        ASSERT_TRUE(taken->IsNull(i));
-        continue;
-      }
-      int32_t taken_index = indices->Value(i);
-      ASSERT_TRUE(values->RangeEquals(taken_index, taken_index + 1, i, taken));
-    }
-  }
 };
 
 TYPED_TEST_SUITE(TestTakeKernelWithNumeric, NumericArrowTypes);
@@ -161,8 +199,8 @@ TYPED_TEST(TestTakeKernelWithNumeric, TakeRandomNumeric) {
       for (auto null_probability : {0.0, 0.01, 0.25, 1.0}) {
         auto values = rand.Numeric<TypeParam>(length, 0, 127, null_probability);
         auto max_index = static_cast<int32_t>(length - 1);
-        auto filter = rand.Int32(indices_length, 0, max_index, null_probability);
-        this->ValidateTake(values, filter);
+        auto indices = rand.Int32(indices_length, 0, max_index, null_probability);
+        ValidateTake<TypeParam>(values, indices);
       }
     }
   }
@@ -179,6 +217,7 @@ class TestTakeKernelWithString : public TestTakeKernel<TypeClass> {
                   const std::string& expected) {
     CheckTake(value_type(), values, indices, expected);
   }
+
   void AssertTakeDictionary(const std::string& dictionary_values,
                             const std::string& dictionary_indices,
                             const std::string& indices,
@@ -211,6 +250,50 @@ TYPED_TEST(TestTakeKernelWithString, TakeString) {
                                      "[2, 5]", &arr));
 }
 
+TEST(TestTakeKernelString, Random) {
+  auto rand = random::RandomArrayGenerator(kRandomSeed);
+  for (size_t i = 3; i < 8; i++) {
+    const int64_t length = static_cast<int64_t>(1ULL << i);
+    for (size_t j = 0; j < 13; j++) {
+      const int64_t indices_length = static_cast<int64_t>(1ULL << j);
+      for (auto null_probability : {0.0, 0.01, 0.25, 1.0}) {
+        auto values = rand.String(length, 0, 32, null_probability);
+        auto max_index = static_cast<int32_t>(length - 1);
+        auto indices = rand.Int32(indices_length, 0, max_index, null_probability);
+        ValidateTake<StringType>(values, indices);
+      }
+    }
+  }
+}
+
+TEST(TestTakeKernelFixedSizeBinary, Random) {
+  auto rand = random::RandomArrayGenerator(kRandomSeed);
+
+  const int32_t value_size = 16;
+  for (size_t i = 3; i < 8; i++) {
+    const int64_t length = static_cast<int64_t>(1ULL << i);
+    for (size_t j = 0; j < 13; j++) {
+      const int64_t indices_length = static_cast<int64_t>(1ULL << j);
+      for (auto null_probability : {0.0, 0.01, 0.25, 1.0}) {
+        int64_t data_nbytes = length * value_size;
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<Buffer> data, AllocateBuffer(data_nbytes));
+        random_bytes(data_nbytes, /*seed=*/0, data->mutable_data());
+        auto validity = rand.Boolean(length, 1 - null_probability);
+
+        auto max_index = static_cast<int32_t>(length - 1);
+        auto indices = rand.Int32(indices_length, 0, max_index, null_probability);
+
+        // Assemble the data for a FixedSizeBinaryArray
+        auto values_data =
+            std::make_shared<ArrayData>(fixed_size_binary(value_size), length);
+        values_data->buffers = {validity->data()->buffers[1], data};
+
+        ValidateTake<FixedSizeBinaryType>(MakeArray(values_data), indices);
+      }
+    }
+  }
+}
+
 TYPED_TEST(TestTakeKernelWithString, TakeDictionary) {
   auto dict = R"(["a", "b", "c", "d", "e"])";
   this->AssertTakeDictionary(dict, "[3, 4, 2]", "[0, 1, 0]", "[3, 4, 3]");
@@ -218,6 +301,29 @@ TYPED_TEST(TestTakeKernelWithString, TakeDictionary) {
   this->AssertTakeDictionary(dict, "[3, 4, 2]", "[null, 1, 0]", "[null, 4, 3]");
 }
 
+class TestTakeKernelFSB : public TestTakeKernel<FixedSizeBinaryType> {
+ public:
+  std::shared_ptr<DataType> value_type() { return fixed_size_binary(3); }
+
+  void AssertTake(const std::string& values, const std::string& indices,
+                  const std::string& expected) {
+    CheckTake(value_type(), values, indices, expected);
+  }
+};
+
+TEST_F(TestTakeKernelFSB, TakeFixedSizeBinary) {
+  this->AssertTake(R"(["aaa", "bbb", "ccc"])", "[0, 1, 0]", R"(["aaa", "bbb", "aaa"])");
+  this->AssertTake(R"([null, "bbb", "ccc"])", "[0, 1, 0]", "[null, \"bbb\", null]");
+  this->AssertTake(R"(["aaa", "bbb", "ccc"])", "[null, 1, 0]", R"([null, "bbb", "aaa"])");
+
+  std::shared_ptr<DataType> type = this->value_type();
+  std::shared_ptr<Array> arr;
+  ASSERT_RAISES(IndexError,
+                TakeJSON(type, R"(["aaa", "bbb", "ccc"])", int8(), "[0, 9, 0]", &arr));
+  ASSERT_RAISES(IndexError, TakeJSON(type, R"(["aaa", "bbb", null, "ddd", "eee"])",
+                                     int64(), "[2, 5]", &arr));
+}
+
 class TestTakeKernelWithList : public TestTakeKernel<ListType> {};
 
 TEST_F(TestTakeKernelWithList, TakeListInt32) {
@@ -356,7 +462,9 @@ TEST_F(TestTakeKernelWithStruct, TakeStruct) {
 
 class TestTakeKernelWithUnion : public TestTakeKernel<UnionType> {};
 
-TEST_F(TestTakeKernelWithUnion, TakeUnion) {
+// TODO: Restore Union take functionality
+
+TEST_F(TestTakeKernelWithUnion, DISABLED_TakeUnion) {
   for (auto mode : {UnionMode::SPARSE, UnionMode::DENSE}) {
     auto union_type = union_({field("a", int32()), field("b", utf8())}, {2, 5}, mode);
     auto union_json = R"([
diff --git a/cpp/src/arrow/type_traits.h b/cpp/src/arrow/type_traits.h
index 937443b3012..d59af0dfe90 100644
--- a/cpp/src/arrow/type_traits.h
+++ b/cpp/src/arrow/type_traits.h
@@ -696,7 +696,6 @@ static inline bool is_floating(Type::type type_id) {
 
 static inline bool is_primitive(Type::type type_id) {
   switch (type_id) {
-    case Type::NA:
     case Type::BOOL:
     case Type::UINT8:
     case Type::INT8:
diff --git a/cpp/src/arrow/util/bit_block_counter.cc b/cpp/src/arrow/util/bit_block_counter.cc
index ca5c159350c..fd94b95a306 100644
--- a/cpp/src/arrow/util/bit_block_counter.cc
+++ b/cpp/src/arrow/util/bit_block_counter.cc
@@ -112,6 +112,18 @@ BitBlockCount BitBlockCounter::NextFourWords() {
   return {256, static_cast<int16_t>(total_popcount)};
 }
 
+OptionalBitBlockCounter::OptionalBitBlockCounter(const uint8_t* validity_bitmap,
+                                                 int64_t offset, int64_t length)
+    : counter_(validity_bitmap, offset, length),
+      position_(0),
+      length_(length),
+      has_bitmap_(validity_bitmap != nullptr) {}
+
+OptionalBitBlockCounter::OptionalBitBlockCounter(
+    const std::shared_ptr<Buffer>& validity_bitmap, int64_t offset, int64_t length)
+    : OptionalBitBlockCounter(validity_bitmap ? validity_bitmap->data() : nullptr, offset,
+                              length) {}
+
 BitBlockCount BinaryBitBlockCounter::NextAndWord() {
   if (!bits_remaining_) {
     return {0, 0};
diff --git a/cpp/src/arrow/util/bit_block_counter.h b/cpp/src/arrow/util/bit_block_counter.h
index 3e01ddafdff..8632686634a 100644
--- a/cpp/src/arrow/util/bit_block_counter.h
+++ b/cpp/src/arrow/util/bit_block_counter.h
@@ -17,14 +17,21 @@
 
 #pragma once
 
+#include <algorithm>
 #include <cstdint>
+#include <limits>
+#include <memory>
 
+#include "arrow/array/array_base.h"
 #include "arrow/util/bit_util.h"
 #include "arrow/util/bitmap_ops.h"
 #include "arrow/util/ubsan.h"
 #include "arrow/util/visibility.h"
 
 namespace arrow {
+
+class Buffer;
+
 namespace internal {
 
 /// \brief Return value from bit block counters: the total number of bits and
@@ -68,6 +75,41 @@ class ARROW_EXPORT BitBlockCounter {
   int64_t offset_;
 };
 
+/// \brief A tool to iterate through a possibly non-existent validity bitmap,
+/// to allow us to write one code path for both the with-nulls and no-nulls
+/// cases without giving up a lot of performance.
+class OptionalBitBlockCounter {
+ public:
+  OptionalBitBlockCounter(const uint8_t* validity_bitmap, int64_t offset, int64_t length);
+
+  OptionalBitBlockCounter(const std::shared_ptr<Buffer>& validity_bitmap, int64_t offset,
+                          int64_t length);
+
+  /// Return block count for next word when the bitmap is available otherwise
+  /// return a block with length up to INT16_MAX when there is no validity
+  /// bitmap (so all the referenced values are not null).
+  BitBlockCount NextBlock() {
+    static constexpr int64_t kMaxBlockSize = std::numeric_limits<int16_t>::max();
+    if (has_bitmap_) {
+      BitBlockCount block = counter_.NextWord();
+      position_ += block.length;
+      return block;
+    } else {
+      int16_t block_size =
+          static_cast<int16_t>(std::min(kMaxBlockSize, length_ - position_));
+      position_ += block_size;
+      // All values are non-null
+      return {block_size, block_size};
+    }
+  }
+
+ private:
+  BitBlockCounter counter_;
+  int64_t position_;
+  int64_t length_;
+  bool has_bitmap_;
+};
+
 /// \brief A class that computes popcounts on the result of bitwise operations
 /// between two bitmaps, 64 bits at a time. A 64-bit word is loaded from each
 /// bitmap, then the popcount is computed on e.g. the bitwise-and of the two
diff --git a/cpp/src/arrow/util/bit_block_counter_test.cc b/cpp/src/arrow/util/bit_block_counter_test.cc
index d08e1f82c9e..822fb16b9c1 100644
--- a/cpp/src/arrow/util/bit_block_counter_test.cc
+++ b/cpp/src/arrow/util/bit_block_counter_test.cc
@@ -246,5 +246,28 @@ TEST(TestBinaryBitBlockCounter, NextAndWord) {
   }
 }
 
+TEST(TestOptionalBitBlockCounter, Basics) {
+  const int64_t nbytes = 1024;
+  auto bitmap = *AllocateBitmap(nbytes * 8);
+  random_bytes(nbytes, 0, bitmap->mutable_data());
+
+  OptionalBitBlockCounter optional_counter(bitmap, 0, nbytes * 8);
+  BitBlockCounter bit_counter(bitmap->data(), 0, nbytes * 8);
+
+  while (true) {
+    BitBlockCount block = bit_counter.NextWord();
+    BitBlockCount optional_block = optional_counter.NextBlock();
+    ASSERT_EQ(optional_block.length, block.length);
+    ASSERT_EQ(optional_block.popcount, block.popcount);
+    if (block.length == 0) {
+      break;
+    }
+  }
+
+  BitBlockCount optional_block = optional_counter.NextBlock();
+  ASSERT_EQ(optional_block.length, 0);
+  ASSERT_EQ(optional_block.popcount, 0);
+}
+
 }  // namespace internal
 }  // namespace arrow
diff --git a/cpp/src/arrow/util/bitmap_reader.h b/cpp/src/arrow/util/bitmap_reader.h
index 470fbfbb918..838781db8fc 100644
--- a/cpp/src/arrow/util/bitmap_reader.h
+++ b/cpp/src/arrow/util/bitmap_reader.h
@@ -20,6 +20,7 @@
 #include <cstdint>
 #include <cstring>
 
+#include "arrow/buffer.h"
 #include "arrow/util/bit_util.h"
 #include "arrow/util/macros.h"
 
@@ -66,5 +67,18 @@ class BitmapReader {
   int64_t bit_offset_;
 };
 
+/// \brief Index into a possibly non-existent bitmap
+struct OptionalBitIndexer {
+  const uint8_t* bitmap;
+  const int64_t offset;
+
+  explicit OptionalBitIndexer(const std::shared_ptr<Buffer>& buffer, int64_t offset = 0)
+      : bitmap(buffer == NULLPTR ? NULLPTR : buffer->data()), offset(offset) {}
+
+  bool operator[](int64_t i) const {
+    return bitmap == NULLPTR ? true : BitUtil::GetBit(bitmap, offset + i);
+  }
+};
+
 }  // namespace internal
 }  // namespace arrow
diff --git a/cpp/src/arrow/util/int_util.cc b/cpp/src/arrow/util/int_util.cc
index 10abe78b778..f88f3333bcf 100644
--- a/cpp/src/arrow/util/int_util.cc
+++ b/cpp/src/arrow/util/int_util.cc
@@ -21,8 +21,10 @@
 #include <cstring>
 #include <limits>
 
+#include "arrow/util/bit_block_counter.h"
 #include "arrow/util/bit_util.h"
 #include "arrow/util/logging.h"
+#include "arrow/util/macros.h"
 
 namespace arrow {
 namespace internal {
@@ -442,5 +444,93 @@ INSTANTIATE_ALL()
 #undef INSTANTIATE_ALL
 #undef INSTANTIATE_ALL_DEST
 
+template <typename IndexCType, bool IsSigned = std::is_signed<IndexCType>::value>
+Status IndexBoundscheckImpl(const ArrayData& indices, uint64_t upper_limit) {
+  // For unsigned integers, if the values array is larger than the maximum
+  // index value (e.g. especially for UINT8 / UINT16), then there is no need to
+  // boundscheck.
+  if (!IsSigned &&
+      upper_limit > static_cast<uint64_t>(std::numeric_limits<IndexCType>::max())) {
+    return Status::OK();
+  }
+
+  const IndexCType* indices_data = indices.GetValues<IndexCType>(1);
+  const uint8_t* bitmap = nullptr;
+  if (indices.buffers[0]) {
+    bitmap = indices.buffers[0]->data();
+  }
+  auto IsOutOfBounds = [&](int64_t i) -> bool {
+    return (IsSigned && indices_data[i] < 0) ||
+           static_cast<uint64_t>(indices_data[i]) >= upper_limit;
+  };
+  OptionalBitBlockCounter indices_bit_counter(bitmap, indices.offset, indices.length);
+  int64_t position = 0;
+  while (position < indices.length) {
+    BitBlockCount block = indices_bit_counter.NextBlock();
+    bool block_out_of_bounds = false;
+    if (block.popcount == block.length) {
+      // Fast path: branchless
+      for (int64_t i = 0; i < block.length; ++i) {
+        block_out_of_bounds |= IsOutOfBounds(i);
+      }
+    } else if (block.popcount > 0) {
+      // Indices have nulls, must only boundscheck non-null values
+      for (int64_t i = 0; i < block.length; ++i) {
+        if (BitUtil::GetBit(bitmap, indices.offset + position + i)) {
+          block_out_of_bounds |= IsOutOfBounds(i);
+        }
+      }
+    }
+    if (ARROW_PREDICT_FALSE(block_out_of_bounds)) {
+      if (indices.GetNullCount() > 0) {
+        for (int64_t i = 0; i < block.length; ++i) {
+          if (BitUtil::GetBit(bitmap, indices.offset + position + i)) {
+            if (IsOutOfBounds(i)) {
+              return Status::IndexError(
+                  "Take index ", static_cast<int64_t>(indices_data[i]), " out of bounds");
+            }
+          }
+        }
+      } else {
+        for (int64_t i = 0; i < block.length; ++i) {
+          if (IsOutOfBounds(i)) {
+            return Status::IndexError(
+                "Take index ", static_cast<int64_t>(indices_data[i]), " out of bounds");
+          }
+        }
+      }
+    }
+    indices_data += block.length;
+    position += block.length;
+  }
+  return Status::OK();
+}
+
+/// \brief Branchless boundschecking of the indices. Processes batches of
+/// indices at a time and shortcircuits when encountering an out-of-bounds
+/// index in a batch
+Status IndexBoundscheck(const ArrayData& indices, uint64_t upper_limit) {
+  switch (indices.type->id()) {
+    case Type::INT8:
+      return IndexBoundscheckImpl<int8_t>(indices, upper_limit);
+    case Type::INT16:
+      return IndexBoundscheckImpl<int16_t>(indices, upper_limit);
+    case Type::INT32:
+      return IndexBoundscheckImpl<int32_t>(indices, upper_limit);
+    case Type::INT64:
+      return IndexBoundscheckImpl<int64_t>(indices, upper_limit);
+    case Type::UINT8:
+      return IndexBoundscheckImpl<uint8_t>(indices, upper_limit);
+    case Type::UINT16:
+      return IndexBoundscheckImpl<uint16_t>(indices, upper_limit);
+    case Type::UINT32:
+      return IndexBoundscheckImpl<uint32_t>(indices, upper_limit);
+    case Type::UINT64:
+      return IndexBoundscheckImpl<uint64_t>(indices, upper_limit);
+    default:
+      return Status::Invalid("Invalid index type for boundschecking");
+  }
+}
+
 }  // namespace internal
 }  // namespace arrow
diff --git a/cpp/src/arrow/util/int_util.h b/cpp/src/arrow/util/int_util.h
index 3316ef59357..095f685a6be 100644
--- a/cpp/src/arrow/util/int_util.h
+++ b/cpp/src/arrow/util/int_util.h
@@ -21,9 +21,13 @@
 #include <limits>
 #include <type_traits>
 
+#include "arrow/status.h"
 #include "arrow/util/visibility.h"
 
 namespace arrow {
+
+struct ArrayData;
+
 namespace internal {
 
 ARROW_EXPORT
@@ -112,5 +116,11 @@ UpcastInt(Integer v) {
   return v;
 }
 
+/// \brief Do vectorized boundschecking of integer-type indices. The indices
+/// must be non-nonnegative and strictly less than the passed upper limit
+/// (which is usually the length of an array that is being indexed-into).
+ARROW_EXPORT
+Status IndexBoundscheck(const ArrayData& indices, uint64_t upper_limit);
+
 }  // namespace internal
 }  // namespace arrow
diff --git a/cpp/src/arrow/util/int_util_test.cc b/cpp/src/arrow/util/int_util_test.cc
index bf458c5df4d..fe0ebf7c709 100644
--- a/cpp/src/arrow/util/int_util_test.cc
+++ b/cpp/src/arrow/util/int_util_test.cc
@@ -18,11 +18,15 @@
 #include <algorithm>
 #include <cstdint>
 #include <random>
+#include <string>
 #include <utility>
 #include <vector>
 
 #include <gtest/gtest.h>
 
+#include "arrow/testing/gtest_util.h"
+#include "arrow/testing/random.h"
+#include "arrow/type.h"
 #include "arrow/util/int_util.h"
 
 namespace arrow {
@@ -382,5 +386,102 @@ TEST(TransposeInts, Int8ToInt64) {
   ASSERT_EQ(dest, std::vector<int64_t>({2222, 4444, 6666, 1111, 4444, 3333}));
 }
 
+void BoundscheckPasses(const std::shared_ptr<DataType>& type,
+                       const std::string& indices_json, uint64_t upper_limit) {
+  auto indices = ArrayFromJSON(type, indices_json);
+  ASSERT_OK(IndexBoundscheck(*indices->data(), upper_limit));
+}
+
+void BoundscheckFails(const std::shared_ptr<DataType>& type,
+                      const std::string& indices_json, uint64_t upper_limit) {
+  auto indices = ArrayFromJSON(type, indices_json);
+  ASSERT_RAISES(IndexError, IndexBoundscheck(*indices->data(), upper_limit));
+}
+
+TEST(IndexBoundscheck, Batching) {
+  auto rand = random::RandomArrayGenerator(/*seed=*/0);
+
+  const int64_t length = 200;
+
+  auto indices = rand.Int16(length, 0, 0, /*null_probability=*/0);
+  ArrayData* index_data = indices->data().get();
+  index_data->buffers[0] = *AllocateBitmap(length);
+
+  int16_t* values = index_data->GetMutableValues<int16_t>(1);
+  uint8_t* bitmap = index_data->buffers[0]->mutable_data();
+  BitUtil::SetBitsTo(bitmap, 0, length, true);
+
+  ASSERT_OK(IndexBoundscheck(*index_data, 1));
+
+  // We'll place out of bounds indices at various locations
+  values[99] = 1;
+  ASSERT_RAISES(IndexError, IndexBoundscheck(*index_data, 1));
+
+  // Make that value null
+  BitUtil::ClearBit(bitmap, 99);
+  ASSERT_OK(IndexBoundscheck(*index_data, 1));
+
+  values[199] = 1;
+  ASSERT_RAISES(IndexError, IndexBoundscheck(*index_data, 1));
+
+  // Make that value null
+  BitUtil::ClearBit(bitmap, 199);
+  ASSERT_OK(IndexBoundscheck(*index_data, 1));
+}
+
+TEST(IndexBoundscheck, SignedInts) {
+  auto CheckCommon = [&](const std::shared_ptr<DataType>& ty) {
+    BoundscheckPasses(ty, "[0, 0, 0]", 1);
+    BoundscheckFails(ty, "[0, 0, 0]", 0);
+    BoundscheckFails(ty, "[-1]", 1);
+    BoundscheckFails(ty, "[-128]", 1);
+    BoundscheckFails(ty, "[0, 100, 127]", 127);
+    BoundscheckPasses(ty, "[0, 100, 127]", 128);
+  };
+
+  CheckCommon(int8());
+
+  CheckCommon(int16());
+  BoundscheckPasses(int16(), "[0, 999, 999]", 1000);
+  BoundscheckFails(int16(), "[0, 1000, 1000]", 1000);
+  BoundscheckPasses(int16(), "[0, 32767]", 1 << 15);
+
+  CheckCommon(int32());
+  BoundscheckPasses(int32(), "[0, 999999, 999999]", 1000000);
+  BoundscheckFails(int32(), "[0, 1000000, 1000000]", 1000000);
+  BoundscheckPasses(int32(), "[0, 2147483647]", 1LL << 31);
+
+  CheckCommon(int64());
+  BoundscheckPasses(int64(), "[0, 9999999999, 9999999999]", 10000000000LL);
+  BoundscheckFails(int64(), "[0, 10000000000, 10000000000]", 10000000000LL);
+}
+
+TEST(IndexBoundscheck, UnsignedInts) {
+  auto CheckCommon = [&](const std::shared_ptr<DataType>& ty) {
+    BoundscheckPasses(ty, "[0, 0, 0]", 1);
+    BoundscheckFails(ty, "[0, 0, 0]", 0);
+    BoundscheckFails(ty, "[0, 100, 200]", 200);
+    BoundscheckPasses(ty, "[0, 100, 200]", 201);
+  };
+
+  CheckCommon(uint8());
+  BoundscheckPasses(uint8(), "[255, 255, 255]", 1000);
+  BoundscheckFails(uint8(), "[255, 255, 255]", 255);
+
+  CheckCommon(uint16());
+  BoundscheckPasses(uint16(), "[0, 999, 999]", 1000);
+  BoundscheckFails(uint16(), "[0, 1000, 1000]", 1000);
+  BoundscheckPasses(uint16(), "[0, 65535]", 1 << 16);
+
+  CheckCommon(uint32());
+  BoundscheckPasses(uint32(), "[0, 999999, 999999]", 1000000);
+  BoundscheckFails(uint32(), "[0, 1000000, 1000000]", 1000000);
+  BoundscheckPasses(uint32(), "[0, 4294967295]", 1LL << 32);
+
+  CheckCommon(uint64());
+  BoundscheckPasses(uint64(), "[0, 9999999999, 9999999999]", 10000000000LL);
+  BoundscheckFails(uint64(), "[0, 10000000000, 10000000000]", 10000000000LL);
+}
+
 }  // namespace internal
 }  // namespace arrow
diff --git a/cpp/src/arrow/util/macros.h b/cpp/src/arrow/util/macros.h
index da68e88863c..4160324f838 100644
--- a/cpp/src/arrow/util/macros.h
+++ b/cpp/src/arrow/util/macros.h
@@ -47,9 +47,11 @@
 #define ARROW_PREDICT_FALSE(x) (__builtin_expect(!!(x), 0))
 #define ARROW_PREDICT_TRUE(x) (__builtin_expect(!!(x), 1))
 #define ARROW_NORETURN __attribute__((noreturn))
+#define ARROW_NOINLINE __attribute__((noinline))
 #define ARROW_PREFETCH(addr) __builtin_prefetch(addr)
 #elif defined(_MSC_VER)
 #define ARROW_NORETURN __declspec(noreturn)
+#define ARROW_NOINLINE __declspec(noinline)
 #define ARROW_PREDICT_FALSE(x) (x)
 #define ARROW_PREDICT_TRUE(x) (x)
 #define ARROW_PREFETCH(addr)
diff --git a/python/pyarrow/_compute.pyx b/python/pyarrow/_compute.pyx
index a2de8ec492d..d6e81646199 100644
--- a/python/pyarrow/_compute.pyx
+++ b/python/pyarrow/_compute.pyx
@@ -416,3 +416,14 @@ cdef class FilterOptions(FunctionOptions):
 
     cdef const CFunctionOptions* get_options(self) except NULL:
         return &self.filter_options
+
+
+cdef class TakeOptions(FunctionOptions):
+    cdef:
+        CTakeOptions take_options
+
+    def __init__(self, boundscheck=True):
+        self.take_options.boundscheck = boundscheck
+
+    cdef const CFunctionOptions* get_options(self) except NULL:
+        return &self.take_options
diff --git a/python/pyarrow/compute.py b/python/pyarrow/compute.py
index 3761c91cc2a..c971f55726d 100644
--- a/python/pyarrow/compute.py
+++ b/python/pyarrow/compute.py
@@ -21,7 +21,8 @@
     Function,
     FunctionRegistry,
     function_registry,
-    call_function
+    call_function,
+    TakeOptions
 )
 import pyarrow._compute as _pc
 
@@ -153,7 +154,7 @@ def filter(data, mask, null_selection_behavior='drop'):
     return call_function('filter', [data, mask], options)
 
 
-def take(data, indices):
+def take(data, indices, boundscheck=True):
     """
     Select values (or records) from array- or table-like data given integer
     selection indices.
@@ -168,6 +169,9 @@ def take(data, indices):
     data : Array, ChunkedArray, RecordBatch, or Table
     indices : Array, ChunkedArray
         Must be of integer type
+    boundscheck : boolean, default True
+        Whether to boundscheck the indices. If False and there is an out of
+        bounds index, will likely cause the process to crash.
 
     Returns
     -------
@@ -187,4 +191,5 @@ def take(data, indices):
       null
     ]
     """
-    return call_function('take', [data, indices])
+    options = TakeOptions(boundscheck)
+    return call_function('take', [data, indices], options)
diff --git a/python/pyarrow/includes/libarrow.pxd b/python/pyarrow/includes/libarrow.pxd
index 0f0dbc3556f..3c503841157 100644
--- a/python/pyarrow/includes/libarrow.pxd
+++ b/python/pyarrow/includes/libarrow.pxd
@@ -1546,9 +1546,6 @@ cdef extern from "arrow/compute/api.h" namespace "arrow::compute" nogil:
         c_bool allow_float_truncate
         c_bool allow_invalid_utf8
 
-    cdef cppclass CTakeOptions" arrow::compute::TakeOptions"(CFunctionOptions):
-        pass
-
     enum CFilterNullSelectionBehavior \
             "arrow::compute::FilterOptions::NullSelectionBehavior":
         CFilterNullSelectionBehavior_DROP \
@@ -1560,6 +1557,10 @@ cdef extern from "arrow/compute/api.h" namespace "arrow::compute" nogil:
             " arrow::compute::FilterOptions"(CFunctionOptions):
         CFilterNullSelectionBehavior null_selection_behavior
 
+    cdef cppclass CTakeOptions \
+            " arrow::compute::TakeOptions"(CFunctionOptions):
+        c_bool boundscheck
+
     enum DatumType" arrow::Datum::type":
         DatumType_NONE" arrow::Datum::NONE"
         DatumType_SCALAR" arrow::Datum::SCALAR"
diff --git a/python/pyarrow/tests/test_pandas.py b/python/pyarrow/tests/test_pandas.py
index 130ac4a8ec7..fc7dd1deddb 100644
--- a/python/pyarrow/tests/test_pandas.py
+++ b/python/pyarrow/tests/test_pandas.py
@@ -3403,6 +3403,7 @@ def test_variable_dictionary_to_pandas():
     expected_dict = pa.concat_arrays([d1, d2, d3])
 
     a = pa.chunked_array([a1, a2, a3, a4])
+
     a_dense = pa.chunked_array([a1.cast('string'),
                                 a2.cast('string'),
                                 a3.cast('string'),

From 8282d38be856ae0a5f37d0d10500557b79637295 Mon Sep 17 00:00:00 2001
From: Wes McKinney <wesm@apache.org>
Date: Tue, 9 Jun 2020 12:05:59 -0500
Subject: [PATCH 2/8] Add 50% / no nulls cases to defaults

---
 cpp/src/arrow/compute/benchmark_util.h | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/cpp/src/arrow/compute/benchmark_util.h b/cpp/src/arrow/compute/benchmark_util.h
index 702f1c7056a..7731b98f0f1 100644
--- a/cpp/src/arrow/compute/benchmark_util.h
+++ b/cpp/src/arrow/compute/benchmark_util.h
@@ -59,7 +59,7 @@ void BenchmarkSetArgsWithSizes(benchmark::internal::Benchmark* bench,
   // 0 is treated as "no nulls"
   for (const auto size : sizes) {
     for (const auto inverse_null_proportion :
-           std::vector<ArgsType>({10000, 100, 50, 10, 2, 1, 0})) {
+         std::vector<ArgsType>({10000, 100, 50, 10, 2, 1, 0})) {
       bench->Args({static_cast<ArgsType>(size), inverse_null_proportion});
     }
   }

From fee36715b01da52b6a423ae69d185eb3e6448959 Mon Sep 17 00:00:00 2001
From: Wes McKinney <wesm@apache.org>
Date: Tue, 9 Jun 2020 17:20:16 -0500
Subject: [PATCH 3/8] Add more comprehensive random data tests, including for
 boolean type, fixed size binary, large string

---
 .../arrow/compute/kernels/vector_take_test.cc | 138 +++++++++++-------
 cpp/src/arrow/util/bit_block_counter.h        |   2 +-
 cpp/src/arrow/util/int_util.cc                |  13 +-
 3 files changed, 95 insertions(+), 58 deletions(-)

diff --git a/cpp/src/arrow/compute/kernels/vector_take_test.cc b/cpp/src/arrow/compute/kernels/vector_take_test.cc
index c30b87ec060..50382c301fa 100644
--- a/cpp/src/arrow/compute/kernels/vector_take_test.cc
+++ b/cpp/src/arrow/compute/kernels/vector_take_test.cc
@@ -15,6 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
+#include <algorithm>
 #include <limits>
 #include <memory>
 #include <string>
@@ -129,6 +130,60 @@ void ValidateTake(const std::shared_ptr<Array>& values,
   }
 }
 
+template <typename T>
+T GetMaxIndex(int64_t values_length) {
+  int64_t max_index = values_length - 1;
+  if (max_index > static_cast<int64_t>(std::numeric_limits<T>::max())) {
+    max_index = std::numeric_limits<T>::max();
+  }
+  return static_cast<T>(max_index);
+}
+
+template <>
+uint64_t GetMaxIndex(int64_t values_length) {
+  return static_cast<uint64_t>(values_length - 1);
+}
+
+template <typename ValuesType, typename IndexType>
+void CheckTakeRandom(const std::shared_ptr<Array>& values, int64_t indices_length,
+                     double null_probability, random::RandomArrayGenerator* rand) {
+  using IndexCType = typename IndexType::c_type;
+  IndexCType max_index = GetMaxIndex<IndexCType>(values->length());
+  auto indices = rand->Numeric<IndexType>(indices_length, static_cast<IndexCType>(0),
+                                          max_index, null_probability);
+  ValidateTake<ValuesType>(values, indices);
+}
+
+template <typename ValuesType, typename DataGenerator>
+void DoRandomTakeTests(DataGenerator&& generate_values) {
+  auto rand = random::RandomArrayGenerator(kRandomSeed);
+  for (size_t i = 5; i < 10; i++) {
+    const int64_t length = static_cast<int64_t>(1ULL << i);
+    for (size_t j = 4; j < 10; j++) {
+      const int64_t indices_length = static_cast<int64_t>(1ULL << j);
+      for (auto null_probability : {0.0, 0.01, 0.25, 1.0}) {
+        auto values = generate_values(length, null_probability, &rand);
+        CheckTakeRandom<ValuesType, Int8Type>(values, indices_length, null_probability,
+                                              &rand);
+        CheckTakeRandom<ValuesType, Int16Type>(values, indices_length, null_probability,
+                                               &rand);
+        CheckTakeRandom<ValuesType, Int32Type>(values, indices_length, null_probability,
+                                               &rand);
+        CheckTakeRandom<ValuesType, Int64Type>(values, indices_length, null_probability,
+                                               &rand);
+        CheckTakeRandom<ValuesType, UInt8Type>(values, indices_length, null_probability,
+                                               &rand);
+        CheckTakeRandom<ValuesType, UInt16Type>(values, indices_length, null_probability,
+                                                &rand);
+        CheckTakeRandom<ValuesType, UInt32Type>(values, indices_length, null_probability,
+                                                &rand);
+        CheckTakeRandom<ValuesType, UInt64Type>(values, indices_length, null_probability,
+                                                &rand);
+      }
+    }
+  }
+}
+
 template <typename ArrowType>
 class TestTakeKernel : public ::testing::Test {};
 
@@ -161,6 +216,13 @@ TEST(TestTakeKernel, TakeBoolean) {
                 TakeJSON(boolean(), "[true, false, true]", int8(), "[0, -1, 0]", &arr));
 }
 
+TEST(TestTakeKernel, TakeBooleanRandom) {
+  DoRandomTakeTests<BooleanType>(
+      [](int64_t length, double null_probability, random::RandomArrayGenerator* rng) {
+        return rng->Boolean(length, 0.5, null_probability);
+      });
+}
+
 template <typename ArrowType>
 class TestTakeKernelWithNumeric : public TestTakeKernel<ArrowType> {
  protected:
@@ -191,19 +253,10 @@ TYPED_TEST(TestTakeKernelWithNumeric, TakeNumeric) {
 }
 
 TYPED_TEST(TestTakeKernelWithNumeric, TakeRandomNumeric) {
-  auto rand = random::RandomArrayGenerator(kRandomSeed);
-  for (size_t i = 3; i < 8; i++) {
-    const int64_t length = static_cast<int64_t>(1ULL << i);
-    for (size_t j = 0; j < 13; j++) {
-      const int64_t indices_length = static_cast<int64_t>(1ULL << j);
-      for (auto null_probability : {0.0, 0.01, 0.25, 1.0}) {
-        auto values = rand.Numeric<TypeParam>(length, 0, 127, null_probability);
-        auto max_index = static_cast<int32_t>(length - 1);
-        auto indices = rand.Int32(indices_length, 0, max_index, null_probability);
-        ValidateTake<TypeParam>(values, indices);
-      }
-    }
-  }
+  DoRandomTakeTests<TypeParam>(
+      [](int64_t length, double null_probability, random::RandomArrayGenerator* rng) {
+        return rng->Numeric<TypeParam>(length, 0, 127, null_probability);
+      });
 }
 
 template <typename TypeClass>
@@ -251,47 +304,30 @@ TYPED_TEST(TestTakeKernelWithString, TakeString) {
 }
 
 TEST(TestTakeKernelString, Random) {
-  auto rand = random::RandomArrayGenerator(kRandomSeed);
-  for (size_t i = 3; i < 8; i++) {
-    const int64_t length = static_cast<int64_t>(1ULL << i);
-    for (size_t j = 0; j < 13; j++) {
-      const int64_t indices_length = static_cast<int64_t>(1ULL << j);
-      for (auto null_probability : {0.0, 0.01, 0.25, 1.0}) {
-        auto values = rand.String(length, 0, 32, null_probability);
-        auto max_index = static_cast<int32_t>(length - 1);
-        auto indices = rand.Int32(indices_length, 0, max_index, null_probability);
-        ValidateTake<StringType>(values, indices);
-      }
-    }
-  }
+  DoRandomTakeTests<StringType>(
+      [](int64_t length, double null_probability, random::RandomArrayGenerator* rng) {
+        return rng->String(length, 0, 32, null_probability);
+      });
+  DoRandomTakeTests<LargeStringType>(
+      [](int64_t length, double null_probability, random::RandomArrayGenerator* rng) {
+        return rng->LargeString(length, 0, 32, null_probability);
+      });
 }
 
 TEST(TestTakeKernelFixedSizeBinary, Random) {
-  auto rand = random::RandomArrayGenerator(kRandomSeed);
-
-  const int32_t value_size = 16;
-  for (size_t i = 3; i < 8; i++) {
-    const int64_t length = static_cast<int64_t>(1ULL << i);
-    for (size_t j = 0; j < 13; j++) {
-      const int64_t indices_length = static_cast<int64_t>(1ULL << j);
-      for (auto null_probability : {0.0, 0.01, 0.25, 1.0}) {
-        int64_t data_nbytes = length * value_size;
-        ASSERT_OK_AND_ASSIGN(std::shared_ptr<Buffer> data, AllocateBuffer(data_nbytes));
-        random_bytes(data_nbytes, /*seed=*/0, data->mutable_data());
-        auto validity = rand.Boolean(length, 1 - null_probability);
-
-        auto max_index = static_cast<int32_t>(length - 1);
-        auto indices = rand.Int32(indices_length, 0, max_index, null_probability);
-
-        // Assemble the data for a FixedSizeBinaryArray
-        auto values_data =
-            std::make_shared<ArrayData>(fixed_size_binary(value_size), length);
-        values_data->buffers = {validity->data()->buffers[1], data};
-
-        ValidateTake<FixedSizeBinaryType>(MakeArray(values_data), indices);
-      }
-    }
-  }
+  DoRandomTakeTests<FixedSizeBinaryType>([](int64_t length, double null_probability,
+                                            random::RandomArrayGenerator* rng) {
+    const int32_t value_size = 16;
+    int64_t data_nbytes = length * value_size;
+    std::shared_ptr<Buffer> data = *AllocateBuffer(data_nbytes);
+    random_bytes(data_nbytes, /*seed=*/0, data->mutable_data());
+    auto validity = rng->Boolean(length, 1 - null_probability);
+
+    // Assemble the data for a FixedSizeBinaryArray
+    auto values_data = std::make_shared<ArrayData>(fixed_size_binary(value_size), length);
+    values_data->buffers = {validity->data()->buffers[1], data};
+    return MakeArray(values_data);
+  });
 }
 
 TYPED_TEST(TestTakeKernelWithString, TakeDictionary) {
diff --git a/cpp/src/arrow/util/bit_block_counter.h b/cpp/src/arrow/util/bit_block_counter.h
index 8632686634a..9ca27603c7c 100644
--- a/cpp/src/arrow/util/bit_block_counter.h
+++ b/cpp/src/arrow/util/bit_block_counter.h
@@ -78,7 +78,7 @@ class ARROW_EXPORT BitBlockCounter {
 /// \brief A tool to iterate through a possibly non-existent validity bitmap,
 /// to allow us to write one code path for both the with-nulls and no-nulls
 /// cases without giving up a lot of performance.
-class OptionalBitBlockCounter {
+class ARROW_EXPORT OptionalBitBlockCounter {
  public:
   OptionalBitBlockCounter(const uint8_t* validity_bitmap, int64_t offset, int64_t length);
 
diff --git a/cpp/src/arrow/util/int_util.cc b/cpp/src/arrow/util/int_util.cc
index f88f3333bcf..48e320031a2 100644
--- a/cpp/src/arrow/util/int_util.cc
+++ b/cpp/src/arrow/util/int_util.cc
@@ -460,8 +460,9 @@ Status IndexBoundscheckImpl(const ArrayData& indices, uint64_t upper_limit) {
     bitmap = indices.buffers[0]->data();
   }
   auto IsOutOfBounds = [&](int64_t i) -> bool {
-    return (IsSigned && indices_data[i] < 0) ||
-           static_cast<uint64_t>(indices_data[i]) >= upper_limit;
+    return (
+        (IsSigned && indices_data[i] < 0) ||
+        (indices_data[i] >= 0 && static_cast<uint64_t>(indices_data[i]) >= upper_limit));
   };
   OptionalBitBlockCounter indices_bit_counter(bitmap, indices.offset, indices.length);
   int64_t position = 0;
@@ -486,16 +487,16 @@ Status IndexBoundscheckImpl(const ArrayData& indices, uint64_t upper_limit) {
         for (int64_t i = 0; i < block.length; ++i) {
           if (BitUtil::GetBit(bitmap, indices.offset + position + i)) {
             if (IsOutOfBounds(i)) {
-              return Status::IndexError(
-                  "Take index ", static_cast<int64_t>(indices_data[i]), " out of bounds");
+              return Status::IndexError("Index ", static_cast<int64_t>(indices_data[i]),
+                                        " out of bounds");
             }
           }
         }
       } else {
         for (int64_t i = 0; i < block.length; ++i) {
           if (IsOutOfBounds(i)) {
-            return Status::IndexError(
-                "Take index ", static_cast<int64_t>(indices_data[i]), " out of bounds");
+            return Status::IndexError("Index ", static_cast<int64_t>(indices_data[i]),
+                                      " out of bounds");
           }
         }
       }

From a1207c3675246508ea5e9a83a2bc4cdfbbb666ac Mon Sep 17 00:00:00 2001
From: Wes McKinney <wesm@apache.org>
Date: Wed, 10 Jun 2020 09:45:26 -0500
Subject: [PATCH 4/8] Some bug fixes and cleaning, address initial feedback.
 Report benchmarks in items/s rather than bytes for comparability across data
 types

---
 cpp/src/arrow/compute/api_vector.h            |   6 +-
 cpp/src/arrow/compute/benchmark_util.h        |  13 +-
 .../kernels/vector_selection_benchmark.cc     |  20 +--
 cpp/src/arrow/compute/kernels/vector_take.cc  | 136 +++++++++---------
 .../arrow/compute/kernels/vector_take_test.cc |   2 +-
 cpp/src/arrow/util/bit_block_counter.h        |   2 +
 cpp/src/arrow/util/bit_block_counter_test.cc  |   7 +-
 cpp/src/arrow/util/int_util.cc                |  20 +--
 cpp/src/arrow/util/int_util.h                 |   2 +-
 cpp/src/arrow/util/int_util_test.cc           |  80 +++++------
 python/pyarrow/tests/test_pandas.py           |   1 -
 11 files changed, 139 insertions(+), 150 deletions(-)

diff --git a/cpp/src/arrow/compute/api_vector.h b/cpp/src/arrow/compute/api_vector.h
index 8aefbcf1ccf..48f9102eb42 100644
--- a/cpp/src/arrow/compute/api_vector.h
+++ b/cpp/src/arrow/compute/api_vector.h
@@ -68,9 +68,9 @@ struct ARROW_EXPORT TakeOptions : public FunctionOptions {
   explicit TakeOptions(bool boundscheck = true) : boundscheck(boundscheck) {}
 
   bool boundscheck = true;
-  static TakeOptions Boundscheck() { return TakeOptions(true); }
-  static TakeOptions NoBoundscheck() { return TakeOptions(false); }
-  static TakeOptions Defaults() { return Boundscheck(); }
+  static TakeOptions BoundsCheck() { return TakeOptions(true); }
+  static TakeOptions NoBoundsCheck() { return TakeOptions(false); }
+  static TakeOptions Defaults() { return BoundsCheck(); }
 };
 
 /// \brief Take from an array of values at indices in another array
diff --git a/cpp/src/arrow/compute/benchmark_util.h b/cpp/src/arrow/compute/benchmark_util.h
index 7731b98f0f1..05cd2814bff 100644
--- a/cpp/src/arrow/compute/benchmark_util.h
+++ b/cpp/src/arrow/compute/benchmark_util.h
@@ -83,8 +83,10 @@ struct RegressionArgs {
   // proportion of nulls in generated arrays
   double null_proportion;
 
-  explicit RegressionArgs(benchmark::State& state)
-      : size(state.range(0)), null_proportion(), state_(state) {
+  // If size_is_bytes is true, then it's a number of bytes, otherwise it's the
+  // number of items processed (for reporting)
+  explicit RegressionArgs(benchmark::State& state, bool size_is_bytes = true)
+      : size(state.range(0)), state_(state), size_is_bytes_(size_is_bytes) {
     if (state.range(1) == 0) {
       this->null_proportion = 0.0;
     } else {
@@ -95,11 +97,16 @@ struct RegressionArgs {
   ~RegressionArgs() {
     state_.counters["size"] = static_cast<double>(size);
     state_.counters["null_percent"] = null_proportion * 100;
-    state_.SetBytesProcessed(state_.iterations() * size);
+    if (size_is_bytes_) {
+      state_.SetBytesProcessed(state_.iterations() * size);
+    } else {
+      state_.SetItemsProcessed(state_.iterations() * size);
+    }
   }
 
  private:
   benchmark::State& state_;
+  bool size_is_bytes_;
 };
 
 }  // namespace compute
diff --git a/cpp/src/arrow/compute/kernels/vector_selection_benchmark.cc b/cpp/src/arrow/compute/kernels/vector_selection_benchmark.cc
index 822363ac554..75fd5716d4b 100644
--- a/cpp/src/arrow/compute/kernels/vector_selection_benchmark.cc
+++ b/cpp/src/arrow/compute/kernels/vector_selection_benchmark.cc
@@ -87,38 +87,28 @@ struct TakeBenchmark {
   TakeBenchmark(benchmark::State& state, bool indices_have_nulls,
                 bool monotonic_indices = false)
       : state(state),
-        args(state),
+        args(state, /*size_is_bytes=*/false),
         rand(kSeed),
         indices_have_nulls(indices_have_nulls),
         monotonic_indices(monotonic_indices) {}
 
   void Int64() {
-    const int64_t array_size = args.size / sizeof(int64_t);
-    auto values = rand.Int64(array_size, -100, 100, args.null_proportion);
+    auto values = rand.Int64(args.size, -100, 100, args.null_proportion);
     Bench(values);
   }
 
   void FSLInt64() {
-    const int64_t array_size = args.size / sizeof(int64_t);
-    auto int_array = rand.Int64(array_size, -100, 100, args.null_proportion);
+    auto int_array = rand.Int64(args.size, -100, 100, args.null_proportion);
     auto values = std::make_shared<FixedSizeListArray>(
-        fixed_size_list(int64(), 1), array_size, int_array, int_array->null_bitmap(),
+        fixed_size_list(int64(), 1), args.size, int_array, int_array->null_bitmap(),
         int_array->null_count());
     Bench(values);
   }
 
   void String() {
     int32_t string_min_length = 0, string_max_length = 32;
-    int32_t string_mean_length = (string_max_length + string_min_length) / 2;
-    // for an array of 50% null strings, we need to generate twice as many strings
-    // to ensure that they have an average of args.size total characters
-    int64_t array_size = args.size;
-    if (args.null_proportion < 1) {
-      array_size = static_cast<int64_t>(args.size / string_mean_length /
-                                        (1 - args.null_proportion));
-    }
     auto values = std::static_pointer_cast<StringArray>(rand.String(
-        array_size, string_min_length, string_max_length, args.null_proportion));
+        args.size, string_min_length, string_max_length, args.null_proportion));
     Bench(values);
   }
 
diff --git a/cpp/src/arrow/compute/kernels/vector_take.cc b/cpp/src/arrow/compute/kernels/vector_take.cc
index 8aae6811a2a..fdf175ccc4a 100644
--- a/cpp/src/arrow/compute/kernels/vector_take.cc
+++ b/cpp/src/arrow/compute/kernels/vector_take.cc
@@ -37,32 +37,21 @@ namespace arrow {
 using internal::BitBlockCount;
 using internal::BitmapReader;
 using internal::GetArrayView;
-using internal::IndexBoundscheck;
+using internal::IndexBoundsCheck;
 using internal::OptionalBitBlockCounter;
 using internal::OptionalBitIndexer;
 
 namespace compute {
 namespace internal {
 
-struct TakeState : public KernelState {
-  explicit TakeState(const TakeOptions& options) : options(options) {}
-  TakeOptions options;
-};
-
-std::unique_ptr<KernelState> InitTake(KernelContext*, const KernelInitArgs& args) {
-  // NOTE: TakeOptions are currently unused, but we pass it through anyway
-  auto take_options = static_cast<const TakeOptions*>(args.options);
-  return std::unique_ptr<KernelState>(new TakeState{*take_options});
-}
-
-namespace {}  // namespace
+using TakeState = OptionsWrapper<TakeOptions>;
 
 // ----------------------------------------------------------------------
 // Implement optimized take for primitive types from boolean to 1/2/4/8-byte
 // C-type based types. Use common implementation for every byte width and only
 // generate code for unsigned integer indices, since after boundschecking to
-// check for negative numbers the indices we can safely reinterpret_cast signed
-// integers as unsigned.
+// check for negative numbers in the indices we can safely reinterpret_cast
+// signed integers as unsigned.
 
 struct PrimitiveTakeArgs {
   const uint8_t* values;
@@ -88,7 +77,7 @@ PrimitiveTakeArgs GetPrimitiveTakeArgs(const ExecBatch& batch) {
   const ArrayData& arg1 = *batch[1].array();
 
   // Values
-  args.values_bit_width = static_cast<const FixedWidthType&>(*arg0.type).bit_width();
+  args.values_bit_width = checked_cast<const FixedWidthType&>(*arg0.type).bit_width();
   args.values = arg0.buffers[1]->data();
   if (args.values_bit_width > 1) {
     args.values += arg0.offset * args.values_bit_width / 8;
@@ -101,7 +90,7 @@ PrimitiveTakeArgs GetPrimitiveTakeArgs(const ExecBatch& batch) {
   }
 
   // Indices
-  args.indices_bit_width = static_cast<const FixedWidthType&>(*arg1.type).bit_width();
+  args.indices_bit_width = checked_cast<const FixedWidthType&>(*arg1.type).bit_width();
   args.indices = arg1.buffers[1]->data() + arg1.offset * args.indices_bit_width / 8;
   args.indices_length = arg1.length;
   args.indices_offset = arg1.offset;
@@ -145,12 +134,8 @@ struct PrimitiveTakeImpl {
                                                 args.indices_length);
     int64_t position = 0;
     int64_t valid_count = 0;
-    while (true) {
+    while (position < args.indices_length) {
       BitBlockCount block = indices_bit_counter.NextBlock();
-      if (block.length == 0) {
-        // All indices processed.
-        break;
-      }
       if (args.values_null_count == 0) {
         // Values are never null, so things are easier
         valid_count += block.popcount;
@@ -171,6 +156,8 @@ struct PrimitiveTakeImpl {
             }
             ++position;
           }
+        } else {
+          position += block.length;
         }
       } else {
         // Values have nulls, so we must do random access into the values bitmap
@@ -201,6 +188,8 @@ struct PrimitiveTakeImpl {
             }
             ++position;
           }
+        } else {
+          position += block.length;
         }
       }
     }
@@ -240,12 +229,8 @@ struct BooleanTakeImpl {
                                                 args.indices_length);
     int64_t position = 0;
     int64_t valid_count = 0;
-    while (true) {
+    while (position < args.indices_length) {
       BitBlockCount block = indices_bit_counter.NextBlock();
-      if (block.length == 0) {
-        // All indices processed.
-        break;
-      }
       if (args.values_null_count == 0) {
         // Values are never null, so things are easier
         valid_count += block.popcount;
@@ -266,6 +251,8 @@ struct BooleanTakeImpl {
             }
             ++position;
           }
+        } else {
+          position += block.length;
         }
       } else {
         // Values have nulls, so we must do random access into the values bitmap
@@ -296,6 +283,8 @@ struct BooleanTakeImpl {
             }
             ++position;
           }
+        } else {
+          position += block.length;
         }
       }
     }
@@ -325,7 +314,7 @@ void TakeIndexDispatch(const PrimitiveTakeArgs& args, Datum* out) {
   }
 }
 
-Status PreallocateTake(KernelContext* ctx, int64_t length, int bit_width, Datum* out) {
+Status PreallocateData(KernelContext* ctx, int64_t length, int bit_width, Datum* out) {
   // Preallocate memory
   ArrayData* out_arr = out->mutable_array();
   out_arr->length = length;
@@ -343,11 +332,11 @@ Status PreallocateTake(KernelContext* ctx, int64_t length, int bit_width, Datum*
 static void PrimitiveTakeExec(KernelContext* ctx, const ExecBatch& batch, Datum* out) {
   const auto& state = checked_cast<const TakeState&>(*ctx->state());
   if (state.options.boundscheck) {
-    KERNEL_RETURN_IF_ERROR(ctx, IndexBoundscheck(*batch[1].array(), batch[0].length()));
+    KERNEL_RETURN_IF_ERROR(ctx, IndexBoundsCheck(*batch[1].array(), batch[0].length()));
   }
   PrimitiveTakeArgs args = GetPrimitiveTakeArgs(batch);
   KERNEL_RETURN_IF_ERROR(
-      ctx, PreallocateTake(ctx, args.indices_length, args.values_bit_width, out));
+      ctx, PreallocateData(ctx, args.indices_length, args.values_bit_width, out));
   switch (args.values_bit_width) {
     case 1:
       return TakeIndexDispatch<BooleanTakeImpl>(args, out);
@@ -368,7 +357,7 @@ static void PrimitiveTakeExec(KernelContext* ctx, const ExecBatch& batch, Datum*
 static void NullTakeExec(KernelContext* ctx, const ExecBatch& batch, Datum* out) {
   const auto& state = checked_cast<const TakeState&>(*ctx->state());
   if (state.options.boundscheck) {
-    KERNEL_RETURN_IF_ERROR(ctx, IndexBoundscheck(*batch[1].array(), batch[0].length()));
+    KERNEL_RETURN_IF_ERROR(ctx, IndexBoundsCheck(*batch[1].array(), batch[0].length()));
   }
   out->value = std::make_shared<NullArray>(batch.length)->data();
 }
@@ -403,7 +392,8 @@ static void ExtensionTakeExec(KernelContext* ctx, const ExecBatch& batch, Datum*
 
 // ----------------------------------------------------------------------
 
-// Use CRTP here
+// Use CRTP to dispatch to type-specific processing of indices for each
+// unsigned integer type.
 template <typename Impl, typename Type>
 struct GenericTakeImpl {
   using ValuesArrayType = typename TypeTraits<Type>::ArrayType;
@@ -421,6 +411,8 @@ struct GenericTakeImpl {
         out(out->mutable_array()),
         validity_builder(ctx->memory_pool()) {}
 
+  virtual ~GenericTakeImpl() = default;
+
   Status FinishCommon() {
     out->buffers.resize(values->buffers.size());
     out->length = validity_builder.length();
@@ -429,34 +421,33 @@ struct GenericTakeImpl {
   }
 
   template <typename IndexCType, typename ValidVisitor, typename NullVisitor>
-  Status VisitIndices(const ArrayData& indices, ValidVisitor&& visit_valid,
-                      NullVisitor&& visit_null) {
-    const auto indices_values = indices.GetValues<IndexCType>(1);
+  Status VisitIndices(ValidVisitor&& visit_valid, NullVisitor&& visit_null) {
+    const auto indices_values = indices->GetValues<IndexCType>(1);
+    const int64_t indices_offset = indices->offset;
     const uint8_t* bitmap = nullptr;
-    if (indices.buffers[0]) {
-      bitmap = indices.buffers[0]->data();
+    if (indices->buffers[0]) {
+      bitmap = indices->buffers[0]->data();
     }
+    OptionalBitIndexer indices_is_valid(indices->buffers[0], values->offset);
     OptionalBitIndexer values_is_valid(values->buffers[0], values->offset);
 
-    OptionalBitBlockCounter bit_counter(bitmap, indices.offset, indices.length);
+    OptionalBitBlockCounter bit_counter(bitmap, indices_offset, indices->length);
     int64_t position = 0;
-    while (position < indices.length) {
+    while (position < indices->length) {
       BitBlockCount block = bit_counter.NextBlock();
-      if (block.popcount == block.length) {
-        // Fast path, no null indices
+      const bool indices_have_nulls = block.popcount < block.length;
+      if (!indices_have_nulls && values->GetNullCount() == 0) {
+        // Fastest path, neither indices nor values have nulls
+        validity_builder.UnsafeAppend(block.length, true);
         for (int64_t i = 0; i < block.length; ++i) {
-          if (values_is_valid[indices_values[position]]) {
-            validity_builder.UnsafeAppend(true);
-            RETURN_NOT_OK(visit_valid(indices_values[position]));
-          } else {
-            validity_builder.UnsafeAppend(false);
-            RETURN_NOT_OK(visit_null());
-          }
-          ++position;
+          RETURN_NOT_OK(visit_valid(indices_values[position++]));
         }
-      } else {
+      } else if (block.popcount > 0) {
+        // Since we have to branch on whether the indices are null or not, we
+        // combine the "non-null indices block but some values null" and
+        // "some-null indices block but values non-null" into a single loop.
         for (int64_t i = 0; i < block.length; ++i) {
-          if (BitUtil::GetBit(bitmap, indices.offset + position) &&
+          if ((!indices_have_nulls || indices_is_valid[position]) &&
               values_is_valid[indices_values[position]]) {
             validity_builder.UnsafeAppend(true);
             RETURN_NOT_OK(visit_valid(indices_values[position]));
@@ -466,6 +457,13 @@ struct GenericTakeImpl {
           }
           ++position;
         }
+      } else {
+        // The whole block is null
+        validity_builder.UnsafeAppend(block.length, false);
+        for (int64_t i = 0; i < block.length; ++i) {
+          RETURN_NOT_OK(visit_null());
+        }
+        position += block.length;
       }
     }
     return Status::OK();
@@ -476,11 +474,13 @@ struct GenericTakeImpl {
   // Implementation specific finish logic
   virtual Status Finish() = 0;
 
-  virtual Status Exec() {
+  Status Exec() {
     RETURN_NOT_OK(this->validity_builder.Reserve(indices->length));
     RETURN_NOT_OK(Init());
     int index_width =
-        static_cast<const FixedWidthType&>(*this->indices->type).bit_width() / 8;
+        checked_cast<const FixedWidthType&>(*this->indices->type).bit_width() / 8;
+
+    // CTRP dispatch here
     switch (index_width) {
       case 1:
         RETURN_NOT_OK(static_cast<Impl*>(this)->template ProcessIndices<uint8_t>());
@@ -538,15 +538,12 @@ struct VarBinaryTakeImpl : public GenericTakeImpl<VarBinaryTakeImpl<Type>, Type>
   Status ProcessIndices() {
     ValuesArrayType typed_values(this->values_as_binary);
 
-    // Allocate at least 32K at a time to avoid having to call Reserve for
-    // every value for lots of small strings
-    static constexpr int64_t kAllocateChunksize = 1 << 15;
-    RETURN_NOT_OK(data_builder.Reserve(kAllocateChunksize));
+    // Start out with at least 64K allocated
+    RETURN_NOT_OK(data_builder.Reserve(1 << 16));
     int64_t space_available = data_builder.capacity();
 
     offset_type offset = 0;
     RETURN_NOT_OK(this->template VisitIndices<IndexCType>(
-        *indices,
         [&](IndexCType index) {
           offset_builder.UnsafeAppend(offset);
           auto val = typed_values.GetView(index);
@@ -557,7 +554,7 @@ struct VarBinaryTakeImpl : public GenericTakeImpl<VarBinaryTakeImpl<Type>, Type>
           }
           offset += value_size;
           if (ARROW_PREDICT_FALSE(value_size > space_available)) {
-            RETURN_NOT_OK(data_builder.Reserve(value_size + kAllocateChunksize));
+            RETURN_NOT_OK(data_builder.Reserve(value_size));
             space_available = data_builder.capacity() - data_builder.length();
           }
           data_builder.UnsafeAppend(reinterpret_cast<const uint8_t*>(val.data()),
@@ -601,7 +598,6 @@ struct FSBTakeImpl : public GenericTakeImpl<FSBTakeImpl, FixedSizeBinaryType> {
 
     RETURN_NOT_OK(data_builder.Reserve(value_size * indices->length));
     RETURN_NOT_OK(this->template VisitIndices<IndexCType>(
-        *indices,
         [&](IndexCType index) {
           auto val = typed_values.GetView(index);
           data_builder.UnsafeAppend(reinterpret_cast<const uint8_t*>(val.data()),
@@ -655,8 +651,8 @@ struct ListTakeImpl : public GenericTakeImpl<ListTakeImpl<Type>, Type> {
       return Status::OK();
     };
 
-    RETURN_NOT_OK(this->template VisitIndices<IndexCType>(
-        *indices, std::move(PushValidIndex), std::move(PushNullIndex)));
+    RETURN_NOT_OK(this->template VisitIndices<IndexCType>(std::move(PushValidIndex),
+                                                          std::move(PushNullIndex)));
     offset_builder.UnsafeAppend(offset);
     return Status::OK();
   }
@@ -675,7 +671,7 @@ struct ListTakeImpl : public GenericTakeImpl<ListTakeImpl<Type>, Type> {
     // No need to boundscheck the child values indices
     ARROW_ASSIGN_OR_RAISE(std::shared_ptr<Array> taken_child,
                           Take(*typed_values.values(), *child_indices,
-                               TakeOptions::NoBoundscheck(), ctx->exec_context()));
+                               TakeOptions::NoBoundsCheck(), ctx->exec_context()));
     RETURN_NOT_OK(offset_builder.Finish(&out->buffers[1]));
     out->child_data = {taken_child->data()};
     return Status::OK();
@@ -700,7 +696,6 @@ struct FSLTakeImpl : public GenericTakeImpl<FSLTakeImpl, FixedSizeListType> {
     /// indices.
     RETURN_NOT_OK(child_index_builder.Reserve(indices->length * list_size));
     return this->template VisitIndices<IndexCType>(
-        *indices,
         [&](IndexCType index) {
           int64_t offset = index * list_size;
           for (int64_t j = offset; j < offset + list_size; ++j) {
@@ -720,7 +715,7 @@ struct FSLTakeImpl : public GenericTakeImpl<FSLTakeImpl, FixedSizeListType> {
     // No need to boundscheck the child values indices
     ARROW_ASSIGN_OR_RAISE(std::shared_ptr<Array> taken_child,
                           Take(*typed_values.values(), *child_indices,
-                               TakeOptions::NoBoundscheck(), ctx->exec_context()));
+                               TakeOptions::NoBoundsCheck(), ctx->exec_context()));
     out->child_data = {taken_child->data()};
     return Status::OK();
   }
@@ -730,14 +725,13 @@ struct StructTakeImpl : public GenericTakeImpl<StructTakeImpl, StructType> {
   using Base = GenericTakeImpl<StructTakeImpl, StructType>;
   LIFT_BASE_MEMBERS();
 
-  StructTakeImpl(KernelContext* ctx, const ExecBatch& batch, Datum* out)
-      : Base(ctx, batch, out) {}
+  using Base::Base;
 
   template <typename IndexCType>
   Status ProcessIndices() {
     StructArray typed_values(values);
     return this->template VisitIndices<IndexCType>(
-        *indices, [&](IndexCType index) { return Status::OK(); },
+        [&](IndexCType index) { return Status::OK(); },
         /*visit_null=*/VisitNoop);
   }
 
@@ -749,7 +743,7 @@ struct StructTakeImpl : public GenericTakeImpl<StructTakeImpl, StructType> {
     for (int field_index = 0; field_index < values->type->num_fields(); ++field_index) {
       ARROW_ASSIGN_OR_RAISE(Datum taken_field,
                             Take(Datum(typed_values.field(field_index)), Datum(indices),
-                                 TakeOptions::NoBoundscheck(), ctx->exec_context()));
+                                 TakeOptions::NoBoundsCheck(), ctx->exec_context()));
       out->child_data[field_index] = taken_field.array();
     }
     return Status::OK();
@@ -760,7 +754,7 @@ template <typename Impl>
 static void GenericTakeExec(KernelContext* ctx, const ExecBatch& batch, Datum* out) {
   const auto& state = checked_cast<const TakeState&>(*ctx->state());
   if (state.options.boundscheck) {
-    KERNEL_RETURN_IF_ERROR(ctx, IndexBoundscheck(*batch[1].array(), batch[0].length()));
+    KERNEL_RETURN_IF_ERROR(ctx, IndexBoundsCheck(*batch[1].array(), batch[0].length()));
   }
   Impl kernel(ctx, batch, out);
   KERNEL_RETURN_IF_ERROR(ctx, kernel.Exec());
@@ -926,7 +920,7 @@ static InputType kTakeIndexType(match::Integer(), ValueDescr::ARRAY);
 
 void RegisterVectorTake(FunctionRegistry* registry) {
   VectorKernel base;
-  base.init = InitTake;
+  base.init = InitWrapOptions<TakeOptions>;
   base.can_execute_chunkwise = false;
 
   auto array_take = std::make_shared<VectorFunction>("array_take", Arity::Binary());
diff --git a/cpp/src/arrow/compute/kernels/vector_take_test.cc b/cpp/src/arrow/compute/kernels/vector_take_test.cc
index 50382c301fa..9f1dd59220c 100644
--- a/cpp/src/arrow/compute/kernels/vector_take_test.cc
+++ b/cpp/src/arrow/compute/kernels/vector_take_test.cc
@@ -161,7 +161,7 @@ void DoRandomTakeTests(DataGenerator&& generate_values) {
     const int64_t length = static_cast<int64_t>(1ULL << i);
     for (size_t j = 4; j < 10; j++) {
       const int64_t indices_length = static_cast<int64_t>(1ULL << j);
-      for (auto null_probability : {0.0, 0.01, 0.25, 1.0}) {
+      for (auto null_probability : {0.0, 0.01, 0.25, 0.999, 1.0}) {
         auto values = generate_values(length, null_probability, &rand);
         CheckTakeRandom<ValuesType, Int8Type>(values, indices_length, null_probability,
                                               &rand);
diff --git a/cpp/src/arrow/util/bit_block_counter.h b/cpp/src/arrow/util/bit_block_counter.h
index 9ca27603c7c..85a5bf49f85 100644
--- a/cpp/src/arrow/util/bit_block_counter.h
+++ b/cpp/src/arrow/util/bit_block_counter.h
@@ -80,8 +80,10 @@ class ARROW_EXPORT BitBlockCounter {
 /// cases without giving up a lot of performance.
 class ARROW_EXPORT OptionalBitBlockCounter {
  public:
+  // validity_bitmap may be nullptr
   OptionalBitBlockCounter(const uint8_t* validity_bitmap, int64_t offset, int64_t length);
 
+  // validity_bitmap may be null
   OptionalBitBlockCounter(const std::shared_ptr<Buffer>& validity_bitmap, int64_t offset,
                           int64_t length);
 
diff --git a/cpp/src/arrow/util/bit_block_counter_test.cc b/cpp/src/arrow/util/bit_block_counter_test.cc
index 822fb16b9c1..a634bfe5319 100644
--- a/cpp/src/arrow/util/bit_block_counter_test.cc
+++ b/cpp/src/arrow/util/bit_block_counter_test.cc
@@ -217,11 +217,8 @@ TEST(TestBinaryBitBlockCounter, NextAndWord) {
     BinaryBitBlockCounter counter(left->data(), left_offset, right->data(), right_offset,
                                   overlap_length);
     int64_t position = 0;
-    while (true) {
+    do {
       BitBlockCount block = counter.NextAndWord();
-      if (block.length == 0) {
-        break;
-      }
       int expected_popcount = 0;
       for (int j = 0; j < block.length; ++j) {
         expected_popcount +=
@@ -230,7 +227,7 @@ TEST(TestBinaryBitBlockCounter, NextAndWord) {
       }
       ASSERT_EQ(block.popcount, expected_popcount);
       position += block.length;
-    }
+    } while (position < overlap_length);
     // We made it through all the data
     ASSERT_EQ(position, overlap_length);
 
diff --git a/cpp/src/arrow/util/int_util.cc b/cpp/src/arrow/util/int_util.cc
index 48e320031a2..e85abb86fcd 100644
--- a/cpp/src/arrow/util/int_util.cc
+++ b/cpp/src/arrow/util/int_util.cc
@@ -445,7 +445,7 @@ INSTANTIATE_ALL()
 #undef INSTANTIATE_ALL_DEST
 
 template <typename IndexCType, bool IsSigned = std::is_signed<IndexCType>::value>
-Status IndexBoundscheckImpl(const ArrayData& indices, uint64_t upper_limit) {
+Status IndexBoundsCheckImpl(const ArrayData& indices, uint64_t upper_limit) {
   // For unsigned integers, if the values array is larger than the maximum
   // index value (e.g. especially for UINT8 / UINT16), then there is no need to
   // boundscheck.
@@ -510,24 +510,24 @@ Status IndexBoundscheckImpl(const ArrayData& indices, uint64_t upper_limit) {
 /// \brief Branchless boundschecking of the indices. Processes batches of
 /// indices at a time and shortcircuits when encountering an out-of-bounds
 /// index in a batch
-Status IndexBoundscheck(const ArrayData& indices, uint64_t upper_limit) {
+Status IndexBoundsCheck(const ArrayData& indices, uint64_t upper_limit) {
   switch (indices.type->id()) {
     case Type::INT8:
-      return IndexBoundscheckImpl<int8_t>(indices, upper_limit);
+      return IndexBoundsCheckImpl<int8_t>(indices, upper_limit);
     case Type::INT16:
-      return IndexBoundscheckImpl<int16_t>(indices, upper_limit);
+      return IndexBoundsCheckImpl<int16_t>(indices, upper_limit);
     case Type::INT32:
-      return IndexBoundscheckImpl<int32_t>(indices, upper_limit);
+      return IndexBoundsCheckImpl<int32_t>(indices, upper_limit);
     case Type::INT64:
-      return IndexBoundscheckImpl<int64_t>(indices, upper_limit);
+      return IndexBoundsCheckImpl<int64_t>(indices, upper_limit);
     case Type::UINT8:
-      return IndexBoundscheckImpl<uint8_t>(indices, upper_limit);
+      return IndexBoundsCheckImpl<uint8_t>(indices, upper_limit);
     case Type::UINT16:
-      return IndexBoundscheckImpl<uint16_t>(indices, upper_limit);
+      return IndexBoundsCheckImpl<uint16_t>(indices, upper_limit);
     case Type::UINT32:
-      return IndexBoundscheckImpl<uint32_t>(indices, upper_limit);
+      return IndexBoundsCheckImpl<uint32_t>(indices, upper_limit);
     case Type::UINT64:
-      return IndexBoundscheckImpl<uint64_t>(indices, upper_limit);
+      return IndexBoundsCheckImpl<uint64_t>(indices, upper_limit);
     default:
       return Status::Invalid("Invalid index type for boundschecking");
   }
diff --git a/cpp/src/arrow/util/int_util.h b/cpp/src/arrow/util/int_util.h
index 095f685a6be..0478c55e623 100644
--- a/cpp/src/arrow/util/int_util.h
+++ b/cpp/src/arrow/util/int_util.h
@@ -120,7 +120,7 @@ UpcastInt(Integer v) {
 /// must be non-nonnegative and strictly less than the passed upper limit
 /// (which is usually the length of an array that is being indexed-into).
 ARROW_EXPORT
-Status IndexBoundscheck(const ArrayData& indices, uint64_t upper_limit);
+Status IndexBoundsCheck(const ArrayData& indices, uint64_t upper_limit);
 
 }  // namespace internal
 }  // namespace arrow
diff --git a/cpp/src/arrow/util/int_util_test.cc b/cpp/src/arrow/util/int_util_test.cc
index fe0ebf7c709..20e6a2eb42d 100644
--- a/cpp/src/arrow/util/int_util_test.cc
+++ b/cpp/src/arrow/util/int_util_test.cc
@@ -386,19 +386,19 @@ TEST(TransposeInts, Int8ToInt64) {
   ASSERT_EQ(dest, std::vector<int64_t>({2222, 4444, 6666, 1111, 4444, 3333}));
 }
 
-void BoundscheckPasses(const std::shared_ptr<DataType>& type,
+void BoundsCheckPasses(const std::shared_ptr<DataType>& type,
                        const std::string& indices_json, uint64_t upper_limit) {
   auto indices = ArrayFromJSON(type, indices_json);
-  ASSERT_OK(IndexBoundscheck(*indices->data(), upper_limit));
+  ASSERT_OK(IndexBoundsCheck(*indices->data(), upper_limit));
 }
 
-void BoundscheckFails(const std::shared_ptr<DataType>& type,
+void BoundsCheckFails(const std::shared_ptr<DataType>& type,
                       const std::string& indices_json, uint64_t upper_limit) {
   auto indices = ArrayFromJSON(type, indices_json);
-  ASSERT_RAISES(IndexError, IndexBoundscheck(*indices->data(), upper_limit));
+  ASSERT_RAISES(IndexError, IndexBoundsCheck(*indices->data(), upper_limit));
 }
 
-TEST(IndexBoundscheck, Batching) {
+TEST(IndexBoundsCheck, Batching) {
   auto rand = random::RandomArrayGenerator(/*seed=*/0);
 
   const int64_t length = 200;
@@ -411,76 +411,76 @@ TEST(IndexBoundscheck, Batching) {
   uint8_t* bitmap = index_data->buffers[0]->mutable_data();
   BitUtil::SetBitsTo(bitmap, 0, length, true);
 
-  ASSERT_OK(IndexBoundscheck(*index_data, 1));
+  ASSERT_OK(IndexBoundsCheck(*index_data, 1));
 
   // We'll place out of bounds indices at various locations
   values[99] = 1;
-  ASSERT_RAISES(IndexError, IndexBoundscheck(*index_data, 1));
+  ASSERT_RAISES(IndexError, IndexBoundsCheck(*index_data, 1));
 
   // Make that value null
   BitUtil::ClearBit(bitmap, 99);
-  ASSERT_OK(IndexBoundscheck(*index_data, 1));
+  ASSERT_OK(IndexBoundsCheck(*index_data, 1));
 
   values[199] = 1;
-  ASSERT_RAISES(IndexError, IndexBoundscheck(*index_data, 1));
+  ASSERT_RAISES(IndexError, IndexBoundsCheck(*index_data, 1));
 
   // Make that value null
   BitUtil::ClearBit(bitmap, 199);
-  ASSERT_OK(IndexBoundscheck(*index_data, 1));
+  ASSERT_OK(IndexBoundsCheck(*index_data, 1));
 }
 
-TEST(IndexBoundscheck, SignedInts) {
+TEST(IndexBoundsCheck, SignedInts) {
   auto CheckCommon = [&](const std::shared_ptr<DataType>& ty) {
-    BoundscheckPasses(ty, "[0, 0, 0]", 1);
-    BoundscheckFails(ty, "[0, 0, 0]", 0);
-    BoundscheckFails(ty, "[-1]", 1);
-    BoundscheckFails(ty, "[-128]", 1);
-    BoundscheckFails(ty, "[0, 100, 127]", 127);
-    BoundscheckPasses(ty, "[0, 100, 127]", 128);
+    BoundsCheckPasses(ty, "[0, 0, 0]", 1);
+    BoundsCheckFails(ty, "[0, 0, 0]", 0);
+    BoundsCheckFails(ty, "[-1]", 1);
+    BoundsCheckFails(ty, "[-128]", 1);
+    BoundsCheckFails(ty, "[0, 100, 127]", 127);
+    BoundsCheckPasses(ty, "[0, 100, 127]", 128);
   };
 
   CheckCommon(int8());
 
   CheckCommon(int16());
-  BoundscheckPasses(int16(), "[0, 999, 999]", 1000);
-  BoundscheckFails(int16(), "[0, 1000, 1000]", 1000);
-  BoundscheckPasses(int16(), "[0, 32767]", 1 << 15);
+  BoundsCheckPasses(int16(), "[0, 999, 999]", 1000);
+  BoundsCheckFails(int16(), "[0, 1000, 1000]", 1000);
+  BoundsCheckPasses(int16(), "[0, 32767]", 1 << 15);
 
   CheckCommon(int32());
-  BoundscheckPasses(int32(), "[0, 999999, 999999]", 1000000);
-  BoundscheckFails(int32(), "[0, 1000000, 1000000]", 1000000);
-  BoundscheckPasses(int32(), "[0, 2147483647]", 1LL << 31);
+  BoundsCheckPasses(int32(), "[0, 999999, 999999]", 1000000);
+  BoundsCheckFails(int32(), "[0, 1000000, 1000000]", 1000000);
+  BoundsCheckPasses(int32(), "[0, 2147483647]", 1LL << 31);
 
   CheckCommon(int64());
-  BoundscheckPasses(int64(), "[0, 9999999999, 9999999999]", 10000000000LL);
-  BoundscheckFails(int64(), "[0, 10000000000, 10000000000]", 10000000000LL);
+  BoundsCheckPasses(int64(), "[0, 9999999999, 9999999999]", 10000000000LL);
+  BoundsCheckFails(int64(), "[0, 10000000000, 10000000000]", 10000000000LL);
 }
 
-TEST(IndexBoundscheck, UnsignedInts) {
+TEST(IndexBoundsCheck, UnsignedInts) {
   auto CheckCommon = [&](const std::shared_ptr<DataType>& ty) {
-    BoundscheckPasses(ty, "[0, 0, 0]", 1);
-    BoundscheckFails(ty, "[0, 0, 0]", 0);
-    BoundscheckFails(ty, "[0, 100, 200]", 200);
-    BoundscheckPasses(ty, "[0, 100, 200]", 201);
+    BoundsCheckPasses(ty, "[0, 0, 0]", 1);
+    BoundsCheckFails(ty, "[0, 0, 0]", 0);
+    BoundsCheckFails(ty, "[0, 100, 200]", 200);
+    BoundsCheckPasses(ty, "[0, 100, 200]", 201);
   };
 
   CheckCommon(uint8());
-  BoundscheckPasses(uint8(), "[255, 255, 255]", 1000);
-  BoundscheckFails(uint8(), "[255, 255, 255]", 255);
+  BoundsCheckPasses(uint8(), "[255, 255, 255]", 1000);
+  BoundsCheckFails(uint8(), "[255, 255, 255]", 255);
 
   CheckCommon(uint16());
-  BoundscheckPasses(uint16(), "[0, 999, 999]", 1000);
-  BoundscheckFails(uint16(), "[0, 1000, 1000]", 1000);
-  BoundscheckPasses(uint16(), "[0, 65535]", 1 << 16);
+  BoundsCheckPasses(uint16(), "[0, 999, 999]", 1000);
+  BoundsCheckFails(uint16(), "[0, 1000, 1000]", 1000);
+  BoundsCheckPasses(uint16(), "[0, 65535]", 1 << 16);
 
   CheckCommon(uint32());
-  BoundscheckPasses(uint32(), "[0, 999999, 999999]", 1000000);
-  BoundscheckFails(uint32(), "[0, 1000000, 1000000]", 1000000);
-  BoundscheckPasses(uint32(), "[0, 4294967295]", 1LL << 32);
+  BoundsCheckPasses(uint32(), "[0, 999999, 999999]", 1000000);
+  BoundsCheckFails(uint32(), "[0, 1000000, 1000000]", 1000000);
+  BoundsCheckPasses(uint32(), "[0, 4294967295]", 1LL << 32);
 
   CheckCommon(uint64());
-  BoundscheckPasses(uint64(), "[0, 9999999999, 9999999999]", 10000000000LL);
-  BoundscheckFails(uint64(), "[0, 10000000000, 10000000000]", 10000000000LL);
+  BoundsCheckPasses(uint64(), "[0, 9999999999, 9999999999]", 10000000000LL);
+  BoundsCheckFails(uint64(), "[0, 10000000000, 10000000000]", 10000000000LL);
 }
 
 }  // namespace internal
diff --git a/python/pyarrow/tests/test_pandas.py b/python/pyarrow/tests/test_pandas.py
index fc7dd1deddb..130ac4a8ec7 100644
--- a/python/pyarrow/tests/test_pandas.py
+++ b/python/pyarrow/tests/test_pandas.py
@@ -3403,7 +3403,6 @@ def test_variable_dictionary_to_pandas():
     expected_dict = pa.concat_arrays([d1, d2, d3])
 
     a = pa.chunked_array([a1, a2, a3, a4])
-
     a_dense = pa.chunked_array([a1.cast('string'),
                                 a2.cast('string'),
                                 a3.cast('string'),

From 8af5d3ebed1b0484608876e1aff384b780319ac3 Mon Sep 17 00:00:00 2001
From: Wes McKinney <wesm@apache.org>
Date: Wed, 10 Jun 2020 10:08:16 -0500
Subject: [PATCH 5/8] Add additional test where indices do not have nulls for
 better coverage

---
 cpp/src/arrow/compute/kernels/vector_take_test.cc | 7 +++++--
 1 file changed, 5 insertions(+), 2 deletions(-)

diff --git a/cpp/src/arrow/compute/kernels/vector_take_test.cc b/cpp/src/arrow/compute/kernels/vector_take_test.cc
index 9f1dd59220c..c93eb6bae72 100644
--- a/cpp/src/arrow/compute/kernels/vector_take_test.cc
+++ b/cpp/src/arrow/compute/kernels/vector_take_test.cc
@@ -151,7 +151,10 @@ void CheckTakeRandom(const std::shared_ptr<Array>& values, int64_t indices_lengt
   IndexCType max_index = GetMaxIndex<IndexCType>(values->length());
   auto indices = rand->Numeric<IndexType>(indices_length, static_cast<IndexCType>(0),
                                           max_index, null_probability);
+  auto indices_no_nulls = rand->Numeric<IndexType>(
+      indices_length, static_cast<IndexCType>(0), max_index, /*null_probability=*/0.0);
   ValidateTake<ValuesType>(values, indices);
+  ValidateTake<ValuesType>(values, indices_no_nulls);
 }
 
 template <typename ValuesType, typename DataGenerator>
@@ -159,8 +162,8 @@ void DoRandomTakeTests(DataGenerator&& generate_values) {
   auto rand = random::RandomArrayGenerator(kRandomSeed);
   for (size_t i = 5; i < 10; i++) {
     const int64_t length = static_cast<int64_t>(1ULL << i);
-    for (size_t j = 4; j < 10; j++) {
-      const int64_t indices_length = static_cast<int64_t>(1ULL << j);
+    for (size_t j = 7; j < 10; j++) {
+      const int64_t indices_length = static_cast<int64_t>(1ULL << j) - 1;
       for (auto null_probability : {0.0, 0.01, 0.25, 0.999, 1.0}) {
         auto values = generate_values(length, null_probability, &rand);
         CheckTakeRandom<ValuesType, Int8Type>(values, indices_length, null_probability,

From e54f1932f9509853c33199b21d61ae6fe043f704 Mon Sep 17 00:00:00 2001
From: Antoine Pitrou <antoine@python.org>
Date: Thu, 11 Jun 2020 12:32:50 +0200
Subject: [PATCH 6/8] Less benchmarks

---
 cpp/src/arrow/compute/benchmark_util.h                      | 2 +-
 cpp/src/arrow/compute/kernels/vector_selection_benchmark.cc | 4 ++--
 2 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/cpp/src/arrow/compute/benchmark_util.h b/cpp/src/arrow/compute/benchmark_util.h
index 05cd2814bff..1259d1b5468 100644
--- a/cpp/src/arrow/compute/benchmark_util.h
+++ b/cpp/src/arrow/compute/benchmark_util.h
@@ -59,7 +59,7 @@ void BenchmarkSetArgsWithSizes(benchmark::internal::Benchmark* bench,
   // 0 is treated as "no nulls"
   for (const auto size : sizes) {
     for (const auto inverse_null_proportion :
-         std::vector<ArgsType>({10000, 100, 50, 10, 2, 1, 0})) {
+         std::vector<ArgsType>({10000, 100, 10, 2, 1, 0})) {
       bench->Args({static_cast<ArgsType>(size), inverse_null_proportion});
     }
   }
diff --git a/cpp/src/arrow/compute/kernels/vector_selection_benchmark.cc b/cpp/src/arrow/compute/kernels/vector_selection_benchmark.cc
index 75fd5716d4b..c0b542cdf21 100644
--- a/cpp/src/arrow/compute/kernels/vector_selection_benchmark.cc
+++ b/cpp/src/arrow/compute/kernels/vector_selection_benchmark.cc
@@ -39,7 +39,7 @@ struct FilterParams {
   const double filter_null_proportion;
 };
 
-std::vector<int64_t> g_data_sizes = {kL1Size, 1 << 20};
+std::vector<int64_t> g_data_sizes = {kL1Size};
 
 // The benchmark state parameter references this vector of cases. Test high and
 // low selectivity filters.
@@ -259,7 +259,7 @@ BENCHMARK(FilterStringFilterWithNulls)->Apply(FilterSetArgs);
 
 void TakeSetArgs(benchmark::internal::Benchmark* bench) {
   for (int64_t size : g_data_sizes) {
-    for (auto nulls : std::vector<ArgsType>({1000, 100, 50, 10, 1, 0})) {
+    for (auto nulls : std::vector<ArgsType>({1000, 10, 2, 1, 0})) {
       bench->Args({static_cast<ArgsType>(size), nulls});
     }
   }

From b5b3eea222f419e0d2a9389c4be6ab614efc0210 Mon Sep 17 00:00:00 2001
From: Antoine Pitrou <antoine@python.org>
Date: Thu, 11 Jun 2020 16:38:10 +0200
Subject: [PATCH 7/8] Some fixes

---
 cpp/src/arrow/compute/kernels/vector_take.cc  | 35 ++++++++++++-------
 .../arrow/compute/kernels/vector_take_test.cc | 31 +++++++++++-----
 cpp/src/arrow/util/bit_block_counter.cc       |  1 +
 cpp/src/arrow/util/bit_block_counter.h        |  1 -
 cpp/src/arrow/util/int_util.cc                |  2 ++
 5 files changed, 47 insertions(+), 23 deletions(-)

diff --git a/cpp/src/arrow/compute/kernels/vector_take.cc b/cpp/src/arrow/compute/kernels/vector_take.cc
index fdf175ccc4a..22d74fd0ca1 100644
--- a/cpp/src/arrow/compute/kernels/vector_take.cc
+++ b/cpp/src/arrow/compute/kernels/vector_take.cc
@@ -103,7 +103,7 @@ PrimitiveTakeArgs GetPrimitiveTakeArgs(const ExecBatch& batch) {
 }
 
 /// \brief The Take implementation for primitive (fixed-width) types does not
-/// use the logical Arrow type but rather then physical C type. This way we
+/// use the logical Arrow type but rather the physical C type. This way we
 /// only generate one take function for each byte width.
 ///
 /// This function assumes that the indices have been boundschecked.
@@ -153,6 +153,8 @@ struct PrimitiveTakeImpl {
               // index is not null
               BitUtil::SetBit(out_bitmap, out_offset + position);
               out[position] = values[indices[position]];
+            } else {
+              out[position] = ValueCType{};
             }
             ++position;
           }
@@ -169,6 +171,8 @@ struct PrimitiveTakeImpl {
               out[position] = values[indices[position]];
               BitUtil::SetBit(out_bitmap, out_offset + position);
               ++valid_count;
+            } else {
+              out[position] = ValueCType{};
             }
             ++position;
           }
@@ -177,18 +181,19 @@ struct PrimitiveTakeImpl {
           // random access in general we have to check the value nullness one by
           // one.
           for (int64_t i = 0; i < block.length; ++i) {
-            if (BitUtil::GetBit(indices_bitmap, indices_offset + position)) {
-              // index is not null
-              if (BitUtil::GetBit(values_bitmap, values_offset + indices[position])) {
-                // value is not null
-                out[position] = values[indices[position]];
-                BitUtil::SetBit(out_bitmap, out_offset + position);
-                ++valid_count;
-              }
+            if (BitUtil::GetBit(indices_bitmap, indices_offset + position) &&
+                BitUtil::GetBit(values_bitmap, values_offset + indices[position])) {
+              // index is not null && value is not null
+              out[position] = values[indices[position]];
+              BitUtil::SetBit(out_bitmap, out_offset + position);
+              ++valid_count;
+            } else {
+              out[position] = ValueCType{};
             }
             ++position;
           }
         } else {
+          memset(out + position, 0, sizeof(ValueCType) * block.length);
           position += block.length;
         }
       }
@@ -219,6 +224,8 @@ struct BooleanTakeImpl {
     if (args.values_null_count > 0 || args.indices_null_count > 0) {
       BitUtil::SetBitsTo(out_bitmap, out_offset, args.indices_length, false);
     }
+    // Avoid uninitialized data in values array
+    BitUtil::SetBitsTo(out, out_offset, args.indices_length, false);
 
     auto PlaceDataBit = [&](int64_t loc, IndexCType index) {
       BitUtil::SetBitTo(out, out_offset + loc,
@@ -423,20 +430,20 @@ struct GenericTakeImpl {
   template <typename IndexCType, typename ValidVisitor, typename NullVisitor>
   Status VisitIndices(ValidVisitor&& visit_valid, NullVisitor&& visit_null) {
     const auto indices_values = indices->GetValues<IndexCType>(1);
-    const int64_t indices_offset = indices->offset;
     const uint8_t* bitmap = nullptr;
     if (indices->buffers[0]) {
       bitmap = indices->buffers[0]->data();
     }
-    OptionalBitIndexer indices_is_valid(indices->buffers[0], values->offset);
+    OptionalBitIndexer indices_is_valid(indices->buffers[0], indices->offset);
     OptionalBitIndexer values_is_valid(values->buffers[0], values->offset);
+    const bool values_have_nulls = (values->GetNullCount() > 0);
 
-    OptionalBitBlockCounter bit_counter(bitmap, indices_offset, indices->length);
+    OptionalBitBlockCounter bit_counter(bitmap, indices->offset, indices->length);
     int64_t position = 0;
     while (position < indices->length) {
       BitBlockCount block = bit_counter.NextBlock();
       const bool indices_have_nulls = block.popcount < block.length;
-      if (!indices_have_nulls && values->GetNullCount() == 0) {
+      if (!indices_have_nulls && !values_have_nulls) {
         // Fastest path, neither indices nor values have nulls
         validity_builder.UnsafeAppend(block.length, true);
         for (int64_t i = 0; i < block.length; ++i) {
@@ -750,6 +757,8 @@ struct StructTakeImpl : public GenericTakeImpl<StructTakeImpl, StructType> {
   }
 };
 
+#undef LIFT_BASE_MEMBERS
+
 template <typename Impl>
 static void GenericTakeExec(KernelContext* ctx, const ExecBatch& batch, Datum* out) {
   const auto& state = checked_cast<const TakeState&>(*ctx->state());
diff --git a/cpp/src/arrow/compute/kernels/vector_take_test.cc b/cpp/src/arrow/compute/kernels/vector_take_test.cc
index c93eb6bae72..4abaf03279e 100644
--- a/cpp/src/arrow/compute/kernels/vector_take_test.cc
+++ b/cpp/src/arrow/compute/kernels/vector_take_test.cc
@@ -24,6 +24,7 @@
 
 #include "arrow/compute/api.h"
 #include "arrow/compute/kernels/test_util.h"
+#include "arrow/pretty_print.h"
 #include "arrow/table.h"
 #include "arrow/testing/gtest_common.h"
 #include "arrow/testing/gtest_util.h"
@@ -84,11 +85,12 @@ void ValidateTakeImpl(const std::shared_ptr<Array>& values,
   auto typed_indices = checked_pointer_cast<IndexArrayType>(indices);
   for (int64_t i = 0; i < indices->length(); ++i) {
     if (typed_indices->IsNull(i) || typed_values->IsNull(typed_indices->Value(i))) {
-      ASSERT_TRUE(result->IsNull(i));
-      continue;
+      ASSERT_TRUE(result->IsNull(i)) << i;
+    } else {
+      ASSERT_FALSE(result->IsNull(i)) << i;
+      ASSERT_EQ(typed_result->GetView(i), typed_values->GetView(typed_indices->Value(i)))
+          << i;
     }
-    ASSERT_EQ(typed_result->GetView(i), typed_values->GetView(typed_indices->Value(i)))
-        << i;
   }
 }
 
@@ -155,16 +157,21 @@ void CheckTakeRandom(const std::shared_ptr<Array>& values, int64_t indices_lengt
       indices_length, static_cast<IndexCType>(0), max_index, /*null_probability=*/0.0);
   ValidateTake<ValuesType>(values, indices);
   ValidateTake<ValuesType>(values, indices_no_nulls);
+  // Sliced indices array
+  if (indices_length >= 2) {
+    indices = indices->Slice(1, indices_length - 2);
+    indices_no_nulls = indices_no_nulls->Slice(1, indices_length - 2);
+    ValidateTake<ValuesType>(values, indices);
+    ValidateTake<ValuesType>(values, indices_no_nulls);
+  }
 }
 
 template <typename ValuesType, typename DataGenerator>
 void DoRandomTakeTests(DataGenerator&& generate_values) {
   auto rand = random::RandomArrayGenerator(kRandomSeed);
-  for (size_t i = 5; i < 10; i++) {
-    const int64_t length = static_cast<int64_t>(1ULL << i);
-    for (size_t j = 7; j < 10; j++) {
-      const int64_t indices_length = static_cast<int64_t>(1ULL << j) - 1;
-      for (auto null_probability : {0.0, 0.01, 0.25, 0.999, 1.0}) {
+  for (const int64_t length : {1, 16, 59}) {
+    for (const int64_t indices_length : {0, 5, 30}) {
+      for (const auto null_probability : {0.0, 0.05, 0.25, 0.95, 1.0}) {
         auto values = generate_values(length, null_probability, &rand);
         CheckTakeRandom<ValuesType, Int8Type>(values, indices_length, null_probability,
                                               &rand);
@@ -182,6 +189,12 @@ void DoRandomTakeTests(DataGenerator&& generate_values) {
                                                 &rand);
         CheckTakeRandom<ValuesType, UInt64Type>(values, indices_length, null_probability,
                                                 &rand);
+        // Sliced values array
+        if (length > 2) {
+          values = values->Slice(1, length - 2);
+          CheckTakeRandom<ValuesType, UInt64Type>(values, indices_length,
+                                                  null_probability, &rand);
+        }
       }
     }
   }
diff --git a/cpp/src/arrow/util/bit_block_counter.cc b/cpp/src/arrow/util/bit_block_counter.cc
index fd94b95a306..648b46b844c 100644
--- a/cpp/src/arrow/util/bit_block_counter.cc
+++ b/cpp/src/arrow/util/bit_block_counter.cc
@@ -21,6 +21,7 @@
 #include <cstdint>
 #include <type_traits>
 
+#include "arrow/buffer.h"
 #include "arrow/util/bit_util.h"
 #include "arrow/util/ubsan.h"
 
diff --git a/cpp/src/arrow/util/bit_block_counter.h b/cpp/src/arrow/util/bit_block_counter.h
index 85a5bf49f85..d6318e7b6d3 100644
--- a/cpp/src/arrow/util/bit_block_counter.h
+++ b/cpp/src/arrow/util/bit_block_counter.h
@@ -22,7 +22,6 @@
 #include <limits>
 #include <memory>
 
-#include "arrow/array/array_base.h"
 #include "arrow/util/bit_util.h"
 #include "arrow/util/bitmap_ops.h"
 #include "arrow/util/ubsan.h"
diff --git a/cpp/src/arrow/util/int_util.cc b/cpp/src/arrow/util/int_util.cc
index e85abb86fcd..d6e3c1e129b 100644
--- a/cpp/src/arrow/util/int_util.cc
+++ b/cpp/src/arrow/util/int_util.cc
@@ -21,6 +21,8 @@
 #include <cstring>
 #include <limits>
 
+#include "arrow/array/data.h"
+#include "arrow/type.h"
 #include "arrow/util/bit_block_counter.h"
 #include "arrow/util/bit_util.h"
 #include "arrow/util/logging.h"

From eaf6e1fdd087cfb703aca026a02a1ea2405a5651 Mon Sep 17 00:00:00 2001
From: Antoine Pitrou <antoine@python.org>
Date: Thu, 11 Jun 2020 16:59:07 +0200
Subject: [PATCH 8/8] Presize binary builder with a heuristic, and use more
 interesting benchmark size

---
 .../compute/kernels/vector_selection_benchmark.cc  |  2 +-
 cpp/src/arrow/compute/kernels/vector_take.cc       | 14 ++++++++++++--
 cpp/src/arrow/compute/kernels/vector_take_test.cc  |  1 -
 3 files changed, 13 insertions(+), 4 deletions(-)

diff --git a/cpp/src/arrow/compute/kernels/vector_selection_benchmark.cc b/cpp/src/arrow/compute/kernels/vector_selection_benchmark.cc
index c0b542cdf21..d03fa574f17 100644
--- a/cpp/src/arrow/compute/kernels/vector_selection_benchmark.cc
+++ b/cpp/src/arrow/compute/kernels/vector_selection_benchmark.cc
@@ -39,7 +39,7 @@ struct FilterParams {
   const double filter_null_proportion;
 };
 
-std::vector<int64_t> g_data_sizes = {kL1Size};
+std::vector<int64_t> g_data_sizes = {kL2Size};
 
 // The benchmark state parameter references this vector of cases. Test high and
 // low selectivity filters.
diff --git a/cpp/src/arrow/compute/kernels/vector_take.cc b/cpp/src/arrow/compute/kernels/vector_take.cc
index 22d74fd0ca1..536797cbc51 100644
--- a/cpp/src/arrow/compute/kernels/vector_take.cc
+++ b/cpp/src/arrow/compute/kernels/vector_take.cc
@@ -545,8 +545,16 @@ struct VarBinaryTakeImpl : public GenericTakeImpl<VarBinaryTakeImpl<Type>, Type>
   Status ProcessIndices() {
     ValuesArrayType typed_values(this->values_as_binary);
 
-    // Start out with at least 64K allocated
-    RETURN_NOT_OK(data_builder.Reserve(1 << 16));
+    // Presize the data builder with a rough estimate of the required data size
+    const auto values_length = values->length;
+    const auto mean_value_length =
+        (values_length > 0) ? ((typed_values.raw_value_offsets()[values_length] -
+                                typed_values.raw_value_offsets()[0]) /
+                               static_cast<double>(values_length))
+                            : 0.0;
+    RETURN_NOT_OK(data_builder.Reserve(static_cast<int64_t>(
+        mean_value_length * (indices->length - indices->GetNullCount()))));
+
     int64_t space_available = data_builder.capacity();
 
     offset_type offset = 0;
@@ -640,6 +648,8 @@ struct ListTakeImpl : public GenericTakeImpl<ListTakeImpl<Type>, Type> {
   Status ProcessIndices() {
     ValuesArrayType typed_values(this->values);
 
+    // TODO presize child_index_builder with a similar heuristic as VarBinaryTakeImpl
+
     offset_type offset = 0;
     auto PushValidIndex = [&](IndexCType index) {
       offset_builder.UnsafeAppend(offset);
diff --git a/cpp/src/arrow/compute/kernels/vector_take_test.cc b/cpp/src/arrow/compute/kernels/vector_take_test.cc
index 4abaf03279e..51d183c6013 100644
--- a/cpp/src/arrow/compute/kernels/vector_take_test.cc
+++ b/cpp/src/arrow/compute/kernels/vector_take_test.cc
@@ -24,7 +24,6 @@
 
 #include "arrow/compute/api.h"
 #include "arrow/compute/kernels/test_util.h"
-#include "arrow/pretty_print.h"
 #include "arrow/table.h"
 #include "arrow/testing/gtest_common.h"
 #include "arrow/testing/gtest_util.h"