[CORE] Preserve fallback tag for nodes without logicalLink in RemoveFallbackTagRule

[liujiayi771]

Summary

For SparkPlan nodes that lack a logicalLink (e.g., SortExec / BroadcastExchange generated by EnsureRequirements), GlutenFallbackReporter cannot copy their FallbackTag onto a logical plan, so the subsequent RemoveFallbackTagRule removed their tag without persisting it anywhere. This caused the fallback reason to be lost when GlutenQueryExecutionListener re-walks the finalized plan at SQL execution end, resulting in a generic "Gluten does not touch it or does not support it" reason instead of the actual fallback reason.

In #11853, the test assertion for "get correct fallback reason on nodes without logicalLink" was changed from forall to exists, which hid this issue — among the fallbackReasons, one event's reason had degraded to the generic message because the physical plan tag was removed by RemoveFallbackTagRule, but exists only required some reasons to be correct, masking the fact that others were wrong.

Fix: In RemoveFallbackTagRule, skip untagging for nodes that have a FallbackTag but no logicalLink. Their tag stays on the physical node and is read by GlutenExplainUtils.handleVanillaSparkPlan via FallbackTags.getOption(p) (the existing fallback branch) in both the per-stage GlutenFallbackReporter event and the end-of-SQL plan walk. The tag is regenerated by Gluten validation on every AQE stage's postStageCreationRules, so it survives across AQE re-planning iterations as well.

With this fix, the test assertion is changed back to forall, verifying that all fallback events now report the correct reason.

Why not synthesize a logicalLink?

An earlier attempt synthesized a dummy LeafNode (FallbackLogicalPlan) and attached it via SparkPlan.LOGICAL_PLAN_TAG so the existing p.logicalLink.flatMap(FallbackTags.getOption) path would find it. That approach poisoned AdaptiveSparkPlanExec.setLogicalLinkForNewQueryStage, which intentionally relies on EnsureRequirements-generated nodes having no logicalLink in order to walk down to the real logical node. With a synthetic link in place, the stage's logicalLink was set to the dummy, which never appeared in the original logical plan; AQE's replaceWithQueryStages then could not insert the LogicalQueryStage wrapper, breaking AQE rules that match LogicalQueryStage(_, stage: ...QueryStageExec) such as AQEPropagateEmptyRelation.
Keeping the tag on the physical node sidesteps the AQE contract entirely.

Test plan

• Existing test "get correct fallback reason on nodes without logicalLink" now uses forall and verifies all fallback reasons are correct

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[liujiayi771]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[liujiayi771]

Hi @zhztheplayer @Zouxxyy. After #11853, GlutenQueryExecutionListener.onSQLExecutionEnd re-walks qe.executedPlan and posts a fresh GlutenPlanFallbackEvent, which overwrites the per-stage events in kvstore (last write wins). By that time every AQE stage has already gone through RemoveFallbackTagRule, so for nodes without a logicalLink (e.g. SortExec / BroadcastExchange from EnsureRequirements) the FallbackTag is gone. End result: the regression #11295 tried to fix is back, just hidden by the test being relaxed from forall to exists in #11853.

This PR takes the simplest fix: don't untag physical nodes that have no logicalLink in RemoveFallbackTagRule. Any downside you can think of to leaving the FallbackTag on the physical node in this case?

[zhztheplayer]

For SparkPlan nodes that lack a logicalLink (e.g., SortExec / BroadcastExchange generated by EnsureRequirements), GlutenFallbackReporter cannot copy their FallbackTag onto a logical plan,

Just curious, does this behavior differ across Spark versions? I checked latest Spark code and the rule adopts transformUp which should've copied logical links to the sort / exchange nodes: https://github.com/apache/spark/blob/484bb5b2cd718d27872235a30c1af69d5d4d34d7/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala#L891

[liujiayi771]

@zhztheplayer You're right that EnsureRequirements.apply uses transformUp, but transformUp only carries over tags (including logicalLink) to the node being directly replaced via copyTagsFrom. It does NOT set logicalLink on newly constructed child nodes.

In ensureDistributionAndOrdering, the new SortExec(...) / ShuffleExchangeExec(...) / BroadcastExchangeExec(...) are created directly without any setLogicalLink call. The transformUp rule matches the parent operator (e.g., SMJ) and calls reordered.withNewChildren(newChildren), which preserves the parent's logicalLink but does not propagate it to the newly inserted children.

[zhztheplayer]

Are fallback tags really as helpful as expected? If they just represent the fallback reasons that a simple logging mechanism can handle. PR LGTM by the way.

[github-actions]

Run Gluten Clickhouse CI on x86