diff --git a/.github/actions/java-test/action.yaml b/.github/actions/java-test/action.yaml
index 1af66f7019d..5a9cae3808e 100644
--- a/.github/actions/java-test/action.yaml
+++ b/.github/actions/java-test/action.yaml
@@ -76,45 +76,8 @@ runs:
         restore-keys: |
           ${{ runner.os }}-java-maven-
 
-    # Maven itself is stored outside the dependency repository. Keep its cache
-    # independent of pom.xml changes and preserve the macOS cache workaround.
-    - name: Restore Maven distribution
-      id: maven-distribution
-      if: ${{ runner.os != 'macOS' }}
-      uses: actions/cache/restore@v5
-      with:
-        path: |
-          ~/.m2/wrapper/dists
-          /root/.m2/wrapper/dists
-        key: ${{ runner.os }}-${{ runner.arch }}-maven-wrapper-${{ hashFiles('.mvn/wrapper/maven-wrapper.properties') }}
-
-    # Retry only the wrapper download, never compilation or test execution.
-    # Delays use exponential backoff (10s, 20s, 40s) plus 0-4s of random jitter.
-    - name: Bootstrap Maven
-      shell: bash
-      run: |
-        for attempt in 1 2 3 4; do
-          if ./mvnw -B --version; then
-            break
-          fi
-          if [ "$attempt" -eq 4 ]; then
-            echo "::error::Maven bootstrap failed after $attempt attempts; tests were not started."
-            exit 1
-          fi
-          delay=$((10 * (1 << (attempt - 1)) + RANDOM % 5))
-          echo "::warning::Maven bootstrap attempt $attempt failed; retrying in ${delay}s."
-          sleep "$delay"
-        done
-
-    # Save a successful bootstrap even when the subsequent tests fail.
-    - name: Save Maven distribution
-      if: ${{ runner.os != 'macOS' && steps.maven-distribution.outputs.cache-hit != 'true' }}
-      uses: actions/cache/save@v5
-      with:
-        path: |
-          ~/.m2/wrapper/dists
-          /root/.m2/wrapper/dists
-        key: ${{ steps.maven-distribution.outputs.cache-primary-key }}
+    - name: Setup Maven
+      uses: ./.github/actions/setup-maven
 
     - name: Run all tests
       shell: bash
@@ -123,7 +86,7 @@ runs:
         SPARK_LOCAL_HOSTNAME: "localhost"
         SPARK_LOCAL_IP: "127.0.0.1"
       run: |
-        MAVEN_OPTS="-Xmx4G -Xms2G -XX:+UnlockDiagnosticVMOptions -XX:+ShowMessageBoxOnError -XX:+HeapDumpOnOutOfMemoryError -XX:ErrorFile=./hs_err_pid%p.log" SPARK_HOME=`pwd` ./mvnw -B -Prelease install ${{ inputs.maven_opts }}
+        MAVEN_OPTS="${MAVEN_OPTS:-} -Xmx4G -Xms2G -XX:+UnlockDiagnosticVMOptions -XX:+ShowMessageBoxOnError -XX:+HeapDumpOnOutOfMemoryError -XX:ErrorFile=./hs_err_pid%p.log" SPARK_HOME=`pwd` ./mvnw -B -Prelease install ${{ inputs.maven_opts }}
     - name: Run specified tests
       shell: bash
       if: ${{ inputs.suites != '' }}
@@ -133,7 +96,7 @@ runs:
       run: |
         MAVEN_SUITES="$(echo "${{ inputs.suites }}" | paste -sd, -)"
         echo "Running with MAVEN_SUITES=$MAVEN_SUITES"
-        MAVEN_OPTS="-Xmx4G -Xms2G -DwildcardSuites=$MAVEN_SUITES -XX:+UnlockDiagnosticVMOptions -XX:+ShowMessageBoxOnError -XX:+HeapDumpOnOutOfMemoryError -XX:ErrorFile=./hs_err_pid%p.log" SPARK_HOME=`pwd` ./mvnw -B -Prelease install ${{ inputs.maven_opts }}
+        MAVEN_OPTS="${MAVEN_OPTS:-} -Xmx4G -Xms2G -DwildcardSuites=$MAVEN_SUITES -XX:+UnlockDiagnosticVMOptions -XX:+ShowMessageBoxOnError -XX:+HeapDumpOnOutOfMemoryError -XX:ErrorFile=./hs_err_pid%p.log" SPARK_HOME=`pwd` ./mvnw -B -Prelease install ${{ inputs.maven_opts }}
     - name: Upload crash logs
       if: failure()
       uses: actions/upload-artifact@v6
diff --git a/.github/actions/rust-test/action.yaml b/.github/actions/rust-test/action.yaml
index c39c2dcd4f9..49d6680d6d8 100644
--- a/.github/actions/rust-test/action.yaml
+++ b/.github/actions/rust-test/action.yaml
@@ -51,6 +51,9 @@ runs:
         restore-keys: |
           ${{ runner.os }}-rust-maven-
 
+    - name: Setup Maven
+      uses: ./.github/actions/setup-maven
+
     - name: Build common module (pre-requisite for Rust tests)
       shell: bash
       run: |
diff --git a/.github/actions/setup-maven/action.yaml b/.github/actions/setup-maven/action.yaml
new file mode 100644
index 00000000000..51cf015c556
--- /dev/null
+++ b/.github/actions/setup-maven/action.yaml
@@ -0,0 +1,85 @@
+# 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: Setup Maven
+description: 'Restore and bootstrap the Maven wrapper before building or testing'
+runs:
+  using: "composite"
+  steps:
+    # Maven 3.9.6 uses Resolver's native HTTP transport. Retry individual
+    # transfers rather than rerunning Maven goals (and potentially their tests).
+    # MAVEN_ARGS is not read by our wrapper; carry these JVM properties through
+    # MAVEN_OPTS, including callers that also configure their heap there.
+    - name: Configure Maven transfer retries
+      shell: bash
+      run: |
+        retry_options="-Daether.connector.http.retryHandler.count=3"
+        retry_options+=" -Daether.connector.http.retryHandler.interval=5000"
+        retry_options+=" -Daether.connector.http.retryHandler.intervalMax=60000"
+        retry_options+=" -Daether.connector.http.retryHandler.serviceUnavailable=429,500,502,503,504"
+        # JVM options are whitespace-separated. Normalize multiline YAML env
+        # values before writing GitHub's single-line environment-file format.
+        maven_options="${MAVEN_OPTS:-}"
+        maven_options="${maven_options//$'\r'/ }"
+        maven_options="${maven_options//$'\n'/ }"
+        printf 'MAVEN_OPTS=%s %s\n' "$maven_options" "$retry_options" >> "$GITHUB_ENV"
+
+    # Maven itself is stored outside the dependency repository. Keep its cache
+    # independent of pom.xml changes and preserve the macOS cache workaround.
+    # Globbing /root on a host runner aborts cache save with EACCES, so select
+    # only the directory this JVM's wrapper actually uses.
+    - name: Locate Maven distribution cache
+      id: maven-cache-path
+      if: ${{ runner.os != 'macOS' }}
+      shell: bash
+      run: |
+        cache_path=$(bash "$GITHUB_ACTION_PATH/cache-path.sh")
+        printf 'path=%s\n' "$cache_path" >> "$GITHUB_OUTPUT"
+
+    - name: Restore Maven distribution
+      id: maven-distribution
+      if: ${{ runner.os != 'macOS' }}
+      uses: actions/cache/restore@v5
+      with:
+        path: ${{ steps.maven-cache-path.outputs.path }}
+        key: ${{ runner.os }}-${{ runner.arch }}-maven-wrapper-${{ hashFiles('.mvn/wrapper/maven-wrapper.properties') }}
+
+    # Retry only the wrapper download, never compilation or test execution.
+    # Delays use exponential backoff (10s, 20s, 40s) plus 0-4s of random jitter.
+    - name: Bootstrap Maven
+      shell: bash
+      run: |
+        for attempt in 1 2 3 4; do
+          if ./mvnw -B --version; then
+            break
+          fi
+          if [ "$attempt" -eq 4 ]; then
+            echo "::error::Maven bootstrap failed after $attempt attempts; tests were not started."
+            exit 1
+          fi
+          delay=$((10 * (1 << (attempt - 1)) + RANDOM % 5))
+          echo "::warning::Maven bootstrap attempt $attempt failed; retrying in ${delay}s."
+          sleep "$delay"
+        done
+
+    # Save a successful bootstrap even when the subsequent tests fail.
+    - name: Save Maven distribution
+      if: ${{ runner.os != 'macOS' && steps.maven-distribution.outputs.cache-hit != 'true' }}
+      uses: actions/cache/save@v5
+      with:
+        path: ${{ steps.maven-cache-path.outputs.path }}
+        key: ${{ steps.maven-distribution.outputs.cache-primary-key }}
diff --git a/.github/actions/setup-maven/cache-path.sh b/.github/actions/setup-maven/cache-path.sh
new file mode 100755
index 00000000000..42bc77f9bda
--- /dev/null
+++ b/.github/actions/setup-maven/cache-path.sh
@@ -0,0 +1,50 @@
+#!/usr/bin/env bash
+#
+# 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.
+
+set -euo pipefail
+
+# Match mvnw's JVM and option order. In job containers, shell HOME is
+# /github/home while the JVM's user.home (and the wrapper cache) is /root.
+java_command="${JAVACMD:-${JAVA_HOME:+$JAVA_HOME/bin/}java}"
+java_options="${MAVEN_OPTS:-}"
+if [ -f .mvn/jvm.config ]; then
+  java_options="$(tr '\n' ' ' < .mvn/jvm.config) $java_options"
+fi
+java_options="${java_options//$'\r'/ }"
+java_options="${java_options//$'\n'/ }"
+read -r -a java_arguments <<< "$java_options"
+properties=$("$java_command" "${java_arguments[@]}" -XshowSettings:properties -version 2>&1) || {
+  status=$?
+  printf '%s\n' "$properties" >&2
+  exit "$status"
+}
+
+# Wrapper 3.2.0 prefers the JVM property, then the environment override,
+# then user.home/.m2. Both its distribution and ZIP paths use wrapper/dists.
+maven_user_home=$(sed -n 's/^[[:space:]]*maven\.user\.home = //p' <<< "$properties")
+maven_user_home="${maven_user_home:-${MAVEN_USER_HOME:-}}"
+if [ -z "$maven_user_home" ]; then
+  jvm_user_home=$(sed -n 's/^[[:space:]]*user\.home = //p' <<< "$properties")
+  if [ -z "$jvm_user_home" ]; then
+    echo "Could not determine the Maven wrapper cache directory from JVM properties." >&2
+    exit 1
+  fi
+  maven_user_home="$jvm_user_home/.m2"
+fi
+printf '%s/wrapper/dists\n' "${maven_user_home%/}"
diff --git a/.github/actions/setup-spark-builder/action.yaml b/.github/actions/setup-spark-builder/action.yaml
index 84804c0a798..582974e986a 100644
--- a/.github/actions/setup-spark-builder/action.yaml
+++ b/.github/actions/setup-spark-builder/action.yaml
@@ -32,6 +32,10 @@ inputs:
     description: 'Skip cloning Spark and applying patches (when apache-spark/ is pre-staged from a build-jvm artifact)'
     required: false
     default: 'false'
