Reorganize the project folders (#2081)

* Reorganize the project folders

* fix: fix hash collision test

* fix: tomlfmt

Co-authored-by: yangzhong <yangzhong@ebay.com>
Co-authored-by: Andrew Lamb <andrew@nerdnetworks.org>

Author: 3 people

.github/workflows/rust.yml

@@ -430,7 +430,7 @@ jobs:
           export PARQUET_TEST_DATA=$(pwd)/parquet-testing/data
           cd datafusion
           # Force all hash values to collide
-          cargo test --features=force_hash_collisions
+          cargo test --all --features=force_hash_collisions
         env:
           CARGO_HOME: "/github/home/.cargo"
           CARGO_TARGET_DIR: "/github/home/target"

Cargo.toml

@@ -17,21 +17,21 @@
 
 [workspace]
 members = [
-    "datafusion",
-    "datafusion-common",
-    "datafusion-expr",
-    "datafusion-jit",
-    "datafusion-physical-expr",
+    "datafusion/core",
+    "datafusion/common",
+    "datafusion/expr",
+    "datafusion/jit",
+    "datafusion/physical-expr",
+    "datafusion/proto",
     "datafusion-cli",
     "datafusion-examples",
-    "datafusion-proto",
-    "datafusion-storage",
     "benchmarks",
     "ballista/rust/client",
     "ballista/rust/core",
     "ballista/rust/executor",
     "ballista/rust/scheduler",
     "ballista-examples",
+    "data-access",
 ]
 
 [profile.release]

ballista-examples/Cargo.toml

@@ -36,7 +36,7 @@ required-features = ["ballista/standalone"]
 
 [dependencies]
 ballista = { path = "../ballista/rust/client", version = "0.6.0" }
-datafusion = { path = "../datafusion" }
+datafusion = { path = "../datafusion/core" }
 futures = "0.3"
 num_cpus = "1.13.0"
 prost = "0.9"

ballista/rust/client/Cargo.toml

@@ -31,7 +31,7 @@ ballista-core = { path = "../core", version = "0.6.0" }
 ballista-executor = { path = "../executor", version = "0.6.0", optional = true }
 ballista-scheduler = { path = "../scheduler", version = "0.6.0", optional = true }
 
-datafusion = { path = "../../../datafusion", version = "7.0.0" }
+datafusion = { path = "../../../datafusion/core", version = "7.0.0" }
 futures = "0.3"
 log = "0.4"
 parking_lot = "0.12"

ballista/rust/core/Cargo.toml

@@ -27,6 +27,8 @@ edition = "2018"
 build = "build.rs"
 
 [features]
+# Used for testing ONLY: causes all values to hash to the same value (test for collisions)
+force_hash_collisions = ["datafusion/force_hash_collisions"]
 simd = ["datafusion/simd"]
 
 [dependencies]
@@ -36,8 +38,8 @@ arrow-flight = { version = "11" }
 async-trait = "0.1.41"
 chrono = { version = "0.4", default-features = false }
 clap = { version = "3", features = ["derive", "cargo"] }
-datafusion = { path = "../../../datafusion", version = "7.0.0" }
-datafusion-proto = { path = "../../../datafusion-proto", version = "7.0.0" }
+datafusion = { path = "../../../datafusion/core", version = "7.0.0" }
+datafusion-proto = { path = "../../../datafusion/proto", version = "7.0.0" }
 futures = "0.3"
 hashbrown = "0.12"
 log = "0.4"

ballista/rust/core/src/execution_plans/shuffle_writer.rs

@@ -452,6 +452,8 @@ mod tests {
     use tempfile::TempDir;
 
     #[tokio::test]
+    // number of rows in each partition is a function of the hash output, so don't test here
+    #[cfg(not(feature = "force_hash_collisions"))]
     async fn test() -> Result<()> {
         let session_ctx = SessionContext::new();
         let task_ctx = session_ctx.task_ctx();
@@ -507,6 +509,8 @@ mod tests {
     }
 
     #[tokio::test]
+    // number of rows in each partition is a function of the hash output, so don't test here
+    #[cfg(not(feature = "force_hash_collisions"))]
     async fn test_partitioned() -> Result<()> {
         let session_ctx = SessionContext::new();
         let task_ctx = session_ctx.task_ctx();

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

@@ -922,7 +922,7 @@ mod roundtrip_tests {
     use core::panic;
     use datafusion::{
         arrow::datatypes::{DataType, Field, Schema},
-        datafusion_storage::{
+        datafusion_data_access::{
             self,
             object_store::{
                 local::LocalFileSystem, FileMetaStream, ListEntryStream, ObjectReader,
@@ -949,7 +949,7 @@ mod roundtrip_tests {
         async fn list_file(
             &self,
             _prefix: &str,
-        ) -> datafusion_storage::Result<FileMetaStream> {
+        ) -> datafusion_data_access::Result<FileMetaStream> {
             Err(io::Error::new(
                 io::ErrorKind::Unsupported,
                 "this is only a test object store".to_string(),
@@ -960,7 +960,7 @@ mod roundtrip_tests {
             &self,
             _prefix: &str,
             _delimiter: Option<String>,
-        ) -> datafusion_storage::Result<ListEntryStream> {
+        ) -> datafusion_data_access::Result<ListEntryStream> {
             Err(io::Error::new(
                 io::ErrorKind::Unsupported,
                 "this is only a test object store".to_string(),
@@ -970,7 +970,7 @@ mod roundtrip_tests {
         fn file_reader(
             &self,
             _file: SizedFile,
-        ) -> datafusion_storage::Result<Arc<dyn ObjectReader>> {
+        ) -> datafusion_data_access::Result<Arc<dyn ObjectReader>> {
             Err(io::Error::new(
                 io::ErrorKind::Unsupported,
                 "this is only a test object store".to_string(),

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

@@ -351,7 +351,7 @@ fn str_to_byte(s: &str) -> Result<u8, BallistaError> {
 mod tests {
     use async_trait::async_trait;
     use datafusion::arrow::datatypes::SchemaRef;
-    use datafusion::datafusion_storage::object_store::local::LocalFileSystem;
+    use datafusion::datafusion_data_access::object_store::local::LocalFileSystem;
     use datafusion::error::DataFusionError;
     use datafusion::execution::context::{QueryPlanner, SessionState, TaskContext};
     use datafusion::execution::runtime_env::{RuntimeConfig, RuntimeEnv};
@@ -715,7 +715,7 @@ mod tests {
 
         let scan = LogicalPlanBuilder::scan_csv(
             store,
-            "../../../datafusion/tests/customer.csv",
+            "../../../datafusion/core/tests/customer.csv",
             CsvReadOptions::default(),
             None,
             1,

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

@@ -26,7 +26,7 @@ use crate::serde::{from_proto_binary_op, proto_error, protobuf};
 use crate::{convert_box_required, convert_required};
 use chrono::{TimeZone, Utc};
 
-use datafusion::datafusion_storage::{
+use datafusion::datafusion_data_access::{
     object_store::local::LocalFileSystem, FileMeta, SizedFile,
 };
 use datafusion::datasource::listing::PartitionedFile;

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

@@ -33,7 +33,7 @@ use crate::serde::{
 use crate::{convert_box_required, convert_required, into_physical_plan, into_required};
 use datafusion::arrow::compute::SortOptions;
 use datafusion::arrow::datatypes::SchemaRef;
-use datafusion::datafusion_storage::object_store::local::LocalFileSystem;
+use datafusion::datafusion_data_access::object_store::local::LocalFileSystem;
 use datafusion::datasource::listing::PartitionedFile;
 use datafusion::execution::runtime_env::RuntimeEnv;
 use datafusion::logical_plan::window_frames::WindowFrame;
@@ -1020,7 +1020,7 @@ mod roundtrip_tests {
             compute::kernels::sort::SortOptions,
             datatypes::{DataType, Field, Schema},
         },
-        datafusion_storage::object_store::local::LocalFileSystem,
+        datafusion_data_access::object_store::local::LocalFileSystem,
         datasource::listing::PartitionedFile,
         logical_plan::{JoinType, Operator},
         physical_plan::{