[native] Retrieve Json function metadata

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

@@ -36,6 +36,7 @@
 #include "presto_cpp/main/operators/PartitionAndSerialize.h"
 #include "presto_cpp/main/operators/ShuffleRead.h"
 #include "presto_cpp/main/operators/UnsafeRowExchangeSource.h"
+#include "presto_cpp/main/types/FunctionMetadata.h"
 #include "presto_cpp/main/types/PrestoToVeloxConnector.h"
 #include "presto_cpp/main/types/PrestoToVeloxQueryPlan.h"
 #include "velox/common/base/Counters.h"
@@ -375,6 +376,15 @@ void PrestoServer::run() {
                 http::kMimeTypeApplicationJson)
             .sendWithEOM();
       });
+  // TODO: Register this endpoint only in the sidecar.
+  httpServer_->registerGet(
+      "/v1/info/workerFunctionSignatures",
+      [server = this](
+          proxygen::HTTPMessage* /*message*/,
+          const std::vector<std::unique_ptr<folly::IOBuf>>& /*body*/,
+          proxygen::ResponseHandler* downstream) {
+        server->getFunctionSignatures(downstream);
+      });
 
   if (systemConfig->enableRuntimeMetricsCollection()) {
     enableWorkerStatsReporting();
@@ -1317,6 +1327,11 @@ void PrestoServer::reportNodeStatus(proxygen::ResponseHandler* downstream) {
   http::sendOkResponse(downstream, json(fetchNodeStatus()));
 }
 
+void PrestoServer::getFunctionSignatures(
+    proxygen::ResponseHandler* downstream) {
+  http::sendOkResponse(downstream, getJsonFunctionMetadata());
+}
+
 protocol::NodeStatus PrestoServer::fetchNodeStatus() {
   auto systemConfig = SystemConfig::instance();
   const int64_t nodeMemoryGb = systemConfig->systemMemoryGb();

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

@@ -193,6 +193,8 @@ class PrestoServer {
 
   void reportNodeStatus(proxygen::ResponseHandler* downstream);
 
+  void getFunctionSignatures(proxygen::ResponseHandler* downstream);
+
   protocol::NodeStatus fetchNodeStatus();
 
   void populateMemAndCPUInfo();

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

@@ -25,6 +25,10 @@ target_link_libraries(presto_types presto_type_converter velox_type_fbhive
 
 set_property(TARGET presto_types PROPERTY JOB_POOL_LINK presto_link_job_pool)
 
+add_library(presto_function_metadata OBJECT FunctionMetadata.cpp)
+
+target_link_libraries(presto_function_metadata velox_function_registry)
+
 if(PRESTO_ENABLE_TESTING)
   add_subdirectory(tests)
 endif()

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

@@ -0,0 +1,222 @@
+/*
+ * 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 "presto_cpp/main/types/FunctionMetadata.h"
+
+using namespace facebook::velox;
+using namespace facebook::velox::exec;
+
+namespace facebook::presto {
+
+namespace {
+
+constexpr char const* kDefaultSchema = "default";
+
+// The keys in velox function maps are of the format
+// catalog.schema.function_name. This utility function extracts the
+// function_name from this string.
+const std::string getFunctionName(const std::string& registeredFunctionName) {
+  std::vector<std::string> pieces;
+  folly::split('.', registeredFunctionName, pieces, true);
+  return pieces.back();
+}
+
+const protocol::AggregationFunctionMetadata getAggregationFunctionMetadata(
+    const AggregateFunctionSignature& aggregateFunctionSignature) {
+  protocol::AggregationFunctionMetadata aggregationFunctionMetadata;
+  aggregationFunctionMetadata.intermediateType =
+      aggregateFunctionSignature.intermediateType().toString();
+  /// TODO: Set to true for now. To be read from a mapping of aggregate to
+  /// order sensitivity which should be added in velox.
+  aggregationFunctionMetadata.isOrderSensitive = true;
+  return aggregationFunctionMetadata;
+}
+
+const protocol::RoutineCharacteristics getRoutineCharacteristics(
+    const FunctionSignature& functionSignature,
+    const std::string& functionName,
+    const protocol::FunctionKind& functionKind) {
+  protocol::Determinism determinism;
+  protocol::NullCallClause nullCallClause;
+  if (functionKind == protocol::FunctionKind::SCALAR) {
+    auto functionMetadata = getFunctionMetadata(functionName);
+    VELOX_USER_CHECK(functionMetadata.has_value());
+    determinism = functionMetadata.value().deterministic
+        ? protocol::Determinism::DETERMINISTIC
+        : protocol::Determinism::NOT_DETERMINISTIC;
+    nullCallClause = functionMetadata.value().defaultNullBehavior
+        ? protocol::NullCallClause::RETURNS_NULL_ON_NULL_INPUT
+        : protocol::NullCallClause::CALLED_ON_NULL_INPUT;
+  } else {
+    // Default metadata values of DETERMINISTIC and CALLED_ON_NULL_INPUT for
+    // non-scalar functions.
+    determinism = protocol::Determinism::DETERMINISTIC;
+    nullCallClause = protocol::NullCallClause::CALLED_ON_NULL_INPUT;
+  }
+
+  protocol::RoutineCharacteristics routineCharacteristics;
+  routineCharacteristics.language =
+      std::make_shared<protocol::Language>(protocol::Language({"CPP"}));
+  routineCharacteristics.determinism =
+      std::make_shared<protocol::Determinism>(determinism);
+  routineCharacteristics.nullCallClause =
+      std::make_shared<protocol::NullCallClause>(nullCallClause);
+  return routineCharacteristics;
+}
+
+void updateFunctionMetadata(
+    const std::string& functionName,
+    const FunctionSignature& functionSignature,
+    protocol::JsonBasedUdfFunctionMetadata& jsonBasedUdfFunctionMetadata) {
+  jsonBasedUdfFunctionMetadata.docString = functionName;
+  jsonBasedUdfFunctionMetadata.schema = kDefaultSchema;
+  jsonBasedUdfFunctionMetadata.outputType =
+      functionSignature.returnType().toString();
+
+  const std::vector<TypeSignature> argumentTypes =
+      functionSignature.argumentTypes();
+  std::vector<std::string> paramTypes;
+  for (const auto& argumentType : argumentTypes) {
+    paramTypes.emplace_back(argumentType.toString());
+  }
+  jsonBasedUdfFunctionMetadata.paramTypes = paramTypes;
+}
+
+const std::vector<protocol::JsonBasedUdfFunctionMetadata>
+getAggregateFunctionMetadata(
+    const std::string& functionName,
+    const std::vector<AggregateFunctionSignaturePtr>&
+        aggregateFunctionSignatures) {
+  std::vector<protocol::JsonBasedUdfFunctionMetadata>
+      jsonBasedUdfFunctionMetadataList;
+  jsonBasedUdfFunctionMetadataList.reserve(aggregateFunctionSignatures.size());
+  const protocol::FunctionKind functionKind = protocol::FunctionKind::AGGREGATE;
+
+  for (const auto& aggregateFunctionSignature : aggregateFunctionSignatures) {
+    protocol::JsonBasedUdfFunctionMetadata jsonBasedUdfFunctionMetadata;
+    jsonBasedUdfFunctionMetadata.functionKind = functionKind;
+    jsonBasedUdfFunctionMetadata.routineCharacteristics =
+        getRoutineCharacteristics(
+            *aggregateFunctionSignature, functionName, functionKind);
+    jsonBasedUdfFunctionMetadata.aggregateMetadata =
+        std::make_shared<protocol::AggregationFunctionMetadata>(
+            getAggregationFunctionMetadata(*aggregateFunctionSignature));
+
+    updateFunctionMetadata(
+        functionName,
+        *aggregateFunctionSignature,
+        jsonBasedUdfFunctionMetadata);
+    jsonBasedUdfFunctionMetadataList.emplace_back(jsonBasedUdfFunctionMetadata);
+  }
+  return jsonBasedUdfFunctionMetadataList;
+}
+
+const std::vector<protocol::JsonBasedUdfFunctionMetadata>
+getScalarFunctionMetadata(
+    const std::string& functionName,
+    const std::vector<const FunctionSignature*>& functionSignatures) {
+  std::vector<protocol::JsonBasedUdfFunctionMetadata>
+      jsonBasedUdfFunctionMetadataList;
+  jsonBasedUdfFunctionMetadataList.reserve(functionSignatures.size());
+  const protocol::FunctionKind functionKind = protocol::FunctionKind::SCALAR;
+
+  for (const auto& functionSignature : functionSignatures) {
+    protocol::JsonBasedUdfFunctionMetadata jsonBasedUdfFunctionMetadata;
+    jsonBasedUdfFunctionMetadata.functionKind = functionKind;
+    jsonBasedUdfFunctionMetadata.routineCharacteristics =
+        getRoutineCharacteristics(
+            *functionSignature, functionName, functionKind);
+
+    updateFunctionMetadata(
+        functionName, *functionSignature, jsonBasedUdfFunctionMetadata);
+    jsonBasedUdfFunctionMetadataList.emplace_back(jsonBasedUdfFunctionMetadata);
+  }
+  return jsonBasedUdfFunctionMetadataList;
+}
+
+const std::vector<protocol::JsonBasedUdfFunctionMetadata>
+getWindowFunctionMetadata(
+    const std::string& functionName,
+    const std::vector<FunctionSignaturePtr>& windowFunctionSignatures) {
+  std::vector<protocol::JsonBasedUdfFunctionMetadata>
+      jsonBasedUdfFunctionMetadataList;
+  jsonBasedUdfFunctionMetadataList.reserve(windowFunctionSignatures.size());
+  const protocol::FunctionKind functionKind = protocol::FunctionKind::WINDOW;
+
+  for (const auto& windowFunctionSignature : windowFunctionSignatures) {
+    protocol::JsonBasedUdfFunctionMetadata jsonBasedUdfFunctionMetadata;
+    jsonBasedUdfFunctionMetadata.functionKind = functionKind;
+    jsonBasedUdfFunctionMetadata.routineCharacteristics =
+        getRoutineCharacteristics(
+            *windowFunctionSignature, functionName, functionKind);
+
+    updateFunctionMetadata(
+        functionName, *windowFunctionSignature, jsonBasedUdfFunctionMetadata);
+    jsonBasedUdfFunctionMetadataList.emplace_back(jsonBasedUdfFunctionMetadata);
+  }
+  return jsonBasedUdfFunctionMetadataList;
+}
+
+const std::vector<protocol::JsonBasedUdfFunctionMetadata> getFunctionMetadata(
+    const std::string& functionName) {
+  if (auto aggregateFunctionSignatures =
+          getAggregateFunctionSignatures(functionName)) {
+    return getAggregateFunctionMetadata(
+        functionName, aggregateFunctionSignatures.value());
+  } else if (
+      auto windowFunctionSignatures =
+          getWindowFunctionSignatures(functionName)) {
+    return getWindowFunctionMetadata(
+        functionName, windowFunctionSignatures.value());
+  } else {
+    auto functionSignatures = getFunctionSignatures();
+    if (functionSignatures.find(functionName) != functionSignatures.end()) {
+      return getScalarFunctionMetadata(
+          functionName, functionSignatures.at(functionName));
+    }
+  }
+  VELOX_UNREACHABLE("Function kind cannot be determined");
+}
+
+} // namespace
+
+void getJsonMetadataForFunction(
+    const std::string& registeredFunctionName,
+    nlohmann::json& jsonMetadataList) {
+  auto functionMetadataList = getFunctionMetadata(registeredFunctionName);
+  auto functionName = getFunctionName(registeredFunctionName);
+  for (const auto& functionMetadata : functionMetadataList) {
+    jsonMetadataList[functionName].emplace_back(functionMetadata);
+  }
+}
+
+json getJsonFunctionMetadata() {
+  auto registeredFunctionNames = getSortedAggregateNames();
+  auto scalarFunctionNames = getSortedScalarNames();
+  for (const auto& scalarFunction : scalarFunctionNames) {
+    registeredFunctionNames.emplace_back(scalarFunction);
+  }
+  auto windowFunctionNames = getSortedWindowNames();
+  for (const auto& windowFunction : windowFunctionNames) {
+    registeredFunctionNames.emplace_back(windowFunction);
+  }
+
+  nlohmann::json j;
+  for (const auto& registeredFunctionName : registeredFunctionNames) {
+    getJsonMetadataForFunction(registeredFunctionName, j);
+  }
+  return j;
+}
+
+} // namespace facebook::presto

presto-native-execution/presto_cpp/main/types/FunctionMetadata.h

@@ -0,0 +1,18 @@
+#include "velox/exec/Aggregate.h"
+#include "velox/exec/AggregateFunctionRegistry.h"
+#include "velox/exec/WindowFunction.h"
+#include "velox/expression/SimpleFunctionRegistry.h"
+#include "velox/functions/CoverageUtil.h"
+#include "velox/functions/FunctionRegistry.h"
+
+#include "presto_cpp/presto_protocol/presto_protocol.h"
+
+namespace facebook::presto {
+
+void getJsonMetadataForFunction(
+    const std::string& functionName,
+    nlohmann::json& jsonMetadataList);
+
+json getJsonFunctionMetadata();
+
+} // namespace facebook::presto

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

@@ -89,3 +89,24 @@ target_link_libraries(
   velox_tpch_connector
   gtest
   gtest_main)
+
+add_executable(presto_function_metadata_test FunctionMetadataTest.cpp)
+
+add_test(
+  NAME presto_function_metadata_test
+  COMMAND presto_function_metadata_test
+  WORKING_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR})
+
+target_link_libraries(
+  presto_function_metadata_test
+  gtest
+  gtest_main
+  presto_function_metadata
+  presto_protocol
+  velox_aggregates
+  velox_coverage_util
+  velox_exec
+  velox_functions_prestosql
+  presto_type_converter
+  velox_window
+  ${ANTLR4_RUNTIME})