[sv-cosco]perf: Support fetch multiple blocks based on max bytes

Differential Revision: D85830101

Author: xiaoxmeng

presto-native-execution/presto_cpp/main/operators/ShuffleExchangeSource.cpp

@@ -31,11 +31,12 @@ namespace facebook::presto::operators {
 
 folly::SemiFuture<ShuffleExchangeSource::Response>
 ShuffleExchangeSource::request(
-    uint32_t /*maxBytes*/,
+    uint32_t maxBytes,
     std::chrono::microseconds /*maxWait*/) {
-  auto nextBatch = [this]() {
-    return std::move(shuffleReader_->next(1))
-        .deferValue([this](std::vector<std::unique_ptr<ReadBatch>>&& batches) {
+  auto nextBatch = [this, maxBytes]() {
+    return std::move(shuffleReader_->next(maxBytes))
+        .deferValue([this, maxBytes](std::vector<std::unique_ptr<ReadBatch>>&& batches) {
+          LOG(ERROR) << "batch size: " << batches.size() << " maxBytes: " << velox::succinctBytes(maxBytes);
           std::vector<velox::ContinuePromise> promises;
           int64_t totalBytes{0};
           {

presto-native-execution/presto_cpp/main/operators/ShuffleInterface.h

@@ -56,8 +56,12 @@ class ShuffleReader {
  public:
   virtual ~ShuffleReader() = default;
 
+  /// Fetch the next batch of rows from the shuffle reader.
+  /// @param bytes Maximum number of bytes to read in this batch.
+  /// @return A semi-future resolving to a vector of ReadBatch pointers, where
+  /// each ReadBatch contains rows and associated data buffers.
   virtual folly::SemiFuture<std::vector<std::unique_ptr<ReadBatch>>> next(
-      size_t numBatches) = 0;
+       uint64_t maxBytes) = 0;
 
   /// Tell the shuffle system the reader is done. May be called with 'success'
   /// true before reading all the data. This happens when a query has a LIMIT or