Iceberg 1.11 support for Spark 411, part (1/3): extract version-divergent scan APIs behind a shim (#14881)

Stacked work for #14853 (1/3) — common-code preparation for adding
iceberg-1-11-x.

### Depends on
* #14866 Rebase on this PR
after #14866 merged.

### Description

Refactors `iceberg/common` so the `SparkScan` / `SparkCopyOnWriteScan` /
`SparkBatch` / `DataWriteResult` APIs that diverge between Iceberg
1.10.x and 1.11.x are hidden behind a small interface, with
per-Iceberg-version implementations in `iceberg-1-6-x` / `iceberg-1-9-x`
/ `iceberg-1-10-x`. **No behavior change for the Iceberg versions this
PR ships**; sets the stage for the follow-up PR that adds
`iceberg-1-11-x`.

Common changes:
- `GpuSparkCopyOnWriteScan` → renamed to `GpuSparkCopyOnWriteScanBase`
(abstract). The runtime-filter trait + `filter` method live in a
per-version concrete subclass (1.6/1.9/1.10 mix in
`SupportsRuntimeFiltering` with `filter(Filter[])`; 1.11 will mix in
`SupportsRuntimeV2Filtering` with `filter(Predicate[])`).
- `GpuSparkScan`: rewrite `hasNestedType` via Spark's `readSchema()` +
Spark types so it no longer depends on the 1.10-only
`cpuScan.expectedSchema()`. Dispatch `SparkCopyOnWriteScan` construction
through the new `ShimUtils.newCopyOnWriteScan` factory.
- `GpuSparkBatchQueryScan.toString` uses `cpuScan.description()`
(available in both 1.10 and 1.11) instead of `branch` / `expectedSchema`
/ `filterExpressions` (1.11 removed these).
- `GpuSparkBatchQueryScan.runtimeFilterExpressions` reflective
field-read tolerates both the 1.10 name (`runtimeFilterExpressions`) and
the 1.11 name (`runtimeFilters`).
- `GpuSparkBatch`: same tolerance for `expectedSchema` (1.10) vs
`projection` (1.11).
- `GpuSparkWrite`: type-annotate `new Array[DataFile](0)` so Scala 2.13
doesn't infer `Array[Nothing]` under 1.11's wildcarded
`DataWriteResult.dataFiles()`.
- `IcebergShimUtils` / `ShimUtils`: add `newCopyOnWriteScan(Scan,
RapidsConf, Boolean): GpuScan` factory. The parameter is Spark's public
`Scan` because Iceberg's `SparkCopyOnWriteScan` is package-private —
cross-package callers cannot reference it directly.

Per-Iceberg-version module changes (1.6 / 1.9 / 1.10, all identical for
the V1 path):
- New `GpuSparkCopyOnWriteScan` in `org.apache.iceberg.spark.source` (so
it can reference the package-private `SparkCopyOnWriteScan`). Companion
object exposes `create(Scan, ...): GpuScan` for cross-package callers.
- `ShimUtilsImpl.java` implements `newCopyOnWriteScan` via
`GpuSparkCopyOnWriteScan.create`.

The two `try/catch` field-name fallbacks (in `GpuSparkBatchQueryScan`
and `GpuSparkBatch`) are tactical and will be pushed behind proper
per-version `IcebergShimUtils` methods in a later cleanup PR.

### Checklists

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

Testing
- [ ] Added or modified tests to cover new code paths
- [x] Covered by existing tests
(3.5.x + 4.0.x iceberg integration tests in
\`integration_tests/src/main/python/iceberg/\` — exercises the new
dispatch path with no behavior change vs. before this PR.)
- [ ] 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>
Co-authored-by: Chong Gao <res_life@163.com>

Author: res-life

iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/IcebergShimUtils.java

@@ -18,6 +18,7 @@
 
 import com.nvidia.spark.rapids.GpuMetric;
 import com.nvidia.spark.rapids.NoopMetric$;
+import com.nvidia.spark.rapids.RapidsConf;
 import com.nvidia.spark.rapids.fileio.iceberg.IcebergInputFile;
 import org.apache.hadoop.fs.Path;
 import org.apache.iceberg.ContentFile;
@@ -28,6 +29,8 @@
 import org.apache.iceberg.parquet.GpuParquetIO;
 import org.apache.iceberg.shaded.org.apache.parquet.ParquetReadOptions;
 import org.apache.iceberg.shaded.org.apache.parquet.hadoop.ParquetFileReader;
+import org.apache.iceberg.spark.source.GpuSparkScan;
+import org.apache.spark.sql.connector.read.Scan;
 import scala.Option;
 
 import java.io.IOException;
@@ -98,4 +101,18 @@ default ParquetFileReader openParquetReader(
         missCounter.$plus$eq(1L);
         return ParquetFileReader.open(GpuParquetIO.file(inputFile.getDelegate()), options);
     }
+
+    /**
+     * Constructs the version-appropriate {@code GpuSparkCopyOnWriteScan} subclass.
+     *
+     * <p>Iceberg 1.6.x, 1.9.x, and 1.10.x have {@code SparkCopyOnWriteScan}
+     * implementing {@code SupportsRuntimeFiltering} with {@code filter(Filter[])};
+     * Iceberg 1.11.x switched to {@code SupportsRuntimeV2Filtering} with
+     * {@code filter(Predicate[])}. The concrete class therefore differs per Iceberg
+     * version and is constructed here rather than directly in common code.
+     */
+    GpuSparkScan newCopyOnWriteScan(
+            Scan cpuScan,
+            RapidsConf rapidsConf,
+            boolean queryUsesInputFile);
 }

iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/ShimUtils.java

@@ -17,6 +17,7 @@
 package com.nvidia.spark.rapids.iceberg;
 
 import com.nvidia.spark.rapids.GpuMetric;
+import com.nvidia.spark.rapids.RapidsConf;
 import com.nvidia.spark.rapids.ShimLoader;
 import com.nvidia.spark.rapids.fileio.iceberg.IcebergInputFile;
 
@@ -28,6 +29,8 @@
 import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.shaded.org.apache.parquet.ParquetReadOptions;
 import org.apache.iceberg.shaded.org.apache.parquet.hadoop.ParquetFileReader;
+import org.apache.iceberg.spark.source.GpuSparkScan;
+import org.apache.spark.sql.connector.read.Scan;
 
 import java.io.IOException;
 import java.util.Map;
@@ -71,4 +74,11 @@ public static ParquetFileReader openParquetReader(
             scala.collection.immutable.Map<String, GpuMetric> metrics) throws IOException {
         return IMPL.openParquetReader(inputFile, filePath, options, metrics);
     }
+
+    public static GpuSparkScan newCopyOnWriteScan(
+            Scan cpuScan,
+            RapidsConf rapidsConf,
+            boolean queryUsesInputFile) {
+        return IMPL.newCopyOnWriteScan(cpuScan, rapidsConf, queryUsesInputFile);
+    }
 }

iceberg/common/src/main/java/org/apache/iceberg/spark/source/GpuSparkScanAccess.java

@@ -17,6 +17,9 @@
 package org.apache.iceberg.spark.source;
 
 import java.lang.reflect.Field;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.Collections;
 import java.util.List;
 
 import org.apache.iceberg.BaseMetadataTable;
@@ -60,31 +63,40 @@ public static boolean isMetadataScan(Scan scan) {
   }
 
   public static SparkReadConf readConf(Scan scan) {
-    return readField(sparkScan(scan), "readConf", SparkReadConf.class);
+    return readField(sparkScan(scan), SparkReadConf.class, "readConf");
   }
 
   public static Table table(Scan scan) {
     return sparkScan(scan).table();
   }
 
   public static String branch(Scan scan) {
-    return sparkScan(scan).branch();
+    // Iceberg 1.10.x and earlier: protected method SparkScan.branch(). Iceberg 1.11.x
+    // removed it entirely; return null for display purposes.
+    return invokeMethod(sparkScan(scan), String.class, "branch");
   }
 
   public static boolean caseSensitive(Scan scan) {
     return sparkScan(scan).caseSensitive();
   }
 
   public static Schema expectedSchema(Scan scan) {
-    return sparkScan(scan).expectedSchema();
+    // Iceberg 1.10.x: SparkScan.expectedSchema(); Iceberg 1.11.x renamed it to
+    // SparkScan.projection().
+    return invokeMethod(sparkScan(scan), Schema.class, "expectedSchema", "projection");
   }
 
   public static Statistics estimateStatistics(Scan scan) {
     return sparkScan(scan).estimateStatistics();
   }
 
+  @SuppressWarnings("unchecked")
   public static List<Expression> filterExpressions(Scan scan) {
-    return sparkScan(scan).filterExpressions();
+    // Iceberg 1.10.x: SparkScan.filterExpressions(); Iceberg 1.11.x renamed it to
+    // SparkScan.filters().
+    List<Expression> r = (List<Expression>)
+        invokeMethod(sparkScan(scan), List.class, "filterExpressions", "filters");
+    return r != null ? r : Collections.emptyList();
   }
 
   public static Types.StructType groupingKeyType(Scan scan) {
@@ -101,11 +113,17 @@ public static Batch toBatch(Scan scan) {
 
   @SuppressWarnings("unchecked")
   public static List<Expression> runtimeFilterExpressions(Scan scan) {
-    return (List<Expression>) readField(scan, "runtimeFilterExpressions", List.class);
+    // Iceberg 1.6.x / 1.9.x / 1.10.x: field "runtimeFilterExpressions" on
+    // SparkBatchQueryScan. Iceberg 1.11.x: field "runtimeFilters" on the new parent
+    // class SparkRuntimeFilterableScan.
+    return (List<Expression>) readField(scan, List.class,
+        "runtimeFilterExpressions", "runtimeFilters");
   }
 
   public static Schema expectedSchema(Batch batch) {
-    return readField(batch, "expectedSchema", Schema.class);
+    // Iceberg 1.10.x: field "expectedSchema" on SparkBatch.
+    // Iceberg 1.11.x: renamed to "projection".
+    return readField(batch, Schema.class, "expectedSchema", "projection");
   }
 
   public static Table table(InputPartition partition) {
@@ -128,26 +146,66 @@ private static SparkInputPartition sparkInputPartition(InputPartition partition)
     return (SparkInputPartition) partition;
   }
 
-  private static <T> T readField(Object target, String fieldName, Class<T> fieldType) {
+  /**
+   * Read the first existing field from {@code fieldNames} on {@code target} (or its
+   * superclasses). Used when a field was renamed across Iceberg versions — list the
+   * candidate names in priority order. Throws if none of the names exist.
+   */
+  private static <T> T readField(Object target, Class<T> fieldType, String... fieldNames) {
+    Field field = findField(target.getClass(), fieldNames);
+    if (field == null) {
+      throw new IllegalStateException(
+          "None of fields " + String.join(",", fieldNames)
+              + " exist on " + target.getClass().getName());
+    }
     try {
-      Field field = findField(target.getClass(), fieldName);
       field.setAccessible(true);
       return fieldType.cast(field.get(target));
     } catch (IllegalAccessException e) {
       throw new IllegalStateException(
-          "Unable to read " + fieldName + " from " + target.getClass().getName(), e);
+          "Unable to read " + field.getName() + " from " + target.getClass().getName(), e);
     }
   }
 
-  private static Field findField(Class<?> targetClass, String fieldName) {
+  private static Field findField(Class<?> targetClass, String... fieldNames) {
     Class<?> current = targetClass;
     while (current != null) {
-      try {
-        return current.getDeclaredField(fieldName);
-      } catch (NoSuchFieldException e) {
-        current = current.getSuperclass();
+      for (String fieldName : fieldNames) {
+        try {
+          return current.getDeclaredField(fieldName);
+        } catch (NoSuchFieldException ignore) {
+          // try the next name (or the superclass)
+        }
+      }
+      current = current.getSuperclass();
+    }
+    return null;
+  }
+
+  /**
+   * Invoke the first existing zero-arg method from {@code methodNames} on
+   * {@code target} (or its superclasses). Used when a protected method was renamed or
+   * removed across Iceberg versions — list the candidate names in priority order.
+   * Returns {@code null} if none of the names exist (caller decides the fallback).
+   */
+  private static <T> T invokeMethod(Object target, Class<T> returnType, String... methodNames) {
+    Class<?> current = target.getClass();
+    while (current != null) {
+      for (String methodName : methodNames) {
+        try {
+          Method m = current.getDeclaredMethod(methodName);
+          m.setAccessible(true);
+          Object result = m.invoke(target);
+          return result == null ? null : returnType.cast(result);
+        } catch (NoSuchMethodException ignore) {
+          // try the next name (or the superclass)
+        } catch (IllegalAccessException | InvocationTargetException e) {
+          throw new IllegalStateException(
+              "Unable to invoke " + methodName + " on " + target.getClass().getName(), e);
+        }
       }
+      current = current.getSuperclass();
     }
-    throw new IllegalStateException("No field " + fieldName + " in " + targetClass.getName());
+    return null;
   }
 }

…ark/source/GpuSparkCopyOnWriteScan.scala …source/GpuSparkCopyOnWriteScanBase.scalaiceberg/common/src/main/scala/org/apache/iceberg/spark/source/GpuSparkCopyOnWriteScan.scala renamed to iceberg/common/src/main/scala/org/apache/iceberg/spark/source/GpuSparkCopyOnWriteScanBase.scala iceberg/common/src/main/scala/org/apache/iceberg/spark/source/GpuSparkCopyOnWriteScan.scala renamed to iceberg/common/src/main/scala/org/apache/iceberg/spark/source/GpuSparkCopyOnWriteScanBase.scala

@@ -18,30 +18,32 @@ package org.apache.iceberg.spark.source
 
 import java.util.Objects
 
-import com.nvidia.spark.rapids.{GpuScan, RapidsConf}
+import com.nvidia.spark.rapids.RapidsConf
 import org.apache.iceberg.FileScanTask
 
-import org.apache.spark.sql.connector.expressions.NamedReference
-import org.apache.spark.sql.connector.read.{Scan, Statistics, SupportsRuntimeFiltering}
-import org.apache.spark.sql.sources.Filter
+import org.apache.spark.sql.connector.read.{Scan, Statistics}
 
-class GpuSparkCopyOnWriteScan(
+/**
+ * Version-agnostic base for the GPU copy-on-write scan. Iceberg 1.6.x, 1.9.x,
+ * and 1.10.x have {@code SparkCopyOnWriteScan} implementing
+ * {@code SupportsRuntimeFiltering} with {@code filter(Filter[])}; Iceberg 1.11.x
+ * switched to {@code SupportsRuntimeV2Filtering} with {@code filter(Predicate[])}.
+ * The per-Iceberg-version concrete subclass lives in {@code iceberg-1-6-x} /
+ * {@code iceberg-1-9-x} / {@code iceberg-1-10-x} (and {@code iceberg-1-11-x}
+ * once it lands) and mixes in the matching Spark runtime-filter trait + delegates
+ * {@code filter} to the matching Iceberg API.
+ */
+abstract class GpuSparkCopyOnWriteScanBase(
     override val cpuScan: Scan,
     override val rapidsConf: RapidsConf,
     override val queryUsesInputFile: Boolean) extends
-  GpuSparkPartitioningAwareScan[FileScanTask](cpuScan, rapidsConf, queryUsesInputFile)
-  with SupportsRuntimeFiltering {
-
-  private def runtimeFilterScan: SupportsRuntimeFiltering =
-    cpuScan.asInstanceOf[SupportsRuntimeFiltering]
-
-  override def filterAttributes(): Array[NamedReference] = runtimeFilterScan.filterAttributes()
+  GpuSparkPartitioningAwareScan[FileScanTask](cpuScan, rapidsConf, queryUsesInputFile) {
 
   override def estimateStatistics(): Statistics = GpuSparkScanAccess.estimateStatistics(cpuScan)
 
   override def equals(obj: Any): Boolean = {
     obj match {
-      case that: GpuSparkCopyOnWriteScan =>
+      case that: GpuSparkCopyOnWriteScanBase =>
         this.cpuScan == that.cpuScan &&
           this.queryUsesInputFile == that.queryUsesInputFile
       case _ => false
@@ -60,11 +62,4 @@ class GpuSparkCopyOnWriteScan(
       s"caseSensitive=${GpuSparkScanAccess.caseSensitive(cpuScan)}, " +
       s"queryUseInputFile=$queryUsesInputFile)"
   }
-
-  /** Create a version of this scan with input file name support */
-  override def withInputFile(): GpuScan = {
-    new GpuSparkCopyOnWriteScan(cpuScan, rapidsConf, true)
-  }
-
-  override def filter(filters: Array[Filter]): Unit = runtimeFilterScan.filter(filters)
 }

iceberg/common/src/main/scala/org/apache/iceberg/spark/source/GpuSparkScan.scala

@@ -20,6 +20,7 @@ import scala.collection.JavaConverters._
 import scala.util.{Failure, Success, Try}
 
 import com.nvidia.spark.rapids._
+import com.nvidia.spark.rapids.iceberg.ShimUtils
 import org.apache.iceberg.ScanTaskGroup
 import org.apache.iceberg.spark.GpuSparkReadConf
 import org.apache.iceberg.types.Types
@@ -81,7 +82,7 @@ object GpuSparkScan {
       if (GpuSparkScanAccess.isBatchQueryScan(cpuScan)) {
         new GpuSparkBatchQueryScan(cpuScan, rapidsConf, false)
       } else if (GpuSparkScanAccess.isCopyOnWriteScan(cpuScan)) {
-        new GpuSparkCopyOnWriteScan(cpuScan, rapidsConf, false)
+        ShimUtils.newCopyOnWriteScan(cpuScan, rapidsConf, false)
       } else {
         throw new IllegalArgumentException(
           s"Currently iceberg support only supports batch query scan and copy-on-write scan, " +

iceberg/iceberg-1-10-x/src/main/java/com/nvidia/spark/rapids/iceberg/iceberg110x/ShimUtilsImpl.java

@@ -17,6 +17,7 @@
 package com.nvidia.spark.rapids.iceberg.iceberg110x;
 
 import com.nvidia.spark.rapids.GpuMetric;
+import com.nvidia.spark.rapids.RapidsConf;
 import com.nvidia.spark.rapids.fileio.iceberg.IcebergInputFile;
 import com.nvidia.spark.rapids.iceberg.IcebergShimUtils;
 import org.apache.hadoop.fs.Path;
@@ -27,8 +28,11 @@
 import org.apache.iceberg.shaded.org.apache.parquet.ParquetReadOptions;
 import org.apache.iceberg.shaded.org.apache.parquet.hadoop.ParquetFileReader;
 import org.apache.iceberg.spark.SparkUtil;
+import org.apache.iceberg.spark.source.GpuSparkCopyOnWriteScan;
+import org.apache.iceberg.spark.source.GpuSparkScan;
 import org.apache.iceberg.types.Types;
 import org.apache.iceberg.util.PartitionUtil;
+import org.apache.spark.sql.connector.read.Scan;
 
 import java.io.IOException;
 import java.util.Collections;
@@ -74,4 +78,12 @@ public ParquetFileReader openParquetReader(
             scala.collection.immutable.Map<String, GpuMetric> metrics) throws IOException {
         return GpuParquetIOShim.openReader(inputFile, filePath, options, metrics);
     }
+
+    @Override
+    public GpuSparkScan newCopyOnWriteScan(
+            Scan cpuScan,
+            RapidsConf rapidsConf,
+            boolean queryUsesInputFile) {
+        return GpuSparkCopyOnWriteScan.create(cpuScan, rapidsConf, queryUsesInputFile);
+    }
 }

iceberg/iceberg-1-10-x/src/main/scala/org/apache/iceberg/spark/source/GpuSparkCopyOnWriteScan.scala

@@ -0,0 +1,52 @@
+/*
+ * 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 com.nvidia.spark.rapids.{GpuScan, RapidsConf}
+
+import org.apache.spark.sql.connector.expressions.NamedReference
+import org.apache.spark.sql.connector.read.{Scan, SupportsRuntimeFiltering}
+import org.apache.spark.sql.sources.Filter
+
+/**
+ * Iceberg 1.10.x copy-on-write scan: {@code SupportsRuntimeFiltering} with
+ * {@code filter(Array[Filter])}.
+ */
+class GpuSparkCopyOnWriteScan(
+    cpuScanArg: Scan,
+    rapidsConfArg: RapidsConf,
+    queryUsesInputFileArg: Boolean)
+  extends GpuSparkCopyOnWriteScanBase(cpuScanArg, rapidsConfArg, queryUsesInputFileArg)
+  with SupportsRuntimeFiltering {
+
+  private def runtimeFilterScan: SupportsRuntimeFiltering =
+    cpuScan.asInstanceOf[SupportsRuntimeFiltering]
+
+  override def filterAttributes(): Array[NamedReference] = runtimeFilterScan.filterAttributes()
+
+  override def filter(filters: Array[Filter]): Unit = runtimeFilterScan.filter(filters)
+
+  override def withInputFile(): GpuScan =
+    new GpuSparkCopyOnWriteScan(cpuScan, rapidsConf, true)
+}
+
+object GpuSparkCopyOnWriteScan {
+  /** Java-callable factory used by {@code ShimUtilsImpl.newCopyOnWriteScan}. */
+  def create(cpuScan: Scan, rapidsConf: RapidsConf, queryUsesInputFile: Boolean)
+      : GpuSparkScan =
+    new GpuSparkCopyOnWriteScan(cpuScan, rapidsConf, queryUsesInputFile)
+}