Initial changes to support using udaf min/max for statistics and opti…

datafusion/core/src/datasource/file_format/parquet.rs

@@ -1321,7 +1321,7 @@ mod tests {
             .map(|i| i.to_string())
             .collect();
         let coll: Vec<_> = schema
-            .all_fields()
+            .flattened_fields()
             .into_iter()
             .map(|i| i.name().to_string())
             .collect();

datafusion/expr/src/udaf.rs

@@ -249,6 +249,14 @@ impl AggregateUDF {
     pub fn simplify(&self) -> Option<AggregateFunctionSimplification> {
         self.inner.simplify()
     }
+
+    /// Returns true if the function is max, false if the function is min
+    /// None in all other cases, used in certain optimizations or
+    /// or aggregate
+    ///
+    pub fn is_descending(&self) -> Option<bool> {
+        self.inner.is_descending()
+    }
 }
 
 impl<F> From<F> for AggregateUDF
@@ -536,6 +544,16 @@ pub trait AggregateUDFImpl: Debug + Send + Sync {
         self.signature().hash(hasher);
         hasher.finish()
     }
+
+    /// If this function is max, return true
+    /// if the function is min, return false
+    /// otherwise return None (the default)
+    ///
+    ///
+    /// Note: this is used to use special aggregate implementations in certain conditions
+    fn is_descending(&self) -> Option<bool> {
+        None
+    }
 }
 
 pub enum ReversedUDAF {

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

@@ -730,6 +730,12 @@ impl AggregateExpr for AggregateFunctionExpr {
             }
         }
     }
+
+    fn get_minmax_desc(&self) -> Option<(Field, bool)> {
+        self.fun
+            .is_descending()
+            .and_then(|flag| self.field().ok().map(|f| (f, flag)))
+    }
 }
 
 impl PartialEq<dyn Any> for AggregateFunctionExpr {