ARROW-8792: [C++][Python][R][GLib] New Array compute kernels implementation and execution framework

This patch is a major reworking of our development strategy for implementing array-valued functions and applying them in a query processing setting.

The design was partly inspired by my previous work designing Ibis (https://github.com/ibis-project/ibis -- the "expr" subsystem and the way that operators validate input types and resolve output types). Using only function names and input types, you can determine the output types of each function and resolve the "execute" function that performs a unit of work processing a batch of data. This will allow us to build deferred column expressions and then (eventually) do parallel execution.

There are a ton of details, but one nice thing is that there is now a single API entry point for invoking any function by its name:

```c++
Result<Datum> CallFunction(ExecContext* ctx, const std::string& func_name,
                           const std::vector<Datum>& args,
                           const FunctionOptions* options = NULLPTR);
```

What occurs when you do this:

* A `Function` instance is looked up in the global `FunctionRegistry`
* Given the descriptors of `args` (their types and shapes -- array or scalar), the Function searches for `Kernel` that is able to process those types and shapes. A kernel might be able to do `array[T0], array[T1]` or only `scalar[T0], scalar[T1]`, for example. This permits kernel specialization to treat different type and shape combinations
* The kernel is executed iteratively against `args` based on what `args` contains -- if there are ChunkedArrays, they will be split into contiguous pieces. Kernels never see ChunkedArray, only Array or Scalar
* The Executor implementation is able to split contiguous Array inputs into smaller chunks, which is important for parallel execution. See `ExecContext::set_exec_chunksize`

To summarize: the REGISTRY contains FUNCTIONS. A FUNCTION contains KERNELS. A KERNEL is a specific implementation of a function that services a particular type combination.

An additional effort in this patch is to radically simplify the process of creating kernels that are based on a scalar function. To do this, there is a growing collection of template-based kernel generation classes in compute/kernels/codegen_internal.h that will surely be the topic of much debate. I want to make it a lot easier for people to add new kernels.

There are some other incidental changes in the PR, such as changing the convenience APIs like `Cast` to return `Result`. I'm afraid we may have to live with the API breakage unless someone else wants to add backward compatibility code for the old APIs.

I have to apologize for making such a large PR. I've been working long hours on this for nearly a month and the process of porting all of our existing functionality and making the unit tests pass caused much iteration in the "framework" part of the code, such that it would have been a huge time drain to review incomplete iterations of the framework that had not been proven to capture the functionality that previously existed in the project.

Given the size of this PR and that fact that it completely blocks any work into src/arrow/compute, I don't think we should let this sit unmerged for more than 4 or 5 days, tops. I'm committed to responding to all of your questions and working to address your feedback about the design and improving the documentation and code comments. I tried to leave copious comments to explain my thought process in various places. Feel free to make any and all comments in this PR or in whatever form you like. I don't think that merging should be blocked on stylistic issues.

Closes #7240 from wesm/ARROW-8792-kernels-revamp

Lead-authored-by: Wes McKinney <wesm+git@apache.org>
Co-authored-by: Sutou Kouhei <kou@clear-code.com>
Signed-off-by: Wes McKinney <wesm+git@apache.org>

Author: wesm

c_glib/arrow-glib/compute.cpp

@@ -39,8 +39,18 @@ G_BEGIN_DECLS
 
 G_DEFINE_QUARK(garrow-error-quark, garrow_error)
 
-static GArrowError
-garrow_error_code(const arrow::Status &status)
+G_END_DECLS
+
+gboolean
+garrow_error_check(GError **error,
+                   const arrow::Status &status,
+                   const char *context)
+{
+  return garrow::check(error, status, context);
+}
+
+GArrowError
+garrow_error_from_status(const arrow::Status &status)
 {
   switch (status.code()) {
   case arrow::StatusCode::OK:
@@ -73,50 +83,38 @@ garrow_error_code(const arrow::Status &status)
     return GARROW_ERROR_EXECUTION;
   case arrow::StatusCode::AlreadyExists:
     return GARROW_ERROR_ALREADY_EXISTS;
-
   default:
     return GARROW_ERROR_UNKNOWN;
   }
 }
 
-G_END_DECLS
+arrow::Status
+garrow_error_to_status(GError *error,
+                       arrow::StatusCode code,
+                       const char *context)
+{
+  std::stringstream message;
+  message << context << ": " << g_quark_to_string(error->domain);
+  message << "(" << error->code << "): ";
+  message << error->message;
+  g_error_free(error);
+  return arrow::Status(code, message.str());
+}
 
 namespace garrow {
-  gboolean
-  check(GError **error,
-        const arrow::Status &status,
-        const char *context) {
+  gboolean check(GError **error,
+                 const arrow::Status &status,
+                 const char *context) {
     if (status.ok()) {
       return TRUE;
     } else {
       g_set_error(error,
                   GARROW_ERROR,
-                  garrow_error_code(status),
+                  garrow_error_from_status(status),
                   "%s: %s",
                   context,
                   status.ToString().c_str());
       return FALSE;
     }
   }
 }
-
-gboolean
-garrow_error_check(GError **error,
-                   const arrow::Status &status,
-                   const char *context)
-{
-  return garrow::check(error, status, context);
-}
-
-arrow::Status
-garrow_error_to_status(GError *error,
-                       arrow::StatusCode code,
-                       const char *context)
-{
-  std::stringstream message;
-  message << context << ": " << g_quark_to_string(error->domain);
-  message << "(" << error->code << "): ";
-  message << error->message;
-  g_error_free(error);
-  return arrow::Status(code, message.str());
-}

c_glib/arrow-glib/error.cpp

@@ -23,26 +23,48 @@
 
 #include <arrow-glib/error.h>
 
+gboolean garrow_error_check(GError **error,
+                            const arrow::Status &status,
+                            const char *context);
+GArrowError garrow_error_from_status(const arrow::Status &status);
+arrow::Status garrow_error_to_status(GError *error,
+                                     arrow::StatusCode code,
+                                     const char *context);
+
 namespace garrow {
   gboolean check(GError **error,
                  const arrow::Status &status,
                  const char *context);
 
-  template <typename TYPE>
+  template <typename CONTEXT_FUNC>
   gboolean check(GError **error,
-                 const arrow::Result<TYPE> &result,
-                 const char *context) {
-    if (result.ok()) {
+                 const arrow::Status &status,
+                 CONTEXT_FUNC &&context_func) {
+    if (status.ok()) {
       return TRUE;
     } else {
-      return check(error, result.status(), context);
+      std::string context = std::move(context_func());
+      g_set_error(error,
+                  GARROW_ERROR,
+                  garrow_error_from_status(status),
+                  "%s: %s",
+                  context.c_str(),
+                  status.ToString().c_str());
+      return FALSE;
     }
   }
-}
 
-gboolean garrow_error_check(GError **error,
-                            const arrow::Status &status,
-                            const char *context);
-arrow::Status garrow_error_to_status(GError *error,
-                                     arrow::StatusCode code,
-                                     const char *context);
+  template <typename TYPE>
+  gboolean check(GError **error,
+                 const arrow::Result<TYPE> &result,
+                 const char *context) {
+    return check(error, result.status(), context);
+  }
+
+  template <typename TYPE, typename CONTEXT_FUNC>
+  gboolean check(GError **error,
+                 const arrow::Result<TYPE> &result,
+                 CONTEXT_FUNC &&context_func) {
+    return check(error, result.status(), context_func);
+  }
+}

c_glib/arrow-glib/error.hpp

@@ -121,7 +121,6 @@ endfunction()
 
 set(ARROW_SRCS
     array.cc
-    builder.cc
     array/builder_adaptive.cc
     array/builder_base.cc
     array/builder_binary.cc
@@ -134,8 +133,10 @@ set(ARROW_SRCS
     array/dict_internal.cc
     array/diff.cc
     array/validate.cc
+    builder.cc
     buffer.cc
     compare.cc
+    datum.cc
     device.cc
     extension_type.cc
     memory_pool.cc
@@ -319,24 +320,30 @@ endif()
 
 if(ARROW_COMPUTE)
   list(APPEND ARROW_SRCS
-              compute/context.cc
-              compute/kernels/aggregate.cc
-              compute/kernels/boolean.cc
-              compute/kernels/cast.cc
-              compute/kernels/compare.cc
-              compute/kernels/count.cc
-              compute/kernels/hash.cc
-              compute/kernels/filter.cc
-              compute/kernels/mean.cc
-              compute/kernels/minmax.cc
-              compute/kernels/sort_to_indices.cc
-              compute/kernels/nth_to_indices.cc
-              compute/kernels/sum.cc
-              compute/kernels/add.cc
-              compute/kernels/take.cc
-              compute/kernels/isin.cc
-              compute/kernels/match.cc
-              compute/kernels/util_internal.cc)
+              compute/api_aggregate.cc
+              compute/api_scalar.cc
+              compute/api_vector.cc
+              compute/cast.cc
+              compute/exec.cc
+              compute/function.cc
+              compute/kernel.cc
+              compute/registry.cc
+              compute/kernels/aggregate_basic.cc
+              compute/kernels/codegen_internal.cc
+              compute/kernels/scalar_arithmetic.cc
+              compute/kernels/scalar_boolean.cc
+              compute/kernels/scalar_cast_boolean.cc
+              compute/kernels/scalar_cast_internal.cc
+              compute/kernels/scalar_cast_nested.cc
+              compute/kernels/scalar_cast_numeric.cc
+              compute/kernels/scalar_cast_string.cc
+              compute/kernels/scalar_cast_temporal.cc
+              compute/kernels/scalar_compare.cc
+              compute/kernels/scalar_set_lookup.cc
+              compute/kernels/vector_filter.cc
+              compute/kernels/vector_hash.cc
+              compute/kernels/vector_sort.cc
+              compute/kernels/vector_take.cc)
 endif()
 
 if(ARROW_FILESYSTEM)
@@ -524,6 +531,7 @@ endif()
 
 add_arrow_test(misc_test
                SOURCES
+               datum_test.cc
                memory_pool_test.cc
                result_test.cc
                pretty_print_test.cc

cpp/src/arrow/CMakeLists.txt

@@ -44,7 +44,7 @@ elseif(NOT MSVC)
   set(ORC_MIN_TEST_LIBS ${ORC_MIN_TEST_LIBS} pthread ${CMAKE_DL_LIBS})
 endif()
 
-set(ORC_STATIC_TEST_LINK_LIBS ${ORC_MIN_TEST_LIBS} ${ARROW_LIBRARIES_FOR_STATIC_TESTS}
+set(ORC_STATIC_TEST_LINK_LIBS ${ARROW_LIBRARIES_FOR_STATIC_TESTS} ${ORC_MIN_TEST_LIBS}
                               orc::liborc)
 
 add_arrow_test(adapter_test

cpp/src/arrow/adapters/orc/CMakeLists.txt

@@ -33,8 +33,7 @@
 #include "arrow/array/diff.h"
 #include "arrow/buffer.h"
 #include "arrow/builder.h"
-#include "arrow/compute/context.h"
-#include "arrow/compute/kernels/filter.h"
+#include "arrow/compute/api.h"
 #include "arrow/status.h"
 #include "arrow/testing/gtest_common.h"
 #include "arrow/testing/random.h"
@@ -119,20 +118,19 @@ class DiffTest : public ::testing::Test {
 
   void BaseAndTargetFromRandomFilter(std::shared_ptr<Array> values,
                                      double filter_probability) {
-    compute::Datum out_datum, base_filter, target_filter;
+    std::shared_ptr<Array> base_filter, target_filter;
     do {
       base_filter = this->rng_.Boolean(values->length(), filter_probability, 0.0);
       target_filter = this->rng_.Boolean(values->length(), filter_probability, 0.0);
-    } while (base_filter.Equals(target_filter));
+    } while (base_filter->Equals(target_filter));
 
-    ASSERT_OK(compute::Filter(&ctx_, values, base_filter, {}, &out_datum));
+    ASSERT_OK_AND_ASSIGN(Datum out_datum, compute::Filter(values, base_filter));
     base_ = out_datum.make_array();
 
-    ASSERT_OK(compute::Filter(&ctx_, values, target_filter, {}, &out_datum));
+    ASSERT_OK_AND_ASSIGN(out_datum, compute::Filter(values, target_filter));
     target_ = out_datum.make_array();
   }
 
-  compute::FunctionContext ctx_;
   random::RandomArrayGenerator rng_;
   std::shared_ptr<StructArray> edits_;
   std::shared_ptr<Array> base_, target_;
@@ -616,7 +614,6 @@ void MakeSameLength(std::shared_ptr<Array>* a, std::shared_ptr<Array>* b) {
 }
 
 TEST_F(DiffTest, CompareRandomStruct) {
-  compute::FunctionContext ctx;
   for (auto null_probability : {0.0, 0.25}) {
     constexpr auto length = 1 << 10;
     auto int32_values = this->rng_.Int32(length, 0, 127, null_probability);

cpp/src/arrow/array/diff_test.cc

@@ -58,7 +58,12 @@ function(ADD_ARROW_COMPUTE_TEST REL_TEST_NAME)
                  ${ARG_UNPARSED_ARGUMENTS})
 endfunction()
 
-add_arrow_compute_test(compute_test)
-add_arrow_benchmark(compute_benchmark)
+add_arrow_compute_test(internals_test
+                       SOURCES
+                       function_test.cc
+                       exec_test.cc
+                       kernel_test.cc
+                       registry_test.cc)
+add_arrow_compute_test(exec_test)
 
 add_subdirectory(kernels)

cpp/src/arrow/compute/CMakeLists.txt

@@ -0,0 +1,58 @@
+<!---
+  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.
+-->
+
+## Apache Arrow C++ Compute Functions
+
+This submodule contains analytical functions that process primarily Arrow
+columnar data; some functions can process scalar or Arrow-based array
+inputs. These are intended for use inside query engines, data frame libraries,
+etc.
+
+Many functions have SQL-like semantics in that they perform elementwise or
+scalar operations on whole arrays at a time. Other functions are not SQL-like
+and compute results that may be a different length or whose results depend on
+the order of the values.
+
+Some basic terminology:
+
+* We use the term "function" to refer to particular general operation that may
+  have many different implementations corresponding to different combinations
+  of types or function behavior options.
+* We call a specific implementation of a function a "kernel". When executing a
+  function on inputs, we must first select a suitable kernel (kernel selection
+  is called "dispatching") corresponding to the value types of the inputs
+* Functions along with their kernel implementations are collected in a
+  "function registry". Given a function name and argument types, we can look up
+  that function and dispatch to a compatible kernel.
+
+Types of functions
+
+* Scalar functions: elementwise functions that perform scalar operations in a
+  vectorized manner. These functions are generally valid for SQL-like
+  context. These are called "scalar" in that the functions executed consider
+  each value in an array independently, and the output array or arrays have the
+  same length as the input arrays. The result for each array cell is generally
+  independent of its position in the array.
+* Vector functions, which produce a result whose output is generally dependent
+  on the entire contents of the input arrays. These functions **are generally
+  not valid** for SQL-like processing because the output size may be different
+  than the input size, and the result may change based on the order of the
+  values in the array. This includes things like array subselection, sorting,
+  hashing, and more.
+* Scalar aggregate functions of which can be used in a SQL-like context

cpp/src/arrow/compute/README.md

@@ -15,20 +15,17 @@
 // specific language governing permissions and limitations
 // under the License.
 
-#pragma once
+// NOTE: API is EXPERIMENTAL and will change without going through a
+// deprecation cycle
 
-#include "arrow/compute/context.h"  // IWYU pragma: export
-#include "arrow/compute/kernel.h"   // IWYU pragma: export
+#pragma once
 
-#include "arrow/compute/kernels/boolean.h"          // IWYU pragma: export
-#include "arrow/compute/kernels/cast.h"             // IWYU pragma: export
-#include "arrow/compute/kernels/compare.h"          // IWYU pragma: export
-#include "arrow/compute/kernels/count.h"            // IWYU pragma: export
-#include "arrow/compute/kernels/filter.h"           // IWYU pragma: export
-#include "arrow/compute/kernels/hash.h"             // IWYU pragma: export
-#include "arrow/compute/kernels/isin.h"             // IWYU pragma: export
-#include "arrow/compute/kernels/mean.h"             // IWYU pragma: export
-#include "arrow/compute/kernels/nth_to_indices.h"   // IWYU pragma: export
-#include "arrow/compute/kernels/sort_to_indices.h"  // IWYU pragma: export
-#include "arrow/compute/kernels/sum.h"              // IWYU pragma: export
-#include "arrow/compute/kernels/take.h"             // IWYU pragma: export
+#include "arrow/compute/api_aggregate.h"  // IWYU pragma: export
+#include "arrow/compute/api_scalar.h"     // IWYU pragma: export
+#include "arrow/compute/api_vector.h"     // IWYU pragma: export
+#include "arrow/compute/cast.h"           // IWYU pragma: export
+#include "arrow/compute/exec.h"           // IWYU pragma: export
+#include "arrow/compute/function.h"       // IWYU pragma: export
+#include "arrow/compute/kernel.h"         // IWYU pragma: export
+#include "arrow/compute/registry.h"       // IWYU pragma: export
+#include "arrow/datum.h"                  // IWYU pragma: export