feat: Support connector factory registeration in Pre… (#26312)

```
== NO RELEASE NOTE ==
```

Author: xiaoxmeng

presto-native-execution/presto_cpp/main/connectors/Registration.cpp

@@ -33,65 +33,21 @@ namespace {
 constexpr char const* kHiveHadoop2ConnectorName = "hive-hadoop2";
 constexpr char const* kIcebergConnectorName = "iceberg";
 
-const std::unordered_map<
-    std::string,
-    const std::shared_ptr<velox::connector::ConnectorFactory>>&
-connectorFactories() {
-  static const std::unordered_map<
-      std::string,
-      const std::shared_ptr<velox::connector::ConnectorFactory>>
-      factories = {
-          {velox::connector::hive::HiveConnectorFactory::kHiveConnectorName,
-           std::make_shared<velox::connector::hive::HiveConnectorFactory>()},
-          {kHiveHadoop2ConnectorName,
-           std::make_shared<velox::connector::hive::HiveConnectorFactory>(
-               kHiveHadoop2ConnectorName)},
-          {velox::connector::tpch::TpchConnectorFactory::kTpchConnectorName,
-           std::make_shared<velox::connector::tpch::TpchConnectorFactory>()},
-          {kIcebergConnectorName,
-           std::make_shared<velox::connector::hive::HiveConnectorFactory>(
-               kIcebergConnectorName)},
-#ifdef PRESTO_ENABLE_ARROW_FLIGHT_CONNECTOR
-          {ArrowFlightConnectorFactory::kArrowFlightConnectorName,
-           std::make_shared<ArrowFlightConnectorFactory>()},
-#endif
-      };
-  return factories;
-}
-
 } // namespace
 
-velox::connector::ConnectorFactory* getConnectorFactory(
-    const std::string& connectorName) {
-  {
-#ifdef PRESTO_ENABLE_CUDF
-    if (velox::cudf_velox::CudfConfig::getInstance().enabled) {
-      if (connectorName ==
-          velox::connector::hive::HiveConnectorFactory::kHiveConnectorName) {
-        static const auto factory = std::make_shared<
-            velox::cudf_velox::connector::hive::CudfHiveConnectorFactory>();
-        return factory.get();
-      }
-      if (connectorName == kHiveHadoop2ConnectorName) {
-        static const auto factory = std::make_shared<
-            velox::cudf_velox::connector::hive::CudfHiveConnectorFactory>(
-            kHiveHadoop2ConnectorName);
-        return factory.get();
-      }
-    }
-#endif
-    auto it = connectorFactories().find(connectorName);
-    if (it != connectorFactories().end()) {
-      return it->second.get();
-    }
-  }
-  if (!velox::connector::hasConnectorFactory(connectorName)) {
-    VELOX_FAIL("ConnectorFactory with name '{}' not registered", connectorName);
+std::vector<std::string> listConnectorFactories() {
+  std::vector<std::string> names;
+  const auto& factories = detail::connectorFactories();
+  names.reserve(factories.size());
+  for (const auto& [name, _] : factories) {
+    names.push_back(name);
   }
-  return velox::connector::getConnectorFactory(connectorName).get();
+  return names;
 }
 
 void registerConnectors() {
+  registerConnectorFactories();
+
   registerPrestoToVeloxConnector(std::make_unique<HivePrestoToVeloxConnector>(
       velox::connector::hive::HiveConnectorFactory::kHiveConnectorName));
   registerPrestoToVeloxConnector(
@@ -118,4 +74,35 @@ void registerConnectors() {
       ArrowFlightConnectorFactory::kArrowFlightConnectorName));
 #endif
 }
+
+void registerConnectorFactories() {
+  // Register all connector factories using the facebook::presto namespace
+  // factory registry
+
+  // Register Hive connector factory
+  facebook::presto::registerConnectorFactory(
+      std::make_shared<facebook::velox::connector::hive::HiveConnectorFactory>());
+
+  // Register Hive Hadoop2 connector factory
+  facebook::presto::registerConnectorFactory(
+      std::make_shared<facebook::velox::connector::hive::HiveConnectorFactory>(
+          kHiveHadoop2ConnectorName));
+
+  // Register TPCH connector factory
+  facebook::presto::registerConnectorFactory(
+      std::make_shared<facebook::velox::connector::tpch::TpchConnectorFactory>());
+
+  // Register Iceberg connector factory (using Hive implementation)
+  facebook::presto::registerConnectorFactory(
+      std::make_shared<facebook::velox::connector::hive::HiveConnectorFactory>(
+          kIcebergConnectorName));
+
+#ifdef PRESTO_ENABLE_ARROW_FLIGHT_CONNECTOR
+  // Note: ArrowFlightConnectorFactory would need to be implemented in Presto
+  // namespace For now, keep the Velox version
+  facebook::presto::registerConnectorFactory(
+      std::make_shared<ArrowFlightConnectorFactory>());
+#endif
+}
+
 } // namespace facebook::presto

presto-native-execution/presto_cpp/main/connectors/Registration.h

@@ -13,17 +13,79 @@
  */
 #pragma once
 
+#include <memory>
 #include <string>
+#include <unordered_map>
+#include <vector>
+
+#include "folly/Executor.h"
+#include "velox/connectors/Connector.h"
 
 // Forward declaration for ConnectorFactory.
 namespace facebook::velox::connector {
 class ConnectorFactory;
 } // namespace facebook::velox::connector
 
 namespace facebook::presto {
+using facebook::velox::connector::ConnectorFactory; 
+
+namespace detail {
+inline std::unordered_map<std::string, std::shared_ptr<ConnectorFactory>>&
+connectorFactories() {
+  static std::unordered_map<std::string, std::shared_ptr<ConnectorFactory>>
+      factories;
+  return factories;
+}
+} // namespace detail
+
+/// Adds a factory for creating connectors to the registry using connector
+/// name as the key. Throws if factory with the same name is already present.
+/// Always returns true. The return value makes it easy to use with
+/// FB_ANONYMOUS_VARIABLE.
+inline bool registerConnectorFactory(
+    std::shared_ptr<ConnectorFactory> factory) {
+  const bool ok = detail::connectorFactories()
+                .insert({factory->connectorName(), factory})
+                .second;
+  VELOX_CHECK(
+      ok,
+      "ConnectorFactory with name '{}' is already registered",
+      factory->connectorName());
+  return true;
+}
+
+/// Returns true if a connector with the specified name has been registered,
+/// false otherwise.
+inline bool hasConnectorFactory(const std::string& connectorName) {
+  return detail::connectorFactories().count(connectorName) == 1;
+}
+
+/// Unregister a connector factory by name.
+/// Returns true if a connector with the specified name has been
+/// unregistered, false otherwise.
+inline bool unregisterConnectorFactory(const std::string& connectorName) {
+  const auto count = detail::connectorFactories().erase(connectorName);
+  return count == 1;
+}
+
+/// Returns a factory for creating connectors with the specified name.
+/// Throws if factory doesn't exist.
+inline std::shared_ptr<ConnectorFactory> getConnectorFactory(
+    const std::string& connectorName) {
+  auto it = detail::connectorFactories().find(connectorName);
+  VELOX_CHECK(
+      it != detail::connectorFactories().end(),
+      "ConnectorFactory with name '{}' not registered",
+      connectorName);
+  return it->second;
+}
+
+/// Returns a list of all registered connector factory names.
+std::vector<std::string> listConnectorFactories();
 
-velox::connector::ConnectorFactory* getConnectorFactory(
-    const std::string& connectorName);
+/// Registers all connector factories using the facebook::presto namespace
+/// factory registry.
+void registerConnectorFactories();
 
 void registerConnectors();
 

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

@@ -12,6 +12,7 @@
 add_executable(
   presto_server_test
   AnnouncerTest.cpp
+  ConnectorTest.cpp
   CoordinatorDiscovererTest.cpp
   HttpServerWrapper.cpp
   PeriodicMemoryCheckerTest.cpp