Merge pull request #7 from robtandy/rob.tandy/pull_arrow_flight

Rob.tandy/pull arrow flight

Author: robtandy

.cargo/config.toml

@@ -1,12 +1,8 @@
 [target.x86_64-apple-darwin]
-rustflags = [
-  "-C", "link-arg=-undefined",
-  "-C", "link-arg=dynamic_lookup",
-]
+rustflags = ["-C", "link-arg=-undefined", "-C", "link-arg=dynamic_lookup"]
 
 [target.aarch64-apple-darwin]
-rustflags = [
-  "-C", "link-arg=-undefined",
-  "-C", "link-arg=dynamic_lookup",
-]
+rustflags = ["-C", "link-arg=-undefined", "-C", "link-arg=dynamic_lookup"]
 
+[build]
+rustflags = ["-C", "target-cpu=native"]

Cargo.lock

@@ -25,7 +25,7 @@ version = "0.1.0"
 edition = "2021"
 readme = "README.md"
 license = "Apache-2.0"
-rust-version = "1.70"
+rust-version = "1.76"
 build = "build.rs"
 
 [dependencies]
@@ -34,6 +34,7 @@ arrow = { version = "53.3", features = ["pyarrow", "ipc"] }
 arrow-flight = "53.3"
 async-stream = "0.3"
 async-channel = "2.3"
+bytesize = "1.3"
 datafusion = { version = "43.0", features = ["pyarrow", "avro"] }
 datafusion-python = { version = "43.1" }
 datafusion-proto = "43.0"
