Speed up optimize_projection (apache#15787)

* save

* fmt

Author: xudong963

datafusion/optimizer/src/optimize_projections/mod.rs

@@ -32,8 +32,7 @@ use datafusion_common::{
 use datafusion_expr::expr::Alias;
 use datafusion_expr::Unnest;
 use datafusion_expr::{
-    logical_plan::LogicalPlan, projection_schema, Aggregate, Distinct, Expr, Projection,
-    TableScan, Window,
+    logical_plan::LogicalPlan, Aggregate, Distinct, Expr, Projection, TableScan, Window,
 };
 
 use crate::optimize_projections::required_indices::RequiredIndicies;
@@ -780,9 +779,24 @@ fn rewrite_projection_given_requirements(
 /// Projection is unnecessary, when
 /// - input schema of the projection, output schema of the projection are same, and
 /// - all projection expressions are either Column or Literal
-fn is_projection_unnecessary(input: &LogicalPlan, proj_exprs: &[Expr]) -> Result<bool> {
-    Ok(&projection_schema(input, proj_exprs)? == input.schema()
-        && proj_exprs.iter().all(is_expr_trivial))
+pub fn is_projection_unnecessary(
+    input: &LogicalPlan,
+    proj_exprs: &[Expr],
+) -> Result<bool> {
+    // First check if the number of expressions is equal to the number of fields in the input schema.
+    if proj_exprs.len() != input.schema().fields().len() {
+        return Ok(false);
+    }
+    Ok(input.schema().iter().zip(proj_exprs.iter()).all(
+        |((field_relation, field_name), expr)| {
+            // Check if the expression is a column and if it matches the field name
+            if let Expr::Column(col) = expr {
+                col.relation.as_ref() == field_relation && col.name.eq(field_name.name())
+            } else {
+                false
+            }
+        },
+    ))
 }
 
 #[cfg(test)]