This is an automated email from the ASF dual-hosted git repository. blaginin pushed a commit to branch db/ci-merge-queue-gate in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit 1cf01519f6da25c25c5a70c839267f307a298d43 Author: blaginin <[email protected]> AuthorDate: Fri Sep 11 15:04:51 2026 +0100 ci: add aggregate required check and merge queue support --- .asf.yaml | 3 + .github/actions/check-ci-result/action.yaml | 31 ++++++ .github/workflows/ci.yml | 47 ++++++++- dev/ci/check-ci-result.py | 69 +++++++++++++ dev/ci/compute-changes.py | 25 ++--- dev/ci/test_ci_queue.py | 145 ++++++++++++++++++++++++++++ 6 files changed, 306 insertions(+), 14 deletions(-) diff --git a/.asf.yaml b/.asf.yaml index 21562b811c..2548cc7806 100644 --- a/.asf.yaml +++ b/.asf.yaml @@ -46,6 +46,9 @@ github: main: required_pull_request_reviews: required_approving_review_count: 1 + required_status_checks: + contexts: + - "CI Required" branch-0.1: required_pull_request_reviews: required_approving_review_count: 1 diff --git a/.github/actions/check-ci-result/action.yaml b/.github/actions/check-ci-result/action.yaml new file mode 100644 index 0000000000..d54142b092 --- /dev/null +++ b/.github/actions/check-ci-result/action.yaml @@ -0,0 +1,31 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +name: Check CI result +description: Require every planned CI workflow to finish successfully +inputs: + needs-json: + description: JSON of the caller job's complete needs context + required: true +runs: + using: composite + steps: + - name: Verify planned CI results + shell: bash + env: + CI_NEEDS_JSON: ${{ inputs.needs-json }} + run: python3 "$GITHUB_WORKSPACE/dev/ci/check-ci-result.py" diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f53d854ba6..98aaf80000 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -27,7 +27,7 @@ name: CI # in their own subgroup so they never cancel the real commit run; opened and # synchronize both map to `commit` so a new push still supersedes its predecessor. concurrency: - group: ${{ github.repository }}-${{ github.head_ref || github.sha }}-${{ github.workflow }}-${{ github.event.action == 'labeled' && github.event.label.name || 'commit' }} + group: ${{ github.repository }}-${{ github.head_ref || github.sha }}-${{ github.workflow }}-${{ github.event_name }}-${{ github.event.action == 'labeled' && github.event.label.name || 'commit' }} cancel-in-progress: true on: @@ -37,6 +37,8 @@ on: branches: - main workflow_dispatch: + merge_group: + types: [checks_requested] jobs: # --------------------------------------------------------------------------- @@ -98,6 +100,9 @@ jobs: - name: Check CI config invariants run: python3 dev/ci/check-ci-config.py + - name: Check CI gate and queue routing + run: python3 -m unittest discover -s dev/ci -p 'test_ci_*.py' + - name: Install actionlint run: | curl -sSfL https://raw.githubusercontent.com/rhysd/actionlint/main/scripts/download-actionlint.bash | bash @@ -150,6 +155,8 @@ jobs: PR_HEAD_SHA: ${{ github.event.pull_request.head.sha }} PUSH_BEFORE: ${{ github.event.before }} PUSH_AFTER: ${{ github.sha }} + QUEUE_BASE_SHA: ${{ github.event.merge_group.base_sha }} + QUEUE_HEAD_SHA: ${{ github.event.merge_group.head_sha }} run: | set -euo pipefail : > changed_files.txt @@ -160,13 +167,19 @@ jobs: : elif [[ "$EVENT_NAME" == "pull_request" ]]; then git diff --name-only "$PR_BASE_SHA"..."$PR_HEAD_SHA" > changed_files.txt - else + elif [[ "$EVENT_NAME" == "merge_group" ]]; then + # Include every queued change relative to the current base branch. + git diff --name-only "${QUEUE_BASE_SHA:?}".."${QUEUE_HEAD_SHA:?}" > changed_files.txt + elif [[ "$EVENT_NAME" == "push" ]]; then # push to main; first push to a branch has all-zero before sha if [[ "$PUSH_BEFORE" =~ ^0+$ ]]; then git ls-tree -r --name-only "$PUSH_AFTER" > changed_files.txt else git diff --name-only "$PUSH_BEFORE".."$PUSH_AFTER" > changed_files.txt fi + else + echo "Unsupported CI event: $EVENT_NAME" >&2 + exit 1 fi echo "Changed files:" cat changed_files.txt @@ -296,7 +309,7 @@ jobs: iceberg_1_11: name: Iceberg Spark SQL Tests (Iceberg 1.11) needs: changes - # Runs on every PR: Iceberg 1.11 is our only Spark 4.1 Iceberg coverage. + # Runs on PRs, main, and the merge queue; this is our Spark 4.1 Iceberg coverage. if: needs.changes.outputs.iceberg_1_11 == 'true' uses: ./.github/workflows/iceberg_spark_test_reusable.yml with: @@ -305,3 +318,31 @@ jobs: spark-short: '4.1' spark-full: '4.1.3' java: 17 + + ci-required: + # Label/manual runs must never overwrite the canonical PR/queue verdict. + name: ${{ (github.event.action == 'labeled' || github.event_name == 'workflow_dispatch') && 'CI Optional' || 'CI Required' }} + if: ${{ always() }} + needs: + - preflight + - changes + - pr_build_linux + - pr_build_macos + - pr_benchmark_check + - docs + - spark_3_4 + - spark_3_5 + - spark_4_0 + - spark_4_1 + - iceberg_1_8 + - iceberg_1_9 + - iceberg_1_10 + - iceberg_1_11 + runs-on: ubuntu-slim + permissions: + contents: read + steps: + - uses: actions/checkout@v7 + - uses: ./.github/actions/check-ci-result + with: + needs-json: ${{ toJSON(needs) }} diff --git a/dev/ci/check-ci-result.py b/dev/ci/check-ci-result.py new file mode 100644 index 0000000000..9ae801a833 --- /dev/null +++ b/dev/ci/check-ci-result.py @@ -0,0 +1,69 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +"""Validate the umbrella's results, including jobs legitimately skipped by policy.""" + +import json +import os +import sys + +# Routing output -> reusable caller id. Keep synchronized with ci.yml. +JOBS = { + "build_linux": "pr_build_linux", + "build_macos": "pr_build_macos", + "benchmark": "pr_benchmark_check", + "docs": "docs", + "spark_3_4": "spark_3_4", + "spark_3_5": "spark_3_5", + "spark_4_0": "spark_4_0", + "spark_4_1": "spark_4_1", + "iceberg_1_8": "iceberg_1_8", + "iceberg_1_9": "iceberg_1_9", + "iceberg_1_10": "iceberg_1_10", + "iceberg_1_11": "iceberg_1_11", +} + + +def failures(needs): + errors = [] + expected = {"preflight", "changes", *JOBS.values()} + if set(needs) != expected: + errors.append(f"Unexpected dependency set: missing {sorted(expected - set(needs))}, " + f"extra {sorted(set(needs) - expected)}") + for job in ("preflight", "changes"): + if needs.get(job, {}).get("result") != "success": + errors.append(f"{job} must succeed") + plan = needs.get("changes", {}).get("outputs", {}) + for output, job in JOBS.items(): + flag = plan.get(output) + result = needs.get(job, {}).get("result") + if flag not in ("true", "false"): + errors.append(f"Missing or invalid plan for {job}: {flag!r}") + elif flag == "true" and result != "success": + errors.append(f"Required {job}: {result!r}") + elif flag == "false" and result != "skipped": + errors.append(f"Unplanned {job}: {result!r} (expected skipped)") + return errors + + +if __name__ == "__main__": + errors = failures(json.loads(os.environ["CI_NEEDS_JSON"])) + for error in errors: + print(error, file=sys.stderr) + if errors: + sys.exit(1) + print("All planned CI checks passed") diff --git a/dev/ci/compute-changes.py b/dev/ci/compute-changes.py index 0eb4cc2446..c84a4f8eee 100644 --- a/dev/ci/compute-changes.py +++ b/dev/ci/compute-changes.py @@ -302,6 +302,7 @@ FILTERS = { # # "pr" every pull request # "push" push to main +# "queue" merge_group/checks_requested # "label:<name>" a pull request carrying that label # # workflow_dispatch always runs everything, so it is not listed. "pr" and @@ -309,20 +310,20 @@ FILTERS = { # requests or opt-in, never both -- and check-ci-config.py rejects a job that # lists both rather than letting the label quietly win. POLICY = { - "build_linux": ["pr", "push"], - "build_macos": ["pr", "push"], - "benchmark": ["pr", "push"], + "build_linux": ["pr", "push", "queue"], + "build_macos": ["pr", "push", "queue"], + "benchmark": ["pr", "push", "queue"], # docs deploys to asf-site, so it must not run from a pull request. "docs": ["push"], - "spark_3_4": ["push", "label:run-spark-3.4-tests"], - "spark_3_5": ["pr", "push"], - "spark_4_0": ["push", "label:run-spark-4.0-tests"], - "spark_4_1": ["pr", "push"], - "iceberg_1_8": ["push", "label:run-iceberg-tests"], - "iceberg_1_9": ["push", "label:run-iceberg-tests"], - "iceberg_1_10": ["push", "label:run-iceberg-tests"], + "spark_3_4": ["push", "queue", "label:run-spark-3.4-tests"], + "spark_3_5": ["pr", "push", "queue"], + "spark_4_0": ["push", "queue", "label:run-spark-4.0-tests"], + "spark_4_1": ["pr", "push", "queue"], + "iceberg_1_8": ["push", "queue", "label:run-iceberg-tests"], + "iceberg_1_9": ["push", "queue", "label:run-iceberg-tests"], + "iceberg_1_10": ["push", "queue", "label:run-iceberg-tests"], # Iceberg 1.11 is our only Spark 4.1 Iceberg coverage, so it is not opt-in. - "iceberg_1_11": ["pr", "push"], + "iceberg_1_11": ["pr", "push", "queue"], } @@ -342,6 +343,8 @@ def event_allows(job, event): if name == "workflow_dispatch": return True + if name == "merge_group": + return event.get("action") == "checks_requested" and "queue" in tiers if name == "push": return "push" in tiers if name != "pull_request": diff --git a/dev/ci/test_ci_queue.py b/dev/ci/test_ci_queue.py new file mode 100644 index 0000000000..13c18eaf85 --- /dev/null +++ b/dev/ci/test_ci_queue.py @@ -0,0 +1,145 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +import importlib.util +from pathlib import Path +import re +import unittest + +ROOT = Path(__file__).resolve().parents[2] + + +def load(name): + spec = importlib.util.spec_from_file_location(name, ROOT / "dev/ci" / f"{name}.py") + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +routing = load("compute-changes") +gate = load("check-ci-result") +PR_JOBS = {"build_linux", "build_macos", "benchmark", "spark_3_5", "spark_4_1", "iceberg_1_11"} +OTHER_VERSIONS = {"spark_3_4", "spark_4_0", "iceberg_1_8", "iceberg_1_9", "iceberg_1_10"} +QUEUE_JOBS = PR_JOBS | OTHER_VERSIONS + + +class QueuePolicyTest(unittest.TestCase): + def event(self, name, action="", labels=(), label=""): + return {"name": name, "action": action, "labels": list(labels), + "label": label} + + def allowed(self, event): + return {job for job in routing.POLICY if routing.event_allows(job, event)} + + def test_queue_and_push_run_full_coverage_without_reducing_prs(self): + self.assertEqual(self.allowed(self.event("pull_request", "synchronize")), PR_JOBS) + self.assertEqual(self.allowed(self.event("push")), QUEUE_JOBS | {"docs"}) + self.assertEqual(self.allowed(self.event("merge_group", "checks_requested")), QUEUE_JOBS) + + def test_unknown_event_and_queue_action_cannot_launch_jobs(self): + for event in (self.event("schedule"), self.event("merge_group", "destroyed")): + self.assertEqual(self.allowed(event), set()) + + def test_label_opt_in_does_not_rerun_primary(self): + for label, expected in (("run-spark-3.4-tests", {"spark_3_4"}), + ("run-spark-4.0-tests", {"spark_4_0"}), + ("run-iceberg-tests", {"iceberg_1_8", "iceberg_1_9", + "iceberg_1_10"})): + with self.subTest(label=label): + self.assertEqual(self.allowed(self.event("pull_request", "labeled", [label], label)), expected) + self.assertEqual(self.allowed(self.event("pull_request", "synchronize", [label])), PR_JOBS | expected) + self.assertEqual(self.allowed(self.event("pull_request", "labeled", [label, "dependencies"], "dependencies")), set()) + + def test_docs_and_combined_queue_changes(self): + event = self.event("merge_group", "checks_requested") + self.assertFalse(any(routing.compute(["docs/source/user-guide/overview.md"], event).values())) + # The queue diff can include different integration edits from two PRs. + jobs = routing.compute(["dev/diffs/3.5.9.diff", "dev/diffs/iceberg/1.11.0.diff"], event) + self.assertEqual({k for k, v in jobs.items() if v}, { + "spark_3_5", "iceberg_1_8", "iceberg_1_9", "iceberg_1_10", "iceberg_1_11"}) + jobs = routing.compute(["native/core/src/lib.rs"], event) + self.assertEqual({k for k, v in jobs.items() if v}, QUEUE_JOBS - {"benchmark"}) + + def test_manual_run_exercises_all_coverage(self): + event = self.event("workflow_dispatch") + self.assertEqual(self.allowed(event), QUEUE_JOBS | {"docs"}) + + +class RequiredGateTest(unittest.TestCase): + def fixture(self, planned=()): + outputs = {key: str(key in planned).lower() for key in gate.JOBS} + needs = {"preflight": {"result": "success"}, + "changes": {"result": "success", "outputs": outputs}} + needs.update({job: {"result": "success" if key in planned else "skipped"} + for key, job in gate.JOBS.items()}) + return needs + + def test_docs_skip_and_full_queue_success(self): + self.assertEqual(gate.failures(self.fixture()), []) + self.assertEqual(gate.failures(self.fixture(QUEUE_JOBS)), []) + + def test_any_required_group_failure_cancellation_or_skip_blocks(self): + for job in gate.JOBS.values(): + for status in ("failure", "cancelled", "skipped", None): + with self.subTest(job=job, status=status): + needs = self.fixture(gate.JOBS) + needs[job]["result"] = status + self.assertTrue(gate.failures(needs)) + + def test_preflight_and_planner_cannot_hide_behind_skipped_children(self): + for job in ("preflight", "changes"): + for status in ("failure", "cancelled", "skipped"): + needs = self.fixture() + needs[job]["result"] = status + self.assertTrue(gate.failures(needs)) + + def test_missing_plan_and_unknown_dependency_fail_closed(self): + needs = self.fixture() + del needs["changes"]["outputs"]["iceberg_1_11"] + self.assertTrue(gate.failures(needs)) + needs = self.fixture() + needs["unknown"] = {"result": "success"} + self.assertTrue(gate.failures(needs)) + needs = self.fixture() + del needs["spark_4_1"] + self.assertTrue(gate.failures(needs)) + + def test_unplanned_failure_is_not_ignored(self): + needs = self.fixture() + needs["spark_4_1"]["result"] = "failure" + self.assertTrue(gate.failures(needs)) + + def test_workflow_wiring_covers_every_caller(self): + workflow = (ROOT / ".github/workflows/ci.yml").read_text() + blocks = dict(re.findall(r"^ ([\w-]+):\n(.*?)(?=^ [\w-]+:|\Z)", workflow.split("\njobs:\n", 1)[1], re.M | re.S)) + self.assertEqual(set(blocks), {"preflight", "changes", "ci-required", *gate.JOBS.values()}) + required = blocks["ci-required"] + for job in {"preflight", "changes", *gate.JOBS.values()}: + self.assertIn(f" - {job}\n", required) + for key, job in gate.JOBS.items(): + self.assertIn(f"if: needs.changes.outputs.{key} == 'true'", blocks[job]) + self.assertIn("if: ${{ always() }}", required) + self.assertIn("github.event.action == 'labeled'", required) + self.assertIn("'CI Optional' || 'CI Required'", required) + self.assertIn(" merge_group:\n types: [checks_requested]", workflow) + self.assertIn('"${QUEUE_BASE_SHA:?}".."${QUEUE_HEAD_SHA:?}"', workflow) + self.assertIn(' - "CI Required"', (ROOT / ".asf.yaml").read_text()) + self.assertEqual(set(gate.JOBS), set(routing.POLICY)) + + +if __name__ == "__main__": + unittest.main() --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
