GH-44393: [C++][Compute] Placement vector functions

cpp/src/arrow/CMakeLists.txt

@@ -787,6 +787,7 @@ if(ARROW_COMPUTE)
        compute/kernels/vector_array_sort.cc
        compute/kernels/vector_cumulative_ops.cc
        compute/kernels/vector_pairwise.cc
+       compute/kernels/vector_placement.cc
        compute/kernels/vector_nested.cc
        compute/kernels/vector_rank.cc
        compute/kernels/vector_replace.cc

cpp/src/arrow/compute/api_vector.cc

@@ -155,6 +155,11 @@ static auto kPairwiseOptionsType = GetFunctionOptionsType<PairwiseOptions>(
     DataMember("periods", &PairwiseOptions::periods));
 static auto kListFlattenOptionsType = GetFunctionOptionsType<ListFlattenOptions>(
     DataMember("recursive", &ListFlattenOptions::recursive));
+static auto kReverseIndicesOptionsType = GetFunctionOptionsType<ReverseIndicesOptions>(
+    DataMember("output_length", &ReverseIndicesOptions::output_length),
+    DataMember("output_type", &ReverseIndicesOptions::output_type));
+static auto kPermuteOptionsType = GetFunctionOptionsType<PermuteOptions>(
+    DataMember("output_length", &PermuteOptions::output_length));
 }  // namespace
 }  // namespace internal
 
@@ -230,6 +235,17 @@ ListFlattenOptions::ListFlattenOptions(bool recursive)
     : FunctionOptions(internal::kListFlattenOptionsType), recursive(recursive) {}
 constexpr char ListFlattenOptions::kTypeName[];
 
+ReverseIndicesOptions::ReverseIndicesOptions(int64_t output_length,
+                                             std::shared_ptr<DataType> output_type)
+    : FunctionOptions(internal::kReverseIndicesOptionsType),
+      output_length(output_length),
+      output_type(std::move(output_type)) {}
+constexpr char ReverseIndicesOptions::kTypeName[];
+
+PermuteOptions::PermuteOptions(int64_t output_length)
+    : FunctionOptions(internal::kPermuteOptionsType), output_length(output_length) {}
+constexpr char PermuteOptions::kTypeName[];
+
 namespace internal {
 void RegisterVectorOptions(FunctionRegistry* registry) {
   DCHECK_OK(registry->AddFunctionOptionsType(kFilterOptionsType));
@@ -244,6 +260,8 @@ void RegisterVectorOptions(FunctionRegistry* registry) {
   DCHECK_OK(registry->AddFunctionOptionsType(kRankOptionsType));
   DCHECK_OK(registry->AddFunctionOptionsType(kPairwiseOptionsType));
   DCHECK_OK(registry->AddFunctionOptionsType(kListFlattenOptionsType));
+  DCHECK_OK(registry->AddFunctionOptionsType(kReverseIndicesOptionsType));
+  DCHECK_OK(registry->AddFunctionOptionsType(kPermuteOptionsType));
 }
 }  // namespace internal
 
@@ -429,5 +447,18 @@ Result<Datum> CumulativeMean(const Datum& values, const CumulativeOptions& optio
   return CallFunction("cumulative_mean", {Datum(values)}, &options, ctx);
 }
 
+// ----------------------------------------------------------------------
+// Placement functions
+
+Result<Datum> ReverseIndices(const Datum& indices, const ReverseIndicesOptions& options,
+                             ExecContext* ctx) {
+  return CallFunction("reverse_indices", {indices}, &options, ctx);
+}
+
+Result<Datum> Permute(const Datum& values, const Datum& indices,
+                      const PermuteOptions& options, ExecContext* ctx) {
+  return CallFunction("permute", {values, indices}, &options, ctx);
+}
+
 }  // namespace compute
 }  // namespace arrow

cpp/src/arrow/compute/api_vector.h

@@ -257,6 +257,37 @@ class ARROW_EXPORT ListFlattenOptions : public FunctionOptions {
   bool recursive = false;
 };
 
+/// \brief Options for reverse_indices function
+class ARROW_EXPORT ReverseIndicesOptions : public FunctionOptions {
+ public:
+  explicit ReverseIndicesOptions(int64_t output_length = -1,
+                                 std::shared_ptr<DataType> output_type = NULLPTR);
+  static constexpr char const kTypeName[] = "ReverseIndicesOptions";
+  static ReverseIndicesOptions Defaults() { return ReverseIndicesOptions(); }
+
+  /// \brief The length of the output reverse indices. If negative, the output will be of
+  /// the same length as the input indices. Any indices that are greater or equal to this
+  /// length will be ignored.
+  int64_t output_length = -1;
+  /// \brief The type of the output reverse indices. If null, the output will be of the
+  /// same type as the input indices, otherwise must be integer types. An invalid error
+  /// will be reported if this type is not able to store the length of the input indices.
+  std::shared_ptr<DataType> output_type = NULLPTR;
+};
+
+/// \brief Options for permute function
+class ARROW_EXPORT PermuteOptions : public FunctionOptions {
+ public:
+  explicit PermuteOptions(int64_t output_length = -1);
+  static constexpr char const kTypeName[] = "PermuteOptions";
+  static PermuteOptions Defaults() { return PermuteOptions(); }
+
+  /// \brief The length of the output permutation. If negative, the output will be of the
+  /// same length as the input values (and indices). Any values with indices that are
+  /// greater or equal to this length will be ignored.
+  int64_t output_length = -1;
+};
+
 /// @}
 
 /// \brief Filter with a boolean selection filter
