refactor: Migrate test and fuzzer code to ConnectorRegistry API

Summary:
Replace legacy free functions (registerConnector, unregisterConnector,
getConnector) with ConnectorRegistry methods across all test, fuzzer,
benchmark, and example code in velox/.

This prepares for deprecating the free functions in a follow-up commit.

Differential Revision: D98961767

Author: mbasmanova

velox/benchmarks/QueryBenchmarkBase.cpp

@@ -18,6 +18,7 @@
 #include <iostream>
 #include "velox/common/base/SuccinctPrinter.h"
 #include "velox/common/file/FileSystems.h"
+#include "velox/connectors/ConnectorRegistry.h"
 #include "velox/connectors/hive/HiveConnector.h"
 #include "velox/dwio/dwrf/RegisterDwrfReader.h"
 #include "velox/dwio/parquet/RegisterParquetReader.h"
@@ -210,7 +211,8 @@ void QueryBenchmarkBase::initialize() {
   connector::hive::HiveConnectorFactory factory;
   auto hiveConnector =
       factory.newConnector(kHiveConnectorId, properties, ioExecutor_.get());
-  connector::registerConnector(hiveConnector);
+  connector::ConnectorRegistry::global().insert(
+      hiveConnector->connectorId(), hiveConnector);
   parquet::registerParquetReaderFactory();
   dwrf::registerDwrfReaderFactory();
 }

velox/connectors/fuzzer/tests/FuzzerConnectorTestBase.h

@@ -14,6 +14,7 @@
  * limitations under the License.
  */
 
+#include "velox/connectors/ConnectorRegistry.h"
 #include "velox/connectors/fuzzer/FuzzerConnector.h"
 #include "velox/exec/tests/utils/OperatorTestBase.h"
 #include "velox/vector/fuzzer/VectorFuzzer.h"
@@ -28,11 +29,12 @@ class FuzzerConnectorTestBase : public exec::test::OperatorTestBase {
     OperatorTestBase::SetUp();
     connector::fuzzer::FuzzerConnectorFactory factory;
     auto fuzzerConnector = factory.newConnector(kFuzzerConnectorId, nullptr);
-    connector::registerConnector(fuzzerConnector);
+    connector::ConnectorRegistry::global().insert(
+        fuzzerConnector->connectorId(), fuzzerConnector);
   }
 
   void TearDown() override {
-    connector::unregisterConnector(kFuzzerConnectorId);
+    connector::ConnectorRegistry::global().erase(kFuzzerConnectorId);
     OperatorTestBase::TearDown();
   }
 

velox/connectors/hive/iceberg/tests/IcebergConnectorTest.cpp

@@ -16,6 +16,7 @@
 
 #include "velox/connectors/hive/iceberg/IcebergConnector.h"
 #include <gtest/gtest.h>
+#include "velox/connectors/ConnectorRegistry.h"
 #include "velox/connectors/hive/HiveConfig.h"
 #include "velox/connectors/hive/iceberg/tests/IcebergTestBase.h"
 
@@ -27,12 +28,13 @@ class IcebergConnectorTest : public test::IcebergTestBase {
  protected:
   static void resetIcebergConnector(
       const std::shared_ptr<const config::ConfigBase>& config) {
-    unregisterConnector(test::kIcebergConnectorId);
+    ConnectorRegistry::global().erase(test::kIcebergConnectorId);
 
     IcebergConnectorFactory factory;
     auto icebergConnector =
         factory.newConnector(test::kIcebergConnectorId, config);
-    registerConnector(icebergConnector);
+    ConnectorRegistry::global().insert(
+        icebergConnector->connectorId(), icebergConnector);
   }
 };
 
