build(native): Update Velox and fix build

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

@@ -1264,14 +1264,12 @@ std::vector<std::string> PrestoServer::registerVeloxConnectors(
 
       // make sure connector type is supported
       getPrestoToVeloxConnector(connectorName);
-
-      std::shared_ptr<velox::connector::Connector> connector =
-          velox::connector::getConnectorFactory(connectorName)
-              ->newConnector(
-                  catalogName,
-                  std::move(properties),
-                  connectorIoExecutor_.get(),
-                  connectorCpuExecutor_.get());
+      auto connector = getConnectorFactory(connectorName)
+                           ->newConnector(
+                               catalogName,
+                               std::move(properties),
+                               connectorIoExecutor_.get(),
+                               connectorCpuExecutor_.get());
       velox::connector::registerConnector(connector);
     }
   }

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

@@ -28,44 +28,45 @@ namespace {
 constexpr char const* kHiveHadoop2ConnectorName = "hive-hadoop2";
 constexpr char const* kIcebergConnectorName = "iceberg";
 
-void registerConnectorFactories() {
-  // These checks for connector factories can be removed after we remove the
-  // registrations from the Velox library.
-  if (!velox::connector::hasConnectorFactory(
-          velox::connector::hive::HiveConnectorFactory::kHiveConnectorName)) {
-    velox::connector::registerConnectorFactory(
-        std::make_shared<velox::connector::hive::HiveConnectorFactory>());
-    velox::connector::registerConnectorFactory(
-        std::make_shared<velox::connector::hive::HiveConnectorFactory>(
-            kHiveHadoop2ConnectorName));
-  }
-  if (!velox::connector::hasConnectorFactory(
-          velox::connector::tpch::TpchConnectorFactory::kTpchConnectorName)) {
-    velox::connector::registerConnectorFactory(
-        std::make_shared<velox::connector::tpch::TpchConnectorFactory>());
-  }
-
-  // Register Velox connector factory for iceberg.
-  // The iceberg catalog is handled by the hive connector factory.
-  if (!velox::connector::hasConnectorFactory(kIcebergConnectorName)) {
-    velox::connector::registerConnectorFactory(
-        std::make_shared<velox::connector::hive::HiveConnectorFactory>(
-            kIcebergConnectorName));
-  }
-
+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
-  if (!velox::connector::hasConnectorFactory(
-          ArrowFlightConnectorFactory::kArrowFlightConnectorName)) {
-    velox::connector::registerConnectorFactory(
-        std::make_shared<ArrowFlightConnectorFactory>());
-  }
+          {ArrowFlightConnectorFactory::kArrowFlightConnectorName,
+           std::make_shared<ArrowFlightConnectorFactory>()},
 #endif
+      };
+  return factories;
 }
+
 } // namespace
 
