[SPARK-58561][INFRA] Support backporting a PR merged into a non-default branch

dev/merge_spark_pr.py

@@ -415,6 +415,118 @@ def get_json(url):
         sys.exit(-1)
 
 
+def merge_footer_pr(message):
+    """The PR number in `message`'s generated merge footer, or None if it has none.
+
+    A commit message cannot be searched for "Closes #N from " to identify the merge of N:
+    `merge_pr` passes the PR body through as its own `git commit -m` paragraph, so a body
+    discussing or reverting another commit can quote that commit's entire footer, structure
+    included. What is unforgeable is *position*: `merge_pr` appends the footer last, so the
+    generated one is the final "Closes" paragraph in the message. Later `cherry-pick -x`
+    lines may follow it, but no further "Closes" paragraph can. So read the last one and
+    compare its number, rather than searching for a number anywhere.
+
+    >>> footer = "Closes #1 from a/b.\\n\\nAuthored-by: A <a@e.org>\\nSigned-off-by: C <c@e.org>"
+    >>> merge_footer_pr("[SPARK-1][SQL] Title\\n\\nSome body.\\n\\n" + footer)
+    1
+    >>> merge_footer_pr("[SPARK-1][SQL] Title\\n\\n" + footer.replace("Authored", "Lead-authored"))
+    1
+    >>> merge_footer_pr("[SPARK-1][SQL] Title\\n\\nNo footer here.") is None
+    True
+
+    A cherry-pick keeps the footer, with `-x` provenance appended after it:
+
+    >>> pick = footer + "\\n(cherry picked from commit abc123)\\nSigned-off-by: C <c@e.org>"
+    >>> merge_footer_pr("[SPARK-1][SQL] Title\\n\\n" + pick)
+    1
+
+    A body quoting another PR's complete footer does not shadow the real one:
+
+    >>> quoted = "Reverting:\\n\\n" + footer + "\\n\\nSee above."
+    >>> own = footer.replace("#1", "#2")
+    >>> merge_footer_pr("[SPARK-2][SQL] Later\\n\\n%s\\n\\n%s" % (quoted, own))
+    2
+    """
+    # \s*$ tolerates trailing whitespace. The blank line and authors paragraph reject prose
+    # that merely mentions a PR; taking the LAST match rejects a body quoting a real footer.
+    footer = re.compile(
+        r"^Closes #(\d+) from \S+\s*$\n\n(?:Lead-authored-by|Authored-by):",
+        re.MULTILINE,
+    )
+    matches = footer.findall(message)
+    return int(matches[-1]) if matches else None
+
+
+def has_merge_footer(message, pr_num):
+    """Whether `message`'s generated merge footer closes `pr_num`. See `merge_footer_pr`.
+
+    >>> footer = "Closes #1 from a/b.\\n\\nAuthored-by: A <a@e.org>\\nSigned-off-by: C <c@e.org>"
+    >>> has_merge_footer("[SPARK-1][SQL] Title\\n\\n" + footer, 1)
+    True
+    >>> has_merge_footer("[SPARK-1][SQL] Title\\n\\n" + footer, 2)
+    False
+
+    A commit whose body quotes another PR's full footer is not taken for that PR's merge:
+
+    >>> quoted = "Reverting:\\n\\n" + footer + "\\n\\nSee above."
+    >>> later = "[SPARK-2][SQL] Later\\n\\n%s\\n\\n%s" % (quoted, footer.replace("#1", "#2"))
+    >>> has_merge_footer(later, 1)
+    False
+    >>> has_merge_footer(later, 2)
+    True
+    """
+    return merge_footer_pr(message) == pr_num
+
+
+def merge_commit_candidates(pr_events):
+    """Split `pr_events` into (closed_commits, referenced_commits), each oldest-first.
+
+    Ordered by time so that a PR reopened and merged again yields its latest merge last.
+
+    >>> merge_commit_candidates([{"event": "closed", "commit_id": "a", "created_at": "t2"},
+    ...                          {"event": "referenced", "commit_id": "b", "created_at": "t1"}])
+    (['a'], ['b'])
+    >>> merge_commit_candidates([{"event": "closed", "commit_id": None, "created_at": "t1"}])
+    ([], [])
+    >>> merge_commit_candidates([{"event": "referenced", "commit_id": "c", "created_at": "t2"},
+    ...                          {"event": "referenced", "commit_id": "b", "created_at": "t1"}])
+    ([], ['b', 'c'])
+    """
+
+    def commits_of(event_name):
+        matched = [e for e in pr_events if e["event"] == event_name and e["commit_id"] is not None]
+        return [e["commit_id"] for e in sorted(matched, key=lambda x: x["created_at"])]
+
+    return commits_of("closed"), commits_of("referenced")
+
+
+def find_merge_commit(pr_num, pr_events):
+    """Return (hash, message) of the commit that merged `pr_num`, or (None, None).
+
+    GitHub attributes the merge commit to the `closed` event only when that commit lands
+    on the default branch (master), because the "Closes #N" keyword in the commit message
+    is what closes the PR and the keyword is honored only there. A PR merged into any
+    other branch -- e.g. one opened against a rolling branch-M.x -- is instead closed by
+    this script through the API, and that `closed` event carries no commit, so the merge
+    survives only as a `referenced` event. Prefer the `closed` commit, which GitHub itself
+    linked; otherwise fall back to `referenced` events, which are also raised by any commit
+    merely mentioning the PR, so confirm each against the merge footer `merge_pr` generates.
+    """
+
+    def message_of(commit_hash):
+        return get_json("%s/commits/%s" % (GITHUB_API_BASE, commit_hash))["commit"]["message"]
+
+    closed_commits, referenced_commits = merge_commit_candidates(pr_events)
+    if closed_commits:
+        return closed_commits[-1], message_of(closed_commits[-1])
+
+    for commit_hash in reversed(referenced_commits):
+        message = message_of(commit_hash)
+        if has_merge_footer(message, pr_num):
+            return commit_hash, message
+    return None, None
+
+
 def close_pr(pr_num):
     url = "%s/pulls/%s" % (GITHUB_API_BASE, pr_num)
     data = json.dumps({"state": "closed"}).encode("utf-8")
