Add stats estimation accuracy to benchmarks

gabotechs · gabotechs · commit ada65596fca8 · 2026-07-16T10:14:28.000+02:00
diff --git a/benchmarks/cdk/bin/@bench-common.ts b/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`);
diff --git a/benchmarks/cdk/bin/@results.ts b/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
+}
diff --git a/benchmarks/cdk/bin/datafusion-bench.ts b/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> {
diff --git a/benchmarks/cdk/bin/worker.rs b/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)| {
diff --git a/benchmarks/src/datasets/common.rs b/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(())
diff --git a/benchmarks/src/lib.rs b/benchmarks/src/lib.rs
@@ -1 +1,2 @@
 pub mod datasets;
+pub mod stats;
diff --git a/benchmarks/src/results.rs b/benchmarks/src/results.rs
@@ -13,6 +13,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"
@@ -21,6 +25,19 @@ pub struct QueryIter {
     pub error: Option<String>,
 }
 
+fn median(mut values: Vec<f64>) -> Option<f64> {
+    if values.is_empty() {
+        return None;
+    }
+    values.sort_unstable_by(f64::total_cmp);
+    let mid = values.len() / 2;
+    Some(if values.len().is_multiple_of(2) {
+        (values[mid - 1] + values[mid]) / 2.0
+    } else {
+        values[mid]
+    })
+}
+
 /// A single benchmark case
 #[derive(Debug, Serialize, Deserialize)]
 pub struct BenchResult {
@@ -279,29 +296,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>(
diff --git a/benchmarks/src/run.rs b/benchmarks/src/run.rs
diff --git a/benchmarks/src/stats.rs b/benchmarks/src/stats.rs

Original file line number	Diff line number	Diff line change
`@@ -1 +1,2 @@`
`1`	`1`	`pub mod datasets;`
	`2`	`+pub mod stats;`