Function RunSubstraitConsumer

cpp/examples/arrow/engine_substrait_consumption.cc:115–171

}

arrow::Status RunSubstraitConsumer(int argc, char** argv) {
  // 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();
  ARROW_RETURN_NOT_OK(maybe_serialized_plan.status());
  ARROW_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Buffer> serialized_plan,
                        std::move(maybe_serialized_plan));

  // Print the received plan to stdout as JSON
  ARROW_ASSIGN_OR_RAISE(auto plan_json,
                        eng::internal::SubstraitToJSON("Plan", *serialized_plan));

  std::cout << std::string(50, '#') << " received substrait::Plan:" << std::endl;
  std::cout << plan_json << 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<ac::SinkNodeConsumer>> consumers;
  std::function<std::shared_ptr<ac::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<ac::Declaration>> maybe_decls =
      eng::DeserializePlans(*serialized_plan, consumer_factory);
  ARROW_RETURN_NOT_OK(maybe_decls.status());
  ARROW_ASSIGN_OR_RAISE(std::vector<ac::Declaration> decls, std::move(maybe_decls));

  // 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_ASSIGN_OR_RAISE(std::shared_ptr<ac::ExecPlan> plan, ac::ExecPlan::Make());

  // Add decls to plan (note: configure ExecNode registry before this point)
  for (const ac::Declaration& decl : decls) {
    ARROW_RETURN_NOT_OK(decl.AddToPlan(plan.get()).status());
  }

  // Validate the plan and print it to stdout
  ARROW_RETURN_NOT_OK(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;
  plan->StartProducing();

  // ... and wait for it to finish
  ARROW_RETURN_NOT_OK(plan->finished().status());
  return arrow::Status::OK();
}