Add stats estimation accuracy to benchmarks

Author: gabotechs

benchmarks/cdk/bin/@bench-common.ts

@@ -68,6 +68,7 @@ export interface ExecuteQueryResult {
     plan: string
     elapsed: number
     tasks: number
+    statsQError?: number
 }
 
 export interface BenchmarkRunner {
@@ -193,11 +194,18 @@ export async function runBenchmark(
                 rowCount: response.rowCount,
                 plan: response.plan,
                 tasks: response.tasks,
+                statsQError: response.statsQError,
             })
 
-            console.log(
-                `Query ${id} iteration ${i} took ${Math.round(response.elapsed)} ms and returned ${response.rowCount} rows`
-            );
+            if (response.statsQError !== undefined) {
+                console.log(
+                    `Query ${id} iteration ${i} took ${Math.round(response.elapsed)} ms, stats q-error ${response.statsQError.toFixed(2)}x and returned ${response.rowCount} rows`
+                );
+            } else {
+                console.log(
+                    `Query ${id} iteration ${i} took ${Math.round(response.elapsed)} ms and returned ${response.rowCount} rows`
+                );
+            }
         }
 
         console.log(`Query ${id} p50 time: ${result.p50()} ms`);

benchmarks/cdk/bin/@results.ts

@@ -12,6 +12,7 @@ export interface QueryIter {
     rowCount: number;
     elapsed: number; // Duration in milliseconds
     tasks: number;
+    statsQError?: number;
     error?: string;
 }
 
