david-mollitor-db opened a new pull request, #59179: URL: https://github.com/apache/spark/pull/59179
### What changes were proposed in this pull request? `GraphOperations` (mixed into `DataflowGraph`) answers the reachability queries the triggered scheduler uses to order flows: `upstreamFlows`, `downstreamFlows`, and the dataset-level `upstreamDatasets`. This PR: - Memoizes the per-flow and per-dataset reachability sets. The graph is immutable once resolved, so a computed set stays valid for its lifetime. - Adds immediate-neighbor adjacency maps (`immediateUpstreamDatasets` / `immediateDownstreamDatasets`) so each traversal step in `dfsInternal` is a map lookup rather than a scan over all flows. Method names, signatures, and result sets are unchanged; the original `upstreamFlows` / `downstreamFlows` bodies are preserved as private `compute*` helpers behind the cache. ### Why are the changes needed? `TriggeredGraphExecution` calls `upstreamFlows` for every queued flow on each scheduler tick (and `downstreamFlows` when a flow fails). Each call runs a full graph traversal plus a filter over all flows, recomputed on every tick -- roughly `O(queued x datasets x flows)` per tick -- which scales poorly on large or wide pipelines. Caching the closures makes these lookups effectively `O(1)` after warmup, and the adjacency index also speeds up the `reanalyzeFlow` traversal performed when each flow starts. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? - New `GraphOperationsSuite`: asserts `upstreamFlows` / `downstreamFlows` / `upstreamDatasets` against a diamond graph and two disconnected components whose reachability is known by inspection, and verifies that repeated queries return the identical cached instance. - Existing pipelines suites that exercise these methods and `reanalyzeFlow` (`TriggeredGraphExecutionSuite`, `GraphExecutionSuite`, `SinkExecutionSuite`, `SqlPipelineSuite`, `ConnectValidPipelineSuite`, `ConnectInvalidPipelineSuite`, `AutoCdc*`). WIP / draft: broader pipelines regression is still being run and will be summarized before this is marked ready for review. ### Was this patch authored or co-authored using generative AI tooling? Yes. Generated-by: Claude Code. This pull request and its description were written by Isaac. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
