Add support for REST based remote functions.

Co-authored-by: Wills Feng <[email protected]>

velox/functions/remote/CMakeLists.txt

@@ -12,6 +12,27 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+if(NOT DEFINED PROXYGEN_LIBRARIES)
+  find_package(Sodium REQUIRED)
+
+  find_library(PROXYGEN proxygen)
+  find_library(PROXYGEN_HTTP_SERVER proxygenhttpserver)
+  find_library(FIZZ fizz)
+  find_library(WANGLE wangle)
+
+  if(NOT PROXYGEN
+     OR NOT PROXYGEN_HTTP_SERVER
+     OR NOT FIZZ
+     OR NOT WANGLE)
+    message(
+      FATAL_ERROR
+        "One or more proxygen libraries were not found. Please ensure proxygen, proxygenhttpserver, fizz, and wangle are installed."
+    )
+  endif()
+
+  set(PROXYGEN_LIBRARIES ${PROXYGEN_HTTP_SERVER} ${PROXYGEN} ${WANGLE} ${FIZZ})
+endif()
+
 add_subdirectory(if)
 add_subdirectory(client)
 add_subdirectory(server)

velox/functions/remote/client/CMakeLists.txt

@@ -16,10 +16,19 @@ velox_add_library(velox_functions_remote_thrift_client ThriftClient.cpp)
 velox_link_libraries(velox_functions_remote_thrift_client
                      PUBLIC remote_function_thrift FBThrift::thriftcpp2)
 
