[native] Fix possible SEGV when registering connectors with same name

This PR addresses two issues:
1. Use "connectorName" for the variable name identifying an entry in the unordered
    connector map. The API used connectorId (in most cases) and "connectorName"
    in the unregister function. It appears best to use "connectorName" because that
    is used in the property name (connector.name=) to avoid confusion.
 2. If a connector is registered a second time the exception is thrown. When
    accessing the connectorName, which was a reference to the member in the
    PrestoToVeloxConnector class, it is deallocated because the object is destructed
    when the insert to the map fails.

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

@@ -37,14 +37,14 @@ connectors() {
 
 void registerPrestoToVeloxConnector(
     std::unique_ptr<const PrestoToVeloxConnector> connector) {
-  auto& connectorId = connector->connectorId();
+  auto connectorName = connector->connectorName();
   auto connectorProtocol = connector->createConnectorProtocol();
   VELOX_CHECK(
-      connectors().insert({connectorId, std::move(connector)}).second,
+      connectors().insert({connectorName, std::move(connector)}).second,
       "Connector {} is already registered",
-      connectorId);
+      connectorName);
   protocol::registerConnectorProtocol(
-      connectorId, std::move(connectorProtocol));
+      connectorName, std::move(connectorProtocol));
 }
 
 void unregisterPrestoToVeloxConnector(const std::string& connectorName) {
@@ -53,10 +53,10 @@ void unregisterPrestoToVeloxConnector(const std::string& connectorName) {
 }
 
 const PrestoToVeloxConnector& getPrestoToVeloxConnector(
-    const std::string& connectorId) {
-  auto it = connectors().find(connectorId);
+    const std::string& connectorName) {
+  auto it = connectors().find(connectorName);
   VELOX_CHECK(
-      it != connectors().end(), "Connector {} not registered", connectorId);
+      it != connectors().end(), "Connector {} not registered", connectorName);
   return *(it->second);
 }
 

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

