ARROW-15238: [C++] ARROW_ENGINE module with substrait consumer

.travis.yml

@@ -99,6 +99,7 @@ jobs:
           -e ARROW_GCS=OFF
           -e ARROW_MIMALLOC=OFF
           -e ARROW_ORC=OFF
+          -e ARROW_ENGINE=OFF
           -e ARROW_PARQUET=OFF
           -e ARROW_S3=OFF
           -e CMAKE_UNITY_BUILD=ON

cpp/CMakeLists.txt

@@ -351,7 +351,9 @@ if(ARROW_CUDA
 endif()
 
 if(ARROW_ENGINE)
+  set(ARROW_PARQUET ON)
   set(ARROW_COMPUTE ON)
+  set(ARROW_DATASET ON)
 endif()
 
 if(ARROW_SKYHOOK)

cpp/cmake_modules/DefineOptions.cmake

@@ -225,7 +225,7 @@ if("${CMAKE_SOURCE_DIR}" STREQUAL "${CMAKE_CURRENT_SOURCE_DIR}")
 
   define_option(ARROW_DATASET "Build the Arrow Dataset Modules" OFF)
 
-  define_option(ARROW_ENGINE "Build the Arrow Execution Engine" OFF)
+  define_option(ARROW_ENGINE "Build the Arrow Query Engine Module" OFF)
 
   define_option(ARROW_FILESYSTEM "Build the Arrow Filesystem Layer" OFF)
 
@@ -478,6 +478,16 @@ advised that if this is enabled 'install' will fail silently on components;\
 that have not been built"
                 OFF)
 
+  set(ARROW_SUBSTRAIT_REPO_DEFAULT "https://github.com/substrait-io/substrait")
+  define_option_string(ARROW_SUBSTRAIT_REPO
+                       "Custom git repository URL for downloading Substrait sources.;\
+See also ARROW_SUBSTRAIT_TAG" "${ARROW_SUBSTRAIT_REPO_DEFAULT}")
+
+  set(ARROW_SUBSTRAIT_TAG_DEFAULT "e1b4c04a1b518912f4c4065b16a1b2c0ac8e14cf")
+  define_option_string(ARROW_SUBSTRAIT_TAG
+                       "Custom git hash/tag/branch for Substrait repository.;\
+See also ARROW_SUBSTRAIT_REPO" "${ARROW_SUBSTRAIT_TAG_DEFAULT}")
+
   option(ARROW_BUILD_CONFIG_SUMMARY_JSON "Summarize build configuration in a JSON file"
          ON)
 endif()

cpp/cmake_modules/FindArrowEngine.cmake

