Map the Photon nodes Databricks 15.4 and 17.3 added to their Spark equivalents

[jtwynne]

Fixes #2158

Problem

On a Databricks 15.4 or 17.3 Photon event log, spark_rapids qualification lists Photon nodes
that did not exist on 13.3 as unsupported execs, Exec, Unsupported, Triage in
unsupported_operators.csv. The Photon-to-Spark mapping is one file,
parser/photon/databricks-13_3.json, loaded by PhotonOssOpMapper for every runtime. A Photon
name with no entry is left as it is and the generic parser reports it as unsupported. On a 17.3
Delta MERGE log, seven of the 21 Photon node names have no entry (PhotonWriteStage,
PhotonParquetWriter, PhotonMetadataSubquery, PhotonRuntimeFilterSource,
PhotonColumnarToRow, PhotonRange, PhotonJsonScan) and they land in 39 of 121 SQLs:

10,35,21,Exec,"PhotonWriteStage","Unsupported",2603,312028,Triage
10,35,22,Exec,"PhotonParquetWriter","Unsupported",2603,312028,Triage
10,22,24,Exec,"PhotonMetadataSubquery","Unsupported",1324,312028,Triage
10,22,28,Exec,"PhotonRuntimeFilterSource","Unsupported",1324,312028,Triage

The write pair sits under Execute WriteIntoDeltaCommand, which the tools already recognise,
so the command is scored while its own writer counts against it. The file's header carries a
TODO for per-version files and #1384 asks for the same.

Fix

Seven entries in databricks-13_3.json, each the Spark node the tools already score or set
aside, the way the existing entries map Photon's stage wrappers and scans.

• PhotonWriteStage -> WholeStageCodegen, as PhotonShuffleMapStage, PhotonResultStage and
  PhotonUnionShuffleMapStage already do. PhotonParquetWriter -> WriteFiles, the physical
  writer under the write command (WriteFilesExec in supportedExecs.csv, and one of the
  supported blank execs).
• PhotonMetadataSubquery -> Subquery and PhotonRuntimeFilterSource -> Subquery. Both are
  pointer nodes: in the log each has zero exec duration and no children and sits inside the
  scan's own stage, whose time is scored through the scan and filter. That is the shape
  SubqueryExecParser describes ("a collect execution pointing to an actual one", driver-side
  metrics) and removes from the estimate, and it is where the OSS forms of the same two things
  land, a runtime filter being a bloom-filter aggregate subquery and metadata pruning a
  dynamic-pruning subquery. EMR's GenerateBloomFilter gets the same treatment on
  execsToBeRemoved. SubqueryBroadcast would instead count a Photon-only filter build as
  supported work, so it was not used. With the mapping these rows reappear as Subquery under
  IgnoreNoPerf and the stage time stays with the scans. This is the one judgment call in the
  set.
• PhotonColumnarToRow -> ColumnarToRow (also on execsToBeRemoved), PhotonRange -> Range
  (RangeExec), PhotonJsonScan -> Scan, which keeps the format suffix the way PhotonScan
  does and gives the read parser Scan json.
• Additive and version-agnostic: a 13.3 plan never contains these names, so the 13.3 fixture
  and its expectation files are unchanged. None of the seven means something different on
  another runtime, so this does not pre-empt [FEA] Implement Version-Specific Photon to Spark Node Mapping for Photon parsing #1384, which stays for a mapping that diverges.
  The file keeps its name, since renaming it touches PhotonOssOpMapper and the design in
  [FEA] Implement Version-Specific Photon to Spark Node Mapping for Photon parsing #1384.
• Databricks AWS and Azure, qualification and profiling, since both go through
  PhotonOssOpMapper.toPlatformAwarePlan.

Tests

• PhotonPlanParserSuite, "Photon operators from Databricks 15.4 and 17.3 map to their Spark
  equivalents": one assertion per entry with the expected name written out, through
  PhotonOssOpMapper.mapContentToOss, plus PhotonJsonScan json -> Scan json and an unmapped
  name left unchanged. Fails on dev at its first pin, "[PhotonWriteStage]" did not equal "[WholeStageCodegen]". The existing 13.3 log test in the suite is unchanged and still
  passes. No log fixture is added: the 17.3 log is far
  above the suite's largest fixture, and the mapping is a name substitution the string test
  exercises directly.
• PhotonPlanParserSuite: 2 succeeded, 0 failed with the change. Scalastyle clean.
• Qualifying the 17.3 Photon log with the change, same tool build, same log: the seven names
  leave unsupported_operators.csv (71 rows over 39 SQLs to 0), the unsupported stage share
  moves from 49.26% to 28.45%, the estimated speedup from 1.05x to 1.15x, and execs.csv shows
  WriteFiles, Subquery, Range and Scan json where the Photon names were. The 71 Triage
  rows become 32 rows under ignore actions, 28 Subquery and 3 ColumnarToRow as
  IgnoreNoPerf and the JSON scan as a ReadDeltaLog IgnorePerf (it is the Delta log read),
  none of which counts toward the estimate. The other 721 rows are unchanged. The app stays
  Not Recommended on the share that remains, which is the Delta commands and the rest of the
  plan, not these nodes.

[greptile-apps]

Confidence Score: 5/5

