Finalize Iceberg root and shim class placement [reduced-it]

dist/build/package-parallel-worlds.py

@@ -204,6 +204,10 @@ def root_safe_module_class_members(classifier):
 root_safe_modules = read_patterns(os.sep.join([dist_dir, 'root-safe-module-classes.txt']))
 from_single_shim_or_each = from_single_shim + from_each
 iceberg_audit_runtimes = {}
+# Classifiers are processed newest-first. Let older classifiers contribute
+# conditional root-safe classes that are absent from newer classifiers, without
+# overwriting a newer implementation of the same class path.
+promoted_root_safe_members = set()
 
 for bv in buildver_list:
     classifier = 'spark' + bv
@@ -225,7 +229,7 @@ def root_safe_module_class_members(classifier):
                 # IMPORTANT unconditional extract from the highest Spark version to the top
                 if bv == buildver_list[0] and art == 'sql-plugin-api':
                     zip_handle.extractall(path=top_dist_jar_dir)
-                if bv == buildver_list[0] and art == 'aggregator':
+                if art == 'aggregator':
                     namelist = zip_handle.namelist()
                     namelist_set = set(namelist)
                     root_safe_members = root_safe_module_class_members(classifier)
@@ -234,9 +238,13 @@ def root_safe_module_class_members(classifier):
                         raise Exception(
                             "root-safe module classes missing from aggregator: %s" %
                             ", ".join(missing_members))
+                    new_root_safe_members = (
+                        root_safe_members - promoted_root_safe_members)
                     zip_handle.extractall(
                         path=top_dist_jar_dir,
-                        members=[name for name in namelist if name in root_safe_members])
+                        members=[name for name in namelist
+                                 if name in new_root_safe_members])
+                    promoted_root_safe_members.update(root_safe_members)
                 # TODO deprecate
                 namelist = zip_handle.namelist()
                 glob_list = from_single_shim_or_each if bv == buildver_list[0] else from_each

dist/keep-in-spark-shim-dirs.txt

@@ -53,3 +53,12 @@ org/apache/spark/sql/rapids/execution/GpuBroadcastExchangeExec*.class
 # exchange execs. Keep the rule and helpers in the selected shim loader.
 com/nvidia/spark/rapids/GpuTransitionOverrides.class
 com/nvidia/spark/rapids/GpuTransitionOverrides$$*.class
+
+# Iceberg implementation classes compile against Spark- and Iceberg-version-specific
+# APIs. Root-selected Iceberg classes are excluded by binary-dedupe.sh; keep every
+# other Iceberg class in the selected shim loader instead of spark-shared.
+com/nvidia/spark/rapids/*Iceberg*.class
+com/nvidia/spark/rapids/fileio/iceberg/*.class
+com/nvidia/spark/rapids/iceberg/*.class
+org/apache/iceberg/*.class
+org/apache/spark/sql/rapids/*Iceberg*.class

dist/scripts/binary-dedupe.sh

@@ -182,6 +182,12 @@ function filter_keep_in_spark_shim_dirs() {
     path_without_leading_slash="${class_resource#/}"
     class_file="${path_without_leading_slash#*/}"
     if keep_in_spark_shim_dirs "$class_file"; then
+      # Root-layout classes were explicitly selected by the packager and must
+      # remain eligible for de-duplication so the later root promotion can
+      # remove their shim copies.
+      if [[ -f "./parallel-world/$class_file" ]]; then
+        echo "$class_resource"
+      fi
       continue
     fi
     echo "$class_resource"

dist/scripts/build-unshim-parallel-world.py

