Wrong aggregation arguments error.

datafusion/src/physical_plan/aggregates.rs

@@ -113,7 +113,14 @@ pub fn create_aggregate_expr(
     name: String,
 ) -> Result<Arc<dyn AggregateExpr>> {
     // coerce
-    let arg = coerce(args, input_schema, &signature(fun))?[0].clone();
+    let arg = coerce(args, input_schema, &signature(fun))?;
+    if arg.is_empty() {
+        return Err(DataFusionError::Plan(format!(
+            "Invalid or wrong number of arguments passed to aggregate: '{}'",
+            name,
+        )));
+    }
+    let arg = arg[0].clone();
 
     let arg_types = args
         .iter()

datafusion/tests/sql.rs

@@ -3437,3 +3437,14 @@ async fn test_physical_plan_display_indent_multi_children() {
         expected, actual
     );
 }
+
+#[tokio::test]
+async fn test_aggregation_with_bad_arguments() -> Result<()> {
+    let mut ctx = ExecutionContext::new();
+    register_aggregate_csv(&mut ctx)?;
+    let sql = "SELECT COUNT(DISTINCT) FROM aggregate_test_100";
+    let logical_plan = ctx.create_logical_plan(&sql)?;
+    let physical_plan = ctx.create_physical_plan(&logical_plan);
+    assert!(physical_plan.is_err());
+    Ok(())
+}