Combine Cargo workspaces

Cargo.toml

@@ -17,12 +17,12 @@
 
 [workspace]
 members = [
-        "datafusion",
-        "datafusion-examples",
+    "datafusion",
+    "datafusion-examples",
 	"benchmarks",
-]
-
-# this package is excluded because it requires different compilation flags, thereby significantly changing
-# how it is compiled within the workspace, causing the whole workspace to be compiled from scratch
-# this way, this is a stand-alone package that compiles independently of the others.
-exclude = ["ballista"]
+    "ballista/rust/benchmarks/tpch",
+    "ballista/rust/client",
+    "ballista/rust/core",
+    "ballista/rust/executor",
+    "ballista/rust/scheduler",
+]

ballista/rust/benchmarks/tpch/Cargo.toml

@@ -17,7 +17,7 @@
 
 [package]
 name = "tpch"
-version = "0.4.2-SNAPSHOT"
+version = "0.5.0-SNAPSHOT"
 homepage = "https://github.com/apache/arrow"
 repository = "https://github.com/apache/arrow"
 authors = ["Apache Arrow <[email protected]>"]

ballista/rust/client/Cargo.toml

@@ -19,7 +19,7 @@
 name = "ballista"
 description = "Ballista Distributed Compute"
 license = "Apache-2.0"
-version = "0.4.2-SNAPSHOT"
+version = "0.5.0-SNAPSHOT"
 homepage = "https://github.com/apache/arrow"
 repository = "https://github.com/apache/arrow"
 authors = ["Apache Arrow <[email protected]>"]

ballista/rust/client/src/context.rs

