ARROW-9430: [C++] Implement replace_with_mask kernel

cpp/src/arrow/CMakeLists.txt

@@ -401,6 +401,7 @@ if(ARROW_COMPUTE)
        compute/kernels/util_internal.cc
        compute/kernels/vector_hash.cc
        compute/kernels/vector_nested.cc
+       compute/kernels/vector_replace.cc
        compute/kernels/vector_selection.cc
        compute/kernels/vector_sort.cc
        compute/exec/key_hash.cc

cpp/src/arrow/array/array_binary.h

@@ -71,6 +71,13 @@ class BaseBinaryArray : public FlatArray {
                              raw_value_offsets_[i + 1] - pos);
   }
 
+  /// \brief Get binary value as a string_view
+  /// Provided for consistency with other arrays.
+  ///
+  /// \param i the value index
+  /// \return the view over the selected value
+  util::string_view Value(int64_t i) const { return GetView(i); }
+
   /// \brief Get binary value as a std::string
   ///
   /// \param i the value index

cpp/src/arrow/array/util.cc

@@ -528,6 +528,11 @@ class RepeatedArrayFactory {
     return FinishFixedWidth(value.data(), value.size());
   }
 
+  Status Visit(const Decimal256Type&) {
+    auto value = checked_cast<const Decimal256Scalar&>(scalar_).value.ToBytes();
+    return FinishFixedWidth(value.data(), value.size());
+  }
+
   template <typename T>
   enable_if_base_binary<T, Status> Visit(const T&) {
     std::shared_ptr<Buffer> value =

cpp/src/arrow/compute/api_vector.cc

@@ -162,6 +162,11 @@ Result<std::shared_ptr<Array>> NthToIndices(const Array& values, int64_t n,
   return result.make_array();
 }
 
+Result<Datum> ReplaceWithMask(const Datum& values, const Datum& mask,
+                              const Datum& replacements, ExecContext* ctx) {
+  return CallFunction("replace_with_mask", {values, mask, replacements}, ctx);
+}
+
 Result<std::shared_ptr<Array>> SortIndices(const Array& values, SortOrder order,
                                            ExecContext* ctx) {
   ArraySortOptions options(order);

cpp/src/arrow/compute/api_vector.h

@@ -171,6 +171,23 @@ Result<std::shared_ptr<ArrayData>> GetTakeIndices(
 
 }  // namespace internal
 
+/// \brief ReplaceWithMask replaces each value in the array corresponding
+/// to a true value in the mask with the next element from `replacements`.
+///
+/// \param[in] values Array input to replace
+/// \param[in] mask Array or Scalar of Boolean mask values
+/// \param[in] replacements The replacement values to draw from. There must
+/// be as many replacement values as true values in the mask.
+/// \param[in] ctx the function execution context, optional
+///
+/// \return the resulting datum
+///
+/// \since 5.0.0
+/// \note API not yet finalized
+ARROW_EXPORT
+Result<Datum> ReplaceWithMask(const Datum& values, const Datum& mask,
+                              const Datum& replacements, ExecContext* ctx = NULLPTR);
+
 /// \brief Take from an array of values at indices in another array
 ///
 /// The output array will be of the same type as the input values

cpp/src/arrow/compute/kernels/CMakeLists.txt

@@ -48,13 +48,15 @@ add_arrow_compute_test(vector_test
                        SOURCES
                        vector_hash_test.cc
                        vector_nested_test.cc
+                       vector_replace_test.cc
                        vector_selection_test.cc
                        vector_sort_test.cc
                        test_util.cc)
 
 add_arrow_benchmark(vector_hash_benchmark PREFIX "arrow-compute")
 add_arrow_benchmark(vector_sort_benchmark PREFIX "arrow-compute")
 add_arrow_benchmark(vector_partition_benchmark PREFIX "arrow-compute")
+add_arrow_benchmark(vector_replace_benchmark PREFIX "arrow-compute")
 add_arrow_benchmark(vector_selection_benchmark PREFIX "arrow-compute")
 
 # ----------------------------------------------------------------------

cpp/src/arrow/compute/kernels/codegen_internal.h

@@ -1240,6 +1240,7 @@ ArrayKernelExec GenerateTypeAgnosticPrimitive(detail::GetTypeId get_id) {
     case Type::FLOAT:
     case Type::DATE32:
     case Type::TIME32:
+    case Type::INTERVAL_MONTHS:
       return Generator<UInt32Type>::Exec;
     case Type::UINT64:
     case Type::INT64:
@@ -1248,6 +1249,7 @@ ArrayKernelExec GenerateTypeAgnosticPrimitive(detail::GetTypeId get_id) {
     case Type::TIMESTAMP:
     case Type::TIME64:
     case Type::DURATION:
+    case Type::INTERVAL_DAY_TIME:
       return Generator<UInt64Type>::Exec;
     default:
       DCHECK(false);

cpp/src/arrow/compute/kernels/test_util.h

@@ -172,5 +172,27 @@ void CheckDispatchBest(std::string func_name, std::vector<ValueDescr> descrs,
 // Check that function fails to produce a Kernel for the set of ValueDescrs.
 void CheckDispatchFails(std::string func_name, std::vector<ValueDescr> descrs);
 
+// Helper to get a default instance of a type, including parameterized types
+template <typename T>
+enable_if_parameter_free<T, std::shared_ptr<DataType>> default_type_instance() {
+  return TypeTraits<T>::type_singleton();
+}
+template <typename T>
+enable_if_time<T, std::shared_ptr<DataType>> default_type_instance() {
+  // Time32 requires second/milli, Time64 requires nano/micro
+  if (bit_width(T::type_id) == 32) {
+    return std::make_shared<T>(TimeUnit::type::SECOND);
+  }
+  return std::make_shared<T>(TimeUnit::type::NANO);
+}
+template <typename T>
+enable_if_timestamp<T, std::shared_ptr<DataType>> default_type_instance() {
+  return std::make_shared<T>(TimeUnit::type::SECOND);
+}
+template <typename T>
+enable_if_decimal<T, std::shared_ptr<DataType>> default_type_instance() {
+  return std::make_shared<T>(5, 2);
+}
+
 }  // namespace compute
 }  // namespace arrow