From 5f34de829abb8abefad19894e66c6fd0e7d348f1 Mon Sep 17 00:00:00 2001 From: Geoffrey Claude Date: Thu, 2 Jul 2026 17:32:04 +0200 Subject: [PATCH] Avoid sort pushdown panic on missing file stats --- .../datasource/src/file_scan_config/mod.rs | 36 ++++++++++++++++ datafusion/datasource/src/statistics.rs | 43 ++++++++++++++++++- 2 files changed, 78 insertions(+), 1 deletion(-) diff --git a/datafusion/datasource/src/file_scan_config/mod.rs b/datafusion/datasource/src/file_scan_config/mod.rs index 4bf86e17d387d..eeca3669dd1f9 100644 --- a/datafusion/datasource/src/file_scan_config/mod.rs +++ b/datafusion/datasource/src/file_scan_config/mod.rs @@ -2592,6 +2592,16 @@ mod tests { )) } + fn make_file_with_empty_column_stats(name: &str) -> PartitionedFile { + PartitionedFile::new(name.to_string(), 1024).with_statistics(Arc::new( + Statistics { + num_rows: Precision::Absent, + total_byte_size: Precision::Absent, + column_statistics: vec![], + }, + )) + } + #[derive(Clone)] struct ExactSortPushdownSource { metrics: ExecutionPlanMetricsSet, @@ -2953,6 +2963,32 @@ mod tests { Ok(()) } + #[test] + fn sort_pushdown_unsupported_source_empty_column_statistics() -> Result<()> { + let file_schema = + Arc::new(Schema::new(vec![Field::new("a", DataType::Float64, false)])); + let table_schema = TableSchema::new(Arc::clone(&file_schema), vec![]); + let file_source = Arc::new(MockSource::new(table_schema)); + + let file_groups = vec![FileGroup::new(vec![ + make_file_with_empty_column_stats("file_b"), + make_file_with_empty_column_stats("file_a"), + ])]; + + let sort_expr = PhysicalSortExpr::new_default(Arc::new(Column::new("a", 0))); + let config = + FileScanConfigBuilder::new(ObjectStoreUrl::local_filesystem(), file_source) + .with_file_groups(file_groups) + .build(); + + let result = config.try_pushdown_sort(&[sort_expr])?; + assert!( + matches!(result, SortOrderPushdownResult::Unsupported), + "expected missing file statistics to skip sort pushdown" + ); + Ok(()) + } + #[test] fn sort_pushdown_inexact_source_with_statistics_sorting() -> Result<()> { let file_schema = diff --git a/datafusion/datasource/src/statistics.rs b/datafusion/datasource/src/statistics.rs index 6abfafe9d39d4..5fe52a36409fa 100644 --- a/datafusion/datasource/src/statistics.rs +++ b/datafusion/datasource/src/statistics.rs @@ -97,7 +97,10 @@ impl MinMaxStatistics { .zip(s.column_statistics[i].max_value.get_value().cloned()) .ok_or_else(|| plan_datafusion_err!("statistics not found")) } else { - let partition_value = &pv[i - s.column_statistics.len()]; + let Some(partition_value) = pv.get(i - s.column_statistics.len()) + else { + return Err(plan_datafusion_err!("statistics not found")); + }; Ok((partition_value.clone(), partition_value.clone())) } }) @@ -623,6 +626,44 @@ mod tests { fn file_with_stats(path: &str, stats: Statistics) -> PartitionedFile { PartitionedFile::new(path, 1).with_statistics(Arc::new(stats)) } + + fn file_with_empty_column_stats(path: &str) -> PartitionedFile { + PartitionedFile::new(path, 1).with_statistics(Arc::new(Statistics { + num_rows: Precision::Absent, + total_byte_size: Precision::Absent, + column_statistics: vec![], + })) + } + + #[test] + fn min_max_statistics_errors_when_column_and_partition_stats_are_missing() { + let schema = + Arc::new(Schema::new(vec![Field::new("a", DataType::Float64, true)])); + let sort_order = LexOrdering::new(vec![PhysicalSortExpr::new_default(Arc::new( + Column::new("a", 0), + ))]) + .unwrap(); + let files = vec![ + file_with_empty_column_stats("first.parquet"), + file_with_empty_column_stats("second.parquet"), + ]; + + let err = match MinMaxStatistics::new_from_files( + &sort_order, + &schema, + None, + files.iter(), + ) { + Ok(_) => panic!("expected missing statistics to return an error"), + Err(err) => err, + }; + + assert!( + err.to_string().contains("statistics not found"), + "unexpected error: {err}" + ); + } + #[tokio::test] #[expect(deprecated)] async fn test_get_statistics_with_limit_casts_first_file_sum_to_sum_type()