diff --git a/velox/experimental/cudf/CudfConfig.h b/velox/experimental/cudf/CudfConfig.h
index c52f61960da..25e4d62d6ba 100644
--- a/velox/experimental/cudf/CudfConfig.h
+++ b/velox/experimental/cudf/CudfConfig.h
@@ -53,6 +53,8 @@ struct CudfConfig {
       "cudf.batch_size_min_threshold_bytes"};
   static constexpr const char* kCudfBatchSizeMaxThreshold{
       "cudf.batch_size_max_threshold"};
+  static constexpr const char* kCudfBatchConcatMaxBytes{
+      "cudf.batch_concat_max_bytes"};
   static constexpr const char* kCudfConcatOptimizationEnabled{
       "cudf.concat_optimization_enabled"};
   static constexpr const char* kCudfGroupbyStreamingMaxDistinctKeys{
@@ -230,6 +232,10 @@ struct CudfConfig {
   /// This field is intentionally appended so incremental builds that reuse
   /// stable UCX objects preserve the offsets of every pre-existing field.
   bool exchangeConcatOptimizationEnabled{true};
+
+  /// Hard upper bound on the estimated input bytes passed to one
+  /// cudf::concatenate call by CudfBatchConcat.
+  uint64_t batchConcatMaxBytes{1ULL << 30};
 };
 
 } // namespace facebook::velox::cudf_velox
diff --git a/velox/experimental/cudf/exec/CudfBatchConcat.cpp b/velox/experimental/cudf/exec/CudfBatchConcat.cpp
index 6bffc3a05fb..eaef2de3da6 100644
--- a/velox/experimental/cudf/exec/CudfBatchConcat.cpp
+++ b/velox/experimental/cudf/exec/CudfBatchConcat.cpp
@@ -20,6 +20,7 @@
 #include "velox/experimental/cudf/exec/GpuResources.h"
 #include "velox/experimental/cudf/exec/Utilities.h"
 
+#include <algorithm>
 #include <cstdlib>
 #include <limits>
 #include <string_view>
@@ -79,6 +80,16 @@ int32_t queryConcatTargetRows(exec::DriverCtx* driverCtx) {
   return static_cast<int32_t>(targetRows);
 }
 
