diff --git a/dist/build/package-parallel-worlds.py b/dist/build/package-parallel-worlds.py index ef279e25f4c..f8ca6738f80 100644 --- a/dist/build/package-parallel-worlds.py +++ b/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 diff --git a/dist/keep-in-spark-shim-dirs.txt b/dist/keep-in-spark-shim-dirs.txt index 41dded9f0ff..46d220ed661 100644 --- a/dist/keep-in-spark-shim-dirs.txt +++ b/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 diff --git a/dist/scripts/binary-dedupe.sh b/dist/scripts/binary-dedupe.sh index a5ac3346c79..7ae2fa62a7c 100755 --- a/dist/scripts/binary-dedupe.sh +++ b/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" diff --git a/dist/scripts/build-unshim-parallel-world.py b/dist/scripts/build-unshim-parallel-world.py index 05e47882618..8ca95a920c0 100644 --- a/dist/scripts/build-unshim-parallel-world.py +++ b/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: diff --git a/dist/scripts/tests/test_root_safe_provider_selection.py b/dist/scripts/tests/test_root_safe_provider_selection.py new file mode 100644 index 00000000000..148286fcdea --- /dev/null +++ b/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() diff --git a/dist/unshimmed-common-from-single-shim.txt b/dist/unshimmed-common-from-single-shim.txt index cdc418a2ce5..4faa1243144 100644 --- a/dist/unshimmed-common-from-single-shim.txt +++ b/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 diff --git a/iceberg-common/pom.xml b/iceberg-common/pom.xml index dcd5bdfabd9..974e0a05676 100644 --- a/iceberg-common/pom.xml +++ b/iceberg-common/pom.xml @@ -37,6 +37,14 @@ package false true + + -Xlint:all,-serial,-path,-try,-processing,-options,-classfile|-Werror diff --git a/iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkCatalog.java b/iceberg-common/src/main/spark350/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkCatalog.java similarity index 84% rename from iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkCatalog.java rename to iceberg-common/src/main/spark350/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkCatalog.java index f86ecf60eee..3fa7c6f1aac 100644 --- a/iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkCatalog.java +++ b/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; diff --git a/iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkSessionCatalog.java b/iceberg-common/src/main/spark350/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkSessionCatalog.java similarity index 87% rename from iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkSessionCatalog.java rename to iceberg-common/src/main/spark350/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkSessionCatalog.java index 1620ba84a75..5e60dcd20bb 100644 --- a/iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/RapidsSparkSessionCatalog.java +++ b/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; diff --git a/iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/source/RapidsSparkTable.java b/iceberg-common/src/main/spark350/java/com/nvidia/spark/rapids/iceberg/spark/source/RapidsSparkTable.java similarity index 97% rename from iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/source/RapidsSparkTable.java rename to iceberg-common/src/main/spark350/java/com/nvidia/spark/rapids/iceberg/spark/source/RapidsSparkTable.java index a631fcee427..0b1128f8db3 100644 --- a/iceberg/common/src/main/java/com/nvidia/spark/rapids/iceberg/spark/source/RapidsSparkTable.java +++ b/iceberg-common/src/main/spark350/java/com/nvidia/spark/rapids/iceberg/spark/source/RapidsSparkTable.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.source; import java.util.Arrays; diff --git a/iceberg/iceberg-1-6-x/src/main/java/org/apache/iceberg/spark/source/GpuBaseReader.java b/iceberg-common/src/main/spark350/java/org/apache/iceberg/spark/source/GpuBaseReader.java similarity index 81% rename from iceberg/iceberg-1-6-x/src/main/java/org/apache/iceberg/spark/source/GpuBaseReader.java rename to iceberg-common/src/main/spark350/java/org/apache/iceberg/spark/source/GpuBaseReader.java index aa76cc468b5..9171ff8c57c 100644 --- a/iceberg/iceberg-1-6-x/src/main/java/org/apache/iceberg/spark/source/GpuBaseReader.java +++ b/iceberg-common/src/main/spark350/java/org/apache/iceberg/spark/source/GpuBaseReader.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2025, NVIDIA CORPORATION. + * Copyright (c) 2025-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. @@ -14,6 +14,13 @@ * limitations under the License. */ +/*** spark-rapids-shim-json-lines +{"spark": "350"} +{"spark": "351"} +{"spark": "352"} +{"spark": "353"} +spark-rapids-shim-json-lines ***/ + package org.apache.iceberg.spark.source; import org.apache.iceberg.types.Type; diff --git a/iceberg/common/src/main/java/org/apache/iceberg/spark/source/GpuSparkPlanningUtil.java b/iceberg-common/src/main/spark350/java/org/apache/iceberg/spark/source/GpuSparkPlanningUtil.java similarity index 72% rename from iceberg/common/src/main/java/org/apache/iceberg/spark/source/GpuSparkPlanningUtil.java rename to iceberg-common/src/main/spark350/java/org/apache/iceberg/spark/source/GpuSparkPlanningUtil.java index e26cb01584b..4a3f453e6cd 100644 --- a/iceberg/common/src/main/java/org/apache/iceberg/spark/source/GpuSparkPlanningUtil.java +++ b/iceberg-common/src/main/spark350/java/org/apache/iceberg/spark/source/GpuSparkPlanningUtil.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2025, NVIDIA CORPORATION. + * Copyright (c) 2025-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. @@ -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 org.apache.iceberg.spark.source; import org.apache.iceberg.ScanTaskGroup; diff --git a/pom.xml b/pom.xml index 8484ed5c513..5d7e55d0ab7 100644 --- a/pom.xml +++ b/pom.xml @@ -2056,6 +2056,11 @@ This will force full Scala code rebuild in downstream modules. + + + + package false true + + -Xlint:all,-serial,-path,-try,-processing,-options,-classfile|-Werror diff --git a/scala2.13/pom.xml b/scala2.13/pom.xml index c5b6b2f0c52..10cc91c211d 100644 --- a/scala2.13/pom.xml +++ b/scala2.13/pom.xml @@ -2056,6 +2056,11 @@ This will force full Scala code rebuild in downstream modules. + + + +