@@ -45,7 +47,7 @@ TEST_F(IcebergConnectorTest, connectorConfiguration) {
   resetIcebergConnector(customConfig);
 
   // Verify connector was registered successfully with custom config.
-  auto icebergConnector = getConnector(test::kIcebergConnectorId);
+  auto icebergConnector = ConnectorRegistry::tryGet(test::kIcebergConnectorId);
   ASSERT_NE(icebergConnector, nullptr);
 
   auto config = icebergConnector->connectorConfig();
@@ -57,7 +59,7 @@ TEST_F(IcebergConnectorTest, connectorConfiguration) {
 }
 
 TEST_F(IcebergConnectorTest, connectorProperties) {
-  auto icebergConnector = getConnector(test::kIcebergConnectorId);
+  auto icebergConnector = ConnectorRegistry::tryGet(test::kIcebergConnectorId);
   ASSERT_NE(icebergConnector, nullptr);
 
   ASSERT_TRUE(icebergConnector->canAddDynamicFilter());

velox/connectors/hive/iceberg/tests/IcebergReadTest.cpp

@@ -19,6 +19,7 @@
 #include "velox/common/base/tests/GTestUtils.h"
 #include "velox/common/encode/Base64.h"
 #include "velox/common/file/FileSystems.h"
+#include "velox/connectors/ConnectorRegistry.h"
 #include "velox/connectors/hive/HiveConnectorSplit.h"
 #include "velox/connectors/hive/iceberg/IcebergConnector.h"
 #include "velox/connectors/hive/iceberg/IcebergDeleteFile.h"
@@ -58,11 +59,12 @@ class HiveIcebergTest : public HiveConnectorTestBase {
         std::make_shared<config::ConfigBase>(
             std::unordered_map<std::string, std::string>()),
         ioExecutor_.get());
-    connector::registerConnector(icebergConnector);
+    connector::ConnectorRegistry::global().insert(
+        icebergConnector->connectorId(), icebergConnector);
   }
 
   void TearDown() override {
-    connector::unregisterConnector(kIcebergConnectorId);
+    connector::ConnectorRegistry::global().erase(kIcebergConnectorId);
     HiveConnectorTestBase::TearDown();
   }
 

velox/connectors/hive/iceberg/tests/IcebergTestBase.cpp

@@ -18,6 +18,7 @@
 
 #include <filesystem>
 
+#include "velox/connectors/ConnectorRegistry.h"
 #include "velox/connectors/hive/TableHandle.h"
 #include "velox/connectors/hive/iceberg/IcebergColumnHandle.h"
 #include "velox/connectors/hive/iceberg/IcebergConfig.h"
@@ -46,7 +47,8 @@ void IcebergTestBase::SetUp() {
       std::make_shared<config::ConfigBase>(
           std::unordered_map<std::string, std::string>()),
       ioExecutor_.get());
-  registerConnector(icebergConnector);
+  ConnectorRegistry::global().insert(
+      icebergConnector->connectorId(), icebergConnector);
 
   connectorSessionProperties_ = std::make_shared<config::ConfigBase>(
       std::unordered_map<std::string, std::string>(), true);
@@ -75,7 +77,7 @@ void IcebergTestBase::TearDown() {
   opPool_.reset();
   root_.reset();
   queryCtx_.reset();
-  unregisterConnector(kIcebergConnectorId);
+  ConnectorRegistry::global().erase(kIcebergConnectorId);
   HiveConnectorTestBase::TearDown();
 }
 

velox/connectors/hive/paimon/tests/PaimonConnectorTest.cpp

@@ -17,6 +17,7 @@
 
 #include <gtest/gtest.h>
 
+#include "velox/connectors/ConnectorRegistry.h"
 #include "velox/connectors/hive/HiveConfig.h"
 #include "velox/exec/tests/utils/HiveConnectorTestBase.h"
 
@@ -33,17 +34,17 @@ class PaimonConnectorTest : public exec::test::HiveConnectorTestBase {
         std::unordered_map<std::string, std::string>{});
     auto connector =
         PaimonConnectorFactory().newConnector(kPaimonConnectorId, config);
-    registerConnector(connector);
+    ConnectorRegistry::global().insert(connector->connectorId(), connector);
   }
 
   void TearDown() override {
-    unregisterConnector(kPaimonConnectorId);
+    ConnectorRegistry::global().erase(kPaimonConnectorId);
     HiveConnectorTestBase::TearDown();
   }
 };
 
 TEST_F(PaimonConnectorTest, connectorRegistration) {
-  auto connector = getConnector(kPaimonConnectorId);
+  auto connector = ConnectorRegistry::tryGet(kPaimonConnectorId);
   ASSERT_NE(connector, nullptr);
   ASSERT_NE(connector->connectorConfig(), nullptr);
 }

velox/connectors/hive/storage_adapters/gcs/tests/GcsMultipleEndpointsTest.cpp

@@ -18,6 +18,7 @@
 #include <folly/init/Init.h>
 
 #include "gtest/gtest.h"
+#include "velox/connectors/ConnectorRegistry.h"
 #include "velox/connectors/hive/HiveConnector.h"
 #include "velox/connectors/hive/HiveConnectorSplit.h"
 #include "velox/connectors/hive/storage_adapters/gcs/RegisterGcsFileSystem.h"
@@ -70,8 +71,10 @@ class GcsMultipleEndpointsTest : public testing::Test,
         std::string(connectorId2),
         gcsEmulatorTwo_->hiveConfig(config2Override),
         ioExecutor_.get());
-    connector::registerConnector(hiveConnector1);
-    connector::registerConnector(hiveConnector2);
+    connector::ConnectorRegistry::global().insert(
+        hiveConnector1->connectorId(), hiveConnector1);
+    connector::ConnectorRegistry::global().insert(
+        hiveConnector2->connectorId(), hiveConnector2);
   }
 
   void TearDown() override {
@@ -192,8 +195,8 @@ TEST_F(GcsMultipleEndpointsTest, baseEndpoints) {
 
   testJoin(kExpectedRows, gcsBucket, kConnectorId1, kConnectorId2);
 
-  connector::unregisterConnector(std::string(kConnectorId1));
-  connector::unregisterConnector(std::string(kConnectorId2));
+  connector::ConnectorRegistry::global().erase(std::string(kConnectorId1));
+  connector::ConnectorRegistry::global().erase(std::string(kConnectorId2));
 }
 
 } // namespace facebook::velox::filesystems

