From 0b5ac64acc33da9d4ffa1cb4f11c9da15cb24c4d Mon Sep 17 00:00:00 2001
From: Wes McKinney <wesm@apache.org>
Date: Tue, 16 Jun 2020 19:29:33 -0500
Subject: [PATCH 01/11] Benchmark enhancements

---
 .../kernels/scalar_compare_benchmark.cc       | 56 ++++++++++++-------
 1 file changed, 35 insertions(+), 21 deletions(-)

diff --git a/cpp/src/arrow/compute/kernels/scalar_compare_benchmark.cc b/cpp/src/arrow/compute/kernels/scalar_compare_benchmark.cc
index f7d74b4843e..8af575f5874 100644
--- a/cpp/src/arrow/compute/kernels/scalar_compare_benchmark.cc
+++ b/cpp/src/arrow/compute/kernels/scalar_compare_benchmark.cc
@@ -30,37 +30,51 @@ namespace compute {
 
 constexpr auto kSeed = 0x94378165;
 
-static void CompareArrayScalarKernel(benchmark::State& state) {
-  RegressionArgs args(state);
-  const int64_t array_size = args.size / sizeof(int64_t);
+template <CompareOperator op, typename Type>
+static void CompareArrayScalar(benchmark::State& state) {
+  RegressionArgs args(state, /*size_is_bytes=*/false);
+  auto ty = TypeTraits<Type>::type_singleton();
   auto rand = random::RandomArrayGenerator(kSeed);
-  auto array = std::static_pointer_cast<NumericArray<Int64Type>>(
-      rand.Int64(array_size, -100, 100, args.null_proportion));
-
-  CompareOptions ge{GREATER_EQUAL};
-
+  auto array = rand.ArrayOf(ty, args.size, args.null_proportion);
+  auto scalar = *rand.ArrayOf(ty, 1, 0)->GetScalar(0);
   for (auto _ : state) {
-    ABORT_NOT_OK(Compare(array, Datum(int64_t(0)), ge).status());
+    ABORT_NOT_OK(Compare(array, Datum(scalar), CompareOptions(op)).status());
   }
 }
 
-static void CompareArrayArrayKernel(benchmark::State& state) {
-  RegressionArgs args(state);
-  const int64_t array_size = args.size / sizeof(int64_t);
+template <CompareOperator op, typename Type>
+static void CompareArrayArray(benchmark::State& state) {
+  RegressionArgs args(state, /*size_is_bytes=*/false);
+  auto ty = TypeTraits<Type>::type_singleton();
   auto rand = random::RandomArrayGenerator(kSeed);
-  auto lhs = std::static_pointer_cast<NumericArray<Int64Type>>(
-      rand.Int64(array_size, -100, 100, args.null_proportion));
-  auto rhs = std::static_pointer_cast<NumericArray<Int64Type>>(
-      rand.Int64(array_size, -100, 100, args.null_proportion));
-
-  CompareOptions ge(GREATER_EQUAL);
+  auto lhs = rand.ArrayOf(ty, args.size, args.null_proportion);
+  auto rhs = rand.ArrayOf(ty, args.size, args.null_proportion);
   for (auto _ : state) {
-    ABORT_NOT_OK(Compare(lhs, rhs, ge).status());
+    ABORT_NOT_OK(Compare(lhs, rhs, CompareOptions(op)).status());
   }
 }
 
-BENCHMARK(CompareArrayScalarKernel)->Apply(RegressionSetArgs);
-BENCHMARK(CompareArrayArrayKernel)->Apply(RegressionSetArgs);
+static void GreaterArrayArrayInt64(benchmark::State& state) {
+  CompareArrayArray<GREATER, Int64Type>(state);
+}
+
+static void GreaterArrayScalarInt64(benchmark::State& state) {
+  CompareArrayScalar<GREATER, Int64Type>(state);
+}
+
+static void GreaterArrayArrayString(benchmark::State& state) {
+  CompareArrayArray<GREATER, StringType>(state);
+}
+
+static void GreaterArrayScalarString(benchmark::State& state) {
+  CompareArrayScalar<GREATER, StringType>(state);
+}
+
+BENCHMARK(GreaterArrayArrayInt64)->Apply(RegressionSetArgs);
+BENCHMARK(GreaterArrayScalarInt64)->Apply(RegressionSetArgs);
+
+BENCHMARK(GreaterArrayArrayString)->Apply(RegressionSetArgs);
+BENCHMARK(GreaterArrayScalarString)->Apply(RegressionSetArgs);
 
 }  // namespace compute
 }  // namespace arrow

From 0d5ed54f209cc351360e4e66141080580d23e05b Mon Sep 17 00:00:00 2001
From: Wes McKinney <wesm@apache.org>
Date: Tue, 16 Jun 2020 10:48:22 -0500
Subject: [PATCH 02/11] Work on paring down comparisons

Finish paring down compare, implement needed simplification of kernel generators

Benchmark improvements
---
 .../arrow/compute/kernels/codegen_internal.cc |  11 +-
 .../arrow/compute/kernels/codegen_internal.h  | 125 +++++++++++++++---
 .../arrow/compute/kernels/scalar_compare.cc   |  98 ++++++++------
 cpp/src/arrow/compute/kernels/vector_sort.cc  |   2 +-
 cpp/src/arrow/scalar.h                        |  19 +--
 5 files changed, 177 insertions(+), 78 deletions(-)

diff --git a/cpp/src/arrow/compute/kernels/codegen_internal.cc b/cpp/src/arrow/compute/kernels/codegen_internal.cc
index 427ad2b0c54..2606c978e39 100644
--- a/cpp/src/arrow/compute/kernels/codegen_internal.cc
+++ b/cpp/src/arrow/compute/kernels/codegen_internal.cc
@@ -31,11 +31,12 @@ void ExecFail(KernelContext* ctx, const ExecBatch& batch, Datum* out) {
   ctx->SetStatus(Status::NotImplemented("This kernel is malformed"));
 }
 
-void BinaryExecFlipped(KernelContext* ctx, ArrayKernelExec exec, const ExecBatch& batch,
-                       Datum* out) {
-  ExecBatch flipped_batch = batch;
-  std::swap(flipped_batch.values[0], flipped_batch.values[1]);
-  exec(ctx, flipped_batch, out);
+ArrayKernelExec MakeFlippedBinaryExec(ArrayKernelExec exec) {
+  return [exec](KernelContext* ctx, const ExecBatch& batch, Datum* out) {
+    ExecBatch flipped_batch = batch;
+    std::swap(flipped_batch.values[0], flipped_batch.values[1]);
+    exec(ctx, flipped_batch, out);
+  };
 }
 
 std::vector<std::shared_ptr<DataType>> g_signed_int_types;
diff --git a/cpp/src/arrow/compute/kernels/codegen_internal.h b/cpp/src/arrow/compute/kernels/codegen_internal.h
index c547c807757..87d2366f428 100644
--- a/cpp/src/arrow/compute/kernels/codegen_internal.h
+++ b/cpp/src/arrow/compute/kernels/codegen_internal.h
@@ -23,8 +23,10 @@
 #include <utility>
 #include <vector>
 
+#include "arrow/array/builder_binary.h"
 #include "arrow/array/data.h"
 #include "arrow/buffer.h"
+#include "arrow/buffer_builder.h"
 #include "arrow/compute/exec.h"
 #include "arrow/compute/kernel.h"
 #include "arrow/datum.h"
@@ -121,10 +123,26 @@ struct ArrayIterator<Type, enable_if_boolean<Type>> {
 
 template <typename Type>
 struct ArrayIterator<Type, enable_if_base_binary<Type>> {
-  int64_t position = 0;
-  typename TypeTraits<Type>::ArrayType arr;
-  explicit ArrayIterator(const ArrayData& data) : arr(data.Copy()) {}
-  util::string_view operator()() { return arr.GetView(position++); }
+  using offset_type = typename Type::offset_type;
+  const ArrayData& arr;
+  const offset_type* offsets;
+  offset_type cur_offset;
+  const char* data;
+  int64_t position;
+  explicit ArrayIterator(const ArrayData& arr)
+      : arr(arr),
+        offsets(reinterpret_cast<const offset_type*>(arr.buffers[1]->data()) +
+                arr.offset),
+        cur_offset(offsets[0]),
+        data(reinterpret_cast<const char*>(arr.buffers[2]->data())),
+        position(0) {}
+
+  util::string_view operator()() {
+    offset_type next_offset = offsets[position++ + 1];
+    auto result = util::string_view(data + cur_offset, next_offset - cur_offset);
+    cur_offset = next_offset;
+    return result;
+  }
 };
 
 template <typename Type, typename Enable = void>
@@ -229,8 +247,7 @@ Result<ValueDescr> FirstType(KernelContext*, const std::vector<ValueDescr>& desc
 
 void ExecFail(KernelContext* ctx, const ExecBatch& batch, Datum* out);
 
-void BinaryExecFlipped(KernelContext* ctx, ArrayKernelExec exec, const ExecBatch& batch,
-                       Datum* out);
+ArrayKernelExec MakeFlippedBinaryExec(ArrayKernelExec exec);
 
 // ----------------------------------------------------------------------
 // Helpers for iterating over common DataType instances for adding kernels to
@@ -485,18 +502,38 @@ struct ScalarUnaryNotNullStateful {
   struct ArrayExec<Type, enable_if_base_binary<Type>> {
     static void Exec(const ThisType& functor, KernelContext* ctx, const ExecBatch& batch,
                      Datum* out) {
-      typename TypeTraits<Type>::BuilderType builder;
-      VisitArrayValuesInline<Arg0Type>(*batch[0].array(), [&](util::optional<ARG0> v) {
-        if (v.has_value()) {
-          KERNEL_RETURN_IF_ERROR(ctx, builder.Append(functor.op.Call(ctx, *v)));
-        } else {
-          KERNEL_RETURN_IF_ERROR(ctx, builder.AppendNull());
+      using offset_type = typename Type::offset_type;
+      const ArrayData& arg0 = *batch[0].array();
+      typename TypeTraits<Type>::ArrayType arg0_boxed(batch[0].array());
+
+      ArrayData* out_arr = out->mutable_array();
+      TypedBufferBuilder<offset_type> offset_builder;
+      TypedBufferBuilder<uint8_t> data_builder;
+
+      KERNEL_RETURN_IF_ERROR(ctx, offset_builder.Reserve(arg0.length + 1));
+      offset_type offset = 0;
+
+      const offset_type* in_offsets = arg0_boxed.raw_value_offsets();
+      const uint8_t* in_data = arg0.buffers[2]->data();
+      offset_type cur_offset = in_offsets[0];
+      for (int64_t i = 0; i < arg0.length; ++i) {
+        offset_type next_offset = in_offsets[i + 1];
+        if (arg0_boxed.IsValid(i)) {
+          auto val_size = next_offset - cur_offset;
+          auto val =
+              functor.op.Call(ctx, util::string_view(in_data + cur_offset, val_size));
+          if (std::is_same<offset_type, int32_t>::value &&
+              (static_cast<int64_t>(offset) + val_size > kBinaryMemoryLimit)) {
+            ctx->SetStatus(Status::OutOfMemory("Overflowed 32-bit binary builder"));
+            return;
+          }
+          KERNEL_RETURN_IF_ERROR(ctx, data_builder.Append(val.data(), val_size));
         }
-      });
+        cur_offset = next_offset;
+      }
       if (!ctx->HasError()) {
-        std::shared_ptr<ArrayData> result;
-        ctx->SetStatus(builder.FinishInternal(&result));
-        out->value = std::move(result);
+        KERNEL_RETURN_IF_ERROR(ctx, offset_builder.Finish(&out_arr->buffers[1]));
+        KERNEL_RETURN_IF_ERROR(ctx, data_builder.Finish(&out_arr->buffers[2]));
       }
     }
   };
@@ -787,6 +824,41 @@ ArrayKernelExec Integer(detail::GetTypeId get_id) {
   }
 }
 
+template <template <typename...> class Generator, typename Type0, typename... Args>
+ArrayKernelExec IntegerBased(detail::GetTypeId get_id) {
+  switch (get_id.id) {
+    case Type::INT8:
+      return Generator<Type0, Int8Type, Args...>::Exec;
+    case Type::INT16:
+      return Generator<Type0, Int16Type, Args...>::Exec;
+    case Type::INT32:
+      return Generator<Type0, Int32Type, Args...>::Exec;
+    case Type::INT64:
+      return Generator<Type0, Int64Type, Args...>::Exec;
+    case Type::UINT8:
+      return Generator<Type0, UInt8Type, Args...>::Exec;
+    case Type::UINT16:
+      return Generator<Type0, UInt16Type, Args...>::Exec;
+    case Type::UINT32:
+      return Generator<Type0, UInt32Type, Args...>::Exec;
+    case Type::UINT64:
+      return Generator<Type0, UInt64Type, Args...>::Exec;
+    case Type::DATE32:
+      return Generator<Type0, Int32Type, Args...>::Exec;
+    case Type::DATE64:
+      return Generator<Type0, Int64Type, Args...>::Exec;
+    case Type::TIMESTAMP:
+      return Generator<Type0, Int64Type, Args...>::Exec;
+    case Type::TIME32:
+      return Generator<Type0, Int32Type, Args...>::Exec;
+    case Type::TIME64:
+      return Generator<Type0, Int64Type, Args...>::Exec;
+    default:
+      DCHECK(false);
+      return ExecFail;
+  }
+}
+
 // Generate a kernel given a templated functor for integer types
 //
 // See "Numeric" above for description of the generator functor
@@ -807,11 +879,30 @@ ArrayKernelExec SignedInteger(detail::GetTypeId get_id) {
   }
 }
 
-// Generate a kernel given a templated functor for base binary types
+// Generate a kernel given a templated functor for base binary types. Generates
+// a single kernel for binary/string and large binary / large string. If your
+// kernel implementation needs access to the specific type at compile time,
+// please use BaseBinarySpecific.
 //
 // See "Numeric" above for description of the generator functor
 template <template <typename...> class Generator, typename Type0, typename... Args>
 ArrayKernelExec BaseBinary(detail::GetTypeId get_id) {
+  switch (get_id.id) {
+    case Type::BINARY:
+    case Type::STRING:
+      return Generator<Type0, BinaryType, Args...>::Exec;
+    case Type::LARGE_BINARY:
+    case Type::LARGE_STRING:
+      return Generator<Type0, LargeBinaryType, Args...>::Exec;
+    default:
+      DCHECK(false);
+      return ExecFail;
+  }
+}
+
+// See BaseBinary documentation
+template <template <typename...> class Generator, typename Type0, typename... Args>
+ArrayKernelExec BaseBinarySpecific(detail::GetTypeId get_id) {
   switch (get_id.id) {
     case Type::BINARY:
       return Generator<Type0, BinaryType, Args...>::Exec;
diff --git a/cpp/src/arrow/compute/kernels/scalar_compare.cc b/cpp/src/arrow/compute/kernels/scalar_compare.cc
index d782a3d41e7..e4faaf01b97 100644
--- a/cpp/src/arrow/compute/kernels/scalar_compare.cc
+++ b/cpp/src/arrow/compute/kernels/scalar_compare.cc
@@ -26,6 +26,8 @@ using util::string_view;
 namespace compute {
 namespace internal {
 
+namespace {
+
 struct Equal {
   template <typename T>
   static constexpr bool Call(KernelContext*, const T& left, const T& right) {
@@ -40,20 +42,6 @@ struct NotEqual {
   }
 };
 
-struct Greater {
-  template <typename T>
-  static constexpr bool Call(KernelContext*, const T& left, const T& right) {
-    return left > right;
-  }
-};
-
-struct GreaterEqual {
-  template <typename T>
-  static constexpr bool Call(KernelContext*, const T& left, const T& right) {
-    return left >= right;
-  }
-};
-
 struct Less {
   template <typename T>
   static constexpr bool Call(KernelContext*, const T& left, const T& right) {
@@ -68,33 +56,45 @@ struct LessEqual {
   }
 };
 
-template <typename InType, typename Op>
-void AddCompare(const std::shared_ptr<DataType>& ty, ScalarFunction* func) {
-  ArrayKernelExec exec = codegen::ScalarBinaryEqualTypes<BooleanType, InType, Op>::Exec;
-  DCHECK_OK(func->AddKernel({ty, ty}, boolean(), exec));
-}
+// Implement Greater, GreaterEqual by flipping arguments to Less, LessEqual
 
 template <typename Op>
-void AddTimestampComparisons(ScalarFunction* func) {
-  ArrayKernelExec exec =
-      codegen::ScalarBinaryEqualTypes<BooleanType, TimestampType, Op>::Exec;
-  for (auto unit : {TimeUnit::SECOND, TimeUnit::MILLI, TimeUnit::MICRO, TimeUnit::NANO}) {
-    InputType in_type(match::TimestampUnit(unit));
-    DCHECK_OK(func->AddKernel({in_type, in_type}, boolean(), exec));
-  }
+void AddIntegerCompare(const std::shared_ptr<DataType>& ty, ScalarFunction* func) {
+  auto exec =
+      codegen::IntegerBased<codegen::ScalarBinaryEqualTypes, BooleanType, Op>(*ty);
+  DCHECK_OK(func->AddKernel({ty, ty}, boolean(), std::move(exec)));
+}
+
+template <typename InType, typename Op>
+void AddGenericCompare(const std::shared_ptr<DataType>& ty, ScalarFunction* func) {
+  DCHECK_OK(
+      func->AddKernel({ty, ty}, boolean(),
+                      codegen::ScalarBinaryEqualTypes<BooleanType, InType, Op>::Exec));
 }
 
 template <typename Op>
-void MakeCompareFunction(std::string name, FunctionRegistry* registry) {
+std::shared_ptr<ScalarFunction> MakeCompareFunction(std::string name) {
   auto func = std::make_shared<ScalarFunction>(name, Arity::Binary());
 
   DCHECK_OK(func->AddKernel(
       {boolean(), boolean()}, boolean(),
       codegen::ScalarBinary<BooleanType, BooleanType, BooleanType, Op>::Exec));
 
-  for (const std::shared_ptr<DataType>& ty : NumericTypes()) {
-    auto exec = codegen::Numeric<codegen::ScalarBinaryEqualTypes, BooleanType, Op>(*ty);
-    DCHECK_OK(func->AddKernel({ty, ty}, boolean(), exec));
+  for (const std::shared_ptr<DataType>& ty : IntTypes()) {
+    AddIntegerCompare<Op>(ty, func.get());
+  }
+  AddIntegerCompare<Op>(date32(), func.get());
+  AddIntegerCompare<Op>(date64(), func.get());
+
+  AddGenericCompare<FloatType, Op>(float32(), func.get());
+  AddGenericCompare<DoubleType, Op>(float64(), func.get());
+
+  // Add timestamp kernels
+  for (auto unit : {TimeUnit::SECOND, TimeUnit::MILLI, TimeUnit::MICRO, TimeUnit::NANO}) {
+    InputType in_type(match::TimestampUnit(unit));
+    auto exec =
+        codegen::IntegerBased<codegen::ScalarBinaryEqualTypes, BooleanType, Op>(int64());
+    DCHECK_OK(func->AddKernel({in_type, in_type}, boolean(), exec));
   }
   for (const std::shared_ptr<DataType>& ty : BaseBinaryTypes()) {
     auto exec =
@@ -102,24 +102,36 @@ void MakeCompareFunction(std::string name, FunctionRegistry* registry) {
     DCHECK_OK(func->AddKernel({ty, ty}, boolean(), exec));
   }
 
-  // Temporal types requires some care because cross-unit comparisons with
-  // everything but DATE32 and DATE64 are not implemented yet
-  AddCompare<Date32Type, Op>(date32(), func.get());
-  AddCompare<Date64Type, Op>(date64(), func.get());
-  AddTimestampComparisons<Op>(func.get());
-
   // TODO: Leave time32, time64, and duration for follow up work
 
-  DCHECK_OK(registry->AddFunction(std::move(func)));
+  return func;
 }
 
+std::shared_ptr<ScalarFunction> MakeFlippedFunction(std::string name,
+                                                    const ScalarFunction& func) {
+  auto flipped_func = std::make_shared<ScalarFunction>(name, Arity::Binary());
+  for (const ScalarKernel* kernel : func.kernels()) {
+    ScalarKernel flipped_kernel = *kernel;
+    flipped_kernel.exec = MakeFlippedBinaryExec(kernel->exec);
+    DCHECK_OK(flipped_func->AddKernel(std::move(flipped_kernel)));
+  }
+  return flipped_func;
+}
+
+}  // namespace
+
 void RegisterScalarComparison(FunctionRegistry* registry) {
-  MakeCompareFunction<Equal>("equal", registry);
-  MakeCompareFunction<NotEqual>("not_equal", registry);
-  MakeCompareFunction<Less>("less", registry);
-  MakeCompareFunction<LessEqual>("less_equal", registry);
-  MakeCompareFunction<Greater>("greater", registry);
-  MakeCompareFunction<GreaterEqual>("greater_equal", registry);
+  DCHECK_OK(registry->AddFunction(MakeCompareFunction<Equal>("equal")));
+  DCHECK_OK(registry->AddFunction(MakeCompareFunction<NotEqual>("not_equal")));
+
+  auto less = MakeCompareFunction<Less>("less");
+  auto greater = MakeFlippedFunction("greater", *less);
+  auto less_equal = MakeCompareFunction<LessEqual>("less_equal");
+  auto greater_equal = MakeFlippedFunction("greater_equal", *less_equal);
+  DCHECK_OK(registry->AddFunction(std::move(less)));
+  DCHECK_OK(registry->AddFunction(std::move(less_equal)));
+  DCHECK_OK(registry->AddFunction(std::move(greater)));
+  DCHECK_OK(registry->AddFunction(std::move(greater_equal)));
 }
 
 }  // namespace internal
diff --git a/cpp/src/arrow/compute/kernels/vector_sort.cc b/cpp/src/arrow/compute/kernels/vector_sort.cc
index 2e2a4454e8b..4ae4355f7f8 100644
--- a/cpp/src/arrow/compute/kernels/vector_sort.cc
+++ b/cpp/src/arrow/compute/kernels/vector_sort.cc
@@ -269,7 +269,7 @@ void AddSortingKernels(VectorKernel base, VectorFunction* func) {
   }
   for (const auto& ty : BaseBinaryTypes()) {
     base.signature = KernelSignature::Make({InputType::Array(ty)}, uint64());
-    base.exec = codegen::BaseBinary<ExecTemplate, UInt64Type>(*ty);
+    base.exec = codegen::BaseBinarySpecific<ExecTemplate, UInt64Type>(*ty);
     DCHECK_OK(func->AddKernel(base));
   }
 }
diff --git a/cpp/src/arrow/scalar.h b/cpp/src/arrow/scalar.h
index b011f27c93a..cc2b7b20614 100644
--- a/cpp/src/arrow/scalar.h
+++ b/cpp/src/arrow/scalar.h
@@ -105,6 +105,11 @@ struct ARROW_EXPORT PrimitiveScalar : public Scalar {
     ARROW_CHECK_EQ(this->type->id(), T::type_id);
   }
 
+  explicit PrimitiveScalar(std::shared_ptr<DataType> type)
+      : Scalar(std::move(type), false) {
+    ARROW_CHECK_EQ(this->type->id(), T::type_id);
+  }
+
   explicit PrimitiveScalar(ValueType value)
       : PrimitiveScalar(value, TypeTraits<T>::type_singleton()) {}
 
@@ -238,18 +243,8 @@ struct ARROW_EXPORT FixedSizeBinaryScalar : public BinaryScalar {
 };
 
 template <typename T>
-struct ARROW_EXPORT TemporalScalar : public Scalar {
-  using Scalar::Scalar;
-  using TypeClass = T;
-  using ValueType = typename T::c_type;
-
-  TemporalScalar(ValueType value, std::shared_ptr<DataType> type)
-      : Scalar(std::move(type), true), value(value) {}
-
-  explicit TemporalScalar(std::shared_ptr<DataType> type)
-      : Scalar(std::move(type), false) {}
-
-  ValueType value;
+struct ARROW_EXPORT TemporalScalar : public internal::PrimitiveScalar<T> {
+  using internal::PrimitiveScalar<T>::PrimitiveScalar;
 };
 
 template <typename T>

From 4ec79f682780c138955b9399835a5e7e52dd471c Mon Sep 17 00:00:00 2001
From: Wes McKinney <wesm@apache.org>
Date: Tue, 16 Jun 2020 19:45:39 -0500
Subject: [PATCH 03/11] 'Fix' not-null binary value iteration, note that it
 isn't sufficiently tested

---
 cpp/src/arrow/compute/kernels/codegen_internal.h | 10 ++++++++--
 1 file changed, 8 insertions(+), 2 deletions(-)

diff --git a/cpp/src/arrow/compute/kernels/codegen_internal.h b/cpp/src/arrow/compute/kernels/codegen_internal.h
index 87d2366f428..5cad9995581 100644
--- a/cpp/src/arrow/compute/kernels/codegen_internal.h
+++ b/cpp/src/arrow/compute/kernels/codegen_internal.h
@@ -502,6 +502,8 @@ struct ScalarUnaryNotNullStateful {
   struct ArrayExec<Type, enable_if_base_binary<Type>> {
     static void Exec(const ThisType& functor, KernelContext* ctx, const ExecBatch& batch,
                      Datum* out) {
+      // TODO: This code path is currently inadequately tested.
+
       using offset_type = typename Type::offset_type;
       const ArrayData& arg0 = *batch[0].array();
       typename TypeTraits<Type>::ArrayType arg0_boxed(batch[0].array());
@@ -511,26 +513,30 @@ struct ScalarUnaryNotNullStateful {
       TypedBufferBuilder<uint8_t> data_builder;
 
       KERNEL_RETURN_IF_ERROR(ctx, offset_builder.Reserve(arg0.length + 1));
-      offset_type offset = 0;
+      offset_type out_offset = 0;
 
       const offset_type* in_offsets = arg0_boxed.raw_value_offsets();
       const uint8_t* in_data = arg0.buffers[2]->data();
       offset_type cur_offset = in_offsets[0];
       for (int64_t i = 0; i < arg0.length; ++i) {
+        offset_builder.UnsafeAppend(out_offset);
         offset_type next_offset = in_offsets[i + 1];
         if (arg0_boxed.IsValid(i)) {
           auto val_size = next_offset - cur_offset;
           auto val =
               functor.op.Call(ctx, util::string_view(in_data + cur_offset, val_size));
           if (std::is_same<offset_type, int32_t>::value &&
-              (static_cast<int64_t>(offset) + val_size > kBinaryMemoryLimit)) {
+              (static_cast<int64_t>(out_offset) + val_size > kBinaryMemoryLimit)) {
             ctx->SetStatus(Status::OutOfMemory("Overflowed 32-bit binary builder"));
             return;
           }
           KERNEL_RETURN_IF_ERROR(ctx, data_builder.Append(val.data(), val_size));
+          out_offset += val_size;
         }
         cur_offset = next_offset;
       }
+      // Last offset
+      offset_builder.UnsafeAppend(out_offset);
       if (!ctx->HasError()) {
         KERNEL_RETURN_IF_ERROR(ctx, offset_builder.Finish(&out_arr->buffers[1]));
         KERNEL_RETURN_IF_ERROR(ctx, data_builder.Finish(&out_arr->buffers[2]));

From 66bdd6a828b1444f754f2587b86779f1b4d22a2b Mon Sep 17 00:00:00 2001
From: Wes McKinney <wesm@apache.org>
Date: Tue, 16 Jun 2020 19:47:17 -0500
Subject: [PATCH 04/11] Move a couple functions

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

diff --git a/cpp/src/arrow/compute/kernels/scalar_compare.cc b/cpp/src/arrow/compute/kernels/scalar_compare.cc
index e4faaf01b97..6aed91b810a 100644
--- a/cpp/src/arrow/compute/kernels/scalar_compare.cc
+++ b/cpp/src/arrow/compute/kernels/scalar_compare.cc
@@ -94,12 +94,12 @@ std::shared_ptr<ScalarFunction> MakeCompareFunction(std::string name) {
     InputType in_type(match::TimestampUnit(unit));
     auto exec =
         codegen::IntegerBased<codegen::ScalarBinaryEqualTypes, BooleanType, Op>(int64());
-    DCHECK_OK(func->AddKernel({in_type, in_type}, boolean(), exec));
+    DCHECK_OK(func->AddKernel({in_type, in_type}, boolean(), std::move(exec)));
   }
   for (const std::shared_ptr<DataType>& ty : BaseBinaryTypes()) {
     auto exec =
         codegen::BaseBinary<codegen::ScalarBinaryEqualTypes, BooleanType, Op>(*ty);
-    DCHECK_OK(func->AddKernel({ty, ty}, boolean(), exec));
+    DCHECK_OK(func->AddKernel({ty, ty}, boolean(), std::move(exec)));
   }
 
   // TODO: Leave time32, time64, and duration for follow up work

From 0a687379adbe7e546714457fc013b7b8e9e0ee60 Mon Sep 17 00:00:00 2001
From: Wes McKinney <wesm@apache.org>
Date: Tue, 16 Jun 2020 20:15:23 -0500
Subject: [PATCH 05/11] Arrange that temporal types have a common base type
 with their integer storage siblings

---
 cpp/src/arrow/scalar.h | 43 +++++++++++++++++++++++++++---------------
 cpp/src/arrow/type.h   | 10 ++++++++++
 2 files changed, 38 insertions(+), 15 deletions(-)

diff --git a/cpp/src/arrow/scalar.h b/cpp/src/arrow/scalar.h
index cc2b7b20614..11bded3408c 100644
--- a/cpp/src/arrow/scalar.h
+++ b/cpp/src/arrow/scalar.h
@@ -101,19 +101,10 @@ struct ARROW_EXPORT PrimitiveScalar : public Scalar {
 
   // Non-null constructor.
   PrimitiveScalar(ValueType value, std::shared_ptr<DataType> type)
-      : Scalar(std::move(type), true), value(value) {
-    ARROW_CHECK_EQ(this->type->id(), T::type_id);
-  }
+      : Scalar(std::move(type), true), value(value) {}
 
   explicit PrimitiveScalar(std::shared_ptr<DataType> type)
-      : Scalar(std::move(type), false) {
-    ARROW_CHECK_EQ(this->type->id(), T::type_id);
-  }
-
-  explicit PrimitiveScalar(ValueType value)
-      : PrimitiveScalar(value, TypeTraits<T>::type_singleton()) {}
-
-  PrimitiveScalar() : Scalar(TypeTraits<T>::type_singleton()) {}
+      : Scalar(std::move(type), false) {}
 
   ValueType value{};
 };
@@ -121,12 +112,26 @@ struct ARROW_EXPORT PrimitiveScalar : public Scalar {
 }  // namespace internal
 
 struct ARROW_EXPORT BooleanScalar : public internal::PrimitiveScalar<BooleanType, bool> {
-  using internal::PrimitiveScalar<BooleanType, bool>::PrimitiveScalar;
+  using Base = internal::PrimitiveScalar<BooleanType, bool>;
+  using Base::Base;
+
+  explicit BooleanScalar(bool value)
+    : Base(value, boolean()) {}
+
+  BooleanScalar() : Base(boolean()) {}
 };
 
 template <typename T>
 struct NumericScalar : public internal::PrimitiveScalar<T> {
-  using internal::PrimitiveScalar<T>::PrimitiveScalar;
+  using Base = typename internal::PrimitiveScalar<T>;
+  using Base::Base;
+  using TypeClass = typename Base::TypeClass;
+  using ValueType = typename Base::ValueType;
+
+  explicit NumericScalar(ValueType value)
+      : Base(value, TypeTraits<T>::type_singleton()) {}
+
+  NumericScalar() : Base(TypeTraits<T>::type_singleton()) {}
 };
 
 struct ARROW_EXPORT Int8Scalar : public NumericScalar<Int8Type> {
@@ -242,9 +247,17 @@ struct ARROW_EXPORT FixedSizeBinaryScalar : public BinaryScalar {
   explicit FixedSizeBinaryScalar(std::shared_ptr<DataType> type) : BinaryScalar(type) {}
 };
 
+template <typename T, typename StorageType = typename T::StorageType,
+          typename Enable = void>
+struct ARROW_EXPORT TemporalScalar : internal::PrimitiveScalar<StorageType> {
+  using internal::PrimitiveScalar<StorageType>::PrimitiveScalar;
+  using TypeClass = T;
+};
+
 template <typename T>
-struct ARROW_EXPORT TemporalScalar : public internal::PrimitiveScalar<T> {
+struct ARROW_EXPORT TemporalScalar<T, void, void> : internal::PrimitiveScalar<T> {
   using internal::PrimitiveScalar<T>::PrimitiveScalar;
+  using TypeClass = T;
 };
 
 template <typename T>
@@ -253,7 +266,7 @@ struct ARROW_EXPORT DateScalar : public TemporalScalar<T> {
   using ValueType = typename TemporalScalar<T>::ValueType;
 
   explicit DateScalar(ValueType value)
-      : TemporalScalar<T>(std::move(value), TypeTraits<T>::type_singleton()) {}
+    : TemporalScalar<T>(std::move(value), TypeTraits<T>::type_singleton()) {}
   DateScalar() : TemporalScalar<T>(TypeTraits<T>::type_singleton()) {}
 };
 
diff --git a/cpp/src/arrow/type.h b/cpp/src/arrow/type.h
index a95f585c54d..e5f1f9a8b6a 100644
--- a/cpp/src/arrow/type.h
+++ b/cpp/src/arrow/type.h
@@ -485,6 +485,7 @@ class ARROW_EXPORT CTypeImpl : public BASE {
  public:
   static constexpr Type::type type_id = TYPE_ID;
   using c_type = C_TYPE;
+  using StorageType = DERIVED;
 
   CTypeImpl() : BASE(TYPE_ID) {}
 
@@ -1124,6 +1125,7 @@ class ARROW_EXPORT Date32Type : public DateType {
   static constexpr Type::type type_id = Type::DATE32;
   static constexpr DateUnit UNIT = DateUnit::DAY;
   using c_type = int32_t;
+  using StorageType = Int32Type;
 
   static constexpr const char* type_name() { return "date32"; }
 
@@ -1146,6 +1148,7 @@ class ARROW_EXPORT Date64Type : public DateType {
   static constexpr Type::type type_id = Type::DATE64;
   static constexpr DateUnit UNIT = DateUnit::MILLI;
   using c_type = int64_t;
+  using StorageType = Int64Type;
 
   static constexpr const char* type_name() { return "date64"; }
 
@@ -1183,6 +1186,7 @@ class ARROW_EXPORT Time32Type : public TimeType {
  public:
   static constexpr Type::type type_id = Type::TIME32;
   using c_type = int32_t;
+  using StorageType = Int32Type;
 
   static constexpr const char* type_name() { return "time32"; }
 
@@ -1201,6 +1205,7 @@ class ARROW_EXPORT Time64Type : public TimeType {
  public:
   static constexpr Type::type type_id = Type::TIME64;
   using c_type = int64_t;
+  using StorageType = Int64Type;
 
   static constexpr const char* type_name() { return "time64"; }
 
@@ -1251,6 +1256,7 @@ class ARROW_EXPORT TimestampType : public TemporalType, public ParametricType {
 
   static constexpr Type::type type_id = Type::TIMESTAMP;
   using c_type = int64_t;
+  using StorageType = Int64Type;
 
   static constexpr const char* type_name() { return "timestamp"; }
 
@@ -1296,6 +1302,7 @@ class ARROW_EXPORT MonthIntervalType : public IntervalType {
  public:
   static constexpr Type::type type_id = Type::INTERVAL_MONTHS;
   using c_type = int32_t;
+  using StorageType = Int32Type;
 
   static constexpr const char* type_name() { return "month_interval"; }
 
@@ -1324,6 +1331,8 @@ class ARROW_EXPORT DayTimeIntervalType : public IntervalType {
     }
   };
   using c_type = DayMilliseconds;
+  using StorageType = DayTimeIntervalType;
+
   static_assert(sizeof(DayMilliseconds) == 8,
                 "DayMilliseconds struct assumed to be of size 8 bytes");
   static constexpr Type::type type_id = Type::INTERVAL_DAY_TIME;
@@ -1347,6 +1356,7 @@ class ARROW_EXPORT DurationType : public TemporalType, public ParametricType {
 
   static constexpr Type::type type_id = Type::DURATION;
   using c_type = int64_t;
+  using StorageType = Int64Type;
 
   static constexpr const char* type_name() { return "duration"; }
 

From 4409558a3da2e8b878d3dd06e7909c850234d490 Mon Sep 17 00:00:00 2001
From: Wes McKinney <wesm@apache.org>
Date: Tue, 16 Jun 2020 21:01:45 -0500
Subject: [PATCH 06/11] Add missing unit tests for temporal comparisons

---
 cpp/src/arrow/compute/kernel.cc               | 81 ++++++++++++-------
 cpp/src/arrow/compute/kernel.h                |  5 +-
 cpp/src/arrow/compute/kernel_test.cc          | 18 ++---
 .../arrow/compute/kernels/codegen_internal.h  |  2 +-
 .../arrow/compute/kernels/scalar_compare.cc   | 25 +++++-
 .../compute/kernels/scalar_compare_test.cc    | 61 +++++++++-----
 cpp/src/arrow/scalar.h                        |  5 +-
 7 files changed, 129 insertions(+), 68 deletions(-)

diff --git a/cpp/src/arrow/compute/kernel.cc b/cpp/src/arrow/compute/kernel.cc
index 4747e63c159..7bb31392c9d 100644
--- a/cpp/src/arrow/compute/kernel.cc
+++ b/cpp/src/arrow/compute/kernel.cc
@@ -112,42 +112,61 @@ std::shared_ptr<TypeMatcher> SameTypeId(Type::type type_id) {
   return std::make_shared<SameTypeIdMatcher>(type_id);
 }
 
-class TimestampUnitMatcher : public TypeMatcher {
- public:
-  explicit TimestampUnitMatcher(TimeUnit::type accepted_unit)
-      : accepted_unit_(accepted_unit) {}
-
-  bool Matches(const DataType& type) const override {
-    if (type.id() != Type::TIMESTAMP) {
-      return false;
-    }
-    const auto& ts_type = checked_cast<const TimestampType&>(type);
-    return ts_type.unit() == accepted_unit_;
+#define TIME_UNIT_MATCHER(NAME)                                         \
+  class NAME##TypeUnitMatcher : public TypeMatcher {                    \
+   public:                                                              \
+    explicit NAME##TypeUnitMatcher(TimeUnit::type accepted_unit)        \
+        : accepted_unit_(accepted_unit) {}                              \
+                                                                        \
+    bool Matches(const DataType& type) const override {                 \
+      if (type.id() != NAME##Type::type_id) {                           \
+        return false;                                                   \
+      }                                                                 \
+      const auto& subtype = checked_cast<const NAME##Type&>(type);      \
+      return subtype.unit() == accepted_unit_;                          \
+    }                                                                   \
+                                                                        \
+    bool Equals(const TypeMatcher& other) const override {              \
+      if (this == &other) {                                             \
+        return true;                                                    \
+      }                                                                 \
+      auto casted = dynamic_cast<const NAME##TypeUnitMatcher*>(&other); \
+      if (casted == nullptr) {                                          \
+        return false;                                                   \
+      }                                                                 \
+      return this->accepted_unit_ == casted->accepted_unit_;            \
+    }                                                                   \
+                                                                        \
+    std::string ToString() const override {                             \
+      std::stringstream ss;                                             \
+      ss << NAME##Type::type_name() << "("                              \
+         << ::arrow::internal::ToString(accepted_unit_) << ")";         \
+      return ss.str();                                                  \
+    }                                                                   \
+                                                                        \
+   private:                                                             \
+    TimeUnit::type accepted_unit_;                                      \
   }
 
-  bool Equals(const TypeMatcher& other) const override {
-    if (this == &other) {
-      return true;
-    }
-    auto casted = dynamic_cast<const TimestampUnitMatcher*>(&other);
-    if (casted == nullptr) {
-      return false;
-    }
-    return this->accepted_unit_ == casted->accepted_unit_;
-  }
+TIME_UNIT_MATCHER(Timestamp);
+TIME_UNIT_MATCHER(Time32);
+TIME_UNIT_MATCHER(Time64);
+TIME_UNIT_MATCHER(Duration);
 
-  std::string ToString() const override {
-    std::stringstream ss;
-    ss << "timestamp(" << ::arrow::internal::ToString(accepted_unit_) << ")";
-    return ss.str();
-  }
+std::shared_ptr<TypeMatcher> TimestampTypeUnit(TimeUnit::type unit) {
+  return std::make_shared<TimestampTypeUnitMatcher>(unit);
+}
 
- private:
-  TimeUnit::type accepted_unit_;
-};
+std::shared_ptr<TypeMatcher> Time32TypeUnit(TimeUnit::type unit) {
+  return std::make_shared<Time32TypeUnitMatcher>(unit);
+}
+
+std::shared_ptr<TypeMatcher> Time64TypeUnit(TimeUnit::type unit) {
+  return std::make_shared<Time64TypeUnitMatcher>(unit);
+}
 
-std::shared_ptr<TypeMatcher> TimestampUnit(TimeUnit::type unit) {
-  return std::make_shared<TimestampUnitMatcher>(unit);
+std::shared_ptr<TypeMatcher> DurationTypeUnit(TimeUnit::type unit) {
+  return std::make_shared<DurationTypeUnitMatcher>(unit);
 }
 
 class IntegerMatcher : public TypeMatcher {
diff --git a/cpp/src/arrow/compute/kernel.h b/cpp/src/arrow/compute/kernel.h
index b51efe5a953..68875ccfd66 100644
--- a/cpp/src/arrow/compute/kernel.h
+++ b/cpp/src/arrow/compute/kernel.h
@@ -148,7 +148,10 @@ ARROW_EXPORT std::shared_ptr<TypeMatcher> SameTypeId(Type::type type_id);
 
 /// \brief Match any TimestampType instance having the same unit, but the time
 /// zones can be different.
-ARROW_EXPORT std::shared_ptr<TypeMatcher> TimestampUnit(TimeUnit::type unit);
+ARROW_EXPORT std::shared_ptr<TypeMatcher> TimestampTypeUnit(TimeUnit::type unit);
+ARROW_EXPORT std::shared_ptr<TypeMatcher> Time32TypeUnit(TimeUnit::type unit);
+ARROW_EXPORT std::shared_ptr<TypeMatcher> Time64TypeUnit(TimeUnit::type unit);
+ARROW_EXPORT std::shared_ptr<TypeMatcher> DurationTypeUnit(TimeUnit::type unit);
 
 // \brief Match any integer type
 ARROW_EXPORT std::shared_ptr<TypeMatcher> Integer();
diff --git a/cpp/src/arrow/compute/kernel_test.cc b/cpp/src/arrow/compute/kernel_test.cc
index fbdaf124d81..fc8a9edeb99 100644
--- a/cpp/src/arrow/compute/kernel_test.cc
+++ b/cpp/src/arrow/compute/kernel_test.cc
@@ -45,23 +45,23 @@ TEST(TypeMatcher, SameTypeId) {
   ASSERT_FALSE(matcher->Equals(*match::SameTypeId(Type::TIMESTAMP)));
 }
 
-TEST(TypeMatcher, TimestampUnit) {
-  std::shared_ptr<TypeMatcher> matcher = match::TimestampUnit(TimeUnit::MILLI);
+TEST(TypeMatcher, TimestampTypeUnit) {
+  std::shared_ptr<TypeMatcher> matcher = match::TimestampTypeUnit(TimeUnit::MILLI);
 
   ASSERT_TRUE(matcher->Matches(*timestamp(TimeUnit::MILLI)));
   ASSERT_TRUE(matcher->Matches(*timestamp(TimeUnit::MILLI, "utc")));
   ASSERT_FALSE(matcher->Matches(*timestamp(TimeUnit::SECOND)));
 
   // Check ToString representation
-  ASSERT_EQ("timestamp(s)", match::TimestampUnit(TimeUnit::SECOND)->ToString());
-  ASSERT_EQ("timestamp(ms)", match::TimestampUnit(TimeUnit::MILLI)->ToString());
-  ASSERT_EQ("timestamp(us)", match::TimestampUnit(TimeUnit::MICRO)->ToString());
-  ASSERT_EQ("timestamp(ns)", match::TimestampUnit(TimeUnit::NANO)->ToString());
+  ASSERT_EQ("timestamp(s)", match::TimestampTypeUnit(TimeUnit::SECOND)->ToString());
+  ASSERT_EQ("timestamp(ms)", match::TimestampTypeUnit(TimeUnit::MILLI)->ToString());
+  ASSERT_EQ("timestamp(us)", match::TimestampTypeUnit(TimeUnit::MICRO)->ToString());
+  ASSERT_EQ("timestamp(ns)", match::TimestampTypeUnit(TimeUnit::NANO)->ToString());
 
   // Equals implementation
   ASSERT_TRUE(matcher->Equals(*matcher));
-  ASSERT_TRUE(matcher->Equals(*match::TimestampUnit(TimeUnit::MILLI)));
-  ASSERT_FALSE(matcher->Equals(*match::TimestampUnit(TimeUnit::MICRO)));
+  ASSERT_TRUE(matcher->Equals(*match::TimestampTypeUnit(TimeUnit::MILLI)));
+  ASSERT_FALSE(matcher->Equals(*match::TimestampTypeUnit(TimeUnit::MICRO)));
 }
 
 // ----------------------------------------------------------------------
@@ -135,7 +135,7 @@ TEST(InputType, Constructors) {
   ASSERT_EQ("array[Type::DECIMAL]", ty2_array.ToString());
   ASSERT_EQ("scalar[Type::DECIMAL]", ty2_scalar.ToString());
 
-  InputType ty7(match::TimestampUnit(TimeUnit::MICRO));
+  InputType ty7(match::TimestampTypeUnit(TimeUnit::MICRO));
   ASSERT_EQ("any[timestamp(us)]", ty7.ToString());
 }
 
diff --git a/cpp/src/arrow/compute/kernels/codegen_internal.h b/cpp/src/arrow/compute/kernels/codegen_internal.h
index 5cad9995581..ecc2972b284 100644
--- a/cpp/src/arrow/compute/kernels/codegen_internal.h
+++ b/cpp/src/arrow/compute/kernels/codegen_internal.h
@@ -150,7 +150,7 @@ struct UnboxScalar;
 
 template <typename Type>
 struct UnboxScalar<Type, enable_if_has_c_type<Type>> {
-  using ScalarType = typename TypeTraits<Type>::ScalarType;
+  using ScalarType = ::arrow::internal::PrimitiveScalar<typename Type::StorageType>;
   static typename Type::c_type Unbox(const Datum& datum) {
     return datum.scalar_as<ScalarType>().value;
   }
diff --git a/cpp/src/arrow/compute/kernels/scalar_compare.cc b/cpp/src/arrow/compute/kernels/scalar_compare.cc
index 6aed91b810a..22f07693bb6 100644
--- a/cpp/src/arrow/compute/kernels/scalar_compare.cc
+++ b/cpp/src/arrow/compute/kernels/scalar_compare.cc
@@ -91,11 +91,34 @@ std::shared_ptr<ScalarFunction> MakeCompareFunction(std::string name) {
 
   // Add timestamp kernels
   for (auto unit : {TimeUnit::SECOND, TimeUnit::MILLI, TimeUnit::MICRO, TimeUnit::NANO}) {
-    InputType in_type(match::TimestampUnit(unit));
+    InputType in_type(match::TimestampTypeUnit(unit));
     auto exec =
         codegen::IntegerBased<codegen::ScalarBinaryEqualTypes, BooleanType, Op>(int64());
     DCHECK_OK(func->AddKernel({in_type, in_type}, boolean(), std::move(exec)));
   }
+
+  // Duration
+  for (auto unit : {TimeUnit::SECOND, TimeUnit::MILLI, TimeUnit::MICRO, TimeUnit::NANO}) {
+    InputType in_type(match::DurationTypeUnit(unit));
+    auto exec =
+        codegen::IntegerBased<codegen::ScalarBinaryEqualTypes, BooleanType, Op>(int64());
+    DCHECK_OK(func->AddKernel({in_type, in_type}, boolean(), std::move(exec)));
+  }
+
+  // Time32 and Time64
+  for (auto unit : {TimeUnit::SECOND, TimeUnit::MILLI}) {
+    InputType in_type(match::Time32TypeUnit(unit));
+    auto exec =
+        codegen::IntegerBased<codegen::ScalarBinaryEqualTypes, BooleanType, Op>(int32());
+    DCHECK_OK(func->AddKernel({in_type, in_type}, boolean(), std::move(exec)));
+  }
+  for (auto unit : {TimeUnit::MICRO, TimeUnit::NANO}) {
+    InputType in_type(match::Time64TypeUnit(unit));
+    auto exec =
+        codegen::IntegerBased<codegen::ScalarBinaryEqualTypes, BooleanType, Op>(int64());
+    DCHECK_OK(func->AddKernel({in_type, in_type}, boolean(), std::move(exec)));
+  }
+
   for (const std::shared_ptr<DataType>& ty : BaseBinaryTypes()) {
     auto exec =
         codegen::BaseBinary<codegen::ScalarBinaryEqualTypes, BooleanType, Op>(*ty);
diff --git a/cpp/src/arrow/compute/kernels/scalar_compare_test.cc b/cpp/src/arrow/compute/kernels/scalar_compare_test.cc
index 8bedb965686..c6bd30feed2 100644
--- a/cpp/src/arrow/compute/kernels/scalar_compare_test.cc
+++ b/cpp/src/arrow/compute/kernels/scalar_compare_test.cc
@@ -371,23 +371,55 @@ TYPED_TEST(TestNumericCompareKernel, TestNullScalar) {
 }
 
 TYPED_TEST_SUITE(TestNumericCompareKernel, NumericArrowTypes);
-TYPED_TEST(TestNumericCompareKernel, RandomCompareArrayScalar) {
-  using ScalarType = typename TypeTraits<TypeParam>::ScalarType;
-  using CType = typename TypeTraits<TypeParam>::CType;
 
+template <typename Type>
+void DoRandomCompare(const std::shared_ptr<DataType>& type) {
+  using ScalarType = typename TypeTraits<Type>::ScalarType;
+  using CType = typename TypeTraits<Type>::CType;
   auto rand = random::RandomArrayGenerator(0x5416447);
   for (size_t i = 3; i < 10; i++) {
     for (auto null_probability : {0.0, 0.01, 0.1, 0.25, 0.5, 1.0}) {
       for (auto op : {EQUAL, NOT_EQUAL, GREATER, LESS_EQUAL}) {
         const int64_t length = static_cast<int64_t>(1ULL << i);
-        auto array = Datum(rand.Numeric<TypeParam>(length, 0, 100, null_probability));
-        auto fifty = Datum(std::make_shared<ScalarType>(CType(50)));
+        auto data =
+            rand.Numeric<typename Type::StorageType>(length, 0, 100, null_probability);
+
+        // Create view of data as the type (e.g. timestamp)
+        auto array = Datum(*data->View(type));
+        auto fifty = Datum(std::make_shared<ScalarType>(CType(50), type));
         auto options = CompareOptions(op);
-        ValidateCompare<TypeParam>(options, array, fifty);
-        ValidateCompare<TypeParam>(options, fifty, array);
+        ValidateCompare<Type>(options, array, fifty);
+        ValidateCompare<Type>(options, fifty, array);
       }
     }
   }
+  for (size_t i = 3; i < 5; i++) {
+    for (auto null_probability : {0.0, 0.01, 0.1, 0.25, 0.5, 1.0}) {
+      for (auto op : {EQUAL, NOT_EQUAL, GREATER, LESS_EQUAL}) {
+        const int64_t length = static_cast<int64_t>(1ULL << i);
+        auto lhs = rand.Numeric<typename Type::StorageType>(length << i, 0, 100,
+                                                            null_probability);
+        auto rhs = rand.Numeric<typename Type::StorageType>(length << i, 0, 100,
+                                                            null_probability);
+        auto options = CompareOptions(op);
+        ValidateCompare<Type>(options, *lhs->View(type), *rhs->View(type));
+      }
+    }
+  }
+}
+
+TYPED_TEST(TestNumericCompareKernel, RandomCompare) {
+  DoRandomCompare<TypeParam>(TypeTraits<TypeParam>::type_singleton());
+}
+
+TEST(TestTemporalCompareKernel, RandomCompare) {
+  DoRandomCompare<Date32Type>(date32());
+  DoRandomCompare<Date64Type>(date64());
+  DoRandomCompare<Time32Type>(time32(TimeUnit::SECOND));
+  DoRandomCompare<Time64Type>(time64(TimeUnit::MICRO));
+  DoRandomCompare<TimestampType>(timestamp(TimeUnit::SECOND));
+  DoRandomCompare<TimestampType>(timestamp(TimeUnit::MICRO));
+  DoRandomCompare<DurationType>(duration(TimeUnit::MILLI));
 }
 
 TYPED_TEST(TestNumericCompareKernel, SimpleCompareArrayArray) {
@@ -404,21 +436,6 @@ TYPED_TEST(TestNumericCompareKernel, SimpleCompareArrayArray) {
   ValidateCompare<TypeParam>(lte, "[1,2,3,4,5]", "[2,3,4,5,6]", "[1,1,1,1,1]");
 }
 
-TYPED_TEST(TestNumericCompareKernel, RandomCompareArrayArray) {
-  auto rand = random::RandomArrayGenerator(0x5416447);
-  for (size_t i = 3; i < 5; i++) {
-    for (auto null_probability : {0.0, 0.01, 0.1, 0.25, 0.5, 1.0}) {
-      for (auto op : {EQUAL, NOT_EQUAL, GREATER, LESS_EQUAL}) {
-        const int64_t length = static_cast<int64_t>(1ULL << i);
-        auto lhs = Datum(rand.Numeric<TypeParam>(length << i, 0, 100, null_probability));
-        auto rhs = Datum(rand.Numeric<TypeParam>(length << i, 0, 100, null_probability));
-        auto options = CompareOptions(op);
-        ValidateCompare<TypeParam>(options, lhs, rhs);
-      }
-    }
-  }
-}
-
 TEST(TestCompareTimestamps, Basics) {
   const char* example1_json = R"(["1970-01-01","2000-02-29","1900-02-28"])";
   const char* example2_json = R"(["1970-01-02","2000-02-01","1900-02-28"])";
diff --git a/cpp/src/arrow/scalar.h b/cpp/src/arrow/scalar.h
index 11bded3408c..5e090c7b755 100644
--- a/cpp/src/arrow/scalar.h
+++ b/cpp/src/arrow/scalar.h
@@ -115,8 +115,7 @@ struct ARROW_EXPORT BooleanScalar : public internal::PrimitiveScalar<BooleanType
   using Base = internal::PrimitiveScalar<BooleanType, bool>;
   using Base::Base;
 
-  explicit BooleanScalar(bool value)
-    : Base(value, boolean()) {}
+  explicit BooleanScalar(bool value) : Base(value, boolean()) {}
 
   BooleanScalar() : Base(boolean()) {}
 };
@@ -266,7 +265,7 @@ struct ARROW_EXPORT DateScalar : public TemporalScalar<T> {
   using ValueType = typename TemporalScalar<T>::ValueType;
 
   explicit DateScalar(ValueType value)
-    : TemporalScalar<T>(std::move(value), TypeTraits<T>::type_singleton()) {}
+      : TemporalScalar<T>(std::move(value), TypeTraits<T>::type_singleton()) {}
   DateScalar() : TemporalScalar<T>(TypeTraits<T>::type_singleton()) {}
 };
 

From 471fecb451a11cc5220ec81c3dad082525adc530 Mon Sep 17 00:00:00 2001
From: Wes McKinney <wesm@apache.org>
Date: Tue, 16 Jun 2020 21:40:49 -0500
Subject: [PATCH 07/11] Use >, >= as primary comparators instead of <, <=

---
 .../arrow/compute/kernels/scalar_compare.cc    | 18 ++++++++++--------
 1 file changed, 10 insertions(+), 8 deletions(-)

diff --git a/cpp/src/arrow/compute/kernels/scalar_compare.cc b/cpp/src/arrow/compute/kernels/scalar_compare.cc
index 22f07693bb6..59c4a43bb77 100644
--- a/cpp/src/arrow/compute/kernels/scalar_compare.cc
+++ b/cpp/src/arrow/compute/kernels/scalar_compare.cc
@@ -42,17 +42,17 @@ struct NotEqual {
   }
 };
 
-struct Less {
+struct Greater {
   template <typename T>
   static constexpr bool Call(KernelContext*, const T& left, const T& right) {
-    return left < right;
+    return left > right;
   }
 };
 
-struct LessEqual {
+struct GreaterEqual {
   template <typename T>
   static constexpr bool Call(KernelContext*, const T& left, const T& right) {
-    return left <= right;
+    return left >= right;
   }
 };
 
@@ -147,10 +147,12 @@ void RegisterScalarComparison(FunctionRegistry* registry) {
   DCHECK_OK(registry->AddFunction(MakeCompareFunction<Equal>("equal")));
   DCHECK_OK(registry->AddFunction(MakeCompareFunction<NotEqual>("not_equal")));
 
-  auto less = MakeCompareFunction<Less>("less");
-  auto greater = MakeFlippedFunction("greater", *less);
-  auto less_equal = MakeCompareFunction<LessEqual>("less_equal");
-  auto greater_equal = MakeFlippedFunction("greater_equal", *less_equal);
+  // Note: benchmarking suggests that >, >= perform better than <, <=
+  auto greater = MakeCompareFunction<Greater>("greater");
+  auto greater_equal = MakeCompareFunction<GreaterEqual>("greater_equal");
+
+  auto less = MakeFlippedFunction("less", *greater);
+  auto less_equal = MakeFlippedFunction("less_equal", *greater_equal);
   DCHECK_OK(registry->AddFunction(std::move(less)));
   DCHECK_OK(registry->AddFunction(std::move(less_equal)));
   DCHECK_OK(registry->AddFunction(std::move(greater)));

From be6a6ff2743fb7fbd2fb7103e3452f3c514b8f5f Mon Sep 17 00:00:00 2001
From: Wes McKinney <wesm@apache.org>
Date: Tue, 16 Jun 2020 21:41:52 -0500
Subject: [PATCH 08/11] remove comment

---
 cpp/src/arrow/compute/kernels/scalar_compare.cc | 1 -
 1 file changed, 1 deletion(-)

diff --git a/cpp/src/arrow/compute/kernels/scalar_compare.cc b/cpp/src/arrow/compute/kernels/scalar_compare.cc
index 59c4a43bb77..7ae344fd742 100644
--- a/cpp/src/arrow/compute/kernels/scalar_compare.cc
+++ b/cpp/src/arrow/compute/kernels/scalar_compare.cc
@@ -147,7 +147,6 @@ void RegisterScalarComparison(FunctionRegistry* registry) {
   DCHECK_OK(registry->AddFunction(MakeCompareFunction<Equal>("equal")));
   DCHECK_OK(registry->AddFunction(MakeCompareFunction<NotEqual>("not_equal")));
 
-  // Note: benchmarking suggests that >, >= perform better than <, <=
   auto greater = MakeCompareFunction<Greater>("greater");
   auto greater_equal = MakeCompareFunction<GreaterEqual>("greater_equal");
 

From 72d5ee66f8ddf8f138801f0f63edd4665aa097d0 Mon Sep 17 00:00:00 2001
From: Antoine Pitrou <antoine@python.org>
Date: Wed, 17 Jun 2020 14:48:32 +0200
Subject: [PATCH 09/11] Replace macro with template class

---
 cpp/src/arrow/compute/kernel.cc      | 78 ++++++++++++++--------------
 cpp/src/arrow/compute/kernel_test.cc |  6 ++-
 2 files changed, 45 insertions(+), 39 deletions(-)

diff --git a/cpp/src/arrow/compute/kernel.cc b/cpp/src/arrow/compute/kernel.cc
index 7bb31392c9d..d52df9a669c 100644
--- a/cpp/src/arrow/compute/kernel.cc
+++ b/cpp/src/arrow/compute/kernel.cc
@@ -112,46 +112,48 @@ std::shared_ptr<TypeMatcher> SameTypeId(Type::type type_id) {
   return std::make_shared<SameTypeIdMatcher>(type_id);
 }
 
-#define TIME_UNIT_MATCHER(NAME)                                         \
-  class NAME##TypeUnitMatcher : public TypeMatcher {                    \
-   public:                                                              \
-    explicit NAME##TypeUnitMatcher(TimeUnit::type accepted_unit)        \
-        : accepted_unit_(accepted_unit) {}                              \
-                                                                        \
-    bool Matches(const DataType& type) const override {                 \
-      if (type.id() != NAME##Type::type_id) {                           \
-        return false;                                                   \
-      }                                                                 \
-      const auto& subtype = checked_cast<const NAME##Type&>(type);      \
-      return subtype.unit() == accepted_unit_;                          \
-    }                                                                   \
-                                                                        \
-    bool Equals(const TypeMatcher& other) const override {              \
-      if (this == &other) {                                             \
-        return true;                                                    \
-      }                                                                 \
-      auto casted = dynamic_cast<const NAME##TypeUnitMatcher*>(&other); \
-      if (casted == nullptr) {                                          \
-        return false;                                                   \
-      }                                                                 \
-      return this->accepted_unit_ == casted->accepted_unit_;            \
-    }                                                                   \
-                                                                        \
-    std::string ToString() const override {                             \
-      std::stringstream ss;                                             \
-      ss << NAME##Type::type_name() << "("                              \
-         << ::arrow::internal::ToString(accepted_unit_) << ")";         \
-      return ss.str();                                                  \
-    }                                                                   \
-                                                                        \
-   private:                                                             \
-    TimeUnit::type accepted_unit_;                                      \
+template <typename ArrowType>
+class TimeUnitMatcher : public TypeMatcher {
+  using ThisType = TimeUnitMatcher<ArrowType>;
+
+ public:
+  explicit TimeUnitMatcher(TimeUnit::type accepted_unit)
+      : accepted_unit_(accepted_unit) {}
+
+  bool Matches(const DataType& type) const override {
+    if (type.id() != ArrowType::type_id) {
+      return false;
+    }
+    const auto& time_type = checked_cast<const ArrowType&>(type);
+    return time_type.unit() == accepted_unit_;
   }
 
-TIME_UNIT_MATCHER(Timestamp);
-TIME_UNIT_MATCHER(Time32);
-TIME_UNIT_MATCHER(Time64);
-TIME_UNIT_MATCHER(Duration);
+  bool Equals(const TypeMatcher& other) const override {
+    if (this == &other) {
+      return true;
+    }
+    auto casted = dynamic_cast<const ThisType*>(&other);
+    if (casted == nullptr) {
+      return false;
+    }
+    return this->accepted_unit_ == casted->accepted_unit_;
+  }
+
+  std::string ToString() const override {
+    std::stringstream ss;
+    ss << ArrowType::type_name() << "(" << ::arrow::internal::ToString(accepted_unit_)
+       << ")";
+    return ss.str();
+  }
+
+ private:
+  TimeUnit::type accepted_unit_;
+};
+
+using DurationTypeUnitMatcher = TimeUnitMatcher<DurationType>;
+using Time32TypeUnitMatcher = TimeUnitMatcher<Time32Type>;
+using Time64TypeUnitMatcher = TimeUnitMatcher<Time64Type>;
+using TimestampTypeUnitMatcher = TimeUnitMatcher<TimestampType>;
 
 std::shared_ptr<TypeMatcher> TimestampTypeUnit(TimeUnit::type unit) {
   return std::make_shared<TimestampTypeUnitMatcher>(unit);
diff --git a/cpp/src/arrow/compute/kernel_test.cc b/cpp/src/arrow/compute/kernel_test.cc
index fc8a9edeb99..2eb7fd11449 100644
--- a/cpp/src/arrow/compute/kernel_test.cc
+++ b/cpp/src/arrow/compute/kernel_test.cc
@@ -46,11 +46,14 @@ TEST(TypeMatcher, SameTypeId) {
 }
 
 TEST(TypeMatcher, TimestampTypeUnit) {
-  std::shared_ptr<TypeMatcher> matcher = match::TimestampTypeUnit(TimeUnit::MILLI);
+  auto matcher = match::TimestampTypeUnit(TimeUnit::MILLI);
+  auto matcher2 = match::Time32TypeUnit(TimeUnit::MILLI);
 
   ASSERT_TRUE(matcher->Matches(*timestamp(TimeUnit::MILLI)));
   ASSERT_TRUE(matcher->Matches(*timestamp(TimeUnit::MILLI, "utc")));
   ASSERT_FALSE(matcher->Matches(*timestamp(TimeUnit::SECOND)));
+  ASSERT_FALSE(matcher->Matches(*time32(TimeUnit::MILLI)));
+  ASSERT_TRUE(matcher2->Matches(*time32(TimeUnit::MILLI)));
 
   // Check ToString representation
   ASSERT_EQ("timestamp(s)", match::TimestampTypeUnit(TimeUnit::SECOND)->ToString());
@@ -62,6 +65,7 @@ TEST(TypeMatcher, TimestampTypeUnit) {
   ASSERT_TRUE(matcher->Equals(*matcher));
   ASSERT_TRUE(matcher->Equals(*match::TimestampTypeUnit(TimeUnit::MILLI)));
   ASSERT_FALSE(matcher->Equals(*match::TimestampTypeUnit(TimeUnit::MICRO)));
+  ASSERT_FALSE(matcher->Equals(*match::Time32TypeUnit(TimeUnit::MILLI)));
 }
 
 // ----------------------------------------------------------------------

From d71e34b0136b52f290652575ed35a182afa9b242 Mon Sep 17 00:00:00 2001
From: Wes McKinney <wesm@apache.org>
Date: Wed, 17 Jun 2020 19:18:16 -0500
Subject: [PATCH 10/11] Rename some things to ease readability, delete some now
 unused kernel codegen stuff

---
 .../arrow/compute/kernels/codegen_internal.h  | 151 ++++++------------
 .../compute/kernels/scalar_cast_boolean.cc    |   7 +-
 .../compute/kernels/scalar_cast_numeric.cc    |   4 +-
 .../compute/kernels/scalar_cast_string.cc     |   2 +-
 .../arrow/compute/kernels/scalar_compare.cc   |  24 ++-
 .../compute/kernels/scalar_compare_test.cc    |  47 +++---
 cpp/src/arrow/compute/kernels/vector_sort.cc  |   4 +-
 cpp/src/arrow/scalar.h                        |   6 +-
 cpp/src/arrow/type.h                          |  18 +--
 9 files changed, 100 insertions(+), 163 deletions(-)

diff --git a/cpp/src/arrow/compute/kernels/codegen_internal.h b/cpp/src/arrow/compute/kernels/codegen_internal.h
index ecc2972b284..aa5292e8411 100644
--- a/cpp/src/arrow/compute/kernels/codegen_internal.h
+++ b/cpp/src/arrow/compute/kernels/codegen_internal.h
@@ -150,7 +150,7 @@ struct UnboxScalar;
 
 template <typename Type>
 struct UnboxScalar<Type, enable_if_has_c_type<Type>> {
-  using ScalarType = ::arrow::internal::PrimitiveScalar<typename Type::StorageType>;
+  using ScalarType = ::arrow::internal::PrimitiveScalar<typename Type::PhysicalType>;
   static typename Type::c_type Unbox(const Datum& datum) {
     return datum.scalar_as<ScalarType>().value;
   }
@@ -325,34 +325,6 @@ void SimpleBinary(KernelContext* ctx, const ExecBatch& batch, Datum* out) {
   }
 }
 
-// A ArrayKernelExec-creation template that iterates over primitive non-boolean
-// inputs and writes into non-boolean primitive outputs.
-//
-// It may be possible to create a more generic template that can deal with any
-// input writing to any output, but we will need to write benchmarks to
-// investigate that on all compiler targets to ensure that the additional
-// template abstractions do not incur performance overhead. This template
-// provides a reference point for performance when there are no templates
-// dealing with value iteration.
-//
-// TODO: Run benchmarks to determine if OutputAdapter is a zero-cost abstraction
-template <typename Op, typename OutType, typename Arg0Type>
-void ScalarPrimitiveExecUnary(KernelContext* ctx, const ExecBatch& batch, Datum* out) {
-  using OUT = typename OutType::c_type;
-  using ARG0 = typename Arg0Type::c_type;
-
-  if (batch[0].kind() == Datum::SCALAR) {
-    ctx->SetStatus(Status::NotImplemented("NYI"));
-  } else {
-    ArrayData* out_arr = out->mutable_array();
-    auto out_data = out_arr->GetMutableValues<OUT>(1);
-    auto arg0_data = batch[0].array()->GetValues<ARG0>(1);
-    for (int64_t i = 0; i < batch.length; ++i) {
-      *out_data++ = Op::template Call<OUT, ARG0>(ctx, *arg0_data++);
-    }
-  }
-}
-
 // OutputAdapter allows passing an inlineable lambda that provides a sequence
 // of output values to write into output memory. Boolean and primitive outputs
 // are currently implemented, and the validity bitmap is presumed to be handled
@@ -687,15 +659,41 @@ struct ScalarBinary {
 template <typename OutType, typename ArgType, typename Op>
 using ScalarBinaryEqualTypes = ScalarBinary<OutType, ArgType, ArgType, Op>;
 
+}  // namespace codegen
+
 // ----------------------------------------------------------------------
-// Dynamic kernel selectors. These functors allow a kernel implementation to be
-// selected given a arrow::DataType instance. Using these functors triggers the
-// corresponding template that generate's the kernel's Exec function to be
-// instantiated
+// BEGIN of kernel generator-dispatchers ("GD")
+//
+// These GD functions instantiate kernel functor templates and select one of
+// the instantiated kernels dynamically based on the data type or Type::type id
+// that is passed. This enables functions to be populated with kernels by
+// looping over vectors of data types rather than using macros or other
+// approaches.
+//
+// The kernel functor must be of the form:
+//
+// template <typename Type0, typename Type1, Args...>
+// struct FUNCTOR {
+//   static void Exec(KernelContext* ctx, const ExecBatch& batch, Datum* out) {
+//     // IMPLEMENTATION
+//   }
+// };
+//
+// When you pass FUNCTOR to a GD function, you must pass at least one static
+// type along with the functor -- this is often the fixed return type of the
+// functor. This Type0 argument is passed as the first argument to the functor
+// during instantiation. The 2nd type passed to the functor is the DataType
+// subclass corresponding to the type passed as argument (not template type) to
+// the function.
+//
+// For example, GenerateNumeric<FUNCTOR, Type0>(int32()) will select a kernel
+// instantiated like FUNCTOR<Type0, Int32Type>. Any additional variadic
+// template arguments will be passed as additional template arguments to the
+// kernel template.
 
 namespace detail {
 
-// Convenience so we can pass DataType or Type::type into these kernel selectors
+// Convenience so we can pass DataType or Type::type for the GD's
 struct GetTypeId {
   Type::type id;
   GetTypeId(const std::shared_ptr<DataType>& type)  // NOLINT implicit construction
@@ -708,57 +706,9 @@ struct GetTypeId {
 
 }  // namespace detail
 
-// Generate a kernel given a functor of type
-//
-// struct OPERATOR_NAME {
-//   template <typename OUT, typename ARG0>
-//   static OUT Call(KernelContext*, ARG0 val) {
-//     // IMPLEMENTATION
-//   }
-// };
-template <typename Op>
-ArrayKernelExec NumericEqualTypesUnary(detail::GetTypeId get_id) {
-  switch (get_id.id) {
-    case Type::INT8:
-      return ScalarPrimitiveExecUnary<Op, Int8Type, Int8Type>;
-    case Type::UINT8:
-      return ScalarPrimitiveExecUnary<Op, UInt8Type, UInt8Type>;
-    case Type::INT16:
-      return ScalarPrimitiveExecUnary<Op, Int16Type, Int16Type>;
-    case Type::UINT16:
-      return ScalarPrimitiveExecUnary<Op, UInt16Type, UInt16Type>;
-    case Type::INT32:
-      return ScalarPrimitiveExecUnary<Op, Int32Type, Int32Type>;
-    case Type::UINT32:
-      return ScalarPrimitiveExecUnary<Op, UInt32Type, UInt32Type>;
-    case Type::INT64:
-      return ScalarPrimitiveExecUnary<Op, Int64Type, Int64Type>;
-    case Type::UINT64:
-      return ScalarPrimitiveExecUnary<Op, UInt64Type, UInt64Type>;
-    case Type::FLOAT:
-      return ScalarPrimitiveExecUnary<Op, FloatType, FloatType>;
-    case Type::DOUBLE:
-      return ScalarPrimitiveExecUnary<Op, DoubleType, DoubleType>;
-    default:
-      DCHECK(false);
-      return ExecFail;
-  }
-}
-
-// Generate a kernel given a templated functor. This template effectively
-// "curries" the first type argument. The functor must be of the form:
-//
-// template <typename Type0, typename Type1, Args...>
-// struct FUNCTOR {
-//   static void Exec(KernelContext* ctx, const ExecBatch& batch, Datum* out) {
-//     // IMPLEMENTATION
-//   }
-// };
-//
-// This function will generate exec functions where Type1 is one of the numeric
-// types
+// GD for numeric types (integer and floating point)
 template <template <typename...> class Generator, typename Type0, typename... Args>
-ArrayKernelExec Numeric(detail::GetTypeId get_id) {
+ArrayKernelExec GenerateNumeric(detail::GetTypeId get_id) {
   switch (get_id.id) {
     case Type::INT8:
       return Generator<Type0, Int8Type, Args...>::Exec;
@@ -790,7 +740,7 @@ ArrayKernelExec Numeric(detail::GetTypeId get_id) {
 //
 // See "Numeric" above for description of the generator functor
 template <template <typename...> class Generator, typename Type0, typename... Args>
-ArrayKernelExec FloatingPoint(detail::GetTypeId get_id) {
+ArrayKernelExec GenerateFloatingPoint(detail::GetTypeId get_id) {
   switch (get_id.id) {
     case Type::FLOAT:
       return Generator<Type0, FloatType, Args...>::Exec;
@@ -806,7 +756,7 @@ ArrayKernelExec FloatingPoint(detail::GetTypeId get_id) {
 //
 // See "Numeric" above for description of the generator functor
 template <template <typename...> class Generator, typename Type0, typename... Args>
-ArrayKernelExec Integer(detail::GetTypeId get_id) {
+ArrayKernelExec GenerateInteger(detail::GetTypeId get_id) {
   switch (get_id.id) {
     case Type::INT8:
       return Generator<Type0, Int8Type, Args...>::Exec;
@@ -831,15 +781,20 @@ ArrayKernelExec Integer(detail::GetTypeId get_id) {
 }
 
 template <template <typename...> class Generator, typename Type0, typename... Args>
-ArrayKernelExec IntegerBased(detail::GetTypeId get_id) {
+ArrayKernelExec GeneratePhysicalInteger(detail::GetTypeId get_id) {
   switch (get_id.id) {
     case Type::INT8:
       return Generator<Type0, Int8Type, Args...>::Exec;
     case Type::INT16:
       return Generator<Type0, Int16Type, Args...>::Exec;
     case Type::INT32:
+    case Type::DATE32:
+    case Type::TIME32:
       return Generator<Type0, Int32Type, Args...>::Exec;
     case Type::INT64:
+    case Type::DATE64:
+    case Type::TIMESTAMP:
+    case Type::TIME64:
       return Generator<Type0, Int64Type, Args...>::Exec;
     case Type::UINT8:
       return Generator<Type0, UInt8Type, Args...>::Exec;
@@ -849,16 +804,6 @@ ArrayKernelExec IntegerBased(detail::GetTypeId get_id) {
       return Generator<Type0, UInt32Type, Args...>::Exec;
     case Type::UINT64:
       return Generator<Type0, UInt64Type, Args...>::Exec;
-    case Type::DATE32:
-      return Generator<Type0, Int32Type, Args...>::Exec;
-    case Type::DATE64:
-      return Generator<Type0, Int64Type, Args...>::Exec;
-    case Type::TIMESTAMP:
-      return Generator<Type0, Int64Type, Args...>::Exec;
-    case Type::TIME32:
-      return Generator<Type0, Int32Type, Args...>::Exec;
-    case Type::TIME64:
-      return Generator<Type0, Int64Type, Args...>::Exec;
     default:
       DCHECK(false);
       return ExecFail;
@@ -869,7 +814,7 @@ ArrayKernelExec IntegerBased(detail::GetTypeId get_id) {
 //
 // See "Numeric" above for description of the generator functor
 template <template <typename...> class Generator, typename Type0, typename... Args>
-ArrayKernelExec SignedInteger(detail::GetTypeId get_id) {
+ArrayKernelExec GenerateSignedInteger(detail::GetTypeId get_id) {
   switch (get_id.id) {
     case Type::INT8:
       return Generator<Type0, Int8Type, Args...>::Exec;
@@ -892,7 +837,7 @@ ArrayKernelExec SignedInteger(detail::GetTypeId get_id) {
 //
 // See "Numeric" above for description of the generator functor
 template <template <typename...> class Generator, typename Type0, typename... Args>
-ArrayKernelExec BaseBinary(detail::GetTypeId get_id) {
+ArrayKernelExec GenerateVarBinaryBase(detail::GetTypeId get_id) {
   switch (get_id.id) {
     case Type::BINARY:
     case Type::STRING:
@@ -908,7 +853,7 @@ ArrayKernelExec BaseBinary(detail::GetTypeId get_id) {
 
 // See BaseBinary documentation
 template <template <typename...> class Generator, typename Type0, typename... Args>
-ArrayKernelExec BaseBinarySpecific(detail::GetTypeId get_id) {
+ArrayKernelExec GenerateVarBinary(detail::GetTypeId get_id) {
   switch (get_id.id) {
     case Type::BINARY:
       return Generator<Type0, BinaryType, Args...>::Exec;
@@ -928,7 +873,7 @@ ArrayKernelExec BaseBinarySpecific(detail::GetTypeId get_id) {
 //
 // See "Numeric" above for description of the generator functor
 template <template <typename...> class Generator, typename Type0, typename... Args>
-ArrayKernelExec Temporal(detail::GetTypeId get_id) {
+ArrayKernelExec GenerateTemporal(detail::GetTypeId get_id) {
   switch (get_id.id) {
     case Type::DATE32:
       return Generator<Type0, Date32Type, Args...>::Exec;
@@ -948,6 +893,8 @@ ArrayKernelExec Temporal(detail::GetTypeId get_id) {
   }
 }
 
-}  // namespace codegen
+// END of kernel generator-dispatchers
+// ----------------------------------------------------------------------
+
 }  // namespace compute
 }  // namespace arrow
diff --git a/cpp/src/arrow/compute/kernels/scalar_cast_boolean.cc b/cpp/src/arrow/compute/kernels/scalar_cast_boolean.cc
index 933dcf984c8..524866ac391 100644
--- a/cpp/src/arrow/compute/kernels/scalar_cast_boolean.cc
+++ b/cpp/src/arrow/compute/kernels/scalar_cast_boolean.cc
@@ -52,13 +52,12 @@ std::vector<std::shared_ptr<CastFunction>> GetBooleanCasts() {
 
   for (const auto& ty : NumericTypes()) {
     ArrayKernelExec exec =
-        codegen::Numeric<codegen::ScalarUnary, BooleanType, IsNonZero>(*ty);
+        GenerateNumeric<codegen::ScalarUnary, BooleanType, IsNonZero>(*ty);
     DCHECK_OK(func->AddKernel(ty->id(), {ty}, boolean(), exec));
   }
   for (const auto& ty : BaseBinaryTypes()) {
-    ArrayKernelExec exec =
-        codegen::BaseBinary<codegen::ScalarUnaryNotNull, BooleanType, ParseBooleanString>(
-            *ty);
+    ArrayKernelExec exec = GenerateVarBinaryBase<codegen::ScalarUnaryNotNull, BooleanType,
+                                                 ParseBooleanString>(*ty);
     DCHECK_OK(func->AddKernel(ty->id(), {ty}, boolean(), exec));
   }
   return {func};
diff --git a/cpp/src/arrow/compute/kernels/scalar_cast_numeric.cc b/cpp/src/arrow/compute/kernels/scalar_cast_numeric.cc
index 9b1588df466..af7632b195b 100644
--- a/cpp/src/arrow/compute/kernels/scalar_cast_numeric.cc
+++ b/cpp/src/arrow/compute/kernels/scalar_cast_numeric.cc
@@ -484,13 +484,13 @@ void AddPrimitiveNumberCasts(const std::shared_ptr<DataType>& out_ty,
 
   // Cast from other numbers
   for (const std::shared_ptr<DataType>& in_ty : NumericTypes()) {
-    auto exec = codegen::Numeric<CastFunctor, OutType>(*in_ty);
+    auto exec = GenerateNumeric<CastFunctor, OutType>(*in_ty);
     DCHECK_OK(func->AddKernel(in_ty->id(), {in_ty}, out_ty, exec));
   }
 
   // Cast from other strings
   for (const std::shared_ptr<DataType>& in_ty : BaseBinaryTypes()) {
-    auto exec = codegen::BaseBinary<CastFunctor, OutType>(*in_ty);
+    auto exec = GenerateVarBinaryBase<CastFunctor, OutType>(*in_ty);
     DCHECK_OK(func->AddKernel(in_ty->id(), {in_ty}, out_ty, exec));
   }
 }
diff --git a/cpp/src/arrow/compute/kernels/scalar_cast_string.cc b/cpp/src/arrow/compute/kernels/scalar_cast_string.cc
index 9830102b6aa..b842d6b2d2c 100644
--- a/cpp/src/arrow/compute/kernels/scalar_cast_string.cc
+++ b/cpp/src/arrow/compute/kernels/scalar_cast_string.cc
@@ -139,7 +139,7 @@ void AddNumberToStringCasts(std::shared_ptr<DataType> out_ty, CastFunction* func
 
   for (const std::shared_ptr<DataType>& in_ty : NumericTypes()) {
     DCHECK_OK(func->AddKernel(in_ty->id(), {in_ty}, out_ty,
-                              codegen::Numeric<CastFunctor, OutType>(*in_ty),
+                              GenerateNumeric<CastFunctor, OutType>(*in_ty),
                               NullHandling::COMPUTED_NO_PREALLOCATE));
   }
 }
diff --git a/cpp/src/arrow/compute/kernels/scalar_compare.cc b/cpp/src/arrow/compute/kernels/scalar_compare.cc
index 7ae344fd742..25cc377efa8 100644
--- a/cpp/src/arrow/compute/kernels/scalar_compare.cc
+++ b/cpp/src/arrow/compute/kernels/scalar_compare.cc
@@ -56,12 +56,12 @@ struct GreaterEqual {
   }
 };
 
-// Implement Greater, GreaterEqual by flipping arguments to Less, LessEqual
+// Implement Less, LessEqual by flipping arguments to Greater, GreaterEqual
 
 template <typename Op>
 void AddIntegerCompare(const std::shared_ptr<DataType>& ty, ScalarFunction* func) {
   auto exec =
-      codegen::IntegerBased<codegen::ScalarBinaryEqualTypes, BooleanType, Op>(*ty);
+      GeneratePhysicalInteger<codegen::ScalarBinaryEqualTypes, BooleanType, Op>(*ty);
   DCHECK_OK(func->AddKernel({ty, ty}, boolean(), std::move(exec)));
 }
 
@@ -92,41 +92,39 @@ std::shared_ptr<ScalarFunction> MakeCompareFunction(std::string name) {
   // Add timestamp kernels
   for (auto unit : {TimeUnit::SECOND, TimeUnit::MILLI, TimeUnit::MICRO, TimeUnit::NANO}) {
     InputType in_type(match::TimestampTypeUnit(unit));
-    auto exec =
-        codegen::IntegerBased<codegen::ScalarBinaryEqualTypes, BooleanType, Op>(int64());
+    auto exec = GeneratePhysicalInteger<codegen::ScalarBinaryEqualTypes, BooleanType, Op>(
+        int64());
     DCHECK_OK(func->AddKernel({in_type, in_type}, boolean(), std::move(exec)));
   }
 
   // Duration
   for (auto unit : {TimeUnit::SECOND, TimeUnit::MILLI, TimeUnit::MICRO, TimeUnit::NANO}) {
     InputType in_type(match::DurationTypeUnit(unit));
-    auto exec =
-        codegen::IntegerBased<codegen::ScalarBinaryEqualTypes, BooleanType, Op>(int64());
+    auto exec = GeneratePhysicalInteger<codegen::ScalarBinaryEqualTypes, BooleanType, Op>(
+        int64());
     DCHECK_OK(func->AddKernel({in_type, in_type}, boolean(), std::move(exec)));
   }
 
   // Time32 and Time64
   for (auto unit : {TimeUnit::SECOND, TimeUnit::MILLI}) {
     InputType in_type(match::Time32TypeUnit(unit));
-    auto exec =
-        codegen::IntegerBased<codegen::ScalarBinaryEqualTypes, BooleanType, Op>(int32());
+    auto exec = GeneratePhysicalInteger<codegen::ScalarBinaryEqualTypes, BooleanType, Op>(
+        int32());
     DCHECK_OK(func->AddKernel({in_type, in_type}, boolean(), std::move(exec)));
   }
   for (auto unit : {TimeUnit::MICRO, TimeUnit::NANO}) {
     InputType in_type(match::Time64TypeUnit(unit));
-    auto exec =
-        codegen::IntegerBased<codegen::ScalarBinaryEqualTypes, BooleanType, Op>(int64());
+    auto exec = GeneratePhysicalInteger<codegen::ScalarBinaryEqualTypes, BooleanType, Op>(
+        int64());
     DCHECK_OK(func->AddKernel({in_type, in_type}, boolean(), std::move(exec)));
   }
 
   for (const std::shared_ptr<DataType>& ty : BaseBinaryTypes()) {
     auto exec =
-        codegen::BaseBinary<codegen::ScalarBinaryEqualTypes, BooleanType, Op>(*ty);
+        GenerateVarBinaryBase<codegen::ScalarBinaryEqualTypes, BooleanType, Op>(*ty);
     DCHECK_OK(func->AddKernel({ty, ty}, boolean(), std::move(exec)));
   }
 
-  // TODO: Leave time32, time64, and duration for follow up work
-
   return func;
 }
 
diff --git a/cpp/src/arrow/compute/kernels/scalar_compare_test.cc b/cpp/src/arrow/compute/kernels/scalar_compare_test.cc
index c6bd30feed2..0ee757b1280 100644
--- a/cpp/src/arrow/compute/kernels/scalar_compare_test.cc
+++ b/cpp/src/arrow/compute/kernels/scalar_compare_test.cc
@@ -377,33 +377,26 @@ void DoRandomCompare(const std::shared_ptr<DataType>& type) {
   using ScalarType = typename TypeTraits<Type>::ScalarType;
   using CType = typename TypeTraits<Type>::CType;
   auto rand = random::RandomArrayGenerator(0x5416447);
-  for (size_t i = 3; i < 10; i++) {
-    for (auto null_probability : {0.0, 0.01, 0.1, 0.25, 0.5, 1.0}) {
-      for (auto op : {EQUAL, NOT_EQUAL, GREATER, LESS_EQUAL}) {
-        const int64_t length = static_cast<int64_t>(1ULL << i);
-        auto data =
-            rand.Numeric<typename Type::StorageType>(length, 0, 100, null_probability);
-
-        // Create view of data as the type (e.g. timestamp)
-        auto array = Datum(*data->View(type));
-        auto fifty = Datum(std::make_shared<ScalarType>(CType(50), type));
-        auto options = CompareOptions(op);
-        ValidateCompare<Type>(options, array, fifty);
-        ValidateCompare<Type>(options, fifty, array);
-      }
-    }
-  }
-  for (size_t i = 3; i < 5; i++) {
-    for (auto null_probability : {0.0, 0.01, 0.1, 0.25, 0.5, 1.0}) {
-      for (auto op : {EQUAL, NOT_EQUAL, GREATER, LESS_EQUAL}) {
-        const int64_t length = static_cast<int64_t>(1ULL << i);
-        auto lhs = rand.Numeric<typename Type::StorageType>(length << i, 0, 100,
-                                                            null_probability);
-        auto rhs = rand.Numeric<typename Type::StorageType>(length << i, 0, 100,
-                                                            null_probability);
-        auto options = CompareOptions(op);
-        ValidateCompare<Type>(options, *lhs->View(type), *rhs->View(type));
-      }
+  const int64_t length = 1000;
+  for (auto null_probability : {0.0, 0.01, 0.1, 0.25, 0.5, 1.0}) {
+    for (auto op : {EQUAL, NOT_EQUAL, GREATER, LESS_EQUAL}) {
+      auto data =
+          rand.Numeric<typename Type::PhysicalType>(length, 0, 100, null_probability);
+
+      auto data1 =
+          rand.Numeric<typename Type::PhysicalType>(length, 0, 100, null_probability);
+      auto data2 =
+          rand.Numeric<typename Type::PhysicalType>(length, 0, 100, null_probability);
+
+      // Create view of data as the type (e.g. timestamp)
+      auto array1 = Datum(*data1->View(type));
+      auto array2 = Datum(*data2->View(type));
+      auto fifty = Datum(std::make_shared<ScalarType>(CType(50), type));
+      auto options = CompareOptions(op);
+
+      ValidateCompare<Type>(options, array1, fifty);
+      ValidateCompare<Type>(options, fifty, array1);
+      ValidateCompare<Type>(options, array1, array2);
     }
   }
 }
diff --git a/cpp/src/arrow/compute/kernels/vector_sort.cc b/cpp/src/arrow/compute/kernels/vector_sort.cc
index 4ae4355f7f8..9a3aa77eec9 100644
--- a/cpp/src/arrow/compute/kernels/vector_sort.cc
+++ b/cpp/src/arrow/compute/kernels/vector_sort.cc
@@ -264,12 +264,12 @@ template <template <typename...> class ExecTemplate>
 void AddSortingKernels(VectorKernel base, VectorFunction* func) {
   for (const auto& ty : NumericTypes()) {
     base.signature = KernelSignature::Make({InputType::Array(ty)}, uint64());
-    base.exec = codegen::Numeric<ExecTemplate, UInt64Type>(*ty);
+    base.exec = GenerateNumeric<ExecTemplate, UInt64Type>(*ty);
     DCHECK_OK(func->AddKernel(base));
   }
   for (const auto& ty : BaseBinaryTypes()) {
     base.signature = KernelSignature::Make({InputType::Array(ty)}, uint64());
-    base.exec = codegen::BaseBinarySpecific<ExecTemplate, UInt64Type>(*ty);
+    base.exec = GenerateVarBinary<ExecTemplate, UInt64Type>(*ty);
     DCHECK_OK(func->AddKernel(base));
   }
 }
diff --git a/cpp/src/arrow/scalar.h b/cpp/src/arrow/scalar.h
index 5e090c7b755..c50833e743c 100644
--- a/cpp/src/arrow/scalar.h
+++ b/cpp/src/arrow/scalar.h
@@ -246,10 +246,10 @@ struct ARROW_EXPORT FixedSizeBinaryScalar : public BinaryScalar {
   explicit FixedSizeBinaryScalar(std::shared_ptr<DataType> type) : BinaryScalar(type) {}
 };
 
-template <typename T, typename StorageType = typename T::StorageType,
+template <typename T, typename PhysicalType = typename T::PhysicalType,
           typename Enable = void>
-struct ARROW_EXPORT TemporalScalar : internal::PrimitiveScalar<StorageType> {
-  using internal::PrimitiveScalar<StorageType>::PrimitiveScalar;
+struct ARROW_EXPORT TemporalScalar : internal::PrimitiveScalar<PhysicalType> {
+  using internal::PrimitiveScalar<PhysicalType>::PrimitiveScalar;
   using TypeClass = T;
 };
 
diff --git a/cpp/src/arrow/type.h b/cpp/src/arrow/type.h
index e5f1f9a8b6a..d1c801f49ea 100644
--- a/cpp/src/arrow/type.h
+++ b/cpp/src/arrow/type.h
@@ -485,7 +485,7 @@ class ARROW_EXPORT CTypeImpl : public BASE {
  public:
   static constexpr Type::type type_id = TYPE_ID;
   using c_type = C_TYPE;
-  using StorageType = DERIVED;
+  using PhysicalType = DERIVED;
 
   CTypeImpl() : BASE(TYPE_ID) {}
 
@@ -1125,7 +1125,7 @@ class ARROW_EXPORT Date32Type : public DateType {
   static constexpr Type::type type_id = Type::DATE32;
   static constexpr DateUnit UNIT = DateUnit::DAY;
   using c_type = int32_t;
-  using StorageType = Int32Type;
+  using PhysicalType = Int32Type;
 
   static constexpr const char* type_name() { return "date32"; }
 
@@ -1148,7 +1148,7 @@ class ARROW_EXPORT Date64Type : public DateType {
   static constexpr Type::type type_id = Type::DATE64;
   static constexpr DateUnit UNIT = DateUnit::MILLI;
   using c_type = int64_t;
-  using StorageType = Int64Type;
+  using PhysicalType = Int64Type;
 
   static constexpr const char* type_name() { return "date64"; }
 
@@ -1186,7 +1186,7 @@ class ARROW_EXPORT Time32Type : public TimeType {
  public:
   static constexpr Type::type type_id = Type::TIME32;
   using c_type = int32_t;
-  using StorageType = Int32Type;
+  using PhysicalType = Int32Type;
 
   static constexpr const char* type_name() { return "time32"; }
 
@@ -1205,7 +1205,7 @@ class ARROW_EXPORT Time64Type : public TimeType {
  public:
   static constexpr Type::type type_id = Type::TIME64;
   using c_type = int64_t;
-  using StorageType = Int64Type;
+  using PhysicalType = Int64Type;
 
   static constexpr const char* type_name() { return "time64"; }
 
@@ -1256,7 +1256,7 @@ class ARROW_EXPORT TimestampType : public TemporalType, public ParametricType {
 
   static constexpr Type::type type_id = Type::TIMESTAMP;
   using c_type = int64_t;
-  using StorageType = Int64Type;
+  using PhysicalType = Int64Type;
 
   static constexpr const char* type_name() { return "timestamp"; }
 
@@ -1302,7 +1302,7 @@ class ARROW_EXPORT MonthIntervalType : public IntervalType {
  public:
   static constexpr Type::type type_id = Type::INTERVAL_MONTHS;
   using c_type = int32_t;
-  using StorageType = Int32Type;
+  using PhysicalType = Int32Type;
 
   static constexpr const char* type_name() { return "month_interval"; }
 
@@ -1331,7 +1331,7 @@ class ARROW_EXPORT DayTimeIntervalType : public IntervalType {
     }
   };
   using c_type = DayMilliseconds;
-  using StorageType = DayTimeIntervalType;
+  using PhysicalType = DayTimeIntervalType;
 
   static_assert(sizeof(DayMilliseconds) == 8,
                 "DayMilliseconds struct assumed to be of size 8 bytes");
@@ -1356,7 +1356,7 @@ class ARROW_EXPORT DurationType : public TemporalType, public ParametricType {
 
   static constexpr Type::type type_id = Type::DURATION;
   using c_type = int64_t;
-  using StorageType = Int64Type;
+  using PhysicalType = Int64Type;
 
   static constexpr const char* type_name() { return "duration"; }
 

From c81846a227f19e7af362390a016e701f1b05c7d8 Mon Sep 17 00:00:00 2001
From: Wes McKinney <wesm@apache.org>
Date: Wed, 17 Jun 2020 19:25:03 -0500
Subject: [PATCH 11/11] Revert an overzealous change and add comment regarding
 performance

---
 .../arrow/compute/kernels/codegen_internal.h  | 50 ++++++-------------
 1 file changed, 14 insertions(+), 36 deletions(-)

diff --git a/cpp/src/arrow/compute/kernels/codegen_internal.h b/cpp/src/arrow/compute/kernels/codegen_internal.h
index aa5292e8411..3941b67fba9 100644
--- a/cpp/src/arrow/compute/kernels/codegen_internal.h
+++ b/cpp/src/arrow/compute/kernels/codegen_internal.h
@@ -474,44 +474,22 @@ struct ScalarUnaryNotNullStateful {
   struct ArrayExec<Type, enable_if_base_binary<Type>> {
     static void Exec(const ThisType& functor, KernelContext* ctx, const ExecBatch& batch,
                      Datum* out) {
-      // TODO: This code path is currently inadequately tested.
-
-      using offset_type = typename Type::offset_type;
-      const ArrayData& arg0 = *batch[0].array();
-      typename TypeTraits<Type>::ArrayType arg0_boxed(batch[0].array());
-
-      ArrayData* out_arr = out->mutable_array();
-      TypedBufferBuilder<offset_type> offset_builder;
-      TypedBufferBuilder<uint8_t> data_builder;
-
-      KERNEL_RETURN_IF_ERROR(ctx, offset_builder.Reserve(arg0.length + 1));
-      offset_type out_offset = 0;
-
-      const offset_type* in_offsets = arg0_boxed.raw_value_offsets();
-      const uint8_t* in_data = arg0.buffers[2]->data();
-      offset_type cur_offset = in_offsets[0];
-      for (int64_t i = 0; i < arg0.length; ++i) {
-        offset_builder.UnsafeAppend(out_offset);
-        offset_type next_offset = in_offsets[i + 1];
-        if (arg0_boxed.IsValid(i)) {
-          auto val_size = next_offset - cur_offset;
-          auto val =
-              functor.op.Call(ctx, util::string_view(in_data + cur_offset, val_size));
-          if (std::is_same<offset_type, int32_t>::value &&
-              (static_cast<int64_t>(out_offset) + val_size > kBinaryMemoryLimit)) {
-            ctx->SetStatus(Status::OutOfMemory("Overflowed 32-bit binary builder"));
-            return;
-          }
-          KERNEL_RETURN_IF_ERROR(ctx, data_builder.Append(val.data(), val_size));
-          out_offset += val_size;
+      // NOTE: This code is not currently used by any kernels and has
+      // suboptimal performance because it's recomputing the validity bitmap
+      // that is already computed by the kernel execution layer. Consider
+      // writing a lower-level "output adapter" for base binary types.
+      typename TypeTraits<Type>::BuilderType builder;
+      VisitArrayValuesInline<Arg0Type>(*batch[0].array(), [&](util::optional<ARG0> v) {
+        if (v.has_value()) {
+          KERNEL_RETURN_IF_ERROR(ctx, builder.Append(functor.op.Call(ctx, *v)));
+        } else {
+          KERNEL_RETURN_IF_ERROR(ctx, builder.AppendNull());
         }
-        cur_offset = next_offset;
-      }
-      // Last offset
-      offset_builder.UnsafeAppend(out_offset);
+      });
       if (!ctx->HasError()) {
-        KERNEL_RETURN_IF_ERROR(ctx, offset_builder.Finish(&out_arr->buffers[1]));
-        KERNEL_RETURN_IF_ERROR(ctx, data_builder.Finish(&out_arr->buffers[2]));
+        std::shared_ptr<ArrayData> result;
+        ctx->SetStatus(builder.FinishInternal(&result));
+        out->value = std::move(result);
       }
     }
   };