@@ -0,0 +1,88 @@
+# 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.
+
+# - Find Arrow Engine (arrow/engine/api.h, libarrow_engine.a, libarrow_engine.so)
+#
+# This module requires Arrow from which it uses
+#  arrow_find_package()
+#
+# This module defines
+#  ARROW_ENGINE_FOUND, whether Arrow Engine has been found
+#  ARROW_ENGINE_IMPORT_LIB,
+#    path to libarrow_engine's import library (Windows only)
+#  ARROW_ENGINE_INCLUDE_DIR, directory containing headers
+#  ARROW_ENGINE_LIB_DIR, directory containing Arrow Engine libraries
+#  ARROW_ENGINE_SHARED_LIB, path to libarrow_engine's shared library
+#  ARROW_ENGINE_STATIC_LIB, path to libarrow_engine.a
+
+if(DEFINED ARROW_ENGINE_FOUND)
+  return()
+endif()
+
+set(find_package_arguments)
+if(${CMAKE_FIND_PACKAGE_NAME}_FIND_VERSION)
+  list(APPEND find_package_arguments "${${CMAKE_FIND_PACKAGE_NAME}_FIND_VERSION}")
+endif()
+if(${CMAKE_FIND_PACKAGE_NAME}_FIND_REQUIRED)
+  list(APPEND find_package_arguments REQUIRED)
+endif()
+if(${CMAKE_FIND_PACKAGE_NAME}_FIND_QUIETLY)
+  list(APPEND find_package_arguments QUIET)
+endif()
+find_package(Arrow ${find_package_arguments})
+find_package(Parquet ${find_package_arguments})
+
+if(ARROW_FOUND AND PARQUET_FOUND)
+  arrow_find_package(ARROW_ENGINE
+                     "${ARROW_HOME}"
+                     arrow_engine
+                     arrow/engine/api.h
+                     ArrowEngine
+                     arrow-engine)
+  if(NOT ARROW_ENGINE_VERSION)
+    set(ARROW_ENGINE_VERSION "${ARROW_VERSION}")
+  endif()
+endif()
+
+if("${ARROW_ENGINE_VERSION}" VERSION_EQUAL "${ARROW_VERSION}")
+  set(ARROW_ENGINE_VERSION_MATCH TRUE)
+else()
+  set(ARROW_ENGINE_VERSION_MATCH FALSE)
+endif()
+
+mark_as_advanced(ARROW_ENGINE_IMPORT_LIB
+                 ARROW_ENGINE_INCLUDE_DIR
+                 ARROW_ENGINE_LIBS
+                 ARROW_ENGINE_LIB_DIR
+                 ARROW_ENGINE_SHARED_IMP_LIB
+                 ARROW_ENGINE_SHARED_LIB
+                 ARROW_ENGINE_STATIC_LIB
+                 ARROW_ENGINE_VERSION
+                 ARROW_ENGINE_VERSION_MATCH)
+
+find_package_handle_standard_args(
+  ArrowEngine
+  REQUIRED_VARS ARROW_ENGINE_INCLUDE_DIR ARROW_ENGINE_LIB_DIR ARROW_ENGINE_VERSION_MATCH
+  VERSION_VAR ARROW_ENGINE_VERSION)
+set(ARROW_ENGINE_FOUND ${ArrowEngine_FOUND})
+
+if(ArrowEngine_FOUND AND NOT ArrowEngine_FIND_QUIETLY)
+  message(STATUS "Found the Arrow Engine by ${ARROW_ENGINE_FIND_APPROACH}")
+  message(STATUS "Found the Arrow Engine shared library: ${ARROW_ENGINE_SHARED_LIB}")
+  message(STATUS "Found the Arrow Engine import library: ${ARROW_ENGINE_IMPORT_LIB}")
+  message(STATUS "Found the Arrow Engine static library: ${ARROW_ENGINE_STATIC_LIB}")
+endif()

cpp/cmake_modules/ThirdpartyToolchain.cmake

@@ -309,7 +309,8 @@ endif()
 
 if(ARROW_ORC
    OR ARROW_FLIGHT
-   OR ARROW_GANDIVA)
+   OR ARROW_GANDIVA
+   OR ARROW_ENGINE)
   set(ARROW_WITH_PROTOBUF ON)
 endif()
 
@@ -1425,6 +1426,11 @@ macro(build_protobuf)
   set(PROTOBUF_VENDORED TRUE)
   set(PROTOBUF_PREFIX "${CMAKE_CURRENT_BINARY_DIR}/protobuf_ep-install")
   set(PROTOBUF_INCLUDE_DIR "${PROTOBUF_PREFIX}/include")
+  # This flag is based on what the user initially requested but if
+  # we've fallen back to building protobuf we always build it statically
+  # so we need to reset the flag so that we can link against it correctly
+  # later.
+  set(Protobuf_USE_STATIC_LIBS ON)
   # Newer protobuf releases always have a lib prefix independent from CMAKE_STATIC_LIBRARY_PREFIX
   set(PROTOBUF_STATIC_LIB
       "${PROTOBUF_PREFIX}/lib/libprotobuf${CMAKE_STATIC_LIBRARY_SUFFIX}")
@@ -1531,7 +1537,7 @@ if(ARROW_WITH_PROTOBUF)
                      PC_PACKAGE_NAMES
                      protobuf)
 
-  if(ARROW_PROTOBUF_USE_SHARED AND MSVC_TOOLCHAIN)
+  if(NOT Protobuf_USE_STATIC_LIBS AND MSVC_TOOLCHAIN)
     add_definitions(-DPROTOBUF_USE_DLLS)
   endif()
 

