From be38639f3023e2cbf7c889f6774f5b50112766f8 Mon Sep 17 00:00:00 2001
From: "Carol (Nichols || Goulding)" <carol.nichols@gmail.com>
Date: Wed, 5 Mar 2025 10:20:51 -0500
Subject: [PATCH 01/10] Document guidelines for physical operator yielding

To start a policy of the behavior physical operator streams should have
and drive improvements in this area to allow for timely cancellation.

Connects to #14036 and related to pull requests such as #14028.
---
 datafusion/physical-plan/src/execution_plan.rs | 17 +++++++++++++++++
 1 file changed, 17 insertions(+)

diff --git a/datafusion/physical-plan/src/execution_plan.rs b/datafusion/physical-plan/src/execution_plan.rs
index 851e504b69af..b595e198d328 100644
--- a/datafusion/physical-plan/src/execution_plan.rs
+++ b/datafusion/physical-plan/src/execution_plan.rs
@@ -260,6 +260,20 @@ pub trait ExecutionPlan: Debug + DisplayAs + Send + Sync {
     /// used.
     /// Thus, [`spawn`] is disallowed, and instead use [`SpawnedTask`].
     ///
+    /// To enable timely cancellation, the [`Stream`] that is returned must not
+    /// pin the CPU and must yield back to the tokio runtime regularly. This can
+    /// be achieved by manually returning [`Poll::Pending`] in regular intervals,
+    /// or the use of [`tokio::task::yield_now()`]. Cooperative scheduling may also
+    /// be a way to achieve this goal, as [tokio support for it improves][coop].
+    /// Determination for "regularly" may be made using a timer (being careful with
+    /// the overhead-heavy syscall needed to take the time) or by counting rows or
+    /// batches.
+    ///
+    /// The goal is for `datafusion`-provided operator implementation to
+    /// strive for [the guideline of not spending a long time without reaching
+    /// an `await`/yield point][async-guideline]. Progress towards this goal
+    /// is tracked partially by the cancellation benchmark.
+    ///
     /// For more details see [`SpawnedTask`], [`JoinSet`] and [`RecordBatchReceiverStreamBuilder`]
     /// for structures to help ensure all background tasks are cancelled.
     ///
@@ -267,6 +281,9 @@ pub trait ExecutionPlan: Debug + DisplayAs + Send + Sync {
     /// [`JoinSet`]: tokio::task::JoinSet
     /// [`SpawnedTask`]: datafusion_common_runtime::SpawnedTask
     /// [`RecordBatchReceiverStreamBuilder`]: crate::stream::RecordBatchReceiverStreamBuilder
+    /// [`Poll::Pending`]: std::task::Poll::Pending
+    /// [coop]: https://github.com/tokio-rs/tokio/pull/7116
+    /// [async-guideline]: https://ryhl.io/blog/async-what-is-blocking/
     ///
     /// # Implementation Examples
     ///

From 14560dfd6c132ef50a1df0684ab8d8ccbbf436dc Mon Sep 17 00:00:00 2001
From: "Carol (Nichols || Goulding)" <carol.nichols@gmail.com>
Date: Thu, 6 Mar 2025 14:27:33 -0500
Subject: [PATCH 02/10] Remove discussion of tokio coop

---
 datafusion/physical-plan/src/execution_plan.rs | 9 +++------
 1 file changed, 3 insertions(+), 6 deletions(-)

diff --git a/datafusion/physical-plan/src/execution_plan.rs b/datafusion/physical-plan/src/execution_plan.rs
index b595e198d328..1800f1433850 100644
--- a/datafusion/physical-plan/src/execution_plan.rs
+++ b/datafusion/physical-plan/src/execution_plan.rs
@@ -263,11 +263,9 @@ pub trait ExecutionPlan: Debug + DisplayAs + Send + Sync {
     /// To enable timely cancellation, the [`Stream`] that is returned must not
     /// pin the CPU and must yield back to the tokio runtime regularly. This can
     /// be achieved by manually returning [`Poll::Pending`] in regular intervals,
-    /// or the use of [`tokio::task::yield_now()`]. Cooperative scheduling may also
-    /// be a way to achieve this goal, as [tokio support for it improves][coop].
-    /// Determination for "regularly" may be made using a timer (being careful with
-    /// the overhead-heavy syscall needed to take the time) or by counting rows or
-    /// batches.
+    /// or the use of [`tokio::task::yield_now()`]. Determination for "regularly"
+    /// may be made using a timer (being careful with the overhead-heavy syscall
+    /// needed to take the time) or by counting rows or batches.
     ///
     /// The goal is for `datafusion`-provided operator implementation to
     /// strive for [the guideline of not spending a long time without reaching
@@ -282,7 +280,6 @@ pub trait ExecutionPlan: Debug + DisplayAs + Send + Sync {
     /// [`SpawnedTask`]: datafusion_common_runtime::SpawnedTask
     /// [`RecordBatchReceiverStreamBuilder`]: crate::stream::RecordBatchReceiverStreamBuilder
     /// [`Poll::Pending`]: std::task::Poll::Pending
-    /// [coop]: https://github.com/tokio-rs/tokio/pull/7116
     /// [async-guideline]: https://ryhl.io/blog/async-what-is-blocking/
     ///
     /// # Implementation Examples

From 5cf7332e93b80b5709d8e2733f058ceb54fd5e39 Mon Sep 17 00:00:00 2001
From: "Carol (Nichols || Goulding)" <carol.nichols@gmail.com>
Date: Thu, 6 Mar 2025 14:33:01 -0500
Subject: [PATCH 03/10] Move rationale up

---
 .../physical-plan/src/execution_plan.rs       | 19 ++++++++++---------
 1 file changed, 10 insertions(+), 9 deletions(-)

diff --git a/datafusion/physical-plan/src/execution_plan.rs b/datafusion/physical-plan/src/execution_plan.rs
index 1800f1433850..ccd10f171749 100644
--- a/datafusion/physical-plan/src/execution_plan.rs
+++ b/datafusion/physical-plan/src/execution_plan.rs
@@ -261,16 +261,17 @@ pub trait ExecutionPlan: Debug + DisplayAs + Send + Sync {
     /// Thus, [`spawn`] is disallowed, and instead use [`SpawnedTask`].
     ///
     /// To enable timely cancellation, the [`Stream`] that is returned must not
-    /// pin the CPU and must yield back to the tokio runtime regularly. This can
-    /// be achieved by manually returning [`Poll::Pending`] in regular intervals,
-    /// or the use of [`tokio::task::yield_now()`]. Determination for "regularly"
-    /// may be made using a timer (being careful with the overhead-heavy syscall
-    /// needed to take the time) or by counting rows or batches.
+    /// pin the CPU and must yield back to the tokio runtime regularly.
+    /// [`ExecutionPlan`] implementations should follow [the guideline of not
+    /// spending a long time without reaching an `await`/yield point][async-guideline].
+    /// This can be achieved by manually returning [`Poll::Pending`] in regular
+    /// intervals, or the use of [`tokio::task::yield_now()`]. Determination
+    /// for "regularly" may be made using a timer (being careful with the
+    /// overhead-heavy syscall needed to take the time) or by counting rows
+    /// or batches.
     ///
-    /// The goal is for `datafusion`-provided operator implementation to
-    /// strive for [the guideline of not spending a long time without reaching
-    /// an `await`/yield point][async-guideline]. Progress towards this goal
-    /// is tracked partially by the cancellation benchmark.
+    /// Progress towards this goal is tracked partially by the cancellation
+    /// benchmark.
     ///
     /// For more details see [`SpawnedTask`], [`JoinSet`] and [`RecordBatchReceiverStreamBuilder`]
     /// for structures to help ensure all background tasks are cancelled.

From 3d1257e3f95c5b53acb3b9fa9460f4ea420496b6 Mon Sep 17 00:00:00 2001
From: "Carol (Nichols || Goulding)" <carol.nichols@gmail.com>
Date: Thu, 6 Mar 2025 14:34:56 -0500
Subject: [PATCH 04/10] TODO ADD LINK

---
 datafusion/physical-plan/src/execution_plan.rs | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/datafusion/physical-plan/src/execution_plan.rs b/datafusion/physical-plan/src/execution_plan.rs
index ccd10f171749..6f9fa9192e22 100644
--- a/datafusion/physical-plan/src/execution_plan.rs
+++ b/datafusion/physical-plan/src/execution_plan.rs
@@ -270,8 +270,8 @@ pub trait ExecutionPlan: Debug + DisplayAs + Send + Sync {
     /// overhead-heavy syscall needed to take the time) or by counting rows
     /// or batches.
     ///
-    /// Progress towards this goal is tracked partially by the cancellation
-    /// benchmark.
+    /// The cancellation benchmark tracks some cases of how quickly queries can
+    /// be cancelled.
     ///
     /// For more details see [`SpawnedTask`], [`JoinSet`] and [`RecordBatchReceiverStreamBuilder`]
     /// for structures to help ensure all background tasks are cancelled.

From d4c1e1489e67a2f6ad8e4fa2e6ed7390c9fd8aad Mon Sep 17 00:00:00 2001
From: "Carol (Nichols || Goulding)" <carol.nichols@gmail.com>
Date: Thu, 6 Mar 2025 14:49:58 -0500
Subject: [PATCH 05/10] Say block the CPU rather than pin the CPU

---
 datafusion/physical-plan/src/execution_plan.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/datafusion/physical-plan/src/execution_plan.rs b/datafusion/physical-plan/src/execution_plan.rs
index 6f9fa9192e22..087754faaf96 100644
--- a/datafusion/physical-plan/src/execution_plan.rs
+++ b/datafusion/physical-plan/src/execution_plan.rs
@@ -261,7 +261,7 @@ pub trait ExecutionPlan: Debug + DisplayAs + Send + Sync {
     /// Thus, [`spawn`] is disallowed, and instead use [`SpawnedTask`].
     ///
     /// To enable timely cancellation, the [`Stream`] that is returned must not
-    /// pin the CPU and must yield back to the tokio runtime regularly.
+    /// block the CPU and must yield back to the tokio runtime regularly.
     /// [`ExecutionPlan`] implementations should follow [the guideline of not
     /// spending a long time without reaching an `await`/yield point][async-guideline].
     /// This can be achieved by manually returning [`Poll::Pending`] in regular

From 286f62e62d484b9f729031aaa6fcd4e5e1a7abab Mon Sep 17 00:00:00 2001
From: "Carol (Nichols || Goulding)" <carol.nichols@gmail.com>
Date: Thu, 6 Mar 2025 14:51:11 -0500
Subject: [PATCH 06/10] Add a caveat to use the right tool for the situation

---
 datafusion/physical-plan/src/execution_plan.rs | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/datafusion/physical-plan/src/execution_plan.rs b/datafusion/physical-plan/src/execution_plan.rs
index 087754faaf96..292673135555 100644
--- a/datafusion/physical-plan/src/execution_plan.rs
+++ b/datafusion/physical-plan/src/execution_plan.rs
@@ -265,10 +265,10 @@ pub trait ExecutionPlan: Debug + DisplayAs + Send + Sync {
     /// [`ExecutionPlan`] implementations should follow [the guideline of not
     /// spending a long time without reaching an `await`/yield point][async-guideline].
     /// This can be achieved by manually returning [`Poll::Pending`] in regular
-    /// intervals, or the use of [`tokio::task::yield_now()`]. Determination
-    /// for "regularly" may be made using a timer (being careful with the
-    /// overhead-heavy syscall needed to take the time) or by counting rows
-    /// or batches.
+    /// intervals, or the use of [`tokio::task::yield_now()`] (as appropriate).
+    /// Determination for "regularly" may be made using a timer (being careful
+    /// with the  overhead-heavy syscall needed to take the time) or by
+    /// counting rows or batches.
     ///
     /// The cancellation benchmark tracks some cases of how quickly queries can
     /// be cancelled.

From c56523b52532bfa9f9e5ec3f46f188fb8c44adb7 Mon Sep 17 00:00:00 2001
From: "Carol (Nichols || Goulding)"
 <193874+carols10cents@users.noreply.github.com>
Date: Thu, 13 Mar 2025 10:35:24 -0400
Subject: [PATCH 07/10] Improve documentation of yield guidelines

Co-authored-by: Mehmet Ozan Kabak <ozankabak@gmail.com>
---
 datafusion/physical-plan/src/execution_plan.rs | 18 ++++++++++--------
 1 file changed, 10 insertions(+), 8 deletions(-)

diff --git a/datafusion/physical-plan/src/execution_plan.rs b/datafusion/physical-plan/src/execution_plan.rs
index 292673135555..0969f54c5838 100644
--- a/datafusion/physical-plan/src/execution_plan.rs
+++ b/datafusion/physical-plan/src/execution_plan.rs
@@ -261,14 +261,16 @@ pub trait ExecutionPlan: Debug + DisplayAs + Send + Sync {
     /// Thus, [`spawn`] is disallowed, and instead use [`SpawnedTask`].
     ///
     /// To enable timely cancellation, the [`Stream`] that is returned must not
-    /// block the CPU and must yield back to the tokio runtime regularly.
-    /// [`ExecutionPlan`] implementations should follow [the guideline of not
-    /// spending a long time without reaching an `await`/yield point][async-guideline].
-    /// This can be achieved by manually returning [`Poll::Pending`] in regular
-    /// intervals, or the use of [`tokio::task::yield_now()`] (as appropriate).
-    /// Determination for "regularly" may be made using a timer (being careful
-    /// with the  overhead-heavy syscall needed to take the time) or by
-    /// counting rows or batches.
+    /// block the CPU indefinitely and must yield back to the tokio runtime regularly.
+    /// In a typical [`ExecutionPlan`], this automatically happens unless there are
+    /// special circumstances; e.g. when the computational complexity of processing a
+    /// batch is superlinear. See this [general guideline][async-guideline] for more context
+    /// on this point, which explains why one should avoid spending a long time without
+    /// reaching an `await`/yield point in asynchronous runtimes.
+    /// This can be achieved by manually returning [`Poll::Pending`] and setting up wakers appropriately, or the use of [`tokio::task::yield_now()`] when appropriate.
+    /// In special cases that warrant manual yielding, determination for "regularly" may be
+    /// made using a timer (being careful with the  overhead-heavy system call needed to
+    /// take the time), or by counting rows or batches.
     ///
     /// The cancellation benchmark tracks some cases of how quickly queries can
     /// be cancelled.

From ab1c0864b5b888bcd0ff513126d0437861c757f2 Mon Sep 17 00:00:00 2001
From: "Carol (Nichols || Goulding)" <carol.nichols@gmail.com>
Date: Thu, 13 Mar 2025 10:37:53 -0400
Subject: [PATCH 08/10] Fix newlines and whitespace in comment

---
 datafusion/physical-plan/src/execution_plan.rs | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/datafusion/physical-plan/src/execution_plan.rs b/datafusion/physical-plan/src/execution_plan.rs
index 7d50dcc76a28..75a779a36796 100644
--- a/datafusion/physical-plan/src/execution_plan.rs
+++ b/datafusion/physical-plan/src/execution_plan.rs
@@ -267,9 +267,10 @@ pub trait ExecutionPlan: Debug + DisplayAs + Send + Sync {
     /// batch is superlinear. See this [general guideline][async-guideline] for more context
     /// on this point, which explains why one should avoid spending a long time without
     /// reaching an `await`/yield point in asynchronous runtimes.
-    /// This can be achieved by manually returning [`Poll::Pending`] and setting up wakers appropriately, or the use of [`tokio::task::yield_now()`] when appropriate.
+    /// This can be achieved by manually returning [`Poll::Pending`] and setting up wakers
+    /// appropriately, or the use of [`tokio::task::yield_now()`] when appropriate.
     /// In special cases that warrant manual yielding, determination for "regularly" may be
-    /// made using a timer (being careful with the  overhead-heavy system call needed to
+    /// made using a timer (being careful with the overhead-heavy system call needed to
     /// take the time), or by counting rows or batches.
     ///
     /// The cancellation benchmark tracks some cases of how quickly queries can

From e5d0f8d3da50394c57843a7a6ee464608c979283 Mon Sep 17 00:00:00 2001
From: "Carol (Nichols || Goulding)" <carol.nichols@gmail.com>
Date: Thu, 13 Mar 2025 10:40:37 -0400
Subject: [PATCH 09/10] Add a link to the cancellation benchmark documented in
 the README

---
 datafusion/physical-plan/src/execution_plan.rs | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/datafusion/physical-plan/src/execution_plan.rs b/datafusion/physical-plan/src/execution_plan.rs
index 75a779a36796..40bff635bcd5 100644
--- a/datafusion/physical-plan/src/execution_plan.rs
+++ b/datafusion/physical-plan/src/execution_plan.rs
@@ -273,13 +273,14 @@ pub trait ExecutionPlan: Debug + DisplayAs + Send + Sync {
     /// made using a timer (being careful with the overhead-heavy system call needed to
     /// take the time), or by counting rows or batches.
     ///
-    /// The cancellation benchmark tracks some cases of how quickly queries can
+    /// The [cancellation benchmark] tracks some cases of how quickly queries can
     /// be cancelled.
     ///
     /// For more details see [`SpawnedTask`], [`JoinSet`] and [`RecordBatchReceiverStreamBuilder`]
     /// for structures to help ensure all background tasks are cancelled.
     ///
     /// [`spawn`]: tokio::task::spawn
+    /// [cancellation benchmark]: https://github.com/apache/datafusion/blob/main/benchmarks/README.md#cancellation
     /// [`JoinSet`]: tokio::task::JoinSet
     /// [`SpawnedTask`]: datafusion_common_runtime::SpawnedTask
     /// [`RecordBatchReceiverStreamBuilder`]: crate::stream::RecordBatchReceiverStreamBuilder

From d42889bf6bba8887fd8a4b49b3f006b5997cc656 Mon Sep 17 00:00:00 2001
From: "Carol (Nichols || Goulding)" <carol.nichols@gmail.com>
Date: Thu, 13 Mar 2025 10:41:21 -0400
Subject: [PATCH 10/10] Fix newlines in benchmarks README

---
 benchmarks/README.md | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/benchmarks/README.md b/benchmarks/README.md
index f17d6b5a07b6..39b4584bd202 100644
--- a/benchmarks/README.md
+++ b/benchmarks/README.md
@@ -333,7 +333,8 @@ The output of `dfbench` help includes a description of each benchmark, which is
 
 ## Cancellation
 
-Test performance of cancelling queries
+Test performance of cancelling queries.
+
 Queries in DataFusion should stop executing "quickly" after they are
 cancelled (the output stream is dropped).
 
