[branch-54] Avoid sort pushdown panic on incomplete file statistics

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 =

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()