@@ -80,16 +81,24 @@ export class BenchmarkRun {
         console.log(`=== Comparing ${this.dataset} results from engine '${other.engine}' [prev] with '${this.engine}' [new] ===`);
         let totalTimePrev = 0
         let totalTimeNew = 0
+        const statsQErrorPrev: number[] = []
+        const statsQErrorNew: number[] = []
         for (const query of this.results) {
             const prevQuery = other.results.find(v => v.id === query.id);
             if (!prevQuery) {
                 continue;
             }
             const timePrev = prevQuery.representativeTime()
             const timeNew = query.representativeTime()
-            if (timePrev && timeNew) {
+            if (timePrev !== undefined && timeNew !== undefined) {
                 totalTimePrev += timePrev
                 totalTimeNew += timeNew
+                statsQErrorPrev.push(...prevQuery.iterations.flatMap(iter =>
+                    iter.statsQError === undefined ? [] : [iter.statsQError]
+                ))
+                statsQErrorNew.push(...query.iterations.flatMap(iter =>
+                    iter.statsQError === undefined ? [] : [iter.statsQError]
+                ))
             }
 
             query.compare(prevQuery);
@@ -108,6 +117,16 @@ export class BenchmarkRun {
         console.log(
             `${"TOTAL".padStart(8)}: prev=${totalTimePrev.toString()} ms, new=${totalTimeNew.toString()} ms, diff=${f.toFixed(2)} ${tag} ${emoji}`
         );
+
+        const qErrorPrev = median(statsQErrorPrev)
+        const qErrorNew = median(statsQErrorNew)
+        if (qErrorPrev !== undefined && qErrorNew !== undefined) {
+            console.log(`${"QERR P50".padStart(8)}: prev=${qErrorPrev.toFixed(2)}x, new=${qErrorNew.toFixed(2)}x`)
+        } else if (qErrorPrev !== undefined) {
+            console.log(`${"QERR P50".padStart(8)}: prev=${qErrorPrev.toFixed(2)}x, new=n/a`)
+        } else if (qErrorNew !== undefined) {
+            console.log(`${"QERR P50".padStart(8)}: prev=n/a, new=${qErrorNew.toFixed(2)}x`)
+        }
     }
 
     compareWithPrevious(): void {
@@ -238,6 +257,7 @@ export class BenchResult {
                     error: z.string().optional(),
                     plan: z.string(),
                     tasks: z.number().default(0),
+                    statsQError: z.number().optional(),
                 }).array(),
             })
             const data = fs.readFileSync(filePath, 'utf-8');
@@ -286,3 +306,12 @@ export class BenchResult {
 function numericId(queryName: string): number {
     return parseInt([...queryName.matchAll(/(\d+)/g)][0][0])
 }
+
+function median(values: number[]): number | undefined {
+    if (values.length === 0) {
+        return undefined
+    }
+    const sorted = [...values].sort((a, b) => a - b)
+    const mid = Math.floor(sorted.length / 2)
+    return sorted.length % 2 ? sorted[mid] : (sorted[mid - 1] + sorted[mid]) / 2
+}

benchmarks/cdk/bin/datafusion-bench.ts

@@ -87,7 +87,8 @@ const QueryResponse = z.object({
     count: z.number(),
     plan: z.string(),
     elapsed_ms: z.number(),
-    tasks: z.number()
+    tasks: z.number(),
+    stats_q_error: z.number().nullable().optional()
 })
 type QueryResponse = z.infer<typeof QueryResponse>
 
@@ -142,7 +143,13 @@ class DataFusionRunner implements BenchmarkRunner {
             response = await this.query(sql)
         }
 
-        return { rowCount: response.count, plan: response.plan, elapsed: response.elapsed_ms, tasks: response.tasks };
+        return {
+            rowCount: response.count,
+            plan: response.plan,
+            elapsed: response.elapsed_ms,
+            tasks: response.tasks,
+            statsQError: response.stats_q_error ?? undefined
+        };
     }
 
     private async query(sql: string): Promise<QueryResponse> {

benchmarks/cdk/bin/worker.rs

@@ -20,6 +20,7 @@ use datafusion_distributed::{
     get_distributed_channel_resolver, get_distributed_worker_resolver,
     rewrite_distributed_plan_with_metrics,
 };
+use datafusion_distributed_benchmarks::stats::stats_estimation_q_error;
 use futures::{StreamExt, TryFutureExt};
 use log::{error, info, warn};
 use object_store::aws::AmazonS3Builder;
@@ -48,6 +49,7 @@ struct QueryResult {
     count: usize,
     elapsed_ms: f64,
     tasks: usize,
+    stats_q_error: Option<f64>,
 }
 
 #[derive(Serialize)]
@@ -208,6 +210,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
                         )
                         .await
                         .map_err(err)?;
+                        let stats_q_error = stats_estimation_q_error(&physical);
                         let plan = display_plan_ascii(physical.as_ref(), true);
                         drop(task);
 
@@ -233,6 +236,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
                             plan,
                             elapsed_ms: ms,
                             tasks: task_count,
+                            stats_q_error,
                         }))
                     }
                     .inspect_err(|(_, msg)| {

benchmarks/src/datasets/common.rs

@@ -77,12 +77,13 @@ pub async fn register_tables(
         let path = entry?.path();
         if path.is_dir() {
             let table_name = path.file_name().unwrap().to_str().unwrap();
-            ctx.register_parquet(
-                table_name,
-                path.to_str().unwrap(),
-                ParquetReadOptions::default(),
-            )
-            .await?;
+            let _ = ctx
+                .register_parquet(
+                    table_name,
+                    path.to_str().unwrap(),
+                    ParquetReadOptions::default(),
+                )
+                .await;
         }
     }
     Ok(())

benchmarks/src/lib.rs

@@ -1 +1,2 @@
 pub mod datasets;
+pub mod stats;

benchmarks/src/results.rs

@@ -1,6 +1,7 @@
 use crate::{DATA_PATH, RESULTS_DIR};
 use datafusion::common::utils::get_available_parallelism;
 use datafusion::common::{Result, internal_datafusion_err};
+use datafusion_distributed_benchmarks::stats::median;
 use serde::ser::SerializeSeq;
 use serde::{Deserialize, Deserializer, Serialize, Serializer};
 use std::fs;
@@ -13,6 +14,10 @@ use std::time::{Duration, SystemTime};
 pub struct QueryIter {
     pub row_count: usize,
     pub n_tasks: usize,
+    /// Median q-error of the byte estimates made at dynamic stage boundaries.
+    /// `None` when dynamic planning was disabled or no sampled boundary was present.
+    #[serde(default)]
+    pub stats_q_error: Option<f64>,
     #[serde(
         serialize_with = "serialize_elapsed",
         deserialize_with = "deserialize_elapsed"
@@ -279,29 +284,49 @@ impl BenchResult {
 pub fn print_comparison_total(base: &[BenchResult], new: &[BenchResult]) {
     let mut total_prev: u128 = 0;
     let mut total_new: u128 = 0;
+    let mut stats_q_error_prev = vec![];
+    let mut stats_q_error_new = vec![];
     for query in new {
         let Some(prev) = base.iter().find(|v| v.id == query.id) else {
             continue;
         };
         if let (Some(p), Some(n)) = (prev.representative_time(), query.representative_time()) {
             total_prev += p;
             total_new += n;
+            stats_q_error_prev.extend(
+                prev.iterations
+                    .iter()
+                    .filter_map(|iteration| iteration.stats_q_error),
+            );
+            stats_q_error_new.extend(
+                query
+                    .iterations
+                    .iter()
+                    .filter_map(|iteration| iteration.stats_q_error),
+            );
         }
     }
-    if total_prev == 0 && total_new == 0 {
-        return;
+
+    if total_prev != 0 || total_new != 0 {
+        let (f, tag, emoji) = if total_new < total_prev {
+            let f = total_prev as f64 / total_new as f64;
+            (f, "faster", if f > 1.2 { "✅" } else { "✔" })
+        } else {
+            let f = total_new as f64 / total_prev.max(1) as f64;
+            (f, "slower", if f > 1.2 { "❌" } else { "✖" })
+        };
+        println!(
+            "{:>8}: prev={total_prev} ms, new={total_new} ms, diff={f:.2} {tag} {emoji}",
+            "TOTAL"
+        );
+    }
+
+    match (median(stats_q_error_prev), median(stats_q_error_new)) {
+        (Some(prev), Some(new)) => println!("{:>8}: prev={prev:.2}x, new={new:.2}x", "QERR P50"),
+        (Some(prev), None) => println!("{:>8}: prev={prev:.2}x, new=n/a", "QERR P50"),
+        (None, Some(new)) => println!("{:>8}: prev=n/a, new={new:.2}x", "QERR P50"),
+        (None, None) => {}
     }
-    let (f, tag, emoji) = if total_new < total_prev {
-        let f = total_prev as f64 / total_new as f64;
-        (f, "faster", if f > 1.2 { "✅" } else { "✔" })
-    } else {
-        let f = total_new as f64 / total_prev.max(1) as f64;
-        (f, "slower", if f > 1.2 { "❌" } else { "✖" })
-    };
-    println!(
-        "{:>8}: prev={total_prev} ms, new={total_new} ms, diff={f:.2} {tag} {emoji}",
-        "TOTAL"
-    );
 }
 
 fn serialize_bench_results<S: Serializer>(

benchmarks/src/run.rs

@@ -25,7 +25,7 @@ use datafusion::common::{config_err, exec_err, not_impl_err};
 use datafusion::datasource::source::DataSourceExec;
 use datafusion::error::{DataFusionError, Result};
 use datafusion::execution::SessionStateBuilder;
-use datafusion::physical_plan::collect;
+use datafusion::physical_plan::{ExecutionPlan, collect};
 use datafusion::prelude::*;
 use datafusion_distributed::test_utils::localhost::LocalHostWorkerResolver;
 use datafusion_distributed::test_utils::work_unit_file_scan::{
@@ -37,6 +37,7 @@ use datafusion_distributed::{
     display_plan_ascii, rewrite_distributed_plan_with_metrics,
 };
 use datafusion_distributed_benchmarks::datasets::{clickbench, register_tables, tpcds, tpch};
+use datafusion_distributed_benchmarks::stats::stats_estimation_q_error;
 use std::error::Error;
 use std::fs;
 use std::path::PathBuf;
@@ -88,14 +89,6 @@ pub struct RunOpt {
     #[structopt(long)]
     cardinality_task_sf: Option<f64>,
 
-    /// Use children isolator UNIONs for distributing UNION operations.
-    #[structopt(long)]
-    children_isolator_unions: bool,
-
-    /// Turns on broadcast joins.
-    #[structopt(long = "broadcast-joins")]
-    broadcast_joins: bool,
-
     /// Collects metrics across network boundaries
     #[structopt(long)]
     collect_metrics: bool,
@@ -216,9 +209,11 @@ impl RunOpt {
                 "none" => None,
                 v => return config_err!("Unknown compression type {v}"),
             })?
-            .with_distributed_children_isolator_unions(self.children_isolator_unions)?
-            .with_distributed_broadcast_joins(self.broadcast_joins)?
-            .with_distributed_metrics_collection(self.collect_metrics || self.debug)?
+            .with_distributed_children_isolator_unions(true)?
+            .with_distributed_broadcast_joins(true)?
+            .with_distributed_metrics_collection(
+                self.collect_metrics || self.debug || self.dynamic,
+            )?
             .with_distributed_max_tasks_per_stage(self.max_tasks_per_stage)?
             .with_distributed_user_codec(WorkUnitFileScanCodec)
             .with_distributed_task_estimator(WorkUnitFileScanTaskEstimator)
@@ -290,18 +285,44 @@ impl RunOpt {
                 }
 
                 match self.execute_query(ctx, query).await {
-                    Ok((result, n_tasks)) => {
+                    Ok((result, n_tasks, physical_plan)) => {
                         let elapsed = start.elapsed();
                         let ms = elapsed.as_secs_f64() * 1000.0;
                         let row_count = result.iter().map(|b| b.num_rows()).sum();
-                        println!(
-                            "Query {id} iteration {i} took {ms:.1} ms and returned {row_count} rows"
-                        );
+                        let physical_plan = if self.dynamic || self.debug {
+                            rewrite_distributed_plan_with_metrics(
+                                physical_plan,
+                                DistributedMetricsFormat::PerTask,
+                            )
+                            .await?
+                        } else {
+                            physical_plan
+                        };
+                        let stats_q_error = match self.dynamic {
+                            true => stats_estimation_q_error(&physical_plan),
+                            false => None,
+                        };
+                        if let Some(q_error) = stats_q_error {
+                            println!(
+                                "Query {id} iteration {i} took {ms:.1} ms, stats q-error {q_error:.2}x and returned {row_count} rows"
+                            );
+                        } else {
+                            println!(
+                                "Query {id} iteration {i} took {ms:.1} ms and returned {row_count} rows"
+                            );
+                        }
+                        if self.debug {
+                            println!(
+                                "=== Physical plan with metrics ===\n{}\n",
+                                display_plan_ascii(physical_plan.as_ref(), true)
+                            );
+                        }
 
                         bench_query.iterations.push(QueryIter {
                             elapsed,
                             row_count,
                             n_tasks,
+                            stats_q_error,
                             error: None,
                         });
                     }
@@ -311,6 +332,7 @@ impl RunOpt {
                             elapsed: Duration::from_millis(0),
                             row_count: 0,
                             n_tasks: 0,
+                            stats_q_error: None,
                             error: Some(err.to_string()),
                         });
                         continue 'outer;
@@ -327,7 +349,7 @@ impl RunOpt {
         &self,
         ctx: &SessionContext,
         sql: &str,
-    ) -> Result<(Vec<RecordBatch>, usize)> {
+    ) -> Result<(Vec<RecordBatch>, usize, Arc<dyn ExecutionPlan>)> {
         let plan = ctx.sql(sql).await?;
         let (state, plan) = plan.into_parts();
 
@@ -341,18 +363,7 @@ impl RunOpt {
             Ok(Transformed::no(node))
         })?;
         let result = collect(physical_plan.clone(), state.task_ctx()).await?;
-        if self.debug {
-            let physical_plan = rewrite_distributed_plan_with_metrics(
-                physical_plan.clone(),
-                DistributedMetricsFormat::PerTask,
-            )
-            .await?;
-            println!(
-                "=== Physical plan with metrics ===\n{}\n",
-                display_plan_ascii(physical_plan.as_ref(), true)
-            );
-        }
-        Ok((result, n_tasks))
+        Ok((result, n_tasks, physical_plan))
     }
 
     fn get_path(&self) -> Result<PathBuf> {