+velox_add_library(velox_functions_remote_rest_client RestClient.cpp)
+velox_link_libraries(velox_functions_remote_rest_client ${PROXYGEN_LIBRARIES}
+                     velox_exec Folly::folly)
+
 velox_add_library(velox_functions_remote Remote.cpp)
 velox_link_libraries(
   velox_functions_remote
-  PUBLIC velox_expression
+  PUBLIC velox_functions_remote_rest_client
+         velox_expression
+         velox_memory
+         velox_exec
+         velox_vector
+         velox_presto_serializer
          velox_functions_remote_thrift_client
          velox_functions_remote_get_serde
          velox_type_fbhive

velox/functions/remote/client/Remote.cpp

@@ -17,14 +17,25 @@
 #include "velox/functions/remote/client/Remote.h"
 
 #include <folly/io/async/EventBase.h>
+#include "velox/common/memory/ByteStream.h"
 #include "velox/expression/Expr.h"
 #include "velox/expression/VectorFunction.h"
+#include "velox/functions/remote/client/RestClient.h"
 #include "velox/functions/remote/client/ThriftClient.h"
 #include "velox/functions/remote/if/GetSerde.h"
 #include "velox/functions/remote/if/gen-cpp2/RemoteFunctionServiceAsyncClient.h"
+#include "velox/serializers/PrestoSerializer.h"
 #include "velox/type/fbhive/HiveTypeSerializer.h"
+#include "velox/vector/FlatVector.h"
 #include "velox/vector/VectorStream.h"
 
+#include <cctype>
+#include <iomanip>
+#include <sstream>
+#include <string>
+
+using namespace folly;
+using namespace proxygen;
 namespace facebook::velox::functions {
 namespace {
 
@@ -33,17 +44,43 @@ std::string serializeType(const TypePtr& type) {
   return type::fbhive::HiveTypeSerializer::serialize(type);
 }
 
+std::string extractFunctionName(const std::string& input) {
+  size_t lastDot = input.find_last_of('.');
+  if (lastDot != std::string::npos) {
+    return input.substr(lastDot + 1);
+  }
+  return input;
+}
+
+std::string urlEncode(const std::string& value) {
+  std::ostringstream escaped;
+  escaped.fill('0');
+  escaped << std::hex;
+  for (char c : value) {
+    if (isalnum(static_cast<unsigned char>(c)) || c == '-' || c == '_' ||
+        c == '.' || c == '~') {
+      escaped << c;
+    } else {
+      escaped << '%' << std::setw(2) << int(static_cast<unsigned char>(c));
+    }
+  }
+  return escaped.str();
+}
+
 class RemoteFunction : public exec::VectorFunction {
  public:
   RemoteFunction(
       const std::string& functionName,
       const std::vector<exec::VectorFunctionArg>& inputArgs,
       const RemoteVectorFunctionMetadata& metadata)
-      : functionName_(functionName),
-        location_(metadata.location),
-        thriftClient_(getThriftClient(location_, &eventBase_)),
-        serdeFormat_(metadata.serdeFormat),
-        serde_(getSerde(serdeFormat_)) {
+      : functionName_(functionName), metadata_(metadata) {
+    if (metadata.location.type() == typeid(SocketAddress)) {
+      location_ = boost::get<SocketAddress>(metadata.location);
+      thriftClient_ = getThriftClient(location_, &eventBase_);
+    } else if (metadata.location.type() == typeid(URL)) {
+      url_ = boost::get<URL>(metadata.location);
+    }
+
     std::vector<TypePtr> types;
     types.reserve(inputArgs.size());
     serializedInputTypes_.reserve(inputArgs.size());
@@ -62,7 +99,11 @@ class RemoteFunction : public exec::VectorFunction {
       exec::EvalCtx& context,
       VectorPtr& result) const override {
     try {
-      applyRemote(rows, args, outputType, context, result);
+      if ((metadata_.location.type() == typeid(SocketAddress))) {
+        applyRemote(rows, args, outputType, context, result);
+      } else if (metadata_.location.type() == typeid(URL)) {
+        applyRestRemote(rows, args, outputType, context, result);
+      }
     } catch (const VeloxRuntimeError&) {
       throw;
     } catch (const std::exception&) {
@@ -71,6 +112,55 @@ class RemoteFunction : public exec::VectorFunction {
   }
 
  private:
+  void applyRestRemote(
+      const SelectivityVector& rows,
+      std::vector<VectorPtr>& args,
+      const TypePtr& outputType,
+      exec::EvalCtx& context,
+      VectorPtr& result) const {
+    try {
+      std::string fullUrl = fmt::format(
+          "{}/v1/functions/{}/{}/{}/{}",
+          url_.getUrl(),
+          metadata_.schema.value_or("default"),
+          extractFunctionName(functionName_),
+          urlEncode(metadata_.functionId.value_or("default_function_id")),
+          metadata_.version.value_or("1"));
+
+      // Serialize the input data
+      serializer::presto::PrestoVectorSerde serde;
+      serializer::presto::PrestoVectorSerde::PrestoOptions options;
+
+      auto remoteRowVector = std::make_shared<RowVector>(
+          context.pool(),
+          remoteInputType_,
+          BufferPtr{},
+          rows.end(),
+          std::move(args));
+
+      // Serialize the RowVector into an IOBuf (binary format)
+      IOBuf payload = rowVectorToIOBuf(
+          remoteRowVector, rows.end(), *context.pool(), &serde);
+
+      // Send the serialized data to the remote function via RestClient
+      RestClient restClient(fullUrl);
+      std::unique_ptr<IOBuf> responseBody;
+      restClient.invoke_function(
+          std::make_unique<IOBuf>(std::move(payload)), (responseBody));
+
+      auto outputRowVector = IOBufToRowVector(
+          *responseBody, ROW({outputType}), *context.pool(), &serde);
+      result = outputRowVector->childAt(0);
+
+    } catch (const std::exception& e) {
+      // Catch and handle any exceptions thrown during the process
+      VELOX_FAIL(
+          "Error while executing remote function '{}': {}",
+          functionName_,
+          e.what());
+    }
+  }
+
   void applyRemote(
       const SelectivityVector& rows,
       std::vector<VectorPtr>& args,
@@ -97,11 +187,14 @@ class RemoteFunction : public exec::VectorFunction {
 
     auto requestInputs = request.inputs_ref();
     requestInputs->rowCount_ref() = remoteRowVector->size();
-    requestInputs->pageFormat_ref() = serdeFormat_;
+    requestInputs->pageFormat_ref() = metadata_.serdeFormat;
 
     // TODO: serialize only active rows.
     requestInputs->payload_ref() = rowVectorToIOBuf(
-        remoteRowVector, rows.end(), *context.pool(), serde_.get());
+        remoteRowVector,
+        rows.end(),
+        *context.pool(),
+        getSerde(metadata_.serdeFormat).get());
 
     try {
       thriftClient_->sync_invokeFunction(remoteResponse, request);
@@ -117,12 +210,15 @@ class RemoteFunction : public exec::VectorFunction {
         remoteResponse.get_result().get_payload(),
         ROW({outputType}),
         *context.pool(),
-        serde_.get());
+        getSerde(metadata_.serdeFormat).get());
     result = outputRowVector->childAt(0);
 
     if (auto errorPayload = remoteResponse.get_result().errorPayload()) {
       auto errorsRowVector = IOBufToRowVector(
-          *errorPayload, ROW({VARCHAR()}), *context.pool(), serde_.get());
+          *errorPayload,
+          ROW({VARCHAR()}),
+          *context.pool(),
+          getSerde(metadata_.serdeFormat).get());
       auto errorsVector =
           errorsRowVector->childAt(0)->asFlatVector<StringView>();
       VELOX_CHECK(errorsVector, "Should be convertible to flat vector");
@@ -142,16 +238,14 @@ class RemoteFunction : public exec::VectorFunction {
   }
 
   const std::string functionName_;
-  folly::SocketAddress location_;
 
-  folly::EventBase eventBase_;
+  EventBase eventBase_;
   std::unique_ptr<RemoteFunctionClient> thriftClient_;
-  remote::PageFormat serdeFormat_;
-  std::unique_ptr<VectorSerde> serde_;
-
-  // Structures we construct once to cache:
+  SocketAddress location_;
+  URL url_;
   RowTypePtr remoteInputType_;
   std::vector<std::string> serializedInputTypes_;
+  const RemoteVectorFunctionMetadata metadata_;
 };
 
 std::shared_ptr<exec::VectorFunction> createRemoteFunction(
@@ -169,7 +263,7 @@ void registerRemoteFunction(
     std::vector<exec::FunctionSignaturePtr> signatures,
     const RemoteVectorFunctionMetadata& metadata,
     bool overwrite) {
-  exec::registerStatefulVectorFunction(
+  registerStatefulVectorFunction(
       name,
       signatures,
       std::bind(

velox/functions/remote/client/Remote.h

@@ -16,20 +16,38 @@
 
 #pragma once
 
+#include <boost/variant.hpp>
 #include <folly/SocketAddress.h>
+#include <proxygen/lib/utils/URL.h>
 #include "velox/expression/VectorFunction.h"
 #include "velox/functions/remote/if/gen-cpp2/RemoteFunction_types.h"
 
 namespace facebook::velox::functions {
 
 struct RemoteVectorFunctionMetadata : public exec::VectorFunctionMetadata {
-  /// Network address of the servr to communicate with. Note that this can hold
-  /// a network location (ip/port pair) or a unix domain socket path (see
+  /// URL of the HTTP/REST server for remote function.
+  /// Or Network address of the servr to communicate with. Note that this can
+  /// hold a network location (ip/port pair) or a unix domain socket path (see
   /// SocketAddress::makeFromPath()).
-  folly::SocketAddress location;
+  boost::variant<folly::SocketAddress, proxygen::URL> location;
 
-  /// The serialization format to be used
+  /// The serialization format to be used when sending data to the remote.
   remote::PageFormat serdeFormat{remote::PageFormat::PRESTO_PAGE};
+
+  /// Optional schema defining the structure of the data or input/output types
+  /// involved in the remote function. This may include details such as column
+  /// names and data types.
+  std::optional<std::string> schema;
+
+  /// Optional identifier for the specific remote function to be invoked.
+  /// This can be useful when the same server hosts multiple functions,
+  /// and the client needs to specify which function to call.
+  std::optional<std::string> functionId;
+
+  /// Optional version information to be used when calling the remote function.
+  /// This can help in ensuring compatibility with a particular version of the
+  /// function if multiple versions are available on the server.
+  std::optional<std::string> version;
 };
 
 /// Registers a new remote function. It will use the meatadata defined in
@@ -38,8 +56,8 @@ struct RemoteVectorFunctionMetadata : public exec::VectorFunctionMetadata {
 //
 /// Remote functions are registered as regular statufull functions (using the
 /// same internal catalog), and hence conflict if there already exists a
-/// (non-remote) function registered with the same name. The `overwrite` flag
-/// controls whether to overwrite in these cases.
+/// (non-remote) function registered with the same name. The `overwrite`
+/// flagwrite controls whether to overwrite in these cases.
 void registerRemoteFunction(
     const std::string& name,
     std::vector<exec::FunctionSignaturePtr> signatures,

velox/functions/remote/client/RestClient.cpp

@@ -0,0 +1,34 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * Licensed 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 "RestClient.h"
+#include <folly/io/IOBuf.h>
+#include <folly/logging/xlog.h>
+
+namespace facebook::velox::functions {
+
+// RestClient Implementation
+RestClient::RestClient(const std::string& url) : url_(URL(url)) {
+  httpClient_ = std::make_shared<HttpClient>(url_);
+}
+
+void RestClient::invoke_function(
+    std::unique_ptr<IOBuf> requestBody,
+    std::unique_ptr<IOBuf>& responseBody) const {
+  httpClient_->send(std::move(requestBody));
+  responseBody = httpClient_->getResponseBody();
+};
+
+} // namespace facebook::velox::functions