velox/connectors/hive/storage_adapters/s3fs/tests/S3MultipleEndpointsTest.cpp

@@ -17,6 +17,7 @@
 #include <folly/init/Init.h>
 
 #include "gtest/gtest.h"
+#include "velox/connectors/ConnectorRegistry.h"
 #include "velox/connectors/hive/HiveConnector.h"
 #include "velox/connectors/hive/storage_adapters/s3fs/RegisterS3FileSystem.h"
 #include "velox/connectors/hive/storage_adapters/s3fs/S3Util.h"
@@ -71,8 +72,10 @@ class S3MultipleEndpoints : public S3Test, public ::test::VectorTestBase {
         std::string(connectorId2),
         minioSecondServer_->hiveConfig(config2Override),
         ioExecutor_.get());
-    connector::registerConnector(hiveConnector1);
-    connector::registerConnector(hiveConnector2);
+    connector::ConnectorRegistry::global().insert(
+        hiveConnector1->connectorId(), hiveConnector1);
+    connector::ConnectorRegistry::global().insert(
+        hiveConnector2->connectorId(), hiveConnector2);
   }
 
   void TearDown() override {
@@ -191,8 +194,8 @@ TEST_F(S3MultipleEndpoints, baseEndpoints) {
 
   testJoin(kExpectedRows, outputDirectory, kConnectorId1, kConnectorId2);
 
-  connector::unregisterConnector(std::string(kConnectorId1));
-  connector::unregisterConnector(std::string(kConnectorId2));
+  connector::ConnectorRegistry::global().erase(std::string(kConnectorId1));
+  connector::ConnectorRegistry::global().erase(std::string(kConnectorId2));
 }
 
 TEST_F(S3MultipleEndpoints, bucketEndpoints) {
@@ -218,8 +221,8 @@ TEST_F(S3MultipleEndpoints, bucketEndpoints) {
 
   testJoin(kExpectedRows, outputDirectory, kConnectorId1, kConnectorId2);
 
-  connector::unregisterConnector(std::string(kConnectorId1));
-  connector::unregisterConnector(std::string(kConnectorId2));
+  connector::ConnectorRegistry::global().erase(std::string(kConnectorId1));
+  connector::ConnectorRegistry::global().erase(std::string(kConnectorId2));
 }
 
 } // namespace facebook::velox

velox/connectors/hive/storage_adapters/s3fs/tests/S3ReadTest.cpp

@@ -18,6 +18,7 @@
 #include <gtest/gtest.h>
 
 #include "velox/common/memory/Memory.h"
+#include "velox/connectors/ConnectorRegistry.h"
 #include "velox/connectors/hive/HiveConnector.h"
 #include "velox/connectors/hive/storage_adapters/s3fs/RegisterS3FileSystem.h"
 #include "velox/connectors/hive/storage_adapters/s3fs/tests/S3Test.h"
@@ -43,14 +44,15 @@ class S3ReadTest : public S3Test, public ::test::VectorTestBase {
     connector::hive::HiveConnectorFactory factory;
     auto hiveConnector =
         factory.newConnector(kHiveConnectorId, minioServer_->hiveConfig());
-    connector::registerConnector(hiveConnector);
+    connector::ConnectorRegistry::global().insert(
+        hiveConnector->connectorId(), hiveConnector);
     parquet::registerParquetReaderFactory();
   }
 
   void TearDown() override {
     parquet::unregisterParquetReaderFactory();
     filesystems::finalizeS3FileSystem();
-    connector::unregisterConnector(kHiveConnectorId);
+    connector::ConnectorRegistry::global().erase(kHiveConnectorId);
     S3Test::TearDown();
   }
 };

velox/connectors/hive/storage_adapters/test_common/InsertTest.h

@@ -18,6 +18,7 @@
 #include <gtest/gtest.h>
 
 #include "velox/common/memory/Memory.h"
+#include "velox/connectors/ConnectorRegistry.h"
 #include "velox/connectors/hive/HiveConnector.h"
 #include "velox/connectors/hive/HiveDataSink.h"
 #include "velox/dwio/parquet/RegisterParquetReader.h"
@@ -38,7 +39,8 @@ class InsertTest : public velox::test::VectorTestBase {
     connector::hive::HiveConnectorFactory factory;
     auto hiveConnector = factory.newConnector(
         exec::test::kHiveConnectorId, hiveConfig, ioExecutor);
-    connector::registerConnector(hiveConnector);
+    connector::ConnectorRegistry::global().insert(
+        hiveConnector->connectorId(), hiveConnector);
 
     parquet::registerParquetReaderFactory();
     parquet::registerParquetWriterFactory();
@@ -48,7 +50,7 @@ class InsertTest : public velox::test::VectorTestBase {
     parquet::unregisterParquetReaderFactory();
     parquet::unregisterParquetWriterFactory();
 
-    connector::unregisterConnector(exec::test::kHiveConnectorId);
+    connector::ConnectorRegistry::global().erase(exec::test::kHiveConnectorId);
   }
 
   void runInsertTest(