Initial base implementation for demean Kernel

cpp/src/arrow/CMakeLists.txt

@@ -771,6 +771,7 @@ if(ARROW_COMPUTE)
        compute/kernels/scalar_validity.cc
        compute/kernels/vector_array_sort.cc
        compute/kernels/vector_cumulative_ops.cc
+       compute/kernels/vector_demean.cc
        compute/kernels/vector_pairwise.cc
        compute/kernels/vector_nested.cc
        compute/kernels/vector_rank.cc

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_demean_test
+                       SOURCES
+                       vector_demean_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/vector_demean.cc

@@ -0,0 +1,81 @@
+
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Include necessary Arrow headers
+#include <arrow/compute/api_vector.h>
+#include "arrow/compute/api_scalar.h"
+#include "arrow/compute/function.h"
+#include "arrow/compute/registry.h"
+#include "arrow/util/logging.h"
+
+namespace arrow {
+namespace compute {
+namespace internal {
+
+namespace {
+// Function documentation
+const FunctionDoc demean_doc{
+    "Perform a demean operation over all the elements of the array",
+    ("Returns an array where every element has been removed from \n"
+     "the mean of the array."),
+    {"array"}};
+
+class DemeanMetaFunction : public MetaFunction {
+ public:
+  DemeanMetaFunction() : MetaFunction("demean", Arity::Unary(), demean_doc) {}
+
+  Result<Datum> ExecuteImpl(const std::vector<Datum>& args,
+                            const FunctionOptions* options,
+                            ExecContext* ctx) const override {
+    switch (args[0].kind()) {
+      case Datum::ARRAY:
+      case Datum::CHUNKED_ARRAY: {
+        return Demean(*args[0].make_array(), ctx);
+      } break;
+      default:
+        break;
+    }
+    return Status::NotImplemented(
+        "Unsupported types for demean operation: "
+        "values=",
+        args[0].ToString());
+  }
+
+ private:
+  template <typename T>
+  static Result<Datum> Demean(const T& input, ExecContext* ctx) {
+    // TODO: Expose options?
+    Datum mean_result;
+    ARROW_ASSIGN_OR_RAISE(
+        mean_result, arrow::compute::CumulativeMean(input, CumulativeOptions(), ctx));
+    Datum demean_result;
+    ARROW_ASSIGN_OR_RAISE(
+        demean_result,
+        arrow::compute::Subtract(input, mean_result, ArithmeticOptions(), ctx));
+
+    return demean_result;
+  }
+};
+}  // namespace
+
+void RegisterVectorDemean(FunctionRegistry* registry) {
+  DCHECK_OK(registry->AddFunction(std::make_shared<DemeanMetaFunction>()));
+}
+}  // namespace internal
+}  // namespace compute
+}  // namespace arrow

cpp/src/arrow/compute/kernels/vector_demean_test.cc

@@ -0,0 +1,52 @@
+
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <gtest/gtest.h>
+
+#include "arrow/compute/api.h"
+#include "arrow/compute/kernels/test_util.h"
+#include "arrow/compute/util.h"
+#include "arrow/result.h"
+#include "arrow/status.h"
+#include "arrow/testing/gtest_util.h"
+#include "arrow/testing/matchers.h"
+
+namespace arrow {
+namespace compute {
+
+TEST(TestDemean, BasicDemean) {
+  constexpr int data_bufndx{1};
+  const std::vector<int32_t> test_result{0, 0, 0, 0, 0, 0};
+  const std::vector<int32_t> test_values{1, 1, 1, 1, 1, 1};
+  Int32Builder input_builder;
+  ASSERT_OK(input_builder.Reserve(test_values.size()));
+  ASSERT_OK(input_builder.AppendValues(test_values));
+  ASSERT_OK_AND_ASSIGN(auto test_inputs, input_builder.Finish());
+
+  ASSERT_OK_AND_ASSIGN(Datum demean_result, CallFunction("demean", {test_inputs}));
+  auto result_data = *(demean_result.array());
+
+  // validate each value
+  for (int val_ndx = 0; val_ndx < test_inputs->length(); ++val_ndx) {
+    int32_t expected_value = test_result[val_ndx];
+    int32_t actual_value = result_data.GetValues<int32_t>(data_bufndx)[val_ndx];
+    ASSERT_EQ(expected_value, actual_value);
+  }
+}
+}  // namespace compute
+}  // namespace arrow

cpp/src/arrow/compute/registry.cc

@@ -313,6 +313,7 @@ static std::unique_ptr<FunctionRegistry> CreateBuiltInRegistry() {
   // Vector functions
   RegisterVectorArraySort(registry.get());
   RegisterVectorCumulativeSum(registry.get());
+  RegisterVectorDemean(registry.get());
   RegisterVectorNested(registry.get());
   RegisterVectorRank(registry.get());
   RegisterVectorReplace(registry.get());

cpp/src/arrow/compute/registry_internal.h

@@ -46,6 +46,7 @@ void RegisterScalarOptions(FunctionRegistry* registry);
 // Vector functions
 void RegisterVectorArraySort(FunctionRegistry* registry);
 void RegisterVectorCumulativeSum(FunctionRegistry* registry);
+void RegisterVectorDemean(FunctionRegistry* registry);
 void RegisterVectorHash(FunctionRegistry* registry);
 void RegisterVectorNested(FunctionRegistry* registry);
 void RegisterVectorRank(FunctionRegistry* registry);