[CARMEL-6674] Project fail to be collapsed (#1283)

* [CARMEL-6674]: Project fail to be collapsed

* fix checkstyle

* fix

Author: fenzhu

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala

@@ -234,7 +234,8 @@ abstract class Optimizer(catalogManager: CatalogManager)
       DeduplicateRightSideOfLeftSemiAntiJoin,
       RemoveNoopOperators) :+
     Batch("Pull out complex join condition", Once,
-      PullOutComplexJoinCondition) :+
+      PullOutComplexJoinCondition,
+      CollapseProject) :+
     // This batch must be executed after the `RewriteSubquery` batch, which creates joins.
     Batch("NormalizeFloatingNumbers", Once, NormalizeFloatingNumbers)
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PullOutComplexJoinCondition.scala

@@ -64,8 +64,7 @@ object PullOutComplexJoinCondition extends Rule[LogicalPlan]
         }
         val newLeft = Project(left.output ++ leftComplexExpMap.values, left)
         val newRight = Project(right.output ++ rightComplexExpMap.values, right)
-        CollapseProject(Project(j.output, j.copy(left = newLeft, right = newRight,
-          condition = Some(newCondition))))
+        Project(j.output, j.copy(left = newLeft, right = newRight, condition = Some(newCondition)))
       } else {
         j
       }

sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala

@@ -29,7 +29,7 @@ import org.apache.spark.sql.catalyst.expressions.{GenericRow, In}
 import org.apache.spark.sql.catalyst.expressions.aggregate.{Complete, Partial}
 import org.apache.spark.sql.catalyst.optimizer._
 import org.apache.spark.sql.catalyst.plans.{Cross, Inner, JoinType}
-import org.apache.spark.sql.catalyst.plans.logical.{Filter, LogicalPlan, Project}
+import org.apache.spark.sql.catalyst.plans.logical.{Filter, Join, LogicalPlan, Project}
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.catalyst.util.StringUtils
 import org.apache.spark.sql.execution.{FileSourceScanExec, SparkPlan}
@@ -6800,6 +6800,22 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark
       }
     }
   }
+
+  test("CARMEL-6674: Project fail to be collapsed") {
+    withTable("t1", "t2") {
+      spark.sql("CREATE TABLE t1(id STRING) USING PARQUET")
+      spark.sql("CREATE TABLE t2(sid DECIMAL(18,0), ct TIMESTAMP) USING PARQUET")
+      withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
+        val plan = spark.sql(
+          """
+            |SELECT ct as run_time
+            |FROM t1 LEFT JOIN t2 ON id = sid
+            |""".stripMargin).queryExecution.optimizedPlan
+        assert(plan.isInstanceOf[Project] &&
+          plan.asInstanceOf[Project].child.isInstanceOf[Join])
+      }
+    }
+  }
 }
 
 case class Foo(bar: Option[String])