This is an automated email from the ASF dual-hosted git repository.
hongze pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-gluten.git
The following commit(s) were added to refs/heads/main by this push:
new 80078cd4c [GLUTEN-6908][VL] Fix error when getting output from a Velox
task that is under spilling by background thread (#6934)
80078cd4c is described below
commit 80078cd4c8b36db39128f93c666af1ee2c8d94dd
Author: Hongze Zhang <[email protected]>
AuthorDate: Wed Aug 21 16:54:07 2024 +0800
[GLUTEN-6908][VL] Fix error when getting output from a Velox task that is
under spilling by background thread (#6934)
---
cpp/velox/compute/WholeStageResultIterator.cc | 17 ++++++++++++++++-
1 file changed, 16 insertions(+), 1 deletion(-)
diff --git a/cpp/velox/compute/WholeStageResultIterator.cc
b/cpp/velox/compute/WholeStageResultIterator.cc
index 34d0d2db0..2edf9a573 100644
--- a/cpp/velox/compute/WholeStageResultIterator.cc
+++ b/cpp/velox/compute/WholeStageResultIterator.cc
@@ -195,7 +195,22 @@ std::shared_ptr<ColumnarBatch>
WholeStageResultIterator::next() {
if (task_->isFinished()) {
return nullptr;
}
- velox::RowVectorPtr vector = task_->next();
+ velox::RowVectorPtr vector;
+ while (true) {
+ auto future = velox::ContinueFuture::makeEmpty();
+ auto out = task_->next(&future);
+ if (!future.valid()) {
+ // Not need to wait. Break.
+ vector = std::move(out);
+ break;
+ }
+ // Velox suggested to wait. This might be because another thread (e.g.,
background io thread) is spilling the task.
+ GLUTEN_CHECK(out == nullptr, "Expected to wait but still got non-null
output from Velox task");
+ VLOG(2) << "Velox task " << task_->taskId()
+ << " is busy when ::next() is called. Will wait and try again.
Task state: "
+ << taskStateString(task_->state());
+ future.wait();
+ }
if (vector == nullptr) {
return nullptr;
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
