[14722] Fix memory leak caused by asynchronous prefetch

Alchemy-item: (ID = 901) [OAP] [14722] Fix memory leak caused by asynchronous prefetch commit 1/1 - 075e790

Author: rui-mo

velox/common/caching/AsyncDataCache.h

@@ -445,7 +445,7 @@ class CoalescedLoad {
     return state_;
   }
 
-  void cancel() {
+  virtual void cancel() {
     setEndState(State::kCancelled);
   }
 

velox/dwio/common/DirectBufferedInput.h

@@ -102,6 +102,17 @@ class DirectCoalescedLoad : public cache::CoalescedLoad {
     return size;
   }
 
+  void cancel() override {
+    folly::SemiFuture<bool> waitFuture(false);
+    if (state() == State::kLoading && !loadOrFuture(&waitFuture)) {
+      waitFuture.wait();
+    }
+    for (auto& request : requests_) {
+      pool_->freeNonContiguous(request.data);
+    }
+    CoalescedLoad::cancel();
+  }
+
  private:
   const std::shared_ptr<IoStatistics> ioStats_;
   const std::shared_ptr<filesystems::File::IoStats> fsStats_;