Remove Rayon-based Scheduler

.github/workflows/docs_pr.yaml

@@ -49,6 +49,6 @@ jobs:
           rust-version: stable
       # Note: this does not include dictionary_expressions to reduce codegen
       - name: Run doctests
-        run: cargo test --doc --features avro,jit,scheduler,json
+        run: cargo test --doc --features avro,jit,json
       - name: Verify Working Directory Clean
         run: git diff --exit-code

.github/workflows/rust.yml

@@ -67,7 +67,7 @@ jobs:
 
       # Note: this does not include dictionary_expressions to reduce codegen
       - name: Check workspace with all features
-        run: cargo check --workspace --benches --features avro,jit,scheduler,json
+        run: cargo check --workspace --benches --features avro,jit,json
 
       - name: Check Cargo.lock for datafusion-cli
         run: |
@@ -97,7 +97,7 @@ jobs:
         with:
           rust-version: stable
       - name: Run tests (excluding doctests)
-        run: cargo test --lib --tests --bins --features avro,jit,scheduler,json,dictionary_expressions
+        run: cargo test --lib --tests --bins --features avro,jit,json,dictionary_expressions
       - name: Verify Working Directory Clean
         run: git diff --exit-code
 
@@ -153,7 +153,7 @@ jobs:
           rust-version: stable
       # Note: this does not include dictionary_expressions to reduce codegen
       - name: Run doctests
-        run: cargo test --doc --features avro,jit,scheduler,json
+        run: cargo test --doc --features avro,jit,json
       - name: Verify Working Directory Clean
         run: git diff --exit-code
 
@@ -272,7 +272,7 @@ jobs:
         shell: bash
         run: |
           export PATH=$PATH:$HOME/d/protoc/bin
-          cargo test --lib --tests --bins --features avro,jit,scheduler,json,dictionary_expressions
+          cargo test --lib --tests --bins --features avro,jit,json,dictionary_expressions
         env:
           # do not produce debug symbols to keep memory usage down
           RUSTFLAGS: "-C debuginfo=0"
@@ -305,7 +305,7 @@ jobs:
       - name: Run tests (excluding doctests)
         shell: bash
         run: |
-          cargo test --lib --tests --bins --features avro,jit,scheduler,json,dictionary_expressions
+          cargo test --lib --tests --bins --features avro,jit,json,dictionary_expressions
         env:
           # do not produce debug symbols to keep memory usage down
           RUSTFLAGS: "-C debuginfo=0"

benchmarks/Cargo.toml

@@ -34,7 +34,7 @@ snmalloc = ["snmalloc-rs"]
 
 [dependencies]
 arrow = { workspace = true }
-datafusion = { path = "../datafusion/core", version = "23.0.0", features = ["scheduler"] }
+datafusion = { path = "../datafusion/core", version = "23.0.0" }
 env_logger = "0.10"
 futures = "0.3"
 mimalloc = { version = "0.1", optional = true, default-features = false }

benchmarks/src/bin/tpch.rs

@@ -37,8 +37,6 @@ use std::{iter::Iterator, path::PathBuf, sync::Arc, time::Instant};
 use datafusion::datasource::file_format::csv::DEFAULT_CSV_EXTENSION;
 use datafusion::datasource::file_format::parquet::DEFAULT_PARQUET_EXTENSION;
 use datafusion::datasource::listing::ListingTableUrl;
-use datafusion::scheduler::Scheduler;
-use futures::TryStreamExt;
 use structopt::StructOpt;
 
 #[cfg(feature = "snmalloc")]
@@ -90,10 +88,6 @@ struct DataFusionBenchmarkOpt {
     /// Whether to disable collection of statistics (and cost based optimizations) or not.
     #[structopt(short = "S", long = "disable-statistics")]
     disable_statistics: bool,
-
-    /// Enable scheduler
-    #[structopt(short = "e", long = "enable-scheduler")]
-    enable_scheduler: bool,
 }
 
 #[derive(Debug, StructOpt)]
