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 apache#14036 and related to pull requests such as apache#14028.

Author: carols10cents

datafusion/physical-plan/src/execution_plan.rs

@@ -260,13 +260,30 @@ 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.
     ///
     /// [`spawn`]: tokio::task::spawn
     /// [`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
     ///