From 6b39df17a0d3b9463f33e8681ecf81df744b44d3 Mon Sep 17 00:00:00 2001 From: Gabriel Musat Mestre Date: Wed, 11 Feb 2026 16:36:43 +0100 Subject: [PATCH 01/10] Add dfbench statistics command --- benchmarks/src/bin/dfbench.rs | 5 +- benchmarks/src/lib.rs | 1 + benchmarks/src/statistics.rs | 653 ++++++++++++++++++ .../src/operator_statistics/mod.rs | 39 +- 4 files changed, 679 insertions(+), 19 deletions(-) create mode 100644 benchmarks/src/statistics.rs diff --git a/benchmarks/src/bin/dfbench.rs b/benchmarks/src/bin/dfbench.rs index 50dd99368b7f0..29cc8d63d2d8d 100644 --- a/benchmarks/src/bin/dfbench.rs +++ b/benchmarks/src/bin/dfbench.rs @@ -32,7 +32,8 @@ static ALLOC: snmalloc_rs::SnMalloc = snmalloc_rs::SnMalloc; static ALLOC: mimalloc::MiMalloc = mimalloc::MiMalloc; use datafusion_benchmarks::{ - cancellation, clickbench, dict, h2o, hj, imdb, nlj, smj, sort_tpch, tpcds, tpch, + cancellation, clickbench, dict, h2o, hj, imdb, nlj, smj, sort_tpch, statistics, + tpcds, tpch, }; #[derive(Debug, Parser)] @@ -52,6 +53,7 @@ enum Options { Imdb(imdb::RunOpt), Nlj(nlj::RunOpt), Smj(smj::RunOpt), + Statistics(statistics::RunOpt), SortPushdown(sort_pushdown::RunOpt), SortTpch(sort_tpch::RunOpt), Tpch(tpch::RunOpt), @@ -73,6 +75,7 @@ pub async fn main() -> Result<()> { Options::Imdb(opt) => Box::pin(opt.run()).await, Options::Nlj(opt) => opt.run().await, Options::Smj(opt) => opt.run().await, + Options::Statistics(opt) => opt.run().await, Options::SortPushdown(opt) => opt.run().await, Options::SortTpch(opt) => opt.run().await, Options::Tpch(opt) => Box::pin(opt.run()).await, diff --git a/benchmarks/src/lib.rs b/benchmarks/src/lib.rs index 7d8b7044bbdd8..0b3783421f840 100644 --- a/benchmarks/src/lib.rs +++ b/benchmarks/src/lib.rs @@ -29,6 +29,7 @@ pub mod sort_tpch; pub mod sql_benchmark; pub mod sql_benchmark_runner; pub mod sql_benchmark_suite; +pub mod statistics; pub mod tpcds; pub mod tpch; pub mod util; diff --git a/benchmarks/src/statistics.rs b/benchmarks/src/statistics.rs new file mode 100644 index 0000000000000..eb9237b204b8d --- /dev/null +++ b/benchmarks/src/statistics.rs @@ -0,0 +1,653 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Reports planning statistics alongside runtime metrics for benchmark queries. + +use std::collections::{BTreeMap, HashMap}; +use std::fs; +use std::path::{Path, PathBuf}; +use std::process::Command; +use std::sync::{Arc, LazyLock}; + +use clap::Args; +use datafusion::error::{DataFusionError, Result}; +use datafusion::physical_plan::metrics::MetricValue; +use datafusion::physical_plan::operator_statistics::StatisticsRegistry; +use datafusion::physical_plan::{ExecutionPlan, collect}; +use datafusion::prelude::{ParquetReadOptions, SessionConfig, SessionContext}; +use datafusion_common::stats::Precision; +use regex::Regex; +use serde::{Deserialize, Serialize}; + +/// Generate reports that compare planning statistics with runtime metrics. +#[derive(Debug, Args)] +#[command(verbatim_doc_comment)] +pub struct RunOpt { + /// Query filename stem. If not specified, runs every `.sql` file. + #[arg(short, long)] + query: Option, + + /// Branch whose results should be compared. Defaults to the previous run on this branch. + #[arg(long)] + compare: Option, + + /// Path to Parquet data. Top-level files and directories are registered as tables. + #[arg(required = true, short = 'p', long)] + path: PathBuf, + + /// Path to a SQL file or directory of SQL query files. + #[arg(required = true, short = 'Q', long = "query_path")] + query_path: PathBuf, +} + +impl RunOpt { + pub async fn run(self) -> Result<()> { + let mut config = SessionConfig::from_env()?.with_collect_statistics(true); + config.options_mut().optimizer.prefer_hash_join = true; + let ctx = SessionContext::new_with_config(config); + register_parquet_files(&ctx, &self.path).await?; + + let branch = current_branch_name(); + let result_path = self.report_path(&branch); + let comparison_branch = self + .compare + .as_deref() + .map_or(branch.as_str(), |branch| branch); + let comparison_path = self.report_path(comparison_branch); + let previous = load_comparison_report(&comparison_path)?; + backup_previous_report(&result_path)?; + let comparison_description = self.compare.as_ref().map_or_else( + || format!("previous run on branch '{branch}'"), + |branch| format!("branch '{branch}'"), + ); + + let mut reports = vec![]; + let mut successful_reports = vec![]; + for query_path in query_files(&self.query_path, self.query.as_deref())? { + let query = query_path + .file_stem() + .expect("query file has a filename") + .to_string_lossy() + .to_string(); + let sql = fs::read_to_string(query_path)?; + for (statement, sql) in sql + .split(';') + .filter(|sql| !sql.trim().is_empty()) + .enumerate() + { + let statement = statement + 1; + let report = match self.report_query(&ctx, sql).await { + Ok(operators) => QueryReport { + query: query.clone(), + statement, + operators, + success: true, + error: None, + }, + Err(error) => QueryReport { + query: query.clone(), + statement, + operators: vec![], + success: false, + error: Some(error.to_string()), + }, + }; + print_query_report(&report, previous.as_deref()); + if report.success { + successful_reports.push(report.clone()); + store_report(&result_path, &successful_reports)?; + } + reports.push(report); + } + } + print_q_error_summary( + &reports, + previous.as_deref(), + &branch, + &comparison_description, + ); + Ok(()) + } + + fn report_path(&self, branch: &str) -> PathBuf { + let report_name = self.query.as_ref().map_or_else( + || "statistics.json".to_string(), + |query| format!("statistics-{query}.json"), + ); + PathBuf::from("target/dfbench/statistics") + .join(normalize_branch_name(branch)) + .join(report_name) + } + + async fn report_query( + &self, + ctx: &SessionContext, + sql: &str, + ) -> Result> { + let dataframe = ctx.sql(sql).await?; + let (state, logical_plan) = dataframe.into_parts(); + let logical_plan = state.optimize(&logical_plan)?; + let physical_plan = state.create_physical_plan(&logical_plan).await?; + + let statistics = capture_statistics(physical_plan.as_ref())?; + collect(Arc::clone(&physical_plan), state.task_ctx()).await?; + + let mut report = Vec::with_capacity(statistics.len()); + append_runtime_metrics(physical_plan.as_ref(), &statistics, &mut report); + Ok(report) + } +} + +#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)] +struct QueryReport { + query: String, + statement: usize, + operators: Vec, + #[serde(default = "default_success")] + success: bool, + #[serde(default)] + error: Option, +} + +fn default_success() -> bool { + true +} + +#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)] +struct OperatorReport { + path: String, + name: String, + estimated_rows: StatisticValue, + estimated_bytes: StatisticValue, + runtime_output_rows: Option, + runtime_output_bytes: Option, + q_error: Option, +} + +#[derive(Debug, Serialize)] +struct CapturedStatistics { + path: String, + name: String, + estimated_rows: StatisticValue, + estimated_bytes: StatisticValue, +} + +#[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)] +#[serde(tag = "precision", content = "value", rename_all = "snake_case")] +enum StatisticValue { + Exact(usize), + Inexact(usize), + Absent, +} + +/// Symmetric multiplicative error between an estimated and actual row count. +/// +/// The q-error metric is infinite when exactly one value is zero. A zero +/// estimate for an actual zero result is reported separately as an exact +/// estimate and is excluded from q-error percentiles. +#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)] +#[serde(tag = "kind", content = "value", rename_all = "snake_case")] +enum QError { + Finite(f64), + Infinite, + ExactZero, +} + +fn capture_statistics(plan: &dyn ExecutionPlan) -> Result> { + let statistics_context = StatisticsRegistry::default_with_builtin_providers(); + let mut result = vec![]; + capture_statistics_inner(plan, &statistics_context, "0", &mut result)?; + Ok(result) +} + +fn capture_statistics_inner( + plan: &dyn ExecutionPlan, + statistics_context: &StatisticsRegistry, + path: &str, + result: &mut Vec, +) -> Result<()> { + let statistics = statistics_context.compute_base(plan)?; + result.push(CapturedStatistics { + path: path.to_string(), + name: plan.name().to_string(), + estimated_rows: statistic_value(statistics.num_rows), + estimated_bytes: statistic_value(statistics.total_byte_size), + }); + for (child_index, child) in plan.children().iter().enumerate() { + capture_statistics_inner( + child.as_ref(), + statistics_context, + &format!("{path}.{child_index}"), + result, + )?; + } + Ok(()) +} + +fn append_runtime_metrics( + plan: &dyn ExecutionPlan, + statistics: &[CapturedStatistics], + report: &mut Vec, +) { + let statistics_entry = &statistics[report.len()]; + let (runtime_output_rows, runtime_output_bytes) = + plan.metrics().map_or((None, None), |m| { + ( + m.output_rows(), + m.sum(|metric| matches!(metric.value(), MetricValue::OutputBytes(_))) + .map(|metric| metric.as_usize()), + ) + }); + report.push(OperatorReport { + path: statistics_entry.path.clone(), + name: statistics_entry.name.clone(), + estimated_rows: statistics_entry.estimated_rows.clone(), + estimated_bytes: statistics_entry.estimated_bytes.clone(), + runtime_output_rows, + runtime_output_bytes, + q_error: q_error(&statistics_entry.estimated_rows, runtime_output_rows), + }); + for child in plan.children() { + append_runtime_metrics(child.as_ref(), statistics, report); + } +} + +fn q_error(estimated: &StatisticValue, actual: Option) -> Option { + let estimated = match estimated { + StatisticValue::Exact(value) | StatisticValue::Inexact(value) => *value, + StatisticValue::Absent => return None, + }; + let actual = actual?; + if estimated == 0 && actual == 0 { + Some(QError::ExactZero) + } else if estimated == 0 || actual == 0 { + Some(QError::Infinite) + } else { + Some(QError::Finite( + estimated.max(actual) as f64 / estimated.min(actual) as f64, + )) + } +} + +fn statistic_value(value: Precision) -> StatisticValue { + match value { + Precision::Exact(value) => StatisticValue::Exact(value), + Precision::Inexact(value) => StatisticValue::Inexact(value), + Precision::Absent => StatisticValue::Absent, + } +} + +fn load_comparison_report(comparison_path: &Path) -> Result>> { + if comparison_path.exists() { + let previous = fs::read_to_string(comparison_path)?; + let previous: Vec = serde_json::from_str(&previous) + .map_err(|error| DataFusionError::External(Box::new(error)))?; + Ok(Some(previous)) + } else { + eprintln!( + "No comparison report found at {}", + comparison_path.display() + ); + Ok(None) + } +} + +fn backup_previous_report(result_path: &Path) -> Result<()> { + if result_path.exists() { + let previous_path = result_path.with_extension("previous.json"); + fs::copy(result_path, previous_path)?; + } + Ok(()) +} + +fn store_report(result_path: &Path, report: &[QueryReport]) -> Result<()> { + let parent = result_path.parent().expect("result path has a parent"); + fs::create_dir_all(parent)?; + let temporary_path = result_path.with_extension("tmp"); + fs::write(&temporary_path, serialize_report(report)?)?; + fs::rename(temporary_path, result_path)?; + Ok(()) +} + +fn current_branch_name() -> String { + Command::new("git") + .args(["rev-parse", "--abbrev-ref", "HEAD"]) + .output() + .ok() + .filter(|output| output.status.success()) + .and_then(|output| String::from_utf8(output.stdout).ok()) + .map(|branch| branch.trim().to_string()) + .filter(|branch| branch != "HEAD") + .unwrap_or_else(|| "detached".to_string()) +} + +fn normalize_branch_name(branch: &str) -> String { + branch + .chars() + .map(|character| match character { + 'a'..='z' | 'A'..='Z' | '0'..='9' | '-' | '_' | '.' => character, + _ => '_', + }) + .collect() +} + +fn print_query_report(query: &QueryReport, previous: Option<&[QueryReport]>) { + if !query.success { + println!( + "=== {} (statement {}) FAILED ===\n{}", + query.query, + query.statement, + query.error.as_deref().unwrap_or("unknown error") + ); + return; + } + let previous = previous.map(operator_reports); + println!("=== {} ===", query.query); + for operator in &query.operators { + let identifier = OperatorIdentifier::new(query, operator); + let previous_q_error = previous + .as_ref() + .and_then(|reports| reports.get(&identifier)) + .and_then(|operator| operator.q_error.as_ref()); + let depth = operator.path.matches('.').count(); + println!( + "{:indent$}{}: rows={} vs {}, q-error: previous={}, current={}, change={}", + "", + operator.name, + display_statistic(&operator.estimated_rows), + display_option(operator.runtime_output_rows), + display_q_error(previous_q_error), + display_q_error(operator.q_error.as_ref()), + display_improvement(previous_q_error, operator.q_error.as_ref()), + indent = depth * 2, + ); + } +} + +fn display_option(value: Option) -> String { + value.map_or_else(|| "?".to_string(), |value| value.to_string()) +} + +fn display_statistic(value: &StatisticValue) -> String { + match value { + StatisticValue::Exact(value) => format!("Exact({value})"), + StatisticValue::Inexact(value) => format!("Inexact({value})"), + StatisticValue::Absent => "Absent".to_string(), + } +} + +fn display_q_error(value: Option<&QError>) -> String { + match value { + Some(QError::Finite(value)) => format!("{value:.2}x"), + Some(QError::Infinite) => "infinite".to_string(), + Some(QError::ExactZero) => "exact zero".to_string(), + None => "?".to_string(), + } +} + +fn display_improvement(previous: Option<&QError>, current: Option<&QError>) -> String { + match (previous, current) { + (Some(QError::Finite(previous)), Some(QError::Finite(current))) => { + let improvement = (previous - current) / previous * 100.0; + if improvement >= 20.0 { + format!("✅ {improvement:.1}%") + } else if improvement <= -20.0 { + format!("❌ {improvement:.1}%") + } else { + format!("{improvement:.1}%") + } + } + (Some(QError::Infinite), Some(QError::Infinite)) => "0.0%".to_string(), + (Some(QError::ExactZero), Some(QError::ExactZero)) => "0.0%".to_string(), + (Some(QError::Infinite | QError::Finite(_)), Some(QError::ExactZero)) => { + "✅ exact zero".to_string() + } + (Some(QError::ExactZero), Some(QError::Infinite | QError::Finite(_))) => { + "❌ no longer exact zero".to_string() + } + (Some(QError::Infinite), Some(QError::Finite(_))) => "✅ resolved".to_string(), + (Some(QError::Finite(_)), Some(QError::Infinite)) => "❌ infinite".to_string(), + _ => "?".to_string(), + } +} + +fn print_q_error_summary( + reports: &[QueryReport], + previous: Option<&[QueryReport]>, + branch: &str, + comparison_description: &str, +) { + let q_errors = sorted_finite_q_errors(reports); + let previous_q_errors = previous.map(sorted_finite_q_errors); + let q_error_counts = q_error_counts(reports); + let total = reports + .iter() + .map(|query| query.operators.len()) + .sum::(); + + println!("=== q-error summary ==="); + println!("current: branch '{branch}'"); + println!("comparison: {comparison_description}"); + let failed = reports.iter().filter(|query| !query.success).count(); + println!( + "queries: {} succeeded, {failed} failed", + reports.len() - failed + ); + println!( + "evaluated operators: {}/{}", + q_error_counts.evaluated, total + ); + println!( + "finite q-errors: {}, exact-zero estimates: {}, infinite q-errors: {}", + q_error_counts.finite, q_error_counts.exact_zero, q_error_counts.infinite + ); + for percentile in [50, 75, 95, 99] { + let current = percentile_q_error(&q_errors, percentile); + let previous = previous_q_errors + .as_ref() + .and_then(|q_errors| percentile_q_error(q_errors, percentile)); + println!( + "p{percentile}: previous={}, current={}, change={}", + display_q_error(previous), + display_q_error(current), + display_improvement(previous, current), + ); + } +} + +struct QErrorCounts { + evaluated: usize, + finite: usize, + exact_zero: usize, + infinite: usize, +} + +fn q_error_counts(reports: &[QueryReport]) -> QErrorCounts { + let mut counts = QErrorCounts { + evaluated: 0, + finite: 0, + exact_zero: 0, + infinite: 0, + }; + for q_error in reports + .iter() + .flat_map(|query| &query.operators) + .filter_map(|operator| operator.q_error.as_ref()) + { + counts.evaluated += 1; + match q_error { + QError::Finite(_) => counts.finite += 1, + QError::ExactZero => counts.exact_zero += 1, + QError::Infinite => counts.infinite += 1, + } + } + counts +} + +fn sorted_finite_q_errors(reports: &[QueryReport]) -> Vec<&QError> { + let mut q_errors = reports + .iter() + .flat_map(|query| &query.operators) + .filter_map(|operator| match operator.q_error.as_ref() { + Some(QError::Finite(_)) => operator.q_error.as_ref(), + Some(QError::ExactZero | QError::Infinite) | None => None, + }) + .collect::>(); + q_errors.sort_by(|left, right| q_error_value(left).total_cmp(&q_error_value(right))); + q_errors +} + +fn percentile_q_error<'a>( + q_errors: &[&'a QError], + percentile: usize, +) -> Option<&'a QError> { + if q_errors.is_empty() { + return None; + } + let index = (percentile * q_errors.len()) + .div_ceil(100) + .saturating_sub(1); + q_errors.get(index).copied() +} + +fn q_error_value(q_error: &QError) -> f64 { + match q_error { + QError::Finite(value) => *value, + QError::Infinite => f64::INFINITY, + QError::ExactZero => 1.0, + } +} + +fn operator_reports( + reports: &[QueryReport], +) -> HashMap, &OperatorReport> { + reports + .iter() + .flat_map(|query| { + query + .operators + .iter() + .map(move |operator| (OperatorIdentifier::new(query, operator), operator)) + }) + .collect() +} + +#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] +struct OperatorIdentifier<'a> { + query: &'a str, + statement: usize, + path: &'a str, + name: &'a str, +} + +impl<'a> OperatorIdentifier<'a> { + fn new(query: &'a QueryReport, operator: &'a OperatorReport) -> Self { + Self { + query: query.query.as_str(), + statement: query.statement, + path: operator.path.as_str(), + name: operator.name.as_str(), + } + } +} + +fn serialize_report(report: &[QueryReport]) -> Result { + serde_json::to_string_pretty(report) + .map_err(|error| DataFusionError::External(Box::new(error))) +} + +fn query_files(path: &Path, query: Option<&str>) -> Result> { + if path.is_file() { + return Ok(vec![path.to_path_buf()]); + } + let mut files = fs::read_dir(path)? + .filter_map(|entry| entry.ok().map(|entry| entry.path())) + .filter(|path| path.extension().is_some_and(|extension| extension == "sql")) + .filter(|path| { + query.is_none_or(|query| path.file_stem().is_some_and(|name| name == query)) + }) + .collect::>(); + files.sort_by(|left, right| { + query_number(left) + .cmp(&query_number(right)) + .then_with(|| left.cmp(right)) + }); + Ok(files) +} + +fn query_number(path: &Path) -> Option { + static QUERY_NUMBER: LazyLock = LazyLock::new(|| Regex::new(r"\d+").unwrap()); + let filename = path.file_stem()?.to_str()?; + QUERY_NUMBER.find(filename)?.as_str().parse().ok() +} + +async fn register_parquet_files(ctx: &SessionContext, path: &Path) -> Result<()> { + let mut files = vec![]; + collect_parquet_files(path, &mut files)?; + + let mut tables = BTreeMap::new(); + for file in files { + let parent = file.parent().expect("Parquet file has a parent directory"); + let relative_parent = parent + .strip_prefix(path) + .expect("Parquet file is within the data path"); + let (table, table_path) = if relative_parent.as_os_str().is_empty() { + ( + file.file_stem() + .expect("Parquet file has a filename") + .to_string_lossy() + .to_string(), + file, + ) + } else { + let table = relative_parent + .components() + .next() + .expect("relative table directory has a component") + .as_os_str() + .to_string_lossy() + .to_string(); + (table.clone(), path.join(table)) + }; + tables.insert(table, table_path); + } + + for (table, table_path) in tables { + ctx.register_parquet( + table, + table_path.to_string_lossy(), + ParquetReadOptions::default(), + ) + .await?; + } + Ok(()) +} + +fn collect_parquet_files(path: &Path, files: &mut Vec) -> Result<()> { + for entry in fs::read_dir(path)? { + let path = entry?.path(); + if path.is_dir() { + collect_parquet_files(&path, files)?; + } else if path + .extension() + .is_some_and(|extension| extension == "parquet") + { + files.push(path); + } + } + Ok(()) +} diff --git a/datafusion/physical-plan/src/operator_statistics/mod.rs b/datafusion/physical-plan/src/operator_statistics/mod.rs index 142768fcf49d2..8c8547accbb9f 100644 --- a/datafusion/physical-plan/src/operator_statistics/mod.rs +++ b/datafusion/physical-plan/src/operator_statistics/mod.rs @@ -784,16 +784,16 @@ impl StatisticsProvider for JoinStatisticsProvider { /// Estimate equi-join output using NDV of join key columns: /// left_rows * right_rows / product(max(left_ndv_i, right_ndv_i)) - /// Falls back to Cartesian product if any key lacks NDV on both sides. + /// Returns `None` when any key lacks NDV information. fn equi_join_estimate( on: JoinOnRef, left: &Statistics, right: &Statistics, left_rows: usize, right_rows: usize, - ) -> usize { + ) -> Option { if on.is_empty() { - return left_rows.saturating_mul(right_rows); + return Some(left_rows.saturating_mul(right_rows)); } let mut ndv_divisor: usize = 1; for (left_key, right_key) in on { @@ -809,21 +809,28 @@ impl StatisticsProvider for JoinStatisticsProvider { (Some(l), Some(r)) if l > 0 && r > 0 => { ndv_divisor = ndv_divisor.saturating_mul(l.max(r)); } - _ => return left_rows.saturating_mul(right_rows), + _ => return None, } } let max_rows = left_rows.saturating_mul(right_rows); - max_rows.checked_div(ndv_divisor).unwrap_or(max_rows) + Some(max_rows.checked_div(ndv_divisor).unwrap_or(max_rows)) } let (inner_estimate, is_exact_cartesian, join_type) = if let Some(hash_join) = plan.downcast_ref::() { - let est = - equi_join_estimate(hash_join.on(), left, right, left_rows, right_rows); + let Some(est) = + equi_join_estimate(hash_join.on(), left, right, left_rows, right_rows) + else { + return Ok(StatisticsResult::Delegate); + }; (est, false, *hash_join.join_type()) } else if let Some(smj) = plan.downcast_ref::() { - let est = equi_join_estimate(smj.on(), left, right, left_rows, right_rows); + let Some(est) = + equi_join_estimate(smj.on(), left, right, left_rows, right_rows) + else { + return Ok(StatisticsResult::Delegate); + }; (est, false, smj.join_type()) } else if plan.downcast_ref::().is_some() { let both_exact = left.num_rows.is_exact().unwrap_or(false) @@ -1026,7 +1033,7 @@ mod tests { use super::*; use crate::filter::FilterExec; use crate::projection::ProjectionExec; - use crate::statistics::StatisticsArgs; + use crate::statistics::{StatisticsArgs, StatisticsContext}; use crate::{DisplayAs, DisplayFormatType, PlanProperties}; use arrow::datatypes::{DataType, Field, Schema}; use datafusion_common::stats::Precision; @@ -1908,8 +1915,8 @@ mod tests { } #[test] - fn test_join_provider_fallback_cartesian() -> Result<()> { - // No NDV available -> Cartesian product estimate + fn test_join_provider_delegates_without_ndv() -> Result<()> { + // No NDV available -> use the built-in join statistics. let left = make_source_with_ndv_2col(100, None); let right = make_source_with_ndv_2col(200, None); let join = make_hash_join(left, right)?; @@ -1919,7 +1926,9 @@ mod tests { Arc::new(DefaultStatisticsProvider), ]); let stats = registry.compute(join.as_ref())?; - assert_eq!(stats.base.num_rows, Precision::Inexact(20_000)); + let expected = + StatisticsContext::new().compute(join.as_ref(), &StatisticsArgs::new())?; + assert_eq!(stats.base.num_rows, expected.num_rows); Ok(()) } @@ -2036,12 +2045,6 @@ mod tests { compute_join_rows(1000, Some(100), 500, Some(50), JoinType::RightSemi)?, Precision::Inexact(500) ); - // Cartesian fallback (no NDV): inner = 1000*500 = 500000, - // left semi = min(500000, 1000) = 1000 (selectivity = 1.0) - assert_eq!( - compute_join_rows(1000, None, 500, None, JoinType::LeftSemi)?, - Precision::Inexact(1000) - ); Ok(()) } From a2123e70e7b8f76e24b996d6212c4f8793bdd41f Mon Sep 17 00:00:00 2001 From: Gabriel Date: Tue, 4 Aug 2026 15:29:09 +0200 Subject: [PATCH 02/10] Use parser for splitting statements --- benchmarks/src/statistics.rs | 16 +++++++++++----- 1 file changed, 11 insertions(+), 5 deletions(-) diff --git a/benchmarks/src/statistics.rs b/benchmarks/src/statistics.rs index eb9237b204b8d..b7781dbd33412 100644 --- a/benchmarks/src/statistics.rs +++ b/benchmarks/src/statistics.rs @@ -29,6 +29,7 @@ use datafusion::physical_plan::metrics::MetricValue; use datafusion::physical_plan::operator_statistics::StatisticsRegistry; use datafusion::physical_plan::{ExecutionPlan, collect}; use datafusion::prelude::{ParquetReadOptions, SessionConfig, SessionContext}; +use datafusion::sql::parser::DFParser; use datafusion_common::stats::Precision; use regex::Regex; use serde::{Deserialize, Serialize}; @@ -84,11 +85,7 @@ impl RunOpt { .to_string_lossy() .to_string(); let sql = fs::read_to_string(query_path)?; - for (statement, sql) in sql - .split(';') - .filter(|sql| !sql.trim().is_empty()) - .enumerate() - { + for (statement, sql) in sql_statements(&sql)?.iter().enumerate() { let statement = statement + 1; let report = match self.report_query(&ctx, sql).await { Ok(operators) => QueryReport { @@ -570,6 +567,15 @@ fn serialize_report(report: &[QueryReport]) -> Result { .map_err(|error| DataFusionError::External(Box::new(error))) } +fn sql_statements(sql: &str) -> Result> { + DFParser::parse_sql(sql).map(|statements| { + statements + .into_iter() + .map(|statement| statement.to_string()) + .collect() + }) +} + fn query_files(path: &Path, query: Option<&str>) -> Result> { if path.is_file() { return Ok(vec![path.to_path_buf()]); From 4d6565e001761d802f794861cd240bf99f747462 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Tue, 4 Aug 2026 16:35:00 +0200 Subject: [PATCH 03/10] Store reports even on failures --- benchmarks/src/statistics.rs | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/benchmarks/src/statistics.rs b/benchmarks/src/statistics.rs index b7781dbd33412..e785fffe5ed5a 100644 --- a/benchmarks/src/statistics.rs +++ b/benchmarks/src/statistics.rs @@ -77,7 +77,6 @@ impl RunOpt { ); let mut reports = vec![]; - let mut successful_reports = vec![]; for query_path in query_files(&self.query_path, self.query.as_deref())? { let query = query_path .file_stem() @@ -104,11 +103,8 @@ impl RunOpt { }, }; print_query_report(&report, previous.as_deref()); - if report.success { - successful_reports.push(report.clone()); - store_report(&result_path, &successful_reports)?; - } reports.push(report); + store_report(&result_path, &reports)?; } } print_q_error_summary( From ffd58c4b50ce6e4ee83b532687a7085dc526cc26 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Tue, 4 Aug 2026 16:36:33 +0200 Subject: [PATCH 04/10] Avoid registering a duplicate table --- benchmarks/src/statistics.rs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/benchmarks/src/statistics.rs b/benchmarks/src/statistics.rs index e785fffe5ed5a..2373dd301cff8 100644 --- a/benchmarks/src/statistics.rs +++ b/benchmarks/src/statistics.rs @@ -30,6 +30,7 @@ use datafusion::physical_plan::operator_statistics::StatisticsRegistry; use datafusion::physical_plan::{ExecutionPlan, collect}; use datafusion::prelude::{ParquetReadOptions, SessionConfig, SessionContext}; use datafusion::sql::parser::DFParser; +use datafusion_common::config_err; use datafusion_common::stats::Precision; use regex::Regex; use serde::{Deserialize, Serialize}; @@ -625,6 +626,9 @@ async fn register_parquet_files(ctx: &SessionContext, path: &Path) -> Result<()> .to_string(); (table.clone(), path.join(table)) }; + if tables.contains_key(&table) { + return config_err!("Tried to register duplicate table {table}"); + } tables.insert(table, table_path); } From e1dad5cbb8da019839e1bbcb8016f97407da6990 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Tue, 4 Aug 2026 16:45:56 +0200 Subject: [PATCH 05/10] Rollback unintended changes --- .../src/operator_statistics/mod.rs | 39 +++++++++---------- 1 file changed, 18 insertions(+), 21 deletions(-) diff --git a/datafusion/physical-plan/src/operator_statistics/mod.rs b/datafusion/physical-plan/src/operator_statistics/mod.rs index 8c8547accbb9f..142768fcf49d2 100644 --- a/datafusion/physical-plan/src/operator_statistics/mod.rs +++ b/datafusion/physical-plan/src/operator_statistics/mod.rs @@ -784,16 +784,16 @@ impl StatisticsProvider for JoinStatisticsProvider { /// Estimate equi-join output using NDV of join key columns: /// left_rows * right_rows / product(max(left_ndv_i, right_ndv_i)) - /// Returns `None` when any key lacks NDV information. + /// Falls back to Cartesian product if any key lacks NDV on both sides. fn equi_join_estimate( on: JoinOnRef, left: &Statistics, right: &Statistics, left_rows: usize, right_rows: usize, - ) -> Option { + ) -> usize { if on.is_empty() { - return Some(left_rows.saturating_mul(right_rows)); + return left_rows.saturating_mul(right_rows); } let mut ndv_divisor: usize = 1; for (left_key, right_key) in on { @@ -809,28 +809,21 @@ impl StatisticsProvider for JoinStatisticsProvider { (Some(l), Some(r)) if l > 0 && r > 0 => { ndv_divisor = ndv_divisor.saturating_mul(l.max(r)); } - _ => return None, + _ => return left_rows.saturating_mul(right_rows), } } let max_rows = left_rows.saturating_mul(right_rows); - Some(max_rows.checked_div(ndv_divisor).unwrap_or(max_rows)) + max_rows.checked_div(ndv_divisor).unwrap_or(max_rows) } let (inner_estimate, is_exact_cartesian, join_type) = if let Some(hash_join) = plan.downcast_ref::() { - let Some(est) = - equi_join_estimate(hash_join.on(), left, right, left_rows, right_rows) - else { - return Ok(StatisticsResult::Delegate); - }; + let est = + equi_join_estimate(hash_join.on(), left, right, left_rows, right_rows); (est, false, *hash_join.join_type()) } else if let Some(smj) = plan.downcast_ref::() { - let Some(est) = - equi_join_estimate(smj.on(), left, right, left_rows, right_rows) - else { - return Ok(StatisticsResult::Delegate); - }; + let est = equi_join_estimate(smj.on(), left, right, left_rows, right_rows); (est, false, smj.join_type()) } else if plan.downcast_ref::().is_some() { let both_exact = left.num_rows.is_exact().unwrap_or(false) @@ -1033,7 +1026,7 @@ mod tests { use super::*; use crate::filter::FilterExec; use crate::projection::ProjectionExec; - use crate::statistics::{StatisticsArgs, StatisticsContext}; + use crate::statistics::StatisticsArgs; use crate::{DisplayAs, DisplayFormatType, PlanProperties}; use arrow::datatypes::{DataType, Field, Schema}; use datafusion_common::stats::Precision; @@ -1915,8 +1908,8 @@ mod tests { } #[test] - fn test_join_provider_delegates_without_ndv() -> Result<()> { - // No NDV available -> use the built-in join statistics. + fn test_join_provider_fallback_cartesian() -> Result<()> { + // No NDV available -> Cartesian product estimate let left = make_source_with_ndv_2col(100, None); let right = make_source_with_ndv_2col(200, None); let join = make_hash_join(left, right)?; @@ -1926,9 +1919,7 @@ mod tests { Arc::new(DefaultStatisticsProvider), ]); let stats = registry.compute(join.as_ref())?; - let expected = - StatisticsContext::new().compute(join.as_ref(), &StatisticsArgs::new())?; - assert_eq!(stats.base.num_rows, expected.num_rows); + assert_eq!(stats.base.num_rows, Precision::Inexact(20_000)); Ok(()) } @@ -2045,6 +2036,12 @@ mod tests { compute_join_rows(1000, Some(100), 500, Some(50), JoinType::RightSemi)?, Precision::Inexact(500) ); + // Cartesian fallback (no NDV): inner = 1000*500 = 500000, + // left semi = min(500000, 1000) = 1000 (selectivity = 1.0) + assert_eq!( + compute_join_rows(1000, None, 500, None, JoinType::LeftSemi)?, + Precision::Inexact(1000) + ); Ok(()) } From 4f1c721d52d2baecd7a7a44c439986cd8e6e7832 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Tue, 4 Aug 2026 17:03:47 +0200 Subject: [PATCH 06/10] Add some focused unit tests --- benchmarks/src/statistics.rs | 44 ++++++++++++++++++++++++++++++++++++ 1 file changed, 44 insertions(+) diff --git a/benchmarks/src/statistics.rs b/benchmarks/src/statistics.rs index 2373dd301cff8..ad8c94978ea3c 100644 --- a/benchmarks/src/statistics.rs +++ b/benchmarks/src/statistics.rs @@ -657,3 +657,47 @@ fn collect_parquet_files(path: &Path, files: &mut Vec) -> Result<()> { } Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + use tempfile::tempdir; + + #[test] + fn parses_sql_statements_without_splitting_string_literals() { + let statements = sql_statements("SELECT ';'; SELECT 2").unwrap(); + + assert_eq!(statements.len(), 2); + assert!(statements[0].contains("';'")); + } + + #[test] + fn persists_failed_reports() { + let directory = tempdir().unwrap(); + let path = directory.path().join("statistics.json"); + let reports = vec![QueryReport { + query: "q1".to_string(), + statement: 1, + operators: vec![], + success: false, + error: Some("expected failure".to_string()), + }]; + + store_report(&path, &reports).unwrap(); + assert_eq!(load_comparison_report(&path).unwrap(), Some(reports)); + } + + #[tokio::test] + async fn rejects_duplicate_table_names() { + let directory = tempdir().unwrap(); + let path = directory.path(); + fs::write(path.join("foo.parquet"), b"").unwrap(); + fs::create_dir(path.join("foo")).unwrap(); + fs::write(path.join("foo/part.parquet"), b"").unwrap(); + + let error = register_parquet_files(&SessionContext::new(), path) + .await + .unwrap_err(); + assert!(error.to_string().contains("duplicate table foo")); + } +} From e2f79d3ce77cea17103087239b8361efb9f218ef Mon Sep 17 00:00:00 2001 From: Gabriel Date: Thu, 6 Aug 2026 12:11:24 +0200 Subject: [PATCH 07/10] Do not abort on parse errors --- benchmarks/src/statistics.rs | 72 +++++++++++++++++++++++++----------- 1 file changed, 50 insertions(+), 22 deletions(-) diff --git a/benchmarks/src/statistics.rs b/benchmarks/src/statistics.rs index ad8c94978ea3c..428e335cc3f22 100644 --- a/benchmarks/src/statistics.rs +++ b/benchmarks/src/statistics.rs @@ -17,7 +17,7 @@ //! Reports planning statistics alongside runtime metrics for benchmark queries. -use std::collections::{BTreeMap, HashMap}; +use std::collections::{BTreeMap, HashMap, VecDeque}; use std::fs; use std::path::{Path, PathBuf}; use std::process::Command; @@ -29,7 +29,7 @@ use datafusion::physical_plan::metrics::MetricValue; use datafusion::physical_plan::operator_statistics::StatisticsRegistry; use datafusion::physical_plan::{ExecutionPlan, collect}; use datafusion::prelude::{ParquetReadOptions, SessionConfig, SessionContext}; -use datafusion::sql::parser::DFParser; +use datafusion::sql::parser::{DFParser, Statement}; use datafusion_common::config_err; use datafusion_common::stats::Precision; use regex::Regex; @@ -85,9 +85,25 @@ impl RunOpt { .to_string_lossy() .to_string(); let sql = fs::read_to_string(query_path)?; - for (statement, sql) in sql_statements(&sql)?.iter().enumerate() { + let statements = match sql_statements(&sql) { + Ok(statements) => statements, + Err(error) => { + let report = QueryReport { + query: query.clone(), + statement: 1, + operators: vec![], + success: false, + error: Some(error.to_string()), + }; + print_query_report(&report, previous.as_deref()); + reports.push(report); + store_report(&result_path, &reports)?; + continue; + } + }; + for (statement, sql) in statements.into_iter().enumerate() { let statement = statement + 1; - let report = match self.report_query(&ctx, sql).await { + let report = match self.report_statement(&ctx, sql).await { Ok(operators) => QueryReport { query: query.clone(), statement, @@ -127,13 +143,13 @@ impl RunOpt { .join(report_name) } - async fn report_query( + async fn report_statement( &self, ctx: &SessionContext, - sql: &str, + statement: Statement, ) -> Result> { - let dataframe = ctx.sql(sql).await?; - let (state, logical_plan) = dataframe.into_parts(); + let state = ctx.state(); + let logical_plan = state.statement_to_plan(statement).await?; let logical_plan = state.optimize(&logical_plan)?; let physical_plan = state.create_physical_plan(&logical_plan).await?; @@ -564,13 +580,8 @@ fn serialize_report(report: &[QueryReport]) -> Result { .map_err(|error| DataFusionError::External(Box::new(error))) } -fn sql_statements(sql: &str) -> Result> { - DFParser::parse_sql(sql).map(|statements| { - statements - .into_iter() - .map(|statement| statement.to_string()) - .collect() - }) +fn sql_statements(sql: &str) -> Result> { + DFParser::parse_sql(sql) } fn query_files(path: &Path, query: Option<&str>) -> Result> { @@ -602,7 +613,7 @@ async fn register_parquet_files(ctx: &SessionContext, path: &Path) -> Result<()> let mut files = vec![]; collect_parquet_files(path, &mut files)?; - let mut tables = BTreeMap::new(); + let mut tables = BTreeMap::::new(); for file in files { let parent = file.parent().expect("Parquet file has a parent directory"); let relative_parent = parent @@ -626,10 +637,17 @@ async fn register_parquet_files(ctx: &SessionContext, path: &Path) -> Result<()> .to_string(); (table.clone(), path.join(table)) }; - if tables.contains_key(&table) { - return config_err!("Tried to register duplicate table {table}"); + if let Some(existing_path) = tables.get(&table) { + if existing_path != &table_path { + return config_err!( + "Tried to register duplicate table {table} from '{}' and '{}'", + existing_path.display(), + table_path.display() + ); + } + } else { + tables.insert(table, table_path); } - tables.insert(table, table_path); } for (table, table_path) in tables { @@ -664,11 +682,21 @@ mod tests { use tempfile::tempdir; #[test] - fn parses_sql_statements_without_splitting_string_literals() { - let statements = sql_statements("SELECT ';'; SELECT 2").unwrap(); + fn parses_statements_without_losing_external_table_details() { + let statements = sql_statements( + "CREATE EXTERNAL TABLE t(c1 int) STORED AS CSV \ + PARTITIONED BY (p1, p2) LOCATION 'foo.csv' \ + OPTIONS (format.delimiter '|'); SELECT ';'", + ) + .unwrap(); assert_eq!(statements.len(), 2); - assert!(statements[0].contains("';'")); + let Statement::CreateExternalTable(table) = &statements[0] else { + panic!("expected CREATE EXTERNAL TABLE"); + }; + assert_eq!(table.columns.len(), 1); + assert_eq!(table.table_partition_cols, ["p1", "p2"]); + assert_eq!(table.options.len(), 1); } #[test] From e18b8213807dbc09792d016f6621bb0dfd4e84cf Mon Sep 17 00:00:00 2001 From: Gabriel Date: Tue, 11 Aug 2026 16:00:44 +0200 Subject: [PATCH 08/10] Use dialect from SessionContext --- benchmarks/src/statistics.rs | 37 ++++++++++++++++++++++++++++++++---- 1 file changed, 33 insertions(+), 4 deletions(-) diff --git a/benchmarks/src/statistics.rs b/benchmarks/src/statistics.rs index 428e335cc3f22..34a0fb2c87006 100644 --- a/benchmarks/src/statistics.rs +++ b/benchmarks/src/statistics.rs @@ -29,7 +29,9 @@ use datafusion::physical_plan::metrics::MetricValue; use datafusion::physical_plan::operator_statistics::StatisticsRegistry; use datafusion::physical_plan::{ExecutionPlan, collect}; use datafusion::prelude::{ParquetReadOptions, SessionConfig, SessionContext}; -use datafusion::sql::parser::{DFParser, Statement}; +use datafusion::sql::parser::{DFParserBuilder, Statement}; +use datafusion::sql::sqlparser::dialect::dialect_from_str; +use datafusion_common::config::{Dialect, SqlParserOptions}; use datafusion_common::config_err; use datafusion_common::stats::Precision; use regex::Regex; @@ -61,6 +63,7 @@ impl RunOpt { let mut config = SessionConfig::from_env()?.with_collect_statistics(true); config.options_mut().optimizer.prefer_hash_join = true; let ctx = SessionContext::new_with_config(config); + let sql_parser_options = ctx.state().config_options().sql_parser.clone(); register_parquet_files(&ctx, &self.path).await?; let branch = current_branch_name(); @@ -85,7 +88,7 @@ impl RunOpt { .to_string_lossy() .to_string(); let sql = fs::read_to_string(query_path)?; - let statements = match sql_statements(&sql) { + let statements = match sql_statements(&sql, &sql_parser_options) { Ok(statements) => statements, Err(error) => { let report = QueryReport { @@ -580,8 +583,20 @@ fn serialize_report(report: &[QueryReport]) -> Result { .map_err(|error| DataFusionError::External(Box::new(error))) } -fn sql_statements(sql: &str) -> Result> { - DFParser::parse_sql(sql) +fn sql_statements(sql: &str, options: &SqlParserOptions) -> Result> { + let dialect = dialect_from_str(options.dialect).ok_or_else(|| { + DataFusionError::Plan(format!( + "Unsupported SQL dialect: {}. Available dialects: {}.", + options.dialect, + Dialect::available() + )) + })?; + + DFParserBuilder::new(sql) + .with_dialect(dialect.as_ref()) + .with_recursion_limit(options.recursion_limit.get()) + .build()? + .parse_statements() } fn query_files(path: &Path, query: Option<&str>) -> Result> { @@ -687,6 +702,7 @@ mod tests { "CREATE EXTERNAL TABLE t(c1 int) STORED AS CSV \ PARTITIONED BY (p1, p2) LOCATION 'foo.csv' \ OPTIONS (format.delimiter '|'); SELECT ';'", + &SessionConfig::new().options().sql_parser, ) .unwrap(); @@ -699,6 +715,19 @@ mod tests { assert_eq!(table.options.len(), 1); } + #[test] + fn parses_statements_with_session_sql_dialect() { + let sql = "# MySQL comment\nSELECT 1"; + assert!(sql_statements(sql, &SessionConfig::new().options().sql_parser).is_err()); + + let mut config = SessionConfig::new(); + config.options_mut().sql_parser.dialect = Dialect::MySQL; + + let statements = sql_statements(sql, &config.options().sql_parser).unwrap(); + + assert_eq!(statements.len(), 1); + } + #[test] fn persists_failed_reports() { let directory = tempdir().unwrap(); From a5085cdfc424f6c8fd78df6709cb3c4b4f2028ad Mon Sep 17 00:00:00 2001 From: Gabriel Date: Wed, 12 Aug 2026 13:28:59 +0200 Subject: [PATCH 09/10] Route Ddl and Statements through the Context --- benchmarks/src/statistics.rs | 40 ++++++++++++++++++++++++++++++++++++ 1 file changed, 40 insertions(+) diff --git a/benchmarks/src/statistics.rs b/benchmarks/src/statistics.rs index 34a0fb2c87006..027a0d12f88ba 100644 --- a/benchmarks/src/statistics.rs +++ b/benchmarks/src/statistics.rs @@ -25,6 +25,7 @@ use std::sync::{Arc, LazyLock}; use clap::Args; use datafusion::error::{DataFusionError, Result}; +use datafusion::logical_expr::LogicalPlan; use datafusion::physical_plan::metrics::MetricValue; use datafusion::physical_plan::operator_statistics::StatisticsRegistry; use datafusion::physical_plan::{ExecutionPlan, collect}; @@ -153,6 +154,16 @@ impl RunOpt { ) -> Result> { let state = ctx.state(); let logical_plan = state.statement_to_plan(statement).await?; + if matches!( + logical_plan, + LogicalPlan::Ddl(_) | LogicalPlan::Statement(_) + ) { + ctx.execute_logical_plan(logical_plan) + .await? + .collect() + .await?; + return Ok(vec![]); + } let logical_plan = state.optimize(&logical_plan)?; let physical_plan = state.create_physical_plan(&logical_plan).await?; @@ -728,6 +739,35 @@ mod tests { assert_eq!(statements.len(), 1); } + #[tokio::test] + async fn applies_session_changing_statements_before_reporting_queries() { + let directory = tempdir().unwrap(); + let options = RunOpt { + query: None, + compare: None, + path: directory.path().to_path_buf(), + query_path: directory.path().to_path_buf(), + }; + let ctx = SessionContext::new(); + let mut statements = sql_statements( + "CREATE TABLE x AS VALUES (1); SELECT * FROM x", + &SessionConfig::new().options().sql_parser, + ) + .unwrap(); + + let reports = options + .report_statement(&ctx, statements.pop_front().unwrap()) + .await + .unwrap(); + assert!(reports.is_empty()); + + let reports = options + .report_statement(&ctx, statements.pop_front().unwrap()) + .await + .unwrap(); + assert!(!reports.is_empty()); + } + #[test] fn persists_failed_reports() { let directory = tempdir().unwrap(); From 2c9a58c958a6b64cdc785fbf8b4477e1c92f56b3 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Sun, 16 Aug 2026 18:29:29 +0200 Subject: [PATCH 10/10] Address feedbac --- benchmarks/src/statistics.rs | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/benchmarks/src/statistics.rs b/benchmarks/src/statistics.rs index 027a0d12f88ba..3f17d0d547b54 100644 --- a/benchmarks/src/statistics.rs +++ b/benchmarks/src/statistics.rs @@ -39,6 +39,9 @@ use regex::Regex; use serde::{Deserialize, Serialize}; /// Generate reports that compare planning statistics with runtime metrics. +/// +/// Parser options are captured when the run starts, so `SET` statements do not +/// affect parsing in later query files. #[derive(Debug, Args)] #[command(verbatim_doc_comment)] pub struct RunOpt { @@ -101,7 +104,6 @@ impl RunOpt { }; print_query_report(&report, previous.as_deref()); reports.push(report); - store_report(&result_path, &reports)?; continue; } }; @@ -125,9 +127,9 @@ impl RunOpt { }; print_query_report(&report, previous.as_deref()); reports.push(report); - store_report(&result_path, &reports)?; } } + store_report(&result_path, &reports)?; print_q_error_summary( &reports, previous.as_deref(),