refactor: Add arg cpuExecutor to newConnector interface of ConnectorF…

…actory 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.
facebookincubator · Dec 14, 2024 · eee9a8e · eee9a8e
1 parent 19c5771
commit eee9a8e
Show file tree

Hide file tree

Showing 6 changed files with 15 additions and 9 deletions.
diff --git a/velox/connectors/Connector.h b/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_;

diff --git a/velox/connectors/fuzzer/FuzzerConnector.h b/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);
   }
 };
 

diff --git a/velox/connectors/hive/HiveConnector.h b/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);
   }
 };
 

diff --git a/velox/connectors/tests/ConnectorTest.cpp b/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);
   }
 };

diff --git a/velox/connectors/tpch/TpchConnector.h b/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);
   }
 };
 

diff --git a/velox/exec/tests/AsyncConnectorTest.cpp b/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);
   }
 };