[native] Add native plan checker and native endpoint for Velox plan conversion

presto-main/src/main/java/com/facebook/presto/connector/ConnectorAwareNodeManager.java

@@ -48,6 +48,7 @@ public Set<Node> getAllNodes()
         return ImmutableSet.<Node>builder()
                 .addAll(getWorkerNodes())
                 .addAll(nodeManager.getCoordinators())
+                .addAll(nodeManager.getCoordinatorSidecars())
                 .build();
     }
 

presto-main/src/main/java/com/facebook/presto/server/PluginManager.java

@@ -354,11 +354,6 @@ public void installPlugin(Plugin plugin)
             log.info("Registering node status notification provider %s", nodeStatusNotificationProviderFactory.getName());
             nodeStatusNotificationManager.addNodeStatusNotificationProviderFactory(nodeStatusNotificationProviderFactory);
         }
-
-        for (PlanCheckerProviderFactory planCheckerProviderFactory : plugin.getPlanCheckerProviderFactories()) {
-            log.info("Registering plan checker provider factory %s", planCheckerProviderFactory.getName());
-            planCheckerProviderManager.addPlanCheckerProviderFactory(planCheckerProviderFactory);
-        }
     }
 
     public void installCoordinatorPlugin(CoordinatorPlugin plugin)
@@ -372,6 +367,11 @@ public void installCoordinatorPlugin(CoordinatorPlugin plugin)
             log.info("Registering system session property provider factory %s", providerFactory.getName());
             metadata.getSessionPropertyManager().addSessionPropertyProviderFactory(providerFactory);
         }
+
+        for (PlanCheckerProviderFactory planCheckerProviderFactory : plugin.getPlanCheckerProviderFactories()) {
+            log.info("Registering plan checker provider factory %s", planCheckerProviderFactory.getName());
+            planCheckerProviderManager.addPlanCheckerProviderFactory(planCheckerProviderFactory);
+        }
     }
 
     private URLClassLoader buildClassLoader(String plugin)

presto-main/src/main/java/com/facebook/presto/server/PrestoServer.java

@@ -26,6 +26,7 @@
 import com.facebook.airlift.json.smile.SmileModule;
 import com.facebook.airlift.log.LogJmxModule;
 import com.facebook.airlift.log.Logger;
+import com.facebook.airlift.node.NodeInfo;
 import com.facebook.airlift.node.NodeModule;
 import com.facebook.airlift.tracetoken.TraceTokenModule;
 import com.facebook.drift.server.DriftServer;
@@ -39,9 +40,11 @@
 import com.facebook.presto.execution.warnings.WarningCollectorModule;
 import com.facebook.presto.metadata.Catalog;
 import com.facebook.presto.metadata.CatalogManager;
-import com.facebook.presto.metadata.SessionPropertyManager;
+import com.facebook.presto.metadata.DiscoveryNodeManager;
+import com.facebook.presto.metadata.InternalNodeManager;
 import com.facebook.presto.metadata.StaticCatalogStore;
 import com.facebook.presto.metadata.StaticFunctionNamespaceStore;
+import com.facebook.presto.nodeManager.PluginNodeManager;
 import com.facebook.presto.security.AccessControlManager;
 import com.facebook.presto.security.AccessControlModule;
 import com.facebook.presto.server.security.PasswordAuthenticatorManager;
@@ -179,8 +182,12 @@ public void run()
             injector.getInstance(TracerProviderManager.class).loadTracerProvider();
             injector.getInstance(NodeStatusNotificationManager.class).loadNodeStatusNotificationProvider();
             injector.getInstance(GracefulShutdownHandler.class).loadNodeStatusNotification();