+  sbt-projects:
+    description: 'Space-separated Spark projects whose test dependencies are needed'
+    required: false
+    default: 'catalyst sql hive'
 runs:
   using: "composite"
   steps:
@@ -61,6 +65,9 @@ runs:
         restore-keys: |
           ${{ runner.os }}-spark-sql-
 
+    - name: Setup Maven
+      uses: ./.github/actions/setup-maven
+
     - name: Build Comet (with native)
       if: ${{ inputs.skip-native-build != 'true' }}
       shell: bash
@@ -93,3 +100,52 @@ runs:
               "$(dirname "$pom")/_remote.repositories"
           done
         done
+
+    - name: Remove incomplete Parquet test dependencies
+      shell: bash
+      run: |
+        # Maven installs Parquet POMs and main JARs, but not all test classifiers.
+        # Coursier treats those entries as local and will not fetch the missing
+        # tests.jar remotely. Do this before dependency resolution, not tests.
+        rm -rf "$HOME/.m2/repository/org/apache/parquet" /root/.m2/repository/org/apache/parquet
+
+    - name: Restore SBT dependency cache
+      id: sbt-dependencies
+      uses: actions/cache/restore@v6
+      with:
+        path: |
+          ~/.cache/coursier
+          /root/.cache/coursier
+          ~/.sbt/boot
+          /root/.sbt/boot
+          ~/.ivy2/cache
+          /root/.ivy2/cache
+        key: ${{ runner.os }}-${{ runner.arch }}-spark-sbt-v1-${{ inputs.spark-version }}-${{ hashFiles('**/pom.xml', '!**/target/**', 'apache-spark/project/build.properties', 'apache-spark/project/*.sbt', 'apache-spark/project/*.scala') }}
+        restore-keys: |
+          ${{ runner.os }}-${{ runner.arch }}-spark-sbt-v1-${{ inputs.spark-version }}-
+
+    - name: Resolve Spark dependencies
+      shell: bash
+      env:
+        SBT_PROJECTS: ${{ inputs.sbt-projects }}
+      run: |
+        cd apache-spark
+        read -r -a projects <<< "$SBT_PROJECTS"
+        ../dev/ci/resolve-spark-dependencies.sh "${projects[@]}"
+
+    # Keep a successful download even if a later compilation or test fails.
+    # Only the build job, which resolves all three projects, may publish this
+    # shared immutable key. A writer-only or shard job must not fill it with
+    # partial coverage and prevent the full build from saving its downloads.
+    - name: Save SBT dependency cache
+      if: inputs.sbt-projects == 'catalyst sql hive' && steps.sbt-dependencies.outputs.cache-hit != 'true'
+      uses: actions/cache/save@v6
+      with:
+        path: |
+          ~/.cache/coursier
+          /root/.cache/coursier
+          ~/.sbt/boot
+          /root/.sbt/boot
+          ~/.ivy2/cache
+          /root/.ivy2/cache
+        key: ${{ steps.sbt-dependencies.outputs.cache-primary-key }}
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 5243c4aaec4..9f360c67512 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -68,6 +68,9 @@ jobs:
           distribution: temurin
           java-version: 11
 
+      - name: Setup Maven
+        uses: ./.github/actions/setup-maven
+
       - name: Apache RAT license check
         run: ./mvnw -B -N apache-rat:check
 
@@ -88,6 +91,13 @@ jobs:
       - name: Check micro benchmark runner
         run: python3 dev/ci/check-benchmark-runner.py
 
+      - name: Check CI download handling
+        env:
+          COMET_TEST_MAVEN_DOWNLOADS: "1"
+        run: |
+          python3 dev/ci/test-download-retry.py
+          python3 dev/ci/test-delta-gate.py
+
       - name: Install actionlint
         run: |
           curl -sSfL https://raw.githubusercontent.com/rhysd/actionlint/main/scripts/download-actionlint.bash | bash
diff --git a/.github/workflows/delta_build_gate.yml b/.github/workflows/delta_build_gate.yml
index 245c2d26738..1f570651ae3 100644
--- a/.github/workflows/delta_build_gate.yml
+++ b/.github/workflows/delta_build_gate.yml
@@ -87,6 +87,32 @@ jobs:
           rust-version: ${{ env.RUST_VERSION }}
           jdk-version: 17
 
+      - name: Compute Delta Maven cache key
+        id: delta-maven-cache-key
+        run: echo "hash=${{ hashFiles('**/pom.xml', '.mvn/wrapper/maven-wrapper.properties') }}" >> "$GITHUB_OUTPUT"
+
+      - name: Cache Delta Maven dependencies
+        uses: actions/cache@v6
+        with:
+          path: |
+            ~/.m2/repository
+            /root/.m2/repository
+          key: ${{ runner.os }}-${{ runner.arch }}-delta-maven-jdk17-${{ steps.delta-maven-cache-key.outputs.hash }}
+          restore-keys: |
+            ${{ runner.os }}-${{ runner.arch }}-delta-maven-jdk17-
+
+      - name: Setup Maven
+        uses: ./.github/actions/setup-maven
+
       - name: Run dev/verify-contrib-delta-gate.sh
         run: |
           dev/verify-contrib-delta-gate.sh
+
+      - name: Upload Delta gate logs
+        if: failure()
+        uses: actions/upload-artifact@v7
+        with:
+          name: delta-build-gate-logs
+          path: artifactlog/delta-build-gate/
+          if-no-files-found: ignore
+          retention-days: 7
diff --git a/.github/workflows/docs.yaml b/.github/workflows/docs.yaml
index 1ca45266304..e475de5f140 100644
--- a/.github/workflows/docs.yaml
+++ b/.github/workflows/docs.yaml
@@ -49,6 +49,9 @@ jobs:
           java-version: '17'
           cache: 'maven'
 
+      - name: Setup Maven
+        uses: ./.github/actions/setup-maven
+
       - name: Install dependencies
         run: |
           set -x
diff --git a/.github/workflows/iceberg_spark_test_reusable.yml b/.github/workflows/iceberg_spark_test_reusable.yml
index 4b72cf47c2b..37552810a28 100644
--- a/.github/workflows/iceberg_spark_test_reusable.yml
+++ b/.github/workflows/iceberg_spark_test_reusable.yml
@@ -129,6 +129,8 @@ jobs:
         with:
           name: native-lib-iceberg
           path: native/target/release/
+      - name: Setup Maven
+        uses: ./.github/actions/setup-maven
       - name: Build Comet
         run: |
           ./mvnw install -Prelease -DskipTests -Pspark-${{ inputs.spark-short }} -Pscala-${{ inputs.scala }}
@@ -164,6 +166,8 @@ jobs:
         with:
           name: native-lib-iceberg
           path: native/target/release/
+      - name: Setup Maven
+        uses: ./.github/actions/setup-maven
       - name: Build Comet
         run: |
           ./mvnw install -Prelease -DskipTests -Pspark-${{ inputs.spark-short }} -Pscala-${{ inputs.scala }}
@@ -199,6 +203,8 @@ jobs:
         with:
           name: native-lib-iceberg
           path: native/target/release/
+      - name: Setup Maven
+        uses: ./.github/actions/setup-maven
       - name: Build Comet
         run: |
           ./mvnw install -Prelease -DskipTests -Pspark-${{ inputs.spark-short }} -Pscala-${{ inputs.scala }}
diff --git a/.github/workflows/pr_benchmark_check.yml b/.github/workflows/pr_benchmark_check.yml
index 933714d9d72..91df16ccdee 100644
--- a/.github/workflows/pr_benchmark_check.yml
+++ b/.github/workflows/pr_benchmark_check.yml
@@ -69,6 +69,9 @@ jobs:
           restore-keys: |
             ${{ runner.os }}-benchmark-maven-
 
+      - name: Setup Maven
+        uses: ./.github/actions/setup-maven
+
       - name: Check Scala compilation and linting
         # Pin to spark-4.0 (Scala 2.13.16) because the default profile is now
         # spark-4.1 / Scala 2.13.17, and semanticdb-scalac_2.13.17 is not yet