@@ -202,6 +202,10 @@ def copy_and_extract_jars(
     cache_root = target_dir / "unshim-parallel-world-cache"
     sorted_buildvers = sorted(buildvers, reverse=True)
     root_buildver = sorted_buildvers[0]
+    # Classifiers are processed newest-first. Let older classifiers contribute
+    # conditional root-safe classes that are absent from newer classifiers,
+    # without overwriting a newer implementation of the same class path.
+    promoted_root_safe_members = set()
 
     for buildver in sorted_buildvers:
         classifier = "spark%s" % buildver
@@ -216,7 +220,7 @@ def copy_and_extract_jars(
             link_tree_contents(contents_dir, parallel_world / classifier)
             if buildver == root_buildver and artifact == "sql-plugin-api":
                 link_tree_contents(contents_dir, parallel_world)
-            if buildver == root_buildver and artifact == "aggregator":
+            if artifact == "aggregator":
                 root_safe_members = root_safe_module_class_members(
                     base_dir,
                     scala_binary_version,
@@ -228,7 +232,11 @@ def copy_and_extract_jars(
                     raise RuntimeError(
                         "root-safe module classes missing from aggregator: %s" %
                         ", ".join(missing_members))
-                link_members(contents_dir, parallel_world, sorted(root_safe_members))
+                new_root_safe_members = (
+                    root_safe_members - promoted_root_safe_members)
+                link_members(contents_dir, parallel_world,
+                             sorted(new_root_safe_members))
+                promoted_root_safe_members.update(root_safe_members)
 
             patterns = from_each
             if buildver == root_buildver:

dist/scripts/tests/test_root_safe_provider_selection.py

@@ -0,0 +1,216 @@
+# 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.
+
+import importlib.util
+import os
+from pathlib import Path
+import subprocess
+import tempfile
+import unittest
+import zipfile
+
+
+DIST_DIR = Path(__file__).resolve().parents[2]
+STANDARD_ASSEMBLER = DIST_DIR / "build" / "package-parallel-worlds.py"
+FAST_ASSEMBLER = DIST_DIR / "scripts" / "build-unshim-parallel-world.py"
+BINARY_DEDUPE = DIST_DIR / "scripts" / "binary-dedupe.sh"
+
+SHARED = "org/apache/iceberg/Shared.class"
+NEWER_ONLY = "org/apache/iceberg/NewerOnly.class"
+OLDER_ONLY = "org/apache/iceberg/OlderOnly.class"
+NEWER_IMPL = "org/apache/iceberg/NewerImpl.class"
+OLDER_IMPL = "org/apache/iceberg/OlderImpl.class"
+
+
+def write_jar(path, entries):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with zipfile.ZipFile(path, "w") as jar:
+        for name, contents in entries.items():
+            jar.writestr(name, contents)
+
+
+def artifact_path(base_dir, artifact, buildver):
+    artifact_id = "rapids-4-spark-%s_2.13" % artifact
+    return (base_dir / artifact / "target" / ("spark%s" % buildver) /
+            ("%s-1.0-spark%s.jar" % (artifact_id, buildver)))
+
+
+def create_artifacts(base_dir):
+    for buildver in ("353", "413"):
+        write_jar(artifact_path(base_dir, "sql-plugin-api", buildver), {})
+
+    write_jar(artifact_path(base_dir, "iceberg-common", "413"), {
+        SHARED: b"module-shared-413",
+        NEWER_ONLY: b"module-newer-only",
+    })
+    write_jar(artifact_path(base_dir, "aggregator", "413"), {
+        SHARED: b"aggregator-shared-413",
+        NEWER_ONLY: b"aggregator-newer-only",
+        NEWER_IMPL: b"aggregator-newer-impl",
+    })
+    write_jar(artifact_path(base_dir, "iceberg-common", "353"), {
+        SHARED: b"module-shared-353",
+        OLDER_ONLY: b"module-older-only",
+    })
+    write_jar(artifact_path(base_dir, "aggregator", "353"), {
+        SHARED: b"aggregator-shared-353",
+        OLDER_ONLY: b"aggregator-older-only",
+        OLDER_IMPL: b"aggregator-older-impl",
+    })
+
+
+def read_bytes(root, entry):
+    return (root / entry).read_bytes()
+
+
+class FakeAttributes:
+    def get(self, name):
+        if name == "artifact_csv":
+            return "sql-plugin-api,aggregator"
+        raise KeyError(name)
+
+
+class FakeProject:
+    def __init__(self, source_dir, project_dir, target_dir, repository_dir):
+        self.properties = {
+            "included_buildvers": "353,413",
+            "spark.rapids.source.basedir": str(source_dir),
+            "spark.rapids.project.basedir": str(project_dir),
+            "project.version": "1.0",
+            "scala.binary.version": "2.13",
+            "project.build.directory": str(target_dir),
+            "env.ART_URL": "",
+            "maven.local.repository": str(repository_dir),
+            "should.build.conventional.jar": False,
+        }
+
+    def getProperty(self, name):
+        return self.properties.get(name)
+
+
+def execfile_compat(path, globals_dict):
+    with open(path, "rb") as source:
+        code = compile(source.read(), str(path), "exec")
+    exec(code, globals_dict)
+
+
+def load_fast_assembler():
+    spec = importlib.util.spec_from_file_location("build_unshim_parallel_world", FAST_ASSEMBLER)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+class RootSafeProviderSelectionTest(unittest.TestCase):
+    def setUp(self):
+        self.temp_dir = tempfile.TemporaryDirectory()
+        self.root = Path(self.temp_dir.name)
+        self.project_dir = self.root / "project"
+        self.source_dir = self.root / "source"
+        self.config_dir = self.source_dir / "dist"
+        self.config_dir.mkdir(parents=True)
+        (self.config_dir / "build").mkdir()
+        (self.config_dir / "unshimmed-common-from-single-shim.txt").write_text("")
+        (self.config_dir / "unshimmed-from-each-spark3xx.txt").write_text("")
+        (self.config_dir / "root-safe-module-classes.txt").write_text("iceberg-common\n")
+        (self.config_dir / "keep-in-spark-shared.txt").write_text("")
+        (self.config_dir / "keep-in-spark-shim-dirs.txt").write_text(
+            "org/apache/iceberg/*.class\n")
+        (self.config_dir / "build" / "iceberg_runtime.py").write_text(
+            "def coordinates(zip_handle, buildver, scala_version, get_property):\n"
+            "    return []\n")
+        create_artifacts(self.project_dir)
+
+    def tearDown(self):
+        self.temp_dir.cleanup()
+
+    def assemble_standard(self):
+        target_dir = self.root / "standard-target"
+        (target_dir / "deps").mkdir(parents=True)
+        globals_dict = {
+            "attributes": FakeAttributes(),
+            "project": FakeProject(
+                self.source_dir, self.project_dir, target_dir, self.root / "repository"),
+            "execfile": execfile_compat,
+            "self": self,
+        }
+        execfile_compat(STANDARD_ASSEMBLER, globals_dict)
+        return target_dir
+
+    def assemble_fast(self):
+        target_dir = self.root / "fast-target"
+        fast = load_fast_assembler()
+        fast.copy_and_extract_jars(
+            self.project_dir, target_dir, "2.13", "1.0", ["353", "413"],
+            [], [], ["iceberg-common"])
+        return target_dir
+
+    def assert_provider_selection(self, target_dir):
+        parallel_world = target_dir / "parallel-world"
+        self.assertEqual(b"aggregator-shared-413", read_bytes(parallel_world, SHARED))
+        self.assertEqual(b"aggregator-newer-only", read_bytes(parallel_world, NEWER_ONLY))
+        self.assertEqual(b"aggregator-older-only", read_bytes(parallel_world, OLDER_ONLY))
+        self.assertFalse((parallel_world / NEWER_IMPL).exists())
+        self.assertFalse((parallel_world / OLDER_IMPL).exists())
+        self.assertTrue((parallel_world / "spark413" / NEWER_IMPL).is_file())
+        self.assertTrue((parallel_world / "spark353" / OLDER_IMPL).is_file())
+
+    def run_dedupe(self, target_dir):
+        parallel_world = target_dir / "parallel-world"
+        # Root-safe classes must remain binary-compatible across the worlds where they occur.
+        # Normalize the synthetic shared class after verifying which provider supplied the root.
+        (parallel_world / "spark353" / SHARED).write_bytes(
+            read_bytes(parallel_world / "spark413", SHARED))
+        env = os.environ.copy()
+        env.update({
+            "UNSHIM_FAST": "1",
+            "UNSHIMMED_COMMON_FROM_SINGLE_SHIM_TXT": str(
+                self.config_dir / "unshimmed-common-from-single-shim.txt"),
+            "KEEP_IN_SPARK_SHARED_TXT": str(
+                self.config_dir / "keep-in-spark-shared.txt"),
+            "KEEP_IN_SPARK_SHIM_DIRS_TXT": str(
+                self.config_dir / "keep-in-spark-shim-dirs.txt"),
+            "UNSHIM_ANALYZER_SCRIPT": str(self.root / "missing-analyzer.py"),
+        })
+        result = subprocess.run([str(BINARY_DEDUPE)], cwd=target_dir, env=env,
+                                stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+                                universal_newlines=True)
+        self.assertEqual(0, result.returncode, result.stdout + result.stderr)
+
+    def assert_final_layout(self, target_dir):
+        parallel_world = target_dir / "parallel-world"
+        for helper in (SHARED, NEWER_ONLY, OLDER_ONLY):
+            self.assertTrue((parallel_world / helper).is_file())
+            self.assertFalse((parallel_world / "spark413" / helper).exists())
+            self.assertFalse((parallel_world / "spark353" / helper).exists())
+        self.assertTrue((parallel_world / "spark413" / NEWER_IMPL).is_file())
+        self.assertTrue((parallel_world / "spark353" / OLDER_IMPL).is_file())
+        self.assertFalse((parallel_world / "spark353" / NEWER_IMPL).exists())
+        self.assertFalse((parallel_world / "spark413" / OLDER_IMPL).exists())
+        shared_iceberg = parallel_world / "spark-shared" / "org/apache/iceberg"
+        self.assertEqual([], list(shared_iceberg.rglob("*.class")))
+
+    def test_provider_selection_and_dedupe_for_both_assemblers(self):
+        for name, assemble in (
+                ("standard", self.assemble_standard),
+                ("fast", self.assemble_fast)):
+            with self.subTest(assembler=name):
+                target_dir = assemble()
+                self.assert_provider_selection(target_dir)
+                self.run_dedupe(target_dir)
+                self.assert_final_layout(target_dir)
+
+
+if __name__ == "__main__":
+    unittest.main()

dist/unshimmed-common-from-single-shim.txt

@@ -11,13 +11,6 @@ com/nvidia/spark/rapids/optimizer/SQLOptimizerPlugin*
 com/nvidia/spark/rapids/ShimLoaderTemp*
 com/nvidia/spark/rapids/SparkShims*
 com/nvidia/shaded/spark/com/google/flatbuffers/*.class
-com/nvidia/spark/rapids/iceberg/parquet/converter/FromIcebergShaded*.class
-com/nvidia/spark/rapids/iceberg/parquet/converter/ToIcebergShaded*.class
-com/nvidia/spark/rapids/iceberg/spark/RapidsSparkCatalog.class
-com/nvidia/spark/rapids/iceberg/spark/RapidsSparkSessionCatalog.class
-com/nvidia/spark/rapids/iceberg/spark/source/RapidsSparkTable.class
-org/apache/iceberg/spark/source/GpuBaseReader.class
-org/apache/iceberg/spark/source/GpuSparkPlanningUtil.class
 org/apache/spark/sql/rapids/AdaptiveSparkPlanHelperShim*
 org/apache/spark/sql/rapids/ExecutionPlanCaptureCallback*
 rapids/*.py

iceberg-common/pom.xml

@@ -37,6 +37,14 @@
         <rapids.shim.jar.phase>package</rapids.shim.jar.phase>
         <rapids.shimplify.skip>false</rapids.shimplify.skip>
         <maven.scaladoc.skip>true</maven.scaladoc.skip>
+        <!--
+        Spark 4.x catalyst class files carry @Deprecated(since = "...")
+        annotations. Reading them under a Java 8 target makes javac emit
+        classfile warnings, which the project's -Werror promotes to an error.
+        Keep source warnings enabled while excluding dependency classfile
+        warnings for the conditional Java helpers in this module.
+        -->
+        <scala.javac.args>-Xlint:all,-serial,-path,-try,-processing,-options,-classfile|-Werror</scala.javac.args>
     </properties>
 
     <dependencies>

iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkCatalog.java → iceberg-common/src/main/spark350/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkCatalog.java

@@ -14,6 +14,27 @@
  * limitations under the License.
  */
 
+/*** spark-rapids-shim-json-lines
+{"spark": "350"}
+{"spark": "351"}
+{"spark": "352"}
+{"spark": "353"}
+{"spark": "354"}
+{"spark": "355"}
+{"spark": "356"}
+{"spark": "357"}
+{"spark": "358"}
+{"spark": "359"}
+{"spark": "400"}
+{"spark": "401"}
+{"spark": "402"}
+{"spark": "403"}
+{"spark": "404"}
+{"spark": "411"}
+{"spark": "412"}
+{"spark": "413"}
+spark-rapids-shim-json-lines ***/
+
 package com.nvidia.spark.rapids.iceberg.spark;
 
 import org.apache.iceberg.spark.SparkCatalog;

iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkSessionCatalog.java → iceberg-common/src/main/spark350/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkSessionCatalog.java

@@ -14,6 +14,27 @@
  * limitations under the License.
  */
 
+/*** spark-rapids-shim-json-lines
+{"spark": "350"}
+{"spark": "351"}
+{"spark": "352"}
+{"spark": "353"}
+{"spark": "354"}
+{"spark": "355"}
+{"spark": "356"}
+{"spark": "357"}
+{"spark": "358"}
+{"spark": "359"}
+{"spark": "400"}
+{"spark": "401"}
+{"spark": "402"}
+{"spark": "403"}
+{"spark": "404"}
+{"spark": "411"}
+{"spark": "412"}
+{"spark": "413"}
+spark-rapids-shim-json-lines ***/
+
 package com.nvidia.spark.rapids.iceberg.spark;
 
 import com.nvidia.spark.rapids.iceberg.spark.source.RapidsSparkTable;