Method invokeFunction

bolt/functions/remote/server/RemoteFunctionService.cpp:82–132

}

void RemoteFunctionServiceHandler::invokeFunction(
    remote::RemoteFunctionResponse& response,
    std::unique_ptr<remote::RemoteFunctionRequest> request) {
  const auto& functionHandle = request->get_remoteFunctionHandle();
  const auto& inputs = request->get_inputs();

  LOG(INFO) << "Got a request for '" << functionHandle.get_name()
            << "': " << inputs.get_rowCount() << " input rows.";

  if (!request->get_throwOnError()) {
    BOLT_NYI("throwOnError not implemented yet on remote server.");
  }

  // Deserialize types and data.
  auto inputType = deserializeArgTypes(functionHandle.get_argumentTypes());
  auto outputType = deserializeType(functionHandle.get_returnType());

  auto serdeFormat = inputs.get_pageFormat();
  auto serde = getSerde(serdeFormat);

  auto inputVector =
      IOBufToRowVector(inputs.get_payload(), inputType, *pool_, serde.get());

  // Execute the expression.
  const vector_size_t numRows = inputVector->size();
  SelectivityVector rows{numRows};

  // Expression boilerplate.
  auto queryCtx = core::QueryCtx::create();
  core::ExecCtx execCtx{pool_.get(), queryCtx.get()};
  exec::ExprSet exprSet{
      getExpressions(
          inputType,
          outputType,
          getFunctionName(functionPrefix_, functionHandle.get_name())),
      &execCtx};
  exec::EvalCtx evalCtx(&execCtx, &exprSet, inputVector.get());

  std::vector<VectorPtr> expressionResult;
  exprSet.eval(rows, evalCtx, expressionResult);

  // Create output vector.
  auto outputRowVector = std::make_shared<RowVector>(
      pool_.get(), ROW({outputType}), BufferPtr(), numRows, expressionResult);

  auto result = response.result_ref();
  result->rowCount_ref() = outputRowVector->size();
  result->pageFormat_ref() = serdeFormat;
  result->payload_ref() =
      rowVectorToIOBuf(outputRowVector, rows.end(), *pool_, serde.get());
}

} // namespace bytedance::bolt::functions