diff --git a/.github/workflows/pr_build_linux.yml b/.github/workflows/pr_build_linux.yml
index b7c88254c93..42a273879dc 100644
--- a/.github/workflows/pr_build_linux.yml
+++ b/.github/workflows/pr_build_linux.yml
@@ -119,6 +119,9 @@ jobs:
           restore-keys: |
             ${{ runner.os }}-java-maven-
 
+      - name: Setup Maven
+        uses: ./.github/actions/setup-maven
+
       - name: Run scalafix check
         run: |
           ./mvnw -B package -DskipTests scalafix:scalafix -Dscalafix.mode=CHECK -Psemanticdb ${{ matrix.profile.maven_opts }}
@@ -173,6 +176,9 @@ jobs:
           restore-keys: |
             ${{ runner.os }}-java-maven-
 
+      - name: Setup Maven
+        uses: ./.github/actions/setup-maven
+
       - name: Compile (skip tests)
         run: ./mvnw -B install -DskipTests -Dmaven.test.skip=true -Pspark-4.1
 
@@ -492,6 +498,9 @@ jobs:
           restore-keys: |
             ${{ runner.os }}-java-maven-
 
+      - name: Setup Maven
+        uses: ./.github/actions/setup-maven
+
       - name: Cache TPC-H data
         id: cache-tpch
         uses: actions/cache@v6
@@ -506,7 +515,7 @@ jobs:
       - name: Generate TPC-H data (SF=1)
         if: steps.cache-tpch.outputs.cache-hit != 'true'
         run: |
-          cd spark && MAVEN_OPTS='-Xmx20g' ../mvnw -B -Prelease exec:java -Dexec.mainClass="org.apache.spark.sql.GenTPCHData" -Dexec.classpathScope="test" -Dexec.cleanupDaemonThreads="false" -Dexec.args="--location `pwd`/.. --scaleFactor 1 --numPartitions 1 --overwrite"
+          cd spark && MAVEN_OPTS="${MAVEN_OPTS:-} -Xmx20g" ../mvnw -B -Prelease exec:java -Dexec.mainClass="org.apache.spark.sql.GenTPCHData" -Dexec.classpathScope="test" -Dexec.cleanupDaemonThreads="false" -Dexec.args="--location `pwd`/.. --scaleFactor 1 --numPartitions 1 --overwrite"
 
       - name: Run TPC-H queries
         run: |
@@ -547,6 +556,9 @@ jobs:
           restore-keys: |
             ${{ runner.os }}-java-maven-
 
+      - name: Setup Maven
+        uses: ./.github/actions/setup-maven
+
       - name: Cache TPC-DS data
         id: cache-tpcds
         uses: actions/cache@v6
@@ -575,7 +587,7 @@ jobs:
       - name: Generate TPC-DS data (SF=1)
         if: steps.cache-tpcds.outputs.cache-hit != 'true'
         run: |
-          cd spark && MAVEN_OPTS='-Xmx20g' ../mvnw -B -Prelease exec:java -Dexec.mainClass="org.apache.spark.sql.GenTPCDSData" -Dexec.classpathScope="test" -Dexec.cleanupDaemonThreads="false" -Dexec.args="--dsdgenDir `pwd`/../tpcds-kit/tools --location `pwd`/../tpcds-sf-1 --scaleFactor 1 --numPartitions 1"
+          cd spark && MAVEN_OPTS="${MAVEN_OPTS:-} -Xmx20g" ../mvnw -B -Prelease exec:java -Dexec.mainClass="org.apache.spark.sql.GenTPCDSData" -Dexec.classpathScope="test" -Dexec.cleanupDaemonThreads="false" -Dexec.args="--dsdgenDir `pwd`/../tpcds-kit/tools --location `pwd`/../tpcds-sf-1 --scaleFactor 1 --numPartitions 1"
 
       - name: Run TPC-DS queries (Sort merge join)
         run: |
diff --git a/.github/workflows/pyarrow_udf_test.yml b/.github/workflows/pyarrow_udf_test.yml
index 1a03962685d..8b52c70e2a2 100644
--- a/.github/workflows/pyarrow_udf_test.yml
+++ b/.github/workflows/pyarrow_udf_test.yml
@@ -43,6 +43,7 @@ on:
       - "spark/src/test/resources/pyspark/test_pyarrow_udf.py"
       - "spark/src/test/spark-3.5/org/apache/spark/sql/comet/CometMapInBatchSuite.scala"
       - "spark/src/test/spark-4.x/org/apache/spark/sql/comet/CometMapInBatchSuite.scala"
+      - ".github/actions/setup-maven/**"
       - ".github/workflows/pyarrow_udf_test.yml"
   pull_request:
     paths: *feature-paths
@@ -100,6 +101,9 @@ jobs:
           restore-keys: |
             ${{ runner.os }}-java-maven-
 
+      - name: Setup Maven
+        uses: ./.github/actions/setup-maven
+
       - name: Build Comet (debug, ${{ matrix.name }} / Scala 2.13)
         run: |
           cd native && cargo build
diff --git a/.github/workflows/spark_sql_test_reusable.yml b/.github/workflows/spark_sql_test_reusable.yml
index 5d4f8f0473e..74f43aebd80 100644
--- a/.github/workflows/spark_sql_test_reusable.yml
+++ b/.github/workflows/spark_sql_test_reusable.yml
@@ -129,13 +129,7 @@ jobs:
       - name: Pre-compile Spark Test classes
         run: |
           cd apache-spark
-          # Mirror the workaround from `Run Spark tests` below: Comet's mvn
-          # install populates partial Parquet entries (main JAR + POM but no
-          # `*-tests.jar` classifier). Coursier then sees the POM in
-          # mavenLocal, declares the artifact "found locally", and refuses
-          # to fall back to Maven Central for the missing test classifier.
-          # Wiping the parquet cache forces a clean remote fetch.
-          rm -rf /root/.m2/repository/org/apache/parquet
+          # setup-spark-builder already resolved these projects' dependencies.
           # Compile Test sources for the three subprojects the matrix touches.
           # SBT will transitively compile Compile/compile of their dependencies
           # plus any Test/compile pulled in by `dependsOn(... % "test->test")`,
@@ -167,14 +161,14 @@ jobs:
     strategy:
       matrix:
         module:
-          - {name: "catalyst", args1: "catalyst/test", args2: ""}
+          - {name: "catalyst", project: "catalyst", args1: "catalyst/test", args2: ""}
           # sql_core-* set HEAP_SIZE / METASPACE_SIZE so SparkBuild.scala caps
-          - {name: "sql_core-1", args1: "", args2: "sql/testOnly * -- -l org.apache.spark.tags.ExtendedSQLTest -l org.apache.spark.tags.SlowSQLTest", heap: "3g", metaspace: "1g"}
-          - {name: "sql_core-2", args1: "", args2: "sql/testOnly * -- -n org.apache.spark.tags.ExtendedSQLTest", heap: "3g", metaspace: "1g"}
-          - {name: "sql_core-3", args1: "", args2: "sql/testOnly * -- -n org.apache.spark.tags.SlowSQLTest", heap: "3g", metaspace: "1g"}
-          - {name: "sql_hive-1", args1: "", args2: "hive/testOnly * -- -l org.apache.spark.tags.ExtendedHiveTest -l org.apache.spark.tags.SlowHiveTest"}
-          - {name: "sql_hive-2", args1: "", args2: "hive/testOnly * -- -n org.apache.spark.tags.ExtendedHiveTest"}
-          - {name: "sql_hive-3", args1: "", args2: "hive/testOnly * -- -n org.apache.spark.tags.SlowHiveTest"}
+          - {name: "sql_core-1", project: "sql", args1: "", args2: "sql/testOnly * -- -l org.apache.spark.tags.ExtendedSQLTest -l org.apache.spark.tags.SlowSQLTest", heap: "3g", metaspace: "1g"}
+          - {name: "sql_core-2", project: "sql", args1: "", args2: "sql/testOnly * -- -n org.apache.spark.tags.ExtendedSQLTest", heap: "3g", metaspace: "1g"}
+          - {name: "sql_core-3", project: "sql", args1: "", args2: "sql/testOnly * -- -n org.apache.spark.tags.SlowSQLTest", heap: "3g", metaspace: "1g"}
+          - {name: "sql_hive-1", project: "hive", args1: "", args2: "hive/testOnly * -- -l org.apache.spark.tags.ExtendedHiveTest -l org.apache.spark.tags.SlowHiveTest"}
+          - {name: "sql_hive-2", project: "hive", args1: "", args2: "hive/testOnly * -- -n org.apache.spark.tags.ExtendedHiveTest"}
+          - {name: "sql_hive-3", project: "hive", args1: "", args2: "hive/testOnly * -- -n org.apache.spark.tags.SlowHiveTest"}
       fail-fast: false
     name: spark-sql-${{ matrix.module.name }}/spark-${{ inputs.spark-full }}-jdk${{ inputs.java }}
     runs-on: ubuntu-24.04
@@ -208,10 +202,10 @@ jobs:
           spark-short-version: ${{ inputs.spark-short }}
           skip-native-build: true
           skip-spark-clone: true
+          sbt-projects: ${{ matrix.module.project }}
       - name: Run Spark tests
         run: |
           cd apache-spark
-          rm -rf /root/.m2/repository/org/apache/parquet # somehow parquet cache requires cleanups
           # set SBTOPTS
           printf -- '-J-Xms1g\n-J-Xmx4g\n-J-XX:MaxMetaspaceSize=1g\n' > .sbtopts
           export SERIAL_SBT_TESTS=1
diff --git a/.github/workflows/spark_sql_writer_tests.yml b/.github/workflows/spark_sql_writer_tests.yml
index c813a8b39e1..f697e17adce 100644
--- a/.github/workflows/spark_sql_writer_tests.yml
+++ b/.github/workflows/spark_sql_writer_tests.yml
@@ -116,11 +116,11 @@ jobs:
           spark-version: ${{ steps.resolve.outputs.spark-full }}
           spark-short-version: ${{ inputs.spark-version }}
           skip-native-build: true
