You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by al...@apache.org on 2025/07/30 21:04:19 UTC
(datafusion) branch main updated: Use tokio::task::coop::poll_proceed by default in CooperativeStream (#16748)
This is an automated email from the ASF dual-hosted git repository.
alamb pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new 3dc3aaa86e Use tokio::task::coop::poll_proceed by default in CooperativeStream (#16748)
3dc3aaa86e is described below
commit 3dc3aaa86e8ed815fcc0619e9bd28966633093f3
Author: Pepijn Van Eeckhoudt <pe...@vaneeckhoudt.net>
AuthorDate: Wed Jul 30 23:04:15 2025 +0200
Use tokio::task::coop::poll_proceed by default in CooperativeStream (#16748)
Co-authored-by: Andrew Lamb <an...@nerdnetworks.org>
---
datafusion/common-runtime/Cargo.toml | 2 +-
datafusion/physical-plan/src/coop.rs | 20 +++++++++-----------
2 files changed, 10 insertions(+), 12 deletions(-)
diff --git a/datafusion/common-runtime/Cargo.toml b/datafusion/common-runtime/Cargo.toml
index 0b987bdea6..e53d97b413 100644
--- a/datafusion/common-runtime/Cargo.toml
+++ b/datafusion/common-runtime/Cargo.toml
@@ -43,4 +43,4 @@ log = { workspace = true }
tokio = { workspace = true }
[dev-dependencies]
-tokio = { version = "1.47", features = ["rt", "rt-multi-thread", "time"] }
+tokio = { workspace = true, features = ["rt", "rt-multi-thread", "time"] }
diff --git a/datafusion/physical-plan/src/coop.rs b/datafusion/physical-plan/src/coop.rs
index be0afa07ea..777fd7d9d6 100644
--- a/datafusion/physical-plan/src/coop.rs
+++ b/datafusion/physical-plan/src/coop.rs
@@ -65,10 +65,7 @@
//! The optimizer rule currently checks the plan for exchange-like operators and leave operators
//! that report [`SchedulingType::NonCooperative`] in their [plan properties](ExecutionPlan::properties).
-#[cfg(any(
- datafusion_coop = "tokio_fallback",
- not(any(datafusion_coop = "tokio", datafusion_coop = "per_stream"))
-))]
+#[cfg(datafusion_coop = "tokio_fallback")]
use futures::Future;
use std::any::Any;
use std::pin::Pin;
@@ -133,10 +130,14 @@ where
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Self::Item>> {
- #[cfg(datafusion_coop = "tokio")]
+ #[cfg(any(
+ datafusion_coop = "tokio",
+ not(any(
+ datafusion_coop = "tokio_fallback",
+ datafusion_coop = "per_stream"
+ ))
+ ))]
{
- // TODO this should be the default implementation
- // Enable once https://github.com/tokio-rs/tokio/issues/7403 is merged and released
let coop = std::task::ready!(tokio::task::coop::poll_proceed(cx));
let value = self.inner.poll_next_unpin(cx);
if value.is_ready() {
@@ -145,10 +146,7 @@ where
value
}
- #[cfg(any(
- datafusion_coop = "tokio_fallback",
- not(any(datafusion_coop = "tokio", datafusion_coop = "per_stream"))
- ))]
+ #[cfg(datafusion_coop = "tokio_fallback")]
{
// This is a temporary placeholder implementation that may have slightly
// worse performance compared to `poll_proceed`
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org