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

Summary:
Pull Request resolved: facebookincubator#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.

Differential Revision: D98963405

Author: mbasmanova

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.
 
+[[deprecated("Use ConnectorRegistry::global().insert() instead.")]]
 bool registerConnector(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/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} {

velox/exec/TableScan.cpp

@@ -16,6 +16,7 @@
 #include "velox/exec/TableScan.h"
 #include "velox/common/testutil/TestValue.h"
 #include "velox/common/time/Timer.h"
+#include "velox/connectors/ConnectorRegistry.h"
 #include "velox/exec/OperatorType.h"
 #include "velox/exec/Task.h"
 
@@ -90,7 +91,8 @@ TableScan::TableScan(
           driverCtx_->driverId,
           operatorType(),
           tableHandle_->connectorId())),
-      connector_(connector::getConnector(tableHandle_->connectorId())),
+      connector_(
+          connector::ConnectorRegistry::tryGet(tableHandle_->connectorId())),
       getOutputTimeLimitMs_(
           driverCtx_->queryConfig().tableScanGetOutputTimeLimitMs()),
       outputBatchRowsOverride_(

velox/exec/TableWriter.cpp

@@ -15,6 +15,7 @@
  */
 
 #include "velox/exec/TableWriter.h"
+#include "velox/connectors/ConnectorRegistry.h"
 #include "velox/exec/OperatorType.h"
 #include "velox/exec/Task.h"
 
@@ -63,7 +64,7 @@ TableWriter::TableWriter(
         &nonReclaimableSection_);
   }
   const auto& connectorId = tableWriteNode->insertTableHandle()->connectorId();
-  connector_ = connector::getConnector(connectorId);
+  connector_ = connector::ConnectorRegistry::tryGet(connectorId);
   connectorQueryCtx_ = operatorCtx_->createConnectorQueryCtx(
       connectorId,
       planNodeId(),

velox/experimental/cudf/exec/OperatorAdapters.cpp

@@ -28,6 +28,7 @@
 #include "velox/experimental/cudf/exec/Utilities.h"
 #include "velox/experimental/cudf/expression/ExpressionEvaluator.h"
 
+#include "velox/connectors/ConnectorRegistry.h"
 #include "velox/exec/AssignUniqueId.h"
 #include "velox/exec/CallbackSink.h"
 #include "velox/exec/FilterProject.h"
@@ -95,7 +96,7 @@ class TableScanAdapter : public OperatorAdapter {
     if (!tableScanNode) {
       return false;
     }
-    auto const& connector = velox::connector::getConnector(
+    auto const& connector = velox::connector::ConnectorRegistry::tryGet(
         tableScanNode->tableHandle()->connectorId());
     auto cudfHiveConnector = std::dynamic_pointer_cast<
         facebook::velox::cudf_velox::connector::hive::CudfHiveConnector>(

velox/experimental/wave/exec/TableScan.h

@@ -15,6 +15,7 @@
  */
 #pragma once
 
+#include "velox/connectors/ConnectorRegistry.h"
 #include "velox/experimental/wave/exec/WaveOperator.h"
 
 #include "velox/common/time/Timer.h"
@@ -49,7 +50,8 @@ class TableScan : public WaveSourceOperator {
                            ->queryConfig()
                            .preferredOutputBatchRows()) {
     defines_ = std::move(defines);
-    connector_ = connector::getConnector(tableHandle_->connectorId());
+    connector_ =
+        connector::ConnectorRegistry::tryGet(tableHandle_->connectorId());
   }
 
   std::vector<AdvanceResult> canAdvance(WaveStream& stream) override;

velox/experimental/wave/exec/WaveHiveDataSource.cpp

@@ -15,6 +15,7 @@
  */
 
 #include "velox/experimental/wave/exec/WaveHiveDataSource.h"
+#include "velox/connectors/ConnectorRegistry.h"
 
 namespace facebook::velox::wave {
 
@@ -164,7 +165,8 @@ void WaveHiveDataSource::registerConnector() {
   // Create hive connector with config...
   connector::hive::HiveConnectorFactory factory;
   auto hiveConnector = factory.newConnector("wavemock", config, nullptr);
-  connector::registerConnector(hiveConnector);
+  connector::ConnectorRegistry::global().insert(
+      hiveConnector->connectorId(), hiveConnector);
   connector::hive::HiveDataSource::registerWaveDelegateHook(
       [](const HiveTableHandlePtr& hiveTableHandle,
          const std::shared_ptr<common::ScanSpec>& scanSpec,

velox/python/runner/PyConnectors.cpp

@@ -15,6 +15,7 @@
  */
 
 #include "velox/python/runner/PyConnectors.h"
+#include "velox/connectors/ConnectorRegistry.h"
 #include "velox/connectors/hive/HiveConnector.h"
 #include "velox/connectors/tpch/TpchConnector.h"
 
@@ -35,7 +36,8 @@ void registerConnector(
   TConnectorFactory factory;
   auto connector = factory.newConnector(
       connectorId, configBase, folly::getGlobalCPUExecutor().get());
-  connector::registerConnector(connector);
+  connector::ConnectorRegistry::global().insert(
+      connector->connectorId(), connector);
   connectorRegistry().insert(connectorId);
 }
 
@@ -57,7 +59,8 @@ void registerTpch(
 
 // Is it ok to unregister connectors that were not registered.
 void unregister(const std::string& connectorId) {
-  if (!facebook::velox::connector::unregisterConnector(connectorId)) {
+  if (!facebook::velox::connector::ConnectorRegistry::global().erase(
+          connectorId)) {
     throw std::runtime_error(
         fmt::format("Unable to unregister connector '{}'", connectorId));
   }

velox/tool/trace/TraceReplayRunner.cpp

@@ -381,14 +381,15 @@ TraceReplayRunner::createReplayer() const {
         taskTraceMetadataReader_->connectorId(FLAGS_node_id);
     VELOX_CHECK(connectorId.has_value());
 
-    if (!connector::hasConnector(connectorId.value())) {
+    if (!connector::ConnectorRegistry::tryGet(connectorId.value())) {
       connector::hive::HiveConnectorFactory factory;
       const auto hiveConnector = factory.newConnector(
           connectorId.value(),
           std::make_shared<config::ConfigBase>(
               std::unordered_map<std::string, std::string>()),
           ioExecutor_.get());
-      connector::registerConnector(hiveConnector);
+      connector::ConnectorRegistry::global().insert(
+          hiveConnector->connectorId(), hiveConnector);
     }
     replayer = std::make_unique<tool::trace::TableScanReplayer>(
         FLAGS_root_dir,