Iceberg 1.11 support for Spark 411, part (3/3): accelerate SparkIncrementalAppendScan on GPU [databricks] (#14883)

Part (3/3) of #14853. Parts (1/3) (#14881) and (2/3) (#14882) are
merged; this PR is now based directly on `main`.

### Description

Iceberg 1.11 split `SparkBatchQueryScan` into a new class hierarchy and
the incremental-append query path (`.option("start-snapshot-id", ...)` +
`.option("end-snapshot-id", ...)`) moved into a brand-new class
`org.apache.iceberg.spark.source.SparkIncrementalAppendScan`
(package-private, 1.11-only, extends `SparkRuntimeFilterableScan`).

Before 1.11 the same query path went through `SparkBatchQueryScan` and
was matched by the existing batch-query `ScanRule` in
`IcebergProviderBase`. Without a rule for the new class, the leaf falls
back to CPU and the `GpuTransitionOverrides` partial-columnar assertion
fires, breaking `test_iceberg_incremental_read` (3 variants: PERFILE /
MULTITHREADED / COALESCING).

Adds:
- `GpuSparkIncrementalAppendScan` in `org.apache.iceberg.spark.source` —
mirrors `GpuSparkBatchQueryScan` since both CPU scans extend
`SparkRuntimeFilterableScan` (a
`SparkPartitioningAwareScan<PartitionScanTask>`). Takes the public
`Scan` type and reaches Iceberg internals only through the root-loadable
`GpuSparkScanAccess` bridge + public `SupportsRuntimeV2Filtering`; it
does **not** reference the package-private `SparkIncrementalAppendScan`
directly, so it works under `extraClassPath` / system-classpath
(#14959).
- `iceberg111x.IcebergProviderImpl` overrides `getScans` to register a
third `ScanRule` for `SparkIncrementalAppendScan` on top of the base
provider's two rules. The CPU class is loaded by string (via
`ShimReflectionUtils.loadClass`) because it is package-private.
- CI: `run_iceberg_tests()` in `jenkins/spark-tests.sh` enables the full
Iceberg integration suite on Spark 4.1 / Iceberg 1.11.0 (nightly).

### Review follow-ups deferred from part (2/3) (#14882)

Addressing the applicable review follow-ups deferred to this final split
PR:
- **Assert the selected shim package, not just `detectedVersion()`.**
Exposes `IcebergProvider.shimPackage()` through `IcebergProviderAccess`;
`iceberg_version_detection_test.py` now also asserts the selected shim
sub-package (e.g. Iceberg 1.11.0 → `iceberg111x`), so a correct
`detectedVersion()` paired with a wrong version→shim mapping now fails.
- **Clarified the version matrix comment** in
`run_iceberg_version_detect_tests()` to note that Spark 4.1 / Iceberg
1.11 is covered by nightly CI rather than pre-merge.

### Checklists

Documentation
- [ ] Updated for new or modified user-facing features or behaviors
- [x] No user-facing change

Testing
- [x] Added or modified tests to cover new code paths
(`iceberg_version_detection_test.py` now asserts the selected shim
package for existing version-detection runs.
`iceberg_test.py::test_iceberg_incremental_read` — 3 variants — covers
the GPU incremental-append path in nightly CI.)
- [ ] Covered by existing tests
- [ ] Not required

Performance
- [ ] Tests ran and results are added in the PR description
- [ ] Issue filed with a link in the PR description
- [x] Not required

---------

Signed-off-by: Chong Gao <res_life@163.com>
Signed-off-by: Chong Gao <chongg@nvidia.com>
Co-authored-by: Chong Gao <res_life@163.com>

Author: res-life

iceberg/iceberg-1-11-x/src/main/scala/com/nvidia/spark/rapids/iceberg/iceberg111x/IcebergProviderImpl.scala

@@ -16,6 +16,48 @@
 
 package com.nvidia.spark.rapids.iceberg.iceberg111x
 
+import scala.reflect.ClassTag
+import scala.util.Try
+
+import com.nvidia.spark.rapids.{GpuScan, ScanMeta, ScanRule, ShimReflectionUtils}
 import com.nvidia.spark.rapids.iceberg.IcebergProviderBase
+import org.apache.iceberg.spark.source.{GpuSparkIncrementalAppendScan, GpuSparkScan}
+
+import org.apache.spark.sql.connector.read.{Scan, SupportsRuntimeV2Filtering}
+
+class IcebergProviderImpl extends IcebergProviderBase {
+
+  /**
+   * Adds a {@code SparkIncrementalAppendScan} rule on top of the base provider's two rules
+   * ({@code SparkBatchQueryScan}, {@code SparkCopyOnWriteScan}). The incremental-append scan
+   * is a 1.11-only class — before 1.11 the same query path went through
+   * {@code SparkBatchQueryScan} and was matched by the base rule. The CPU class is loaded
+   * by string here because it is package-private and not directly referenceable from
+   * outside {@code org.apache.iceberg.spark.source}.
+   */
+  override def getScans: Map[Class[_ <: Scan], ScanRule[_ <: Scan]] = {
+    val cpuIncrementalAppendScanClass = ShimReflectionUtils.loadClass(
+      "org.apache.iceberg.spark.source.SparkIncrementalAppendScan")
+
+    val incrementalRule = new ScanRule[Scan](
+      (a, conf, p, r) => new ScanMeta[Scan](a, conf, p, r) {
+        private lazy val convertedScan: Try[GpuSparkScan] = Try(
+          GpuSparkIncrementalAppendScan.create(a, this.conf, false))
+
+        override def supportsRuntimeFilters: Boolean =
+          a.isInstanceOf[SupportsRuntimeV2Filtering]
+
+        override def tagSelfForGpu(): Unit = {
+          GpuSparkScan.tagForGpu(this, convertedScan)
+        }
+
+        override def convertToGpu(): GpuScan = convertedScan.get
+      },
+      "Iceberg incremental append scan",
+      ClassTag(cpuIncrementalAppendScanClass)
+    )
 
-class IcebergProviderImpl extends IcebergProviderBase
+    super.getScans + (
+      cpuIncrementalAppendScanClass.asSubclass(classOf[Scan]) -> incrementalRule)
+  }
+}

iceberg/iceberg-1-11-x/src/main/scala/org/apache/iceberg/spark/source/GpuSparkIncrementalAppendScan.scala

@@ -0,0 +1,97 @@
+/*
+ * Copyright (c) 2026, NVIDIA CORPORATION.
+ *
+ * Licensed 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.
+ */
+
+package org.apache.iceberg.spark.source
+
+import java.util.Objects
+
+import scala.collection.JavaConverters._
+
+import com.nvidia.spark.rapids.{GpuScan, RapidsConf}
+import org.apache.iceberg.expressions.Expression
+
+import org.apache.spark.sql.connector.expressions.NamedReference
+import org.apache.spark.sql.connector.expressions.filter.Predicate
+import org.apache.spark.sql.connector.read.{Scan, Statistics, SupportsRuntimeV2Filtering}
+
+/**
+ * GPU wrapper for Iceberg 1.11's {@code SparkIncrementalAppendScan} (the
+ * `.option("start-snapshot-id", ...)` / `.option("end-snapshot-id", ...)` read
+ * path). This class was introduced in Iceberg 1.11 — before 1.11 the same
+ * incremental-read path went through {@code SparkBatchQueryScan} (accelerated by
+ * {@link GpuSparkBatchQueryScan}). It mirrors {@link GpuSparkBatchQueryScan}
+ * because both CPU scans extend {@code SparkRuntimeFilterableScan} (a
+ * {@code SparkPartitioningAwareScan<PartitionScanTask>}).
+ *
+ * <p>Takes the public {@code Scan} type and reaches Iceberg internals only through
+ * the root-loadable {@link GpuSparkScanAccess} bridge and public Spark interfaces
+ * ({@code SupportsRuntimeV2Filtering}). It must NOT reference the package-private
+ * {@code SparkIncrementalAppendScan} directly: under {@code extraClassPath}
+ * (system-classpath) the Iceberg classes load in the app classloader while this
+ * shimmed class loads in Spark's MutableURLClassLoader, so any same-package access
+ * would fail with {@code IllegalAccessError} (see issue #14959).
+ */
+class GpuSparkIncrementalAppendScan(
+    override val cpuScan: Scan,
+    override val rapidsConf: RapidsConf,
+    override val queryUsesInputFile: Boolean) extends
+  GpuSparkPartitioningAwareScan[org.apache.iceberg.PartitionScanTask](
+    cpuScan, rapidsConf, queryUsesInputFile)
+  with SupportsRuntimeV2Filtering {
+
+  private val runtimeFilterExpressions: List[Expression] =
+    GpuSparkScanAccess.runtimeFilterExpressions(cpuScan)
+    .asScala
+    .toList
+
+  private def runtimeFilterScan: SupportsRuntimeV2Filtering =
+    cpuScan.asInstanceOf[SupportsRuntimeV2Filtering]
+
+  override def filterAttributes(): Array[NamedReference] = runtimeFilterScan.filterAttributes()
+
+  override def filter(predicates: Array[Predicate]): Unit = runtimeFilterScan.filter(predicates)
+
+  override def estimateStatistics(): Statistics = GpuSparkScanAccess.estimateStatistics(cpuScan)
+
+  override def equals(obj: Any): Boolean = obj match {
+    case that: GpuSparkIncrementalAppendScan =>
+      this.cpuScan == that.cpuScan &&
+        this.queryUsesInputFile == that.queryUsesInputFile
+    case _ => false
+  }
+
+  override def hashCode(): Int =
+    Objects.hash(cpuScan, Boolean.box(queryUsesInputFile))
+
+  override def toString: String =
+    s"GpuSparkIncrementalAppendScan(table=${GpuSparkScanAccess.table(cpuScan)}, " +
+      s"branch=${GpuSparkScanAccess.branch(cpuScan)}, " +
+      s"type=${GpuSparkScanAccess.expectedSchema(cpuScan).asStruct()}, " +
+      s"filters=${GpuSparkScanAccess.filterExpressions(cpuScan)}, " +
+      s"runtimeFilters=$runtimeFilterExpressions, " +
+      s"caseSensitive=${GpuSparkScanAccess.caseSensitive(cpuScan)}, " +
+      s"queryUseInputFile=$queryUsesInputFile)"
+
+  override def withInputFile(): GpuScan =
+    new GpuSparkIncrementalAppendScan(cpuScan, rapidsConf, true)
+}
+
+object GpuSparkIncrementalAppendScan {
+  /** Java-callable factory used by {@code IcebergProviderImpl}. Takes the public
+   *  {@code Scan} type — never the package-private {@code SparkIncrementalAppendScan}. */
+  def create(cpuScan: Scan, rapidsConf: RapidsConf, queryUsesInputFile: Boolean): GpuSparkScan =
+    new GpuSparkIncrementalAppendScan(cpuScan, rapidsConf, queryUsesInputFile)
+}

integration_tests/src/main/python/iceberg/iceberg_version_detection_test.py

@@ -32,11 +32,24 @@ def test_iceberg_version_detection():
     if expected is None:
         pytest.skip("EXPECTED_ICEBERG_VERSION env var not set")
 
+    # Shim sub-package selected per Iceberg major.minor, e.g. 1.11.x -> iceberg111x.
+    # Mirrors IcebergProbeImpl.icebergVersionToShim.
+    major_minor = ".".join(expected.split(".")[:2])
+    expected_shim_package = \
+        "com.nvidia.spark.rapids.iceberg.iceberg{}x".format(major_minor.replace(".", ""))
+
     def check(spark):
-        jvm = spark.sparkContext._jvm
-        actual = jvm.com.nvidia.spark.rapids.iceberg.IcebergProviderAccess.detectedVersion()
+        access = spark.sparkContext._jvm.com.nvidia.spark.rapids.iceberg.IcebergProviderAccess
+        actual = access.detectedVersion()
         assert actual == expected, \
             "Iceberg version detection mismatch: expected '{}' on Spark {}, got '{}'".format(
                 expected, spark_version(), actual)
+        # Assert the shim package too: a correct detectedVersion() paired with a wrong
+        # version -> shim mapping (e.g. the 1.11 -> iceberg111x row) would otherwise
+        # still pass.
+        actual_shim_package = access.shimPackage()
+        assert actual_shim_package == expected_shim_package, \
+            "Iceberg shim package mismatch for {} on Spark {}: expected '{}', got '{}'".format(
+                expected, spark_version(), expected_shim_package, actual_shim_package)
 
     with_gpu_session(check)

jenkins/spark-premerge-build.sh

@@ -167,11 +167,10 @@ run_iceberg_version_detect_tests() {
         return 0
     fi
 
-    # Supported Iceberg versions per Spark version — must stay in sync with
-    # run_iceberg_tests() in spark-tests.sh. Note: the Spark 4.1 -> 1.11.0 row is
-    # listed here, but ci_scala213() currently uses SPARK_VER=4.0.1 so the 4.1 branch
-    # is not exercised in pre-merge CI. The full 4.1 integration suite is added in the
-    # stacked follow-up PR; until then, the 1.11.0 commit-ID mapping is nightly-only.
+    # Supported Iceberg versions per Spark version. The 3.5.x / 4.0.x rows mirror
+    # run_iceberg_tests() in spark-tests.sh. The Spark 4.1 -> 1.11.0 row is kept here
+    # for callers that explicitly test Spark 4.1, while the regular pre-merge job
+    # below runs on Spark 4.0.1. Spark 4.1 is covered by nightly run_iceberg_tests().
     local iceberg_versions
     if [[ "$iceberg_spark_ver" == "4.1" ]]; then
         iceberg_versions="1.11.0"
@@ -269,6 +268,7 @@ ci_scala213() {
     # Moved out of spark-tests.sh DEFAULT mode where JDK 8 causes
     # UnsupportedClassVersionError for Iceberg 1.9+ runtime JARs.
     run_iceberg_version_detect_tests $SPARK_VER 2.13
+
 }
 
 prepare_spark() {

jenkins/spark-tests.sh

@@ -268,7 +268,8 @@ run_iceberg_tests() {
   # get the patch version of Spark
   SPARK_PATCH_VER=$(echo "$SPARK_VER" | cut -d. -f3)
 
-  if [[ "$ICEBERG_SPARK_VER" != "3.5" && "$ICEBERG_SPARK_VER" != "4.0" ]]; then
+  if [[ "$ICEBERG_SPARK_VER" != "3.5" && "$ICEBERG_SPARK_VER" != "4.0" \
+        && "$ICEBERG_SPARK_VER" != "4.1" ]]; then
     echo "!!!! Skipping Iceberg tests. GPU acceleration of Iceberg is not supported on $ICEBERG_SPARK_VER"
     return 0
   fi
@@ -277,8 +278,15 @@ run_iceberg_tests() {
   # Spark 3.5.0-3.5.3 -> Iceberg 1.6.1
   # Spark 3.5.4+       -> Iceberg 1.9.2, 1.10.1
   # Spark 4.0.x        -> Iceberg 1.10.1
+  # Spark 4.1.x        -> Iceberg 1.11.0
   local supported_versions
-  if [[ "$ICEBERG_SPARK_VER" == "4.0" ]]; then
+  if [[ "$ICEBERG_SPARK_VER" == "4.1" ]]; then
+    if [[ "$SCALA_BINARY_VER" != "2.13" ]]; then
+      echo "!!!! Skipping Iceberg tests. Spark 4.1 Iceberg tests require Scala 2.13"
+      return 0
+    fi
+    supported_versions="1.11.0"
+  elif [[ "$ICEBERG_SPARK_VER" == "4.0" ]]; then
     if [[ "$SCALA_BINARY_VER" != "2.13" ]]; then
       echo "!!!! Skipping Iceberg tests. Spark 4.0 Iceberg tests require Scala 2.13"
       return 0
@@ -302,7 +310,9 @@ run_iceberg_tests() {
     echo "Using user-specified ICEBERG_VERSIONS=$ICEBERG_VERSIONS"
   else
     # Default: test one representative version per Spark patch range
-    if [[ "$ICEBERG_SPARK_VER" == "4.0" ]]; then
+    if [[ "$ICEBERG_SPARK_VER" == "4.1" ]]; then
+      ICEBERG_VERSIONS="1.11.0"
+    elif [[ "$ICEBERG_SPARK_VER" == "4.0" ]]; then
       ICEBERG_VERSIONS="1.10.1"
     elif [[ "$SPARK_PATCH_VER" -le 3 ]]; then
       ICEBERG_VERSIONS="1.6.1"

sql-plugin/src/main/java/com/nvidia/spark/rapids/iceberg/IcebergProviderAccess.java

@@ -24,4 +24,13 @@ private IcebergProviderAccess() {
   public static String detectedVersion() {
     return IcebergProvider$.MODULE$.detectedVersion();
   }
+
+  /**
+   * Exposes the selected shim package to the Py4J integration test in
+   * {@code iceberg_version_detection_test.py}, which verifies that the detected Iceberg version
+   * maps to the expected shim implementation.
+   */
+  public static String shimPackage() {
+    return IcebergProvider$.MODULE$.shimPackage();
+  }
 }