+          sbt-projects: sql
 
       - name: Run Parquet writer tests
         run: |
           cd apache-spark
-          rm -rf /root/.m2/repository/org/apache/parquet # somehow parquet cache requires cleanups
           # SERIAL_SBT_TESTS gates SparkParallelTestGrouping in
           # project/SparkBuild.scala. We always set it to reduce peak memory
           # on standard 7 GB runners (3.5 and 4.1 are unaffected by the
diff --git a/dev/ci/compute-changes.py b/dev/ci/compute-changes.py
index 9b7cd2f1691..5e4c977d09c 100644
--- a/dev/ci/compute-changes.py
+++ b/dev/ci/compute-changes.py
@@ -41,6 +41,7 @@
         ".github/workflows/ci.yml",
         ".github/workflows/pr_build_linux.yml",
         ".github/actions/setup-builder/**",
+        ".github/actions/setup-maven/**",
         ".github/actions/java-test/**",
         ".github/actions/rust-test/**",
         "!**.md",
@@ -64,6 +65,7 @@
         ".github/workflows/ci.yml",
         ".github/workflows/pr_build_macos.yml",
         ".github/actions/setup-macos-builder/**",
+        ".github/actions/setup-maven/**",
         ".github/actions/java-test/**",
         "!**.md",
         "!native/core/benches/**",
@@ -72,11 +74,13 @@
         "!spark/src/main/scala/org/apache/comet/GenerateDocs.scala",
     ],
     "benchmark": [
+        ".github/actions/setup-maven/**",
         "native/core/benches/**",
         "native/spark-expr/benches/**",
         "spark/src/test/scala/org/apache/spark/sql/benchmark/**",
     ],
     "docs": [
+        ".github/actions/setup-maven/**",
         ".asf.yaml",
         ".github/workflows/docs.yaml",
         "docs/**",
@@ -104,12 +108,15 @@
         "!spark/src/main/scala/org/apache/comet/GenerateDocs.scala",
         "spark/pom.xml",
         "dev/diffs/3.4.3.diff",
+        "dev/ci/retry-download.sh",
+        "dev/ci/resolve-spark-dependencies.sh",
         "pom.xml",
         "rust-toolchain.toml",
         ".github/workflows/ci.yml",
         ".github/workflows/spark_sql_test_reusable.yml",
         ".github/actions/setup-builder/**",
         ".github/actions/setup-spark-builder/**",
+        ".github/actions/setup-maven/**",
     ],
     "spark_3_5": [
         "native/**/src/**",
@@ -126,12 +133,15 @@
         "!spark/src/main/scala/org/apache/comet/GenerateDocs.scala",
         "spark/pom.xml",
         "dev/diffs/3.5.9.diff",
+        "dev/ci/retry-download.sh",
+        "dev/ci/resolve-spark-dependencies.sh",
         "pom.xml",
         "rust-toolchain.toml",
         ".github/workflows/ci.yml",
         ".github/workflows/spark_sql_test_reusable.yml",
         ".github/actions/setup-builder/**",
         ".github/actions/setup-spark-builder/**",
+        ".github/actions/setup-maven/**",
     ],
     "spark_4_0": [
         "native/**/src/**",
@@ -148,12 +158,15 @@
         "!spark/src/main/scala/org/apache/comet/GenerateDocs.scala",
         "spark/pom.xml",
         "dev/diffs/4.0.4.diff",
+        "dev/ci/retry-download.sh",
+        "dev/ci/resolve-spark-dependencies.sh",
         "pom.xml",
         "rust-toolchain.toml",
         ".github/workflows/ci.yml",
         ".github/workflows/spark_sql_test_reusable.yml",
         ".github/actions/setup-builder/**",
         ".github/actions/setup-spark-builder/**",
+        ".github/actions/setup-maven/**",
     ],
     "spark_4_1": [
         "native/**/src/**",
@@ -170,12 +183,15 @@
         "!spark/src/main/scala/org/apache/comet/GenerateDocs.scala",
         "spark/pom.xml",
         "dev/diffs/4.1.3.diff",
+        "dev/ci/retry-download.sh",
+        "dev/ci/resolve-spark-dependencies.sh",
         "pom.xml",
         "rust-toolchain.toml",
         ".github/workflows/ci.yml",
         ".github/workflows/spark_sql_test_reusable.yml",
         ".github/actions/setup-builder/**",
         ".github/actions/setup-spark-builder/**",
+        ".github/actions/setup-maven/**",
     ],
     "iceberg_1_8": [
         "native/**/src/**",
@@ -193,6 +209,7 @@
         ".github/workflows/iceberg_spark_test_reusable.yml",
         ".github/actions/setup-builder/**",
         ".github/actions/setup-iceberg-builder/**",
+        ".github/actions/setup-maven/**",
     ],
     "iceberg_1_9": [
         "native/**/src/**",
@@ -210,6 +227,7 @@
         ".github/workflows/iceberg_spark_test_reusable.yml",
         ".github/actions/setup-builder/**",
         ".github/actions/setup-iceberg-builder/**",
+        ".github/actions/setup-maven/**",
     ],
     "iceberg_1_10": [
         "native/**/src/**",
@@ -227,6 +245,7 @@
         ".github/workflows/iceberg_spark_test_reusable.yml",
         ".github/actions/setup-builder/**",
         ".github/actions/setup-iceberg-builder/**",
+        ".github/actions/setup-maven/**",
     ],
     "iceberg_1_11": [
         "native/**/src/**",
@@ -244,6 +263,7 @@
         ".github/workflows/iceberg_spark_test_reusable.yml",
         ".github/actions/setup-builder/**",
         ".github/actions/setup-iceberg-builder/**",
+        ".github/actions/setup-maven/**",
     ],
 }
 
