)]}'
{
  "commit": "688a73cd32a4d3fdf8446053ded2049fa47b5487",
  "tree": "5d1edb2c8504182cecda3f4315911da1ad925d03",
  "parents": [
    "976b87ff45962de6ed7d16667f2049e8d926559d"
  ],
  "author": {
    "name": "Chao Sun",
    "email": "chao@openai.com",
    "time": "Fri Aug 21 10:06:08 2026 -0700"
  },
  "committer": {
    "name": "Chao Sun",
    "email": "chao@openai.com",
    "time": "Fri Aug 21 10:06:08 2026 -0700"
  },
  "message": "[SPARK-56842][SQL] Short-circuit AQE when materialized stages are empty\n\n### Why are the changes needed?\n\nAdaptive Query Execution can miss a provable empty or nonempty result when a materialized query stage is represented through a supported physical sort, limit, or aggregate operator. In particular, a global aggregate over empty input still produces one output row: treating that aggregate as empty changes the result of conditionless semi/anti joins.\n\nAdaptive replanning can also remove exchange stages that are still materializing. Leaving a private obsolete stage running wastes work and can let an irrelevant materialization failure abort the selected query plan. Reusing the same exchange inside one adaptive plan must not permanently protect that result after every local alias disappears, while reuse by another adaptive plan or subquery must remain protected. Some obsolete stages cannot safely be cancelled at all, and an opaque delegated shuffle may additionally conceal the shuffle ID needed for cleanup. Correctness therefore requires distinguishing stages whose eventual ordinary failure can safely be ignored from stages whose failure must still propagate, while preserving exchange ownership, shuffle cleanup, fatal-error propagation, and stages carried across multiple adopted adaptive plans.\n\n### What changes were proposed in this PR?\n\n- Infer runtime row counts only through the supported, explicitly safe `SortExec`, `BaseLimitExec`, and `BaseAggregateExec` cases. A global aggregate over an empty stage reports one row; grouped aggregates preserve the child estimate; limits propagate only provable emptiness, including `LIMIT 0`.\n- Remove the unsupported `AQEShuffleReadExec`, `ProjectExec`, and `ColumnarToRowExec` recursion paths instead of claiming coverage for physical shapes that the adaptive rule cannot reach.\n- Derive obsolete-stage candidates from both the previous physical plan and the current round of replacement stages, deduplicating stage IDs before applying the unfinished/private-exchange checks so an older stage remains discoverable when a later replan drops its branch.\n- Record the owning adaptive-plan instance when an exchange result enters the shared stage cache. Reuse within that same plan remains cancellable once every local alias disappears; reuse by a different plan or subquery, or reuse of an entry without a recorded owner, remains permanently protected. Compare owners by reference identity rather than the plan\u0027s structural `equals` implementation.\n- Track every historical local stage ID associated with a reused result, group cancellation by result identity, and cancel its underlying exchange at most once. Return or mark all alias IDs so every materialization callback suppresses an irrelevant ordinary failure while retained/shared and fatal failures continue to propagate.\n- Permanently remember a failed cancellation by result identity and never retry it for a later same-plan reuse alias: a native shuffle marks itself cancelled before attempting to cancel its `FutureAction`, so a retry could otherwise report false success while its job is still running. Keep the original cache entry available and track later aliases for the same guarded obsolete-failure handling.\n- Proactively cancel only eligible, private obsolete shuffle stages. Never cancel broadcast stages, stages already shared or retained by the adopted plan, or opaque delegated shuffle exchanges that cannot expose their own cancellable submission. Mark only private obsolete broadcasts and identifiable submitted shuffles that cannot safely be cancelled for guarded eventual-failure handling; opaque delegated shuffles are neither cancelled nor marked for failure suppression because their hidden delegate shuffle ID cannot safely be registered.\n- Register an already-submitted shuffle ID before any cancellation attempt, without creating a lazy shuffle dependency for an unsubmitted stage. Resolve the underlying exchange directly from `ShuffleQueryStageExec.shuffle`, including aliases wrapped in the leaf `ReusedExchangeExec`. Under `RemoveShuffleFiles`, do not cancel a submitted shuffle: scheduler cancellation is asynchronous, so deleting its files could race map writers. Submitted shuffles remain eligible under cleanup modes that do not remove their files.\n- Preserve the exchange-cache entry when cancellation fails or an obsolete stage cannot safely be cancelled. Remove it immediately only after successful cancellation. For a failed cancellation, an obsolete broadcast, or a known submitted shuffle protected by `RemoveShuffleFiles`, ignore an eventual nonfatal failure only after confirming that the stage remains private and absent from the current physical plan, registering any visible submitted shuffle ID, and evicting the failed cache entry. Fatal failures, failures from subsequently retained/shared stages, and failures from opaque delegated shuffles always propagate; opaque delegated stages remain cached.\n- Serialize exchange-cache lookup, owner registration, reuse, and cross-plan shared-result marking under short adaptive stage-lifecycle lock sections. Reserve cancellation using a stage-result-scoped `CompletableFuture`, then acquire the shuffle monitor and cancel outside the query-global lock. Both the initial cache lookup and the post-child-planning cache lookup await cancellation of that same result outside the global lock and retry; unrelated exchanges continue. Keep stage-local submission/cancellation eligibility checks atomic.\n- Precompute retained stage IDs and shared-result identities and remove exchange-cache entries through canonical-key lookup, avoiding repeated full-cache scans during adaptive replanning.\n- Document nested-stage materialization, same-plan versus cross-subquery exchange ownership, lifecycle lock ordering, alias failure callbacks, and the permanent protection of externally shared stage results. Keep lifecycle helpers scoped to `private[adaptive]`.\n\n### Does this PR introduce _any_ user-facing change?\n\nYes. Eligible obsolete shuffle jobs may be cancelled under cleanup modes that do not remove submitted shuffle files, and listeners may observe corresponding failed/cancelled job events. Broadcast jobs, opaque delegated shuffles, and submitted shuffles subject to immediate file removal are intentionally left running. An eventual ordinary failure from a private obsolete broadcast or identifiable cleanup-protected shuffle may be ignored after cleanup checks; an opaque delegated shuffle\u0027s failure still propagates because its hidden shuffle cannot be safely accounted for.\n\n### How was this PR tested?\n\nThe latest owner-aware exchange reuse, alias-wide failure handling, direct shuffle-registration, and sticky failed-cancellation changes passed both validation runs on top of published PR head `4baed923d832a9452d48d6a08a97be04c06e73e6`. The owner-aware follow-up remains local, uncommitted, and unpublished.\n\n- Focused exchange-reuse regressions: **13 tests passed**, with **0 failures, 0 errors, and 0 skipped tests**; JUnit XML timestamp `2026-08-19T14:03:59`, duration `4.893s`.\n- Complete `AdaptiveQueryExecSuite`: **144 tests passed**, with **0 failures, 0 errors, and 0 skipped tests**; JUnit XML timestamp `2026-08-19T14:07:18`, duration `127.294s`.\n\nThe regression coverage includes:\n\n- A mutation-sensitive conditionless left-semi join over a global aggregate on empty input, with assertions for the original join and its adaptive rewrite.\n- A real planned `ORDER BY ... LIMIT ... DISTINCT` query that establishes a logically linked physical sort/shuffle shape, plus direct rule assertions for sort and limit wrappers derived from that shape.\n- Zero and positive limits, offsets, unknown statistics, grouped aggregates, and the single output row of global aggregates over empty input.\n- Exchange reuse, private/shared-stage protection, cache retention after failed cancellation, fatal-stage failure propagation, and broadcast cancellation safety.\n- Same-plan reuse with multiple dropped aliases, one underlying cancellation, and an ordinary failure callback for each alias; retained aliases and fatal alias failures remain protected.\n- Reuse by a separately owned adaptive subquery remains protected even when its plan compares structurally equal to the original owner; cache entries with unknown owners also remain protected.\n- A failed cancellation followed by later same-plan reuse keeps the original exchange cached, never retries the already-marked native cancellation, and guards eventual failures from the additional aliases.\n- Historical same-plan alias IDs remain tracked when an earlier adaptive-plan adoption has already dropped those aliases; subsequent cancellation returns every alias ID and suppresses each obsolete ordinary failure.\n- Submitted-shuffle ID registration, including a reused-exchange alias backed by its actual owner-created shuffle, unsubmitted lazy-dependency safety, and `RemoveShuffleFiles` protection against active map writers. A submitted-shuffle regression checks that a wrapped fatal failure still propagates, a later ordinary obsolete-stage failure is ignored, its shuffle ID is registered and its failed cache entry is removed, and its underlying shuffle action is never cancelled.\n- An opaque delegated exchange whose own `futureAction` is empty remains uncancelled and cached; its later ordinary failure is not suppressed because the hidden delegate\u0027s shuffle ID cannot safely be registered.\n- A stage present in a constructed previous physical plan but absent from the latest replacement batch, including deduplication, cancellation, later nonfatal cancelled-stage failure suppression, and cache retention for the stage kept by the new plan.\n- Unrelated-exchange progress while either a generic cancellation callback or a real shuffle monitor is blocked; same-result reuse waits outside the global lock and reuses the original cached stage if cancellation fails.\n\nThe focused regressions and complete suite were run against the updated integration worktree:\n\n```bash\n/private/tmp/spark-pr55839-sbt.sh \\\n  \u0027sql/testOnly org.apache.spark.sql.execution.adaptive.AdaptiveQueryExecSuite -- -z \"reuse\"\u0027\n/private/tmp/spark-pr55839-sbt.sh \\\n  \u0027sql/testOnly org.apache.spark.sql.execution.adaptive.AdaptiveQueryExecSuite\u0027\n```\n\n### Was this patch authored or co-authored using generative AI tooling?\n\nGenerated-by: OpenAI Codex\n\nCloses #55839 from sunchao/dev/chao/codex/oss-aqe-empty-stage.\n\nLead-authored-by: Chao Sun \u003cchao@openai.com\u003e\nCo-authored-by: fredliu-data \u003cfredliu@openai.com\u003e\nSigned-off-by: Chao Sun \u003cchao@openai.com\u003e\n",
  "tree_diff": [
    {
      "type": "modify",
      "old_id": "e2a013b9e814c2d7c3513710d280d03e4cf44177",
      "old_mode": 33188,
      "old_path": "sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AQEPropagateEmptyRelation.scala",
      "new_id": "d9107c6a128e78357b58d9a6fba0deb5ccd8bf5d",
      "new_mode": 33188,
      "new_path": "sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AQEPropagateEmptyRelation.scala"
    },
    {
      "type": "modify",
      "old_id": "bfe6a9a3f633254b99838efdfa1df859eb8c4e72",
      "old_mode": 33188,
      "old_path": "sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AdaptiveSparkPlanExec.scala",
      "new_id": "a12b202b5b0090141160259941098f9773cadb51",
      "new_mode": 33188,
      "new_path": "sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AdaptiveSparkPlanExec.scala"
    },
    {
      "type": "modify",
      "old_id": "9523ce897ecf7a62cd31f71f5daa4493307693bd",
      "old_mode": 33188,
      "old_path": "sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala",
      "new_id": "8e98d7785f6c0dd1d8b07597899032c67760e70f",
      "new_mode": 33188,
      "new_path": "sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala"
    }
  ]
}