@@ -30,7 +30,7 @@ use ballista_core::serde::protobuf::{
 };
 use ballista_core::{
     client::BallistaClient,
-    datasource::DFTableAdapter,
+    datasource::DfTableAdapter,
     error::{BallistaError, Result},
     memory_stream::MemoryStream,
     utils::create_datafusion_context,
@@ -151,7 +151,7 @@ impl BallistaContext {
             let execution_plan = ctx.create_physical_plan(&plan)?;
             ctx.register_table(
                 TableReference::Bare { table: name },
-                Arc::new(DFTableAdapter::new(plan, execution_plan)),
+                Arc::new(DfTableAdapter::new(plan, execution_plan)),
             )?;
         }
         let df = ctx.sql(sql)?;

ballista/rust/core/Cargo.toml

@@ -19,7 +19,7 @@
 name = "ballista-core"
 description = "Ballista Distributed Compute"
 license = "Apache-2.0"
-version = "0.4.2-SNAPSHOT"
+version = "0.5.0-SNAPSHOT"
 homepage = "https://github.com/apache/arrow"
 repository = "https://github.com/apache/arrow"
 authors = ["Apache Arrow <[email protected]>"]

ballista/rust/core/src/datasource.rs

@@ -30,20 +30,20 @@ use datafusion::{
 /// TableProvider which is effectively a wrapper around a physical plan. We need to be able to
 /// register tables so that we can create logical plans from SQL statements that reference these
 /// tables.
-pub struct DFTableAdapter {
+pub struct DfTableAdapter {
     /// DataFusion logical plan
     pub logical_plan: LogicalPlan,
     /// DataFusion execution plan
     plan: Arc<dyn ExecutionPlan>,
 }
 
-impl DFTableAdapter {
+impl DfTableAdapter {
     pub fn new(logical_plan: LogicalPlan, plan: Arc<dyn ExecutionPlan>) -> Self {
         Self { logical_plan, plan }
     }
 }
 
-impl TableProvider for DFTableAdapter {
+impl TableProvider for DfTableAdapter {
     fn as_any(&self) -> &dyn Any {
         self
     }

ballista/rust/core/src/error.rs

@@ -49,6 +49,7 @@ pub enum BallistaError {
     TokioError(tokio::task::JoinError),
 }
 
+#[allow(clippy::from_over_into)]
 impl<T> Into<Result<T>> for BallistaError {
     fn into(self) -> Result<T> {
         Err(self)

ballista/rust/core/src/serde/logical_plan/from_proto.rs

@@ -436,6 +436,7 @@ impl TryInto<arrow::datatypes::DataType> for &protobuf::arrow_type::ArrowTypeEnu
     }
 }
 
+#[allow(clippy::from_over_into)]
 impl Into<arrow::datatypes::DataType> for protobuf::PrimitiveScalarType {
     fn into(self) -> arrow::datatypes::DataType {
         use arrow::datatypes::DataType;
@@ -1170,6 +1171,7 @@ impl TryFrom<i32> for protobuf::FileType {
     }
 }
 
+#[allow(clippy::from_over_into)]
 impl Into<datafusion::sql::parser::FileType> for protobuf::FileType {
     fn into(self) -> datafusion::sql::parser::FileType {
         use datafusion::sql::parser::FileType;

ballista/rust/core/src/serde/logical_plan/to_proto.rs

@@ -24,7 +24,7 @@ use std::{
     convert::{TryFrom, TryInto},
 };
 
-use crate::datasource::DFTableAdapter;
+use crate::datasource::DfTableAdapter;
 use crate::serde::{protobuf, BallistaError};
 
 use arrow::datatypes::{DataType, Schema};
@@ -679,7 +679,7 @@ impl TryInto<protobuf::LogicalPlanNode> for &LogicalPlan {
 
                 // unwrap the DFTableAdapter to get to the real TableProvider
                 let source = if let Some(adapter) =
-                    source.as_any().downcast_ref::<DFTableAdapter>()
+                    source.as_any().downcast_ref::<DfTableAdapter>()
                 {
                     match &adapter.logical_plan {
                         LogicalPlan::TableScan { source, .. } => Ok(source.as_any()),
@@ -1020,7 +1020,7 @@ impl TryInto<protobuf::LogicalExprNode> for &Expr {
                 let fun: protobuf::ScalarFunction = fun.try_into()?;
                 let expr: Vec<protobuf::LogicalExprNode> = args
                     .iter()
-                    .map(|e| Ok(e.try_into()?))
+                    .map(|e| e.try_into())
                     .collect::<Result<Vec<protobuf::LogicalExprNode>, BallistaError>>()?;
                 Ok(protobuf::LogicalExprNode {
                     expr_type: Some(
@@ -1163,6 +1163,7 @@ impl TryInto<protobuf::LogicalExprNode> for &Expr {
     }
 }
 
+#[allow(clippy::from_over_into)]
 impl Into<protobuf::Schema> for &Schema {
     fn into(self) -> protobuf::Schema {
         protobuf::Schema {

ballista/rust/core/src/serde/scheduler/from_proto.rs

@@ -72,6 +72,7 @@ impl TryInto<PartitionId> for protobuf::PartitionId {
     }
 }
 
+#[allow(clippy::from_over_into)]
 impl Into<PartitionStats> for protobuf::PartitionStats {
     fn into(self) -> PartitionStats {
         PartitionStats::new(

ballista/rust/core/src/serde/scheduler/mod.rs

@@ -75,6 +75,7 @@ pub struct ExecutorMeta {
     pub port: u16,
 }
 
+#[allow(clippy::from_over_into)]
 impl Into<protobuf::ExecutorMetadata> for ExecutorMeta {
     fn into(self) -> protobuf::ExecutorMetadata {
         protobuf::ExecutorMetadata {

ballista/rust/core/src/serde/scheduler/to_proto.rs

@@ -55,6 +55,7 @@ impl TryInto<protobuf::ExecutePartition> for ExecutePartition {
     }
 }
 
+#[allow(clippy::from_over_into)]
 impl Into<protobuf::PartitionId> for PartitionId {
     fn into(self) -> protobuf::PartitionId {
         protobuf::PartitionId {
@@ -77,6 +78,7 @@ impl TryInto<protobuf::PartitionLocation> for PartitionLocation {
     }
 }
 
+#[allow(clippy::from_over_into)]
 impl Into<protobuf::PartitionStats> for PartitionStats {
     fn into(self) -> protobuf::PartitionStats {
         let none_value = -1_i64;

ballista/rust/executor/Cargo.toml

@@ -19,14 +19,13 @@
 name = "ballista-executor"
 description = "Ballista Distributed Compute - Executor"
 license = "Apache-2.0"
-version = "0.4.2-SNAPSHOT"
+version = "0.5.0-SNAPSHOT"
 homepage = "https://github.com/apache/arrow"
 repository = "https://github.com/apache/arrow"
 authors = ["Apache Arrow <[email protected]>"]
 edition = "2018"
 
 [features]
-default = ["snmalloc"]
 snmalloc = ["snmalloc-rs"]
 
 [dependencies]

ballista/rust/scheduler/Cargo.toml

@@ -19,7 +19,7 @@
 name = "ballista-scheduler"
 description = "Ballista Distributed Compute - Scheduler"
 license = "Apache-2.0"
-version = "0.4.2-SNAPSHOT"
+version = "0.5.0-SNAPSHOT"
 homepage = "https://github.com/apache/arrow"
 repository = "https://github.com/apache/arrow"
 authors = ["Apache Arrow <[email protected]>"]

ballista/rust/scheduler/src/planner.rs

@@ -25,7 +25,7 @@ use std::time::Instant;
 use std::{collections::HashMap, future::Future};
 
 use ballista_core::client::BallistaClient;
-use ballista_core::datasource::DFTableAdapter;
+use ballista_core::datasource::DfTableAdapter;
 use ballista_core::error::{BallistaError, Result};
 use ballista_core::serde::scheduler::ExecutorMeta;
 use ballista_core::serde::scheduler::PartitionId;
@@ -138,7 +138,7 @@ impl DistributedPlanner {
             stages.append(&mut child_stages);
         }
 
-        if let Some(adapter) = execution_plan.as_any().downcast_ref::<DFTableAdapter>() {
+        if let Some(adapter) = execution_plan.as_any().downcast_ref::<DfTableAdapter>() {
             // remove Repartition rule because that isn't supported yet
             let rules: Vec<Arc<dyn PhysicalOptimizerRule + Send + Sync>> = vec![
                 Arc::new(CoalesceBatches::new()),

dev/docker/rust.dockerfile

@@ -59,20 +59,18 @@ ARG RELEASE_FLAG=--release
 # force build.rs to run to generate configure_me code.
 ENV FORCE_REBUILD='true'
 RUN cargo build $RELEASE_FLAG
-RUN cd ballista/rust && \
-    cargo build $RELEASE_FLAG
 
 # put the executor on /executor (need to be copied from different places depending on FLAG)
 ENV RELEASE_FLAG=${RELEASE_FLAG}
-RUN if [ -z "$RELEASE_FLAG" ]; then mv /tmp/ballista/ballista/rust/target/debug/ballista-executor /executor; else mv /tmp/ballista/ballista/rust/target/release/ballista-executor /executor; fi
+RUN if [ -z "$RELEASE_FLAG" ]; then mv /tmp/ballista/target/debug/ballista-executor /executor; else mv /tmp/ballista/target/release/ballista-executor /executor; fi
 
 # put the scheduler on /scheduler (need to be copied from different places depending on FLAG)
 ENV RELEASE_FLAG=${RELEASE_FLAG}
-RUN if [ -z "$RELEASE_FLAG" ]; then mv /tmp/ballista/ballista/rust/target/debug/ballista-scheduler /scheduler; else mv /tmp/ballista/ballista/rust/target/release/ballista-scheduler /scheduler; fi
+RUN if [ -z "$RELEASE_FLAG" ]; then mv /tmp/ballista/target/debug/ballista-scheduler /scheduler; else mv /tmp/ballista/target/release/ballista-scheduler /scheduler; fi
 
 # put the tpch on /tpch (need to be copied from different places depending on FLAG)
 ENV RELEASE_FLAG=${RELEASE_FLAG}
-RUN if [ -z "$RELEASE_FLAG" ]; then mv /tmp/ballista/ballista/rust/target/debug/tpch /tpch; else mv /tmp/ballista/ballista/rust/target/release/tpch /tpch; fi
+RUN if [ -z "$RELEASE_FLAG" ]; then mv /tmp/ballista/target/debug/tpch /tpch; else mv /tmp/ballista/target/release/tpch /tpch; fi
 
 # Copy the binary into a new container for a smaller docker image
 FROM ballistacompute/rust-base:0.4.0-20210213