save

rust/ballista/rust/benchmarks/tpch/Cargo.toml

@@ -27,10 +27,9 @@ edition = "2018"
 [dependencies]
 ballista = { path="../../client" }
 
-arrow = { git = "https://github.com/apache/arrow", rev="f1f4f2b" }
-datafusion = { git = "https://github.com/apache/arrow", rev="f1f4f2b" }
-parquet = { git = "https://github.com/apache/arrow", rev="f1f4f2b" }
-
+arrow = { path = "../../../../arrow"  }
+datafusion = { path = "../../../../datafusion" }
+parquet = { path = "../../../../parquet"  }
 
 env_logger = "0.8"
 tokio = { version = "1.0", features = ["macros", "rt", "rt-multi-thread"] }

rust/ballista/rust/client/Cargo.toml

@@ -30,5 +30,6 @@ ballista-core = { path = "../core" }
 futures = "0.3"
 log = "0.4"
 tokio = "1.0"
-arrow = { git = "https://github.com/apache/arrow", rev="f1f4f2b" }
-datafusion = { git = "https://github.com/apache/arrow", rev="f1f4f2b" }
+
+arrow = { path = "../../../arrow"  }
+datafusion = { path = "../../../datafusion" }

rust/ballista/rust/core/Cargo.toml

@@ -39,9 +39,10 @@ sqlparser = "0.8"
 tokio = "1.0"
 tonic = "0.4"
 uuid = { version = "0.8", features = ["v4"] }
-arrow = { git = "https://github.com/apache/arrow", rev="f1f4f2b" }
-arrow-flight = { git = "https://github.com/apache/arrow", rev="f1f4f2b" }
-datafusion = { git = "https://github.com/apache/arrow", rev="f1f4f2b" }
+
+arrow = { path = "../../../arrow"  }
+arrow-flight = { path = "../../../arrow-flight"  }
+datafusion = { path = "../../../datafusion" }
 
 
 [dev-dependencies]

rust/ballista/rust/executor/Cargo.toml

@@ -45,9 +45,9 @@ tokio-stream = "0.1"
 tonic = "0.4"
 uuid = { version = "0.8", features = ["v4"] }
 
-arrow = { git = "https://github.com/apache/arrow", rev="f1f4f2b" }
-arrow-flight = { git = "https://github.com/apache/arrow", rev="f1f4f2b" }
-datafusion = { git = "https://github.com/apache/arrow", rev="f1f4f2b" }
+arrow = { path = "../../../arrow"  }
+arrow-flight = { path = "../../../arrow-flight"  }
+datafusion = { path = "../../../datafusion" }
 
 [dev-dependencies]
 

rust/ballista/rust/scheduler/Cargo.toml

@@ -52,8 +52,8 @@ tonic = "0.4"
 tower = { version = "0.4" }
 warp = "0.3"
 
-arrow = { git = "https://github.com/apache/arrow", rev="f1f4f2b" }
-datafusion = { git = "https://github.com/apache/arrow", rev="f1f4f2b" }
+arrow = { path = "../../../arrow"  }
+datafusion = { path = "../../../datafusion" }
 
 [dev-dependencies]
 ballista-core = { path = "../core" }

rust/ballista/rust/scheduler/src/planner.rs

@@ -35,9 +35,12 @@ use ballista_core::{
     serde::scheduler::PartitionLocation,
 };
 use datafusion::execution::context::ExecutionContext;
+use datafusion::physical_optimizer::coalesce_batches::CoalesceBatches;
+use datafusion::physical_optimizer::optimizer::PhysicalOptimizerRule;
 use datafusion::physical_plan::hash_aggregate::{AggregateMode, HashAggregateExec};
 use datafusion::physical_plan::hash_join::HashJoinExec;
 use datafusion::physical_plan::merge::MergeExec;
+use datafusion::physical_optimizer::merge_exec::AddMergeExec;
 use datafusion::physical_plan::ExecutionPlan;
 use log::{debug, info};
 use tokio::task::JoinHandle;
@@ -136,7 +139,12 @@ impl DistributedPlanner {
         }
 
         if let Some(adapter) = execution_plan.as_any().downcast_ref::<DFTableAdapter>() {
-            let ctx = ExecutionContext::new();
+            let rules: Vec<Arc<dyn PhysicalOptimizerRule + Send + Sync>> = vec![
+                Arc::new(CoalesceBatches::new()),
+                Arc::new(AddMergeExec::new()),
+            ];
+            let mut ctx = ExecutionContext::new();
+            let ctx = ctx.with_physical_optimizers(rules);
             Ok((ctx.create_physical_plan(&adapter.logical_plan)?, stages))
         } else if let Some(merge) = execution_plan.as_any().downcast_ref::<MergeExec>() {
             let query_stage = create_query_stage(