@@ -32,14 +32,14 @@ void registerPrestoToVeloxConnector(
 void unregisterPrestoToVeloxConnector(const std::string& connectorName);
 
 const PrestoToVeloxConnector& getPrestoToVeloxConnector(
-    const std::string& connectorId);
+    const std::string& connectorName);
 
 class PrestoToVeloxConnector {
  public:
   virtual ~PrestoToVeloxConnector() = default;
 
-  [[nodiscard]] const std::string& connectorId() const {
-    return connectorId_;
+  [[nodiscard]] const std::string& connectorName() const {
+    return connectorName_;
   }
 
   [[nodiscard]] virtual std::unique_ptr<velox::connector::ConnectorSplit>
@@ -103,15 +103,15 @@ class PrestoToVeloxConnector {
   createConnectorProtocol() const = 0;
 
  protected:
-  explicit PrestoToVeloxConnector(std::string connectorId)
-      : connectorId_(std::move(connectorId)) {}
-  const std::string connectorId_;
+  explicit PrestoToVeloxConnector(std::string connectorName)
+      : connectorName_(std::move(connectorName)) {}
+  const std::string connectorName_;
 };
 
 class HivePrestoToVeloxConnector final : public PrestoToVeloxConnector {
  public:
-  explicit HivePrestoToVeloxConnector(std::string connectorId)
-      : PrestoToVeloxConnector(std::move(connectorId)) {}
+  explicit HivePrestoToVeloxConnector(std::string connectorName)
+      : PrestoToVeloxConnector(std::move(connectorName)) {}
 
   std::unique_ptr<velox::connector::ConnectorSplit> toVeloxSplit(
       const protocol::ConnectorId& catalogId,
@@ -161,8 +161,8 @@ class HivePrestoToVeloxConnector final : public PrestoToVeloxConnector {
 
 class IcebergPrestoToVeloxConnector final : public PrestoToVeloxConnector {
  public:
-  explicit IcebergPrestoToVeloxConnector(std::string connectorId)
-      : PrestoToVeloxConnector(std::move(connectorId)) {}
+  explicit IcebergPrestoToVeloxConnector(std::string connectorName)
+      : PrestoToVeloxConnector(std::move(connectorName)) {}
 
   std::unique_ptr<velox::connector::ConnectorSplit> toVeloxSplit(
       const protocol::ConnectorId& catalogId,
@@ -187,8 +187,8 @@ class IcebergPrestoToVeloxConnector final : public PrestoToVeloxConnector {
 
 class TpchPrestoToVeloxConnector final : public PrestoToVeloxConnector {
  public:
-  explicit TpchPrestoToVeloxConnector(std::string connectorId)
-      : PrestoToVeloxConnector(std::move(connectorId)) {}
+  explicit TpchPrestoToVeloxConnector(std::string connectorName)
+      : PrestoToVeloxConnector(std::move(connectorName)) {}
 
   std::unique_ptr<velox::connector::ConnectorSplit> toVeloxSplit(
       const protocol::ConnectorId& catalogId,

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

@@ -71,3 +71,21 @@ target_link_libraries(
 
 set_property(TARGET presto_expressions_test PROPERTY JOB_POOL_LINK
                                                      presto_link_job_pool)
+
+add_executable(presto_to_velox_connector_test PrestoToVeloxConnectorTest.cpp)
+
+add_test(
+  NAME presto_to_velox_connector_test
+  COMMAND presto_to_velox_connector_test
+  WORKING_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR})
+
+target_link_libraries(
+  presto_to_velox_connector_test
+  presto_protocol
+  presto_operators
+  presto_type_converter
+  presto_types
+  velox_hive_connector
+  velox_tpch_connector
+  gtest
+  gtest_main)

presto-native-execution/presto_cpp/main/types/tests/PrestoToVeloxConnectorTest.cpp

@@ -0,0 +1,57 @@
+/*
+ * 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/PrestoToVeloxConnector.h"
+#include <gtest/gtest.h>
+#include "velox/common/base/tests/GTestUtils.h"
+
+using namespace facebook::presto;
+using namespace facebook::velox;
+
+class PrestoToVeloxConnectorTest : public ::testing::Test {};
+
+TEST_F(PrestoToVeloxConnectorTest, registerVariousConnectors) {
+  std::vector<std::pair<std::string, std::unique_ptr<PrestoToVeloxConnector>>>
+      connectorList;
+  connectorList.emplace_back(std::move(std::pair(
+      "hive",
+      std::move(std::make_unique<HivePrestoToVeloxConnector>("hive")))));
+  connectorList.emplace_back(std::move(std::pair(
+      "hive-hadoop2",
+      std::move(
+          std::make_unique<HivePrestoToVeloxConnector>("hive-hadoop2")))));
+  connectorList.emplace_back(std::move(std::pair(
+      "iceberg",
+      std::move(std::make_unique<IcebergPrestoToVeloxConnector>("iceberg")))));
+  connectorList.emplace_back(std::move(std::pair(
+      "tpch",
+      std::move(std::make_unique<HivePrestoToVeloxConnector>("tpch")))));
+
+  for (auto& [connectorName, connector] : connectorList) {
+    registerPrestoToVeloxConnector(std::move(connector));
+    EXPECT_EQ(
+        connectorName,
+        getPrestoToVeloxConnector(connectorName).connectorName());
+    unregisterPrestoToVeloxConnector(connectorName);
+  }
+}
+
+TEST_F(PrestoToVeloxConnectorTest, addDuplicates) {
+  constexpr auto kConnectorName = "hive";
+  registerPrestoToVeloxConnector(
+      std::make_unique<HivePrestoToVeloxConnector>(kConnectorName));
+  VELOX_ASSERT_THROW(
+      registerPrestoToVeloxConnector(
+          std::make_unique<HivePrestoToVeloxConnector>(kConnectorName)),
+      fmt::format("Connector {} is already registered", kConnectorName));
+}