@@ -83,9 +84,9 @@ tonic-build = { version = "0.8", default-features = false, features = [
 url = "2"
 
 [dev-dependencies]
-anyhow = "1.0.89"
-pretty_assertions = "1.4.0"
-regex = "1.11.0"
+#anyhow = "1.0.89"
+#pretty_assertions = "1.4.0"
+#regex = "1.11.0"
 
 [lib]
 name = "datafusion_ray"
@@ -95,5 +96,13 @@ crate-type = ["cdylib", "rlib"]
 name = "datafusion_ray._datafusion_ray_internal"
 
 [profile.release]
+lto = "thin"
 codegen-units = 1
-lto = true
+opt-level = 3
+debug = 0
+
+[profile.dev]
+opt-level = 1
+
+[profile.dev.package."*"]
+opt-level = 3

Cargo.toml

@@ -48,19 +48,62 @@ of [Apache Arrow], [Apache DataFusion], and [Ray].
 
 [DataFusion Python]: https://github.com/apache/datafusion-python
 
+## Building
+
+To build DataFusion Ray, you will need rust installed, as well as [https://github.com/PyO3/maturin](maturin).
+
+Install maturin in your current python environment (a virtual environment is recommended), with
+
+```bash
+pip install maturin
+```
+
+Then build the project with the following command:
+
+```bash
+maturin develop # --release for a release build
+```
+
+- In the `examples` directory, run
+
 ## Example
 
-- In the `tpch` directory, use `make_data.py` to create a TPCH dataset at the specifed scale factor, then
+- In the `examples` directory, run
 
 ```bash
-RAY_COLOR_PREFIX=1 RAY_DEDUP_LOGS=0 python tpcbench.py --data=file:///path/to/your/tpch/directory/ --concurrency=4 --batch-size=4096 --validate
+RAY_COLOR_PREFIX=1 RAY_DEDUP_LOGS=0 python tips.py --data-dir=$(pwd)/../testdata/tips/
 ```
 
+- In the `tpch` directory, use `make_data.py` to create a TPCH dataset at a provided scale factor, then
+
+```bash
+RAY_COLOR_PREFIX=1 RAY_DEDUP_LOGS=0 python tpc.py --data=file:///path/to/your/tpch/directory/ --concurrency=2 --batch-size=8182 --qnum 2
+```
+
+To execute the TPCH query #2. To execute an arbitrary query against the TPCH dataset, provide it with `--query` instead of `--qnum`. This is useful for validating plans that DataFusion Ray will create.
+
+For example, to execute the following query:
+
+```bash
+RAY_COLOR_PREFIX=1 RAY_DEDUP_LOGS=0 python tpc.py --data=file:///path/to/your/tpch/directory/ --concurrency=2 --batch-size=8182 --query `select c.c_name, sum(o.o_totalprice) as total from orders o inner join customer c on o.o_custkey = c.c_custkey group by c_name limit 1`
+```
+
+To further parallelize execution, and host each partition of each stage as a Ray Actor, add `--isolate`. Note that this can drastically increase the number of Actors. A future version of DataFusion Ray will provide a`--split-factor` which will let you configure how the Stages are split.
+
+To validate the output against non-ray single node datafusion, add `--validate` which will ensure that both systems produce the same output.
+
+To run the entire TPCH benchmark use
+
+```bash
+RAY_COLOR_PREFIX=1 RAY_DEDUP_LOGS=0 python tpcbench.py --data=file:///path/to/your/tpch/directory/ --concurrency=2 --batch-size=8182 [--isolate] [--validate]
+```
+
+This will output a json file in the current directory with query timings.
+
 ## Status
 
-- DataFusion Ray can execute all TPCH queries.
+- DataFusion Ray can execute all TPCH queries. Tested up to SF100.
 
 ## Known Issues
 
-- Using `--isolate` (in `tpcbench.py`) to execute individual partitions in their own Ray Actors currently can produce incorrect results.
-- The DataFusion config setting, `datafusion.execution.parquet.pushdown_filters`, can produce incorrect results. We think this could be related to an issue with round trip physical path serialization.
+- The DataFusion config setting, `datafusion.execution.parquet.pushdown_filters`, can produce incorrect results. We think this could be related to an issue with round trip physical path serialization. At the moment, do not enable this setting, as it prevents physical plans from serializing correctly.

README.md

@@ -18,39 +18,39 @@
 use std::path::Path;
 
 fn main() -> Result<(), String> {
-    /*    use std::io::Write;
-
-        let out = std::path::PathBuf::from(std::env::var("OUT_DIR").unwrap());
-
-        // for use in docker build where file changes can be wonky
-        println!("cargo:rerun-if-env-changed=FORCE_REBUILD");
-
-        let version = rustc_version::version().unwrap();
-        println!("cargo:rustc-env=RUSTC_VERSION={version}");
-
-        let path = "src/proto/generated/protobuf.rs";
-
-        // We don't include the proto files in releases so that downstreams
-        // do not need to have PROTOC included
-        if Path::new("src/proto/datafusion_ray.proto").exists() {
-            println!("cargo:rerun-if-changed=src/proto/datafusion_common.proto");
-            println!("cargo:rerun-if-changed=src/proto/datafusion.proto");
-            println!("cargo:rerun-if-changed=src/proto/datafusion_ray.proto");
-            tonic_build::configure()
-                .extern_path(".datafusion", "::datafusion_proto::protobuf")
-                .extern_path(".datafusion_common", "::datafusion_proto::protobuf")
-                .compile(&["src/proto/datafusion_ray.proto"], &["src/proto"])
-                .map_err(|e| format!("protobuf compilation failed: {e}"))?;
-            let generated_source_path = out.join("datafusion_ray.protobuf.rs");
-            let code = std::fs::read_to_string(generated_source_path).unwrap();
-            let mut file = std::fs::OpenOptions::new()
-                .write(true)
-                .truncate(true)
-                .create(true)
-                .open(path)
-                .unwrap();
-            file.write_all(code.as_str().as_ref()).unwrap();
-        }
-    */
+    use std::io::Write;
+
+    let out = std::path::PathBuf::from(std::env::var("OUT_DIR").unwrap());
+
+    // for use in docker build where file changes can be wonky
+    println!("cargo:rerun-if-env-changed=FORCE_REBUILD");
+
+    let version = rustc_version::version().unwrap();
+    println!("cargo:rustc-env=RUSTC_VERSION={version}");
+
+    let path = "src/proto/generated/protobuf.rs";
+
+    // We don't include the proto files in releases so that downstreams
+    // do not need to have PROTOC included
+    if Path::new("src/proto/datafusion_ray.proto").exists() {
+        println!("cargo:rerun-if-changed=src/proto/datafusion_common.proto");
+        println!("cargo:rerun-if-changed=src/proto/datafusion.proto");
+        println!("cargo:rerun-if-changed=src/proto/datafusion_ray.proto");
+        tonic_build::configure()
+            .extern_path(".datafusion", "::datafusion_proto::protobuf")
+            .extern_path(".datafusion_common", "::datafusion_proto::protobuf")
+            .compile(&["src/proto/datafusion_ray.proto"], &["src/proto"])
+            .map_err(|e| format!("protobuf compilation failed: {e}"))?;
+        let generated_source_path = out.join("datafusion_ray.protobuf.rs");
+        let code = std::fs::read_to_string(generated_source_path).unwrap();
+        let mut file = std::fs::OpenOptions::new()
+            .write(true)
+            .truncate(true)
+            .create(true)
+            .open(path)
+            .unwrap();
+        file.write_all(code.as_str().as_ref()).unwrap();
+    }
+
     Ok(())
 }