Register REST Remote function

Author: Joe-Abraham

presto-native-execution/CMakeLists.txt

@@ -57,7 +57,7 @@ option(PRESTO_ENABLE_ABFS "Build ABFS support" OFF)
 option(PRESTO_ENABLE_PARQUET "Enable Parquet support" OFF)
 
 # Forwards user input to VELOX_ENABLE_REMOTE_FUNCTIONS.
-option(PRESTO_ENABLE_REMOTE_FUNCTIONS "Enable remote function support" OFF)
+option(PRESTO_ENABLE_REMOTE_FUNCTIONS "Enable remote function support" ON)
 
 option(PRESTO_ENABLE_TESTING "Enable tests" ON)
 

presto-native-execution/presto_cpp/main/PrestoServer.cpp

@@ -1246,10 +1246,11 @@ void PrestoServer::registerRemoteFunctions() {
     } else {
       VELOX_FAIL(
           "To register remote functions using a json file path you need to "
-          "specify the remote server location using '{}', '{}' or '{}'.",
+          "specify the remote server location using '{}', '{}' or '{}' or {}.",
           SystemConfig::kRemoteFunctionServerThriftAddress,
           SystemConfig::kRemoteFunctionServerThriftPort,
-          SystemConfig::kRemoteFunctionServerThriftUdsPath);
+          SystemConfig::kRemoteFunctionServerThriftUdsPath,
+          SystemConfig::kRemoteFunctionServerRestURL);
     }
   }
 #endif

presto-native-execution/presto_cpp/main/common/Configs.h