@@ -705,5 +736,56 @@ Result<std::shared_ptr<Array>> PairwiseDiff(const Array& array,
                                             bool check_overflow = false,
                                             ExecContext* ctx = NULLPTR);
 
+/// \brief Return the reverse indices of the given indices.
+///
+/// For indices[i] = x, reverse_indices[x] = i. And reverse_indices[x] = null if x does
+/// not appear in the input indices. For indices[i] = x where x < 0 or x >= output_length,
+/// it is ignored. If multiple indices point to the same value, the last one is used.
+///
+/// For example, with indices = [null, 0, 3, 2, 4, 1, 1], the reverse indices is
+///   [1, 6, 3]                    if output_length = 3,
+///   [1, 6, 3, 2, 4, null, null]  if output_length = 7.
+/// output_length can also be negative, in which case the reverse indices is of the same
+/// length as the indices.
+///
+/// \param[in] indices array-like indices
+/// \param[in] options configures the output length and the output type
+/// \param[in] ctx the function execution context, optional
+/// \return the resulting reverse indices
+///
+/// \since 19.0.0
+/// \note API not yet finalized
+ARROW_EXPORT
+Result<Datum> ReverseIndices(
+    const Datum& indices,
+    const ReverseIndicesOptions& options = ReverseIndicesOptions::Defaults(),
+    ExecContext* ctx = NULLPTR);
+
+/// \brief Permute the values into specified positions according to the indices.
+///
+/// For indices[i] = x, output[x] = values[i]. And output[x] = null if x does not appear
+/// in the input indices. For indices[i] = x where x < 0 or x >= output_length, values[i]
+/// is ignored. If multiple indices point to the same value, the last one is used.
+///
+/// For example, with values = [a, b, c, d, e, f, g] and indices = [null, 0,
+/// 3, 2, 4, 1, 1], the permutation is
+///   [b, g, d]                    if output_length = 3,
+///   [b, g, d, c, e, null, null]  if output_length = 7.
+/// output_length can also be negative, in which case the permutation is of the same
+/// length as the values (and indices).
+///
+/// \param[in] values datum to permute
+/// \param[in] indices array-like indices
+/// \param[in] options configures the output length of the permutation
+/// \param[in] ctx the function execution context, optional
+/// \return the resulting permutation
+///
+/// \since 19.0.0
+/// \note API not yet finalized
+ARROW_EXPORT
+Result<Datum> Permute(const Datum& values, const Datum& indices,
+                      const PermuteOptions& options = PermuteOptions::Defaults(),
+                      ExecContext* ctx = NULLPTR);
+
 }  // namespace compute
 }  // namespace arrow

cpp/src/arrow/compute/function_test.cc

@@ -136,6 +136,10 @@ TEST(FunctionOptions, Equality) {
   options.emplace_back(new SelectKOptions(5, {{SortKey("key", SortOrder::Ascending)}}));
   options.emplace_back(new Utf8NormalizeOptions());
   options.emplace_back(new Utf8NormalizeOptions(Utf8NormalizeOptions::NFD));
+  options.emplace_back(
+      new ReverseIndicesOptions(/*output_length=*/42, /*output_type=*/int32()));
+  options.emplace_back(new PermuteOptions());
+  options.emplace_back(new PermuteOptions(/*output_length=*/42));
 
   for (size_t i = 0; i < options.size(); i++) {
     const size_t prev_i = i == 0 ? options.size() - 1 : i - 1;

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

@@ -115,6 +115,12 @@ add_arrow_compute_test(vector_selection_test
                        EXTRA_LINK_LIBS
                        arrow_compute_kernels_testing)
 
+add_arrow_compute_test(vector_placement_test
+                       SOURCES
+                       vector_placement_test.cc
+                       EXTRA_LINK_LIBS
+                       arrow_compute_kernels_testing)
+
 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")

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

@@ -1037,8 +1037,9 @@ ArrayKernelExec GenerateFloatingPoint(detail::GetTypeId get_id) {
 // Generate a kernel given a templated functor for integer types
 //
 // See "Numeric" above for description of the generator functor
-template <template <typename...> class Generator, typename Type0, typename... Args>
-ArrayKernelExec GenerateInteger(detail::GetTypeId get_id) {
+template <template <typename...> class Generator, typename Type0,
+          typename KernelType = ArrayKernelExec, typename... Args>
+KernelType GenerateInteger(detail::GetTypeId get_id) {
   switch (get_id.id) {
     case Type::INT8:
       return Generator<Type0, Int8Type, Args...>::Exec;