Use Tokio's task budget consistently, better APIs to support task cancellation (#16398)

* Use Tokio's task budget consistently

* Rework `ensure_coop` to base itself on evaluation and scheduling properties

* Iterating on documentation

* Improve robustness of cooperative yielding test cases

* Reorganize tests by operator a bit better

* Coop documentation

* More coop documentation

* Avoid Box in temporary CooperativeStream::poll_next implementation

* Adapt interleave test cases for range generator

* Add temporary `tokio_coop` feature to unblock merging

* Extract magic number to constant

* Fix documentation error

* Push scheduling type down from DataSourceExec to DataSource

* Use custom configuration instead of feature to avoid exposing internal cooperation variants

* Use dedicated enum for yield results

* Documentation improvements from review

* More documentation

* Change default coop strategy to 'tokio_fallback'

* Documentation refinement

* Re-enable interleave test cases

* fix logical merge conflict

---------

Co-authored-by: Andrew Lamb <[email protected]>

Author: pepijnve

Cargo.toml

@@ -218,5 +218,9 @@ unnecessary_lazy_evaluations = "warn"
 uninlined_format_args = "warn"
 
 [workspace.lints.rust]
-unexpected_cfgs = { level = "warn", check-cfg = ["cfg(tarpaulin)", "cfg(tarpaulin_include)"] }
+unexpected_cfgs = { level = "warn", check-cfg = [
+    'cfg(datafusion_coop, values("tokio", "tokio_fallback", "per_stream"))',
+    "cfg(tarpaulin)",
+    "cfg(tarpaulin_include)",
+] }
 unused_qualifications = "deny"

datafusion/common/src/config.rs

@@ -796,15 +796,6 @@ config_namespace! {
         /// then the output will be coerced to a non-view.
         /// Coerces `Utf8View` to `LargeUtf8`, and `BinaryView` to `LargeBinary`.
         pub expand_views_at_output: bool, default = false
-
-        /// When DataFusion detects that a plan might not be promply cancellable
-        /// due to the presence of tight-looping operators, it will attempt to
-        /// mitigate this by inserting explicit yielding (in as few places as
-        /// possible to avoid performance degradation). This value represents the
-        /// yielding period (in batches) at such explicit yielding points. The
-        /// default value is 64. If set to 0, no DataFusion will not perform
-        /// any explicit yielding.
-        pub yield_period: usize, default = 64
     }
 }