@@ -227,16 +221,14 @@ async fn benchmark_query(
         if query_id == 15 {
             for (n, query) in sql.iter().enumerate() {
                 if n == 1 {
-                    result = execute_query(&ctx, query, opt.debug, opt.enable_scheduler)
-                        .await?;
+                    result = execute_query(&ctx, query, opt.debug).await?;
                 } else {
-                    execute_query(&ctx, query, opt.debug, opt.enable_scheduler).await?;
+                    execute_query(&ctx, query, opt.debug).await?;
                 }
             }
         } else {
             for query in sql {
-                result =
-                    execute_query(&ctx, query, opt.debug, opt.enable_scheduler).await?;
+                result = execute_query(&ctx, query, opt.debug).await?;
             }
         }
 
@@ -295,7 +287,6 @@ async fn execute_query(
     ctx: &SessionContext,
     sql: &str,
     debug: bool,
-    enable_scheduler: bool,
 ) -> Result<Vec<RecordBatch>> {
     let plan = ctx.sql(sql).await?;
     let (state, plan) = plan.into_parts();
@@ -315,15 +306,7 @@ async fn execute_query(
             displayable(physical_plan.as_ref()).indent()
         );
     }
-    let result = if enable_scheduler {
-        let scheduler = Scheduler::new(num_cpus::get());
-        let results = scheduler
-            .schedule(physical_plan.clone(), state.task_ctx())
-            .unwrap();
-        results.stream().try_collect().await?
-    } else {
-        collect(physical_plan.clone(), state.task_ctx()).await?
-    };
+    let result = collect(physical_plan.clone(), state.task_ctx()).await?;
     if debug {
         println!(
             "=== Physical plan with metrics ===\n{}\n",
@@ -530,7 +513,7 @@ mod tests {
             if sql.starts_with("select") {
                 let explain = "explain ".to_string() + sql;
                 let result_batch =
-                    execute_query(&ctx, explain.as_str(), false, false).await?;
+                    execute_query(&ctx, explain.as_str(), false).await?;
                 if !actual.is_empty() {
                     actual += "\n";
                 }
@@ -542,7 +525,7 @@ mod tests {
                 // let mut file = File::create(format!("expected-plans/q{}.txt", query))?;
                 // file.write_all(actual.as_bytes())?;
             } else {
-                execute_query(&ctx, sql.as_str(), false, false).await?;
+                execute_query(&ctx, sql.as_str(), false).await?;
             }
         }
 
@@ -726,7 +709,7 @@ mod tests {
 
         let sql = &get_query_sql(n)?;
         for query in sql {
-            execute_query(&ctx, query, false, false).await?;
+            execute_query(&ctx, query, false).await?;
         }
 
         Ok(())
@@ -757,7 +740,6 @@ mod ci {
             mem_table: false,
             output_path: None,
             disable_statistics: false,
-            enable_scheduler: false,
         };
         register_tables(&opt, &ctx).await?;
         let queries = get_query_sql(query)?;
@@ -1064,7 +1046,6 @@ mod ci {
             mem_table: false,
             output_path: None,
             disable_statistics: false,
-            enable_scheduler: false,
         };
         let mut results = benchmark_datafusion(opt).await?;
         assert_eq!(results.len(), 1);

ci/scripts/rust_clippy.sh

@@ -18,4 +18,4 @@
 # under the License.
 
 set -ex
-cargo clippy --all-targets --workspace --features avro,jit,pyarrow,scheduler -- -D warnings
+cargo clippy --all-targets --workspace --features avro,jit,pyarrow -- -D warnings

datafusion/core/Cargo.toml

@@ -48,8 +48,6 @@ force_hash_collisions = []
 jit = ["datafusion-jit", "datafusion-row/jit"]
 pyarrow = ["datafusion-common/pyarrow"]
 regex_expressions = ["datafusion-physical-expr/regex_expressions", "datafusion-optimizer/regex_expressions"]
-# Used to enable scheduler
-scheduler = ["rayon"]
 simd = ["arrow/simd"]
 unicode_expressions = ["datafusion-physical-expr/unicode_expressions", "datafusion-optimizer/unicode_expressions", "datafusion-sql/unicode_expressions"]
 
@@ -89,7 +87,6 @@ parquet = { workspace = true }
 percent-encoding = "2.2.0"
 pin-project-lite = "^0.2.7"
 rand = "0.8"
-rayon = { version = "1.5", optional = true }
 smallvec = { version = "1.6", features = ["union"] }
 sqlparser = { version = "0.33", features = ["visitor"] }
 tempfile = "3"
@@ -153,7 +150,6 @@ name = "physical_plan"
 [[bench]]
 harness = false
 name = "parquet_query_sql"
-required-features = ["scheduler"]
 
 [[bench]]
 harness = false

datafusion/core/benches/parquet_query_sql.rs

@@ -25,7 +25,6 @@ use arrow::datatypes::{
 use arrow::record_batch::RecordBatch;
 use criterion::{criterion_group, criterion_main, Criterion};
 use datafusion::prelude::{SessionConfig, SessionContext};
-use datafusion::scheduler::Scheduler;
 use futures::stream::StreamExt;
 use parquet::arrow::ArrowWriter;
 use parquet::file::properties::{WriterProperties, WriterVersion};
@@ -196,8 +195,6 @@ fn criterion_benchmark(c: &mut Criterion) {
     let config = SessionConfig::new().with_target_partitions(partitions);
     let context = SessionContext::with_config(config);
 
-    let scheduler = Scheduler::new(partitions);
-
     let local_rt = tokio::runtime::Builder::new_current_thread()
         .build()
         .unwrap();
@@ -249,22 +246,6 @@ fn criterion_benchmark(c: &mut Criterion) {
                 })
             });
         });
-
-        c.bench_function(&format!("scheduled: {query}"), |b| {
-            b.iter(|| {
-                let query = query.clone();
-                let context = context.clone();
-
-                local_rt.block_on(async {
-                    let query = context.sql(&query).await.unwrap();
-                    let plan = query.create_physical_plan().await.unwrap();
-                    let results = scheduler.schedule(plan, context.task_ctx()).unwrap();
-
-                    let mut stream = results.stream();
-                    while stream.next().await.transpose().unwrap().is_some() {}
-                });
-            });
-        });
     }
 
     // Temporary file must outlive the benchmarks, it is deleted when dropped

datafusion/core/src/lib.rs

@@ -422,8 +422,6 @@ pub mod physical_optimizer;
 pub mod physical_plan;
 pub mod prelude;
 pub mod scalar;
-#[cfg(feature = "scheduler")]
-pub mod scheduler;
 pub mod variable;
 
 // re-export dependencies from arrow-rs to minimise version maintenance for crate users