Back out "perf: Do not request data size if there is only a single source as in persistent shuffle" (#15406)

Summary:
Pull Request resolved: #15406

Original commit changeset: 4424ea18f7d4

Original Phabricator Diff: D85647156

Reviewed By: amitkdutta, xiaoxmeng

Differential Revision: D86272441

fbshipit-source-id: 01aa02cf6e4354e69732f552e0bba45851eb8950

Author: Ke Wang

velox/exec/ExchangeClient.cpp

@@ -185,7 +185,7 @@ void ExchangeClient::request(std::vector<RequestSpec>&& requestSpecs) {
                 RECORD_METRIC_VALUE(kMetricExchangeDataCount);
               }
 
-              bool pauseCurrentSource{false};
+              bool pauseCurrentSource = false;
               std::vector<RequestSpec> requestSpecs;
               std::shared_ptr<ExchangeSource> currentSource = spec.source;
               {
@@ -232,9 +232,6 @@ ExchangeClient::pickSourcesToRequestLocked() {
   if (closed_) {
     return {};
   }
-  if (sources_.size() == 1) {
-    return pickupSingleSourceToRequestLocked();
-  }
   std::vector<RequestSpec> requestSpecs;
   while (!emptySources_.empty()) {
     auto& source = emptySources_.front();
@@ -286,42 +283,6 @@ ExchangeClient::pickSourcesToRequestLocked() {
   return requestSpecs;
 }
 
-std::vector<ExchangeClient::RequestSpec>
-ExchangeClient::pickupSingleSourceToRequestLocked() {
-  VELOX_CHECK_EQ(sources_.size(), 1);
-  VELOX_CHECK(!closed_);
-  if (emptySources_.empty() && producingSources_.empty()) {
-    return {};
-  }
-
-  VELOX_CHECK_EQ(totalPendingBytes_, 0);
-  VELOX_CHECK_LE(!!emptySources_.empty() + !!producingSources_.empty(), 1);
-  const auto requestBytes = maxQueuedBytes_ - queue_->totalBytes();
-  if (requestBytes <= 0) {
-    return {};
-  }
-  std::vector<RequestSpec> requestSpecs;
-  SCOPE_EXIT {
-    totalPendingBytes_ += requestBytes;
-  };
-  if (!emptySources_.empty()) {
-    VELOX_CHECK_EQ(emptySources_.size(), 1);
-    auto& source = emptySources_.front();
-    VELOX_CHECK(source->shouldRequestLocked());
-    requestSpecs.push_back({std::move(source), requestBytes});
-    emptySources_.pop();
-    return requestSpecs;
-  }
-
-  VELOX_CHECK_EQ(producingSources_.size(), 1);
-  auto& source = producingSources_.front().source;
-  VELOX_CHECK(source->shouldRequestLocked());
-  VELOX_CHECK(!producingSources_.front().remainingBytes.empty());
-  requestSpecs.push_back({std::move(source), requestBytes});
-  producingSources_.pop();
-  return requestSpecs;
-}
-
 ExchangeClient::~ExchangeClient() {
   close();
 }

velox/exec/ExchangeClient.h

@@ -132,27 +132,8 @@ class ExchangeClient : public std::enable_shared_from_this<ExchangeClient> {
     std::vector<int64_t> remainingBytes;
   };
 
-  // Selects exchange sources to request data from based on available queue
-  // capacity. Handles multiple sources by first requesting data sizes from all
-  // empty sources, then requesting actual data from producing sources based on
-  // their remaining bytes and available capacity. May initiate out-of-band
-  // transfers for large pages that exceed capacity to avoid deadlock
-  // situations. For single source case, delegates to
-  // pickupSingleSourceToRequestLocked which sets max request bytes based on
-  // available queue space instead of reported remaining bytes from exchange
-  // sources.
   std::vector<RequestSpec> pickSourcesToRequestLocked();
 
-  // Specialized single-source request picker for single-source exchange
-  // clients. Sets the max request bytes based on available space in the queue
-  // rather than the reported remaining bytes from exchange sources. The reason
-  // is that single source has no other alternative so just fetch as much as
-  // possible from that source. Returns a request spec for the single source
-  // when there is available capacity in the queue and no pending requests. If
-  // capacity is unavailable or requests are already pending, returns empty
-  // vector.
-  std::vector<RequestSpec> pickupSingleSourceToRequestLocked();
-
   void request(std::vector<RequestSpec>&& requestSpecs);
 
   // Handy for ad-hoc logging.

velox/exec/tests/ExchangeClientTest.cpp

@@ -589,7 +589,7 @@ TEST_P(ExchangeClientTest, acknowledge) {
   SCOPED_TESTVALUE_SET(
       "facebook::velox::exec::test::LocalExchangeSource::pause",
       std::function<void(void*)>(([&numberOfAcknowledgeRequests](void*) {
-        ++numberOfAcknowledgeRequests;
+        numberOfAcknowledgeRequests++;
       })));
 
   {
@@ -665,7 +665,7 @@ TEST_P(ExchangeClientTest, acknowledge) {
     int attempts = 100;
     bool outputBuffersEmpty;
     while (attempts > 0) {
-      --attempts;
+      attempts--;
       outputBuffersEmpty = bufferManager_->getUtilization(sourceTaskId) == 0;
       if (outputBuffersEmpty) {
         break;
@@ -677,7 +677,7 @@ TEST_P(ExchangeClientTest, acknowledge) {
     // The output buffer is empty now
     // Explicit acknowledge is not necessary as a blocking getDataSize is sent
     // right away
-    EXPECT_EQ(numberOfAcknowledgeRequests, 3);
+    EXPECT_EQ(numberOfAcknowledgeRequests, 2);
 #endif
   }
 
@@ -985,10 +985,7 @@ VELOX_INSTANTIATE_TEST_SUITE_P(
     testing::Values(
         VectorSerde::Kind::kPresto,
         VectorSerde::Kind::kCompactRow,
-        VectorSerde::Kind::kUnsafeRow),
-    [](const testing::TestParamInfo<VectorSerde::Kind>& info) {
-      return fmt::format("{}", info.param);
-    });
+        VectorSerde::Kind::kUnsafeRow));
 
 } // namespace
 } // namespace facebook::velox::exec