cpp/examples/arrow/CMakeLists.txt

@@ -21,6 +21,10 @@ if(ARROW_COMPUTE)
   add_arrow_example(compute_register_example)
 endif()
 
+if(ARROW_ENGINE)
+  add_arrow_example(engine_substrait_consumption EXTRA_LINK_LIBS arrow_engine_shared)
+endif()
+
 if(ARROW_COMPUTE AND ARROW_CSV)
   add_arrow_example(compute_and_write_csv_example)
 endif()

cpp/examples/arrow/engine_substrait_consumption.cc

@@ -0,0 +1,186 @@
+// 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 <arrow/api.h>
+#include <arrow/compute/api.h>
+#include <arrow/engine/api.h>
+
+#include <cstdlib>
+#include <iostream>
+#include <memory>
+#include <vector>
+
+namespace eng = arrow::engine;
+namespace cp = arrow::compute;
+
+#define ABORT_ON_FAILURE(expr)                     \
+  do {                                             \
+    arrow::Status status_ = (expr);                \
+    if (!status_.ok()) {                           \
+      std::cerr << status_.message() << std::endl; \
+      abort();                                     \
+    }                                              \
+  } while (0);
+
+class IgnoringConsumer : public cp::SinkNodeConsumer {
+ public:
+  explicit IgnoringConsumer(size_t tag) : tag_{tag} {}
+
+  arrow::Status Consume(cp::ExecBatch batch) override {
+    // Consume a batch of data
+    // (just print its row count to stdout)
+    std::cout << "-" << tag_ << " consumed " << batch.length << " rows" << std::endl;
+    return arrow::Status::OK();
+  }
+
+  arrow::Future<> Finish() override {
+    // Signal to the consumer that the last batch has been delivered
+    // (we don't do any real work in this consumer so mark it finished immediately)
+    //
+    // The returned future should only finish when all outstanding tasks have completed
+    // (after this method is called Consume is guaranteed not to be called again)
+    std::cout << "-" << tag_ << " finished" << std::endl;
+    return arrow::Future<>::MakeFinished();
+  }
+
+ private:
+  // A unique label for instances to help distinguish logging output if a plan has
+  // multiple sinks
+  //
+  // In this example, this is set to the zero-based index of the relation tree in the plan
+  size_t tag_;
+};
+
+arrow::Future<std::shared_ptr<arrow::Buffer>> GetSubstraitFromServer(
+    const std::string& filename) {
+  // Emulate server interaction by parsing hard coded JSON
+  std::string substrait_json = R"({
+    "relations": [
+      {"rel": {
+        "read": {
+          "base_schema": {
+            "struct": {
+              "types": [ {"i64": {}}, {"bool": {}} ]
+            },
+            "names": ["i", "b"]
+          },
+          "local_files": {
+            "items": [
+              {
+                "uri_file": "file://FILENAME_PLACEHOLDER",
+                "format": "FILE_FORMAT_PARQUET"
+              }
+            ]
+          }
+        }
+      }}
+    ],
+    "extension_uris": [
+      {
+        "extension_uri_anchor": 7,
+        "uri": "https://github.com/apache/arrow/blob/master/format/substrait/extension_types.yaml"
+      }
+    ],
+    "extensions": [
+      {"extension_type": {
+        "extension_uri_reference": 7,
+        "type_anchor": 42,
+        "name": "null"
+      }},
+      {"extension_type_variation": {
+        "extension_uri_reference": 7,
+        "type_variation_anchor": 23,
+        "name": "u8"
+      }},
+      {"extension_function": {
+        "extension_uri_reference": 7,
+        "function_anchor": 42,
+        "name": "add"
+      }}
+    ]
+  })";
+  std::string filename_placeholder = "FILENAME_PLACEHOLDER";
+  substrait_json.replace(substrait_json.find(filename_placeholder),
+                         filename_placeholder.size(), filename);
+  return eng::internal::SubstraitFromJSON("Plan", substrait_json);
+}
+
+int main(int argc, char** argv) {
+  if (argc < 2) {
+    std::cout << "Please specify a parquet file to scan" << std::endl;
+    // Fake pass for CI
+    return EXIT_SUCCESS;
+  }
+
+  // Plans arrive at the consumer serialized in a Buffer, using the binary protobuf
+  // serialization of a substrait Plan
+  auto maybe_serialized_plan = GetSubstraitFromServer(argv[1]).result();
+  ABORT_ON_FAILURE(maybe_serialized_plan.status());
+  std::shared_ptr<arrow::Buffer> serialized_plan =
+      std::move(maybe_serialized_plan).ValueOrDie();
+
+  // Print the received plan to stdout as JSON
+  arrow::Result<std::string> maybe_plan_json =
+      eng::internal::SubstraitToJSON("Plan", *serialized_plan);
+  ABORT_ON_FAILURE(maybe_plan_json.status());
+  std::cout << std::string(50, '#') << " received substrait::Plan:" << std::endl;
+  std::cout << maybe_plan_json.ValueOrDie() << std::endl;
+
+  // The data sink(s) for plans is/are implicit in substrait plans, but explicit in
+  // Arrow. Therefore, deserializing a plan requires a factory for consumers: each
+  // time the root of a substrait relation tree is deserialized, an Arrow consumer is
+  // constructed into which its batches will be piped.
+  std::vector<std::shared_ptr<cp::SinkNodeConsumer>> consumers;
+  std::function<std::shared_ptr<cp::SinkNodeConsumer>()> consumer_factory = [&] {
+    // All batches produced by the plan will be fed into IgnoringConsumers:
+    auto tag = consumers.size();
+    consumers.emplace_back(new IgnoringConsumer{tag});
+    return consumers.back();
+  };
+
+  // Deserialize each relation tree in the substrait plan to an Arrow compute Declaration
+  arrow::Result<std::vector<cp::Declaration>> maybe_decls =
+      eng::DeserializePlan(*serialized_plan, consumer_factory);
+  ABORT_ON_FAILURE(maybe_decls.status());
+  std::vector<cp::Declaration> decls = std::move(maybe_decls).ValueOrDie();
+
+  // It's safe to drop the serialized plan; we don't leave references to its memory
+  serialized_plan.reset();
+
+  // Construct an empty plan (note: configure Function registry and ThreadPool here)
+  arrow::Result<std::shared_ptr<cp::ExecPlan>> maybe_plan = cp::ExecPlan::Make();
+  ABORT_ON_FAILURE(maybe_plan.status());
+  std::shared_ptr<cp::ExecPlan> plan = std::move(maybe_plan).ValueOrDie();
+
+  // Add decls to plan (note: configure ExecNode registry before this point)
+  for (const cp::Declaration& decl : decls) {
+    ABORT_ON_FAILURE(decl.AddToPlan(plan.get()).status());
+  }
+
+  // Validate the plan and print it to stdout
+  ABORT_ON_FAILURE(plan->Validate());
+  std::cout << std::string(50, '#') << " produced arrow::ExecPlan:" << std::endl;
+  std::cout << plan->ToString() << std::endl;
+
+  // Start the plan...
+  std::cout << std::string(50, '#') << " consuming batches:" << std::endl;
+  ABORT_ON_FAILURE(plan->StartProducing());
+
+  // ... and wait for it to finish
+  ABORT_ON_FAILURE(plan->finished().status());
+  return EXIT_SUCCESS;
+}

cpp/src/arrow/CMakeLists.txt

@@ -722,6 +722,10 @@ if(ARROW_COMPUTE)
   add_subdirectory(compute)
 endif()
 
+if(ARROW_ENGINE)
+  add_subdirectory(engine)
+endif()
+
 if(ARROW_CUDA)
   add_subdirectory(gpu)
 endif()

cpp/src/arrow/array/array_base.cc

@@ -282,6 +282,8 @@ std::string Array::ToString() const {
   return ss.str();
 }
 
+void PrintTo(const Array& x, std::ostream* os) { *os << x.ToString(); }
+
 Result<std::shared_ptr<Array>> Array::View(
     const std::shared_ptr<DataType>& out_type) const {
   ARROW_ASSIGN_OR_RAISE(std::shared_ptr<ArrayData> result,