cargo changes and logging

Signed-off-by: bharath-techie <[email protected]>

Author: bharath-techie

plugins/engine-datafusion/Cargo.toml

@@ -0,0 +1,74 @@
+[workspace]
+resolver = "2"
+members = [
+    "jni"
+]
+
+[workspace.dependencies]
+# DataFusion dependencies
+datafusion = "49.0.0"
+datafusion-expr = "49.0.0"
+datafusion-datasource = "49.0.0"
+arrow-json = "55.2"
+arrow = { version = "55.2", features = ["ffi", "ipc_compression"] }
+#arrow = "55.2.0"
+arrow-array = "55.2.0"
+arrow-schema = "55.2.0"
+arrow-buffer = "55.2.0"
+
+# JNI dependencies
+jni = "0.21"
+
+# Substrait support
+datafusion-substrait = "49.0.0"
+prost = "0.13"
+
+
+# Async runtime
+tokio = { version = "1.0", features = ["full"] }
+futures = "0.3"
+#tokio = { version = "1.0", features = ["rt", "rt-multi-thread", "macros"] }
+
+# Serialization
+serde = { version = "1.0", features = ["derive"] }
+serde_json = "1.0"
+
+# Error handling
+anyhow = "1.0"
+thiserror = "1.0"
+
+# Logging
+log = "0.4"
+# Parquet support
+parquet = "53.0.0"
+
+# Object store for file access
+object_store = "=0.12.3"
+url = "2.0"
+
+# Substrait support
+substrait = "0.47"
+
+# Temporary directory support
+tempfile = "3.0"
+chrono = "0.4.41"
+
+async-trait = "0.1.89"
+itertools = "0.14.0"
+rstest = "0.26.1"
+regex = "1.11.2"
+#
+#[build-dependencies]
+#cbindgen = "0.27"
+
+
+[profile.release]
+lto = true
+codegen-units = 1
+panic = "abort"
+
+[profile.dev]
+opt-level = 1          # Some optimization for reasonable performance
+lto = false           # Disable LTO for faster builds
+codegen-units = 16    # More parallel compilation
+incremental = true    # Enable incremental compilation

plugins/engine-datafusion/jni/Cargo.toml

@@ -11,66 +11,68 @@ crate-type = ["cdylib"]
 
 [dependencies]
 # DataFusion dependencies
-datafusion = "49.0.0"
-datafusion-expr = "49.0.0"
-datafusion-datasource = "49.0.0"
-arrow-json = "55.2"
-arrow = { version = "55.2", features = ["ffi", "ipc_compression"] }
+datafusion = { workspace = true }
+datafusion-expr = { workspace = true }
+datafusion-datasource = { workspace = true }
+arrow-json = { workspace = true }
+arrow = { workspace = true }
 #arrow = "55.2.0"
-arrow-array = "55.2.0"
-arrow-schema = "55.2.0"
-arrow-buffer = "55.2.0"
+arrow-array = { workspace = true }
+arrow-schema = { workspace = true }
+arrow-buffer = { workspace = true }
 
 # JNI dependencies
-jni = "0.21"
+jni = { workspace = true }
 
 # Substrait support
-datafusion-substrait = "49.0.0"
-prost = "0.13"
+datafusion-substrait = { workspace = true }
+prost = { workspace = true }
 
 
 # Async runtime
-tokio = { version = "1.0", features = ["full"] }
-futures = "0.3"
+tokio = { workspace = true }
+futures = { workspace = true }
 #tokio = { version = "1.0", features = ["rt", "rt-multi-thread", "macros"] }
 
 # Serialization
-serde = { version = "1.0", features = ["derive"] }
-serde_json = "1.0"
+serde = { workspace = true }
+serde_json = { workspace = true }
 
 # Error handling
-anyhow = "1.0"
-thiserror = "1.0"
+anyhow = { workspace = true }
+thiserror = { workspace = true }
 
 # Logging
-log = "0.4"
+log ={ workspace = true }
 # Parquet support