+uint64_t queryConcatMaxBytes(exec::DriverCtx* driverCtx) {
+  const auto maxBytes = concatThreshold(
+      driverCtx,
+      CudfConfig::kCudfBatchConcatMaxBytes,
+      "GLUTEN_CUDF_BATCH_CONCAT_MAX_BYTES",
+      CudfConfig::getInstance().batchConcatMaxBytes);
+  VELOX_CHECK_GT(maxBytes, 0, "CudfBatchConcat hard byte cap must be positive");
+  return maxBytes;
+}
+
 std::string getAggregationStep(
     const std::shared_ptr<const core::PlanNode>& planNode) {
   const auto aggregation =
@@ -141,6 +152,23 @@ CudfBatchConcat::CudfBatchConcat(
     RowTypePtr outputType,
     int32_t targetRows,
     uint64_t targetBytes)
+    : CudfBatchConcat(
+          operatorId,
+          driverCtx,
+          std::move(planNode),
+          std::move(outputType),
+          targetRows,
+          targetBytes,
+          queryConcatMaxBytes(driverCtx)) {}
+
+CudfBatchConcat::CudfBatchConcat(
+    int32_t operatorId,
+    exec::DriverCtx* driverCtx,
+    std::shared_ptr<const core::PlanNode> planNode,
+    RowTypePtr outputType,
+    int32_t targetRows,
+    uint64_t targetBytes,
+    uint64_t maxConcatBytes)
     : CudfOperatorBase(
           operatorId,
           driverCtx,
@@ -154,17 +182,68 @@ CudfBatchConcat::CudfBatchConcat(
       driverCtx_(driverCtx),
       aggregationStep_(getAggregationStep(planNode)),
       targetRows_(checkedConcatTargetRows(targetRows)),
-      targetBytes_(targetBytes) {
+      targetBytes_(targetBytes),
+      maxConcatBytes_(maxConcatBytes) {
+  VELOX_CHECK_GT(
+      maxConcatBytes_, 0, "CudfBatchConcat hard byte cap must be positive");
   if (logConcatConfig()) {
     LOG(WARNING) << "CudfBatchConcat configured targetRows=" << targetRows_
-                 << ", targetBytes=" << targetBytes_;
+                 << ", targetBytes=" << targetBytes_
+                 << ", maxConcatBytes=" << maxConcatBytes_;
   }
 }
 
 bool CudfBatchConcat::needsInput() const {
   return !noMoreInput_ && outputQueue_.empty() &&
       currentNumRows_ < targetRows_ &&
-      (targetBytes_ == 0 || currentNumBytes_ < targetBytes_);
+      (targetBytes_ == 0 || currentNumBytes_ < targetBytes_) &&
+      currentNumBytes_ < maxConcatBytes_;
+}
+
+void CudfBatchConcat::flushBufferedInputs() {
+  if (buffer_.empty()) {
+    return;
+  }
+
+  if (buffer_.size() == 1) {
+    outputQueue_.push(std::move(buffer_.front()));
+    buffer_.clear();
+  } else {
+    const auto outputStream = buffer_.front()->stream();
+    const auto bufferedInputs = buffer_.size();
+    ConcatenateBatchStats concatStats;
+    std::vector<CudfVectorPtr> outputVectors;
+    try {
+      outputVectors = getConcatenatedCudfVectorsBatched(
+          pool(),
+          std::exchange(buffer_, {}),
+          outputType_,
+          outputStream,
+          get_output_mr(),
+          maxConcatBytes_,
+          &concatStats);
+    } catch (...) {
+      concatenateCalls_ += concatStats.concatenateCalls;
+      maxConcatenateInputBytes_ =
+          std::max(maxConcatenateInputBytes_, concatStats.maxInputBytes);
+      publishRuntimeStats();
+      LOG(ERROR) << "CudfBatchConcat concatenate failed planNode="
+                 << planNodeId() << ", bufferedInputs=" << bufferedInputs
+                 << ", bufferedBytes=" << currentNumBytes_
+                 << ", maxConcatBytes=" << maxConcatBytes_
+                 << ", outputStream=" << outputStream.value();
+      throw;
+    }
+    concatenateCalls_ += concatStats.concatenateCalls;
+    maxConcatenateInputBytes_ =
+        std::max(maxConcatenateInputBytes_, concatStats.maxInputBytes);
+    publishRuntimeStats();
+    for (auto& output : outputVectors) {
+      outputQueue_.push(std::move(output));
+    }
+  }
+  currentNumRows_ = 0;
+  currentNumBytes_ = 0;
 }
 
 void CudfBatchConcat::doAddInput(RowVectorPtr input) {
@@ -175,38 +254,50 @@ void CudfBatchConcat::doAddInput(RowVectorPtr input) {
     return;
   }
 
-  // Push input cudf table to buffer
+  const auto inputRows = static_cast<size_t>(cudfVector->size());
+  const auto inputBytes = cudfVector->estimateFlatSize();
   ++inputBatches_;
-  totalInputRows_ += cudfVector->size();
-  totalInputBytes_ += cudfVector->estimateFlatSize();
-  currentNumRows_ += cudfVector->size();
-  currentNumBytes_ += cudfVector->estimateFlatSize();
+  VELOX_CHECK_LE(
+      inputRows,
+      std::numeric_limits<uint64_t>::max() - totalInputRows_,
+      "CudfBatchConcat total input row count overflow");
+  totalInputRows_ += inputRows;
+  VELOX_CHECK_LE(
+      inputBytes,
+      std::numeric_limits<uint64_t>::max() - totalInputBytes_,
+      "CudfBatchConcat total input byte count overflow");
+  totalInputBytes_ += inputBytes;
+
+  if (inputBytes > maxConcatBytes_) {
+    if (!buffer_.empty()) {
+      ++hardByteCapFlushes_;
+      flushBufferedInputs();
+    }
+    ++oversizedInputPassthroughs_;
+    outputQueue_.push(std::move(cudfVector));
+    return;
+  }
+
+  if (!buffer_.empty() && inputBytes > maxConcatBytes_ - currentNumBytes_) {
+    ++hardByteCapFlushes_;
+    flushBufferedInputs();
+  }
+
+  VELOX_CHECK_LE(
+      inputRows,
+      std::numeric_limits<size_t>::max() - currentNumRows_,
+      "CudfBatchConcat buffered row count overflow");
+  currentNumRows_ += inputRows;
+  currentNumBytes_ += inputBytes;
   buffer_.push_back(std::move(cudfVector));
 
-  // Enforce the bound here as well as in needsInput(). Source pipelines may
-  // have already scheduled another input before the driver observes the
-  // updated needsInput() result. Keeping a ready output in outputQueue_ makes
-  // the backpressure explicit and prevents scan batches from accumulating up
-  // to device capacity. Avoid a redundant D2D concatenate for one large input.
   if (currentNumRows_ >= targetRows_ ||
-      (targetBytes_ != 0 && currentNumBytes_ >= targetBytes_)) {
-    if (buffer_.size() == 1) {
-      outputQueue_.push(std::move(buffer_.front()));
-      buffer_.clear();
-    } else {
-      const auto outputStream = buffer_.front()->stream();
-      auto outputVectors = getConcatenatedCudfVectorsBatched(
-          pool(),
-          std::exchange(buffer_, {}),
-          outputType_,
-          outputStream,
-          get_output_mr());
-      for (auto& output : outputVectors) {
-        outputQueue_.push(std::move(output));
-      }
+      (targetBytes_ != 0 && currentNumBytes_ >= targetBytes_) ||
+      currentNumBytes_ >= maxConcatBytes_) {
+    if (currentNumBytes_ >= maxConcatBytes_) {
+      ++hardByteCapFlushes_;
     }
-    currentNumRows_ = 0;
-    currentNumBytes_ = 0;
+    flushBufferedInputs();
   }
 }
 
@@ -223,50 +314,8 @@ RowVectorPtr CudfBatchConcat::doGetOutput() {
   if (!buffer_.empty() &&
       (currentNumRows_ >= targetRows_ ||
        (targetBytes_ != 0 && currentNumBytes_ >= targetBytes_) ||
-       noMoreInput_)) {
-    // Preserve the zero-copy Exchange fast path when a single received batch
-    // already satisfies the target (or is the final tail batch). CudfVector
-    // carries its producing stream, so downstream operators can consume it
-    // directly without rebinding allocation ownership to another stream.
-    if (buffer_.size() == 1) {
-      auto output = std::move(buffer_.front());
-      buffer_.clear();
-      currentNumRows_ = 0;
-      currentNumBytes_ = 0;
-      ++outputBatches_;
-      return output;
-    }
-    // Use stream from existing buffer vectors
-    const auto outputStream = buffer_[0]->stream();
-    auto outputVectors = getConcatenatedCudfVectorsBatched(
-        pool(),
-        std::exchange(buffer_, {}),
-        outputType_,
-        outputStream,
-        get_output_mr());
-
-    currentNumRows_ = 0;
-    currentNumBytes_ = 0;
-    VELOX_CHECK_GT(outputVectors.size(), 0);
-
-    for (auto it = outputVectors.begin(); it + 1 != outputVectors.end(); ++it) {
-      outputQueue_.push(std::move(*it));
-    }
-
-    // If last table is a smaller batch and we still expect more input and keep
-    // it in buffer.
-    auto& last = outputVectors.back();
-    auto rowCount = last->size();
-
-    const auto lastBytes = last->estimateFlatSize();
-    if (!noMoreInput_ && rowCount < targetRows_ &&
-        (targetBytes_ == 0 || lastBytes < targetBytes_)) {
-      currentNumRows_ = rowCount;
-      currentNumBytes_ = lastBytes;
-      buffer_.push_back(std::move(last));
-    } else {
-      outputQueue_.push(std::move(last));
-    }
+       currentNumBytes_ >= maxConcatBytes_ || noMoreInput_)) {
+    flushBufferedInputs();
 
     // Return the first batch from the new queue
     if (!outputQueue_.empty()) {
@@ -280,16 +329,42 @@ RowVectorPtr CudfBatchConcat::doGetOutput() {
   return nullptr;
 }
 
+void CudfBatchConcat::publishRuntimeStats() {
+  auto lockedStats = stats_.wlock();
+  lockedStats->setRuntimeStat(
+      "concatenateCalls", RuntimeMetric(saturateCast(concatenateCalls_)));
+  lockedStats->setRuntimeStat(
+      "hardByteCapFlushes", RuntimeMetric(saturateCast(hardByteCapFlushes_)));
+  lockedStats->setRuntimeStat(
+      "oversizedInputPassthroughs",
+      RuntimeMetric(saturateCast(oversizedInputPassthroughs_)));
+  lockedStats->setRuntimeStat(
+      "maxConcatenateInputBytes",
+      RuntimeMetric(
+          saturateCast(maxConcatenateInputBytes_),
+          RuntimeCounter::Unit::kBytes));
+}
+
 bool CudfBatchConcat::isFinished() {
   const bool finished = noMoreInput_ && buffer_.empty() && outputQueue_.empty();
-  if (finished && !summaryLogged_ && logConcatConfig()) {
+  if (finished && !summaryLogged_) {
     summaryLogged_ = true;
-    LOG(WARNING) << "CudfBatchConcat summary planNode=" << planNodeId()
-                 << ", step=" << aggregationStep_
-                 << ", inputBatches=" << inputBatches_
-                 << ", outputBatches=" << outputBatches_
-                 << ", inputRows=" << totalInputRows_
-                 << ", inputBytes=" << totalInputBytes_;
+    publishRuntimeStats();
+    if (logConcatConfig()) {
+      LOG(WARNING) << "CudfBatchConcat summary planNode=" << planNodeId()
+                   << ", step=" << aggregationStep_
+                   << ", inputBatches=" << inputBatches_
+                   << ", outputBatches=" << outputBatches_
+                   << ", inputRows=" << totalInputRows_
+                   << ", inputBytes=" << totalInputBytes_
+                   << ", maxConcatBytes=" << maxConcatBytes_
+                   << ", hardByteCapFlushes=" << hardByteCapFlushes_
+                   << ", oversizedInputPassthroughs="
+                   << oversizedInputPassthroughs_
+                   << ", concatenateCalls=" << concatenateCalls_
+                   << ", maxConcatenateInputBytes="
+                   << maxConcatenateInputBytes_;
+    }
   }
   return finished;
 }
diff --git a/velox/experimental/cudf/exec/CudfBatchConcat.h b/velox/experimental/cudf/exec/CudfBatchConcat.h
index eff9aa2f4c7..5aaaf52d622 100644
--- a/velox/experimental/cudf/exec/CudfBatchConcat.h
+++ b/velox/experimental/cudf/exec/CudfBatchConcat.h
@@ -55,6 +55,15 @@ class CudfBatchConcat : public CudfOperatorBase {
       int32_t targetRows,
       uint64_t targetBytes);
 
+  CudfBatchConcat(
+      int32_t operatorId,
+      exec::DriverCtx* driverCtx,
+      std::shared_ptr<const core::PlanNode> planNode,
+      RowTypePtr outputType,
+      int32_t targetRows,
+      uint64_t targetBytes,
+      uint64_t maxConcatBytes);
+
   bool needsInput() const override;
 
   exec::BlockingReason isBlocked(ContinueFuture* /*future*/) override {
@@ -68,6 +77,9 @@ class CudfBatchConcat : public CudfOperatorBase {
   RowVectorPtr doGetOutput() override;
 
  private:
+  void flushBufferedInputs();
+  void publishRuntimeStats();
+
   exec::DriverCtx* const driverCtx_;
   const std::string aggregationStep_;
   std::vector<CudfVectorPtr> buffer_;
@@ -80,6 +92,11 @@ class CudfBatchConcat : public CudfOperatorBase {
   uint64_t currentNumBytes_{0};
   const size_t targetRows_{0};
   const uint64_t targetBytes_{0};
+  const uint64_t maxConcatBytes_{0};
+  uint64_t hardByteCapFlushes_{0};
+  uint64_t oversizedInputPassthroughs_{0};
+  uint64_t concatenateCalls_{0};
+  uint64_t maxConcatenateInputBytes_{0};
   bool summaryLogged_{false};
 };
 
diff --git a/velox/experimental/cudf/exec/CudfGroupby.cpp b/velox/experimental/cudf/exec/CudfGroupby.cpp
index 46d2502d97a..ac3a85d190e 100644
--- a/velox/experimental/cudf/exec/CudfGroupby.cpp
+++ b/velox/experimental/cudf/exec/CudfGroupby.cpp
@@ -91,6 +91,12 @@ constexpr const char* kIntermediateAggregationMaxLevel =
     "cudfIntermediateAggregationMaxLevel";
 constexpr const char* kIntermediateAggregationSerializedMerges =
     "cudfIntermediateAggregationSerializedMerges";
+constexpr const char* kIntermediateAggregationMemoryLimitFlushes =
+    "cudfIntermediateAggregationMemoryLimitFlushes";
+constexpr const char* kIntermediateAggregationMemoryLimitFlushBytes =
+    "cudfIntermediateAggregationMemoryLimitFlushBytes";
+constexpr const char* kIntermediateAggregationBufferedBytes =
+    "cudfIntermediateAggregationBufferedBytes";
 
 bool serializeLargeIntermediateAggregationMergesEnabled() {
   static const bool enabled = [] {
@@ -1117,6 +1123,16 @@ CudfGroupby::CudfGroupby(
       maxPartialAggregationMemoryUsage_(
           driverCtx->queryConfig().maxPartialAggregationMemoryUsage()) {}
 
+bool CudfGroupby::needsInput() const {
+  if (noMoreInput_) {
+    return false;
+  }
+  if (!isPartialOutput_ || !streamingEnabled_) {
+    return true;
+  }
+  return bufferedResult_ == nullptr && !partialAggregationMemoryLimitReached();
+}
+
 void CudfGroupby::initialize() {
   Operator::initialize();
 
@@ -1696,6 +1712,7 @@ void CudfGroupby::addIntermediateAggregationRun(
   ++intermediateInputRunCount_;
 
   auto level = aggregationRunLevel(run.representedRows);
+  const auto runBytes = run.data->estimateFlatSize();
   {
     auto lockedStats = stats_.wlock();
     lockedStats->addRuntimeStat(
@@ -1705,13 +1722,34 @@ void CudfGroupby::addIntermediateAggregationRun(
         RuntimeCounter(static_cast<int64_t>(level)));
   }
 
+  if (isPartialOutput_ && maxPartialAggregationMemoryUsage_ > 0 &&
+      intermediateBufferedBytes_ > 0) {
+    const auto memoryLimit =
+        static_cast<uint64_t>(maxPartialAggregationMemoryUsage_);
+    if (runBytes > memoryLimit ||
+        intermediateBufferedBytes_ > memoryLimit - runBytes) {
+      bufferIntermediateAggregationRunsForMemoryLimit(run.representedRows);
+    }
+  }
+
   for (;;) {
     if (intermediateRunLevels_.size() <= level) {
       intermediateRunLevels_.resize(level + 1);
     }
     if (!intermediateRunLevels_[level].has_value()) {
-      intermediateBufferedBytes_ += run.data->estimateFlatSize();
+      const auto bufferedRunBytes = run.data->estimateFlatSize();
+      VELOX_CHECK_LE(
+          intermediateBufferedBytes_,
+          std::numeric_limits<uint64_t>::max() - bufferedRunBytes);
+      intermediateBufferedBytes_ += bufferedRunBytes;
       intermediateRunLevels_[level] = std::move(run);
+      VELOX_CHECK_LE(
+          intermediateBufferedBytes_,
+          static_cast<uint64_t>(std::numeric_limits<int64_t>::max()));
+      auto lockedStats = stats_.wlock();
+      lockedStats->addRuntimeStat(
+          kIntermediateAggregationBufferedBytes,
+          RuntimeCounter(static_cast<int64_t>(intermediateBufferedBytes_)));
       return;
     }
 
@@ -1731,6 +1769,36 @@ void CudfGroupby::addIntermediateAggregationRun(
   }
 }
 
+bool CudfGroupby::partialAggregationMemoryLimitReached() const {
+  return maxPartialAggregationMemoryUsage_ > 0 &&
+      intermediateBufferedBytes_ >=
+      static_cast<uint64_t>(maxPartialAggregationMemoryUsage_);
+}
+
+void CudfGroupby::bufferIntermediateAggregationRunsForMemoryLimit(
+    uint64_t retainedInputRows) {
+  VELOX_CHECK(isPartialOutput_);
+  VELOX_CHECK(streamingEnabled_);
+  VELOX_CHECK_NULL(bufferedResult_);
+  VELOX_CHECK_GT(intermediateBufferedBytes_, 0);
+  VELOX_CHECK_LE(retainedInputRows, static_cast<uint64_t>(numInputRows_));
+
+  const auto flushBytes = intermediateBufferedBytes_;
+  numInputRows_ -= static_cast<int64_t>(retainedInputRows);
+  inputRowsRetainedAfterFlush_ = retainedInputRows;
+  bufferedResult_ = drainIntermediateAggregationRuns();
+  VELOX_CHECK_NOT_NULL(bufferedResult_);
+
+  VELOX_CHECK_LE(
+      flushBytes, static_cast<uint64_t>(std::numeric_limits<int64_t>::max()));
+  auto lockedStats = stats_.wlock();
+  lockedStats->addRuntimeStat(
+      kIntermediateAggregationMemoryLimitFlushes, RuntimeCounter(1));
+  lockedStats->addRuntimeStat(
+      kIntermediateAggregationMemoryLimitFlushBytes,
+      RuntimeCounter(static_cast<int64_t>(flushBytes)));
+}
+
 CudfGroupby::IntermediateAggregationRun
 CudfGroupby::mergeIntermediateAggregationRuns(
     IntermediateAggregationRun left,
@@ -1748,6 +1816,12 @@ CudfGroupby::mergeIntermediateAggregationRuns(
       left.data->estimateFlatSize() + right.data->estimateFlatSize();
   VELOX_CHECK_LE(
       inputBytes, static_cast<uint64_t>(std::numeric_limits<int64_t>::max()));
+  if (isPartialOutput_ && maxPartialAggregationMemoryUsage_ > 0) {
+    VELOX_CHECK_LE(
+        inputBytes,
+        static_cast<uint64_t>(maxPartialAggregationMemoryUsage_),
+        "CudfGroupby partial merge input exceeded its memory limit");
+  }
   const auto representedRows =
       addRepresentedRows(left.representedRows, right.representedRows);
   // The pre-rebase MPP path deliberately serialized very large merge kernels
@@ -1973,7 +2047,11 @@ CudfVectorPtr CudfGroupby::releaseAndResetBufferedResult() {
         RuntimeCounter(aggregationPct));
   }
 
-  numInputRows_ = 0;
+  VELOX_CHECK_LE(
+      inputRowsRetainedAfterFlush_,
+      static_cast<uint64_t>(std::numeric_limits<int64_t>::max()));
+  numInputRows_ = static_cast<int64_t>(inputRowsRetainedAfterFlush_);
+  inputRowsRetainedAfterFlush_ = 0;
   // We're moving bufferedResult_ to the caller because we want it to be null
   // after this call.
   return std::move(bufferedResult_);
@@ -1982,10 +2060,8 @@ CudfVectorPtr CudfGroupby::releaseAndResetBufferedResult() {
 RowVectorPtr CudfGroupby::doGetOutput() {
   // Handle partial streaming groupby.
   if (isPartialOutput_ && streamingEnabled_) {
-    if (!bufferedResult_ && maxPartialAggregationMemoryUsage_ > 0 &&
-        intermediateBufferedBytes_ >
-            static_cast<uint64_t>(maxPartialAggregationMemoryUsage_)) {
-      bufferedResult_ = drainIntermediateAggregationRuns();
+    if (!bufferedResult_ && partialAggregationMemoryLimitReached()) {
+      bufferIntermediateAggregationRunsForMemoryLimit(0);
     }
     if (bufferedResult_) {
       return releaseAndResetBufferedResult();
@@ -2100,7 +2176,7 @@ RowVectorPtr CudfGroupby::doGetOutput() {
 
 void CudfGroupby::doNoMoreInput() {
   Operator::noMoreInput();
-  if (isPartialOutput_ && inputs_.empty()) {
+  if (isPartialOutput_ && !streamingEnabled_ && inputs_.empty()) {
     finished_ = true;
   }
 }
@@ -2118,6 +2194,7 @@ void CudfGroupby::doClose() {
   finalStreamingRequestAggregationCounts_.clear();
   intermediateRunLevels_.clear();
   intermediateBufferedBytes_ = 0;
+  inputRowsRetainedAfterFlush_ = 0;
   finalRunLevels_.clear();
   bufferedResult_.reset();
   inputs_.clear();
diff --git a/velox/experimental/cudf/exec/CudfGroupby.h b/velox/experimental/cudf/exec/CudfGroupby.h
index 468e50efa68..2939d2a7094 100644
--- a/velox/experimental/cudf/exec/CudfGroupby.h
+++ b/velox/experimental/cudf/exec/CudfGroupby.h
@@ -94,9 +94,7 @@ class CudfGroupby : public CudfOperatorBase {
 
   void initialize() override;
 
-  bool needsInput() const override {
-    return !noMoreInput_;
-  }
+  bool needsInput() const override;
 
   exec::BlockingReason isBlocked(ContinueFuture* /* unused */) override {
     return exec::BlockingReason::kNotBlocked;
@@ -147,6 +145,9 @@ class CudfGroupby : public CudfOperatorBase {
   void computeSingleGroupbyStreaming(CudfVectorPtr tbl);
 
   void addIntermediateAggregationRun(IntermediateAggregationRun run);
+  bool partialAggregationMemoryLimitReached() const;
+  void bufferIntermediateAggregationRunsForMemoryLimit(
+      uint64_t retainedInputRows);
   IntermediateAggregationRun mergeIntermediateAggregationRuns(
       IntermediateAggregationRun left,
       IntermediateAggregationRun right,
@@ -207,6 +208,7 @@ class CudfGroupby : public CudfOperatorBase {
   uint64_t intermediateBufferedBytes_{0};
   uint64_t intermediateInputRunCount_{0};
   uint64_t intermediateRunMergeCount_{0};
+  uint64_t inputRowsRetainedAfterFlush_{0};
 
   // Supported FINAL aggregates use cuDF's persistent hash state directly
   // across exchange pages. The choice is made from the first page and is never
diff --git a/velox/experimental/cudf/exec/ToCudf.cpp b/velox/experimental/cudf/exec/ToCudf.cpp
index 2c33cfc2a7f..eb925c9bb3e 100644
--- a/velox/experimental/cudf/exec/ToCudf.cpp
+++ b/velox/experimental/cudf/exec/ToCudf.cpp
@@ -584,6 +584,12 @@ void CudfConfig::initialize(
     batchSizeMaxThreshold =
         folly::to<int32_t>(config[kCudfBatchSizeMaxThreshold]);
   }
+  if (config.find(kCudfBatchConcatMaxBytes) != config.end()) {
+    const auto value = folly::to<uint64_t>(config[kCudfBatchConcatMaxBytes]);
+    VELOX_USER_CHECK_GT(
+        value, 0, "{} must be positive", kCudfBatchConcatMaxBytes);
+    batchConcatMaxBytes = value;
+  }
   if (config.find(kCudfConcatOptimizationEnabled) != config.end()) {
     concatOptimizationEnabled =
         folly::to<bool>(config[kCudfConcatOptimizationEnabled]);
diff --git a/velox/experimental/cudf/exec/Utilities.cpp b/velox/experimental/cudf/exec/Utilities.cpp
index d4e89980c1a..ea35907790d 100644
--- a/velox/experimental/cudf/exec/Utilities.cpp
+++ b/velox/experimental/cudf/exec/Utilities.cpp
@@ -28,6 +28,7 @@
 
 #include <algorithm>
 #include <limits>
+#include <sstream>
 #include <vector>
 
 namespace facebook::velox::cudf_velox {
@@ -217,7 +218,9 @@ std::vector<std::unique_ptr<cudf::table>> getConcatenatedTableBatched(
     std::vector<CudfVectorPtr>&& tables,
     const TypePtr& tableType,
     rmm::cuda_stream_view stream,
-    rmm::device_async_resource_ref mr) {
+    rmm::device_async_resource_ref mr,
+    uint64_t maxConcatBytes,
+    ConcatenateBatchStats* concatStats) {
   std::vector<std::unique_ptr<cudf::table>> concatTables;
   // Check for empty vector
   if (tables.size() == 0) {
@@ -227,14 +230,22 @@ std::vector<std::unique_ptr<cudf::table>> getConcatenatedTableBatched(
 
   auto inputStreams = std::vector<rmm::cuda_stream_view>();
   auto tableViews = std::vector<cudf::table_view>();
+  auto inputBytes = std::vector<uint64_t>();
+  const bool observeInputBytes = maxConcatBytes != 0 || concatStats != nullptr;
 
   inputStreams.reserve(tables.size());
   tableViews.reserve(tables.size());
+  if (observeInputBytes) {
+    inputBytes.reserve(tables.size());
+  }
 
   for (const auto& table : tables) {
     VELOX_CHECK_NOT_NULL(table);
     tableViews.push_back(table->getTableView());
     inputStreams.push_back(table->stream());
+    if (observeInputBytes) {
+      inputBytes.push_back(table->estimateFlatSize());
+    }
   }
 
   cudf::detail::join_streams(inputStreams, stream);
@@ -244,30 +255,60 @@ std::vector<std::unique_ptr<cudf::table>> getConcatenatedTableBatched(
     auto const maxRows = maxBatchRows();
     size_t startpos = 0;
     size_t runningRows = 0;
+    uint64_t runningBytes = 0;
+    const auto flushBatch = [&](size_t endpos) {
+      VELOX_CHECK_LT(startpos, endpos);
+      VELOX_CHECK(
+          maxConcatBytes == 0 || runningBytes <= maxConcatBytes,
+          "Concatenate input estimate {} exceeds hard byte cap {}",
+          runningBytes,
+          maxConcatBytes);
+      if (concatStats != nullptr) {
+        ++concatStats->concatenateCalls;
+        concatStats->maxInputBytes =
+            std::max(concatStats->maxInputBytes, runningBytes);
+      }
+      outputTables.push_back(cudf::concatenate(
+          std::vector<cudf::table_view>(
+              tableViews.begin() + startpos, tableViews.begin() + endpos),
+          stream,
+          mr));
+    };
     for (size_t i = 0; i < tableViews.size(); ++i) {
       auto const numRows = static_cast<size_t>(tableViews[i].num_rows());
-      // If adding this table would exceed the limit, flush current batch
-      // [startpos, i).
-      if (runningRows > 0 && runningRows + numRows > maxRows) {
-        outputTables.push_back(
-            cudf::concatenate(
-                std::vector<cudf::table_view>(
-                    tableViews.begin() + startpos, tableViews.begin() + i),
-                stream,
-                mr));
+      const auto bytes = observeInputBytes ? inputBytes[i] : 0;
+      VELOX_CHECK(
+          maxConcatBytes == 0 || bytes <= maxConcatBytes,
+          "A single concatenate input estimate {} exceeds hard byte cap {}. "
+          "The caller must pass it through without concatenating.",
+          bytes,
+          maxConcatBytes);
+      const bool exceedsRows =
+          runningRows > maxRows || numRows > maxRows - runningRows;
+      const bool exceedsBytes = maxConcatBytes != 0 &&
+          (runningBytes > maxConcatBytes ||
+           bytes > maxConcatBytes - runningBytes);
+      // Flush before adding the input that would cross either hard bound.
+      if (i > startpos && (exceedsRows || exceedsBytes)) {
+        flushBatch(i);
         startpos = i;
         runningRows = 0;
+        runningBytes = 0;
       }
+      VELOX_CHECK_LE(
+          numRows,
+          std::numeric_limits<size_t>::max() - runningRows,
+          "Concatenate input row count overflow");
+      VELOX_CHECK_LE(
+          bytes,
+          std::numeric_limits<uint64_t>::max() - runningBytes,
+          "Concatenate input byte estimate overflow");
       runningRows += numRows;
+      runningBytes += bytes;
     }
     // Flush the final batch [startpos, end).
     if (startpos < tableViews.size()) {
-      outputTables.push_back(
-          cudf::concatenate(
-              std::vector<cudf::table_view>(
-                  tableViews.begin() + startpos, tableViews.end()),
-              stream,
-              mr));
+      flushBatch(tableViews.size());
     }
     orderCudfVectorDeallocationsAfterStream(tables, inputStreams, stream);
 
@@ -276,6 +317,19 @@ std::vector<std::unique_ptr<cudf::table>> getConcatenatedTableBatched(
   } catch (...) {
     // A failed later batch may leave earlier concatenate kernels in flight.
     stream.synchronize();
+    std::ostringstream streamList;
+    for (size_t i = 0; i < std::min<size_t>(inputStreams.size(), 16); ++i) {
+      if (i > 0) {
+        streamList << ",";
+      }
+      streamList << inputStreams[i].value();
+    }
+    if (inputStreams.size() > 16) {
+      streamList << ",...";
+    }
+    LOG(ERROR) << "getConcatenatedTableBatched failed outputStream="
+               << stream.value() << ", inputStreams=[" << streamList.str()
+               << "]";
     throw;
   }
 }
@@ -285,13 +339,15 @@ std::vector<CudfVectorPtr> getConcatenatedCudfVectorsBatched(
     std::vector<CudfVectorPtr>&& vectors,
     const TypePtr& tableType,
     rmm::cuda_stream_view stream,
-    rmm::device_async_resource_ref mr) {
+    rmm::device_async_resource_ref mr,
+    uint64_t maxConcatBytes,
+    ConcatenateBatchStats* concatStats) {
   VELOX_CHECK_NOT_NULL(pool);
 
   std::vector<CudfVectorPtr> outputVectors;
   if (tableType->size() > 0) {
-    auto tables =
-        getConcatenatedTableBatched(std::move(vectors), tableType, stream, mr);
+    auto tables = getConcatenatedTableBatched(
+        std::move(vectors), tableType, stream, mr, maxConcatBytes, concatStats);
     outputVectors.reserve(tables.size());
     for (auto& table : tables) {
       VELOX_CHECK_NOT_NULL(table);
diff --git a/velox/experimental/cudf/exec/Utilities.h b/velox/experimental/cudf/exec/Utilities.h
index deb834036eb..e7abe33d960 100644
--- a/velox/experimental/cudf/exec/Utilities.h
+++ b/velox/experimental/cudf/exec/Utilities.h
@@ -27,6 +27,11 @@
 
 namespace facebook::velox::cudf_velox {
 
+struct ConcatenateBatchStats {
+  uint64_t concatenateCalls{0};
+  uint64_t maxInputBytes{0};
+};
+
 // Concatenate a vector of cuDF tables into a single table
 [[nodiscard]] std::unique_ptr<cudf::table> concatenateTables(
     std::vector<std::unique_ptr<cudf::table>> tables,
@@ -62,12 +67,10 @@ namespace facebook::velox::cudf_velox {
  * @brief Concatenates multiple CUDF tables with automatic batching based on
  * size limits.
  *
- * This function concatenates a vector of CUDF tables while respecting size
- * limits imposed by cudf::size_type i.e. 32-bit signed integer. Unlike
- * getConcatenatedTable that returns a single concatenated table, this batched
- * version splits the concatenation into multiple output tables when the total
- * number of rows would exceeds ~2.1 billion, the maximum value representable by
- * cudf::size_type
+ * This function concatenates a vector of CUDF tables while respecting the row
+ * limit and, when maxConcatBytes is non-zero, an estimated-input-byte limit.
+ * A single input larger than maxConcatBytes is rejected so the caller can
+ * preserve it without a redundant device-to-device copy.
  *
  * The function is stream-safe and handles proper stream synchronization. All
  * input streams from individual tables are collected and joined on the provided
@@ -80,6 +83,9 @@ namespace facebook::velox::cudf_velox {
  * @param tableType Velox type representation for creating empty tables when
  * needed
  * @param stream CUDA stream for asynchronous operations and memory management
+ * @param maxConcatBytes Hard estimated input byte limit for each concatenate
+ * call, or zero to preserve row-only batching
+ * @param concatStats Optional concatenate-call observations
  * @return Vector of concatenated tables (multiple if input exceeded size
  * limits)
  *
@@ -89,7 +95,9 @@ getConcatenatedTableBatched(
     std::vector<CudfVectorPtr>&& tables,
     const TypePtr& tableType,
     rmm::cuda_stream_view stream,
-    rmm::device_async_resource_ref mr);
+    rmm::device_async_resource_ref mr,
+    uint64_t maxConcatBytes = 0,
+    ConcatenateBatchStats* concatStats = nullptr);
 
 /**
  * @brief Concatenates multiple CudfVectors into CudfVector output batches.
@@ -103,7 +111,9 @@ getConcatenatedTableBatched(
     std::vector<CudfVectorPtr>&& vectors,
     const TypePtr& tableType,
     rmm::cuda_stream_view stream,
-    rmm::device_async_resource_ref mr);
+    rmm::device_async_resource_ref mr,
+    uint64_t maxConcatBytes = 0,
+    ConcatenateBatchStats* concatStats = nullptr);
 
 /**
  * @brief Wrapper for CUDA events used for stream synchronization.
diff --git a/velox/experimental/cudf/tests/AggregationTest.cpp b/velox/experimental/cudf/tests/AggregationTest.cpp
index c88af75c227..1c25c78fb67 100644
--- a/velox/experimental/cudf/tests/AggregationTest.cpp
+++ b/velox/experimental/cudf/tests/AggregationTest.cpp
@@ -1000,6 +1000,14 @@ TEST_F(AggregationTest, partialAggregationMemoryLimit) {
                       .customStats.at("flushRowCount");
   EXPECT_GT(rowFlushStats.sum, 0);
   EXPECT_GT(rowFlushStats.max, 0);
+  const auto countPlanStats = toPlanStats(task->taskStats());
+  const auto& oversizedRunStats = countPlanStats.at(aggNodeId).customStats;
+  EXPECT_EQ(oversizedRunStats.count("cudfIntermediateAggregationRunMerges"), 0);
+  EXPECT_EQ(
+      oversizedRunStats.at("cudfIntermediateAggregationMemoryLimitFlushes").sum,
+      vectors.size());
+  EXPECT_GT(
+      oversizedRunStats.at("cudfIntermediateAggregationBufferedBytes").max, 1);
 
   // Global aggregation.
   task = AssertQueryBuilder(duckDbQueryRunner_)
@@ -1072,6 +1080,50 @@ TEST_F(AggregationTest, partialAggregationUsesBalancedRunMerges) {
   EXPECT_EQ(stats.count("cudfIntermediateAggregationFinalizeMerges"), 0);
 }
 
+TEST_F(AggregationTest, partialAggregationEnforcesMemoryLimitDuringInput) {
+  std::vector<RowVectorPtr> vectors;
+  constexpr int32_t kBatches = 4;
+  constexpr int32_t kRowsPerBatch = 100;
+  constexpr int64_t kMemoryLimit = 4'000;
+  for (int32_t batch = 0; batch < kBatches; ++batch) {
+    vectors.push_back(makeRowVector(
+        {makeFlatVector<int64_t>(kRowsPerBatch, [batch](vector_size_t row) {
+          return static_cast<int64_t>(batch) * kRowsPerBatch + row;
+        })}));
+  }
+
+  createDuckDbTable(vectors);
+  core::PlanNodeId partialAggId;
+  auto plan = PlanBuilder()
+                  .values(vectors)
+                  .partialAggregation({"c0"}, {"count(1)"})
+                  .capturePlanNodeId(partialAggId)
+                  .finalAggregation()
+                  .planNode();
+
+  auto task =
+      AssertQueryBuilder(duckDbQueryRunner_)
+          .maxDrivers(1)
+          .config(QueryConfig::kMaxPartialAggregationMemory, kMemoryLimit)
+          .plan(plan)
+          .assertResults("SELECT c0, count(1) FROM tmp GROUP BY c0");
+
+  const auto planStats = toPlanStats(task->taskStats());
+  const auto& partialStats = planStats.at(partialAggId);
+  EXPECT_EQ(partialStats.inputVectors, kBatches);
+  EXPECT_EQ(partialStats.inputRows, kBatches * kRowsPerBatch);
+  const auto& stats = partialStats.customStats;
+  ASSERT_EQ(stats.count("flushTimes"), 1);
+  EXPECT_EQ(stats.at("flushTimes").sum, 2);
+  ASSERT_EQ(stats.count("cudfIntermediateAggregationRunMerges"), 1);
+  EXPECT_EQ(stats.at("cudfIntermediateAggregationRunMerges").sum, 2);
+  EXPECT_EQ(stats.at("cudfIntermediateAggregationMemoryLimitFlushes").sum, 1);
+  EXPECT_LE(
+      stats.at("cudfIntermediateAggregationBufferedBytes").max, kMemoryLimit);
+  EXPECT_LE(
+      stats.at("cudfIntermediateAggregationMergeBytes").max, kMemoryLimit);
+}
+
 class FinalAggregationStreamingTest : public AggregationTest {
  protected:
   class ScopedStreamingCapacity {
diff --git a/velox/experimental/cudf/tests/BatchConcatTest.cpp b/velox/experimental/cudf/tests/BatchConcatTest.cpp
index 639033d3263..a7f149dee4f 100644
--- a/velox/experimental/cudf/tests/BatchConcatTest.cpp
+++ b/velox/experimental/cudf/tests/BatchConcatTest.cpp
@@ -15,13 +15,18 @@
  */
 
 #include "velox/experimental/cudf/CudfConfig.h"
+#include "velox/experimental/cudf/exec/GpuResources.h"
 #include "velox/experimental/cudf/exec/ToCudf.h"
+#include "velox/experimental/cudf/exec/Utilities.h"
+#include "velox/experimental/cudf/exec/VeloxCudfInterop.h"
 
 #include "velox/exec/PlanNodeStats.h"
 #include "velox/exec/tests/utils/AssertQueryBuilder.h"
 #include "velox/exec/tests/utils/OperatorTestBase.h"
 #include "velox/exec/tests/utils/PlanBuilder.h"
 
+#include <rmm/cuda_stream.hpp>
+
 using namespace facebook::velox;
 using namespace facebook::velox::exec;
 using namespace facebook::velox::exec::test;
@@ -36,7 +41,12 @@ class CudfBatchConcatTest : public OperatorTestBase {
   }
 
   void TearDown() override {
-    CudfConfig::getInstance().concatOptimizationEnabled = false;
+    auto& config = CudfConfig::getInstance();
+    config.concatOptimizationEnabled = false;
+    config.batchSizeMinThreshold = 100000;
+    config.batchSizeMaxThreshold = std::nullopt;
+    config.batchSizeMinThresholdBytes = 0;
+    config.batchConcatMaxBytes = 1ULL << 30;
     cudf_velox::unregisterCudf();
     OperatorTestBase::TearDown();
   }
@@ -44,11 +54,13 @@ class CudfBatchConcatTest : public OperatorTestBase {
   void updateCudfConfig(
       int32_t min,
       std::optional<int32_t> max,
-      uint64_t minBytes = 0) {
+      uint64_t minBytes = 0,
+      uint64_t maxConcatBytes = 1ULL << 30) {
     auto& config = CudfConfig::getInstance();
     config.batchSizeMinThreshold = min;
     config.batchSizeMaxThreshold = max;
     config.batchSizeMinThresholdBytes = minBytes;
+    config.batchConcatMaxBytes = maxConcatBytes;
   }
 
   template <typename T>
@@ -67,6 +79,30 @@ class CudfBatchConcatTest : public OperatorTestBase {
     }
     return PlanBuilder(generator).localPartitionRoundRobin(sources).planNode();
   }
+
+  PlanNodeStats runGlobalAggregation(
+      const std::vector<RowVectorPtr>& vectors,
+      const std::string& aggregate,
+      const std::string& query) {
+    createDuckDbTable(vectors);
+    auto generator = std::make_shared<core::PlanNodeIdGenerator>();
+    core::PlanNodeId aggNodeId;
+    auto plan = PlanBuilder(generator)
+                    .addNode([&](auto, auto) {
+                      return createFragmentedSource(vectors, generator);
+                    })
+                    .singleAggregation({}, {aggregate})
+                    .capturePlanNodeId(aggNodeId)
+                    .planNode();
+    auto task = AssertQueryBuilder(duckDbQueryRunner_)
+                    .plan(plan)
+                    .maxDrivers(1)
+                    .assertResults(query);
+    auto planStats = toPlanStats(task->taskStats());
+    auto concatStats =
+        std::move(planStats.at(aggNodeId).operatorStats.at("CudfBatchConcat"));
+    return std::move(*concatStats);
+  }
 };
 
 // Verifies that CudfBatchConcat is inserted before aggregation and reduces
@@ -153,6 +189,84 @@ TEST_F(CudfBatchConcatTest, concatFlushesAtByteThreshold) {
   EXPECT_LT(concatStats.outputVectors, concatStats.inputVectors);
 }
 
+TEST_F(CudfBatchConcatTest, concatFastPathStaysSingleBatchBelowHardCap) {
+  updateCudfConfig(
+      /*min=*/100000,
+      /*max=*/std::nullopt,
+      /*minBytes=*/0,
+      /*maxConcatBytes=*/500);
+  CudfConfig::getInstance().concatOptimizationEnabled = true;
+
+  std::vector<RowVectorPtr> vectors;
+  for (int i = 0; i < 6; ++i) {
+    vectors.push_back(makeRowVector({makeFlatSequence<int64_t>(i * 10, 10)}));
+  }
+  const auto concatStats =
+      runGlobalAggregation(vectors, "sum(c0)", "SELECT sum(c0) FROM tmp");
+  EXPECT_EQ(concatStats.outputVectors, 1);
+  EXPECT_EQ(concatStats.customStats.at("concatenateCalls").sum, 1);
+  EXPECT_EQ(concatStats.customStats.at("maxConcatenateInputBytes").max, 480);
+  EXPECT_EQ(concatStats.customStats.at("hardByteCapFlushes").sum, 0);
+}
+
+TEST_F(CudfBatchConcatTest, concatFlushesBeforeNextInputExceedsHardCap) {
+  updateCudfConfig(
+      /*min=*/100000,
+      /*max=*/std::nullopt,
+      /*minBytes=*/0,
+      /*maxConcatBytes=*/200);
+  CudfConfig::getInstance().concatOptimizationEnabled = true;
+
+  std::vector<RowVectorPtr> vectors;
+  for (int i = 0; i < 6; ++i) {
+    vectors.push_back(makeRowVector({makeFlatSequence<int64_t>(i * 10, 10)}));
+  }
+  const auto concatStats =
+      runGlobalAggregation(vectors, "sum(c0)", "SELECT sum(c0) FROM tmp");
+  EXPECT_EQ(concatStats.inputVectors, 6);
+  EXPECT_EQ(concatStats.outputVectors, 3);
+  EXPECT_EQ(concatStats.customStats.at("concatenateCalls").sum, 3);
+  EXPECT_EQ(concatStats.customStats.at("maxConcatenateInputBytes").max, 160);
+  EXPECT_EQ(concatStats.customStats.at("hardByteCapFlushes").sum, 2);
+}
+
+TEST_F(CudfBatchConcatTest, concatHonorsRowAndByteLimitsTogether) {
+  updateCudfConfig(
+      /*min=*/30,
+      /*max=*/20,
+      /*minBytes=*/0,
+      /*maxConcatBytes=*/250);
+  CudfConfig::getInstance().concatOptimizationEnabled = true;
+
+  std::vector<RowVectorPtr> vectors;
+  for (int i = 0; i < 6; ++i) {
+    vectors.push_back(makeRowVector({makeFlatSequence<int64_t>(i * 10, 10)}));
+  }
+  const auto concatStats =
+      runGlobalAggregation(vectors, "sum(c0)", "SELECT sum(c0) FROM tmp");
+  EXPECT_EQ(concatStats.outputVectors, 4);
+  EXPECT_EQ(concatStats.customStats.at("concatenateCalls").sum, 4);
+  EXPECT_LE(concatStats.customStats.at("maxConcatenateInputBytes").max, 250);
+}
+
+TEST_F(CudfBatchConcatTest, oversizedInputIsPassedThroughWithoutConcatenate) {
+  updateCudfConfig(
+      /*min=*/100000,
+      /*max=*/std::nullopt,
+      /*minBytes=*/0,
+      /*maxConcatBytes=*/64);
+  CudfConfig::getInstance().concatOptimizationEnabled = true;
+
+  std::vector<RowVectorPtr> vectors{
+      makeRowVector({makeFlatSequence<int64_t>(0, 10)})};
+  const auto concatStats =
+      runGlobalAggregation(vectors, "sum(c0)", "SELECT sum(c0) FROM tmp");
+  EXPECT_EQ(concatStats.outputVectors, 1);
+  EXPECT_EQ(concatStats.customStats.at("concatenateCalls").sum, 0);
+  EXPECT_EQ(concatStats.customStats.at("oversizedInputPassthroughs").sum, 1);
+  EXPECT_EQ(concatStats.customStats.at("maxConcatenateInputBytes").max, 0);
+}
+
 // Verifies that CudfBatchConcat is not inserted when the optimization is
 // disabled, even when aggregation is present.
 TEST_F(CudfBatchConcatTest, concatNotInsertedWhenDisabled) {
@@ -264,6 +378,163 @@ TEST_F(CudfBatchConcatTest, concatWithGroupedAggregation) {
   EXPECT_LT(concatIt->second->outputVectors, 6);
 }
 
+TEST_F(CudfBatchConcatTest, concatBoundsWideStringInputs) {
+  updateCudfConfig(
+      /*min=*/100000,
+      /*max=*/std::nullopt,
+      /*minBytes=*/0,
+      /*maxConcatBytes=*/1200);
+  CudfConfig::getInstance().concatOptimizationEnabled = true;
+
+  std::vector<RowVectorPtr> vectors;
+  for (int batch = 0; batch < 6; ++batch) {
+    vectors.push_back(makeRowVector(
+        {makeFlatSequence<int64_t>(batch * 4, 4),
+         makeFlatVector<std::string>(4, [batch](auto row) {
+           return std::string(128, static_cast<char>('a' + batch + row));
+         })}));
+  }
+  createDuckDbTable(vectors);
+
+  auto generator = std::make_shared<core::PlanNodeIdGenerator>();
+  core::PlanNodeId aggNodeId;
+  auto plan = PlanBuilder(generator)
+                  .addNode([&](auto, auto) {
+                    return createFragmentedSource(vectors, generator);
+                  })
+                  .singleAggregation({}, {"sum(c0)"})
+                  .capturePlanNodeId(aggNodeId)
+                  .planNode();
+
+  auto task = AssertQueryBuilder(duckDbQueryRunner_)
+                  .plan(plan)
+                  .maxDrivers(1)
+                  .assertResults("SELECT sum(c0) FROM tmp");
+
+  const auto planStats = toPlanStats(task->taskStats());
+  const auto& concatStats =
+      *planStats.at(aggNodeId).operatorStats.at("CudfBatchConcat");
+  EXPECT_EQ(concatStats.inputVectors, 6);
+  EXPECT_EQ(concatStats.outputVectors, 3);
+  EXPECT_EQ(concatStats.customStats.at("concatenateCalls").sum, 3);
+  EXPECT_LE(concatStats.customStats.at("maxConcatenateInputBytes").max, 1200);
+  EXPECT_GT(concatStats.customStats.at("hardByteCapFlushes").sum, 0);
+}
+
+TEST_F(CudfBatchConcatTest, concatSupportsNestedInputsBelowHardCap) {
+  updateCudfConfig(
+      /*min=*/100000,
+      /*max=*/std::nullopt,
+      /*minBytes=*/0,
+      /*maxConcatBytes=*/1 << 20);
+  CudfConfig::getInstance().concatOptimizationEnabled = true;
+
+  std::vector<RowVectorPtr> vectors;
+  for (int batch = 0; batch < 3; ++batch) {
+    vectors.push_back(makeRowVector(
+        {makeArrayVector<int64_t>(
+             {{batch, batch + 1}, {}, {batch + 2}, {batch + 3, batch + 4}}),
+         makeFlatSequence<int64_t>(batch * 4, 4)}));
+  }
+  createDuckDbTable(vectors);
+
+  auto generator = std::make_shared<core::PlanNodeIdGenerator>();
+  core::PlanNodeId aggNodeId;
+  auto plan = PlanBuilder(generator)
+                  .addNode([&](auto, auto) {
+                    return createFragmentedSource(vectors, generator);
+                  })
+                  .singleAggregation({}, {"sum(c1)"})
+                  .capturePlanNodeId(aggNodeId)
+                  .planNode();
+
+  auto task = AssertQueryBuilder(duckDbQueryRunner_)
+                  .plan(plan)
+                  .maxDrivers(1)
+                  .assertResults("SELECT sum(c1) FROM tmp");
+
+  const auto planStats = toPlanStats(task->taskStats());
+  const auto& concatStats =
+      *planStats.at(aggNodeId).operatorStats.at("CudfBatchConcat");
+  EXPECT_EQ(concatStats.outputVectors, 1);
+  EXPECT_EQ(concatStats.customStats.at("concatenateCalls").sum, 1);
+  EXPECT_LE(
+      concatStats.customStats.at("maxConcatenateInputBytes").max, 1 << 20);
+}
+
+TEST_F(CudfBatchConcatTest, concatJoinsInputStreamsAndPreservesOrder) {
+  auto inputType = ROW({"c0", "c1"}, {BIGINT(), VARCHAR()});
+  auto first = makeRowVector(
+      {"c0", "c1"},
+      {makeFlatVector<int64_t>({0, 1, 2, 3}),
+       makeFlatVector<std::string>({"s0", "s1", "s2", "s3"})});
+  auto second = makeRowVector(
+      {"c0", "c1"},
+      {makeFlatVector<int64_t>({4, 5, 6, 7}),
+       makeFlatVector<std::string>({"s4", "s5", "s6", "s7"})});
+  auto third = makeRowVector(
+      {"c0", "c1"},
+      {makeFlatVector<int64_t>({8, 9, 10, 11}),
+       makeFlatVector<std::string>({"s8", "s9", "s10", "s11"})});
+  auto expected = makeRowVector(
+      {"c0", "c1"},
+      {makeFlatSequence<int64_t>(0, 12),
+       makeFlatVector<std::string>(
+           {"s0",
+            "s1",
+            "s2",
+            "s3",
+            "s4",
+            "s5",
+            "s6",
+            "s7",
+            "s8",
+            "s9",
+            "s10",
+            "s11"})});
+
+  rmm::cuda_stream firstStream{rmm::cuda_stream::flags::non_blocking};
+  rmm::cuda_stream secondStream{rmm::cuda_stream::flags::non_blocking};
+  rmm::cuda_stream thirdStream{rmm::cuda_stream::flags::non_blocking};
+  rmm::cuda_stream outputStream{rmm::cuda_stream::flags::non_blocking};
+  ASSERT_NE(firstStream.value(), secondStream.value());
+  ASSERT_NE(secondStream.value(), thirdStream.value());
+  ASSERT_NE(thirdStream.value(), outputStream.value());
+
+  std::vector<CudfVectorPtr> inputs;
+  const auto addInput = [&](const RowVectorPtr& input, auto stream) {
+    auto table = with_arrow::toCudfTable(
+        input, pool(), stream, cudf_velox::get_output_mr());
+    inputs.push_back(std::make_shared<CudfVector>(
+        pool(), inputType, input->size(), std::move(table), stream));
+  };
+  addInput(first, firstStream.view());
+  addInput(second, secondStream.view());
+  addInput(third, thirdStream.view());
+
+  ConcatenateBatchStats stats;
+  auto outputs = getConcatenatedCudfVectorsBatched(
+      pool(),
+      std::move(inputs),
+      inputType,
+      outputStream.view(),
+      cudf_velox::get_output_mr(),
+      /*maxConcatBytes=*/1 << 20,
+      &stats);
+  ASSERT_EQ(outputs.size(), 1);
+  EXPECT_EQ(stats.concatenateCalls, 1);
+  EXPECT_LE(stats.maxInputBytes, 1 << 20);
+
+  auto result = with_arrow::toVeloxColumn(
+      outputs.front()->getTableView(),
+      pool(),
+      inputType,
+      outputStream.view(),
+      cudf_velox::get_output_mr());
+  outputStream.view().synchronize();
+  facebook::velox::test::assertEqualVectors(expected, result);
+}
+
 TEST_F(CudfBatchConcatTest, concatPreservesZeroColumnRowCountForCountStar) {
   updateCudfConfig(/*min=*/30, /*max=*/std::nullopt);
   CudfConfig::getInstance().concatOptimizationEnabled = true;
diff --git a/velox/experimental/cudf/tests/ConfigTest.cpp b/velox/experimental/cudf/tests/ConfigTest.cpp
index bfa2529ea2f..3df165445d0 100644
--- a/velox/experimental/cudf/tests/ConfigTest.cpp
+++ b/velox/experimental/cudf/tests/ConfigTest.cpp
@@ -29,6 +29,7 @@ TEST(ConfigTest, CudfConfig) {
   EXPECT_EQ(defaults.batchSizeMinThreshold, 100000);
   EXPECT_EQ(defaults.exchangeBatchSizeMinThreshold, 32000000);
   EXPECT_EQ(defaults.exchangeBatchSizeMinThresholdBytes, 0);
+  EXPECT_EQ(defaults.batchConcatMaxBytes, 1ULL << 30);
 
   std::unordered_map<std::string, std::string> options = {
       {CudfConfig::kCudfEnabled, "false"},
@@ -43,6 +44,7 @@ TEST(ConfigTest, CudfConfig) {
       {CudfConfig::kCudfOrderByMaxOutputRows, "1048576"},
       {CudfConfig::kCudfExchangeConcatOptimizationEnabled, "false"},
       {CudfConfig::kCudfExchangeBatchSizeMinThresholdBytes, "8388608"},
+      {CudfConfig::kCudfBatchConcatMaxBytes, "536870912"},
       {CudfConfig::kCudfOrderByMergeFanIn, "7"},
       {CudfConfig::kCudfWindowSortedRunBytes, "134217728"}};
 
@@ -55,6 +57,7 @@ TEST(ConfigTest, CudfConfig) {
   ASSERT_EQ(config.functionNamePrefix, "presto");
   ASSERT_EQ(config.exchangeConcatOptimizationEnabled, false);
   ASSERT_EQ(config.exchangeBatchSizeMinThresholdBytes, 8388608);
+  ASSERT_EQ(config.batchConcatMaxBytes, 536870912);
   ASSERT_EQ(config.allowCpuFallback, false);
   ASSERT_EQ(config.groupbyStreamingMaxDistinctKeys, 16777216);
   ASSERT_EQ(config.orderBySortedRunBytes, 67108864);
@@ -64,6 +67,12 @@ TEST(ConfigTest, CudfConfig) {
   ASSERT_EQ(config.orderByMaxOutputRows, 1048576);
 }
 
+TEST(ConfigTest, BatchConcatMaxBytesMustBePositive) {
+  CudfConfig config;
+  EXPECT_ANY_THROW(
+      config.initialize({{CudfConfig::kCudfBatchConcatMaxBytes, "0"}}));
+}
+
 TEST(ConfigTest, WindowBounds) {
   CudfConfig defaultConfig;
   EXPECT_EQ(defaultConfig.windowSortedRunBytes, 3ULL << 30);
diff --git a/velox/experimental/ucx-exchange/UcxExchangeServer.cpp b/velox/experimental/ucx-exchange/UcxExchangeServer.cpp
index 3e324745191..aa3a57ef01b 100644
--- a/velox/experimental/ucx-exchange/UcxExchangeServer.cpp
+++ b/velox/experimental/ucx-exchange/UcxExchangeServer.cpp
@@ -14,11 +14,13 @@
  * limitations under the License.
  */
 #include "velox/experimental/ucx-exchange/UcxExchangeServer.h"
+
 #include <glog/logging.h>
 #include <malloc.h>
 #include <rmm/cuda_stream_view.hpp>
 #include <algorithm>
 #include <cstdlib>
+#include <cstring>
 #include <limits>
 #include <string>
 #include "cuda_runtime.h"
@@ -30,6 +32,12 @@
 namespace facebook::velox::ucx_exchange {
 
 namespace {
+std::unique_ptr<std::vector<uint8_t>> transferMetadata(
+    const UcxTransferData& data) {
+  VELOX_CHECK_NOT_NULL(data.metadata);
+  return std::make_unique<std::vector<uint8_t>>(*data.metadata);
+}
+
 void accountFreedHostBytesAndTrim(uint64_t bytes) {
   constexpr uint64_t kTrimInterval = 64ULL * 1024 * 1024;
   static std::atomic<uint64_t> freedSinceTrim{0};
@@ -181,7 +189,7 @@ struct MetaSendContext {
 };
 
 struct DataSendContext {
-  std::shared_ptr<cudf::packed_columns> data;
+  std::shared_ptr<UcxTransferData> data;
   // The UCX build used by Gluten MPP may not include CUDA memory-type
   // transports.  In that case handing an rmm device pointer to tagSend makes
   // the shared-memory transport memcpy from an inaccessible address.  Keep a
@@ -276,7 +284,7 @@ void UcxExchangeServer::process() {
       // callback
       {
         std::weak_ptr<UcxExchangeServer> weakQueue = weak_from_this();
-        queueMgr_->getData(
+        queueMgr_->getTransferData(
             partitionKey_.taskId,
             partitionKey_.destination,
             // Unbounded per-fetch cap; rendezvous + queue-occupancy
@@ -284,7 +292,7 @@ void UcxExchangeServer::process() {
             std::numeric_limits<uint64_t>::max(),
             static_cast<int64_t>(sequenceNumber_),
             [weakQueue](
-                std::shared_ptr<cudf::packed_columns> data,
+                std::shared_ptr<UcxTransferData> data,
                 int64_t sequence,
                 std::vector<int64_t> remainingBytes) {
               auto self = weakQueue.lock();
@@ -463,9 +471,7 @@ void UcxExchangeServer::sendData() {
   VLOG(2) << (isIntraNodeTransfer_ ? "[INTRA]" : "[REMOTE]") << " [ExSrv "
           << partitionKey_.toString() << " seq=" << sequenceNumber_
           << "] sendData hasData=" << (dataPtr_ != nullptr)
-          << (dataPtr_ && dataPtr_->gpu_data
-                  ? " size=" + std::to_string(dataPtr_->gpu_data->size())
-                  : "");
+          << (dataPtr_ ? " size=" + std::to_string(dataPtr_->size()) : "");
 
   if (isIntraNodeTransfer_) {
     // INTRA-NODE TRANSFER PATH: Use registry for all communication, no UCXX
@@ -473,7 +479,21 @@ void UcxExchangeServer::sendData() {
     sendStart_ = std::chrono::high_resolution_clock::now();
 
     if (dataPtr_) {
-      bytes_ = dataPtr_->gpu_data->size();
+      bytes_ = dataPtr_->size();
+      if (dataPtr_->isHostStaged()) {
+        auto metadata = transferMetadata(*dataPtr_);
+        auto deviceBuffer = std::make_unique<rmm::device_buffer>(
+            bytes_, rmm::cuda_stream_default);
+        CUDF_CUDA_TRY(cudaMemcpy(
+            deviceBuffer->data(),
+            dataPtr_->hostData->data(),
+            bytes_,
+            cudaMemcpyHostToDevice));
+        dataPtr_->deviceData = std::make_shared<cudf::packed_columns>(
+            std::move(metadata), std::move(deviceBuffer));
+        dataPtr_->hostData.reset();
+        dataPtr_->metadata.reset();
+      }
 
       VLOG(3) << "@" << partitionKey_.taskId
               << " Intra-node transfer: publishing data for sequence "
@@ -481,14 +501,14 @@ void UcxExchangeServer::sendData() {
 
       IntraNodeTransferKey key{
           partitionKey_.taskId, partitionKey_.destination, sequenceNumber_};
-      const auto stream = dataPtr_->gpu_data->stream();
+      const auto stream = dataPtr_->deviceData->gpu_data->stream();
       // The consumer tags uniquely owned pages with this stream so downstream
       // reads and stream-ordered async frees remain ordered with the buffer.
       // dataPtr_ is already a shared_ptr, pass directly to share ownership.
       intraNodeRetrieveFuture_ =
           IntraNodeTransferRegistry::getInstance()->publish(
               key,
-              dataPtr_,
+              dataPtr_->deviceData,
               stream,
               /*atEnd=*/false,
               makeIntraNodeRetrieveWakeup());
@@ -530,12 +550,14 @@ void UcxExchangeServer::sendData() {
     }
   } else {
     // REMOTE EXCHANGE PATH: Use UCXX for metadata and data transfer
-    const bool useHostStaging = !communicator->hasCudaTransport();
+    const bool preStaged = dataPtr_ && dataPtr_->isHostStaged();
+    const bool useHostStaging = preStaged || !communicator->hasCudaTransport();
     std::shared_ptr<DataSendContext> dataCtx;
     if (dataPtr_) {
-      const auto hostBytes = static_cast<int64_t>(dataPtr_->gpu_data->size());
+      const auto hostBytes = dataPtr_->size();
       dataCtx = std::make_shared<DataSendContext>();
-      if (useHostStaging && !dataCtx->reserveHostBytes(hostBytes)) {
+      if (!preStaged && useHostStaging &&
+          !dataCtx->reserveHostBytes(hostBytes)) {
         // Keep dataPtr_ and state=DataReady.  Completed UCX callbacks release
         // process-wide credit; requeueing lets this server retry without
         // dequeuing or staging another packed table.
@@ -549,9 +571,11 @@ void UcxExchangeServer::sendData() {
       // Copy metadata (not move) because in broadcast mode, the same
       // packed_columns may be shared across multiple destination queues.
       // Metadata is small (CPU-side), so copying is negligible.
-      metadataMsg->cudfMetadata =
-          std::make_unique<std::vector<uint8_t>>(*dataPtr_->metadata);
-      metadataMsg->dataSizeBytes = dataPtr_->gpu_data->size();
+      metadataMsg->cudfMetadata = preStaged
+          ? transferMetadata(*dataPtr_)
+          : std::make_unique<std::vector<uint8_t>>(
+                *dataPtr_->deviceData->metadata);
+      metadataMsg->dataSizeBytes = dataPtr_->size();
       metadataMsg->remainingBytes = {};
       metadataMsg->atEnd = false;
     } else {
@@ -620,13 +644,10 @@ void UcxExchangeServer::sendData() {
     // send the data chunk (if any)
     if (dataPtr_) {
       sendStart_ = std::chrono::high_resolution_clock::now();
-      bytes_ = dataPtr_->gpu_data->size();
+      bytes_ = dataPtr_->size();
 
-      VLOG(3) << "@" << partitionKey_.taskId
-              << " Sending rmm::buffer: " << std::hex
-              << dataPtr_->gpu_data.get()
-              << " pointing to device memory: " << std::hex
-              << dataPtr_->gpu_data->data() << std::dec << " to task "
+      VLOG(3) << "@" << partitionKey_.taskId << " Sending "
+              << (preStaged ? "host buffer" : "rmm::buffer") << " to task "
               << partitionKey_.toString() << ":" << this->sequenceNumber_
               << std::dec << " of size " << bytes_;
 
@@ -642,20 +663,25 @@ void UcxExchangeServer::sendData() {
       // it after the DMA completes, while the Request (and context shell)
       // stays alive for UCP wireup replay.
       dataCtx->data = dataPtr_;
-      void* sendBuffer = dataCtx->data->gpu_data->data();
-      if (useHostStaging) {
+      void* sendBuffer = preStaged
+          ? static_cast<void*>(dataCtx->data->hostData->data())
+          : dataCtx->data->deviceData->gpu_data->data();
+      if (useHostStaging && !preStaged) {
         dataCtx->hostData = std::make_shared<std::vector<uint8_t>>(bytes_);
-        const auto producerStream = dataCtx->data->gpu_data->stream();
+        const auto producerStream =
+            dataCtx->data->deviceData->gpu_data->stream();
         CUDF_CUDA_TRY(cudaStreamSynchronize(producerStream.value()));
         CUDF_CUDA_TRY(cudaMemcpy(
             dataCtx->hostData->data(),
-            dataCtx->data->gpu_data->data(),
+            dataCtx->data->deviceData->gpu_data->data(),
             bytes_,
             cudaMemcpyDeviceToHost));
         sendBuffer = dataCtx->hostData->data();
       }
       VLOG(2) << "@" << partitionKey_.taskId << " posting "
-              << (useHostStaging ? "host-staged" : "direct-device")
+              << (preStaged            ? "pre-staged-host"
+                      : useHostStaging ? "host-staged"
+                                       : "direct-device")
               << " send for " << bytes_ << " bytes";
 
       dataRequest_ = endpointRef_->endpoint_->tagSend(
@@ -673,23 +699,22 @@ void UcxExchangeServer::sendData() {
             // callback means UCX has finished with both payloads; only the
             // empty context shell must remain alive with the Request.
             auto ctx = std::static_pointer_cast<DataSendContext>(arg);
+            const auto releasedHostBytes =
+                ctx->data != nullptr && ctx->data->isHostStaged()
+                ? ctx->data->size()
+                : ctx->reservedHostBytes;
             auto dataHolder = std::move(ctx->data);
             auto hostDataHolder = std::move(ctx->hostData);
-            const auto releasedHostBytes = ctx->reservedHostBytes;
             ctx->releaseHostReservation();
             hostDataHolder.reset();
-            // The default allocator retains these very large vector arenas in
-            // the executor even after free().  A long exchange therefore has
-            // bounded live staging but unbounded RSS. Return completed large
-            // transfers to the OS instead of waiting for process teardown.
-            accountFreedHostBytesAndTrim(releasedHostBytes);
-
+            dataHolder.reset();
             if (auto self = weakData.lock()) {
               self->sendComplete(status, arg);
             }
-            // The holders are destroyed here, releasing the GPU buffer if
-            // sendComplete() already reset the server's dataPtr_, and always
-            // releasing the completed transfer's host staging allocation.
+            // sendComplete() drops the server's last payload reference. Return
+            // completed large host allocations to the OS instead of retaining
+            // their arenas for the lifetime of the executor.
+            accountFreedHostBytesAndTrim(releasedHostBytes);
           },
           dataCtx);
     } else {
diff --git a/velox/experimental/ucx-exchange/UcxExchangeServer.h b/velox/experimental/ucx-exchange/UcxExchangeServer.h
index e4ac9585104..a0527814485 100644
--- a/velox/experimental/ucx-exchange/UcxExchangeServer.h
+++ b/velox/experimental/ucx-exchange/UcxExchangeServer.h
@@ -126,7 +126,7 @@ class UcxExchangeServer
   bool isIntraNodeTransfer_{false};
 
   std::atomic<ServerState> state_;
-  std::shared_ptr<cudf::packed_columns> dataPtr_{nullptr};
+  std::shared_ptr<UcxTransferData> dataPtr_{nullptr};
   /// Protects dataPtr_. Must be recursive because sendData() holds the lock
   /// when calling tagSend(), and for small messages UCX completes inline via
   /// its fast-completion path, firing the sendComplete() callback on the same
diff --git a/velox/experimental/ucx-exchange/UcxOutputQueueManager.cpp b/velox/experimental/ucx-exchange/UcxOutputQueueManager.cpp
index e12d7a13dcd..8445539602d 100644
--- a/velox/experimental/ucx-exchange/UcxOutputQueueManager.cpp
+++ b/velox/experimental/ucx-exchange/UcxOutputQueueManager.cpp
@@ -40,6 +40,8 @@ void UcxOutputQueueManager::initializeTask(
     int numDestinations,
     int numDrivers) {
   const auto& taskId = task->taskId();
+  const bool hostSpooling = hostSpoolingTasks_.withLock(
+      [&](const auto& tasks) { return tasks.count(taskId) != 0; });
   queues_.withLock([&](auto& queues) {
     auto it = queues.find(taskId);
     if (it == queues.end()) {
@@ -55,6 +57,9 @@ void UcxOutputQueueManager::initializeTask(
                 << " initializeTask ignored (already initialized)";
       }
     }
+    if (hostSpooling) {
+      queues[taskId]->enableHostSpooling();
+    }
   });
   // Clear any stale "removed" state so that getData() calls after this
   // initializeTask() create proper placeholder queues if needed.
@@ -100,6 +105,14 @@ void UcxOutputQueueManager::enqueue(
   getQueue(taskId)->enqueue(destination, std::move(txData), numRows);
 }
 
+void UcxOutputQueueManager::enableHostSpooling(std::string_view taskId) {
+  const std::string taskIdStr{taskId};
+  hostSpoolingTasks_.withLock([&](auto& tasks) { tasks.emplace(taskIdStr); });
+  if (auto queue = getQueueIfExists(taskId)) {
+    queue->enableHostSpooling();
+  }
+}
+
 bool UcxOutputQueueManager::checkBlocked(
     std::string_view taskId,
     ContinueFuture* future) {
@@ -200,6 +213,37 @@ void UcxOutputQueueManager::getData(
   outputQueue->getData(destination, maxBytes, sequence, notify);
 }
 
+void UcxOutputQueueManager::getTransferData(
+    std::string_view taskId,
+    int destination,
+    uint64_t maxBytes,
+    int64_t sequence,
+    UcxTransferDataAvailableCallback notify) {
+  std::shared_ptr<UcxOutputQueue> outputQueue;
+  bool taskRemoved = false;
+  std::string taskIdStr{taskId};
+  queues_.withLock([&](auto& queues) {
+    auto it = queues.find(taskIdStr);
+    if (it == queues.end()) {
+      if (removedTasks_.withLock(
+              [&](auto& removed) { return removed.count(taskIdStr) > 0; })) {
+        taskRemoved = true;
+        return;
+      }
+      outputQueue = std::make_shared<UcxOutputQueue>(nullptr, destination, 0);
+      queues[taskIdStr] = outputQueue;
+    } else {
+      outputQueue = it->second;
+    }
+  });
+  if (taskRemoved) {
+    notify(nullptr, sequence, {});
+    return;
+  }
+  outputQueue->getTransferData(
+      destination, maxBytes, sequence, std::move(notify));
+}
+
 bool UcxOutputQueueManager::canUseIntraNode(std::string_view taskId) {
   auto queue = getQueueIfExists(taskId);
   if (!queue) {
@@ -247,6 +291,7 @@ void UcxOutputQueueManager::removeTask(std::string_view taskId) {
   if (queue != nullptr) {
     queue->terminate();
   }
+  hostSpoolingTasks_.withLock([&](auto& tasks) { tasks.erase(taskIdStr); });
   // Notify the intra-node registry so that any sources polling for this
   // task get an atEnd result instead of spinning forever.
   IntraNodeTransferRegistry::getInstance()->cancelTask(taskId);
diff --git a/velox/experimental/ucx-exchange/UcxOutputQueueManager.h b/velox/experimental/ucx-exchange/UcxOutputQueueManager.h
index 5d54da4aa6f..f4ac88254d7 100644
--- a/velox/experimental/ucx-exchange/UcxOutputQueueManager.h
+++ b/velox/experimental/ucx-exchange/UcxOutputQueueManager.h
@@ -81,6 +81,10 @@ class UcxOutputQueueManager {
       std::unique_ptr<cudf::packed_columns> txData,
       int32_t numRows);
 
+  /// Marks a producer task for eager host materialization before it starts.
+  /// The marker is retained until initializeTask() creates its output queue.
+  void enableHostSpooling(std::string_view taskId);
+
   /// @brief Checks if the queue for a task is over capacity.
   /// Should be called after enqueueing all partitions for a batch.
   /// @param taskId The unique task Id.
@@ -119,6 +123,13 @@ class UcxOutputQueueManager {
       int64_t sequence,
       UcxDataAvailableCallbackV2 notify);
 
+  void getTransferData(
+      std::string_view taskId,
+      int destination,
+      uint64_t maxBytes,
+      int64_t sequence,
+      UcxTransferDataAvailableCallback notify);
+
   /// Returns true if the given task can use intra-node transfer.
   /// Returns false until the task queue is initialized. Initialized broadcast
   /// queues are safe because shared pages are cloned by UcxExchangeSource.
@@ -153,6 +164,9 @@ class UcxOutputQueueManager {
   // that exceed the placeholder's undersized queues_ vector.
   folly::Synchronized<std::unordered_set<std::string>, std::mutex>
       removedTasks_;
+
+  folly::Synchronized<std::unordered_set<std::string>, std::mutex>
+      hostSpoolingTasks_;
 };
 
 } // namespace facebook::velox::ucx_exchange
diff --git a/velox/experimental/ucx-exchange/UcxQueues.cpp b/velox/experimental/ucx-exchange/UcxQueues.cpp
index 373f79754a7..b3c72e58491 100644
--- a/velox/experimental/ucx-exchange/UcxQueues.cpp
+++ b/velox/experimental/ucx-exchange/UcxQueues.cpp
@@ -17,9 +17,14 @@
 
 #include "velox/experimental/cudf/exec/GpuResources.h"
 
+#include <malloc.h>
 #include <atomic>
+#include <cerrno>
+#include <cstdlib>
+#include <cstring>
 #include <limits>
 #include <sstream>
+#include "cuda_runtime.h"
 
 namespace facebook::velox::ucx_exchange {
 
@@ -28,6 +33,48 @@ std::atomic<int64_t> diagnosticGlobalQueuedBytes{0};
 std::atomic<int64_t> diagnosticGlobalQueuedColumns{0};
 std::atomic<int64_t> diagnosticGlobalQueueGiB{0};
 
+struct PinnedScratch {
+  ~PinnedScratch() {
+    if (data != nullptr) {
+      cudaFreeHost(data);
+    }
+  }
+
+  uint8_t* ensure(size_t bytes) {
+    if (capacity >= bytes) {
+      return data;
+    }
+    if (data != nullptr) {
+      CUDF_CUDA_TRY(cudaFreeHost(data));
+      data = nullptr;
+      capacity = 0;
+    }
+    CUDF_CUDA_TRY(cudaMallocHost(reinterpret_cast<void**>(&data), bytes));
+    capacity = bytes;
+    return data;
+  }
+
+  uint8_t* data{nullptr};
+  size_t capacity{0};
+};
+
+thread_local PinnedScratch pinnedScratch;
+
+uint64_t configuredHostSpoolMaxBytes() {
+  const auto* value = std::getenv("GLUTEN_UCX_HOST_SPOOL_MAX_BYTES");
+  if (value == nullptr || *value == '\0') {
+    return 0;
+  }
+  errno = 0;
+  char* end = nullptr;
+  const auto parsed = std::strtoull(value, &end, 10);
+  if (errno == 0 && end != value && *end == '\0' && parsed > 0) {
+    return parsed;
+  }
+  LOG(WARNING) << "Ignoring invalid GLUTEN_UCX_HOST_SPOOL_MAX_BYTES=" << value;
+  return 0;
+}
+
 void updateDiagnosticGlobalQueue(
     int64_t bytes,
     int64_t columns,
@@ -54,21 +101,31 @@ void updateDiagnosticGlobalQueue(
 }
 } // namespace
 
-void UcxDestinationQueue::Stats::recordEnqueue(
-    const cudf::packed_columns* data) {
+void UcxDestinationQueue::Stats::recordEnqueue(const UcxTransferData* data) {
   if (data != nullptr) {
-    bytesQueued += data->gpu_data->size();
+    bytesQueued += data->size();
+    backpressureBytesQueued += data->backpressureSize();
+    deviceBytesQueued += data->isHostStaged() ? 0 : data->backpressureSize();
     packedColumnsQueued++;
   }
 }
 
-void UcxDestinationQueue::Stats::recordDequeue(
-    const cudf::packed_columns* data) {
+void UcxDestinationQueue::Stats::recordDequeue(const UcxTransferData* data) {
   if (data != nullptr) {
-    const int64_t size = data->gpu_data->size();
+    const int64_t size = data->size();
+    const int64_t backpressureSize = data->backpressureSize();
+    const int64_t deviceBytes = data->isHostStaged() ? 0 : backpressureSize;
 
     bytesQueued -= size;
     VELOX_DCHECK_GE(bytesQueued, 0, "bytesQueued must be non-negative");
+    backpressureBytesQueued -= backpressureSize;
+    VELOX_DCHECK_GE(
+        backpressureBytesQueued,
+        0,
+        "backpressureBytesQueued must be non-negative");
+    deviceBytesQueued -= deviceBytes;
+    VELOX_DCHECK_GE(
+        deviceBytesQueued, 0, "deviceBytesQueued must be non-negative");
     --packedColumnsQueued;
     VELOX_DCHECK_GE(
         packedColumnsQueued, 0, "packedColumnsQueued must be non-negative");
@@ -78,8 +135,7 @@ void UcxDestinationQueue::Stats::recordDequeue(
   }
 }
 
-void UcxDestinationQueue::enqueueBack(
-    std::shared_ptr<cudf::packed_columns> data) {
+void UcxDestinationQueue::enqueueBack(std::shared_ptr<UcxTransferData> data) {
   // drop duplicate end markers.
   if (data == nullptr && !queue_.empty() && queue_.back() == nullptr) {
     return;
@@ -91,8 +147,7 @@ void UcxDestinationQueue::enqueueBack(
   queue_.push_back(std::move(data));
 }
 
-void UcxDestinationQueue::enqueueFront(
-    std::shared_ptr<cudf::packed_columns> data) {
+void UcxDestinationQueue::enqueueFront(std::shared_ptr<UcxTransferData> data) {
   // ignore nullptr.
   if (data == nullptr) {
     return;
@@ -103,24 +158,15 @@ void UcxDestinationQueue::enqueueFront(
 }
 
 UcxDestinationQueue::Data UcxDestinationQueue::getData(
-    UcxDataAvailableCallback notify) {
+    UcxTransferDataAvailableCallback notify) {
   return getData(
-      std::numeric_limits<uint64_t>::max(),
-      sequence_,
-      [notify = std::move(notify)](
-          std::shared_ptr<cudf::packed_columns> data,
-          int64_t /*sequence*/,
-          std::vector<int64_t> remainingBytes) mutable {
-        if (notify) {
-          notify(std::move(data), std::move(remainingBytes));
-        }
-      });
+      std::numeric_limits<uint64_t>::max(), sequence_, std::move(notify));
 }
 
 UcxDestinationQueue::Data UcxDestinationQueue::getData(
     uint64_t maxBytes,
     int64_t sequence,
-    UcxDataAvailableCallbackV2 notify) {
+    UcxTransferDataAvailableCallback notify) {
   if (sequence < sequence_) {
     // A retried/duplicate UCX connection can race with task abort after the
     // original server has already advanced this destination queue.  Treat
@@ -132,7 +178,7 @@ UcxDestinationQueue::Data UcxDestinationQueue::getData(
                  << sequence << " acknowledgedSequence=" << sequence_;
     return {nullptr, sequence_, {}, true};
   }
-  if (notifyV2_ != nullptr && notify != nullptr) {
+  if (notify_ != nullptr && notify != nullptr) {
     // A second server for the same task/destination/sequence must not replace
     // the active server's waiter.  Return a deliberately different sequence
     // so the duplicate UcxExchangeServer follows its stale-connection close
@@ -142,13 +188,13 @@ UcxDestinationQueue::Data UcxDestinationQueue::getData(
     return {nullptr, sequence_ + 1, {}, true};
   }
   VELOX_CHECK(
-      notify_ == nullptr && notifyV2_ == nullptr,
+      notify_ == nullptr,
       "UcxDestinationQueue already has a pending data notification");
   if (sequence > sequence_) {
     // Minimal V2 implementation only supports in-order requests. The full
     // Presto-style ack path can skip prefixes later; for now, install the
     // notify and wait for the requested sequence to become available.
-    notifyV2_ = std::move(notify);
+    notify_ = std::move(notify);
     notifySequence_ = sequence;
     notifyMaxBytes_ = maxBytes;
     return {};
@@ -156,7 +202,7 @@ UcxDestinationQueue::Data UcxDestinationQueue::getData(
 
   if (queue_.empty()) {
     // delay notification.
-    notifyV2_ = std::move(notify);
+    notify_ = std::move(notify);
     notifySequence_ = sequence;
     notifyMaxBytes_ = maxBytes;
     return {};
@@ -189,18 +235,16 @@ UcxDataAvailable UcxDestinationQueue::deleteResults() {
 
   UcxDataAvailable result;
   result.callback = std::move(notify_);
-  result.callbackV2 = std::move(notifyV2_);
   result.sequence = notifySequence_;
   clearNotify();
   return result;
 }
 
 UcxDataAvailable UcxDestinationQueue::getAndClearNotify() {
-  if (notify_ == nullptr && notifyV2_ == nullptr) {
+  if (notify_ == nullptr) {
     return UcxDataAvailable();
   }
-  auto savedV1 = std::move(notify_);
-  auto savedV2 = std::move(notifyV2_);
+  auto saved = std::move(notify_);
   const auto savedSequence = notifySequence_;
   const auto savedMaxBytes = notifyMaxBytes_;
   clearNotify();
@@ -210,16 +254,14 @@ UcxDataAvailable UcxDestinationQueue::getAndClearNotify() {
       savedSequence,
       nullptr);
   if (!data.immediate) {
-    notify_ = std::move(savedV1);
-    notifyV2_ = std::move(savedV2);
+    notify_ = std::move(saved);
     notifySequence_ = savedSequence;
     notifyMaxBytes_ = savedMaxBytes;
     return UcxDataAvailable();
   }
 
   UcxDataAvailable result;
-  result.callback = std::move(savedV1);
-  result.callbackV2 = std::move(savedV2);
+  result.callback = std::move(saved);
   result.sequence = data.sequence;
   result.data = std::move(data.data);
   result.remainingBytes = std::move(data.remainingBytes);
@@ -228,14 +270,12 @@ UcxDataAvailable UcxDestinationQueue::getAndClearNotify() {
 
 void UcxDestinationQueue::clearNotify() {
   notify_ = nullptr;
-  notifyV2_ = nullptr;
   notifySequence_ = 0;
   notifyMaxBytes_ = 0;
 }
 
 void UcxDestinationQueue::finish() {
   VELOX_CHECK_NULL(notify_, "notify must be cleared before finish");
-  VELOX_CHECK_NULL(notifyV2_, "V2 notify must be cleared before finish");
   VELOX_CHECK(queue_.empty(), "data must be fetched before finish");
 }
 
@@ -247,7 +287,6 @@ std::string UcxDestinationQueue::toString() {
   std::stringstream out;
   out << "[available: " << queue_.size() << ", "
       << "sequence: " << sequence_ << ", "
-      << (notifyV2_ ? "notifyV2 registered, " : "")
       << (notify_ ? "notify registered, " : "") << this << "]";
   return out.str();
 }
@@ -289,6 +328,10 @@ bool UcxOutputQueue::initialize(
   task_ = task;
   maxSize_ = task_->queryCtx()->queryConfig().maxOutputBufferSize();
   continueSize_ = (maxSize_ * kContinuePct) / 100;
+  if (hostSpooling_) {
+    hostSpoolMaxSize_ = std::max(hostSpoolMaxSize_, maxSize_);
+    hostSpoolContinueSize_ = (hostSpoolMaxSize_ * kContinuePct) / 100;
+  }
   // Publish task metadata before destination queue expansion. Acceptor only
   // needs task/kind to choose the intra-node path; getData() takes mutex_ and
   // waits for any queue expansion in this function to finish.
@@ -324,17 +367,48 @@ void UcxOutputQueue::enqueue(
   VELOX_CHECK_NOT_NULL(task_);
   VELOX_CHECK(
       task_->isRunning(), "Task is terminated, cannot add data to output.");
-  std::vector<UcxDataAvailable> dataAvailableCallbacks;
+
+  bool hostSpooling = false;
   {
     std::lock_guard<std::mutex> l(mutex_);
-    auto numBytes = data->gpu_data->size();
-    auto sharedData = std::shared_ptr<cudf::packed_columns>(std::move(data));
+    VELOX_CHECK_GE(destination, 0);
+    VELOX_CHECK_LT(destination, queues_.size());
+    hostSpooling = hostSpooling_;
+  }
+
+  const auto numBytes = static_cast<int64_t>(data->gpu_data->size());
+  auto transfer = std::make_shared<UcxTransferData>();
+  if (hostSpooling) {
+    VELOX_CHECK_NE(
+        kind_,
+        core::PartitionedOutputNode::Kind::kBroadcast,
+        "Host spooling is only supported for partitioned output");
+    transfer->metadata =
+        std::shared_ptr<std::vector<uint8_t>>(std::move(data->metadata));
+    auto* hostStaging = pinnedScratch.ensure(static_cast<size_t>(numBytes));
+    const auto stream = data->gpu_data->stream();
+    CUDF_CUDA_TRY(cudaStreamSynchronize(stream.value()));
+    CUDF_CUDA_TRY(cudaMemcpy(
+        hostStaging, data->gpu_data->data(), numBytes, cudaMemcpyDeviceToHost));
+    transfer->hostData =
+        std::make_shared<std::vector<uint8_t>>(static_cast<size_t>(numBytes));
+    std::memcpy(transfer->hostData->data(), hostStaging, numBytes);
+    data.reset();
+  } else {
+    transfer->deviceData =
+        std::shared_ptr<cudf::packed_columns>(std::move(data));
+  }
+  const auto hostResidentBytes = transfer->hostResidentSize();
 
+  std::vector<UcxDataAvailable> dataAvailableCallbacks;
+  bool trimHostAllocator = false;
+  {
+    std::lock_guard<std::mutex> l(mutex_);
     bool success = false;
     if (kind_ == core::PartitionedOutputNode::Kind::kBroadcast) {
       VELOX_CHECK_EQ(destination, 0, "Broadcast uses destination 0");
       enqueueBroadcastOutputLocked(
-          std::move(sharedData), dataAvailableCallbacks);
+          transfer->deviceData, dataAvailableCallbacks);
       // For broadcast, count queuedBytes_ once per active destination so
       // that each destination's dequeue symmetrically decrements it. The
       // total sent stats count the logical data once.
@@ -346,6 +420,7 @@ void UcxOutputQueue::enqueue(
       }
       updateTotalQueuedBytesMsLocked();
       queuedBytes_ += numBytes * numActive;
+      deviceQueuedBytes_ += numBytes * numActive;
       queuedPackedColumns_ += numActive;
       totalBytesSent_ += numBytes;
       totalRowsSent_ += numRows;
@@ -354,9 +429,34 @@ void UcxOutputQueue::enqueue(
     } else {
       VELOX_CHECK_LT(destination, queues_.size());
       success = enqueuePartitionedOutputLocked(
-          destination, std::move(sharedData), dataAvailableCallbacks);
+          destination, std::move(transfer), dataAvailableCallbacks);
       if (success) {
-        updateStatsWithEnqueuedLocked(numBytes, numRows);
+        updateStatsWithEnqueuedLocked(
+            numBytes,
+            hostSpooling ? hostResidentBytes : numBytes,
+            hostSpooling ? 0 : numBytes,
+            numRows);
+        if (hostSpooling) {
+          hostSpooledBytes_ += numBytes;
+          hostSpoolPeakBytes_ =
+              std::max(hostSpoolPeakBytes_, hostSpooledBytes_);
+          hostSpoolResidentBytes_ += hostResidentBytes;
+          hostSpoolPeakResidentBytes_ =
+              std::max(hostSpoolPeakResidentBytes_, hostSpoolResidentBytes_);
+          const auto currentGiB = hostSpooledBytes_ >> 30;
+          const auto previousGiB = (hostSpooledBytes_ - numBytes) >> 30;
+          if (currentGiB != previousGiB) {
+            trimHostAllocator = true;
+            LOG(WARNING) << "CUDF_UCX_HOST_SPOOL task=" << task_->taskId()
+                         << " event=enqueue hostQueuedBytes="
+                         << hostSpooledBytes_
+                         << " hostPeakBytes=" << hostSpoolPeakBytes_
+                         << " hostResidentBytes=" << hostSpoolResidentBytes_
+                         << " hostPeakResidentBytes="
+                         << hostSpoolPeakResidentBytes_
+                         << " packedColumns=" << queuedPackedColumns_;
+          }
+        }
       }
     }
   }
@@ -364,14 +464,42 @@ void UcxOutputQueue::enqueue(
   for (auto& callback : dataAvailableCallbacks) {
     callback.notify();
   }
+  if (trimHostAllocator) {
+    malloc_trim(0);
+  }
+}
+
+void UcxOutputQueue::enableHostSpooling() {
+  std::lock_guard<std::mutex> l(mutex_);
+  VELOX_CHECK_EQ(
+      queuedPackedColumns_, 0, "Host spooling must be enabled before enqueue");
+  VELOX_CHECK_NE(
+      kind_,
+      core::PartitionedOutputNode::Kind::kBroadcast,
+      "Host spooling is not supported for broadcast output");
+  if (hostSpooling_) {
+    return;
+  }
+  hostSpooling_ = true;
+  hostSpoolMaxSize_ =
+      std::max<uint64_t>(maxSize_, configuredHostSpoolMaxBytes());
+  hostSpoolContinueSize_ = (hostSpoolMaxSize_ * kContinuePct) / 100;
+  LOG(WARNING) << "CUDF_UCX_HOST_SPOOL task="
+               << (task_ ? task_->taskId() : "pending")
+               << " event=enabled hostMaxBytes=" << hostSpoolMaxSize_
+               << " deviceMaxBytes=" << maxSize_;
 }
 
 bool UcxOutputQueue::checkBlocked(ContinueFuture* future) {
   std::lock_guard<std::mutex> l(mutex_);
-  if (queuedBytes_ >= maxSize_ && future) {
+  const bool deviceBlocked = deviceQueuedBytes_ >= maxSize_;
+  const bool hostBlocked = hostSpooling_ && queuedBytes_ >= hostSpoolMaxSize_;
+  if ((deviceBlocked || hostBlocked) && future) {
     VLOG(2) << "[BACKPRESSURE] task=" << (task_ ? task_->taskId() : "n/a")
             << " BLOCKED queuedBytes=" << queuedBytes_
-            << " maxSize=" << maxSize_
+            << " hostMaxSize=" << hostSpoolMaxSize_
+            << " deviceQueuedBytes=" << deviceQueuedBytes_
+            << " deviceMaxSize=" << maxSize_
             << " waitingProducers=" << (promises_.size() + 1);
     promises_.emplace_back("UcxOutputQueue::checkBlocked");
     *future = promises_.back().getSemiFuture();
@@ -381,109 +509,110 @@ bool UcxOutputQueue::checkBlocked(ContinueFuture* future) {
 }
 
 void UcxOutputQueue::getData(int destination, UcxDataAvailableCallback notify) {
+  getTransferData(
+      destination,
+      std::numeric_limits<uint64_t>::max(),
+      -1,
+      [notify = std::move(notify)](
+          std::shared_ptr<UcxTransferData> data,
+          int64_t /*sequence*/,
+          std::vector<int64_t> remainingBytes) mutable {
+        VELOX_CHECK(
+            data == nullptr || data->deviceData != nullptr,
+            "Legacy UCX queue fetch cannot consume host-spooled data");
+        notify(
+            data == nullptr ? nullptr : data->deviceData,
+            std::move(remainingBytes));
+      });
+}
+
+void UcxOutputQueue::getData(
+    int destination,
+    uint64_t maxBytes,
+    int64_t sequence,
+    UcxDataAvailableCallbackV2 notify) {
+  getTransferData(
+      destination,
+      maxBytes,
+      sequence,
+      [notify = std::move(notify)](
+          std::shared_ptr<UcxTransferData> data,
+          int64_t sequence,
+          std::vector<int64_t> remainingBytes) mutable {
+        VELOX_CHECK(
+            data == nullptr || data->deviceData != nullptr,
+            "Legacy UCX queue fetch cannot consume host-spooled data");
+        notify(
+            data == nullptr ? nullptr : data->deviceData,
+            sequence,
+            std::move(remainingBytes));
+      });
+}
+
+void UcxOutputQueue::getTransferData(
+    int destination,
+    uint64_t maxBytes,
+    int64_t sequence,
+    UcxTransferDataAvailableCallback notify) {
   UcxDestinationQueue::Data data;
   std::vector<ContinuePromise> promises;
   {
     std::lock_guard<std::mutex> l(mutex_);
-    // If the queue doesn't exist yet, create an empty queue to store
-    // the notify callback. The queue will eventually be initialized when
-    // the task is being created.
     for (int i = queues_.size(); i <= destination; ++i) {
-      // create the destination queues inside the vector using emplace_back.
       queues_.emplace_back(std::make_unique<UcxDestinationQueue>());
     }
     auto* queue = queues_[destination].get();
-    // queue can be nullptr here if the task has terminated and results
-    // have been removed. In this case, no data is returned.
     if (queue) {
-      // Capture weak_ptr instead of raw `this` to prevent use-after-free.
-      // The callback fires outside the lock (from enqueue() or terminate()),
-      // and concurrent removeTask() can destroy the UcxOutputQueue while
-      // the callback is still executing.
       std::weak_ptr<UcxOutputQueue> weakSelf = shared_from_this();
-      data = queue->getData([notify, weakSelf](
-                                std::shared_ptr<cudf::packed_columns> data,
-                                std::vector<int64_t> remainingBytes) {
+      auto callback = [notify, weakSelf](
+                          std::shared_ptr<UcxTransferData> data,
+                          int64_t sequence,
+                          std::vector<int64_t> remainingBytes) {
         std::vector<ContinuePromise> promises;
-        int64_t bytes = data ? data->gpu_data->size() : -1L;
-        notify(std::move(data), std::move(remainingBytes));
+        int64_t bytes = data ? data->size() : -1L;
+        int64_t backpressureBytes = data ? data->backpressureSize() : -1L;
+        const bool hostStaged = data && data->isHostStaged();
+        const int64_t hostResidentBytes =
+            hostStaged ? data->hostResidentSize() : 0;
+        notify(std::move(data), sequence, std::move(remainingBytes));
         if (bytes >= 0L) {
           auto self = weakSelf.lock();
           if (!self) {
-            // Queue was destroyed by removeTask(), safe to skip stats update.
             return;
           }
           std::lock_guard<std::mutex> l(self->mutex_);
-          self->updateStatsWithFreedLocked(bytes, 1L, promises);
+          self->updateStatsWithFreedLocked(
+              backpressureBytes,
+              hostStaged ? 0 : backpressureBytes,
+              1L,
+              promises);
+          if (hostStaged) {
+            self->hostSpooledBytes_ -= bytes;
+            VELOX_CHECK_GE(self->hostSpooledBytes_, 0);
+            self->hostSpoolResidentBytes_ -= hostResidentBytes;
+            VELOX_CHECK_GE(self->hostSpoolResidentBytes_, 0);
+          }
         }
-        // outside of lock:
-        // wake up any producers that are waiting for queue to become less full.
         for (auto& promise : promises) {
           promise.setValue();
         }
-      });
-      if (data.data) {
-        // This implies data.immediate and no notify upcall will be done.
-        // Need to update the stats here.
-        updateStatsWithFreedLocked(data.data->gpu_data->size(), 1L, promises);
-      }
-    } else {
-      data = UcxDestinationQueue::Data{nullptr, 0, {}, true};
-    }
-  }
-  // outside lock: If we have data, then return it immediately.
-  if (data.immediate) {
-    notify(std::move(data.data), std::move(data.remainingBytes));
-  } else {
-    VLOG(2) << "[QUEUE] task=" << (task_ ? task_->taskId() : "n/a")
-            << " dest=" << destination
-            << " server waiting for data (callback installed)";
-  }
-  // wake up any producers that are waiting for queue to become less full.
-  for (auto& promise : promises) {
-    promise.setValue();
-  }
-}
-
-void UcxOutputQueue::getData(
-    int destination,
-    uint64_t maxBytes,
-    int64_t sequence,
-    UcxDataAvailableCallbackV2 notify) {
-  UcxDestinationQueue::Data data;
-  std::vector<ContinuePromise> promises;
-  {
-    std::lock_guard<std::mutex> l(mutex_);
-    for (int i = queues_.size(); i <= destination; ++i) {
-      queues_.emplace_back(std::make_unique<UcxDestinationQueue>());
-    }
-    auto* queue = queues_[destination].get();
-    if (queue) {
-      std::weak_ptr<UcxOutputQueue> weakSelf = shared_from_this();
-      data = queue->getData(
-          maxBytes,
-          sequence,
-          [notify, weakSelf](
-              std::shared_ptr<cudf::packed_columns> data,
-              int64_t sequence,
-              std::vector<int64_t> remainingBytes) {
-            std::vector<ContinuePromise> promises;
-            int64_t bytes = data ? data->gpu_data->size() : -1L;
-            notify(std::move(data), sequence, std::move(remainingBytes));
-            if (bytes >= 0L) {
-              auto self = weakSelf.lock();
-              if (!self) {
-                return;
-              }
-              std::lock_guard<std::mutex> l(self->mutex_);
-              self->updateStatsWithFreedLocked(bytes, 1L, promises);
-            }
-            for (auto& promise : promises) {
-              promise.setValue();
-            }
-          });
+      };
+      data = sequence < 0
+          ? queue->getData(std::move(callback))
+          : queue->getData(maxBytes, sequence, std::move(callback));
       if (data.data) {
-        updateStatsWithFreedLocked(data.data->gpu_data->size(), 1L, promises);
+        const auto bytes = data.data->size();
+        updateStatsWithFreedLocked(
+            data.data->backpressureSize(),
+            data.data->isHostStaged() ? 0 : data.data->backpressureSize(),
+            1L,
+            promises);
+        if (data.data->isHostStaged()) {
+          hostSpooledBytes_ -= bytes;
+          VELOX_CHECK_GE(hostSpooledBytes_, 0);
+          hostSpoolResidentBytes_ -= data.data->hostResidentSize();
+          VELOX_CHECK_GE(hostSpoolResidentBytes_, 0);
+        }
       }
     } else {
       data = UcxDestinationQueue::Data{nullptr, sequence, {}, true};
@@ -536,6 +665,17 @@ void UcxOutputQueue::checkIfDone(bool oneDriverFinished) {
               << " chunks=" << totalPackedColumnsSent_
               << " avgRowsPerChunk=" << avgRows
               << " totalBytes=" << totalBytesSent_;
+      if (hostSpooling_) {
+        LOG(WARNING) << "CUDF_UCX_HOST_SPOOL task="
+                     << (task_ ? task_->taskId() : "n/a")
+                     << " event=producer_finished"
+                     << " hostQueuedBytes=" << hostSpooledBytes_
+                     << " hostPeakBytes=" << hostSpoolPeakBytes_
+                     << " hostResidentBytes=" << hostSpoolResidentBytes_
+                     << " hostPeakResidentBytes=" << hostSpoolPeakResidentBytes_
+                     << " totalBytes=" << totalBytesSent_
+                     << " packedColumns=" << totalPackedColumnsSent_;
+      }
     }
     for (auto& queue : queues_) {
       if (queue != nullptr) {
@@ -552,7 +692,7 @@ void UcxOutputQueue::checkIfDone(bool oneDriverFinished) {
 
 bool UcxOutputQueue::enqueuePartitionedOutputLocked(
     int destination,
-    std::shared_ptr<cudf::packed_columns> data,
+    std::shared_ptr<UcxTransferData> data,
     std::vector<UcxDataAvailable>& dataAvailableCbs) {
   VELOX_DCHECK(dataAvailableCbs.empty());
   VELOX_CHECK_LT(destination, queues_.size());
@@ -571,9 +711,11 @@ void UcxOutputQueue::enqueueBroadcastOutputLocked(
     std::vector<UcxDataAvailable>& dataAvailableCbs) {
   VELOX_DCHECK(dataAvailableCbs.empty());
 
+  auto transfer = std::make_shared<UcxTransferData>();
+  transfer->deviceData = data;
   for (auto& queue : queues_) {
     if (queue != nullptr) {
-      queue->enqueueBack(data);
+      queue->enqueueBack(transfer);
       dataAvailableCbs.emplace_back(queue->getAndClearNotify());
     }
   }
@@ -626,10 +768,13 @@ void UcxOutputQueue::updateOutputBuffers(int numBuffers, bool noMoreBuffers) {
       for (int32_t i = 0; i < numNewBuffers; ++i) {
         auto buffer = std::make_unique<UcxDestinationQueue>();
         for (const auto& data : dataToBroadcast_) {
-          buffer->enqueueBack(data);
+          auto transfer = std::make_shared<UcxTransferData>();
+          transfer->deviceData = data;
+          buffer->enqueueBack(std::move(transfer));
           // Account for backfilled data in queuedBytes_ so that dequeue
           // decrements don't drive it negative.
           queuedBytes_ += data->gpu_data->size();
+          deviceQueuedBytes_ += data->gpu_data->size();
           queuedPackedColumns_++;
         }
         if (atEnd_) {
@@ -670,7 +815,8 @@ void UcxOutputQueue::deleteResults(int destination) {
       return;
     }
     // remember destination queue fill stats
-    int64_t bytes = queue->stats().bytesQueued;
+    int64_t bytes = queue->stats().backpressureBytesQueued;
+    int64_t deviceBytes = queue->stats().deviceBytesQueued;
     int64_t packedCols = queue->stats().packedColumnsQueued;
     dataAvailable = queue->deleteResults();
     queue->finish();
@@ -678,7 +824,7 @@ void UcxOutputQueue::deleteResults(int destination) {
     isFinished = isFinishedLocked();
     // update UcxOutputQueue stats
     if (bytes > 0 || packedCols > 0) {
-      updateStatsWithFreedLocked(bytes, packedCols, promises);
+      updateStatsWithFreedLocked(bytes, deviceBytes, packedCols, promises);
     } else {
       promises = std::move(promises_);
     }
@@ -751,41 +897,52 @@ exec::OutputBuffer::Stats UcxOutputQueue::stats() {
 }
 
 void UcxOutputQueue::updateStatsWithEnqueuedLocked(
-    int64_t bytes,
+    int64_t logicalBytes,
+    int64_t backpressureBytes,
+    int64_t deviceBytes,
     int64_t rows) {
   updateTotalQueuedBytesMsLocked();
 
-  queuedBytes_ += bytes;
+  queuedBytes_ += backpressureBytes;
+  deviceQueuedBytes_ += deviceBytes;
   queuedPackedColumns_++;
 
-  totalBytesSent_ += bytes;
+  totalBytesSent_ += logicalBytes;
   totalRowsSent_ += rows;
   totalPackedColumnsSent_++;
-  updateDiagnosticGlobalQueue(bytes, 1, "enqueue", task_);
+  updateDiagnosticGlobalQueue(backpressureBytes, 1, "enqueue", task_);
   logDeviceQueueResidencyLocked("enqueue");
 }
 
 void UcxOutputQueue::updateStatsWithFreedLocked(
     int64_t bytes,
+    int64_t deviceBytes,
     int64_t numPackedCols,
     std::vector<ContinuePromise>& promises) {
   updateTotalQueuedBytesMsLocked();
 
   queuedBytes_ -= bytes;
+  deviceQueuedBytes_ -= deviceBytes;
   queuedPackedColumns_ -= numPackedCols;
 
   VELOX_CHECK_GE(queuedBytes_, 0);
+  VELOX_CHECK_GE(deviceQueuedBytes_, 0);
   VELOX_CHECK_GE(queuedPackedColumns_, 0);
   updateDiagnosticGlobalQueue(-bytes, -numPackedCols, "dequeue", task_);
   logDeviceQueueResidencyLocked("dequeue");
 
   // Check whether queue is below low-water mark and return outstanding
   // promises
-  if (queuedBytes_ <= continueSize_ && !promises_.empty()) {
+  const bool belowDeviceLowWater = deviceQueuedBytes_ <= continueSize_;
+  const bool belowHostLowWater =
+      !hostSpooling_ || queuedBytes_ <= hostSpoolContinueSize_;
+  if (belowDeviceLowWater && belowHostLowWater && !promises_.empty()) {
     VLOG(2) << "[BACKPRESSURE] task=" << (task_ ? task_->taskId() : "n/a")
             << " UNBLOCKING " << promises_.size() << " producers"
             << " queuedBytes=" << queuedBytes_
-            << " continueSize=" << continueSize_;
+            << " hostContinueSize=" << hostSpoolContinueSize_
+            << " deviceQueuedBytes=" << deviceQueuedBytes_
+            << " deviceContinueSize=" << continueSize_;
     promises = std::move(promises_);
   }
 }
@@ -803,6 +960,7 @@ void UcxOutputQueue::logDeviceQueueResidencyLocked(const char* event) {
   LOG(WARNING) << "CUDF_DEVICE_QUEUE event=" << event
                << " task=" << (task_ ? task_->taskId() : "n/a")
                << " queuedBytes=" << queuedBytes_
+               << " deviceQueuedBytes=" << deviceQueuedBytes_
                << " queuedPackedColumns=" << queuedPackedColumns_
                << " maxSize=" << maxSize_ << " continueSize=" << continueSize_;
 }
diff --git a/velox/experimental/ucx-exchange/UcxQueues.h b/velox/experimental/ucx-exchange/UcxQueues.h
index a00d0a48c36..6da7d722269 100644
--- a/velox/experimental/ucx-exchange/UcxQueues.h
+++ b/velox/experimental/ucx-exchange/UcxQueues.h
@@ -27,6 +27,36 @@
 
 namespace facebook::velox::ucx_exchange {
 
+/// A packed table waiting in a producer output queue. The normal path keeps
+/// the original device-resident packed_columns. Phase-separated exchanges may
+/// instead keep the contiguous payload in host memory until its destination is
+/// active.
+struct UcxTransferData {
+  std::shared_ptr<cudf::packed_columns> deviceData;
+  std::shared_ptr<std::vector<uint8_t>> hostData;
+  std::shared_ptr<std::vector<uint8_t>> metadata;
+
+  int64_t size() const {
+    if (deviceData != nullptr) {
+      return deviceData->gpu_data->size();
+    }
+    return hostData == nullptr ? 0 : hostData->size();
+  }
+
+  int64_t hostResidentSize() const {
+    return (hostData == nullptr ? 0 : hostData->size()) +
+        (metadata == nullptr ? 0 : metadata->size());
+  }
+
+  int64_t backpressureSize() const {
+    return isHostStaged() ? hostResidentSize() : size();
+  }
+
+  bool isHostStaged() const {
+    return hostData != nullptr;
+  }
+};
+
 /// @brief  Callback function for getting data from the queues.
 /// A nullptr indicates that there is no more data.
 /// The remainingBytes vector contains the sizes for the
@@ -42,18 +72,20 @@ using UcxDataAvailableCallbackV2 = std::function<void(
     int64_t sequence,
     std::vector<int64_t> remainingBytes)>;
 
+using UcxTransferDataAvailableCallback = std::function<void(
+    std::shared_ptr<UcxTransferData> data,
+    int64_t sequence,
+    std::vector<int64_t> remainingBytes)>;
+
 struct UcxDataAvailable {
-  UcxDataAvailableCallback callback{nullptr};
-  UcxDataAvailableCallbackV2 callbackV2{nullptr};
-  std::shared_ptr<cudf::packed_columns> data;
+  UcxTransferDataAvailableCallback callback{nullptr};
+  std::shared_ptr<UcxTransferData> data;
   int64_t sequence{0};
   std::vector<int64_t> remainingBytes;
 
   void notify() {
-    if (callbackV2) {
-      callbackV2(std::move(data), sequence, remainingBytes);
-    } else if (callback) {
-      callback(std::move(data), remainingBytes);
+    if (callback) {
+      callback(std::move(data), sequence, remainingBytes);
     }
   }
 };
@@ -67,12 +99,14 @@ struct UcxDataAvailable {
 class UcxDestinationQueue {
  public:
   struct Stats {
-    void recordEnqueue(const cudf::packed_columns* data);
+    void recordEnqueue(const UcxTransferData* data);
 
-    void recordDequeue(const cudf::packed_columns* data);
+    void recordDequeue(const UcxTransferData* data);
 
     // what has been queued
     int64_t bytesQueued{0};
+    int64_t backpressureBytesQueued{0};
+    int64_t deviceBytesQueued{0};
     int64_t packedColumnsQueued{0};
 
     // what has been dequeued
@@ -82,15 +116,15 @@ class UcxDestinationQueue {
 
   /// @brief Enqueues the data to the back of the queue.
   /// @param data Corresponds to a RowVector
-  void enqueueBack(std::shared_ptr<cudf::packed_columns> data);
+  void enqueueBack(std::shared_ptr<UcxTransferData> data);
 
   /// @brief Enqueues the data to the front of the queue. This is needed when
   /// a transfer fails.
   /// @param data
-  void enqueueFront(std::shared_ptr<cudf::packed_columns> data);
+  void enqueueFront(std::shared_ptr<UcxTransferData> data);
 
   struct Data {
-    std::shared_ptr<cudf::packed_columns> data;
+    std::shared_ptr<UcxTransferData> data;
     int64_t sequence{0};
     std::vector<int64_t> remainingBytes;
     /// Whether the result is returned immediately without invoking the `notify'
@@ -102,12 +136,12 @@ class UcxDestinationQueue {
   /// ownership to the caller. If there is no data, 'notify' is installed and it
   /// will be called when data becomes available. In this case, a nullptr is
   /// returned.
-  [[nodiscard]] Data getData(UcxDataAvailableCallback notify);
+  [[nodiscard]] Data getData(UcxTransferDataAvailableCallback notify);
 
   [[nodiscard]] Data getData(
       uint64_t maxBytes,
       int64_t sequence,
-      UcxDataAvailableCallbackV2 notify);
+      UcxTransferDataAvailableCallback notify);
 
   /// Removes all remaining data from the queue and returns any pending waiter
   /// so it can be woken with an end marker.
@@ -128,9 +162,8 @@ class UcxDestinationQueue {
  private:
   void clearNotify();
 
-  std::deque<std::shared_ptr<cudf::packed_columns>> queue_;
-  UcxDataAvailableCallback notify_{nullptr};
-  UcxDataAvailableCallbackV2 notifyV2_{nullptr};
+  std::deque<std::shared_ptr<UcxTransferData>> queue_;
+  UcxTransferDataAvailableCallback notify_{nullptr};
   int64_t sequence_{0};
   int64_t notifySequence_{0};
   uint64_t notifyMaxBytes_{0};
@@ -204,6 +237,10 @@ class UcxOutputQueue : public std::enable_shared_from_this<UcxOutputQueue> {
       std::unique_ptr<cudf::packed_columns> data,
       int32_t numRows);
 
+  /// Materializes subsequently enqueued partitioned payloads in host memory.
+  /// This must be enabled before the producer task starts.
+  void enableHostSpooling();
+
   /// @brief Checks if the queue is over capacity and returns a future if so.
   /// This should be called after enqueueing all partitions for a batch.
   /// @param future Output parameter - populated with a future if blocked.
@@ -222,6 +259,12 @@ class UcxOutputQueue : public std::enable_shared_from_this<UcxOutputQueue> {
       int64_t sequence,
       UcxDataAvailableCallbackV2 notify);
 
+  void getTransferData(
+      int destination,
+      uint64_t maxBytes,
+      int64_t sequence,
+      UcxTransferDataAvailableCallback notify);
+
   /// @brief Indicates that a driver is done and won't enqueue any more data.
   void noMoreData();
 
@@ -260,13 +303,18 @@ class UcxOutputQueue : public std::enable_shared_from_this<UcxOutputQueue> {
   static constexpr int32_t kContinuePct = 90;
 
   // Methods that update the statistics.
-  void updateStatsWithEnqueuedLocked(int64_t bytes, int64_t rows);
+  void updateStatsWithEnqueuedLocked(
+      int64_t logicalBytes,
+      int64_t backpressureBytes,
+      int64_t deviceBytes,
+      int64_t rows);
 
   // updates the counters and returns promises if the queuedBytes_ counter falls
   // below the continueSize_ low water mark. These promises then need to be
   // realized outside the lock.
   void updateStatsWithFreedLocked(
       int64_t bytes,
+      int64_t deviceBytes,
       int64_t numPackedCols,
       std::vector<ContinuePromise>& promises);
 
@@ -288,7 +336,7 @@ class UcxOutputQueue : public std::enable_shared_from_this<UcxOutputQueue> {
 
   bool enqueuePartitionedOutputLocked(
       int destination,
-      std::shared_ptr<cudf::packed_columns> data,
+      std::shared_ptr<UcxTransferData> data,
       std::vector<UcxDataAvailable>& dataAvailableCbs);
 
   void enqueueBroadcastOutputLocked(
@@ -311,6 +359,14 @@ class UcxOutputQueue : public std::enable_shared_from_this<UcxOutputQueue> {
   // backfill. Cleared once noMoreQueues_ is set.
   std::vector<std::shared_ptr<cudf::packed_columns>> dataToBroadcast_;
 
+  bool hostSpooling_{false};
+  int64_t hostSpooledBytes_{0};
+  int64_t hostSpoolPeakBytes_{0};
+  int64_t hostSpoolResidentBytes_{0};
+  int64_t hostSpoolPeakResidentBytes_{0};
+  uint64_t hostSpoolMaxSize_{0};
+  uint64_t hostSpoolContinueSize_{0};
+
   /// If 'queuedBytes_' > 'maxSize_', each producer is blocked after adding
   /// data.
   uint64_t maxSize_{0};
@@ -343,6 +399,7 @@ class UcxOutputQueue : public std::enable_shared_from_this<UcxOutputQueue> {
 
   // actual data in 'queues_'
   int64_t queuedBytes_{0};
+  int64_t deviceQueuedBytes_{0};
   int64_t queuedPackedColumns_{0};
 
   // Last reported 256 MiB bucket. Diagnostic-only; it does not cap the queue.
diff --git a/velox/experimental/ucx-exchange/tests/UcxOutputQueueManagerTest.cpp b/velox/experimental/ucx-exchange/tests/UcxOutputQueueManagerTest.cpp
index 1b3ae98b03a..cd28488cd83 100644
--- a/velox/experimental/ucx-exchange/tests/UcxOutputQueueManagerTest.cpp
+++ b/velox/experimental/ucx-exchange/tests/UcxOutputQueueManagerTest.cpp
@@ -14,6 +14,7 @@
  * limitations under the License.
  */
 #include "velox/experimental/ucx-exchange/UcxOutputQueueManager.h"
+
 #include <cudf/column/column_factories.hpp>
 #include <cudf/contiguous_split.hpp>
 #include <cudf/table/table.hpp>
@@ -22,8 +23,11 @@
 #include <folly/synchronization/EventCount.h>
 #include <gtest/gtest.h>
 #include <rmm/device_buffer.hpp>
+#include <cstdlib>
 #include <limits>
 #include <memory>
+#include <optional>
+#include <string>
 #include <vector>
 #include "velox/common/memory/MemoryPool.h"
 #include "velox/exec/tests/utils/PlanBuilder.h"
@@ -35,6 +39,30 @@ using namespace facebook::velox;
 using namespace facebook::velox::exec;
 using namespace facebook::velox::core;
 
+namespace {
+class ScopedEnvironment {
+ public:
+  ScopedEnvironment(const char* name, const char* value) : name_(name) {
+    if (const auto* previous = std::getenv(name)) {
+      previous_ = previous;
+    }
+    setenv(name, value, 1);
+  }
+
+  ~ScopedEnvironment() {
+    if (previous_.has_value()) {
+      setenv(name_.c_str(), previous_->c_str(), 1);
+    } else {
+      unsetenv(name_.c_str());
+    }
+  }
+
+ private:
+  std::string name_;
+  std::optional<std::string> previous_;
+};
+} // namespace
+
 class UcxOutputQueueManagerTest : public testing::Test {
  protected:
   UcxOutputQueueManagerTest() {}
@@ -54,12 +82,14 @@ class UcxOutputQueueManagerTest : public testing::Test {
       int numDrivers,
       bool cleanup = true,
       core::PartitionedOutputNode::Kind kind =
-          core::PartitionedOutputNode::Kind::kPartitioned) {
+          core::PartitionedOutputNode::Kind::kPartitioned,
+      uint64_t maxOutputBufferSize = FOUR_GBYTES) {
     if (cleanup) {
       queueManager_->removeTask(taskId);
     }
 
-    auto task = createSourceTask(taskId, pool_, UcxTestData::kTestRowType);
+    auto task = createSourceTask(
+        taskId, pool_, UcxTestData::kTestRowType, maxOutputBufferSize);
 
     queueManager_->initializeTask(task, kind, numDestinations, numDrivers);
     return task;
@@ -341,6 +371,172 @@ TEST_F(UcxOutputQueueManagerTest, basicPartitioned) {
   EXPECT_TRUE(task->isFinished());
 }
 
+TEST_F(UcxOutputQueueManagerTest, hostSpoolingReleasesQueuedDevicePayload) {
+  const std::string taskId = "hostSpooling";
+  constexpr int32_t numRows = 100;
+  queueManager_->removeTask(taskId);
+  queueManager_->enableHostSpooling(taskId);
+  auto task = initializeTask(
+      taskId, 2 /* numDestinations*/, 1 /*numDrivers*/, false /*cleanup*/);
+
+  auto packed = makePackedColumns(numRows);
+  const auto expectedBytes = packed->gpu_data->size();
+  queueManager_->enqueue(taskId, 1, std::move(packed), numRows);
+
+  bool received = false;
+  queueManager_->getTransferData(
+      taskId,
+      1,
+      std::numeric_limits<uint64_t>::max(),
+      0,
+      [&](std::shared_ptr<UcxTransferData> data,
+          int64_t sequence,
+          std::vector<int64_t> remainingBytes) {
+        ASSERT_NE(data, nullptr);
+        EXPECT_EQ(sequence, 0);
+        EXPECT_TRUE(remainingBytes.empty());
+        EXPECT_TRUE(data->isHostStaged());
+        EXPECT_EQ(data->deviceData, nullptr);
+        ASSERT_NE(data->hostData, nullptr);
+        ASSERT_NE(data->metadata, nullptr);
+        EXPECT_EQ(data->size(), expectedBytes);
+
+        auto deviceBuffer = std::make_unique<rmm::device_buffer>(
+            expectedBytes, rmm::cuda_stream_default);
+        CUDF_CUDA_TRY(cudaMemcpy(
+            deviceBuffer->data(),
+            data->hostData->data(),
+            expectedBytes,
+            cudaMemcpyHostToDevice));
+        auto metadata = std::make_unique<std::vector<uint8_t>>(*data->metadata);
+        cudf::packed_columns restored(
+            std::move(metadata), std::move(deviceBuffer));
+        EXPECT_EQ(cudf::unpack(restored).num_rows(), numRows);
+        received = true;
+      });
+  EXPECT_TRUE(received);
+
+  noMoreData(taskId);
+  fetchEndMarker(taskId, 0);
+  fetchEndMarker(taskId, 1);
+  queueManager_->removeTask(taskId);
+  EXPECT_TRUE(task->isFinished());
+}
+
+TEST_F(UcxOutputQueueManagerTest, hostSpoolingSupportsCapacityOverride) {
+  ScopedEnvironment maxBytes{"GLUTEN_UCX_HOST_SPOOL_MAX_BYTES", "1073741824"};
+  const std::string taskId = "hostSpoolingCapacityOverride";
+  constexpr int32_t numRows = 1'000;
+  auto packed = makePackedColumns(numRows);
+  const auto logicalBytes = packed->gpu_data->size();
+  const auto maxOutputBufferSize = logicalBytes - 1;
+
+  queueManager_->removeTask(taskId);
+  queueManager_->enableHostSpooling(taskId);
+  auto task = initializeTask(
+      taskId,
+      1,
+      1,
+      false,
+      core::PartitionedOutputNode::Kind::kPartitioned,
+      maxOutputBufferSize);
+  queueManager_->enqueue(taskId, 0, std::move(packed), numRows);
+
+  ContinueFuture future;
+  EXPECT_FALSE(queueManager_->checkBlocked(taskId, &future));
+  bool received = false;
+  queueManager_->getTransferData(
+      taskId,
+      0,
+      std::numeric_limits<uint64_t>::max(),
+      0,
+      [&](std::shared_ptr<UcxTransferData> data,
+          int64_t sequence,
+          std::vector<int64_t> remainingBytes) {
+        ASSERT_NE(data, nullptr);
+        EXPECT_EQ(sequence, 0);
+        EXPECT_TRUE(remainingBytes.empty());
+        EXPECT_EQ(data->hostData->size(), logicalBytes);
+        received = true;
+      });
+  EXPECT_TRUE(received);
+
+  noMoreData(taskId);
+  fetchEndMarker(taskId, 0);
+  queueManager_->removeTask(taskId);
+  EXPECT_TRUE(task->isFinished());
+}
+
+TEST_F(UcxOutputQueueManagerTest, hostSpoolingBackpressureUsesResidentBytes) {
+  ScopedEnvironment maxBytes{"GLUTEN_UCX_HOST_SPOOL_MAX_BYTES", "1073741824"};
+  const std::string hostTaskId = "hostSpoolingBackpressure";
+  constexpr int32_t numRows = 1000;
+  auto packed = makePackedColumns(numRows);
+  const auto logicalBytes = packed->gpu_data->size();
+  const auto maxOutputBufferSize = logicalBytes - 1;
+
+  queueManager_->removeTask(hostTaskId);
+  queueManager_->enableHostSpooling(hostTaskId);
+  auto hostTask = initializeTask(
+      hostTaskId,
+      1,
+      1,
+      false,
+      core::PartitionedOutputNode::Kind::kPartitioned,
+      maxOutputBufferSize);
+  queueManager_->enqueue(hostTaskId, 0, std::move(packed), numRows);
+
+  ContinueFuture hostFuture;
+  EXPECT_FALSE(queueManager_->checkBlocked(hostTaskId, &hostFuture));
+  auto hostStats = queueManager_->stats(hostTaskId);
+  ASSERT_TRUE(hostStats.has_value());
+  EXPECT_GT(hostStats->bufferedBytes, logicalBytes);
+  EXPECT_EQ(hostStats->totalBytesSent, logicalBytes);
+
+  bool received = false;
+  std::shared_ptr<UcxTransferData> response;
+  queueManager_->getTransferData(
+      hostTaskId,
+      0,
+      std::numeric_limits<uint64_t>::max(),
+      0,
+      [&](std::shared_ptr<UcxTransferData> data,
+          int64_t sequence,
+          std::vector<int64_t> remainingBytes) {
+        received = true;
+        EXPECT_EQ(sequence, 0);
+        EXPECT_TRUE(remainingBytes.empty());
+        response = std::move(data);
+      });
+  EXPECT_TRUE(received);
+  ASSERT_NE(response, nullptr);
+  EXPECT_TRUE(response->isHostStaged());
+  EXPECT_EQ(response->size(), logicalBytes);
+  EXPECT_EQ(response->backpressureSize(), hostStats->bufferedBytes);
+  noMoreData(hostTaskId);
+  fetchEndMarker(hostTaskId, 0);
+  queueManager_->removeTask(hostTaskId);
+  EXPECT_TRUE(hostTask->isFinished());
+
+  const std::string deviceTaskId = "deviceBackpressure";
+  auto deviceTask = initializeTask(
+      deviceTaskId,
+      1,
+      1,
+      true,
+      core::PartitionedOutputNode::Kind::kPartitioned,
+      maxOutputBufferSize);
+  queueManager_->enqueue(deviceTaskId, 0, makePackedColumns(numRows), numRows);
+  ContinueFuture deviceFuture;
+  EXPECT_TRUE(queueManager_->checkBlocked(deviceTaskId, &deviceFuture));
+  fetch(deviceTaskId, 0);
+  deviceFuture.wait();
+  noMoreData(deviceTaskId);
+  fetchEndMarker(deviceTaskId, 0);
+  queueManager_->removeTask(deviceTaskId);
+  EXPECT_TRUE(deviceTask->isFinished());
+}
+
 TEST_F(UcxOutputQueueManagerTest, v1RepeatedFetchAdvancesQueue) {
   const std::string taskId = "v1RepeatedFetch";
   const int destination = 0;
@@ -846,6 +1042,39 @@ TEST_F(UcxOutputQueueManagerTest, broadcastBasic) {
   queueManager_->removeTask(taskId);
 }
 
+TEST_F(UcxOutputQueueManagerTest, broadcastDeviceBackpressureAccounting) {
+  const vector_size_t size = 100;
+  const std::string taskId = "broadcastDeviceBackpressure";
+  constexpr int numDestinations = 3;
+  auto packed = makePackedColumns(size);
+  const auto bytes = packed->gpu_data->size();
+
+  auto task = initializeTask(
+      taskId,
+      numDestinations,
+      1 /* numDrivers */,
+      true /* cleanup */,
+      core::PartitionedOutputNode::Kind::kBroadcast,
+      bytes * 2);
+  queueManager_->updateOutputBuffers(taskId, numDestinations, true);
+  queueManager_->enqueue(taskId, 0, std::move(packed), size);
+
+  ContinueFuture future;
+  EXPECT_TRUE(queueManager_->checkBlocked(taskId, &future));
+  fetch(taskId, 0);
+  EXPECT_FALSE(future.isReady());
+  fetch(taskId, 1);
+  future.wait();
+  fetch(taskId, 2);
+
+  noMoreData(taskId);
+  for (int destination = 0; destination < numDestinations; ++destination) {
+    fetchEndMarker(taskId, destination);
+  }
+  queueManager_->removeTask(taskId);
+  EXPECT_TRUE(task->isFinished());
+}
+
 // Broadcast: late destination receives backfilled data.
 TEST_F(UcxOutputQueueManagerTest, broadcastLateDestination) {
   const vector_size_t size = 50;