Cleanup logical optimizer rules. (#7919)

* Initial commit

* Address todos

* Update comments

* Simplifications

* Minor simplifications

* Address reviews

* Add TableScan constructor

* Minor changes

* make try_new_with_schema method of Aggregate private

* Use projection try_new instead of try_new_schema

* Simplifications, add comment

* Review changes

* Improve comments

* Move get_wider_type to type_coercion module

* Clean up type coercion file

---------

Co-authored-by: berkaysynnada <berkay.sahin@synnada.ai>
Co-authored-by: Mehmet Ozan Kabak <ozankabak@gmail.com>

Author: 3 people

datafusion/common/src/dfschema.rs

@@ -444,6 +444,14 @@ impl DFSchema {
                         .zip(iter2)
                         .all(|((t1, f1), (t2, f2))| t1 == t2 && Self::field_is_semantically_equal(f1, f2))
             }
+            (
+                DataType::Decimal128(_l_precision, _l_scale),
+                DataType::Decimal128(_r_precision, _r_scale),
+            ) => true,
+            (
+                DataType::Decimal256(_l_precision, _l_scale),
+                DataType::Decimal256(_r_precision, _r_scale),
+            ) => true,
             _ => dt1 == dt2,
         }
     }

datafusion/common/src/functional_dependencies.rs

@@ -558,4 +558,21 @@ mod tests {
         assert_eq!(iter.next(), Some(&Constraint::Unique(vec![20])));
         assert_eq!(iter.next(), None);
     }
+
+    #[test]
+    fn test_get_updated_id_keys() {
+        let fund_dependencies =
+            FunctionalDependencies::new(vec![FunctionalDependence::new(
+                vec![1],
+                vec![0, 1, 2],
+                true,
+            )]);
+        let res = fund_dependencies.project_functional_dependencies(&[1, 2], 2);
+        let expected = FunctionalDependencies::new(vec![FunctionalDependence::new(
+            vec![0],
+            vec![0, 1],
+            true,
+        )]);
+        assert_eq!(res, expected);
+    }
 }

datafusion/core/tests/sql/group_by.rs

@@ -231,13 +231,13 @@ async fn group_by_dictionary() {
         .expect("ran plan correctly");
 
         let expected = [
-            "+-------+------------------------+",
-            "| t.val | COUNT(DISTINCT t.dict) |",
-            "+-------+------------------------+",
-            "| 1     | 2                      |",
-            "| 2     | 2                      |",
-            "| 4     | 1                      |",
-            "+-------+------------------------+",
+            "+-----+------------------------+",
+            "| val | COUNT(DISTINCT t.dict) |",
+            "+-----+------------------------+",
+            "| 1   | 2                      |",
+            "| 2   | 2                      |",
+            "| 4   | 1                      |",
+            "+-----+------------------------+",
         ];
         assert_batches_sorted_eq!(expected, &results);
     }

datafusion/expr/src/built_in_function.rs

@@ -17,21 +17,26 @@
 
 //! Built-in functions module contains all the built-in functions definitions.
 
+use std::cmp::Ordering;
+use std::collections::HashMap;
+use std::fmt;
+use std::str::FromStr;
+use std::sync::{Arc, OnceLock};
+
 use crate::nullif::SUPPORTED_NULLIF_TYPES;
 use crate::signature::TIMEZONE_WILDCARD;
+use crate::type_coercion::binary::get_wider_type;
 use crate::type_coercion::functions::data_types;
 use crate::{
     conditional_expressions, struct_expressions, utils, FuncMonotonicity, Signature,
     TypeSignature, Volatility,
 };
+
 use arrow::datatypes::{DataType, Field, Fields, IntervalUnit, TimeUnit};
 use datafusion_common::{
     internal_err, plan_datafusion_err, plan_err, DataFusionError, Result,
 };
-use std::collections::HashMap;
-use std::fmt;
-use std::str::FromStr;
-use std::sync::{Arc, OnceLock};
+
 use strum::IntoEnumIterator;
 use strum_macros::EnumIter;
 
@@ -468,18 +473,14 @@ impl BuiltinScalarFunction {
     /// * `List(Int64)` has dimension 2
     /// * `List(List(Int64))` has dimension 3
     /// * etc.
-    fn return_dimension(self, input_expr_type: DataType) -> u64 {
-        let mut res: u64 = 1;
+    fn return_dimension(self, input_expr_type: &DataType) -> u64 {
+        let mut result: u64 = 1;
         let mut current_data_type = input_expr_type;
-        loop {
-            match current_data_type {
-                DataType::List(field) => {
-                    current_data_type = field.data_type().clone();
-                    res += 1;
-                }
-                _ => return res,
-            }
+        while let DataType::List(field) = current_data_type {
+            current_data_type = field.data_type();
+            result += 1;
         }
+        result
     }
 
     /// Returns the output [`DataType`] of this function
@@ -538,11 +539,17 @@ impl BuiltinScalarFunction {
                     match input_expr_type {
                         List(field) => {
                             if !field.data_type().equals_datatype(&Null) {
-                                let dims = self.return_dimension(input_expr_type.clone());
-                                if max_dims < dims {
-                                    max_dims = dims;
-                                    expr_type = input_expr_type.clone();
-                                }
+                                let dims = self.return_dimension(input_expr_type);
+                                expr_type = match max_dims.cmp(&dims) {
+                                    Ordering::Greater => expr_type,
+                                    Ordering::Equal => {
+                                        get_wider_type(&expr_type, input_expr_type)?
+                                    }
+                                    Ordering::Less => {
+                                        max_dims = dims;
+                                        input_expr_type.clone()
+                                    }
+                                };
                             }
                         }
                         _ => {