diff --git a/dev/ci/resolve-spark-dependencies.sh b/dev/ci/resolve-spark-dependencies.sh
new file mode 100755
index 00000000000..fe3fe11c260
--- /dev/null
+++ b/dev/ci/resolve-spark-dependencies.sh
@@ -0,0 +1,40 @@
+#!/usr/bin/env bash
+#
+# 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.
+
+# Run from the patched Spark checkout after Comet's Maven install. SBT's
+# update task resolves dependencies without compiling Spark or running tests.
+set -euo pipefail
+
+if [ "$#" -eq 0 ]; then
+  echo "Usage: $0 catalyst|sql|hive [...]" >&2
+  exit 2
+fi
+
+update_tasks=()
+for project in "$@"; do
+  case "$project" in
+    catalyst|sql|hive) update_tasks+=("$project/Test/update") ;;
+    *) echo "Unsupported Spark dependency project: $project" >&2; exit 2 ;;
+  esac
+done
+
+script_dir=$(cd "$(dirname "$0")" && pwd)
+export NOLINT_ON_COMPILE=true
+exec "$script_dir/retry-download.sh" build/sbt -batch -Dsbt.log.noformat=true \
+  -mem 1024 "${update_tasks[@]}"
diff --git a/dev/ci/retry-download.sh b/dev/ci/retry-download.sh
new file mode 100755
index 00000000000..3f2bfde123b
--- /dev/null
+++ b/dev/ci/retry-download.sh
@@ -0,0 +1,61 @@
+#!/usr/bin/env bash
+#
+# 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.
+
+# Only wrap dependency acquisition or tool bootstrap, never a build or test.
+# A missing artifact, a bad build definition, and other permanent failures must
+# still fail immediately. Keep the output visible and preserve the exit status.
+set -uo pipefail
+
+if [ "$#" -eq 0 ]; then
+  echo "Usage: $0 command [argument ...]" >&2
+  exit 2
+fi
+
+download_log=$(mktemp "${TMPDIR:-/tmp}/comet-download.XXXXXX") || exit 1
+trap 'rm -f "$download_log"' EXIT
+trap 'exit 130' INT
+trap 'exit 143' TERM
+
+for attempt in 1 2 3 4; do
+  "$@" 2>&1 | tee "$download_log"
+  command_status=("${PIPESTATUS[@]}")
+  status=${command_status[0]}
+  if [ "${command_status[1]}" -ne 0 ]; then
+    echo "Could not capture dependency download output." >&2
+    exit "${command_status[1]}"
+  fi
+  if [ "$status" -eq 0 ]; then
+    exit 0
+  fi
+
+  # Signals (including cancellation and OOM kills) are not download failures.
+  if [ "$status" -ge 128 ] || ! grep -Eiq \
+    '(status code:|response code:|HTTP/[0-9.]+|HTTP error|returned error:)[[:space:]]*(429|500|502|503|504)([^0-9]|$)|Connection reset|Connection timed out|ConnectTimeoutException|SocketTimeoutException|Read timed out|Temporary failure in name resolution|Network is unreachable|Remote host terminated the handshake' \
+    "$download_log"; then
+    exit "$status"
+  fi
+  if [ "$attempt" -eq 4 ]; then
+    echo "::error::Dependency download failed after $attempt attempts."
+    exit "$status"
+  fi
+
+  delay=$((10 * (1 << (attempt - 1)) + RANDOM % 5))
+  echo "::warning::Transient download failure; retrying in ${delay}s (attempt $attempt of 4)."
+  sleep "$delay" || exit "$?"
+done
diff --git a/dev/ci/test-delta-gate.py b/dev/ci/test-delta-gate.py
new file mode 100644
index 00000000000..8d7ca8e99ed
--- /dev/null
+++ b/dev/ci/test-delta-gate.py
@@ -0,0 +1,204 @@
+#!/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.
+
+"""Run the actual Delta gate and retry helper without network or build tools."""
+
+import json
+import os
+from pathlib import Path
+import shutil
+import subprocess
+import sys
+import tempfile
+import unittest
+
+
+REPO = Path(__file__).resolve().parents[2]
+PROFILES = ("spark-4.1", "spark-4.1,contrib-delta", "spark-3.5,contrib-delta", "spark-4.0,contrib-delta")
+
+# All fake commands record their arguments. Only Maven model resolution can
+# recover; a compiler failure includes the same network text to catch retries
+# accidentally being widened to compilation or tests.
+FAKE_TOOL = r'''
+import json, os, sys
+from pathlib import Path
+
+tool, args = Path(sys.argv[0]).name, sys.argv[1:]
+root = Path(os.environ["COMET_TEST_GATE_ROOT"])
+scenario = os.environ["COMET_TEST_GATE_SCENARIO"]
+calls_file = root / "calls.jsonl"
+calls = [json.loads(line) for line in calls_file.read_text().splitlines()] if calls_file.exists() else []
+with calls_file.open("a") as out:
+    out.write(json.dumps([tool, *args]) + "\n")
+
+def compile_stage(name):
+    if scenario == name:
+        for stream in (sys.stdout, sys.stderr):
+            for index in range(100):
+                print(f"compiler output {index}", file=stream)
+        print("Connection reset during compiler fixture", file=sys.stderr)
+        sys.exit(71)
+
+if tool == "sleep":
+    sys.exit(0)
+if tool == "nm":
+    if Path(args[0]).read_bytes().endswith(b"C"):
+        print("000001 T comet_contrib_delta")
+    sys.exit(0)
+if tool == "cargo":
+    if args[0] == "tree":
+        print("datafusion-comet 0.1.0")
+        print("Downloading delta_kernel registry metadata fixture", file=sys.stderr)
+        if "--features" in args:
+            print("comet-contrib-delta 0.1.0")
+    elif args[0] == "build":
+        contrib = "--features" in args
+        compile_stage("cargo-build-contrib" if contrib else "cargo-build-default")
+        lib = root / "native/target/debug/libcomet.so"
+        lib.parent.mkdir(parents=True, exist_ok=True)
+        lib.write_bytes(b"N" * 100 + (b"C" * 100 if contrib else b""))
+    sys.exit(0)
+
+assert tool == "mvnw", tool
+if "help:effective-pom" in args:
+    profile = next(arg[2:] for arg in args if arg.startswith("-P"))
+    attempt = 1 + sum(call[0] == "mvnw" and "-P" + profile in call for call in calls)
+    if scenario == "permanent:" + profile:
+        print("Permanent Maven model failure: " + profile, file=sys.stderr)
+        sys.exit(73)
+    if profile == "spark-4.1" and (scenario == "exhausted" or (scenario == "recover" and attempt <= 2)):
+        print(f"Could not transfer artifact: Connection reset attempt {attempt}", file=sys.stderr)
+        sys.exit(74)
+    pom = Path(next(arg[len("-Doutput="):] for arg in args if arg.startswith("-Doutput=")))
+    lines = ["<project>", "<dependencyManagement>", "</dependencyManagement>", "<dependencies>",
+             "<dependency>", "<groupId>org.apache.spark</groupId>", "</dependency>"]
+    if "contrib-delta" in profile:
+        version = {"spark-4.1": "4.1.0", "spark-3.5": "3.3.2", "spark-4.0": "4.0.0"}[profile.split(",")[0]]
+        lines += ["<dependency>", "<groupId>io.delta</groupId>", "<artifactId>delta-spark_2.13</artifactId>",
+                  f"<version>{version}</version>", "</dependency>"]
+    lines += ["</dependencies>", "<profiles>", "<groupId>io.delta</groupId>", "</profiles>", "</project>"]
+    pom.write_text("\n".join(lines) + "\n")
+    print("effective-pom success: " + profile)
+else:
+    assert "test-compile" in args, args
+    compile_stage("maven-test-compile")
+    classes = root / "spark/target/classes"
+    classes.mkdir(parents=True, exist_ok=True)
+    if scenario == "leaked-class":
+        leaked = classes / "org/apache/comet/contrib/Delta.class"
+        leaked.parent.mkdir(parents=True, exist_ok=True)
+        leaked.touch()
+'''
+
+
+class DeltaGateTest(unittest.TestCase):
+    def run_gate(self, scenario="success"):
+        with tempfile.TemporaryDirectory(prefix="comet-delta-gate-test-") as temp:
+            root = Path(temp)
+            (root / "dev/ci").mkdir(parents=True)
+            (root / "native").mkdir()
+            (root / "bin").mkdir()
+            for script in ("dev/verify-contrib-delta-gate.sh", "dev/ci/retry-download.sh"):
+                shutil.copy2(REPO / script, root / script)
+            for tool in ("mvnw", "cargo", "nm", "sleep"):
+                path = root / ("mvnw" if tool == "mvnw" else "bin/" + tool)
+                path.write_text(f"#!{sys.executable}\n" + FAKE_TOOL)
+                path.chmod(0o755)
+            env = dict(
+                os.environ,
+                PATH=f"{root / 'bin'}{os.pathsep}{os.environ['PATH']}",
+                COMET_TEST_GATE_ROOT=str(root),
+                COMET_TEST_GATE_SCENARIO=scenario,
+                COMET_DELTA_GATE_LOG_DIR=str(root / "logs"),
+            )
+            result = subprocess.run(
+                ["bash", str(root / "dev/verify-contrib-delta-gate.sh")],
+                cwd=root, env=env, capture_output=True, text=True, timeout=20,
+            )
+            calls = [json.loads(line) for line in (root / "calls.jsonl").read_text().splitlines()]
+            logs = {path.name: path.read_text() for path in (root / "logs").glob("*.log")}
+            return result, calls, logs
+
+    def test_normal_gate_keeps_download_stderr_out_of_dependency_tree(self):
+        result, calls, logs = self.run_gate()
+        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+        self.assertIn("All gate checks passed", result.stdout)
+        self.assertIn("delta_kernel", logs["cargo-tree-default.stderr.log"])
+        self.assertFalse(any(call[0] == "sleep" for call in calls))
+
+    def test_permanent_model_failures_keep_each_profile_status(self):
+        for profile in PROFILES:
+            with self.subTest(profile=profile):
+                result, calls, _ = self.run_gate("permanent:" + profile)
+                self.assertEqual(result.returncode, 73, result.stdout + result.stderr)
+                self.assertIn("Permanent Maven model failure: " + profile, result.stderr)
+                self.assertEqual(sum("help:effective-pom" in call and "-P" + profile in call for call in calls), 1)
+                self.assertFalse(any(call[0] == "sleep" or "test-compile" in call for call in calls))
+
+    def test_transient_model_download_recovers(self):
+        result, calls, logs = self.run_gate("recover")
+        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+        model_calls = [call for call in calls if "help:effective-pom" in call]
+        self.assertEqual(sum("-Pspark-4.1" in call for call in model_calls), 3)
+        self.assertTrue(all("-U" in call for call in model_calls))
+        self.assertEqual(sum(call[0] == "sleep" for call in calls), 2)
+        self.assertIn("All gate checks passed", result.stdout)
+        log = logs["maven-default-spark-4.1.stdout.log"]
+        for message in ("attempt 1", "attempt 2", "effective-pom success"):
+            self.assertIn(message, log)
+
+    def test_transient_model_download_exhaustion_stays_fatal(self):
+        result, calls, logs = self.run_gate("exhausted")
+        self.assertEqual(result.returncode, 74, result.stdout + result.stderr)
+        self.assertEqual(sum("help:effective-pom" in call for call in calls), 4)
+        self.assertEqual(sum(call[0] == "sleep" for call in calls), 3)
+        self.assertFalse(any("test-compile" in call for call in calls))
+        self.assertIn("failed after 4 attempts", result.stderr)
+        self.assertIn("attempt 1", logs["maven-default-spark-4.1.stdout.log"])
+        self.assertIn("attempt 4", logs["maven-default-spark-4.1.stdout.log"])
+
+    def test_compilation_is_not_retried_and_retains_complete_logs(self):
+        for stage in ("maven-test-compile", "cargo-build-default", "cargo-build-contrib"):
+            with self.subTest(stage=stage):
+                result, calls, logs = self.run_gate(stage)
+                self.assertEqual(result.returncode, 71, result.stdout + result.stderr)
+                self.assertFalse(any(call[0] == "sleep" for call in calls))
+                if stage == "maven-test-compile":
+                    attempts = [call for call in calls if "test-compile" in call]
+                else:
+                    attempts = [call for call in calls if call[:2] == ["cargo", "build"]
+                                and ("--features" in call) == (stage == "cargo-build-contrib")]
+                self.assertEqual(len(attempts), 1)
+                for stream in ("stdout", "stderr"):
+                    expected = [f"compiler output {index}" for index in range(100)]
+                    if stream == "stderr":
+                        expected.append("Connection reset during compiler fixture")
+                    self.assertEqual(logs[f"{stage}.{stream}.log"].splitlines(), expected)
+                self.assertIn("Connection reset during compiler fixture", result.stderr)
+                self.assertNotIn("compiler output 0\n", result.stdout + result.stderr)  # Bounded console tail.
+
+    def test_delta_class_leak_still_fails_gate(self):
+        result, calls, _ = self.run_gate("leaked-class")
+        self.assertEqual(result.returncode, 1, result.stdout + result.stderr)
+        self.assertIn("default Maven build compiled contrib classes", result.stdout)
+        self.assertFalse(any(call[:2] == ["cargo", "build"] for call in calls))
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/dev/ci/test-download-retry.py b/dev/ci/test-download-retry.py
new file mode 100644
index 00000000000..12c320a57a8
--- /dev/null
+++ b/dev/ci/test-download-retry.py
@@ -0,0 +1,396 @@
+#!/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.
+
+"""Regression tests for CI download handling; cache-path checks require Java."""
+
+import hashlib
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+import json
+import os
+from pathlib import Path
+import shutil
+import subprocess
+import sys
+import tempfile
+import threading
+import unittest
+
+
+RETRY = Path(__file__).with_name("retry-download.sh")
+RESOLVE_SPARK = Path(__file__).with_name("resolve-spark-dependencies.sh")
+REPO = Path(__file__).resolve().parents[2]
+MAVEN_CACHE_PATH = REPO / ".github/actions/setup-maven/cache-path.sh"
+
+
+class DownloadRetryTest(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory(prefix="comet-download-test-")
+        self.addCleanup(self.temp.cleanup)
+        self.root = Path(self.temp.name)
+        self.calls = self.root / "calls.jsonl"
+        self.delays = self.root / "delays"
+        self.command = self.root / "download.py"
+        self.command.write_text(
+            f"#!{sys.executable}\n"
+            "import json, os, pathlib, sys\n"
+            "calls = pathlib.Path(os.environ['COMET_TEST_CALLS'])\n"
+            "attempt = len(calls.read_text().splitlines()) if calls.exists() else 0\n"
+            "with calls.open('a') as out: out.write(json.dumps(sys.argv[1:]) + '\\n')\n"
+            "failures = int(os.environ['COMET_TEST_FAILURES'])\n"
+            "print('download output', flush=True)\n"
+            "if attempt < failures:\n"
+            "    print(os.environ['COMET_TEST_ERROR'], file=sys.stderr)\n"
+            "    sys.exit(int(os.environ['COMET_TEST_STATUS']))\n"
+        )
+        self.command.chmod(0o755)
+        (self.root / "build").mkdir()
+        (self.root / "build" / "sbt").symlink_to(self.command)
+        sleeper = self.root / "sleep"
+        sleeper.write_text(
+            '#!/bin/sh\nprintf "%s\\n" "$1" >> "$COMET_TEST_DELAYS"\n'
+            'exit "${COMET_TEST_SLEEP_STATUS:-0}"\n'
+        )
+        sleeper.chmod(0o755)
+
+    def run_download(
+        self, message="Connection reset", failures=0, status=17,
+        args=(), projects=None, sleep_status=0
+    ):
+        env = dict(
+            os.environ,
+            PATH=f"{self.root}{os.pathsep}{os.environ['PATH']}",
+            COMET_TEST_CALLS=str(self.calls),
+            COMET_TEST_DELAYS=str(self.delays),
+            COMET_TEST_ERROR=message,
+            COMET_TEST_FAILURES=str(failures),
+            COMET_TEST_STATUS=str(status),
+            COMET_TEST_SLEEP_STATUS=str(sleep_status),
+        )
+        command = ["bash", str(RETRY), sys.executable, str(self.command), *args]
+        if projects is not None:
+            command = ["bash", str(RESOLVE_SPARK), *projects]
+        result = subprocess.run(
+            command,
+            cwd=self.root,
+            env=env,
+            capture_output=True,
+            text=True,
+            timeout=15,
+        )
+        calls = (
+            [json.loads(line) for line in self.calls.read_text().splitlines()]
+            if self.calls.exists() else []
+        )
+        delays = (
+            [int(line) for line in self.delays.read_text().splitlines()]
+            if self.delays.exists() else []
+        )
+        return result, calls, delays
+
+    def assert_backoff(self, delays):
+        for index, delay in enumerate(delays):
+            self.assertIn(delay, range(10 * 2**index, 10 * 2**index + 5))
+
+    def test_success_does_not_retry_and_preserves_arguments(self):
+        args = ("one argument", "*", "-Dkey=literal $value")
+        result, calls, delays = self.run_download(args=args)
+        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+        self.assertEqual(calls, [list(args)])
+        self.assertEqual(delays, [])
+        self.assertIn("download output", result.stdout)
+
+    def test_transient_download_errors_retry(self):
+        messages = (
+            "Could not transfer artifact: Connection reset",
+            "Server returned HTTP response code: 502 for URL: https://repo.invalid/a.jar",
+            "status code: 429, reason phrase: Too Many Requests (429)",
+            "status code: 500, reason phrase: Internal Server Error (500)",
+            "HTTP/1.1 503 Service Unavailable",
+            "curl: (22) The requested URL returned error: 504",
+            "java.net.SocketTimeoutException: Read timed out",
+            "Network is unreachable (os error 101)",
+        )
+        for message in messages:
+            with self.subTest(message=message):
+                self.calls.unlink(missing_ok=True)
+                self.delays.unlink(missing_ok=True)
+                result, calls, delays = self.run_download(message, failures=2)
+                self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+                self.assertEqual(len(calls), 3)
+                self.assertEqual(len(delays), 2)
+                self.assert_backoff(delays)
+                self.assertIn(message, result.stdout)
+
+    def test_fourth_attempt_can_succeed(self):
+        result, calls, delays = self.run_download(failures=3)
+        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+        self.assertEqual(len(calls), 4)
+        self.assertEqual(len(delays), 3)
+        self.assert_backoff(delays)
+
+    def test_exhaustion_preserves_exit_code_without_extra_sleep(self):
+        result, calls, delays = self.run_download(failures=10, status=23)
+        self.assertEqual(result.returncode, 23)
+        self.assertEqual(len(calls), 4)
+        self.assertEqual(len(delays), 3)
+        self.assertIn("failed after 4 attempts", result.stdout)
+
+    def test_permanent_failures_do_not_retry(self):
+        for message in (
+            "status code: 404, reason phrase: Not Found (404)",
+            "status code: 401, reason phrase: Unauthorized (401)",
+            "[error] Could not find artifact missing:dependency:jar:1.0",
+            "[error] not found: value invalidBuildSetting",
+            "[ERROR] COMPILATION ERROR",
+            "[ERROR] There are test failures.",
+        ):
+            with self.subTest(message=message):
+                self.calls.unlink(missing_ok=True)
+                self.delays.unlink(missing_ok=True)
+                result, calls, delays = self.run_download(message, failures=10, status=42)
+                self.assertEqual(result.returncode, 42)
+                self.assertEqual(len(calls), 1)
+                self.assertEqual(delays, [])
+
+    def test_signal_exit_does_not_retry_even_after_network_message(self):
+        for status in (130, 137, 139, 143):
+            with self.subTest(status=status):
+                self.calls.unlink(missing_ok=True)
+                self.delays.unlink(missing_ok=True)
+                result, calls, delays = self.run_download(failures=10, status=status)
+                self.assertEqual(result.returncode, status)
+                self.assertEqual(len(calls), 1)
+                self.assertEqual(delays, [])
+
+    def test_missing_command_fails(self):
+        result = subprocess.run(["bash", str(RETRY)], capture_output=True, text=True)
+        self.assertEqual(result.returncode, 2)
+        self.assertIn("Usage:", result.stderr)
+
+    def test_interrupted_backoff_does_not_start_another_attempt(self):
+        result, calls, delays = self.run_download(failures=1, sleep_status=143)
+        self.assertEqual(result.returncode, 143)
+        self.assertEqual(len(calls), 1)
+        self.assertEqual(len(delays), 1)
+
+    def test_spark_retries_dependency_tasks_only(self):
+        result, calls, delays = self.run_download(
+            "Server returned HTTP response code: 502", failures=2, projects=("sql", "hive")
+        )
+        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+        expected = [
+            "-batch", "-Dsbt.log.noformat=true", "-mem", "1024",
+            "sql/Test/update", "hive/Test/update"
+        ]
+        self.assertEqual(calls, [expected] * 3)
+        self.assertEqual(len(delays), 2)
+        self.assert_backoff(delays)
+
+    def test_spark_build_definition_failure_does_not_retry(self):
+        result, calls, delays = self.run_download(
+            "[error] not found: value invalidBuildSetting", failures=10, projects=("catalyst",)
+        )
+        self.assertEqual(result.returncode, 17)
+        self.assertEqual(len(calls), 1)
+        self.assertEqual(delays, [])
+
+    def test_spark_rejects_commands_outside_dependency_resolution(self):
+        for projects in ((), ("sql/test",), ("sql", "hive/Test/compile")):
+            with self.subTest(projects=projects):
+                result, calls, delays = self.run_download(projects=projects)
+                self.assertEqual(result.returncode, 2)
+                self.assertEqual(calls, [])
+                self.assertEqual(delays, [])
+
+
+class MavenCachePathTest(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory(prefix="comet-maven-cache-test-")
+        self.addCleanup(self.temp.cleanup)
+        self.root = Path(self.temp.name)
+        self.java = shutil.which("java")
+        self.assertIsNotNone(self.java, "Maven cache-path tests require Java")
+
+    def cache_path(self, **overrides):
+        env = dict(os.environ)
+        for name in (
+            "MAVEN_OPTS", "MAVEN_USER_HOME", "JAVA_HOME", "JAVACMD",
+            "JAVA_TOOL_OPTIONS", "JDK_JAVA_OPTIONS", "_JAVA_OPTIONS",
+        ):
+            env.pop(name, None)
+        env["JAVACMD"] = self.java
+        env.update(overrides)
+        return subprocess.run(
+            ["bash", str(MAVEN_CACHE_PATH)], cwd=self.root, env=env,
+            capture_output=True, text=True, timeout=15,
+        )
+
+    def test_host_runner_caches_only_its_accessible_distribution(self):
+        runner_home = self.root / "runner"
+        distribution = runner_home / ".m2/wrapper/dists"
+        distribution.mkdir(parents=True)
+        marker = distribution / "downloaded-maven"
+        marker.write_text("cached")
+        result = self.cache_path(MAVEN_OPTS=f"-Duser.home={runner_home}")
+        self.assertEqual(result.returncode, 0, result.stderr)
+        paths = result.stdout.splitlines()
+        self.assertEqual(paths, [str(distribution)])
+        # Every emitted path can be traversed without touching /root. The old
+        # shared list aborted cache save with EACCES on the host runner.
+        self.assertEqual([p for path in paths for p in Path(path).iterdir()], [marker])
+
+    def test_container_uses_jvm_home_instead_of_shell_home(self):
+        result = self.cache_path(MAVEN_OPTS="-Duser.home=/root")
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(result.stdout.strip(), "/root/.m2/wrapper/dists")
+
+    def test_uses_java_home_or_path_when_javacmd_is_unset(self):
+        runner_home = self.root / "runner"
+        for java_home in (str(Path(self.java).resolve().parents[1]), ""):
+            with self.subTest(java_home=java_home):
+                result = self.cache_path(
+                    JAVACMD="", JAVA_HOME=java_home,
+                    MAVEN_OPTS=f"-Duser.home={runner_home}",
+                )
+                self.assertEqual(result.returncode, 0, result.stderr)
+                self.assertEqual(result.stdout.strip(), str(runner_home / ".m2/wrapper/dists"))
+
+    def test_wrapper_home_overrides_follow_wrapper_precedence(self):
+        configured_home = self.root / "wrapper"
+        env_home = self.root / "wrapper with spaces"
+        (self.root / ".mvn").mkdir()
+        config = self.root / ".mvn/jvm.config"
+        config.write_text(f"-Dmaven.user.home={configured_home}\n-Xmx128m\n")
+        result = self.cache_path(MAVEN_USER_HOME=str(env_home))
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(result.stdout.strip(), str(configured_home / "wrapper/dists"))
+        config.unlink()
+        result = self.cache_path(MAVEN_USER_HOME=str(env_home))
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(result.stdout.strip(), str(env_home / "wrapper/dists"))
+
+    def test_maven_opts_override_jvm_config_and_javacmd_overrides_java_home(self):
+        configured_home = self.root / "configured"
+        effective_home = self.root / "effective"
+        (self.root / ".mvn").mkdir()
+        (self.root / ".mvn/jvm.config").write_text(f"-Dmaven.user.home={configured_home}\n")
+        result = self.cache_path(
+            JAVA_HOME=str(self.root / "missing-jdk"),
+            MAVEN_OPTS=f"-Xmx128m\n-Dmaven.user.home={effective_home}",
+        )
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(result.stdout.strip(), str(effective_home / "wrapper/dists"))
+
+    def test_invalid_jvm_options_fail_instead_of_caching_a_guessed_path(self):
+        result = self.cache_path(MAVEN_OPTS="-XX:CometInvalidOption")
+        self.assertNotEqual(result.returncode, 0)
+        self.assertEqual(result.stdout, "")
+        self.assertIn("CometInvalidOption", result.stderr)
+
+
+@unittest.skipUnless(
+    os.environ.get("COMET_TEST_MAVEN_DOWNLOADS") == "1",
+    "requires Maven bootstrapped/configured by the setup-maven action",
+)
+class MavenTransferRetryTest(unittest.TestCase):
+    """Exercise CI's actual MAVEN_OPTS with Maven, not a simulated resolver.
+
+    The local mirror serves a parent POM. `validate` needs no plugins, does not
+    compile/run tests, and all artifact requests stay on this loopback server.
+    Preflight already bootstrapped Maven; only shorten the retry delay here.
+    """
+
+    def test_transfer_retries_are_bounded_and_permanent_failures_still_fail(self):
+        parent = (
+            '<project xmlns="http://maven.apache.org/POM/4.0.0">'
+            '<modelVersion>4.0.0</modelVersion><groupId>comet.ci</groupId>'
+            '<artifactId>parent</artifactId><version>1</version>'
+            '<packaging>pom</packaging></project>'
+        ).encode()
+        state = {}
+
+        class Handler(BaseHTTPRequestHandler):
+            def log_message(self, *_args):
+                pass
+
+            def do_GET(self):
+                if self.path.endswith(".pom"):
+                    state["requests"] += 1
+                    if state["requests"] <= state["failures"]:
+                        self.send_response(state["status"])
+                        self.send_header("Content-Length", "0")
+                        self.end_headers()
+                        return
+                    body = parent
+                elif self.path.endswith(".sha1"):
+                    body = hashlib.sha1(parent).hexdigest().encode()
+                else:
+                    self.send_error(404)
+                    return
+                self.send_response(200)
+                self.send_header("Content-Length", str(len(body)))
+                self.end_headers()
+                self.wfile.write(body)
+
+        server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+        self.addCleanup(server.server_close)
+        threading.Thread(target=server.serve_forever, daemon=True).start()
+        self.addCleanup(server.shutdown)
+        with tempfile.TemporaryDirectory(prefix="comet-maven-test-") as directory:
+            root = Path(directory)
+            (root / "pom.xml").write_text(
+                '<project xmlns="http://maven.apache.org/POM/4.0.0">'
+                '<modelVersion>4.0.0</modelVersion><parent><groupId>comet.ci</groupId>'
+                '<artifactId>parent</artifactId><version>1</version><relativePath/>'
+                '</parent><artifactId>child</artifactId><packaging>pom</packaging></project>'
+            )
+            (root / "settings.xml").write_text(
+                '<settings><mirrors><mirror><id>test</id><mirrorOf>*</mirrorOf>'
+                f'<url>http://127.0.0.1:{server.server_port}</url>'
+                '</mirror></mirrors></settings>'
+            )
+            env = dict(os.environ)
+            env["MAVEN_OPTS"] = (
+                env.get("MAVEN_OPTS", "")
+                + " -Daether.connector.http.retryHandler.interval=100"
+            )
+            for status, failures, expected_requests, succeeds in (
+                (429, 2, 3, True), (502, 2, 3, True),
+                (503, 10, 4, False), (404, 10, 1, False),
+            ):
+                with self.subTest(status=status):
+                    state.update(status=status, failures=failures, requests=0)
+                    result = subprocess.run(
+                        [
+                            str(REPO / "mvnw"), "-B", "-ntp",
+                            "-f", str(root / "pom.xml"),
+                            "-s", str(root / "settings.xml"),
+                            f"-Dmaven.repo.local={root / str(status)}", "validate",
+                        ],
+                        cwd=REPO, env=env, capture_output=True, text=True, timeout=30,
+                    )
+                    self.assertEqual(
+                        result.returncode == 0, succeeds,
+                        result.stdout + result.stderr,
+                    )
+                    self.assertEqual(state["requests"], expected_requests)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/dev/verify-contrib-delta-gate.sh b/dev/verify-contrib-delta-gate.sh
index 3ce3a63ae01..5a8a51035fe 100755
--- a/dev/verify-contrib-delta-gate.sh
+++ b/dev/verify-contrib-delta-gate.sh
@@ -66,20 +66,41 @@ export GIT_CONFIG_COUNT=$((GIT_CONFIG_IDX + 1))
 # launch, produce no dependency output, and trip the anti-vacuous guard below.
 # `./mvnw` self-provisions the pinned Maven version regardless of the image.
 MVNW="$ROOT/mvnw"
+RETRY_DOWNLOAD="$ROOT/dev/ci/retry-download.sh"
+
+# Keep a fixed set of command logs outside Maven's target directories, which the
+# compiled-class gate cleans. CI uploads this directory if any gate fails.
+LOG_DIR="${COMET_DELTA_GATE_LOG_DIR:-$ROOT/artifactlog/delta-build-gate}"
+mkdir -p "$LOG_DIR"
+LOG_DIR="$(cd "$LOG_DIR" && pwd)"
 
 red()   { printf '\033[31m%s\033[0m\n' "$*"; }
 green() { printf '\033[32m%s\033[0m\n' "$*"; }
 hdr()   { printf '\n\033[36m==> %s\033[0m\n' "$*"; }
 
+# Keep stdout separate so cargo-tree output can still be inspected without
+# mistaking dependency-download messages on stderr for entries in the tree.
+run_logged() { # args: log name, command, arguments
+  local name="$1"
+  shift
+  local status
+  if "$@" >"$LOG_DIR/$name.stdout.log" 2>"$LOG_DIR/$name.stderr.log"; then
+    cat "$LOG_DIR/$name.stdout.log"
+  else
+    status=$?
+    red "FAIL: $name exited with status $status (logs: $LOG_DIR/$name.*.log)" >&2
+    tail -n 60 "$LOG_DIR/$name.stdout.log" "$LOG_DIR/$name.stderr.log" >&2 || true
+    return "$status"
+  fi
+}
+
 # ---- Cargo gate -----------------------------------------------------------
 
 hdr "Cargo: default build does not depend on comet-contrib-delta / delta_kernel"
 cd "$NATIVE_DIR"
-TREE_DEFAULT="$(cargo tree -p datafusion-comet --no-default-features 2>/dev/null)"
-# Anti-vacuous (mirrors the Maven gate below): a failing `cargo tree` yields empty output, and the
-# command-substitution failure doesn't trip `set -e` in an assignment -- so assert the root crate we
-# KNOW is always present before concluding "no Delta deps", otherwise a broken cargo-tree run would
-# pass the leak check vacuously. (`datafusion-comet ` with a trailing space matches only the root
+TREE_DEFAULT="$(run_logged cargo-tree-default cargo tree -p datafusion-comet --no-default-features)"
+# Also reject unexpectedly empty output after a successful command before concluding
+# "no Delta deps". (`datafusion-comet ` with a trailing space matches only the root
 # crate line, not `datafusion-comet-proto`/`-common`.)
 if ! grep -q 'datafusion-comet ' <<<"$TREE_DEFAULT"; then
   red "FAIL: default cargo tree produced no datafusion-comet entry (cargo tree likely failed;"
@@ -93,7 +114,7 @@ if grep -qE 'comet-contrib-delta|delta_kernel|delta-kernel' <<<"$TREE_DEFAULT";
 fi
 green "OK: cargo tree default is clean of contrib + kernel"
 
-TREE_CONTRIB="$(cargo tree -p datafusion-comet --features contrib-delta 2>/dev/null)"
+TREE_CONTRIB="$(run_logged cargo-tree-contrib cargo tree -p datafusion-comet --features contrib-delta)"
 # The build-gate unit ships a STUB contrib crate, so the gated tree pulls in
 # `comet-contrib-delta` but not yet the heavy `delta_kernel` (that arrives with the
 # native read-path unit). Require the contrib crate to be present; the symbol check
@@ -111,37 +132,57 @@ hdr "Maven: default profile excludes io.delta:* dependencies"
 cd "$ROOT"
 # `dependency:list` can't run in a fresh CI checkout: it needs the sibling reactor JARs
 # (comet-common, the shims) which aren't built, so it fails with a resolution error and no
-# output. `help:effective-pom` only merges POM models (no artifact resolution), so it works
-# without a build. We extract the ACTIVE top-level <dependencies> -- after </dependencyManagement>,
+# output. `help:effective-pom` only merges POM models (no reactor JAR resolution), so it works
+# without a build, but its Maven plugins still need to be downloaded. We extract the ACTIVE
+# top-level <dependencies> -- after </dependencyManagement>,
 # before the <profiles> listing -- which is exactly what dependency:list would have shown for the
 # active profiles (and excludes the inactive contrib-delta profile's own io.delta declaration).
-# Last effective-pom invocation's combined output, kept so the anti-vacuous guard can SHOW why
-# mvn failed instead of swallowing it.
-EPOM_LOG="$(mktemp)"
-delta_active_deps() { # args: -P / -D flags
-  local epom
-  epom="$(mktemp)"
-  "$MVNW" help:effective-pom -Djava.version=17 -Dmaven.gitcommitid.skip -pl spark \
-    -Doutput="$epom" "$@" >"$EPOM_LOG" 2>&1 || true
-  awk '/<\/dependencyManagement>/{f=1} /<profiles>/{f=0} f' "$epom"
-  rm -f "$epom"
+delta_active_deps() { # args: log name, -P / -D flags
+  local name="$1"
+  shift
+  local epom="$LOG_DIR/$name.pom.log"
+  : >"$epom"
+  # Retry only transient download failures during model/plugin resolution. -U
+  # lets a retry resolve an artifact whose earlier transfer failure was cached.
+  # Compilation and tests below are deliberately not retried.
+  if run_logged "$name" "$RETRY_DOWNLOAD" "$MVNW" -U help:effective-pom \
+      -Djava.version=17 -Dmaven.gitcommitid.skip -pl spark \
+      -Doutput="$epom" "$@" >/dev/null; then
+    awk '/<\/dependencyManagement>/{f=1} /<profiles>/{f=0} f' "$epom"
+  else
+    return $?
+  fi
 }
 # Resolved delta-spark version from the active deps (empty if absent).
-delta_spark_version() { # args: -P flags
-  delta_active_deps "$@" |
-    grep -A2 'artifactId>delta-spark' |
-    grep -oE '<version>[^<]+' | sed -n '1s/<version>//p'
+delta_spark_version() { # args: log name, -P flags
+  local deps
+  deps="$(delta_active_deps "$@")" || return $?
+  # Keep an absent version empty so the profile-specific assertion below reports
+  # it, while an actual Maven failure retains its status and diagnostic log.
+  awk '
+    /artifactId>delta-spark/ { remaining = 3 }
+    remaining > 0 {
+      if (/<version>/) {
+        sub(/.*<version>/, "")
+        sub(/<.*/, "")
+        print
+        exit
+      }
+      remaining--
+    }
+  ' <<<"$deps"
 }
 
-DEPS_DEFAULT="$(delta_active_deps -Pspark-4.1)"
-# Anti-vacuous: a broken mvn run yields empty output; assert a dep we KNOW is always present so a
-# broken run fails loudly instead of "passing" the io.delta check by finding nothing.
+DEPS_DEFAULT="$(delta_active_deps maven-default-spark-4.1 -Pspark-4.1)"
+# Even after Maven succeeds, assert a dep we KNOW is always present rather than
+# "passing" the io.delta check on unexpectedly empty effective-pom output.
 if ! grep -q '<groupId>org.apache.spark</groupId>' <<<"$DEPS_DEFAULT"; then
-  red "FAIL: default effective-pom produced no org.apache.spark deps (mvn likely failed;"
-  red "      refusing to conclude 'zero io.delta' vacuously)"
+  red "FAIL: default effective-pom produced no org.apache.spark deps;"
+  red "      refusing to conclude 'zero io.delta' vacuously"
   red "      --- mvnw: $MVNW (java=${JAVA_HOME:-unset}) ---"
   red "      --- effective-pom output (last 60 lines) ---"
-  tail -60 "$EPOM_LOG" >&2 || true
+  tail -n 60 "$LOG_DIR/maven-default-spark-4.1.stdout.log" \
+    "$LOG_DIR/maven-default-spark-4.1.stderr.log" >&2 || true
   exit 1
 fi
 if grep -q '<groupId>io.delta</groupId>' <<<"$DEPS_DEFAULT"; then
@@ -154,17 +195,17 @@ green "OK: default Maven build has zero io.delta dependencies"
 # Per-Spark Delta version pinning: spark-4.1 -> delta-spark 4.1.x, spark-3.5 -> 3.x, spark-4.0 ->
 # 4.0.x (Delta 4.1 needs Spark 4.1 internals; 4.0 must stay on 4.0.x to avoid a runtime
 # NoSuchMethodError on ParserInterface.$init$).
-V41="$(delta_spark_version -Pspark-4.1,contrib-delta)"
+V41="$(delta_spark_version maven-contrib-spark-4.1 -Pspark-4.1,contrib-delta)"
 case "$V41" in
   4.1.*) green "OK: -Pcontrib-delta + spark-4.1 correctly pulls delta-spark $V41" ;;
   *) red "FAIL: -Pcontrib-delta + spark-4.1 expected delta-spark 4.1.x, got '${V41:-<none>}'"; exit 1 ;;
 esac
-V35="$(delta_spark_version -Pspark-3.5,contrib-delta)"
+V35="$(delta_spark_version maven-contrib-spark-3.5 -Pspark-3.5,contrib-delta)"
 case "$V35" in
   3.*) green "OK: -Pcontrib-delta + spark-3.5 correctly pulls delta-spark $V35" ;;
   *) red "FAIL: -Pcontrib-delta + spark-3.5 expected delta-spark 3.x, got '${V35:-<none>}'"; exit 1 ;;
 esac
