Covert grouping to udaf

datafusion/expr/src/aggregate_function.rs

@@ -41,8 +41,6 @@ pub enum AggregateFunction {
     ArrayAgg,
     /// N'th value in a group according to some ordering
     NthValue,
-    /// Grouping
-    Grouping,
 }
 
 impl AggregateFunction {
@@ -53,7 +51,6 @@ impl AggregateFunction {
             Max => "MAX",
             ArrayAgg => "ARRAY_AGG",
             NthValue => "NTH_VALUE",
-            Grouping => "GROUPING",
         }
     }
 }
@@ -73,8 +70,6 @@ impl FromStr for AggregateFunction {
             "min" => AggregateFunction::Min,
             "array_agg" => AggregateFunction::ArrayAgg,
             "nth_value" => AggregateFunction::NthValue,
-            // other
-            "grouping" => AggregateFunction::Grouping,
             _ => {
                 return plan_err!("There is no built-in function named {name}");
             }
@@ -119,7 +114,6 @@ impl AggregateFunction {
                 coerced_data_types[0].clone(),
                 input_expr_nullable[0],
             )))),
-            AggregateFunction::Grouping => Ok(DataType::Int32),
             AggregateFunction::NthValue => Ok(coerced_data_types[0].clone()),
         }
     }
@@ -130,7 +124,6 @@ impl AggregateFunction {
         match self {
             AggregateFunction::Max | AggregateFunction::Min => Ok(true),
             AggregateFunction::ArrayAgg => Ok(false),
-            AggregateFunction::Grouping => Ok(true),
             AggregateFunction::NthValue => Ok(true),
         }
     }
