Emulate blob storage latency in benchmarks (#2049)

The nvme devices we use for benchmarks (both in the cloud and locally)
are incredibly fast and have incredibly low latency, which makes them
forgiving in exactly the way real object storage isnt.
This is an attempt at allowing us to emulate the performance of real
cloud system in a more reproducible way, the constants and distributions
are not set in stone and we can defiantly change them around, especially
for ranged read where we have more specific numbers from AnyBlob.

Author: AdamGS

Cargo.lock

@@ -101,6 +101,7 @@ libfuzzer-sys = "0.4"
 log = "0.4.21"
 mimalloc = "0.1.42"
 moka = "0.12"
+governor = "0.8"
 num-traits = "0.2.18"
 num_enum = "0.7.2"
 object_store = "0.11.0"
@@ -116,6 +117,7 @@ pyo3 = { version = ">= 0.22", features = ["extension-module", "abi3-py310"] }
 pyo3-log = ">= 0.11"
 rancor = "0.1.0"
 rand = "0.8.5"
+rand_distr = "0.4"
 rayon = "1.10.0"
 regex = "1.11.0"
 reqwest = { version = "0.12.0", features = ["blocking"] }

Cargo.toml

@@ -26,6 +26,7 @@ anyhow = { workspace = true }
 arrow-array = { workspace = true }
 arrow-schema = { workspace = true }
 arrow-select = { workspace = true }
+async-trait = { workspace = true }
 bytes = { workspace = true }
 bzip2 = { workspace = true }
 clap = { workspace = true, features = ["derive"] }
@@ -37,6 +38,7 @@ datafusion-common = { workspace = true }
 datafusion-physical-plan = { workspace = true }
 enum-iterator = { workspace = true }
 futures = { workspace = true, features = ["executor"] }
+governor = { workspace = true }
 homedir = { workspace = true }
 humansize = { workspace = true }
 indicatif = { workspace = true }
@@ -46,6 +48,7 @@ mimalloc = { workspace = true }
 object_store = { workspace = true, features = ["aws"] }
 parquet = { workspace = true, features = ["async"] }
 rand = { workspace = true }
+rand_distr = { workspace = true }
 rayon = { workspace = true }
 regex = { workspace = true }
 reqwest = { workspace = true }

bench-vortex/Cargo.toml

@@ -41,7 +41,7 @@ fn benchmark(c: &mut Criterion) {
         .unwrap();
     }
 
-    let session_context = get_session_with_cache();
+    let session_context = get_session_with_cache(false);
     let context = session_context.clone();
 
     runtime.block_on(async move {

bench-vortex/benches/clickbench.rs

@@ -20,20 +20,22 @@ fn benchmark(c: &mut Criterion) {
             Format::InMemoryVortex {
                 enable_pushdown: true,
             },
+            false,
         ))
         .unwrap();
     let arrow_ctx = runtime
-        .block_on(load_datasets(&data_dir, Format::Arrow))
+        .block_on(load_datasets(&data_dir, Format::Arrow, false))
         .unwrap();
     let parquet_ctx = runtime
-        .block_on(load_datasets(&data_dir, Format::Parquet))
+        .block_on(load_datasets(&data_dir, Format::Parquet, false))
         .unwrap();
     let vortex_compressed_ctx = runtime
         .block_on(load_datasets(
             &data_dir,
             Format::OnDiskVortex {
                 enable_compression: true,
             },
+            false,
         ))
         .unwrap();
 

bench-vortex/benches/tpch.rs

@@ -37,6 +37,8 @@ struct Args {
     queries: Option<Vec<usize>>,
     #[arg(long, default_value = "false")]
     emit_plan: bool,
+    #[arg(long, default_value = "false")]
+    emulate_object_store: bool,
 }
 
 fn main() {
@@ -120,7 +122,7 @@ fn main() {
     let mut all_measurements = Vec::default();
 
     for format in &formats {
-        let session_context = get_session_with_cache();
+        let session_context = get_session_with_cache(args.emulate_object_store);
         let context = session_context.clone();
         match format {
             Format::Parquet => runtime.block_on(async {

bench-vortex/src/bin/clickbench.rs

@@ -33,6 +33,8 @@ struct Args {
     verbose: bool,
     #[arg(short, long, default_value_t, value_enum)]
     display_format: DisplayFormat,
+    #[arg(long, default_value = "false")]
+    emulate_object_store: bool,
 }
 
 fn main() -> ExitCode {
@@ -62,6 +64,7 @@ fn main() -> ExitCode {
         args.warmup,
         args.only_vortex,
         args.display_format,
+        args.emulate_object_store,
     ))
 }
 
@@ -72,6 +75,7 @@ async fn bench_main(
     warmup: bool,
     only_vortex: bool,
     display_format: DisplayFormat,
+    emulate_object_store: bool,
 ) -> ExitCode {
     // uncomment the below to enable trace logging of datafusion execution
     // setup_logger(LevelFilter::Trace);
@@ -101,7 +105,7 @@ async fn bench_main(
     let ctxs = try_join_all(
         formats
             .iter()
-            .map(|format| load_datasets(&data_dir, *format)),
+            .map(|format| load_datasets(&data_dir, *format, emulate_object_store)),
     )
     .await
     .unwrap();