[WIP] parallelize connector reclaim

velox/exec/TableWriter.cpp

@@ -340,7 +340,7 @@ uint64_t TableWriter::ConnectorReclaimer::reclaim(
     return 0;
   }
   RuntimeStatWriterScopeGuard opStatsGuard(op_);
-  return memory::MemoryReclaimer::reclaim(pool, targetBytes, maxWaitMs, stats);
+  return ParallelMemoryReclaimer::reclaim(pool, targetBytes, maxWaitMs, stats);
 }
 
 // static

velox/exec/TableWriter.h

@@ -97,9 +97,6 @@ class TableWriteTraits {
       velox::memory::MemoryPool* pool);
 };
 
-/**
- * Implements a simple table writer operator.
- */
 class TableWriter : public Operator {
  public:
   TableWriter(
@@ -149,8 +146,12 @@ class TableWriter : public Operator {
   // The memory reclaimer customized for connector which interface with the
   // memory arbitrator to reclaim memory from the file writers created within
   // the connector.
-  class ConnectorReclaimer : public Operator::MemoryReclaimer {
+  class ConnectorReclaimer : public exec::ParallelMemoryReclaimer {
    public:
+    static inline std::shared_ptr<folly::CPUThreadPoolExecutor>
+        connectorExecutor = std::make_shared<folly::CPUThreadPoolExecutor>(
+            std::thread::hardware_concurrency() * 2,
+            std::make_shared<folly::NamedThreadFactory>("ConnectorExecutor"));
     static std::unique_ptr<memory::MemoryReclaimer>
     create(DriverCtx* driverCtx, Operator* op, bool canReclaim);
 
@@ -171,14 +172,23 @@ class TableWriter : public Operator {
     void abort(memory::MemoryPool* pool, const std::exception_ptr& /* error */)
         override {}
 
+    std::shared_ptr<Driver> ensureDriver() const {
+      return driver_.lock();
+    }
+
    private:
     ConnectorReclaimer(
         const std::shared_ptr<Driver>& driver,
         Operator* op,
         bool canReclaim)
-        : Operator::MemoryReclaimer(driver, op), canReclaim_(canReclaim) {}
+        : ParallelMemoryReclaimer(connectorExecutor.get()),
+          canReclaim_(canReclaim),
+          driver_(driver),
+          op_(op) {}
 
     const bool canReclaim_{false};
+    std::weak_ptr<Driver> driver_;
+    Operator* op_;
   };
 
   void createDataSink();