-            injector.getInstance(PlanCheckerProviderManager.class).loadPlanCheckerProviders();
-            injector.getInstance(SessionPropertyManager.class).loadSessionPropertyProviders();
+            PlanCheckerProviderManager planCheckerProviderManager = injector.getInstance(PlanCheckerProviderManager.class);
+            InternalNodeManager nodeManager = injector.getInstance(DiscoveryNodeManager.class);
+            NodeInfo nodeInfo = injector.getInstance(NodeInfo.class);
+            PluginNodeManager pluginNodeManager = new PluginNodeManager(nodeManager, nodeInfo.getEnvironment());
+            planCheckerProviderManager.loadPlanCheckerProviders(pluginNodeManager);
+
             startAssociatedProcesses(injector);
 
             injector.getInstance(Announcer.class).start();

presto-main/src/main/java/com/facebook/presto/sql/planner/sanity/PlanChecker.java

@@ -107,7 +107,15 @@ public void validatePlanFragment(PlanFragment planFragment, Session session, Met
         checkers.get(Stage.FRAGMENT).forEach(checker -> checker.validateFragment(planFragment, session, metadata, warningCollector));
         for (PlanCheckerProvider provider : planCheckerProviderManager.getPlanCheckerProviders()) {
             for (com.facebook.presto.spi.plan.PlanChecker checker : provider.getFragmentPlanCheckers()) {
-                checker.validateFragment(toSimplePlanFragment(planFragment), warningCollector);
+                checker.validateFragment(new SimplePlanFragment(
+                        planFragment.getId(),
+                        planFragment.getRoot(),
+                        planFragment.getVariables(),
+                        planFragment.getPartitioning(),
+                        planFragment.getTableScanSchedulingOrder(),
+                        planFragment.getPartitioningScheme(),
+                        planFragment.getStageExecutionDescriptor(),
+                        planFragment.isOutputTableWriterFragment()), warningCollector);
             }
         }
     }
@@ -126,17 +134,4 @@ private enum Stage
     {
         INTERMEDIATE, FINAL, FRAGMENT
     }
-
-    private static SimplePlanFragment toSimplePlanFragment(PlanFragment planFragment)
-    {
-        return new SimplePlanFragment(
-                planFragment.getId(),
-                planFragment.getRoot(),
-                planFragment.getVariables(),
-                planFragment.getPartitioning(),
-                planFragment.getTableScanSchedulingOrder(),
-                planFragment.getPartitioningScheme(),
-                planFragment.getStageExecutionDescriptor(),
-                planFragment.isOutputTableWriterFragment());
-    }
 }

presto-main/src/main/java/com/facebook/presto/sql/planner/sanity/PlanCheckerProviderManager.java

@@ -14,6 +14,7 @@
 package com.facebook.presto.sql.planner.sanity;
 
 import com.facebook.airlift.log.Logger;
+import com.facebook.presto.spi.NodeManager;
 import com.facebook.presto.spi.plan.PlanCheckerProvider;
 import com.facebook.presto.spi.plan.PlanCheckerProviderContext;
 import com.facebook.presto.spi.plan.PlanCheckerProviderFactory;
@@ -40,15 +41,15 @@ public class PlanCheckerProviderManager
     private static final Logger log = Logger.get(PlanCheckerProviderManager.class);
     private static final String PLAN_CHECKER_PROVIDER_NAME = "plan-checker-provider.name";
 
-    private final PlanCheckerProviderContext planCheckerProviderContext;
+    private final SimplePlanFragmentSerde simplePlanFragmentSerde;
     private final Map<String, PlanCheckerProviderFactory> providerFactories = new ConcurrentHashMap<>();
     private final CopyOnWriteArrayList<PlanCheckerProvider> providers = new CopyOnWriteArrayList<>();
     private final File configDirectory;
 
     @Inject
     public PlanCheckerProviderManager(SimplePlanFragmentSerde simplePlanFragmentSerde, PlanCheckerProviderManagerConfig config)
     {
-        this.planCheckerProviderContext = new PlanCheckerProviderContext(requireNonNull(simplePlanFragmentSerde, "planNodeSerde is null"));
+        this.simplePlanFragmentSerde = requireNonNull(simplePlanFragmentSerde, "planNodeSerde is null");
         requireNonNull(config, "config is null");
         this.configDirectory = requireNonNull(config.getPlanCheckerConfigurationDir(), "configDirectory is null");
     }