@@ -615,6 +727,83 @@ def _do_cherry_pick(pr_num, merge_hash, pick_ref):
     return pick_ref, pick_hash
 
 
+def branches_with_merge_footer(pr_num, branch_names):
+    """Release branches from `branch_names` that already carry `pr_num`'s merge footer.
+
+    A cherry-pick is a new commit, so `git branch --contains <merge_hash>` finds only the
+    branch the change was merged into; what identifies a backport is the footer, which
+    `cherry-pick -x` copies verbatim (the same signal `dev/pr_merge_status.py` reads).
+    `git log --grep` only narrows the walk -- it matches the fragment anywhere in a message,
+    including inside a copied PR body -- so every candidate is confirmed with
+    `has_merge_footer` before its branches count as already backported.
+    Best-effort: this only sees branches already fetched into PUSH_REMOTE_NAME's tracking
+    refs, so a backport pushed from elsewhere and not yet fetched is simply not reported --
+    the committer is still prompted and can type any branch.
+    """
+    trailer = "Closes #%s from " % pr_num
+    try:
+        # %x00 delimits records so a commit message (which spans lines) stays one field.
+        out = run_cmd(
+            [
+                "git",
+                "log",
+                "--remotes=%s" % PUSH_REMOTE_NAME,
+                "--fixed-strings",
+                "--grep",
+                trailer,
+                "--format=%H %B%x00",
+            ]
+        )
+    except Exception as e:
+        print_error("Could not scan for existing backports of #%s (%s)." % (pr_num, e))
+        return []
+
+    landed = set()
+    prefix = "%s/" % PUSH_REMOTE_NAME
+    for record in out.split("\0"):
+        record = record.strip("\n")
+        if not record:
+            continue
+        commit_hash, _, message = record.partition(" ")
+        # --grep matched somewhere in the message; only the generated footer counts.
+        if not has_merge_footer(message, pr_num):
+            continue
+        refs = run_cmd(
+            [
+                "git",
+                "for-each-ref",
+                "--contains",
+                commit_hash,
+                "--format=%(refname:short)",
+                "refs/remotes/%s/" % PUSH_REMOTE_NAME,
+            ]
+        )
+        for ref in refs.splitlines():
+            if ref.startswith(prefix):
+                landed.add(ref[len(prefix) :])
+    # Keep branch_names' newest-first order, and drop anything not a known release branch.
+    return [b for b in branch_names if b in landed]
+
+
+def default_pick_branch(branch_names, already_picked):
+    """Highest-ranked release branch that has not already received the change, or None.
+
+    `branch_names` is ordered newest-first (see `semver_branch_rank`) and `already_picked`
+    holds the branches the change is known to be on, so the prompt never defaults to a
+    branch where the cherry-pick would come up empty. Returns None when every known branch
+    already has it, so callers can say so instead of offering an empty pick.
+
+    >>> default_pick_branch(["branch-4.x", "branch-4.3", "branch-4.2"], ("branch-4.x",))
+    'branch-4.3'
+    >>> default_pick_branch(["branch-4.x", "branch-4.3"], ())
+    'branch-4.x'
+    >>> default_pick_branch(["branch-4.x"], ("branch-4.x",)) is None
+    True
+    """
+    remaining = [b for b in branch_names if b not in already_picked]
+    return remaining[0] if remaining else None
+
+
 def _upstream_first_sibling(target_ref, pick_ref, branch_names, already_picked):
     """Return the sibling branch-M.x if Upstream-First should prompt, else None.
 
@@ -1663,17 +1852,15 @@ def main():
 
     # Merged pull requests don't appear as merged in the GitHub API;
     # Instead, they're closed by committers.
-    merge_commits = [e for e in pr_events if e["event"] == "closed" and e["commit_id"] is not None]
-
-    if merge_commits and pr["state"] == "closed":
-        # A PR might have multiple merge commits, if it's reopened and merged again. We shall
-        # cherry-pick PRs in closed state with the latest merge hash.
-        # If the PR is still open(reopened), we shall not cherry-pick it but perform the normal
-        # merge as it could have been reverted earlier.
-        merge_commits = sorted(merge_commits, key=lambda x: x["created_at"])
-        merge_hash = merge_commits[-1]["commit_id"]
-        message = get_json("%s/commits/%s" % (GITHUB_API_BASE, merge_hash))["commit"]["message"]
-
+    # A PR might have multiple merge commits, if it's reopened and merged again. We shall
+    # cherry-pick PRs in closed state with the latest merge hash.
+    # If the PR is still open(reopened), we shall not cherry-pick it but perform the normal
+    # merge as it could have been reverted earlier.
+    merge_hash, message = (None, None)
+    if pr["state"] == "closed":
+        merge_hash, message = find_merge_commit(pr_num, pr_events)
+
+    if merge_hash is not None:
         print("Pull request %s has already been merged, assuming you want to backport" % pr_num)
         commit_is_downloaded = (
             run_cmd(["git", "rev-parse", "--quiet", "--verify", "%s^{commit}" % merge_hash]).strip()
@@ -1683,11 +1870,41 @@ def main():
             fail("Couldn't find any merge commit for #%s, you may need to update HEAD." % pr_num)
 
         print("Found commit %s:\n%s" % (merge_hash, message))
-        default = branch_names[0]
-        picked = cherry_pick(
-            pr_num, merge_hash, default, branch_names, target_ref, already_picked=()
-        )
-        post_merge_comment(pr_num, picked)
+        # The change is already on target_ref and on any branch a previous run backported it
+        # to, so exclude all of them: defaulting to one would cherry-pick an empty commit.
+        picked_refs = [target_ref] + [
+            b for b in branches_with_merge_footer(pr_num, branch_names) if b != target_ref
+        ]
+        if len(picked_refs) > 1:
+            print("Already backported to: %s" % ", ".join(picked_refs[1:]))
+        # Loop so one invocation can reach several maintenance branches, as the merge path does.
+        picked_commits = []
+        try:
+            while True:
+                default = default_pick_branch(branch_names, tuple(picked_refs))
+                if default is None:
+                    print(
+                        "Every known release branch already contains #%s; nothing to pick." % pr_num
+                    )
+                    break
+                picked = cherry_pick(
+                    pr_num,
+                    merge_hash,
+                    default,
+                    branch_names,
+                    target_ref,
+                    already_picked=tuple(picked_refs),
+                )
+                picked_refs = picked_refs + [ref for ref, _ in picked]
+                picked_commits = picked_commits + picked
+                prompt = "Would you like to pick %s into another branch?" % merge_hash
+                if get_input(f"\n{prompt} (y/N): ", ["y", "n", ""]) != "y":
+                    break
+        finally:
+            # Report whatever was pushed even if a later pick is aborted, since the earlier
+            # pushes have already landed.
+            if picked_commits:
+                post_merge_comment(pr_num, picked_commits)
         sys.exit(0)
 
     if not bool(pr["mergeable"]):
@@ -1765,19 +1982,20 @@ def main():
     # then each cherry-pick target as it is picked.
     merged_commits = [(target_ref, merge_hash)]
 
-    # Walk a mutable remaining-branches list so the next default correctly skips any
-    # branches already picked, including branches consumed by the Upstream-First two-branch
-    # path inside cherry_pick (e.g. picking branch-M.x + branch-M.N in a single prompt).
-    # merged_refs doubles as the already_picked set passed to cherry_pick: it starts with
-    # target_ref (the merge sink, never to be re-picked) and grows with every cherry-pick.
-    remaining_branches = [b for b in branch_names if b != target_ref]
+    # merged_refs drives both the next prompt default and the already_picked set passed to
+    # cherry_pick, so each grows with every cherry-pick -- including branches consumed by the
+    # Upstream-First two-branch path inside cherry_pick (e.g. picking branch-M.x + branch-M.N
+    # in a single prompt). It starts with target_ref, the merge sink, never to be re-picked.
     pick_prompt = "Would you like to pick %s into another branch?" % merge_hash
     # Always record the merge summary for what actually landed, even if a later
     # cherry-pick is aborted or cancelled: the merge into the target branch has
     # already been pushed, so cancelling a backport must not drop that line.
     try:
         while get_input(f"\n{pick_prompt} (y/N): ", ["y", "n", ""]) == "y":
-            default = remaining_branches[0] if remaining_branches else branch_names[0]
+            default = default_pick_branch(branch_names, tuple(merged_refs))
+            if default is None:
+                print("Every known release branch already contains #%s; nothing to pick." % pr_num)
+                break
             picked = cherry_pick(
                 pr_num,
                 merge_hash,
@@ -1786,12 +2004,8 @@ def main():
                 target_ref,
                 already_picked=tuple(merged_refs),
             )
-            picked_refs = [ref for ref, _ in picked]
-            merged_refs = merged_refs + picked_refs
+            merged_refs = merged_refs + [ref for ref, _ in picked]
             merged_commits = merged_commits + picked
-            for b in picked_refs:
-                if b in remaining_branches:
-                    remaining_branches.remove(b)
     finally:
         if merged_commits:
             # The "Closes #N" keyword in the commit message only auto-closes the PR when the