[CORE] Add logical link to rewritten spark plan

gluten-core/src/main/scala/io/glutenproject/extension/RewriteSparkPlanRulesManager.scala

@@ -16,7 +16,7 @@
  */
 package io.glutenproject.extension
 
-import io.glutenproject.extension.columnar.{AddTransformHintRule, TRANSFORM_UNSUPPORTED, TransformHint, TransformHints}
+import io.glutenproject.extension.columnar.{AddTransformHintRule, TransformHint, TransformHints}
 
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.catalyst.InternalRow
@@ -66,11 +66,7 @@ class RewriteSparkPlanRulesManager(rewriteRules: Seq[Rule[SparkPlan]]) extends R
       case p if p.nodeName == origin.nodeName => p
     }
     assert(target.size == 1)
-    if (TransformHints.isTransformable(target.head)) {
-      None
-    } else {
-      Some(TransformHints.getHint(target.head))
-    }
+    TransformHints.getHintOption(target.head)
   }
 
   private def applyRewriteRules(origin: SparkPlan): (SparkPlan, Option[String]) = {
@@ -109,15 +105,19 @@ class RewriteSparkPlanRulesManager(rewriteRules: Seq[Rule[SparkPlan]]) extends R
         } else {
           addHint.apply(rewrittenPlan)
           val hint = getTransformHintBack(origin, rewrittenPlan)
-          hint match {
-            case Some(tu @ TRANSFORM_UNSUPPORTED(_, _)) =>
-              // If the rewritten plan is still not transformable, return the original plan.
-              TransformHints.tag(origin, tu)
-              origin
-            case None =>
-              rewrittenPlan.transformUp { case wall: RewrittenNodeWall => wall.originalChild }
-            case _ =>
-              throw new IllegalStateException("Unreachable code")
+          if (hint.isDefined) {
+            // If the rewritten plan is still not transformable, return the original plan.
+            TransformHints.tag(origin, hint.get)
+            origin
+          } else {
+            rewrittenPlan.transformUp {
+              case wall: RewrittenNodeWall => wall.originalChild
+              case p if p.logicalLink.isEmpty =>
+                // Add logical link to pull out project to make fallback reason work,
+                // see `GlutenFallbackReporter`.
+                origin.logicalLink.foreach(p.setLogicalLink)
+                p
+            }
           }
         }
     }

gluten-core/src/main/scala/io/glutenproject/extension/columnar/TransformHintRule.scala

@@ -146,7 +146,7 @@ object TransformHints {
       throw new IllegalStateException("Transform hint tag not set in plan: " + plan.toString()))
   }
 
-  private def getHintOption(plan: SparkPlan): Option[TransformHint] = {
+  def getHintOption(plan: SparkPlan): Option[TransformHint] = {
     plan.getTagValue(TAG)
   }
 

gluten-ut/spark34/src/test/scala/org/apache/spark/sql/gluten/GlutenFallbackSuite.scala

@@ -22,6 +22,7 @@ import io.glutenproject.execution.FileSourceScanExecTransformer
 
 import org.apache.spark.scheduler.{SparkListener, SparkListenerEvent}
 import org.apache.spark.sql.{GlutenSQLTestsTrait, Row}
+import org.apache.spark.sql.execution.ProjectExec
 import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
 import org.apache.spark.sql.execution.ui.{GlutenSQLAppStatusStore, SparkListenerSQLExecutionStart}
 import org.apache.spark.status.ElementTrackingStore
@@ -133,4 +134,33 @@ class GlutenFallbackSuite extends GlutenSQLTestsTrait with AdaptiveSparkPlanHelp
       }
     }
   }
+
+  test("Add logical link to rewritten spark plan") {
+    val events = new ArrayBuffer[GlutenPlanFallbackEvent]
+    val listener = new SparkListener {
+      override def onOtherEvent(event: SparkListenerEvent): Unit = {
+        event match {
+          case e: GlutenPlanFallbackEvent => events.append(e)
+          case _ =>
+        }
+      }
+    }
+    spark.sparkContext.addSparkListener(listener)
+    withSQLConf(GlutenConfig.EXPRESSION_BLACK_LIST.key -> "add") {
+      try {
+        val df = spark.sql("select sum(id +1) from range(10)")
+        spark.sparkContext.listenerBus.waitUntilEmpty()
+        df.collect()
+        val project = find(df.queryExecution.executedPlan) {
+          _.isInstanceOf[ProjectExec]
+        }
+        assert(project.isDefined)
+        events.exists(
+          _.fallbackNodeToReason.values.toSet
+            .contains("Project: Not supported to map spark function name"))
+      } finally {
+        spark.sparkContext.removeSparkListener(listener)
+      }
+    }
+  }
 }