-V40="$(delta_spark_version -Pspark-4.0,contrib-delta)"
+V40="$(delta_spark_version maven-contrib-spark-4.0 -Pspark-4.0,contrib-delta)"
 case "$V40" in
   4.0.*) green "OK: -Pcontrib-delta + spark-4.0 correctly pulls delta-spark $V40" ;;
   *) red "FAIL: -Pcontrib-delta + spark-4.0 expected delta-spark 4.0.x, got '${V40:-<none>}'"; exit 1 ;;
@@ -181,7 +222,7 @@ cd "$ROOT"
 # resources. Without `clean`, this gate reports a leak that is really just a stale artifact, and
 # a developer who sees one false FAIL learns to ignore the gate. Build from scratch so what we
 # find in target/classes is exactly what THIS default build produced.
-"$MVNW" -Pspark-4.1 -Djava.version=17 -Dmaven.compiler.source=17 -Dmaven.compiler.target=17 -Dmaven.gitcommitid.skip -pl spark -am clean test-compile -q -DskipTests=true >/dev/null 2>&1
+run_logged maven-test-compile "$MVNW" -Pspark-4.1 -Djava.version=17 -Dmaven.compiler.source=17 -Dmaven.compiler.target=17 -Dmaven.gitcommitid.skip -pl spark -am clean test-compile -q -DskipTests=true
 LEAK_CLASSES="$(find spark/target/classes -path '*comet/contrib*' -name '*.class' 2>/dev/null)"
 if [[ -n "$LEAK_CLASSES" ]]; then
   red "FAIL: default Maven build compiled contrib classes:"
