[BUG] A Photon stage wrapper mapped to WholeStageCodegen takes its input Exchange into the cluster, so that exec is apportioned to the wrapper's stage

[jtwynne]

Describe the bug

In a Photon plan the stage wrappers (PhotonShuffleMapStage, PhotonResultStage, PhotonUnionShuffleMapStage, and PhotonWriteStage from #2159) map to WholeStageCodegen and become clusters in ToolsPlanGraph. buildSparkPlanGraphNode closes a cluster only at an InputAdapter or a query-stage node, and a Photon plan has neither between the wrapper and its input Exchange (PhotonShuffleExchangeSource), so the Exchange and an AQEShuffleRead next to it become children of the cluster. QualificationAppInfo.flattenedExecs then apportions every cluster child under the cluster's stage, while the Exchange's own accumulators put it in the map stage that wrote the shuffle. The child leaves its real stage's denominator and joins the wrapper's, so the per-stage "Unsupported Task Duration" in stages.csv shifts between the two stages.

The repo's own fixture shows it. In core/src/test/resources/QualificationExpectations/photon_db_13_3/app-20240919162642-0000/execs.csv (NDS q88 on Databricks 13.3 Photon), 14 cluster children have an accumulator stage set that does not contain their cluster's stage, across stages 46, 48, 50, 52, 54, 56 and 58, all under PhotonShuffleMapStage. On a Databricks 17.3 Photon MERGE log the count is 14 under the older wrappers and 16 once PhotonWriteStage is mapped.

The effect on the estimate is small when the absorbed child is a supported Exchange (it only dilutes a denominator), and visible when the stage it leaves holds an unsupported exec. On the 17.3 log, stage 456 (task time 3,376 ms) goes from 723 to 1,012 ms unsupported and stage 301 (1,338 ms) from 286 to 400 after the write-stage mapping, with the HashAggregate in them unchanged. The summary's "Unsupported Operators Stage Duration" is a wall-clock sum keyed on the unsupported operators and does not move.

Steps to reproduce

1. Qualify a Databricks Photon event log (the 13.3 fixture nds_q88_photon_db_13_3.zstd does).
2. In execs.csv, take each WholeStageCodegen row's Exec Children Node Ids and compare every child's Exec Stages with the cluster row's. Children whose stage set is non-empty and disjoint from the cluster's are the absorbed inputs, Exchange in every case in the fixture.
3. In stages.csv, the child's stage carries fewer execs in its denominator than its accumulators say, and the cluster's stage more.

Expected behavior

A cluster child is apportioned to the stage its own accumulators put it in, or the cluster closes at the Photon input node the way it closes at an InputAdapter for a Spark plan.

Environment

• cudf-spark-tools dev at 2073060, the 13.3 fixture in the repo and a Databricks 17.3 Photon log (17.3.x-photon-scala2.13).
• ToolsPlanGraph.buildSparkPlanGraphNode (plangraph/ToolsPlanGraph.scala), QualificationAppInfo.flattenedExecs and stagesSummary (qualification/QualificationAppInfo.scala), PhotonStageExecParser.

Additional context

Found while checking the metrics of the entries in #2159 against a plain-Spark run of the same workload. The mapping there stands on its own. The fix here is either a Photon cluster boundary in the graph builder (close the cluster at the Exchange-mapped input, as InputAdapter does for Spark) or per-child stage attribution in flattenedExecs. Either changes the 13.3 expectation files, which is the test.