@@ -600,6 +600,10 @@ class SystemConfig : public ConfigBase {
   static constexpr std::string_view kRemoteFunctionServerThriftUdsPath{
       "remote-function-server.thrift.uds-path"};
 
+  /// HTTP URL used by the remote function rest server.
+  static constexpr std::string_view kRemoteFunctionServerRestURL{
+      "remote-function-server.rest.url"};
+
   /// Path where json files containing signatures for remote functions can be
   /// found.
   static constexpr std::string_view

presto-native-execution/presto_cpp/main/types/CMakeLists.txt

@@ -18,12 +18,20 @@ add_library(
   presto_types OBJECT
   PrestoToVeloxQueryPlan.cpp PrestoToVeloxExpr.cpp VeloxPlanValidator.cpp
   PrestoToVeloxSplit.cpp PrestoToVeloxConnector.cpp)
+
 add_dependencies(presto_types presto_operators presto_type_converter velox_type
                  velox_type_fbhive velox_dwio_dwrf_proto)
 
 target_link_libraries(presto_types presto_type_converter velox_type_fbhive
                       velox_hive_partition_function velox_tpch_gen)
 
+if(PRESTO_ENABLE_REMOTE_FUNCTIONS)
+  add_dependencies(presto_types velox_expression presto_server_remote_function
+                   velox_functions_remote)
+  target_link_libraries(presto_types presto_server_remote_function
+                        velox_functions_remote)
+endif()
+
 set_property(TARGET presto_types PROPERTY JOB_POOL_LINK presto_link_job_pool)
 
 if(PRESTO_ENABLE_TESTING)

presto-native-execution/presto_cpp/main/types/PrestoToVeloxExpr.cpp

@@ -14,14 +14,23 @@
 
 #include "presto_cpp/main/types/PrestoToVeloxExpr.h"
 #include <boost/algorithm/string/case_conv.hpp>
+#include "presto_cpp/main/common/Configs.h"
 #include "presto_cpp/presto_protocol/Base64Util.h"
 #include "velox/common/base/Exceptions.h"
 #include "velox/functions/prestosql/types/JsonType.h"
 #include "velox/vector/ComplexVector.h"
 #include "velox/vector/ConstantVector.h"
 #include "velox/vector/FlatVector.h"
+#ifdef PRESTO_ENABLE_REMOTE_FUNCTIONS
+#include "presto_cpp/main/JsonSignatureParser.h"
+#include "velox/expression/FunctionSignature.h"
+#include "velox/functions/remote/client/Remote.h"
+#endif
 
 using namespace facebook::velox::core;
+#ifdef PRESTO_ENABLE_REMOTE_FUNCTIONS
+using facebook::velox::functions::remote::PageFormat;
+#endif
 using facebook::velox::TypeKind;
 
 namespace facebook::presto {
@@ -412,6 +421,19 @@ std::optional<TypedExprPtr> VeloxExprConverter::tryConvertLike(
       returnType, args, getFunctionName(signature));
 }
 
+#ifdef PRESTO_ENABLE_REMOTE_FUNCTIONS
+PageFormat fromSerdeString(const std::string_view& serdeName) {
+  if (serdeName == "presto_page") {
+    return PageFormat::PRESTO_PAGE;
+  } else if (serdeName == "spark_unsafe_row") {
+    return PageFormat::SPARK_UNSAFE_ROW;
+  } else {
+    VELOX_FAIL(
+        "Unknown serde name for remote function server: '{}'", serdeName);
+  }
+}
+#endif
+
 TypedExprPtr VeloxExprConverter::toVeloxExpr(
     const protocol::CallExpression& pexpr) const {
   if (auto builtin = std::dynamic_pointer_cast<protocol::BuiltInFunctionHandle>(
@@ -458,10 +480,43 @@ TypedExprPtr VeloxExprConverter::toVeloxExpr(
               pexpr.functionHandle)) {
     auto args = toVeloxExpr(pexpr.arguments);
     auto returnType = typeParser_->parse(pexpr.returnType);
+
     return std::make_shared<CallTypedExpr>(
         returnType, args, getFunctionName(sqlFunctionHandle->functionId));
   }
+#ifdef PRESTO_ENABLE_REMOTE_FUNCTIONS
+  else if (
+      auto RestFunctionHandle =
+          std::dynamic_pointer_cast<protocol::RestFunctionHandle>(
+              pexpr.functionHandle)) {
 
+    auto args = toVeloxExpr(pexpr.arguments);
+    auto returnType = typeParser_->parse(pexpr.returnType);
+
+    const auto* systemConfig = SystemConfig::instance();
+
+    velox::functions::RemoteVectorFunctionMetadata metadata;
+    metadata.serdeFormat =
+        fromSerdeString(systemConfig->remoteFunctionServerSerde());
+    proxygen::URL url(systemConfig->kRemoteFunctionServerRestURL);
+    metadata.location = url;
+
+    json signatureJson;
+    to_json(signatureJson, RestFunctionHandle->signature);
+
+    JsonSignatureParser parser(signatureJson.dump());
+    for (const auto& [functionName, signatureItems] : parser) {
+      for (const auto& item : signatureItems) {
+        velox::functions::registerRemoteFunction(
+            getFunctionName(RestFunctionHandle->functionId),
+            {item.signature},
+            metadata);
+      }
+    }
+    return std::make_shared<CallTypedExpr>(
+        returnType, args, getFunctionName(RestFunctionHandle->functionId));
+  }
+#endif
   VELOX_FAIL("Unsupported function handle: {}", pexpr.functionHandle->_type);
 }
 

presto-native-execution/presto_cpp/main/types/tests/CMakeLists.txt

@@ -69,6 +69,16 @@ target_link_libraries(
   ${GFLAGS_LIBRARIES}
   pthread)
 
+if(PRESTO_ENABLE_REMOTE_FUNCTIONS)
+  add_dependencies(presto_expressions_test presto_server_remote_function
+                   velox_expression velox_functions_remote)
+
+  target_link_libraries(
+    presto_expressions_test GTest::gmock GTest::gmock_main
+    presto_server_remote_function velox_expression velox_functions_remote)
+
+endif()
+
 set_property(TARGET presto_expressions_test PROPERTY JOB_POOL_LINK
                                                      presto_link_job_pool)