-parquet = "53.0.0"
+parquet = { workspace = true }
 
 # Object store for file access
-object_store = "=0.12.3"
-url = "2.0"
+object_store = { workspace = true }
+url = { workspace = true }
 
 # Substrait support
-substrait = "0.47"
+substrait = { workspace = true }
 
 # Temporary directory support
-tempfile = "3.0"
-chrono = "0.4.41"
+tempfile ={ workspace = true }
+chrono = { workspace = true }
 
-async-trait = "0.1.89"
-itertools = "0.14.0"
-rstest = "0.26.1"
-regex = "1.11.2"
+async-trait = { workspace = true }
+itertools = { workspace = true }
+rstest = { workspace = true }
+regex = { workspace = true }
 
 [build-dependencies]
 cbindgen = "0.27"
-
+[build]
+rustflags = ["-C", "force-frame-pointers=yes","symbol-mangling-version=v0"]
 
 [profile.release]
 lto = true
 codegen-units = 1
 panic = "abort"
+incremental = true
 
 [profile.dev]
 opt-level = 1          # Some optimization for reasonable performance

plugins/engine-datafusion/jni/src/lib.rs

@@ -13,6 +13,7 @@ use jni::sys::{jbyteArray, jlong, jstring};
 use jni::JNIEnv;
 use std::ptr::addr_of_mut;
 use std::sync::Arc;
+use std::time::Instant;
 
 mod util;
 mod row_id_optimizer;
@@ -208,6 +209,7 @@ pub extern "system" fn Java_org_opensearch_datafusion_DataFusionQueryJNI_execute
     tokio_runtime_env_ptr: jlong,
     // callback: JObject,
 ) -> jlong {
+    let overall = Instant::now();
     let shard_view = unsafe { &*(shard_view_ptr as *const ShardView) };
     let runtime_ptr = unsafe { &*(tokio_runtime_env_ptr as *const Runtime)};
     let table_name: String = env.get_string(&table_name).expect("Couldn't get java string!").into();
@@ -228,7 +230,8 @@ pub extern "system" fn Java_org_opensearch_datafusion_DataFusionQueryJNI_execute
 
     // TODO: get config from CSV DataFormat
     let mut config = SessionConfig::new();
-    // config.options_mut().execution.parquet.pushdown_filters = true;
+    config.options_mut().execution.parquet.pushdown_filters = false;
+    config.options_mut().execution.target_partitions = 1;
 
     let state = datafusion::execution::SessionStateBuilder::new()
         .with_config(config)
@@ -265,6 +268,7 @@ pub extern "system" fn Java_org_opensearch_datafusion_DataFusionQueryJNI_execute
 
     });
 
+    let start = Instant::now();
     // TODO : how to close ctx ?
     // Convert Java byte array to Rust Vec<u8>
     let plan_bytes_obj = unsafe { JByteArray::from_raw(substrait_bytes) };
@@ -293,18 +297,22 @@ pub extern "system" fn Java_org_opensearch_datafusion_DataFusionQueryJNI_execute
         let logical_plan = match from_substrait_plan(&ctx.state(), &substrait_plan).await {
             Ok(plan) => {
                 // println!("SUBSTRAIT Rust: LogicalPlan: {:?}", plan);
+                  let duration = start.elapsed();
+                         println!("Rust: Substrait decoding time in milliseconds: {}", duration.as_millis());
                 plan
             },
             Err(e) => {
                 println!("SUBSTRAIT Rust: Failed to convert Substrait plan: {}", e);
                 return 0;
             }
         };
-
         let dataframe = ctx.execute_logical_plan(logical_plan).await.unwrap();
         let stream = dataframe.execute_stream().await.unwrap();
         let stream_ptr = Box::into_raw(Box::new(stream)) as jlong;
         // println!("The memory used currently right now: {:?}", jemalloc_stats::refresh_allocated());
+        let duration1 = overall.elapsed();
+        println!("Rust: Overall query setup time in milliseconds: {}", duration1.as_millis());
+
         stream_ptr
 
     })