@@ -61,9 +62,11 @@ public void addPlanCheckerProviderFactory(PlanCheckerProviderFactory planChecker
         }
     }
 
-    public void loadPlanCheckerProviders()
+    public void loadPlanCheckerProviders(NodeManager nodeManager)
             throws IOException
     {
+        PlanCheckerProviderContext planCheckerProviderContext = new PlanCheckerProviderContext(simplePlanFragmentSerde, nodeManager);
+
         for (File file : listFiles(configDirectory)) {
             if (file.isFile() && file.getName().endsWith(".properties")) {
                 // unlike function namespaces and connectors, we don't have a concept of catalog
@@ -75,7 +78,7 @@ public void loadPlanCheckerProviders()
                         file.getAbsoluteFile(),
                         PLAN_CHECKER_PROVIDER_NAME);
                 String planCheckerProviderName = properties.remove(PLAN_CHECKER_PROVIDER_NAME);
-                log.info("-- Loading plan checker provider %s--", planCheckerProviderName);
+                log.info("-- Loading plan checker provider [%s] --", planCheckerProviderName);
                 PlanCheckerProviderFactory providerFactory = providerFactories.get(planCheckerProviderName);
                 checkState(providerFactory != null,
                         "No planCheckerProviderFactory found for '%s'. Available factories were %s", planCheckerProviderName, providerFactories.keySet());

presto-main/src/test/java/com/facebook/presto/sql/planner/sanity/TestPlanCheckerProviderManager.java

@@ -20,6 +20,7 @@
 import com.facebook.presto.spi.plan.PlanCheckerProviderFactory;
 import com.facebook.presto.spi.plan.SimplePlanFragment;
 import com.facebook.presto.sql.planner.plan.JsonCodecSimplePlanFragmentSerde;
+import com.facebook.presto.testing.TestingNodeManager;
 import com.google.common.collect.ImmutableList;
 import org.testng.annotations.Test;
 
@@ -40,7 +41,7 @@ public void testLoadPlanCheckerProviders()
                 .setPlanCheckerConfigurationDir(new File("src/test/resources/plan-checkers"));
         PlanCheckerProviderManager planCheckerProviderManager = new PlanCheckerProviderManager(new JsonCodecSimplePlanFragmentSerde(JsonCodec.jsonCodec(SimplePlanFragment.class)), planCheckerProviderManagerConfig);
         planCheckerProviderManager.addPlanCheckerProviderFactory(new TestingPlanCheckerProviderFactory());
-        planCheckerProviderManager.loadPlanCheckerProviders();
+        planCheckerProviderManager.loadPlanCheckerProviders(new TestingNodeManager());
         assertEquals(planCheckerProviderManager.getPlanCheckerProviders(), ImmutableList.of(TESTING_PLAN_CHECKER_PROVIDER));
     }
 
@@ -51,7 +52,7 @@ public void testLoadUnregisteredPlanCheckerProvider()
         PlanCheckerProviderManagerConfig planCheckerProviderManagerConfig = new PlanCheckerProviderManagerConfig()
                 .setPlanCheckerConfigurationDir(new File("src/test/resources/plan-checkers"));
         PlanCheckerProviderManager planCheckerProviderManager = new PlanCheckerProviderManager(new JsonCodecSimplePlanFragmentSerde(JsonCodec.jsonCodec(SimplePlanFragment.class)), planCheckerProviderManagerConfig);
-        planCheckerProviderManager.loadPlanCheckerProviders();
+        planCheckerProviderManager.loadPlanCheckerProviders(new TestingNodeManager());
     }
 
     public static class TestingPlanCheckerProviderFactory

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

