Fixes 3 bugs during serialization and deserialization of physical plans (#16858)

Author: NGA-TRAN

Cargo.lock

@@ -1358,6 +1358,9 @@ impl DefaultPhysicalPlanner {
                     physical_name(expr),
                 ))?])),
             }
+        } else if group_expr.is_empty() {
+            // No GROUP BY clause - create empty PhysicalGroupBy
+            Ok(PhysicalGroupBy::new(vec![], vec![], vec![]))
         } else {
             Ok(PhysicalGroupBy::new_single(
                 group_expr

datafusion/core/src/physical_planner.rs

@@ -75,6 +75,7 @@ impl TryFrom<AccumulatorArgs<'_>> for FFI_AccumulatorArgs {
             ignore_nulls: args.ignore_nulls,
             fun_definition: None,
             aggregate_function: None,
+            human_display: args.name.to_string(),
         };
         let physical_expr_def = physical_expr_def.encode_to_vec().into();
 

datafusion/ffi/src/udaf/accumulator_args.rs

@@ -332,10 +332,17 @@ impl PhysicalGroupBy {
                 )
                 .collect();
         let num_exprs = expr.len();
+        let groups = if self.expr.is_empty() {
+            // No GROUP BY expressions - should have no groups
+            vec![]
+        } else {
+            // Has GROUP BY expressions - create a single group
+            vec![vec![false; num_exprs]]
+        };
         Self {
             expr,
             null_expr: vec![],
-            groups: vec![vec![false; num_exprs]],
+            groups,
         }
     }
 }

datafusion/physical-plan/src/aggregates/mod.rs

@@ -60,4 +60,5 @@ datafusion-functions = { workspace = true, default-features = true }
 datafusion-functions-aggregate = { workspace = true }
 datafusion-functions-window-common = { workspace = true }
 doc-comment = { workspace = true }
+pretty_assertions = "1.4"
 tokio = { workspace = true, features = ["rt-multi-thread"] }

datafusion/proto/Cargo.toml

@@ -859,6 +859,7 @@ message PhysicalScalarUdfNode {
   optional bytes fun_definition = 3;
   datafusion_common.ArrowType return_type = 4;
   bool nullable = 5;
+  string return_field_name = 6;
 }
 
 message PhysicalAggregateExprNode {
@@ -870,6 +871,7 @@ message PhysicalAggregateExprNode {
   bool distinct = 3;
   bool ignore_nulls = 6;
   optional bytes fun_definition = 7;
+  string human_display = 8;
 }
 
 message PhysicalWindowExprNode {

datafusion/proto/proto/datafusion.proto

@@ -368,7 +368,12 @@ pub fn parse_physical_expr(
                     e.name.as_str(),
                     scalar_fun_def,
                     args,
-                    Field::new("f", convert_required!(e.return_type)?, true).into(),
+                    Field::new(
+                        &e.return_field_name,
+                        convert_required!(e.return_type)?,
+                        true,
+                    )
+                    .into(),
                 )
                 .with_nullable(e.nullable),
             )

datafusion/proto/src/generated/pbjson.rs

@@ -1108,6 +1108,7 @@ impl protobuf::PhysicalPlanNode {
                                     AggregateExprBuilder::new(agg_udf, input_phy_expr)
                                         .schema(Arc::clone(&physical_schema))
                                         .alias(name)
+                                        .human_display(agg_node.human_display.clone())
                                         .with_ignore_nulls(agg_node.ignore_nulls)
                                         .with_distinct(agg_node.distinct)
                                         .order_by(order_bys)