@@ -141,9 +134,7 @@ impl AggregateFunction {
     pub fn signature(&self) -> Signature {
         // note: the physical expression must accept the type returned by this function or the execution panics.
         match self {
-            AggregateFunction::Grouping | AggregateFunction::ArrayAgg => {
-                Signature::any(1, Volatility::Immutable)
-            }
+            AggregateFunction::ArrayAgg => Signature::any(1, Volatility::Immutable),
             AggregateFunction::Min | AggregateFunction::Max => {
                 let valid = STRINGS
                     .iter()

datafusion/expr/src/type_coercion/aggregates.rs

@@ -102,7 +102,6 @@ pub fn coerce_types(
             get_min_max_result_type(input_types)
         }
         AggregateFunction::NthValue => Ok(input_types.to_vec()),
-        AggregateFunction::Grouping => Ok(vec![input_types[0].clone()]),
     }
 }
 

datafusion/functions-aggregate/src/grouping.rs

@@ -0,0 +1,97 @@
+// 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.
+
+//! Defines physical expressions that can evaluated at runtime during query execution
+
+use std::any::Any;
+use std::fmt;
+
+use arrow::datatypes::DataType;
+use arrow::datatypes::Field;
+use datafusion_common::{not_impl_err, Result};
+use datafusion_expr::function::AccumulatorArgs;
+use datafusion_expr::function::StateFieldsArgs;
+use datafusion_expr::utils::format_state_name;
+use datafusion_expr::{Accumulator, AggregateUDFImpl, Signature, Volatility};
+
+make_udaf_expr_and_func!(
+    Grouping,
+    grouping,
+    expression,
+    "Returns 1 if the data is aggregated across the specified column or 0 for not aggregated in the result set.",
+    grouping_udaf
+);
+
+pub struct Grouping {
+    signature: Signature,
+}
+
+impl fmt::Debug for Grouping {
+    fn fmt(&self, f: &mut std::fmt::Formatter) -> fmt::Result {
+        f.debug_struct("Grouping")
+            .field("name", &self.name())
+            .field("signature", &self.signature)
+            .finish()
+    }
+}
+
+impl Default for Grouping {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Grouping {
+    /// Create a new GROUPING aggregate function.
+    pub fn new() -> Self {
+        Self {
+            signature: Signature::any(1, Volatility::Immutable),
+        }
+    }
+}
+
+impl AggregateUDFImpl for Grouping {
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+
+    fn name(&self) -> &str {
+        "grouping"
+    }
+
+    fn signature(&self) -> &Signature {
+        &self.signature
+    }
+
+    fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
+        Ok(DataType::Int32)
+    }
+
+    fn state_fields(&self, args: StateFieldsArgs) -> Result<Vec<Field>> {
+        Ok(vec![Field::new(
+            format_state_name(args.name, "grouping"),
+            DataType::Int32,
+            true,
+        )])
+    }
+
+    fn accumulator(&self, _acc_args: AccumulatorArgs) -> Result<Box<dyn Accumulator>> {
+        not_impl_err!(
+            "physical plan is not yet implemented for GROUPING aggregate function"
+        )
+    }
+}

datafusion/functions-aggregate/src/lib.rs

@@ -73,6 +73,7 @@ pub mod approx_percentile_cont_with_weight;
 pub mod average;
 pub mod bit_and_or_xor;
 pub mod bool_and_or;
+pub mod grouping;
 pub mod string_agg;
 
 use crate::approx_percentile_cont::approx_percentile_cont_udaf;
@@ -154,6 +155,7 @@ pub fn all_default_aggregate_functions() -> Vec<Arc<AggregateUDF>> {
         bool_and_or::bool_and_udaf(),
         bool_and_or::bool_or_udaf(),
         average::avg_udaf(),
+        grouping::grouping_udaf(),
     ]
 }
 

datafusion/physical-expr/src/aggregate/build_in.rs

@@ -60,11 +60,6 @@ pub fn create_aggregate_expr(
         .collect::<Result<Vec<_>>>()?;
     let input_phy_exprs = input_phy_exprs.to_vec();
     Ok(match (fun, distinct) {
-        (AggregateFunction::Grouping, _) => Arc::new(expressions::Grouping::new(
-            input_phy_exprs[0].clone(),
-            name,
-            data_type,
-        )),
         (AggregateFunction::ArrayAgg, false) => {
             let expr = input_phy_exprs[0].clone();
             let nullable = expr.nullable(input_schema)?;

datafusion/physical-expr/src/aggregate/mod.rs

@@ -20,7 +20,6 @@ pub use datafusion_physical_expr_common::aggregate::AggregateExpr;
 pub(crate) mod array_agg;
 pub(crate) mod array_agg_distinct;
 pub(crate) mod array_agg_ordered;
-pub(crate) mod grouping;
 pub(crate) mod nth_value;
 #[macro_use]
 pub(crate) mod min_max;

datafusion/physical-expr/src/expressions/mod.rs

@@ -38,7 +38,6 @@ pub use crate::aggregate::array_agg::ArrayAgg;
 pub use crate::aggregate::array_agg_distinct::DistinctArrayAgg;
 pub use crate::aggregate::array_agg_ordered::OrderSensitiveArrayAgg;
 pub use crate::aggregate::build_in::create_aggregate_expr;
-pub use crate::aggregate::grouping::Grouping;
 pub use crate::aggregate::min_max::{Max, MaxAccumulator, Min, MinAccumulator};
 pub use crate::aggregate::nth_value::NthValueAgg;
 pub use crate::aggregate::stats::StatsType;

datafusion/proto/proto/datafusion.proto

@@ -483,7 +483,7 @@ enum AggregateFunction {
   // APPROX_PERCENTILE_CONT = 14;
   // APPROX_MEDIAN = 15;
   // APPROX_PERCENTILE_CONT_WITH_WEIGHT = 16;
-  GROUPING = 17;
+  // GROUPING = 17;
   // MEDIAN = 18;
   // BIT_AND = 19;
   // BIT_OR = 20;

datafusion/proto/src/logical_plan/from_proto.rs

@@ -140,7 +140,6 @@ impl From<protobuf::AggregateFunction> for AggregateFunction {
             protobuf::AggregateFunction::Min => Self::Min,
             protobuf::AggregateFunction::Max => Self::Max,
             protobuf::AggregateFunction::ArrayAgg => Self::ArrayAgg,
-            protobuf::AggregateFunction::Grouping => Self::Grouping,
             protobuf::AggregateFunction::NthValueAgg => Self::NthValue,
         }
     }

datafusion/proto/src/logical_plan/to_proto.rs

@@ -111,7 +111,6 @@ impl From<&AggregateFunction> for protobuf::AggregateFunction {
             AggregateFunction::Min => Self::Min,
             AggregateFunction::Max => Self::Max,
             AggregateFunction::ArrayAgg => Self::ArrayAgg,
-            AggregateFunction::Grouping => Self::Grouping,
             AggregateFunction::NthValue => Self::NthValueAgg,
         }
     }
@@ -372,7 +371,6 @@ pub fn serialize_expr(
                     AggregateFunction::ArrayAgg => protobuf::AggregateFunction::ArrayAgg,
                     AggregateFunction::Min => protobuf::AggregateFunction::Min,
                     AggregateFunction::Max => protobuf::AggregateFunction::Max,
-                    AggregateFunction::Grouping => protobuf::AggregateFunction::Grouping,
                     AggregateFunction::NthValue => {
                         protobuf::AggregateFunction::NthValueAgg
                     }