@@ -251,8 +292,9 @@ delta_syms() {
   nm "$1" 2>/dev/null | grep -ciE 'comet_contrib_delta|delta_kernel|deltadvfilter|deltasynthetic' || true
 }
 
-cargo clean -p comet-contrib-delta -p datafusion-comet >/dev/null 2>&1 || true
-cargo build -j 4 -p datafusion-comet >/dev/null 2>&1
+cargo clean -p comet-contrib-delta -p datafusion-comet >"$LOG_DIR/cargo-clean.stdout.log" \
+  2>"$LOG_DIR/cargo-clean.stderr.log" || true
+run_logged cargo-build-default cargo build -j 4 -p datafusion-comet
 LIB_DEFAULT="$(comet_lib)"
 if [[ -z "$LIB_DEFAULT" ]]; then
   red "FAIL: default build produced no libcomet.{so,dylib} under $NATIVE_DIR/target/debug"
@@ -270,7 +312,7 @@ if [[ "$EXT_SYMS" -ne 0 ]]; then
 fi
 green "OK: default libcomet has 0 Delta symbols (size=$SIZE_DEFAULT bytes)"
 
-cargo build -j 4 -p datafusion-comet --features contrib-delta >/dev/null 2>&1
+run_logged cargo-build-contrib cargo build -j 4 -p datafusion-comet --features contrib-delta
 LIB_CONTRIB="$(comet_lib)"
 SIZE_CONTRIB="$(lib_size "$LIB_CONTRIB")"
 if [[ "$SIZE_CONTRIB" -le "$SIZE_DEFAULT" ]]; then