refactor: Add arg cpuExecutor to ConnectorFactory::newConnector (#11861)

Summary:
Add an additional argument, cpuExecutor, to the ConnectorFactory's
newConnector interface. Connectors could send async operators to this
cpuExecutor to prevent occupying the driver executor.

Pull Request resolved: #11861

Reviewed By: xiaoxmeng

Differential Revision: D67231282

Pulled By: gggrace14

fbshipit-source-id: a92bb2c27ca26bf86a083ffea2a62b6c89231093

velox/connectors/Connector.h

@@ -482,7 +482,8 @@ class ConnectorFactory {
   virtual std::shared_ptr<Connector> newConnector(
       const std::string& id,
       std::shared_ptr<const config::ConfigBase> config,
-      folly::Executor* executor = nullptr) = 0;
+      folly::Executor* ioExecutor = nullptr,
+      folly::Executor* cpuExecutor = nullptr) = 0;
 
  private:
   const std::string name_;

velox/connectors/fuzzer/FuzzerConnector.h

@@ -141,8 +141,9 @@ class FuzzerConnectorFactory : public ConnectorFactory {
   std::shared_ptr<Connector> newConnector(
       const std::string& id,
       std::shared_ptr<const config::ConfigBase> config,
-      folly::Executor* executor = nullptr) override {
-    return std::make_shared<FuzzerConnector>(id, config, executor);
+      folly::Executor* ioExecutor = nullptr,
+      folly::Executor* cpuExecutor = nullptr) override {
+    return std::make_shared<FuzzerConnector>(id, config, ioExecutor);
   }
 };
 

velox/connectors/hive/HiveConnector.h

@@ -93,8 +93,9 @@ class HiveConnectorFactory : public ConnectorFactory {
   std::shared_ptr<Connector> newConnector(
       const std::string& id,
       std::shared_ptr<const config::ConfigBase> config,
-      folly::Executor* executor = nullptr) override {
-    return std::make_shared<HiveConnector>(id, config, executor);
+      folly::Executor* ioExecutor = nullptr,
+      folly::Executor* cpuExecutor = nullptr) override {
+    return std::make_shared<HiveConnector>(id, config, ioExecutor);
   }
 };
 

velox/connectors/tests/ConnectorTest.cpp

@@ -59,7 +59,8 @@ class TestConnectorFactory : public connector::ConnectorFactory {
   std::shared_ptr<Connector> newConnector(
       const std::string& id,
       std::shared_ptr<const config::ConfigBase> /*config*/,
-      folly::Executor* /*executor*/ = nullptr) override {
+      folly::Executor* /*ioExecutor*/ = nullptr,
+      folly::Executor* /*cpuExecutor*/ = nullptr) override {
     return std::make_shared<TestConnector>(id);
   }
 };

velox/connectors/tpch/TpchConnector.h

@@ -171,8 +171,9 @@ class TpchConnectorFactory : public ConnectorFactory {
   std::shared_ptr<Connector> newConnector(
       const std::string& id,
       std::shared_ptr<const config::ConfigBase> config,
-      folly::Executor* executor = nullptr) override {
-    return std::make_shared<TpchConnector>(id, config, executor);
+      folly::Executor* ioExecutor = nullptr,
+      folly::Executor* cpuExecutor = nullptr) override {
+    return std::make_shared<TpchConnector>(id, config, ioExecutor);
   }
 };
 

velox/exec/tests/AsyncConnectorTest.cpp

@@ -164,7 +164,8 @@ class TestConnectorFactory : public connector::ConnectorFactory {
   std::shared_ptr<connector::Connector> newConnector(
       const std::string& id,
       std::shared_ptr<const config::ConfigBase> config,
-      folly::Executor* /* executor */) override {
+      folly::Executor* /* ioExecutor */,
+      folly::Executor* /* cpuExecutor */) override {
     return std::make_shared<TestConnector>(id);
   }
 };