-void registerConnectors() {
-  registerConnectorFactories();
+velox::connector::ConnectorFactory* getConnectorFactory(
+    const std::string& connectorName) {
+  auto it = connectorFactories().find(connectorName);
+  VELOX_CHECK(
+      it != connectorFactories().end(),
+      "ConnectorFactory with name '{}' not registered",
+      connectorName);
+  return it->second.get();
+}
 
+void registerConnectors() {
   registerPrestoToVeloxConnector(std::make_unique<HivePrestoToVeloxConnector>(
       velox::connector::hive::HiveConnectorFactory::kHiveConnectorName));
   registerPrestoToVeloxConnector(

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

@@ -13,8 +13,18 @@
  */
 #pragma once
 
+#include <string>
+
+// Forward declaration for ConnectorFactory.
+namespace facebook::velox::connector {
+class ConnectorFactory;
+} // namespace facebook::velox::connector
+
 namespace facebook::presto {
 
+velox::connector::ConnectorFactory* getConnectorFactory(
+    const std::string& connectorName);
+
 void registerConnectors();
 
 } // namespace facebook::presto

presto-native-execution/presto_cpp/main/connectors/arrow_flight/tests/utils/ArrowFlightConnectorTestBase.cpp

@@ -25,16 +25,9 @@ namespace facebook::presto::test {
 
 void ArrowFlightConnectorTestBase::SetUp() {
   OperatorTestBase::SetUp();
-
-  if (!velox::connector::hasConnectorFactory(
-          presto::ArrowFlightConnectorFactory::kArrowFlightConnectorName)) {
-    velox::connector::registerConnectorFactory(
-        std::make_shared<presto::ArrowFlightConnectorFactory>());
-  }
+  presto::ArrowFlightConnectorFactory factory;
   velox::connector::registerConnector(
-      velox::connector::getConnectorFactory(
-          ArrowFlightConnectorFactory::kArrowFlightConnectorName)
-          ->newConnector(kFlightConnectorId, config_));
+      factory.newConnector(kFlightConnectorId, config_));
 
   ArrowFlightConfig config(config_);
   if (config.defaultServerPort().has_value()) {

presto-native-execution/presto_cpp/main/tests/ServerOperationTest.cpp

@@ -140,18 +140,11 @@ TEST_F(ServerOperationTest, buildServerOp) {
 
 TEST_F(ServerOperationTest, taskEndpoint) {
   // Setup environment for TaskManager
-  if (!connector::hasConnectorFactory(
-          connector::hive::HiveConnectorFactory::kHiveConnectorName)) {
-    connector::registerConnectorFactory(
-        std::make_shared<connector::hive::HiveConnectorFactory>());
-  }
-  auto hiveConnector =
-      connector::getConnectorFactory(
-          connector::hive::HiveConnectorFactory::kHiveConnectorName)
-          ->newConnector(
-              "test-hive",
-              std::make_shared<config::ConfigBase>(
-                  std::unordered_map<std::string, std::string>()));
+  connector::hive::HiveConnectorFactory factory;
+  auto hiveConnector = factory.newConnector(
+      "test-hive",
+      std::make_shared<config::ConfigBase>(
+          std::unordered_map<std::string, std::string>()));
   connector::registerConnector(hiveConnector);
 
   const auto driverExecutor = std::make_shared<folly::CPUThreadPoolExecutor>(

presto-native-execution/presto_cpp/main/tests/TaskManagerTest.cpp

@@ -195,11 +195,6 @@ class TaskManagerTest : public exec::test::OperatorTestBase,
   static void SetUpTestCase() {
     OperatorTestBase::SetUpTestCase();
     filesystems::registerLocalFileSystem();
-    if (!connector::hasConnectorFactory(
-            connector::hive::HiveConnectorFactory::kHiveConnectorName)) {
-      connector::registerConnectorFactory(
-          std::make_shared<connector::hive::HiveConnectorFactory>());
-    }
     test::setupMutableSystemConfig();
     SystemConfig::instance()->setValue(
         std::string(SystemConfig::kMemoryArbitratorKind), "SHARED");
@@ -233,13 +228,11 @@ class TaskManagerTest : public exec::test::OperatorTestBase,
 
     registerPrestoToVeloxConnector(std::make_unique<HivePrestoToVeloxConnector>(
         connector::hive::HiveConnectorFactory::kHiveConnectorName));
-    auto hiveConnector =
-        connector::getConnectorFactory(
-            connector::hive::HiveConnectorFactory::kHiveConnectorName)
-            ->newConnector(
-                kHiveConnectorId,
-                std::make_shared<config::ConfigBase>(
-                    std::unordered_map<std::string, std::string>()));
+    connector::hive::HiveConnectorFactory factory;
+    auto hiveConnector = factory.newConnector(
+        kHiveConnectorId,
+        std::make_shared<config::ConfigBase>(
+            std::unordered_map<std::string, std::string>()));
     connector::registerConnector(hiveConnector);
 
     rowType_ = ROW({"c0", "c1"}, {INTEGER(), VARCHAR()});