The PR appears safe to merge; the new aliases consistently route newer Photon nodes through existing Spark parser behavior.

Summary

• Maps Photon write-stage, writer, subquery, transition, range, and JSON-scan nodes to existing Spark parser classifications.
• Preserves trailing scan-format text when replacing Photon operator names.
• Verifies every new mapping and confirms that unknown Photon operators remain unchanged.

Diagram

%%{init: {'theme': 'neutral'}}%%
flowchart LR
  A[Databricks Photon plan node] --> B[PhotonOssOpMapper]
  B --> C[Mapped Spark operator name]
  C --> D[Spark plan graph construction]
  D --> E[Existing operator parser]
  E --> F[Qualification and profiling output]

Loading

_{Reviews (1) · Last reviewed commit: "Map the Photon nodes Databricks 15.4 and..."}

[amahussein]

Thanks @jtwynne for the changes!
IIRC, there is a tricky part in adding photon ops. We need to take a deeper look on those new ops and their metrics.
Otherwise, the metrics won't be aggregated correctly.
The starting point is to take a deeper look at a photon eventlog Vs its Spark equivalent eventlog.

CC: @parthosa @mattahrens

[jtwynne]

@amahussein I ran the comparison. Same MERGE workload on plain 17.3 (the gpu-ml runtime with no plugin and no Photon) next to the Photon log the PR came from, every metric name per node pulled from both, then checked against what each mapped parser reads.

Photon node	Photon metrics	Same spot in the Spark plan	What the mapped parser reads
PhotonWriteStage	stage duration, peak memory usage, rows	WholeStageCodegen (duration) under the same Execute WriteIntoDeltaCommand	PhotonStageExecParser reads stage duration, the path PhotonShuffleMapStage and PhotonResultStage already take
PhotonParquetWriter	50, cumulative time, files written, row groups	WriteFiles, 0 metrics	blank exec, no duration read on either side, the time is in the stage above
PhotonRange	cumulative time, rows	Range, number of output rows	generic parser, no duration on either side
PhotonColumnarToRow	12	ColumnarToRow, 2	removed on both sides
PhotonJsonScan json	91, cumulative time	none, Spark reads the Delta state through Scan ExistingRDD	the cumulative time to scanTime swap keys on the name containing Scan, true before and after, and the row files as ReadDeltaLog
PhotonMetadataSubquery	time to execute	none	SubqueryExecParser reads time to collect, so no duration, and the exec is removed from the estimate
PhotonRuntimeFilterSource	50, shuffle-read shaped (fetch wait time, remote and local blocks read)	none, Spark folds it into the scan's dynamic pruning - num filters (DFP) metrics	same

The two subquery nodes are the only ones whose metrics don't match the parser they land on, and they're also the two with nothing to compare against. No Subquery, SubqueryBroadcast or bloom-filter subquery appears in the 144 executions of the Spark run, and the GPU leg of the same workload has no counterpart node either, so dropping them from the estimate is what the Spark plan gets by construction. Their time to execute and shuffle-read metrics go unused, which is worth knowing rather than hiding. PhotonAppSparkMetricsAnalyzer matches accumulators by name and not by node, so the peak-memory and shuffle-write aggregation doesn't see the rename.

Stage level on the Photon log, same jar with and without the change, 190 stages, task totals identical: 40 stages move, 38 down and 2 up. The write stages go to 0 unsupported (12: 93,060 to 0, 453: 126,099 to 0), the runtime-filter stages roughly halve as two Photon rows become one Subquery row (22: 1,179 to 655), the Range stages drop by one exec's share (283: 6,028 to 4,521), and qualifying the plain-Spark log with the same jar gives WriteFiles and Range supported and ColumnarToRow removed, the same answers the mapped Photon plan gets.

The 2 that go up are the tricky part you were remembering, I think. Stages 456 (723 to 1,012) and 301 (286 to 400) hold an unsupported HashAggregate that never changes classification. What changes is that PhotonWriteStage, once it's a WholeStageCodegen, becomes a cluster, and the graph builder only closes a cluster at an InputAdapter or a query stage (ToolsPlanGraph.buildSparkPlanGraphNode), which the Photon plan doesn't emit at that spot. The write stage's cluster then takes in the input Exchange and AQEShuffleRead, and flattenedExecs in QualificationAppInfo apportions every child under the cluster's stage, so those two leave stage 456's denominator and the HashAggregate gets a bigger share. It isn't new with this PR though. The same log already has 14 such children under PhotonShuffleMapStage and PhotonResultStage clusters, and the repo's own 13.3 fixture (NDS q88) has 14 across 7 stages, all under PhotonShuffleMapStage. They never showed because those children are supported Exchange nodes and only dilute a denominator. The write-stage mapping adds 2 instances, with an effect of 289 and 114 ms on stages of 3,376 and 1,338 ms and none on the summary metric, which is a wall-clock sum keyed on the unsupported operators.

So the mapping holds up on metrics, and the stage boundary for Photon clusters is a separate defect in the graph builder that predates it. I submitted that as a new issue in #2161 with the fixture counts, and the fix (close a Photon cluster at the Exchange-mapped input, or apportion cluster children by their own stages) probably belongs in its own PR with its own test. So, I've kept this one as the mapping change, with that as the follow-up.