refactor: Migrate production code to ConnectorRegistry API and deprecate free functions (#16986)

Summary:
Pull Request resolved: #16986

Replace legacy free functions (registerConnector, unregisterConnector,
getConnector, hasConnector) with ConnectorRegistry methods in all
remaining velox/ production code: TableScan, TableWriter,
IndexLookupJoin, TraceReplayRunner, PyConnectors, and experimental
operators.

Mark the free functions [[deprecated]] to prevent new usages.

Reviewed By: jagill

Differential Revision: D98963405

Author: mbasmanova

velox/common/ScopedRegistry.h

@@ -16,6 +16,10 @@
 
 #pragma once
 
+#include <memory>
+#include <utility>
+#include <vector>
+
 #include "folly/Synchronized.h"
 #include "folly/container/F14Map.h"
 #include "folly/container/F14Set.h"

velox/common/tests/ScopedRegistryTest.cpp

@@ -16,7 +16,11 @@
 
 #include "velox/common/ScopedRegistry.h"
 
-#include <gmock/gmock.h>
+#include <map>
+#include <memory>
+#include <set>
+#include <string>
+
 #include <gtest/gtest.h>
 
 namespace facebook::velox {
@@ -25,7 +29,7 @@ namespace {
 // Minimal value type for testing.
 class TestEntry {
  public:
-  explicit TestEntry(const std::string& name) : name_{name} {}
+  explicit TestEntry(std::string name) : name_{std::move(name)} {}
 
   const std::string& name() const {
     return name_;
@@ -94,15 +98,17 @@ TEST(ScopedRegistryTest, snapshot) {
   for (const auto& [key, _] : entries) {
     keys.insert(key);
   }
-  EXPECT_THAT(keys, testing::UnorderedElementsAre("a", "b"));
+  EXPECT_EQ(keys.size(), 2);
+  EXPECT_TRUE(keys.count("a"));
+  EXPECT_TRUE(keys.count("b"));
 }
 
 TEST(ScopedRegistryTest, parentFallback) {
   ScopedRegistry<std::string, TestEntry> parent;
   auto entry = std::make_shared<TestEntry>("from-parent");
   parent.insert("key", entry);
 
-  ScopedRegistry<std::string, TestEntry> child(&parent);
+  const ScopedRegistry<std::string, TestEntry> child(&parent);
   EXPECT_EQ(child.find("key"), entry);
 }
 

velox/connectors/Connector.cpp

@@ -15,6 +15,12 @@
  */
 
 #include "velox/connectors/Connector.h"
+
+#include <memory>
+#include <string>
+
+#include "velox/common/ScopedRegistry.h"
+#include "velox/common/base/Exceptions.h"
 #include "velox/connectors/ConnectorRegistryInternal.h"
 
 namespace facebook::velox::connector {
@@ -24,7 +30,7 @@ ScopedRegistry<std::string, Connector>& connectors() {
   return instance;
 }
 
-bool registerConnector(std::shared_ptr<Connector> connector) {
+bool registerConnector(const std::shared_ptr<Connector>& connector) {
   connectors().insert(connector->connectorId(), connector);
   return true;
 }

velox/connectors/Connector.h

@@ -768,14 +768,18 @@ class Connector {
       trackers_;
 };
 
-// Legacy free functions. Prefer ConnectorRegistry methods for new code.
+/// Deprecated free functions. Use ConnectorRegistry methods instead.
 
-bool registerConnector(std::shared_ptr<Connector> connector);
+[[deprecated("Use ConnectorRegistry::global().insert() instead.")]]
+bool registerConnector(const std::shared_ptr<Connector>& connector);
 
+[[deprecated("Use ConnectorRegistry::tryGet() instead.")]]
 bool hasConnector(const std::string& connectorId);
 
+[[deprecated("Use ConnectorRegistry::global().erase() instead.")]]
 bool unregisterConnector(const std::string& connectorId);
 
+[[deprecated("Use ConnectorRegistry::tryGet() instead.")]]
 std::shared_ptr<Connector> getConnector(const std::string& connectorId);
 
 } // namespace facebook::velox::connector

velox/connectors/ConnectorRegistry.cpp

@@ -15,8 +15,14 @@
  */
 
 #include "velox/connectors/ConnectorRegistry.h"
-#include "velox/connectors/ConnectorRegistryInternal.h"
 
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "velox/connectors/Connector.h"
+#include "velox/connectors/ConnectorRegistryInternal.h"
 #include "velox/core/QueryCtx.h"
 
 namespace facebook::velox::connector {

velox/connectors/ConnectorRegistry.h

@@ -16,6 +16,12 @@
 
 #pragma once
 
+#include <memory>
+#include <string>
+#include <string_view>
+#include <utility>
+#include <vector>
+
 #include "velox/common/ScopedRegistry.h"
 #include "velox/connectors/Connector.h"
 

velox/connectors/ConnectorRegistryInternal.h

@@ -16,6 +16,8 @@
 
 #pragma once
 
+#include <string>
+
 #include "velox/common/ScopedRegistry.h"
 #include "velox/connectors/Connector.h"
 

velox/connectors/tests/ConnectorTest.cpp

@@ -15,14 +15,18 @@
  */
 
 #include "velox/connectors/Connector.h"
+
+#include <memory>
+#include <utility>
+
+#include <fmt/format.h>
+#include <gtest/gtest.h>
+
 #include "velox/common/config/Config.h"
 #include "velox/common/memory/Memory.h"
 #include "velox/connectors/ConnectorRegistry.h"
 #include "velox/core/QueryCtx.h"
 
-#include <gmock/gmock.h>
-#include <gtest/gtest.h>
-
 namespace facebook::velox::connector {
 namespace {
 

velox/core/QueryCtx.h

@@ -19,9 +19,9 @@
 #include <folly/Executor.h>
 #include <folly/Synchronized.h>
 #include <folly/container/F14Map.h>
-#include <folly/executors/CPUThreadPoolExecutor.h>
 #include <deque>
 #include <functional>
+#include <string_view>
 #include <typeindex>
 #include "velox/common/base/Exceptions.h"
 #include "velox/common/caching/AsyncDataCache.h"

velox/exec/IndexLookupJoin.cpp

@@ -18,6 +18,7 @@
 #include "velox/buffer/Buffer.h"
 #include "velox/common/testutil/TestValue.h"
 #include "velox/connectors/Connector.h"
+#include "velox/connectors/ConnectorRegistry.h"
 #include "velox/core/QueryConfig.h"
 #include "velox/exec/OperatorTraceWriter.h"
 #include "velox/exec/OperatorType.h"
@@ -347,7 +348,9 @@ IndexLookupJoin::IndexLookupJoin(
               operatorType(),
               lookupTableHandle_->connectorId()),
           spillConfig_.has_value() ? &(spillConfig_.value()) : nullptr)},
-      connector_(connector::getConnector(lookupTableHandle_->connectorId())),
+      connector_(
+          connector::ConnectorRegistry::tryGet(
+              lookupTableHandle_->connectorId())),
       maxNumInputBatches_(
           1 + driverCtx->queryConfig().indexLookupJoinMaxPrefetchBatches()),
       joinNode_{joinNode} {