Add stats estimation accuracy to benchmarks

benchmarks/Cargo.toml

@@ -32,6 +32,7 @@ aws-config = "1"
 aws-sdk-ec2 = "1"
 openssl = { version = "0.10", features = ["vendored"] } # Keep this. Necessary for the remote benchmarks worker.
 mimalloc = "0.1"
+sketches-ddsketch = "0.3"
 
 [dev-dependencies]
 criterion = "0.5"

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

@@ -68,6 +68,8 @@ export interface ExecuteQueryResult {
     plan: string
     elapsed: number
     tasks: number
+    statsQErrorP50?: number
+    statsQErrorP95?: number
 }
 
 export interface BenchmarkRunner {
@@ -193,11 +195,19 @@ export async function runBenchmark(
                 rowCount: response.rowCount,
                 plan: response.plan,
                 tasks: response.tasks,
+                statsQErrorP50: response.statsQErrorP50,
+                statsQErrorP95: response.statsQErrorP95,
             })
 
-            console.log(
-                `Query ${id} iteration ${i} took ${Math.round(response.elapsed)} ms and returned ${response.rowCount} rows`
-            );
+            if (response.statsQErrorP50 !== undefined && response.statsQErrorP95 !== undefined) {
+                console.log(
+                    `Query ${id} iteration ${i} took ${Math.round(response.elapsed)} ms, stats q-error P50 ${response.statsQErrorP50.toFixed(2)}x, P95 ${response.statsQErrorP95.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,8 @@ export interface QueryIter {
     rowCount: number;
     elapsed: number; // Duration in milliseconds
     tasks: number;
+    statsQErrorP50?: number;
+    statsQErrorP95?: number;
     error?: string;
 }
 
@@ -80,16 +82,32 @@ 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 statsQErrorP50Prev: number[] = []
+        const statsQErrorP50New: number[] = []
+        const statsQErrorP95Prev: number[] = []
+        const statsQErrorP95New: 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
+                statsQErrorP50Prev.push(...prevQuery.iterations.flatMap(iter =>
+                    iter.statsQErrorP50 === undefined ? [] : [iter.statsQErrorP50]
+                ))
+                statsQErrorP50New.push(...query.iterations.flatMap(iter =>
+                    iter.statsQErrorP50 === undefined ? [] : [iter.statsQErrorP50]
+                ))
+                statsQErrorP95Prev.push(...prevQuery.iterations.flatMap(iter =>
+                    iter.statsQErrorP95 === undefined ? [] : [iter.statsQErrorP95]
+                ))
+                statsQErrorP95New.push(...query.iterations.flatMap(iter =>
+                    iter.statsQErrorP95 === undefined ? [] : [iter.statsQErrorP95]
+                ))
             }
 
             query.compare(prevQuery);
@@ -108,6 +126,9 @@ export class BenchmarkRun {
         console.log(
             `${"TOTAL".padStart(8)}: prev=${totalTimePrev.toString()} ms, new=${totalTimeNew.toString()} ms, diff=${f.toFixed(2)} ${tag} ${emoji}`
         );
+
+        printQErrorComparison("QERR P50", statsQErrorP50Prev, statsQErrorP50New)
+        printQErrorComparison("QERR P95", statsQErrorP95Prev, statsQErrorP95New)
     }
 
     compareWithPrevious(): void {
@@ -238,6 +259,8 @@ export class BenchResult {
                     error: z.string().optional(),
                     plan: z.string(),
                     tasks: z.number().default(0),
+                    statsQErrorP50: z.number().optional(),
+                    statsQErrorP95: z.number().optional(),
                 }).array(),
             })
             const data = fs.readFileSync(filePath, 'utf-8');
@@ -286,3 +309,24 @@ export class BenchResult {
 function numericId(queryName: string): number {
     return parseInt([...queryName.matchAll(/(\d+)/g)][0][0])
 }
+
+function printQErrorComparison(label: string, prev: number[], next: number[]): void {
+    const prevValue = median(prev)
+    const nextValue = median(next)
+    if (prevValue !== undefined && nextValue !== undefined) {
+        console.log(`${label.padStart(8)}: prev=${prevValue.toFixed(2)}x, new=${nextValue.toFixed(2)}x`)
+    } else if (prevValue !== undefined) {
+        console.log(`${label.padStart(8)}: prev=${prevValue.toFixed(2)}x, new=n/a`)
+    } else if (nextValue !== undefined) {
+        console.log(`${label.padStart(8)}: prev=n/a, new=${nextValue.toFixed(2)}x`)
+    }
+}
+
+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,9 @@ const QueryResponse = z.object({
     count: z.number(),
     plan: z.string(),
     elapsed_ms: z.number(),
-    tasks: z.number()
+    tasks: z.number(),
+    stats_q_error_p50: z.number().nullable(),
+    stats_q_error_p95: z.number().nullable()
 })
 type QueryResponse = z.infer<typeof QueryResponse>
 
@@ -142,7 +144,14 @@ 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,
+            statsQErrorP50: response.stats_q_error_p50 ?? undefined,
+            statsQErrorP95: response.stats_q_error_p95 ?? 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,8 @@ struct QueryResult {
     count: usize,
     elapsed_ms: f64,
     tasks: usize,
+    stats_q_error_p50: Option<f64>,
+    stats_q_error_p95: Option<f64>,
 }
 
 #[derive(Serialize)]
@@ -208,6 +211,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 +237,8 @@ async fn main() -> Result<(), Box<dyn Error>> {
                             plan,
                             elapsed_ms: ms,
                             tasks: task_count,
+                            stats_q_error_p50: stats_q_error.map(|q_error| q_error.p50),
+                            stats_q_error_p95: stats_q_error.map(|q_error| q_error.p95),
                         }))
                     }
                     .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,12 @@ use std::time::{Duration, SystemTime};
 pub struct QueryIter {
     pub row_count: usize,
     pub n_tasks: usize,
+    /// P50 q-error of the byte estimates made at dynamic stage boundaries.
+    /// `None` when dynamic planning was disabled or no sampled boundary was present.
+    pub stats_q_error_p50: Option<f64>,
+    /// P95 q-error of the byte estimates made at dynamic stage boundaries.
+    /// `None` when dynamic planning was disabled or no sampled boundary was present.
+    pub stats_q_error_p95: Option<f64>,
     #[serde(
         serialize_with = "serialize_elapsed",
         deserialize_with = "deserialize_elapsed"
@@ -279,29 +286,67 @@ 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_p50_prev = vec![];
+    let mut stats_q_error_p50_new = vec![];
+    let mut stats_q_error_p95_prev = vec![];
+    let mut stats_q_error_p95_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_p50_prev.extend(
+                prev.iterations
+                    .iter()
+                    .filter_map(|iteration| iteration.stats_q_error_p50),
+            );
+            stats_q_error_p50_new.extend(
+                query
+                    .iterations
+                    .iter()
+                    .filter_map(|iteration| iteration.stats_q_error_p50),
+            );
+            stats_q_error_p95_prev.extend(
+                prev.iterations
+                    .iter()
+                    .filter_map(|iteration| iteration.stats_q_error_p95),
+            );
+            stats_q_error_p95_new.extend(
+                query
+                    .iterations
+                    .iter()
+                    .filter_map(|iteration| iteration.stats_q_error_p95),
+            );
         }
     }
-    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"
+        );
+    }
+
+    print_q_error_comparison("QERR P50", stats_q_error_p50_prev, stats_q_error_p50_new);
+    print_q_error_comparison("QERR P95", stats_q_error_p95_prev, stats_q_error_p95_new);
+}
+
+fn print_q_error_comparison(label: &str, prev: Vec<f64>, new: Vec<f64>) {
+    match (median(prev), median(new)) {
+        (Some(prev), Some(new)) => println!("{label:>8}: prev={prev:.2}x, new={new:.2}x"),
+        (Some(prev), None) => println!("{label:>8}: prev={prev:.2}x, new=n/a"),
+        (None, Some(new)) => println!("{label:>8}: prev=n/a, new={new:.2}x"),
+        (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>(