@@ -50,6 +50,7 @@ target_link_libraries(
   presto_function_metadata
   presto_http
   presto_operators
+  presto_velox_conversion
   velox_aggregates
   velox_caching
   velox_common_base

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

@@ -39,6 +39,7 @@
 #include "presto_cpp/main/operators/UnsafeRowExchangeSource.h"
 #include "presto_cpp/main/types/FunctionMetadata.h"
 #include "presto_cpp/main/types/PrestoToVeloxQueryPlan.h"
+#include "presto_cpp/main/types/VeloxPlanConversion.h"
 #include "velox/common/base/Counters.h"
 #include "velox/common/base/StatsReporter.h"
 #include "velox/common/caching/CacheTTLController.h"
@@ -478,6 +479,9 @@ void PrestoServer::run() {
 
   pool_ =
       velox::memory::MemoryManager::getInstance()->addLeafPool("PrestoServer");
+  nativeWorkerPool_ = velox::memory::MemoryManager::getInstance()->addLeafPool(
+      "PrestoNativeWorker");
+
   taskManager_ = std::make_unique<TaskManager>(
       driverExecutor_.get(), httpSrvCpuExecutor_.get(), spillerExecutor_.get());
 
@@ -1475,6 +1479,24 @@ void PrestoServer::registerSidecarEndpoints() {
          proxygen::ResponseHandler* downstream) {
         http::sendOkResponse(downstream, getFunctionsMetadata());
       });
+  httpServer_->registerPost(
+      "/v1/velox/plan",
+      [server = this](
+          proxygen::HTTPMessage* message,
+          const std::vector<std::unique_ptr<folly::IOBuf>>& body,
+          proxygen::ResponseHandler* downstream) {
+        std::string planFragmentJson = util::extractMessageBody(body);
+        protocol::PlanConversionResponse response = prestoToVeloxPlanConversion(
+            planFragmentJson,
+            server->nativeWorkerPool_.get(),
+            server->getVeloxPlanValidator());
+        if (response.failures.empty()) {
+          http::sendOkResponse(downstream, json(response));
+        } else {
+          http::sendResponse(
+              downstream, json(response), http::kHttpUnprocessableContent);
+        }
+      });
 }
 
 protocol::NodeStatus PrestoServer::fetchNodeStatus() {

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

@@ -268,6 +268,7 @@ class PrestoServer {
   std::unique_ptr<Announcer> announcer_;
   std::unique_ptr<PeriodicHeartbeatManager> heartbeatManager_;
   std::shared_ptr<velox::memory::MemoryPool> pool_;
+  std::shared_ptr<velox::memory::MemoryPool> nativeWorkerPool_;
   std::unique_ptr<TaskManager> taskManager_;
   std::unique_ptr<TaskResource> taskResource_;
   std::atomic<NodeState> nodeState_{NodeState::kActive};

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

@@ -221,13 +221,7 @@ proxygen::RequestHandler* TaskResource::createOrUpdateTaskImpl(
             httpSrvCpuExecutor_,
             [this, &body, taskId, createOrUpdateFunc]() {
               const auto startProcessCpuTimeNs = util::getProcessCpuTimeNs();
-
-              // TODO Avoid copy
-              std::ostringstream oss;
-              for (auto& buf : body) {
-                oss << std::string((const char*)buf->data(), buf->length());
-              }
-              std::string updateJson = oss.str();
+              std::string updateJson = util::extractMessageBody(body);
 
               std::unique_ptr<protocol::TaskInfo> taskInfo;
               try {

presto-native-execution/presto_cpp/main/common/Utils.cpp

@@ -64,4 +64,13 @@ void installSignalHandler() {
 #endif // __APPLE__
 }
 
+std::string extractMessageBody(
+    const std::vector<std::unique_ptr<folly::IOBuf>>& body) {
+  // TODO Avoid copy
+  std::ostringstream oss;
+  for (auto& buf : body) {
+    oss << std::string((const char*)buf->data(), buf->length());
+  }
+  return oss.str();
+}
 } // namespace facebook::presto::util

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

@@ -44,4 +44,7 @@ long getProcessCpuTimeNs();
 /// context such as the queryId.
 void installSignalHandler();
 
+std::string extractMessageBody(
+    const std::vector<std::unique_ptr<folly::IOBuf>>& body);
+
 } // namespace facebook::presto::util

presto-native-execution/presto_cpp/main/http/HttpClient.cpp

@@ -20,6 +20,7 @@
 #include <folly/synchronization/Latch.h>
 #include <velox/common/base/Exceptions.h>
 #include "presto_cpp/main/common/Configs.h"
+#include "presto_cpp/main/common/Utils.h"
 #include "presto_cpp/main/http/HttpClient.h"
 
 namespace facebook::presto::http {
@@ -169,11 +170,7 @@ HttpResponse::nextAllocationSize(uint64_t dataLength) const {
 std::string HttpResponse::dumpBodyChain() const {
   std::string responseBody;
   if (!bodyChain_.empty()) {
-    std::ostringstream oss;
-    for (const auto& buf : bodyChain_) {
-      oss << std::string((const char*)buf->data(), buf->length());
-    }
-    responseBody = oss.str();
+    responseBody = util::extractMessageBody(bodyChain_);
   }
   return responseBody;
 }

presto-native-execution/presto_cpp/main/http/HttpConstants.h

@@ -20,6 +20,7 @@ const uint16_t kHttpNoContent = 204;
 const uint16_t kHttpBadRequest = 400;
 const uint16_t kHttpUnauthorized = 401;
 const uint16_t kHttpNotFound = 404;
+const uint16_t kHttpUnprocessableContent = 422;
 const uint16_t kHttpInternalServerError = 500;
 
 const char kMimeTypeApplicationJson[] = "application/json";

presto-native-execution/presto_cpp/main/http/HttpServer.cpp

@@ -25,22 +25,7 @@ void sendOkResponse(proxygen::ResponseHandler* downstream) {
 }
 
 void sendOkResponse(proxygen::ResponseHandler* downstream, const json& body) {
-  // nlohmann::json throws when it finds invalid UTF-8 characters. In that case
-  // the server will crash. We handle such situation here and generate body
-  // replacing the faulty UTF-8 sequences.
-  std::string messageBody;
-  try {
-    messageBody = body.dump();
-  } catch (const std::exception& e) {
-    messageBody =
-        body.dump(-1, ' ', false, nlohmann::detail::error_handler_t::replace);
-    LOG(WARNING) << "Failed to serialize json to string. "
-                    "Will retry with 'replace' option. "
-                    "Json Dump:\n"
-                 << messageBody;
-  }
-
-  sendOkResponse(downstream, messageBody);
+  sendResponse(downstream, body, http::kHttpOk);
 }
 
 void sendOkResponse(
@@ -75,6 +60,33 @@ void sendErrorResponse(
       .sendWithEOM();
 }
 
+void sendResponse(
+    proxygen::ResponseHandler* downstream,
+    const json& body,
+    uint16_t status) {
+  // nlohmann::json throws when it finds invalid UTF-8 characters. In that case
+  // the server will crash. We handle such situation here and generate body
+  // replacing the faulty UTF-8 sequences.
+  std::string messageBody;
+  try {
+    messageBody = body.dump();
+  } catch (const std::exception& e) {
+    messageBody =
+        body.dump(-1, ' ', false, nlohmann::detail::error_handler_t::replace);
+    LOG(WARNING) << "Failed to serialize json to string. "
+                    "Will retry with 'replace' option. "
+                    "Json Dump:\n"
+                 << messageBody;
+  }
+
+  proxygen::ResponseBuilder(downstream)
+      .status(status, "")
+      .header(
+          proxygen::HTTP_HEADER_CONTENT_TYPE, http::kMimeTypeApplicationJson)
+      .body(messageBody)
+      .sendWithEOM();
+}
+
 HttpConfig::HttpConfig(const folly::SocketAddress& address, bool reusePort)
     : address_(address), reusePort_(reusePort) {}
 

presto-native-execution/presto_cpp/main/http/HttpServer.h

@@ -42,6 +42,11 @@ void sendErrorResponse(
     const std::string& error = "",
     uint16_t status = http::kHttpInternalServerError);
 
+void sendResponse(
+    proxygen::ResponseHandler* downstream,
+    const json& body,
+    uint16_t status);
+
 class AbstractRequestHandler : public proxygen::RequestHandler {
  public:
   void onRequest(