This is an automated email from the ASF dual-hosted git repository.
sunchao pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
The following commit(s) were added to refs/heads/main by this push:
new 9b608cc509 ci: shard Iceberg Spark tests across four runners (#5459)
9b608cc509 is described below
commit 9b608cc50922bfeac83dde2199847d1c509967cb
Author: Chao Sun <[email protected]>
AuthorDate: Wed Sep 9 16:11:23 2026 -0700
ci: shard Iceberg Spark tests across four runners (#5459)
* ci: shard Iceberg Spark tests across four runners
* fix(ci): verify complete Iceberg shard inventories
---
.github/workflows/ci.yml | 3 +
.github/workflows/iceberg_spark_test_reusable.yml | 45 ++-
dev/ci/check-iceberg-shards.py | 313 +++++++++++++++++++++
dev/ci/compute-changes.py | 12 +
dev/ci/iceberg-test-shards.gradle | 110 ++++++++
dev/ci/test-iceberg-shards.py | 144 ++++++++++
.../contributor-guide/iceberg-spark-tests.md | 22 ++
7 files changed, 648 insertions(+), 1 deletion(-)
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 87a4ec53fd..a4484def25 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -91,6 +91,9 @@ jobs:
- name: Check pull request type labeling
run: node --test dev/ci/pr-type-label.test.mjs
+ - name: Check Iceberg shard inventory validation
+ run: python3 dev/ci/test-iceberg-shards.py
+
- name: Install actionlint
run: |
curl -sSfL
https://raw.githubusercontent.com/rhysd/actionlint/main/scripts/download-actionlint.bash
| bash
diff --git a/.github/workflows/iceberg_spark_test_reusable.yml
b/.github/workflows/iceberg_spark_test_reusable.yml
index 4b72cf47c2..c0ae5427e7 100644
--- a/.github/workflows/iceberg_spark_test_reusable.yml
+++ b/.github/workflows/iceberg_spark_test_reusable.yml
@@ -63,6 +63,9 @@ jobs:
build-native:
name: Build Native Library
runs-on: ubuntu-24.04
+ outputs:
+ shard-matrix: ${{ steps.shards.outputs.matrix }}
+ shard-count: ${{ steps.shards.outputs.count }}
container:
image: amd64/rust
steps:
@@ -74,6 +77,10 @@ jobs:
rust-version: ${{ env.RUST_VERSION }}
jdk-version: 17
+ - name: Define Iceberg test shards
+ id: shards
+ run: python3 dev/ci/check-iceberg-shards.py --github-output
"$GITHUB_OUTPUT"
+
- name: Restore Cargo cache
uses: actions/cache/restore@v6
with:
@@ -111,7 +118,10 @@ jobs:
iceberg-spark:
needs: build-native
- name: iceberg-spark/iceberg-${{ inputs.iceberg-full }}/spark-${{
inputs.spark-full }}/scala-${{ inputs.scala }}/java-${{ inputs.java }}
+ strategy:
+ fail-fast: false
+ matrix: ${{ fromJSON(needs.build-native.outputs.shard-matrix) }}
+ name: iceberg-spark/iceberg-${{ inputs.iceberg-full }}/spark-${{
inputs.spark-full }}/scala-${{ inputs.scala }}/java-${{ inputs.java
}}/shard-${{ matrix.shard }}
runs-on: ubuntu-24.04
container:
image: amd64/rust
@@ -142,7 +152,38 @@ jobs:
rm -rf /root/.m2/repository/org/apache/parquet # somehow parquet
cache requires cleanups
ENABLE_COMET=true ENABLE_COMET_ONHEAP=true ./gradlew
-DsparkVersions=${{ inputs.spark-short }} -DscalaVersion=${{ inputs.scala }}
-DflinkVersions= -DkafkaVersions= \
:iceberg-spark:iceberg-spark-${{ inputs.spark-short }}_${{
inputs.scala }}:test \
+ --init-script ../dev/ci/iceberg-test-shards.gradle \
+ -PcometShardTask=:iceberg-spark:iceberg-spark-${{
inputs.spark-short }}_${{ inputs.scala }}:test \
+ -PcometShardIndex=${{ matrix.shard }} -PcometShardCount=${{
needs.build-native.outputs.shard-count }} \
-Pquick=true -x javadoc
+ - name: Upload Iceberg shard inventory and test reports
+ if: ${{ !cancelled() }}
+ uses: actions/upload-artifact@v7
+ with:
+ name: iceberg-spark-${{ inputs.iceberg-full }}-spark-${{
inputs.spark-full }}-scala-${{ inputs.scala }}-jdk${{ inputs.java }}-shard-${{
matrix.shard }}-attempt-${{ github.run_attempt }}
+ path: |
+ apache-iceberg/**/build/comet-shards/*.json
+ apache-iceberg/**/build/test-results/test/*.xml
+ retention-days: 7
+
+ iceberg-spark-shard-coverage:
+ needs: iceberg-spark
+ if: ${{ !cancelled() }}
+ name: iceberg-spark-shard-coverage/iceberg-${{ inputs.iceberg-full
}}/spark-${{ inputs.spark-full }}/scala-${{ inputs.scala }}/java-${{
inputs.java }}
+ runs-on: ubuntu-slim
+ steps:
+ - uses: actions/checkout@v7
+ - name: Download Iceberg shard inventories
+ uses: actions/download-artifact@v8
+ with:
+ # Download all attempts: a failed-job rerun retains earlier
successful
+ # shards. The checker selects the latest inventory for each index.
+ pattern: iceberg-spark-${{ inputs.iceberg-full }}-spark-${{
inputs.spark-full }}-scala-${{ inputs.scala }}-jdk${{ inputs.java
}}-shard-*-attempt-*
+ path: iceberg-shard-reports
+ - name: Verify complete, disjoint Iceberg candidate coverage
+ run: |
+ python3 dev/ci/check-iceberg-shards.py --manifests
iceberg-shard-reports \
+ --task :iceberg-spark:iceberg-spark-${{ inputs.spark-short }}_${{
inputs.scala }}:test
iceberg-spark-extensions:
needs: build-native
@@ -206,6 +247,8 @@ jobs:
uses: ./.github/actions/setup-iceberg-builder
with:
iceberg-version: ${{ inputs.iceberg-full }}
+ - name: Verify test sharding preserves discovery
+ run: python3 dev/ci/check-iceberg-shards.py --gradle
"$PWD/apache-iceberg/gradlew"
- name: Run Iceberg Spark runtime tests
run: |
cd apache-iceberg
diff --git a/dev/ci/check-iceberg-shards.py b/dev/ci/check-iceberg-shards.py
new file mode 100644
index 0000000000..39de880a4c
--- /dev/null
+++ b/dev/ci/check-iceberg-shards.py
@@ -0,0 +1,313 @@
+# 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.
+
+"""Exercise the real Gradle/JUnit shard filter without building Spark or
Iceberg.
+
+Run with the Iceberg checkout's wrapper (or a Gradle executable):
+ python3 dev/ci/check-iceberg-shards.py --gradle
"$PWD/apache-iceberg/gradlew"
+
+The fixture downloads only JUnit. --junit-classpath accepts local JUnit jars
for
+offline checks. --work-dir retains the fixture, logs, and comparison results.
+--github-output exports the workflow matrix/count from one shared definition.
+--manifests verifies the real inventories downloaded from an Iceberg CI run.
+"""
+
+import argparse
+from collections import Counter
+import json
+import os
+from pathlib import Path
+import shutil
+import subprocess
+import tempfile
+import xml.etree.ElementTree as ET
+
+
+INIT_SCRIPT = Path(__file__).with_name("iceberg-test-shards.gradle")
+# The workflow matrix and Gradle's partition count must come from the same
value,
+# not strategy.job-total (which includes any other matrix dimensions).
+SHARD_COUNT = 4
+
+
+def verify_manifests(root, task):
+ """Check each latest shard attempt against its independently captured
baseline.
+
+ Rerunning only failed jobs leaves successful jobs' artifacts in an earlier
+ attempt. Keep those, but never count two attempts of the same shard twice.
+ Artifact download is scoped to one run and Iceberg/Spark/Scala/JDK tuple.
+ """
+ attempts = {}
+ for path in sorted(root.rglob("*.json")):
+ manifest = json.loads(path.read_text())
+ for key in ("shard", "count", "attempt"):
+ if type(manifest.get(key)) is not int or manifest[key] < 1:
+ raise ValueError(f"{path}: {key} must be a positive integer")
+ key = manifest["shard"], manifest["attempt"]
+ if key in attempts:
+ raise ValueError(f"Duplicate inventory for shard/attempt {key}:
{path}")
+ attempts[key] = manifest
+
+ expected = set(range(1, SHARD_COUNT + 1))
+ indices = {index for index, _ in attempts}
+ if indices != expected:
+ raise ValueError(f"Expected shard indices {sorted(expected)}, found
{sorted(indices)}")
+
+ baseline = None
+ combined = Counter()
+ for index in sorted(expected):
+ attempt = max(attempt for shard, attempt in attempts if shard == index)
+ manifest = attempts[index, attempt]
+ if manifest.get("task") != task or manifest["count"] != SHARD_COUNT:
+ raise ValueError(f"Shard {index}: mismatched task or shard count")
+ for field in ("candidates", "unshardedCandidates"):
+ values = manifest.get(field)
+ if not isinstance(values, list) or not all(isinstance(v, str) for
v in values):
+ raise ValueError(f"Shard {index}: {field} must be a list of
class paths")
+ if len(set(values)) != len(values):
+ raise ValueError(f"Shard {index}: duplicate class paths in
{field}")
+ inventory = set(manifest["unshardedCandidates"])
+ if not inventory:
+ raise ValueError(f"Shard {index}: empty unsharded candidate
inventory")
+ if baseline is not None and baseline != inventory:
+ raise ValueError(f"Shard {index}: unsharded inventories disagree")
+ baseline = inventory
+ combined.update(manifest["candidates"])
+
+ if set(combined) != baseline:
+ missing, extra = baseline - set(combined), set(combined) - baseline
+ raise ValueError(f"Shard coverage mismatch: missing={sorted(missing)},
extra={sorted(extra)}")
+ duplicates = sorted(name for name, count in combined.items() if count != 1)
+ if duplicates:
+ raise ValueError(f"Candidate classes selected by multiple shards:
{duplicates}")
+ print(f"Verified {SHARD_COUNT} shards and {len(baseline)} candidate
classes: "
+ "the selected inventories equal the unsharded inventory exactly
once.", flush=True)
+
+
+def write_workflow_matrix(output):
+ with output.open("a") as stream:
+ stream.write(f"matrix={json.dumps({'shard': list(range(1, SHARD_COUNT
+ 1))})}\n")
+ stream.write(f"count={SHARD_COUNT}\n")
+
+
+def groovy_string(value):
+ return "'" + str(value).replace("\\", "\\\\").replace("'", "\\'") + "'"
+
+
+def create_fixture(root, junit_classpath):
+ (root / "settings.gradle").write_text("rootProject.name =
'comet-iceberg-shard-fixture'\n")
+ if junit_classpath:
+ jars = [str(Path(p).resolve()) for p in
junit_classpath.split(os.pathsep)]
+ if not all(Path(p).is_file() for p in jars):
+ raise ValueError("Every --junit-classpath entry must be an
existing jar")
+ dependencies = "testImplementation files(" + ",
".join(map(groovy_string, jars)) + ")"
+ else:
+ dependencies = """
+ testImplementation 'org.junit.jupiter:junit-jupiter:5.11.4'
+ testRuntimeOnly 'org.junit.platform:junit-platform-launcher:1.11.4'
+ """
+ (root / "build.gradle").write_text("""
+ import groovy.json.JsonOutput
+ plugins { id 'java' }
+ repositories { mavenCentral() }
+ dependencies { DEPENDENCIES }
+ test {
+ useJUnitPlatform { excludeTags 'excluded-tag' }
+ include '**/Test*.class'
+ exclude '**/TestExcludedByPattern.class', '**/TestOtherTask.class'
+ exclude { it.name == 'TestExcludedBySpec.class' }
+ systemProperty 'breakShardFixture',
project.findProperty('breakShardFixture') ?: 'false'
+ doFirst {
+ def candidates = new TreeSet<String>()
+ candidateClassFiles.visit { entry ->
+ if (!entry.directory && entry.name.endsWith('.class')) {
+ candidates.add(entry.relativePath.pathString)
+ }
+ }
+ file('build/candidates.json').text = JsonOutput.toJson(candidates)
+ }
+ }
+ tasks.register('otherTest', Test) {
+ testClassesDirs = sourceSets.test.output.classesDirs
+ classpath = sourceSets.test.runtimeClasspath
+ useJUnitPlatform()
+ include '**/TestOtherTask.class'
+ }
+ """.replace("DEPENDENCIES", dependencies))
+ source_dir = root / "src/test/java/fixture"
+ source_dir.mkdir(parents=True)
+ sources = {
+ "TestStructuredStreamingRead3": """
+ @org.junit.jupiter.params.ParameterizedTest
+ @org.junit.jupiter.params.provider.ValueSource(ints = {1, 2, 3})
+ void parameterized(int value) {
org.junit.jupiter.api.Assertions.assertTrue(value > 0); }
+ @org.junit.jupiter.api.Nested class Nested {
+ @org.junit.jupiter.api.Test void nested() {}
+ }
+ """,
+ "TestAlpha": "@org.junit.jupiter.api.Test void alpha() {}",
+ "TestBeta": "@org.junit.jupiter.api.Test void beta() {}",
+ "TestDelta": "@org.junit.jupiter.api.Test void delta() {}",
+ "TestNewlyAdded": "@org.junit.jupiter.api.Test void
automaticallyDiscovered() {}",
+ "TestInherited": "",
+ "TestDynamic": """
+ @org.junit.jupiter.api.TestFactory
java.util.stream.Stream<org.junit.jupiter.api.DynamicTest> generated() {
+ return java.util.stream.Stream.of("first", "second").map(name ->
+ org.junit.jupiter.api.DynamicTest.dynamicTest(name, () -> {}));
+ }
+ """,
+ "TestFailurePropagation": """
+ @org.junit.jupiter.api.Test void failurePropagates() {
+
org.junit.jupiter.api.Assertions.assertFalse(Boolean.getBoolean("breakShardFixture"));
+ }
+ """,
+ "TestOtherTask": "@org.junit.jupiter.api.Test void unaffectedTask()
{}",
+ "TestExcludedByPattern": "@org.junit.jupiter.api.Test void excluded()
{ throw new AssertionError(); }",
+ "TestExcludedBySpec": "@org.junit.jupiter.api.Test void excluded() {
throw new AssertionError(); }",
+ "NotIncluded": "@org.junit.jupiter.api.Test void excluded() { throw
new AssertionError(); }",
+ "TestExcludedByTag": """
+ @org.junit.jupiter.api.Tag("excluded-tag")
+ @org.junit.jupiter.api.Test void excluded() { throw new
AssertionError(); }
+ """,
+ }
+ for name, body in sources.items():
+ superclass = " extends FixtureBase" if name == "TestInherited" else ""
+ (source_dir / f"{name}.java").write_text(
+ f"package fixture;\npublic class {name}{superclass}
{{\n{body}\n}}\n")
+ (source_dir / "FixtureBase.java").write_text("""
+ package fixture;
+ abstract class FixtureBase {
+ @org.junit.jupiter.api.Test void inherited() {}
+ }
+ """)
+
+
+def read_cases(root, task):
+ cases = Counter()
+ for report in (root / "build/test-results" / task).glob("TEST-*.xml"):
+ for case in ET.parse(report).getroot().iter("testcase"):
+ state = "failed" if case.find("failure") is not None else "passed"
+ if case.find("skipped") is not None:
+ state = "skipped"
+ cases[case.attrib["classname"], case.attrib["name"], state] += 1
+ return cases
+
+
+def check(root, gradle, junit_classpath):
+ create_fixture(root, junit_classpath)
+ results = root / "results"
+ results.mkdir()
+ common = [gradle, "--project-dir", str(root), "--console=plain",
"--no-daemon",
+ "-Dorg.gradle.jvmargs=-Xmx256m", "--max-workers=2"]
+ if junit_classpath:
+ common.append("--offline")
+
+ def run(label, extra=(), task="test", expect_failure=False):
+ # Never let an UP-TO-DATE or NO-SOURCE task reuse another run's
reports.
+ reports = root / "build/test-results" / task
+ if reports.exists():
+ shutil.rmtree(reports)
+ if task == "test":
+ (root / "build/candidates.json").unlink(missing_ok=True)
+ completed = subprocess.run(common + [task] + list(extra), text=True,
+ stdout=subprocess.PIPE,
stderr=subprocess.STDOUT)
+ (results / f"{label}.log").write_text(completed.stdout)
+ if expect_failure:
+ if completed.returncode == 0:
+ raise AssertionError(f"{label}: expected Gradle to fail")
+ elif completed.returncode:
+ raise AssertionError(f"{label}
failed:\n{completed.stdout[-8000:]}")
+ print(f"{label}: {'expected failure' if expect_failure else
'passed'}", flush=True)
+ return read_cases(root, task)
+
+ def shard_args(index, count=SHARD_COUNT, task=":test"):
+ return ["--init-script", str(INIT_SCRIPT.resolve()),
f"-PcometShardTask={task}",
+ f"-PcometShardIndex={index}", f"-PcometShardCount={count}"]
+
+ baseline = run("baseline")
+ inventory = set(json.loads((root / "build/candidates.json").read_text()))
+ assert sum(baseline.values()) == 12, baseline
+ assert all(state == "passed" for _, _, state in baseline), baseline
+ assert any(cls == "fixture.TestNewlyAdded" for cls, _, _ in baseline)
+ assert any("$Nested" in cls for cls, _, _ in baseline)
+ assert all("Excluded" not in cls and "NotIncluded" not in cls for cls, _,
_ in baseline)
+
+ combined_cases = Counter()
+ combined_candidates = Counter()
+ failure_owner = None
+ for index in range(1, SHARD_COUNT + 1):
+ cases = run(f"shard-{index}", shard_args(index))
+ manifest = json.loads((root /
f"build/comet-shards/test-{index}.json").read_text())
+ candidates = json.loads((root / "build/candidates.json").read_text())
+ assert candidates == manifest["candidates"]
+ assert set(manifest["unshardedCandidates"]) == inventory
+ assert cases, f"empty fixture shard {index}"
+ combined_cases.update(cases)
+ combined_candidates.update(candidates)
+ if any(cls == "fixture.TestFailurePropagation" for cls, _, _ in cases):
+ failure_owner = index
+ (results / f"shard-{index}.json").write_text(json.dumps(manifest,
indent=2) + "\n")
+ assert set(combined_candidates) == inventory, (set(combined_candidates),
inventory)
+ assert set(combined_candidates.values()) == {1}, combined_candidates
+ assert combined_cases == baseline, (combined_cases, baseline)
+ verify_manifests(results, ":test")
+ assert run("single-shard", shard_args(1, count=1)) == baseline
+
+ other_baseline = run("other-baseline", task="otherTest")
+ assert sum(other_baseline.values()) == 1
+ assert run("other-unmodified", shard_args(1), task="otherTest") ==
other_baseline
+
+ failed = run("failure-propagation", shard_args(failure_owner) +
["-PbreakShardFixture=true"],
+ expect_failure=True)
+ assert any(cls == "fixture.TestFailurePropagation" and state == "failed"
+ for cls, _, state in failed), failed
+ run("invalid-index", shard_args(0), task="help", expect_failure=True)
+ run("unknown-task", shard_args(1, task=":missing"), task="help",
expect_failure=True)
+ print(f"Verified {len(inventory)} candidate classes and
{sum(baseline.values())} test cases: "
+ "the four shards equal the unsharded inventory exactly once.",
flush=True)
+
+
+def main():
+ parser = argparse.ArgumentParser(description=__doc__)
+ parser.add_argument("--gradle", default="gradle")
+ parser.add_argument("--junit-classpath")
+ parser.add_argument("--work-dir", type=Path)
+ mode = parser.add_mutually_exclusive_group()
+ mode.add_argument("--github-output", type=Path)
+ mode.add_argument("--manifests", type=Path)
+ parser.add_argument("--task", help="Expected Gradle task path when
checking real manifests")
+ args = parser.parse_args()
+ if args.github_output:
+ write_workflow_matrix(args.github_output)
+ return
+ if args.manifests:
+ if not args.task:
+ parser.error("--manifests requires --task")
+ verify_manifests(args.manifests, args.task)
+ return
+ gradle = shutil.which(args.gradle)
+ if not gradle:
+ parser.error(f"Gradle executable not found: {args.gradle}")
+ if args.work_dir:
+ args.work_dir.mkdir(parents=True, exist_ok=False)
+ check(args.work_dir.resolve(), gradle, args.junit_classpath)
+ else:
+ with tempfile.TemporaryDirectory(prefix="comet-iceberg-shards-") as
tmp:
+ check(Path(tmp), gradle, args.junit_classpath)
+
+
+if __name__ == "__main__":
+ main()
diff --git a/dev/ci/compute-changes.py b/dev/ci/compute-changes.py
index 9b7cd2f169..c385fb8412 100644
--- a/dev/ci/compute-changes.py
+++ b/dev/ci/compute-changes.py
@@ -193,6 +193,9 @@ FILTERS = {
".github/workflows/iceberg_spark_test_reusable.yml",
".github/actions/setup-builder/**",
".github/actions/setup-iceberg-builder/**",
+ "dev/ci/iceberg-test-shards.gradle",
+ "dev/ci/check-iceberg-shards.py",
+ "dev/ci/test-iceberg-shards.py",
],
"iceberg_1_9": [
"native/**/src/**",
@@ -210,6 +213,9 @@ FILTERS = {
".github/workflows/iceberg_spark_test_reusable.yml",
".github/actions/setup-builder/**",
".github/actions/setup-iceberg-builder/**",
+ "dev/ci/iceberg-test-shards.gradle",
+ "dev/ci/check-iceberg-shards.py",
+ "dev/ci/test-iceberg-shards.py",
],
"iceberg_1_10": [
"native/**/src/**",
@@ -227,6 +233,9 @@ FILTERS = {
".github/workflows/iceberg_spark_test_reusable.yml",
".github/actions/setup-builder/**",
".github/actions/setup-iceberg-builder/**",
+ "dev/ci/iceberg-test-shards.gradle",
+ "dev/ci/check-iceberg-shards.py",
+ "dev/ci/test-iceberg-shards.py",
],
"iceberg_1_11": [
"native/**/src/**",
@@ -244,6 +253,9 @@ FILTERS = {
".github/workflows/iceberg_spark_test_reusable.yml",
".github/actions/setup-builder/**",
".github/actions/setup-iceberg-builder/**",
+ "dev/ci/iceberg-test-shards.gradle",
+ "dev/ci/check-iceberg-shards.py",
+ "dev/ci/test-iceberg-shards.py",
],
}
diff --git a/dev/ci/iceberg-test-shards.gradle
b/dev/ci/iceberg-test-shards.gradle
new file mode 100644
index 0000000000..e8a39a073a
--- /dev/null
+++ b/dev/ci/iceberg-test-shards.gradle
@@ -0,0 +1,110 @@
+/*
+ * 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 groovy.json.JsonOutput
+import org.gradle.api.file.FileTreeElement
+import org.gradle.api.tasks.testing.Test
+
+// An exclusion predicate partitions Gradle's existing candidates without
broadening
+// its includes, replacing its exclusions, or changing JUnit's
discovery/parameters.
+// All classes remain on the test classpath, including inherited test fixtures.
+class CometIcebergTestShard {
+ static int owner(String classFileName, int count) {
+ if (count == 1) {
+ return 1
+ }
+ // Keep JUnit @Nested classes with their enclosing class.
+ String outer = classFileName.substring(0, classFileName.length() -
'.class'.length())
+ .split('\\$', 2)[0]
+ // This family exists in all supported Iceberg versions. In the 1.11 run
used
+ // to size these shards it took ~19 of ~67 test minutes. The remaining
classes
+ // hash to ~18.5 / 16.4 / 13.2 minutes; new classes need no allowlist
update.
+ if (outer.startsWith('TestStructuredStreamingRead')) {
+ return 1
+ }
+ return 2 + Math.floorMod(outer.hashCode(), count - 1)
+ }
+}
+
+def properties = gradle.startParameter.projectProperties
+def required = ['cometShardTask', 'cometShardIndex', 'cometShardCount']
+if (required.any { !properties[it] }) {
+ throw new GradleException("Iceberg sharding requires ${required.join(', ')}")
+}
+int shardIndex
+int shardCount
+try {
+ shardIndex = Integer.parseInt(properties.cometShardIndex)
+ shardCount = Integer.parseInt(properties.cometShardCount)
+} catch (NumberFormatException e) {
+ throw new GradleException('Iceberg shard index/count must be integers', e)
+}
+if (shardCount < 1 || shardIndex < 1 || shardIndex > shardCount) {
+ throw new GradleException('Iceberg shard index must be between 1 and
cometShardCount')
+}
+
+gradle.projectsEvaluated {
+ def testTask = gradle.rootProject.tasks.findByPath(properties.cometShardTask)
+ if (!(testTask instanceof Test)) {
+ throw new GradleException("Not a Test task: ${properties.cometShardTask}")
+ }
+ testTask.inputs.property('cometShardIndex', shardIndex)
+ testTask.inputs.property('cometShardCount', shardCount)
+ int runAttempt = Integer.parseInt(System.getenv('GITHUB_RUN_ATTEMPT') ?: '1')
+ testTask.inputs.property('cometShardRunAttempt', runAttempt)
+
testTask.inputs.files(gradle.startParameter.initScripts).withPropertyName('cometShardScripts')
+ boolean applyShardFilter = true
+ testTask.exclude { FileTreeElement entry ->
+ applyShardFilter && !entry.directory && entry.name.endsWith('.class') &&
+ CometIcebergTestShard.owner(entry.name, shardCount) != shardIndex
+ }
+
+ // Keep a machine-readable inventory next to the normal JUnit reports. This
is
+ // Gradle's candidate set, not the number of tests that JUnit actually
executes.
+ def manifest = testTask.project.layout.buildDirectory
+ .file("comet-shards/${testTask.name}-${shardIndex}.json").get().asFile
+ testTask.outputs.file(manifest)
+ testTask.doFirst {
+ def inventory = {
+ def candidates = new TreeSet<String>()
+ testTask.candidateClassFiles.visit { entry ->
+ if (!entry.directory && entry.name.endsWith('.class')) {
+ candidates.add(entry.relativePath.pathString)
+ }
+ }
+ candidates
+ }
+ // Compilation has completed by this point. Disable only our predicate
while
+ // recording the baseline, preserving even upstream exclusion closures.
Then
+ // restore it before collecting the selected candidates and executing
tests.
+ def unshardedCandidates
+ applyShardFilter = false
+ try {
+ unshardedCandidates = inventory()
+ } finally {
+ applyShardFilter = true
+ }
+ def candidates = inventory()
+ manifest.parentFile.mkdirs()
+ manifest.text = JsonOutput.prettyPrint(JsonOutput.toJson([
+ task: testTask.path, shard: shardIndex, count: shardCount, attempt:
runAttempt,
+ candidates: candidates, unshardedCandidates: unshardedCandidates])) +
'\n'
+ testTask.logger.lifecycle("Iceberg shard ${shardIndex}/${shardCount}:
${candidates.size()} candidate classes")
+ }
+}
diff --git a/dev/ci/test-iceberg-shards.py b/dev/ci/test-iceberg-shards.py
new file mode 100644
index 0000000000..084facb622
--- /dev/null
+++ b/dev/ci/test-iceberg-shards.py
@@ -0,0 +1,144 @@
+#!/usr/bin/env python3
+#
+# 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.
+
+"""Fast regression tests for the real Iceberg inventory and workflow-matrix
guard."""
+
+import importlib.util
+import json
+from pathlib import Path
+import tempfile
+import unittest
+
+
+SPEC = importlib.util.spec_from_file_location(
+ "check_iceberg_shards",
Path(__file__).with_name("check-iceberg-shards.py"))
+CHECK = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(CHECK)
+
+
+class IcebergShardManifestTest(unittest.TestCase):
+ def setUp(self):
+ self.temp =
tempfile.TemporaryDirectory(prefix="comet-shard-manifest-test-")
+ self.addCleanup(self.temp.cleanup)
+ self.root = Path(self.temp.name)
+ self.candidates = [f"fixture/Test{index}.class"
+ for index in range(1, CHECK.SHARD_COUNT + 1)]
+ for index in range(1, CHECK.SHARD_COUNT + 1):
+ self.write(index)
+
+ def write(self, index, attempt=1, **overrides):
+ manifest = dict(task=":test", count=CHECK.SHARD_COUNT, shard=index,
attempt=attempt,
+ candidates=[self.candidates[index - 1]],
+ unshardedCandidates=self.candidates)
+ manifest.update(overrides)
+ path = self.root / f"shard-{index}-attempt-{attempt}.json"
+ path.write_text(json.dumps(manifest))
+ return path
+
+ def test_complete_inventory(self):
+ CHECK.verify_manifests(self.root, ":test")
+
+ def test_matrix_and_partition_count_share_one_definition(self):
+ path = self.root / "github-output"
+ CHECK.write_workflow_matrix(path)
+ values = dict(line.split("=", 1) for line in
path.read_text().splitlines())
+ matrix = json.loads(values["matrix"])
+ indices = matrix["shard"]
+ self.assertEqual(indices, list(range(1, int(values["count"]) + 1)))
+ self.assertEqual(len(indices), len(set(indices)))
+ # A second dimension changes job-total but must not change shard count.
+ matrix["environment"] = ["first", "second"]
+ self.assertEqual(int(values["count"]), len(matrix["shard"]))
+
+ def test_missing_shard_fails(self):
+ (self.root / "shard-1-attempt-1.json").unlink()
+ with self.assertRaisesRegex(ValueError, "Expected shard indices"):
+ CHECK.verify_manifests(self.root, ":test")
+
+ def test_unexpected_shard_fails(self):
+ self.write(1, shard=CHECK.SHARD_COUNT + 1)
+ with self.assertRaisesRegex(ValueError, "Expected shard indices"):
+ CHECK.verify_manifests(self.root, ":test")
+
+ def test_overlapping_shards_fail(self):
+ self.write(2, candidates=self.candidates[:2])
+ with self.assertRaisesRegex(ValueError, "multiple shards"):
+ CHECK.verify_manifests(self.root, ":test")
+
+ def test_missing_candidate_fails(self):
+ self.write(1, candidates=[])
+ with self.assertRaisesRegex(ValueError, "coverage mismatch"):
+ CHECK.verify_manifests(self.root, ":test")
+
+ def test_extra_candidate_fails(self):
+ self.write(1, candidates=[self.candidates[0],
"fixture/Unexpected.class"])
+ with self.assertRaisesRegex(ValueError, "coverage mismatch"):
+ CHECK.verify_manifests(self.root, ":test")
+
+ def test_inconsistent_baselines_fail(self):
+ self.write(1, unshardedCandidates=self.candidates[:1])
+ with self.assertRaisesRegex(ValueError, "inventories disagree"):
+ CHECK.verify_manifests(self.root, ":test")
+
+ def test_wrong_count_or_task_fails(self):
+ for overrides in ({"count": CHECK.SHARD_COUNT * 2}, {"task":
":otherTest"}):
+ with self.subTest(overrides=overrides):
+ self.write(1, **overrides)
+ with self.assertRaisesRegex(ValueError, "mismatched task or
shard count"):
+ CHECK.verify_manifests(self.root, ":test")
+
+ def test_duplicate_candidate_in_one_manifest_fails(self):
+ self.write(1, candidates=[self.candidates[0]] * 2)
+ with self.assertRaisesRegex(ValueError, "duplicate class paths"):
+ CHECK.verify_manifests(self.root, ":test")
+
+ def test_duplicate_same_attempt_fails(self):
+ contents = (self.root / "shard-1-attempt-1.json").read_text()
+ (self.root / "duplicate.json").write_text(contents)
+ with self.assertRaisesRegex(ValueError, "Duplicate inventory"):
+ CHECK.verify_manifests(self.root, ":test")
+
+ def test_failed_job_rerun_uses_latest_attempt_per_shard(self):
+ # The other shards need not rerun when just shard 1 is retried.
+ self.write(1, candidates=[])
+ self.write(1, attempt=2)
+ CHECK.verify_manifests(self.root, ":test")
+
+ def test_newer_bad_inventory_cannot_be_hidden_by_old_attempt(self):
+ self.write(1, attempt=2, candidates=[])
+ with self.assertRaisesRegex(ValueError, "coverage mismatch"):
+ CHECK.verify_manifests(self.root, ":test")
+
+ def test_empty_inventory_directory_fails(self):
+ with self.assertRaisesRegex(ValueError, "Expected shard indices"):
+ CHECK.verify_manifests(self.root / "missing", ":test")
+
+ def test_malformed_metadata_fails(self):
+ for overrides in ({"attempt": 0}, {"shard": True}, {"count": "4"}):
+ with self.subTest(overrides=overrides):
+ path = self.write(1)
+ manifest = json.loads(path.read_text())
+ manifest.update(overrides)
+ path.write_text(json.dumps(manifest))
+ with self.assertRaisesRegex(ValueError, "positive integer"):
+ CHECK.verify_manifests(self.root, ":test")
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/docs/source/contributor-guide/iceberg-spark-tests.md
b/docs/source/contributor-guide/iceberg-spark-tests.md
index df6c168af6..4b0e827333 100644
--- a/docs/source/contributor-guide/iceberg-spark-tests.md
+++ b/docs/source/contributor-guide/iceberg-spark-tests.md
@@ -105,6 +105,28 @@ run against Spark 3.5.9 with Java 17; Iceberg 1.11.0 runs
against Spark 4.1.3 wi
(1.8, 1.9, 1.10) run only on pushes to main, or on a pull request labeled
`run-iceberg-tests`. All caller
workflows delegate to `iceberg_spark_test_reusable.yml`, which holds the build
and test job logic.
+The core Spark test target runs in four independent workers. The workflow
passes
+`dev/ci/iceberg-test-shards.gradle` as a Gradle init script: one worker runs
the long
+`TestStructuredStreamingRead` family, and the others hash the remaining class
names into three
+buckets. New tests are assigned automatically. Nested classes and all
parameterized cases stay
+with their enclosing class; Gradle's existing includes, exclusions, and JUnit
configuration are
+unchanged. The extensions and shaded-runtime targets remain unsharded.
+
+The matrix and partition count come from the same definition in
`dev/ci/check-iceberg-shards.py`;
+adding another matrix dimension does not change the partition count. Each
worker records its
+unsharded candidate set with only the Comet shard predicate disabled, then
restores the predicate
+before recording its selected set and executing tests. Both inventories and
the JUnit XML reports
+are uploaded. A dependent coverage job requires all shard indices, matching
unsharded inventories,
+and selected sets whose disjoint union equals that inventory. It downloads
only artifacts for the
+same Iceberg/Spark/Scala/JDK configuration in the current workflow run and
uses the latest available
+attempt per shard, so rerunning only failed jobs can reuse earlier successful
shards' inventories.
+
+These candidate inventories include classes that JUnit may not execute, so the
runtime job also
+runs `dev/ci/check-iceberg-shards.py`, a small Gradle/JUnit fixture that
checks the four shards'
+combined candidate classes and executed test cases equal an unsharded run
exactly once. It also
+checks nested, parameterized, inherited, and dynamically generated tests,
existing exclusions,
+and failure propagation. The fixture does not compile Spark or Iceberg.
+
Apply the `run-iceberg-tests` label to a pull request whenever it touches
reflection code
(`org.apache.comet.iceberg.IcebergReflection`) or other logic whose behavior
can differ across Iceberg
versions, since Iceberg 1.11 alone will not catch a regression that only
affects 1.8, 1.9, or 1.10.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
