From ddd2038f57e8551495a748df60cda7db687ad0c6 Mon Sep 17 00:00:00 2001 From: Ray Liu Date: Wed, 2 Sep 2026 17:03:41 +0800 Subject: [PATCH 1/3] Add Iceberg fast test matrix Signed-off-by: Ray Liu --- iceberg/README.md | 19 +- integration_tests/README.md | 6 + integration_tests/requirements.txt | 1 + integration_tests/src/main/python/conftest.py | 4 + .../python/iceberg/iceberg_append_test.py | 38 ++-- .../main/python/iceberg/iceberg_ctas_test.py | 26 ++- .../python/iceberg/iceberg_delete_test.py | 35 ++- .../main/python/iceberg/iceberg_merge_test.py | 32 ++- .../iceberg/iceberg_overwrite_dynamic_test.py | 23 +- .../iceberg/iceberg_overwrite_static_test.py | 35 ++- .../main/python/iceberg/iceberg_rtas_test.py | 29 ++- .../python/iceberg/iceberg_update_test.py | 35 ++- jenkins/get_iceberg_versions.py | 212 ++++++++++++++++++ jenkins/iceberg-test-matrix.yaml | 135 +++++++++++ jenkins/spark-premerge-build.sh | 34 +-- jenkins/spark-tests.sh | 84 ++++--- scripts/tests/test_get_iceberg_versions.py | 118 ++++++++++ .../tests/test_iceberg_fast_test_selection.py | 67 ++++++ 18 files changed, 766 insertions(+), 167 deletions(-) create mode 100644 jenkins/get_iceberg_versions.py create mode 100644 jenkins/iceberg-test-matrix.yaml create mode 100644 scripts/tests/test_get_iceberg_versions.py create mode 100644 scripts/tests/test_iceberg_fast_test_selection.py diff --git a/iceberg/README.md b/iceberg/README.md index 191131c4cd2..efc522aafe8 100644 --- a/iceberg/README.md +++ b/iceberg/README.md @@ -11,18 +11,23 @@ and the directory that contains the corresponding support code. | Iceberg Version | Spark Version | Directory | |-----------------|----------------------------|-------------------| -| 1.6.x | Spark 3.5.0-3.5.3 | `iceberg-1-6-x` | -| 1.9.x | Spark 3.5.4-3.5.9 | `iceberg-1-9-x` | -| 1.10.x | Spark 3.5.4-3.5.9, 4.0.x | `iceberg-1-10-x` | +| 1.6.x | Spark 3.5.1-3.5.3 | `iceberg-1-6-x` | +| 1.9.x | Spark 3.5.5-3.5.9 | `iceberg-1-9-x` | +| 1.10.x | Spark 3.5.6-3.5.9, 4.0.x | `iceberg-1-10-x` | | 1.11.x | Spark 4.0.2+, 4.1.x | `iceberg-1-11-x` | Iceberg GPU acceleration is currently supported on Spark 3.5.x, 4.0.x, and 4.1.x. +The authoritative integration-test compatibility list, including upstream-compatible +combinations that are not currently packaged, is maintained in +[`jenkins/iceberg-test-matrix.yaml`](../jenkins/iceberg-test-matrix.yaml). For Spark 3.5.4+, both `iceberg-1-9-x` and `iceberg-1-10-x` modules are compiled into the -build. The correct version-specific implementation is selected at runtime by probing the -`iceberg-spark-runtime` jar on the classpath. Version-specific code lives in distinct -sub-packages (`iceberg19x`, `iceberg110x`, `iceberg111x`) to avoid class conflicts, and the -common `ShimUtils` dispatcher delegates to the appropriate implementation. +build. The integration-test support baseline follows the Spark patch versions used to build +the corresponding Apache Iceberg release. The correct version-specific implementation is +selected at runtime by probing the `iceberg-spark-runtime` jar on the classpath. +Version-specific code lives in distinct sub-packages (`iceberg19x`, `iceberg110x`, +`iceberg111x`) to avoid class conflicts, and the common `ShimUtils` dispatcher delegates to +the appropriate implementation. For Spark 4.0.0-4.0.1, only `iceberg-1-10-x` is compiled during the build. For Spark 4.0.2+, both `iceberg-1-10-x` and `iceberg-1-11-x` are compiled, and the correct diff --git a/integration_tests/README.md b/integration_tests/README.md index 35fd21cdcfc..3d781b5b45d 100644 --- a/integration_tests/README.md +++ b/integration_tests/README.md @@ -86,6 +86,8 @@ For manual installation, you need to setup your environment: - pytest : A framework that makes it easy to write small, readable tests, and can scale to support complex functional testing for applications and libraries (requires Python 3.6+). +- PyYAML + : Parses the Iceberg and Spark integration-test compatibility matrix used by CI. - sre_yield : Provides a set of APIs to generate string data from a regular expression. - pandas @@ -539,6 +541,10 @@ properly without it. These tests assume Iceberg is not configured and are disabl If Spark has been configured to support Iceberg then these tests can be enabled by adding the `--iceberg` option to the command. +Set `ICEBERG_TEST_FAST_RUN=1` to skip redundant, high-cost cases while retaining tests that +specifically require a local Hadoop catalog. CI uses this mode when it expands the supported +Iceberg and Spark combinations from `jenkins/iceberg-test-matrix.yaml`. + When testing Iceberg package-private access paths, load the local Iceberg runtime jar with `ICEBERG_EXTRA_CLASSPATH` instead of `PYSP_TEST_spark_jars` or `PYSP_TEST_spark_jars_packages`. The test driver will place the RAPIDS, test, and Iceberg diff --git a/integration_tests/requirements.txt b/integration_tests/requirements.txt index 1bd53040f56..0b58b30bafb 100644 --- a/integration_tests/requirements.txt +++ b/integration_tests/requirements.txt @@ -12,6 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. pytest +PyYAML sre_yield pandas pyarrow == 17.0.0 ; python_version == '3.8' diff --git a/integration_tests/src/main/python/conftest.py b/integration_tests/src/main/python/conftest.py index 0c929af6fd4..a3f2e887997 100644 --- a/integration_tests/src/main/python/conftest.py +++ b/integration_tests/src/main/python/conftest.py @@ -119,6 +119,10 @@ def is_iceberg_remote_catalog(): v = os.environ.get('ICEBERG_TEST_REMOTE_CATALOG') return v == "1" +def is_iceberg_test_fast_run(): + v = os.environ.get('ICEBERG_TEST_FAST_RUN') + return v == "1" + def is_iceberg_rest_catalog(): v = os.environ.get('ICEBERG_TEST_CATALOG_TYPE') return v == "rest" diff --git a/integration_tests/src/main/python/iceberg/iceberg_append_test.py b/integration_tests/src/main/python/iceberg/iceberg_append_test.py index e352ae39d27..0bffc8ba2e1 100644 --- a/integration_tests/src/main/python/iceberg/iceberg_append_test.py +++ b/integration_tests/src/main/python/iceberg/iceberg_append_test.py @@ -17,7 +17,7 @@ from asserts import assert_equal_with_local_sort, assert_gpu_fallback_collect, \ assert_gpu_fallback_write_sql -from conftest import is_iceberg_remote_catalog +from conftest import is_iceberg_remote_catalog, is_iceberg_test_fast_run from data_gen import gen_df, copy_and_update from iceberg import create_iceberg_table, \ iceberg_base_table_cols, iceberg_gens_list, get_full_table_name, \ @@ -97,7 +97,8 @@ def insert_data(spark, table_name): @iceberg @ignore_order(local=True) @allow_non_gpu('AppendDataExec') -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize("partition_table", [True, False], ids=lambda x: f"partition_table={x}") def test_insert_into_unpartitioned_table_values(spark_tmp_table_factory, partition_table): @@ -137,7 +138,8 @@ def insert_data(spark, table_name: str): @iceberg @ignore_order(local=True) @allow_non_gpu('LocalTableScanExec', 'ShuffleExchangeExec') -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize("partition_table", [True, False], ids=lambda x: f"partition_table={x}") def test_insert_into_table_values_aqe(spark_tmp_table_factory, partition_table): """Regression test for GPU V2 writes with AQE and a CPU VALUES input plan.""" @@ -175,7 +177,8 @@ def insert_data(spark, table_name: str): @iceberg @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") def test_insert_into_unpartitioned_table_all_cols(spark_tmp_table_factory): table_prop = {"format-version": "2"} cols = [f"_c{idx}" for idx, _ in enumerate(iceberg_full_gens_list)] @@ -241,7 +244,8 @@ def test_insert_into_partitioned_table(spark_tmp_table_factory, partition_col_sq @iceberg @datagen_overrides(seed=0, reason='https://github.com/NVIDIA/spark-rapids-jni/issues/4016') @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize("partition_col_sql", full_coverage_partition_transforms) def test_insert_into_partitioned_table_full_coverage(spark_tmp_table_factory, partition_col_sql): """Partition-transform coverage anchor: this is the single test that exercises @@ -254,7 +258,8 @@ def test_insert_into_partitioned_table_full_coverage(spark_tmp_table_factory, pa @iceberg @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") def test_insert_into_partitioned_table_all_cols(spark_tmp_table_factory): table_prop = {"format-version": "2"} cols = [f"_c{idx}" for idx, _ in enumerate(iceberg_full_gens_list)] @@ -299,7 +304,8 @@ def insert_data(spark, table_name: str): @iceberg @ignore_order(local=True) @allow_non_gpu('AppendDataExec', 'ShuffleExchangeExec', 'ProjectExec') -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize("file_format", ["orc", "avro"], ids=lambda x: f"file_format={x}") def test_insert_into_table_unsupported_file_format_fallback( spark_tmp_table_factory, file_format): @@ -322,7 +328,8 @@ def insert_data(spark, table_name: str): @iceberg @ignore_order(local=True) @allow_non_gpu('AppendDataExec', 'ShuffleExchangeExec', 'ProjectExec') -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize("partition_col_sql", [ pytest.param("bucket(4, contact.email)", id="bucket_nested_struct_field"), pytest.param("truncate(3, contact.email)", id="truncate_nested_struct_field"), @@ -358,7 +365,8 @@ def insert_data(spark): @iceberg @ignore_order(local=True) @allow_non_gpu('AppendDataExec', 'ShuffleExchangeExec', 'ProjectExec') -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize("conf_key", ["spark.rapids.sql.format.iceberg.enabled", "spark.rapids.sql.format.iceberg.write.enabled"], ids=lambda x: f"{x}=False") @@ -428,7 +436,8 @@ def insert_data(spark, table_name: str): @iceberg @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") def test_insert_after_drop_partition_field(spark_tmp_table_factory): """Test INSERT on table after dropping a partition field (void transform). @@ -485,7 +494,8 @@ def insert_data(spark, table_name): @iceberg @datagen_overrides(seed=0, reason='https://github.com/NVIDIA/spark-rapids-jni/issues/4016') @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") def test_insert_into_partitioned_table_fanout_enabled(spark_tmp_table_factory): # Use bucket(2, ...) to keep partition count low and avoid OOM from Iceberg's FanoutDataWriter. _do_test_insert_into_partitioned_table( @@ -500,7 +510,8 @@ def test_insert_into_partitioned_table_fanout_enabled(spark_tmp_table_factory): # GPU would silently use Spark's session codec instead of Iceberg's zstd default. @iceberg @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") # Restricted to codecs whose footer metadata is reliable on small inputs. cuDF skips # compression for tiny row groups (rapidsai/cudf#14017), so codecs like snappy can leave # `UNCOMPRESSED` in the footer of small per-task files and make the assertion flaky; @@ -549,7 +560,8 @@ def create_table(spark): @iceberg @ignore_order(local=True) @allow_non_gpu('AppendDataExec', 'ShuffleExchangeExec', 'ProjectExec') -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize("codec", ["gzip", "lz4"]) def test_insert_into_table_falls_back_on_unsupported_codec(spark_tmp_table_factory, codec): table_name = get_full_table_name(spark_tmp_table_factory) diff --git a/integration_tests/src/main/python/iceberg/iceberg_ctas_test.py b/integration_tests/src/main/python/iceberg/iceberg_ctas_test.py index fae07bb9999..25e23516a90 100644 --- a/integration_tests/src/main/python/iceberg/iceberg_ctas_test.py +++ b/integration_tests/src/main/python/iceberg/iceberg_ctas_test.py @@ -19,7 +19,7 @@ from asserts import (assert_equal_with_local_sort, assert_gpu_and_cpu_are_equal_collect, assert_gpu_fallback_collect) -from conftest import is_iceberg_remote_catalog +from conftest import is_iceberg_remote_catalog, is_iceberg_test_fast_run from data_gen import gen_df, copy_and_update, RepeatSeqGen from iceberg import (create_iceberg_table, iceberg_base_table_cols, @@ -199,7 +199,8 @@ def test_ctas_partitioned_table(spark_tmp_table_factory, partition_col_sql): @iceberg @datagen_overrides(seed=0, reason='https://github.com/NVIDIA/spark-rapids-jni/issues/4016') @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize("partition_col_sql", ctas_partition_transforms) @allow_non_gpu_conditional(is_spark_400_or_later(), "EmptyRelationExec") def test_ctas_partitioned_table_full_coverage(spark_tmp_table_factory, partition_col_sql): @@ -212,7 +213,8 @@ def test_ctas_partitioned_table_full_coverage(spark_tmp_table_factory, partition @iceberg @ignore_order(local=True) @allow_non_gpu('AtomicCreateTableAsSelectExec', 'AppendDataExec') -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize("file_format", ["orc", "avro"], ids=lambda x: f"file_format={x}") def test_ctas_unsupported_file_format_fallback(spark_tmp_table_factory, file_format): @@ -237,7 +239,8 @@ def run_ctas(spark): @iceberg @ignore_order(local=True) @allow_non_gpu('AtomicCreateTableAsSelectExec', 'AppendDataExec') -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize("conf_key", ["spark.rapids.sql.format.iceberg.enabled", "spark.rapids.sql.format.iceberg.write.enabled"], ids=lambda x: f"{x}=False") @@ -263,7 +266,8 @@ def run_ctas(spark): @iceberg @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize("gen_list", _BINARY_CTAS_GEN_LISTS, ids=["binary", "array_binary"]) def test_ctas_unpartitioned_table_binary_types(spark_tmp_table_factory, gen_list): table_prop = { @@ -277,7 +281,8 @@ def test_ctas_unpartitioned_table_binary_types(spark_tmp_table_factory, gen_list @iceberg @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @allow_non_gpu_conditional(is_spark_400_or_later(), "EmptyRelationExec") def test_ctas_unpartitioned_table_all_cols(spark_tmp_table_factory): table_prop = { @@ -293,7 +298,8 @@ def test_ctas_unpartitioned_table_all_cols(spark_tmp_table_factory): @iceberg @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @allow_non_gpu_conditional(is_spark_400_or_later(), "EmptyRelationExec") def test_ctas_partitioned_table_all_cols(spark_tmp_table_factory): table_prop = { @@ -312,7 +318,8 @@ def test_ctas_partitioned_table_all_cols(spark_tmp_table_factory): @iceberg @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize("partition_table", [True, False], ids=lambda x: f"partition_table={x}") @allow_non_gpu('AtomicCreateTableAsSelectExec', 'AppendDataExec', 'ShuffleExchangeExec', 'SortExec', 'ProjectExec') @allow_non_gpu_conditional(is_spark_400_or_later(), "EmptyRelationExec") @@ -386,7 +393,8 @@ def test_ctas_aqe(spark_tmp_table_factory, partition_col_sql): @iceberg @datagen_overrides(seed=0, reason='https://github.com/NVIDIA/spark-rapids-jni/issues/4016') @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") def test_ctas_partitioned_table_fanout_enabled(spark_tmp_table_factory): # Use bucket(2, ...) to keep partition count low and avoid OOM from Iceberg's FanoutDataWriter. _do_test_ctas_partitioned_table( diff --git a/integration_tests/src/main/python/iceberg/iceberg_delete_test.py b/integration_tests/src/main/python/iceberg/iceberg_delete_test.py index ce680ba15c4..1be7726c940 100644 --- a/integration_tests/src/main/python/iceberg/iceberg_delete_test.py +++ b/integration_tests/src/main/python/iceberg/iceberg_delete_test.py @@ -16,7 +16,7 @@ from asserts import assert_equal_with_local_sort, assert_gpu_and_cpu_are_equal_collect, \ assert_gpu_fallback_write_sql -from conftest import is_iceberg_remote_catalog +from conftest import is_iceberg_remote_catalog, is_iceberg_test_fast_run from data_gen import * from iceberg import (create_iceberg_table, get_full_table_name, iceberg_write_enabled_conf, iceberg_base_table_cols, iceberg_gens_list, iceberg_nested_write_gens_list, @@ -231,7 +231,8 @@ def test_iceberg_delete_partitioned_table(spark_tmp_table_factory, partition_col @datagen_overrides(seed=0, reason='https://github.com/NVIDIA/spark-rapids-jni/issues/4016') @ignore_order(local=True) @pytest.mark.datagen_overrides(seed=DELETE_TEST_SEED, reason=DELETE_TEST_SEED_OVERRIDE_REASON) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize("partition_col_sql,delete_mode", delete_partition_transforms_distributed) @allow_non_gpu_conditional(is_spark_400_or_later(), "EmptyRelationExec") def test_iceberg_delete_partitioned_table_full_coverage(spark_tmp_table_factory, partition_col_sql, delete_mode): @@ -244,7 +245,8 @@ def test_iceberg_delete_partitioned_table_full_coverage(spark_tmp_table_factory, @iceberg @ignore_order(local=True) @pytest.mark.datagen_overrides(seed=DELETE_TEST_SEED, reason=DELETE_TEST_SEED_OVERRIDE_REASON) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize('delete_mode', ['copy-on-write', 'merge-on-read']) @allow_non_gpu_conditional(is_spark_400_or_later(), "EmptyRelationExec") def test_iceberg_delete_with_complex_predicate(spark_tmp_table_factory, delete_mode): @@ -262,7 +264,8 @@ def test_iceberg_delete_with_complex_predicate(spark_tmp_table_factory, delete_m @iceberg @ignore_order(local=True) @pytest.mark.datagen_overrides(seed=DELETE_TEST_SEED, reason=DELETE_TEST_SEED_OVERRIDE_REASON) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize('delete_mode,fallback_exec', [ pytest.param('copy-on-write', 'ReplaceDataExec', id='cow'), pytest.param('merge-on-read', 'WriteDeltaExec', id='mor') @@ -300,7 +303,8 @@ def read_func(spark, table_name): @iceberg @ignore_order(local=True) @pytest.mark.datagen_overrides(seed=DELETE_TEST_SEED, reason=DELETE_TEST_SEED_OVERRIDE_REASON) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize('delete_mode,fallback_exec', [ pytest.param('copy-on-write', 'ReplaceDataExec', id='cow'), pytest.param('merge-on-read', 'WriteDeltaExec', id='mor') @@ -366,7 +370,8 @@ def read_func(spark, table_name): @iceberg @ignore_order(local=True) @pytest.mark.datagen_overrides(seed=DELETE_TEST_SEED, reason=DELETE_TEST_SEED_OVERRIDE_REASON) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize('delete_mode', ['copy-on-write', 'merge-on-read']) @allow_non_gpu_conditional(is_spark_400_or_later(), "EmptyRelationExec") def test_iceberg_delete_nested_types(spark_tmp_table_factory, delete_mode): @@ -386,7 +391,8 @@ def test_iceberg_delete_nested_types(spark_tmp_table_factory, delete_mode): @iceberg @ignore_order(local=True) @pytest.mark.datagen_overrides(seed=DELETE_TEST_SEED, reason=DELETE_TEST_SEED_OVERRIDE_REASON) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize('delete_mode,fallback_exec', [ pytest.param('copy-on-write', 'ReplaceDataExec', id='cow'), pytest.param('merge-on-read', 'WriteDeltaExec', id='mor') @@ -423,7 +429,8 @@ def read_func(spark, table_name): @iceberg @ignore_order(local=True) @pytest.mark.datagen_overrides(seed=DELETE_TEST_SEED, reason=DELETE_TEST_SEED_OVERRIDE_REASON) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") def test_iceberg_delete_mor_fallback_writedelta_disabled(spark_tmp_table_factory): """Test merge-on-read DELETE falls back when WriteDeltaExec is disabled @@ -506,7 +513,8 @@ def delete_from_table(spark, table_name): @iceberg @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.datagen_overrides(seed=DELETE_TEST_SEED, reason=DELETE_TEST_SEED_OVERRIDE_REASON) @pytest.mark.parametrize('delete_mode', ['copy-on-write', 'merge-on-read']) @allow_non_gpu_conditional(is_spark_400_or_later(), "EmptyRelationExec") @@ -554,7 +562,8 @@ def do_delete(spark, table_name): @iceberg @datagen_overrides(seed=0, reason='https://github.com/NVIDIA/spark-rapids-jni/issues/4016') @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") def test_iceberg_delete_partitioned_table_fanout_enabled(spark_tmp_table_factory): # Use bucket(2, ...) to keep partition count low and avoid OOM from Iceberg's FanoutDataWriter. _do_test_iceberg_delete_partitioned_table( @@ -571,7 +580,8 @@ def test_iceberg_delete_partitioned_table_fanout_enabled(spark_tmp_table_factory # writes position-delete files. @iceberg @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize("table_codec,expected_codec", [ (None, "zstd"), ("zstd", "zstd"), @@ -619,7 +629,8 @@ def create_and_populate(spark): @iceberg @ignore_order(local=True) @pytest.mark.datagen_overrides(seed=DELETE_TEST_SEED, reason=DELETE_TEST_SEED_OVERRIDE_REASON) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") def test_mor_delete_falls_back_on_divergent_delete_codec(spark_tmp_table_factory): base_table_name = get_full_table_name(spark_tmp_table_factory) cpu_table_name = f"{base_table_name}_cpu" diff --git a/integration_tests/src/main/python/iceberg/iceberg_merge_test.py b/integration_tests/src/main/python/iceberg/iceberg_merge_test.py index 0d26fd341aa..bf3cc5426b8 100644 --- a/integration_tests/src/main/python/iceberg/iceberg_merge_test.py +++ b/integration_tests/src/main/python/iceberg/iceberg_merge_test.py @@ -16,7 +16,7 @@ from asserts import assert_equal_with_local_sort, assert_gpu_and_cpu_are_equal_collect, \ assert_gpu_fallback_write_sql -from conftest import is_iceberg_remote_catalog +from conftest import is_iceberg_remote_catalog, is_iceberg_test_fast_run from data_gen import * from iceberg import (create_iceberg_table, get_full_table_name, iceberg_write_enabled_conf, iceberg_base_table_cols, iceberg_gens_list, iceberg_nested_write_gens_list, @@ -268,7 +268,8 @@ def setup_iceberg_table(spark): @iceberg @datagen_overrides(seed=0, reason='https://github.com/NVIDIA/spark-rapids-jni/issues/4016') @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize("partition_col_sql,merge_mode", merge_partition_transforms_distributed) @allow_non_gpu_conditional(is_spark_400_or_later(), "EmptyRelationExec") def test_iceberg_merge_full_coverage(spark_tmp_table_factory, partition_col_sql, merge_mode): @@ -282,7 +283,8 @@ def test_iceberg_merge_full_coverage(spark_tmp_table_factory, partition_col_sql, @allow_non_gpu("MergeRows$Keep", "MergeRows$Discard", "MergeRows$Split") @iceberg @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize('merge_mode', ['copy-on-write', 'merge-on-read']) @pytest.mark.parametrize('partition_col_sql', [ pytest.param(None, id="unpartitioned"), @@ -345,7 +347,8 @@ def test_iceberg_merge_additional_patterns(spark_tmp_table_factory, partition_co @allow_non_gpu("MergeRows$Keep", "MergeRows$Discard", "MergeRows$Split") @iceberg @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize('merge_mode', ['copy-on-write']) @pytest.mark.parametrize('partition_col_sql', [pytest.param("year(_c9)", id="year(timestamp_col)")]) @pytest.mark.parametrize('merge_sql', [ @@ -371,7 +374,8 @@ def test_iceberg_merge_additional_patterns_bug(spark_tmp_table_factory, partitio @allow_non_gpu("ReplaceDataExec", "WriteDeltaExec", "MergeRowsExec", "BatchScanExec", "ColumnarToRowExec", "ShuffleExchangeExec", "SortExec", "ProjectExec") @iceberg @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize('merge_mode,fallback_exec', [ pytest.param('copy-on-write', 'ReplaceDataExec', id='cow'), pytest.param('merge-on-read', 'WriteDeltaExec', id='mor') @@ -418,7 +422,8 @@ def read_func(spark, table_name): @allow_non_gpu("ReplaceDataExec", "WriteDeltaExec", "MergeRowsExec", "BatchScanExec", "ColumnarToRowExec", "ShuffleExchangeExec", "ProjectExec") @iceberg @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize('merge_mode,fallback_exec', [ pytest.param('copy-on-write', 'ReplaceDataExec', id='cow'), pytest.param('merge-on-read', 'WriteDeltaExec', id='mor') @@ -507,7 +512,8 @@ def read_func(spark, table_name): @allow_non_gpu("MergeRows$Keep", "MergeRows$Discard", "MergeRows$Split") @iceberg @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize('merge_mode', ['copy-on-write', 'merge-on-read']) @allow_non_gpu_conditional(is_spark_400_or_later(), "EmptyRelationExec") def test_iceberg_merge_nested_types(spark_tmp_table_factory, merge_mode): @@ -530,7 +536,8 @@ def test_iceberg_merge_nested_types(spark_tmp_table_factory, merge_mode): @allow_non_gpu("ReplaceDataExec", "WriteDeltaExec", "MergeRowsExec", "BatchScanExec", "ColumnarToRowExec") @iceberg @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize('merge_mode,fallback_exec', [ pytest.param('copy-on-write', 'ReplaceDataExec', id='cow'), pytest.param('merge-on-read', 'WriteDeltaExec', id='mor') @@ -573,7 +580,8 @@ def read_func(spark, table_name): @allow_non_gpu("WriteDeltaExec", "MergeRowsExec", "BatchScanExec", "ColumnarToRowExec") @iceberg @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") def test_iceberg_merge_mor_fallback_writedelta_disabled(spark_tmp_table_factory): """Test merge-on-read MERGE falls back when WriteDeltaExec is disabled @@ -668,7 +676,8 @@ def merge_table(spark, target_table): @iceberg @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize('merge_mode', ['copy-on-write', 'merge-on-read']) @allow_non_gpu_conditional(is_spark_400_or_later(), "EmptyRelationExec") def test_iceberg_merge_after_drop_partition_field(spark_tmp_table_factory, merge_mode): @@ -727,7 +736,8 @@ def do_merge(spark, target_table): @iceberg @datagen_overrides(seed=0, reason='https://github.com/NVIDIA/spark-rapids-jni/issues/4016') @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") def test_iceberg_merge_partitioned_fanout_enabled(spark_tmp_table_factory): # Use bucket(2, ...) to keep partition count low and avoid OOM from Iceberg's FanoutDataWriter. _do_test_iceberg_merge( diff --git a/integration_tests/src/main/python/iceberg/iceberg_overwrite_dynamic_test.py b/integration_tests/src/main/python/iceberg/iceberg_overwrite_dynamic_test.py index 384974514f8..a66f14f3db2 100644 --- a/integration_tests/src/main/python/iceberg/iceberg_overwrite_dynamic_test.py +++ b/integration_tests/src/main/python/iceberg/iceberg_overwrite_dynamic_test.py @@ -16,7 +16,7 @@ import pytest from asserts import assert_equal_with_local_sort, assert_gpu_fallback_collect -from conftest import is_iceberg_remote_catalog +from conftest import is_iceberg_remote_catalog, is_iceberg_test_fast_run from data_gen import gen_df, copy_and_update from iceberg import create_iceberg_table, \ iceberg_base_table_cols, iceberg_gens_list, \ @@ -148,7 +148,8 @@ def test_insert_overwrite_dynamic_bucket_partitioned(spark_tmp_table_factory, pa @iceberg @datagen_overrides(seed=0, reason='https://github.com/NVIDIA/spark-rapids-jni/issues/4016') @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize("partition_col_sql", overwrite_dynamic_partition_transforms) @allow_non_gpu_conditional(is_spark_400_or_later(), "EmptyRelationExec") def test_insert_overwrite_dynamic_bucket_partitioned_full_coverage(spark_tmp_table_factory, partition_col_sql): @@ -161,7 +162,8 @@ def test_insert_overwrite_dynamic_bucket_partitioned_full_coverage(spark_tmp_tab @iceberg @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") def test_insert_overwrite_dynamic_nested_types(spark_tmp_table_factory): """Test INSERT OVERWRITE with dynamic mode on Iceberg-native nested types on GPU.""" table_prop = {"format-version": "2"} @@ -207,7 +209,8 @@ def overwrite_data(spark, table_name): @iceberg @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @allow_non_gpu_conditional(is_spark_400_or_later(), "EmptyRelationExec") def test_insert_overwrite_dynamic_all_cols(spark_tmp_table_factory): """Test INSERT OVERWRITE with dynamic mode on all Iceberg write types on GPU.""" @@ -255,7 +258,8 @@ def overwrite_data(spark, table_name): @iceberg @ignore_order(local=True) @allow_non_gpu('OverwritePartitionsDynamicExec', 'ShuffleExchangeExec', 'SortExec', 'ProjectExec') -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize("file_format", ["orc", "avro"], ids=lambda x: f"file_format={x}") @allow_non_gpu_conditional(is_spark_400_or_later(), "EmptyRelationExec") def test_insert_overwrite_dynamic_unsupported_file_format_fallback( @@ -296,7 +300,8 @@ def overwrite_data(spark, table_name: str): @iceberg @ignore_order(local=True) @allow_non_gpu('OverwritePartitionsDynamicExec', 'ShuffleExchangeExec', 'SortExec', 'ProjectExec') -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize("conf_key", ["spark.rapids.sql.format.iceberg.enabled", "spark.rapids.sql.format.iceberg.write.enabled"], ids=lambda x: f"{x}=False") @@ -384,7 +389,8 @@ def overwrite_dynamic(spark, table_name): @allow_non_gpu("BatchScanExec", "ColumnarToRowExec") @iceberg @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") def test_insert_overwrite_dynamic_after_drop_partition_field(spark_tmp_table_factory): """Test INSERT OVERWRITE (dynamic mode) on table after dropping a partition field (void transform). @@ -449,7 +455,8 @@ def overwrite_data(spark, table_name): @iceberg @datagen_overrides(seed=0, reason='https://github.com/NVIDIA/spark-rapids-jni/issues/4016') @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") def test_insert_overwrite_dynamic_partitioned_fanout_enabled(spark_tmp_table_factory): # Use bucket(2, ...) to keep partition count low and avoid OOM from Iceberg's FanoutDataWriter. _do_test_insert_overwrite_dynamic_partitioned( diff --git a/integration_tests/src/main/python/iceberg/iceberg_overwrite_static_test.py b/integration_tests/src/main/python/iceberg/iceberg_overwrite_static_test.py index 417f4c040fd..6403c232069 100644 --- a/integration_tests/src/main/python/iceberg/iceberg_overwrite_static_test.py +++ b/integration_tests/src/main/python/iceberg/iceberg_overwrite_static_test.py @@ -18,7 +18,7 @@ from asserts import assert_equal_with_local_sort, assert_gpu_and_cpu_are_equal_collect, \ assert_gpu_fallback_collect -from conftest import is_iceberg_remote_catalog +from conftest import is_iceberg_remote_catalog, is_iceberg_test_fast_run from data_gen import DEFAULT_DATA_GEN_LENGTH, StringGen, copy_and_update, gen_df from iceberg import create_iceberg_table, \ iceberg_base_table_cols, iceberg_gens_list, \ @@ -153,7 +153,8 @@ def setup_iceberg_table(spark): @iceberg @ignore_order(local=True) @allow_non_gpu('OverwriteByExpressionExec', 'AppendDataExec') -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") def test_insert_overwrite_unpartitioned_table_values(spark_tmp_table_factory): """Test INSERT OVERWRITE on unpartitioned tables with VALUES syntax.""" base_table_name = get_full_table_name(spark_tmp_table_factory) @@ -225,7 +226,8 @@ def test_insert_overwrite_partitioned_table(spark_tmp_table_factory, partition_c @iceberg @datagen_overrides(seed=0, reason='https://github.com/NVIDIA/spark-rapids-jni/issues/4016') @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize("partition_col_sql", overwrite_static_partition_transforms) def test_insert_overwrite_partitioned_table_full_coverage(spark_tmp_table_factory, partition_col_sql): """Sanity-check INSERT OVERWRITE against a few partition transforms distinct @@ -236,7 +238,8 @@ def test_insert_overwrite_partitioned_table_full_coverage(spark_tmp_table_factor @iceberg @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") def test_insert_overwrite_unpartitioned_table_nested_types(spark_tmp_table_factory): """Test INSERT OVERWRITE with Iceberg-native nested types on GPU.""" table_prop = {"format-version": "2"} @@ -282,7 +285,8 @@ def overwrite_data(spark, table_name): @iceberg @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") def test_insert_overwrite_unpartitioned_table_all_cols(spark_tmp_table_factory): """Test INSERT OVERWRITE on unpartitioned table with all Iceberg write types on GPU.""" table_prop = {"format-version": "2"} @@ -328,7 +332,8 @@ def overwrite_data(spark, table_name): @iceberg @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") def test_insert_overwrite_partitioned_table_nested_types(spark_tmp_table_factory): """Test INSERT OVERWRITE on partitioned table with Iceberg-native nested types on GPU.""" table_prop = {"format-version": "2"} @@ -377,7 +382,8 @@ def overwrite_data(spark, table_name): @iceberg @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") def test_insert_overwrite_partitioned_table_all_cols(spark_tmp_table_factory): """Test INSERT OVERWRITE on partitioned table with all Iceberg write types on GPU.""" table_prop = {"format-version": "2"} @@ -427,7 +433,8 @@ def overwrite_data(spark, table_name): @iceberg @ignore_order(local=True) @allow_non_gpu('OverwriteByExpressionExec', 'ShuffleExchangeExec', 'SortExec', 'ProjectExec') -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize("file_format", ["orc", "avro"], ids=lambda x: f"file_format={x}") def test_insert_overwrite_table_unsupported_file_format_fallback( spark_tmp_table_factory, file_format): @@ -463,7 +470,8 @@ def overwrite_data(spark, table_name: str): @iceberg @ignore_order(local=True) @allow_non_gpu('OverwriteByExpressionExec', 'ShuffleExchangeExec', 'SortExec', 'ProjectExec') -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize("conf_key", ["spark.rapids.sql.format.iceberg.enabled", "spark.rapids.sql.format.iceberg.write.enabled"], ids=lambda x: f"{x}=False") @@ -500,7 +508,8 @@ def overwrite_data(spark, table_name: str): @iceberg @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") def test_insert_overwrite_static_after_drop_partition_field(spark_tmp_table_factory): """Test INSERT OVERWRITE (static mode) on table after dropping a partition field (void transform). @@ -556,7 +565,8 @@ def overwrite_data(spark, table_name): @iceberg @ignore_order(local=True) @allow_non_gpu('ShuffleExchangeExec') -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") def test_insert_overwrite_static_df_api_truncate_string(spark_tmp_table_factory): """Test static overwrite via DataFrame writeTo().overwrite() API with truncate(5, string_col) partitioning. Verifies GPU writes produce Parquet files with correct Iceberg field IDs @@ -614,7 +624,8 @@ def overwrite_data(spark, table_name): @iceberg @datagen_overrides(seed=0, reason='https://github.com/NVIDIA/spark-rapids-jni/issues/4016') @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") def test_insert_overwrite_partitioned_table_fanout_enabled(spark_tmp_table_factory): # Use bucket(2, ...) to keep partition count low and avoid OOM from Iceberg's FanoutDataWriter. _do_test_insert_overwrite_partitioned_table( diff --git a/integration_tests/src/main/python/iceberg/iceberg_rtas_test.py b/integration_tests/src/main/python/iceberg/iceberg_rtas_test.py index dd3af2518c1..db254a5c658 100644 --- a/integration_tests/src/main/python/iceberg/iceberg_rtas_test.py +++ b/integration_tests/src/main/python/iceberg/iceberg_rtas_test.py @@ -17,7 +17,7 @@ import pytest from asserts import assert_equal_with_local_sort, assert_gpu_fallback_collect -from conftest import is_iceberg_remote_catalog +from conftest import is_iceberg_remote_catalog, is_iceberg_test_fast_run from data_gen import gen_df, copy_and_update from iceberg import (create_iceberg_table, iceberg_base_table_cols, @@ -164,7 +164,8 @@ def test_rtas_partitioned_table(spark_tmp_table_factory, partition_col_sql): @iceberg @datagen_overrides(seed=0, reason='https://github.com/NVIDIA/spark-rapids-jni/issues/4016') @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize("partition_col_sql", rtas_partition_transforms) @allow_non_gpu_conditional(is_spark_400_or_later(), "EmptyRelationExec") def test_rtas_partitioned_table_full_coverage(spark_tmp_table_factory, partition_col_sql): @@ -205,7 +206,8 @@ def test_create_or_replace_table(spark_tmp_table_factory): @iceberg @ignore_order(local=True) @allow_non_gpu('AtomicReplaceTableAsSelectExec', 'AppendDataExec') -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize("file_format", ["orc", "avro"], ids=lambda x: f"file_format={x}") @allow_non_gpu_conditional(is_spark_400_or_later(), "EmptyRelationExec") def test_rtas_unsupported_file_format_fallback(spark_tmp_table_factory, @@ -236,7 +238,8 @@ def run_rtas(spark): @iceberg @ignore_order(local=True) @allow_non_gpu('AtomicReplaceTableAsSelectExec', 'AppendDataExec') -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize("conf_key", ["spark.rapids.sql.format.iceberg.enabled", "spark.rapids.sql.format.iceberg.write.enabled"], ids=lambda x: f"{x}=False") @@ -268,7 +271,8 @@ def run_rtas(spark): @iceberg @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") def test_rtas_unpartitioned_table_nested_types(spark_tmp_table_factory): table_prop = { "format-version": "2" @@ -283,7 +287,8 @@ def test_rtas_unpartitioned_table_nested_types(spark_tmp_table_factory): @iceberg @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @allow_non_gpu_conditional(is_spark_400_or_later(), "EmptyRelationExec") def test_rtas_unpartitioned_table_all_cols(spark_tmp_table_factory): """Test RTAS on unpartitioned table with all Iceberg write types on GPU.""" @@ -300,7 +305,8 @@ def test_rtas_unpartitioned_table_all_cols(spark_tmp_table_factory): @iceberg @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") def test_rtas_partitioned_table_nested_types(spark_tmp_table_factory): table_prop = { "format-version": "2" @@ -318,7 +324,8 @@ def test_rtas_partitioned_table_nested_types(spark_tmp_table_factory): @iceberg @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @allow_non_gpu_conditional(is_spark_400_or_later(), "EmptyRelationExec") def test_rtas_partitioned_table_all_cols(spark_tmp_table_factory): """Test RTAS on partitioned table with all Iceberg write types on GPU.""" @@ -339,7 +346,8 @@ def test_rtas_partitioned_table_all_cols(spark_tmp_table_factory): @iceberg @ignore_order(local=True) @allow_non_gpu('AppendDataExec') -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize("partition_table", [True, False], ids=lambda x: f"partition_table={x}") @allow_non_gpu_conditional(is_spark_400_or_later(), "EmptyRelationExec") def test_rtas_from_values(spark_tmp_table_factory, @@ -410,7 +418,8 @@ def test_rtas_aqe(spark_tmp_table_factory, partition_col_sql): @iceberg @datagen_overrides(seed=0, reason='https://github.com/NVIDIA/spark-rapids-jni/issues/4016') @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") def test_rtas_partitioned_table_fanout_enabled(spark_tmp_table_factory): # Use bucket(2, ...) to keep partition count low and avoid OOM from Iceberg's FanoutDataWriter. _do_test_rtas_partitioned_table( diff --git a/integration_tests/src/main/python/iceberg/iceberg_update_test.py b/integration_tests/src/main/python/iceberg/iceberg_update_test.py index c30f1384783..adb98cff53d 100644 --- a/integration_tests/src/main/python/iceberg/iceberg_update_test.py +++ b/integration_tests/src/main/python/iceberg/iceberg_update_test.py @@ -16,7 +16,7 @@ from asserts import assert_equal_with_local_sort, assert_gpu_and_cpu_are_equal_collect, \ assert_gpu_fallback_write_sql -from conftest import is_iceberg_remote_catalog +from conftest import is_iceberg_remote_catalog, is_iceberg_test_fast_run from data_gen import * from iceberg import (create_iceberg_table, get_full_table_name, iceberg_write_enabled_conf, iceberg_base_table_cols, iceberg_gens_list, iceberg_nested_write_gens_list, @@ -191,7 +191,8 @@ def setup_iceberg_table(spark): @iceberg @ignore_order(local=True) @pytest.mark.datagen_overrides(seed=UPDATE_TEST_SEED, reason=UPDATE_TEST_SEED_OVERRIDE_REASON) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize('update_mode', ['copy-on-write', 'merge-on-read']) @allow_non_gpu_conditional(is_spark_400_or_later(), "EmptyRelationExec") def test_iceberg_update_unpartitioned_table_multiple_columns(spark_tmp_table_factory, update_mode): @@ -231,7 +232,8 @@ def test_iceberg_update_partitioned_table_single_column(spark_tmp_table_factory, @datagen_overrides(seed=0, reason='https://github.com/NVIDIA/spark-rapids-jni/issues/4016') @ignore_order(local=True) @pytest.mark.datagen_overrides(seed=UPDATE_TEST_SEED, reason=UPDATE_TEST_SEED_OVERRIDE_REASON) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize("partition_col_sql,update_mode", update_partition_transforms_distributed) @allow_non_gpu_conditional(is_spark_400_or_later(), "EmptyRelationExec") def test_iceberg_update_partitioned_table_single_column_full_coverage(spark_tmp_table_factory, update_mode, partition_col_sql): @@ -244,7 +246,8 @@ def test_iceberg_update_partitioned_table_single_column_full_coverage(spark_tmp_ @iceberg @ignore_order(local=True) @pytest.mark.datagen_overrides(seed=UPDATE_TEST_SEED, reason=UPDATE_TEST_SEED_OVERRIDE_REASON) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize('update_mode', ['copy-on-write', 'merge-on-read']) @allow_non_gpu_conditional(is_spark_400_or_later(), "EmptyRelationExec") def test_iceberg_update_partitioned_table_multiple_columns(spark_tmp_table_factory, update_mode): @@ -260,7 +263,8 @@ def test_iceberg_update_partitioned_table_multiple_columns(spark_tmp_table_facto @iceberg @ignore_order(local=True) @disable_ansi_mode -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @allow_non_gpu_conditional(is_spark_400_or_later(), "EmptyRelationExec") def test_iceberg_update_mor_then_select_count(spark_tmp_table_factory): """Test UPDATE with merge-on-read mode, then select count with the same update filter. @@ -305,7 +309,8 @@ def _query_count(spark, table_name): @iceberg @ignore_order(local=True) @pytest.mark.datagen_overrides(seed=UPDATE_TEST_SEED, reason=UPDATE_TEST_SEED_OVERRIDE_REASON) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize('update_mode,fallback_exec', [ pytest.param('copy-on-write', 'ReplaceDataExec', id='cow'), pytest.param('merge-on-read', 'WriteDeltaExec', id='mor') @@ -343,7 +348,8 @@ def read_func(spark, table_name): @iceberg @ignore_order(local=True) @pytest.mark.datagen_overrides(seed=UPDATE_TEST_SEED, reason=UPDATE_TEST_SEED_OVERRIDE_REASON) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize('update_mode,fallback_exec', [ pytest.param('copy-on-write', 'ReplaceDataExec', id='cow'), pytest.param('merge-on-read', 'WriteDeltaExec', id='mor') @@ -414,7 +420,8 @@ def read_func(spark, table_name): @iceberg @ignore_order(local=True) @pytest.mark.datagen_overrides(seed=UPDATE_TEST_SEED, reason=UPDATE_TEST_SEED_OVERRIDE_REASON) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @allow_non_gpu_conditional(is_spark_400_or_later(), "EmptyRelationExec") @pytest.mark.parametrize('update_mode', ['copy-on-write', 'merge-on-read']) def test_iceberg_update_nested_types(spark_tmp_table_factory, update_mode): @@ -433,7 +440,8 @@ def test_iceberg_update_nested_types(spark_tmp_table_factory, update_mode): @iceberg @ignore_order(local=True) @pytest.mark.datagen_overrides(seed=UPDATE_TEST_SEED, reason=UPDATE_TEST_SEED_OVERRIDE_REASON) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.parametrize('update_mode,fallback_exec', [ pytest.param('copy-on-write', 'ReplaceDataExec', id='cow'), pytest.param('merge-on-read', 'WriteDeltaExec', id='mor') @@ -471,7 +479,8 @@ def read_func(spark, table_name): @iceberg @ignore_order(local=True) @pytest.mark.datagen_overrides(seed=UPDATE_TEST_SEED, reason=UPDATE_TEST_SEED_OVERRIDE_REASON) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @allow_non_gpu_conditional(is_spark_400_or_later(), "EmptyRelationExec") def test_iceberg_update_mor_fallback_writedelta_disabled(spark_tmp_table_factory): """Test merge-on-read UPDATE falls back when WriteDeltaExec is disabled @@ -554,7 +563,8 @@ def update_table(spark, table_name): @iceberg @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") @pytest.mark.datagen_overrides(seed=UPDATE_TEST_SEED, reason=UPDATE_TEST_SEED_OVERRIDE_REASON) @pytest.mark.parametrize('update_mode', ['copy-on-write', 'merge-on-read']) @allow_non_gpu_conditional(is_spark_400_or_later(), "EmptyRelationExec") @@ -603,7 +613,8 @@ def do_update(spark, table_name): @iceberg @datagen_overrides(seed=0, reason='https://github.com/NVIDIA/spark-rapids-jni/issues/4016') @ignore_order(local=True) -@pytest.mark.skipif(is_iceberg_remote_catalog(), reason="Skip for remote catalog to reduce test time") +@pytest.mark.skipif(is_iceberg_remote_catalog() or is_iceberg_test_fast_run(), + reason="Skip for remote catalog or fast Iceberg run to reduce test time") def test_iceberg_update_partitioned_table_fanout_enabled(spark_tmp_table_factory): # Use bucket(2, ...) to keep partition count low and avoid OOM from Iceberg's FanoutDataWriter. _do_test_iceberg_update_partitioned_table_single_column( diff --git a/jenkins/get_iceberg_versions.py b/jenkins/get_iceberg_versions.py new file mode 100644 index 00000000000..7b856f4f234 --- /dev/null +++ b/jenkins/get_iceberg_versions.py @@ -0,0 +1,212 @@ +#!/usr/bin/env python3 + +# 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. + +"""Read and validate the Iceberg integration-test compatibility matrix.""" + +import argparse +import re +import sys +import xml.etree.ElementTree as ET +from pathlib import Path + +import yaml + + +REPO_ROOT = Path(__file__).resolve().parents[1] +DEFAULT_MATRIX = Path(__file__).with_name("iceberg-test-matrix.yaml") +DEFAULT_POM = REPO_ROOT / "pom.xml" +VERSION_PATTERN = re.compile(r"^[0-9]+\.[0-9]+\.[0-9]+$") +SPARK_PROPERTY_PATTERN = re.compile(r"^spark[0-9]+\.version$") + + +class MatrixError(ValueError): + pass + + +def _version_tuple(version): + if not isinstance(version, str) or not VERSION_PATTERN.fullmatch(version): + raise MatrixError(f"invalid version: {version!r}") + return tuple(int(part) for part in version.split(".")) + + +def _spark_family(version): + return ".".join(version.split(".")[:2]) + + +def read_spark_shims(pom_path): + root = ET.parse(pom_path).getroot() + namespace = {"pom": "http://maven.apache.org/POM/4.0.0"} + properties = root.find("pom:properties", namespace) + if properties is None: + raise MatrixError(f"properties not found in {pom_path}") + + versions = set() + for child in properties: + name = child.tag.rsplit("}", 1)[-1] + value = (child.text or "").strip() + if SPARK_PROPERTY_PATTERN.fullmatch(name) and VERSION_PATTERN.fullmatch(value): + versions.add(value) + return versions + + +def load_matrix(matrix_path=DEFAULT_MATRIX, pom_path=DEFAULT_POM): + with open(matrix_path, encoding="utf-8") as stream: + document = yaml.safe_load(stream) + if not isinstance(document, dict) or set(document) != {"iceberg_versions"}: + raise MatrixError("matrix must contain only an iceberg_versions list") + entries = document["iceberg_versions"] + if not isinstance(entries, list) or not entries: + raise MatrixError("iceberg_versions must be a non-empty list") + + spark_shims = read_spark_shims(pom_path) + seen_iceberg_versions = set() + for entry in entries: + if not isinstance(entry, dict) or set(entry) != { + "version", "upstream_minimums", "spark_versions"}: + raise MatrixError( + "each Iceberg entry requires version, upstream_minimums, and spark_versions") + iceberg_version = entry["version"] + _version_tuple(iceberg_version) + if iceberg_version in seen_iceberg_versions: + raise MatrixError(f"duplicate Iceberg version: {iceberg_version}") + seen_iceberg_versions.add(iceberg_version) + + minimums = entry["upstream_minimums"] + spark_versions = entry["spark_versions"] + if not isinstance(minimums, dict) or not minimums: + raise MatrixError(f"upstream_minimums for Iceberg {iceberg_version} must be a mapping") + if not isinstance(spark_versions, list) or not spark_versions: + raise MatrixError(f"spark_versions for Iceberg {iceberg_version} must be a list") + + expected_versions = set() + for family, minimum in minimums.items(): + minimum_tuple = _version_tuple(minimum) + if not isinstance(family, str) or _spark_family(minimum) != family: + raise MatrixError( + f"minimum {minimum!r} does not belong to Spark family {family!r}") + expected_versions.update( + version for version in spark_shims + if _spark_family(version) == family and _version_tuple(version) >= minimum_tuple) + + actual_versions = set() + for spark_entry in spark_versions: + if not isinstance(spark_entry, dict): + raise MatrixError(f"invalid Spark entry for Iceberg {iceberg_version}") + required_keys = {"version", "supported"} + if not required_keys.issubset(spark_entry) or not set(spark_entry).issubset( + required_keys | {"reason"}): + raise MatrixError( + f"Spark entries for Iceberg {iceberg_version} require version and supported") + spark_version = spark_entry["version"] + _version_tuple(spark_version) + if spark_version in actual_versions: + raise MatrixError( + f"duplicate Spark version {spark_version} for Iceberg {iceberg_version}") + actual_versions.add(spark_version) + if type(spark_entry["supported"]) is not bool: + raise MatrixError( + f"supported must be boolean for Iceberg {iceberg_version}, " + f"Spark {spark_version}") + reason = spark_entry.get("reason") + if not spark_entry["supported"] and (not isinstance(reason, str) or not reason.strip()): + raise MatrixError( + f"unsupported Iceberg {iceberg_version}, Spark {spark_version} needs a reason") + if spark_entry["supported"] and reason is not None: + raise MatrixError( + f"supported Iceberg {iceberg_version}, Spark {spark_version} " + "cannot have a reason") + + missing = expected_versions - actual_versions + extra = actual_versions - expected_versions + if missing or extra: + details = [] + if missing: + details.append(f"missing {', '.join(sorted(missing, key=_version_tuple))}") + if extra: + details.append(f"unexpected {', '.join(sorted(extra, key=_version_tuple))}") + raise MatrixError(f"Iceberg {iceberg_version} Spark entries: {'; '.join(details)}") + return entries + + +def supported_iceberg_versions(entries, spark_version): + _version_tuple(spark_version) + return [ + entry["version"] + for entry in entries + for spark_entry in entry["spark_versions"] + if spark_entry["version"] == spark_version and spark_entry["supported"] + ] + + +def validate_requested_versions(entries, spark_version, requested_versions): + _version_tuple(spark_version) + by_iceberg = {entry["version"]: entry for entry in entries} + for iceberg_version in requested_versions: + if iceberg_version not in by_iceberg: + raise MatrixError( + f"Iceberg version {iceberg_version} is not present in the test matrix") + spark_entry = next((item for item in by_iceberg[iceberg_version]["spark_versions"] + if item["version"] == spark_version), None) + if spark_entry is None: + raise MatrixError( + f"Iceberg {iceberg_version} is not upstream-compatible with Spark {spark_version}") + if not spark_entry["supported"]: + raise MatrixError( + f"Iceberg {iceberg_version} is not supported with Spark {spark_version}: " + f"{spark_entry['reason']}") + return requested_versions + + +def parse_args(arguments=None): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--spark-version", help="Spark version whose Iceberg versions should be listed") + parser.add_argument( + "--requested-versions", + help="comma- or whitespace-separated Iceberg versions to validate and echo") + parser.add_argument("--matrix", type=Path, default=DEFAULT_MATRIX) + parser.add_argument("--pom", type=Path, default=DEFAULT_POM) + parser.add_argument( + "--validate", action="store_true", help="validate the matrix without querying it") + args = parser.parse_args(arguments) + if not args.validate and not args.spark_version: + parser.error("--spark-version is required unless --validate is used") + if args.requested_versions and not args.spark_version: + parser.error("--requested-versions requires --spark-version") + return args + + +def main(arguments=None): + args = parse_args(arguments) + try: + entries = load_matrix(args.matrix, args.pom) + if args.validate: + return 0 + if args.requested_versions: + requested = [ + version for version in re.split(r"[\s,]+", args.requested_versions) if version] + versions = validate_requested_versions(entries, args.spark_version, requested) + else: + versions = supported_iceberg_versions(entries, args.spark_version) + print(" ".join(versions)) + return 0 + except (MatrixError, ET.ParseError, OSError, yaml.YAMLError) as error: + print(f"Iceberg test matrix error: {error}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/jenkins/iceberg-test-matrix.yaml b/jenkins/iceberg-test-matrix.yaml new file mode 100644 index 00000000000..e7e295f03df --- /dev/null +++ b/jenkins/iceberg-test-matrix.yaml @@ -0,0 +1,135 @@ +# 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. + +# The Spark versions published in Apache Iceberg's gradle/libs.versions.toml are +# treated as the minimum supported patch for each Spark minor release. Every +# cudf-spark shim at or above that minimum is recorded below. Combinations that +# Iceberg supports but cudf-spark does not currently package stay in the matrix +# with an explanation so future support gaps are visible. +iceberg_versions: + - version: "1.6.1" + upstream_minimums: + "3.3": "3.3.4" + "3.4": "3.4.3" + "3.5": "3.5.1" + spark_versions: + - version: "3.3.4" + supported: false + reason: "Iceberg GPU acceleration is not currently packaged for Spark 3.3.x" + - version: "3.4.3" + supported: false + reason: "Iceberg GPU acceleration is not currently packaged for Spark 3.4.x" + - version: "3.4.4" + supported: false + reason: "Iceberg GPU acceleration is not currently packaged for Spark 3.4.x" + - version: "3.5.1" + supported: true + - version: "3.5.2" + supported: true + - version: "3.5.3" + supported: true + - version: "3.5.4" + supported: false + reason: "The Iceberg 1.6.x integration module is not packaged for Spark 3.5.4 and later" + - version: "3.5.5" + supported: false + reason: "The Iceberg 1.6.x integration module is not packaged for Spark 3.5.4 and later" + - version: "3.5.6" + supported: false + reason: "The Iceberg 1.6.x integration module is not packaged for Spark 3.5.4 and later" + - version: "3.5.7" + supported: false + reason: "The Iceberg 1.6.x integration module is not packaged for Spark 3.5.4 and later" + - version: "3.5.8" + supported: false + reason: "The Iceberg 1.6.x integration module is not packaged for Spark 3.5.4 and later" + - version: "3.5.9" + supported: false + reason: "The Iceberg 1.6.x integration module is not packaged for Spark 3.5.4 and later" + + - version: "1.9.2" + upstream_minimums: + "3.4": "3.4.4" + "3.5": "3.5.5" + spark_versions: + - version: "3.4.4" + supported: false + reason: "Iceberg GPU acceleration is not currently packaged for Spark 3.4.x" + - version: "3.5.5" + supported: true + - version: "3.5.6" + supported: true + - version: "3.5.7" + supported: true + - version: "3.5.8" + supported: true + - version: "3.5.9" + supported: true + + - version: "1.10.1" + upstream_minimums: + "3.4": "3.4.4" + "3.5": "3.5.6" + "4.0": "4.0.0" + spark_versions: + - version: "3.4.4" + supported: false + reason: "Iceberg GPU acceleration is not currently packaged for Spark 3.4.x" + - version: "3.5.6" + supported: true + - version: "3.5.7" + supported: true + - version: "3.5.8" + supported: true + - version: "3.5.9" + supported: true + - version: "4.0.0" + supported: true + - version: "4.0.1" + supported: true + - version: "4.0.2" + supported: true + - version: "4.0.3" + supported: true + - version: "4.0.4" + supported: true + + - version: "1.11.0" + upstream_minimums: + "3.4": "3.4.4" + "3.5": "3.5.8" + "4.0": "4.0.2" + "4.1": "4.1.1" + spark_versions: + - version: "3.4.4" + supported: false + reason: "Iceberg GPU acceleration is not currently packaged for Spark 3.4.x" + - version: "3.5.8" + supported: false + reason: "The Iceberg 1.11.x integration module is not currently packaged for Spark 3.5.x" + - version: "3.5.9" + supported: false + reason: "The Iceberg 1.11.x integration module is not currently packaged for Spark 3.5.x" + - version: "4.0.2" + supported: true + - version: "4.0.3" + supported: true + - version: "4.0.4" + supported: true + - version: "4.1.1" + supported: true + - version: "4.1.2" + supported: true + - version: "4.1.3" + supported: true diff --git a/jenkins/spark-premerge-build.sh b/jenkins/spark-premerge-build.sh index bc53dea59d6..c02ee29fa3c 100755 --- a/jenkins/spark-premerge-build.sh +++ b/jenkins/spark-premerge-build.sh @@ -184,36 +184,16 @@ run_iceberg_version_detect_tests() { local scala_ver=${2:?'scala_ver is required'} echo "Running Iceberg version detection tests for Spark $spark_ver (Scala $scala_ver)..." - local iceberg_spark_ver - iceberg_spark_ver=$(echo "$spark_ver" | cut -d. -f1,2) - local spark_patch_ver - spark_patch_ver=$(echo "$spark_ver" | cut -d. -f3) - - if [[ "$iceberg_spark_ver" != "3.5" && "$iceberg_spark_ver" != "4.0" \ - && "$iceberg_spark_ver" != "4.1" ]]; then - echo "!!!! Skipping Iceberg version detection. Not supported on Spark $iceberg_spark_ver" - return 0 - fi - - # 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" - elif [[ "$iceberg_spark_ver" == "4.0" ]]; then - if [[ "$spark_patch_ver" -ge 2 ]]; then - iceberg_versions="1.10.1 1.11.0" - else - iceberg_versions="1.10.1" - fi - elif [[ "$spark_patch_ver" -le 3 ]]; then - iceberg_versions="1.6.1" - else - iceberg_versions="1.9.2 1.10.1" + iceberg_versions=$(python jenkins/get_iceberg_versions.py \ + --spark-version "$spark_ver") || return 1 + if [[ -z "$iceberg_versions" ]]; then + echo "!!!! Skipping Iceberg version detection. No supported Iceberg version for Spark $spark_ver" + return 0 fi + local iceberg_spark_ver + iceberg_spark_ver=$(echo "$spark_ver" | cut -d. -f1,2) for ICEBERG_VERSION in $iceberg_versions; do echo "!!! Running iceberg version detection test for Iceberg $ICEBERG_VERSION" EXPECTED_ICEBERG_VERSION=${ICEBERG_VERSION} \ diff --git a/jenkins/spark-tests.sh b/jenkins/spark-tests.sh index 3662dfed523..882b76f813d 100755 --- a/jenkins/spark-tests.sh +++ b/jenkins/spark-tests.sh @@ -357,73 +357,63 @@ run_iceberg_tests() { ICEBERG_SPARK_VER=$(echo "$SPARK_VER" | cut -d. -f1,2) # get the patch version of Spark SPARK_PATCH_VER=$(echo "$SPARK_VER" | cut -d. -f3) + local test_type=${1:-'default'} - 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 - - # Supported Iceberg versions per Spark patch version: - # 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.0-4.0.1 -> Iceberg 1.10.1 - # Spark 4.0.2+ -> Iceberg 1.10.1, 1.11.0 - # Spark 4.1.x -> Iceberg 1.11.0 - local supported_versions - if [[ "$ICEBERG_SPARK_VER" == "4.1" ]]; then + if [[ "$ICEBERG_SPARK_VER" == "4.0" || "$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" + echo "!!!! Skipping Iceberg tests. Spark $ICEBERG_SPARK_VER 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 - fi - if [[ "$SPARK_PATCH_VER" -ge 2 ]]; then - supported_versions="1.10.1 1.11.0" - else - supported_versions="1.10.1" - fi - elif [[ "$SPARK_PATCH_VER" -le 3 ]]; then - supported_versions="1.6.1" - else - supported_versions="1.9.2 1.10.1" fi - local test_type=${1:-'default'} - - if [[ -n "$ICEBERG_VERSIONS" ]]; then - for ver in $ICEBERG_VERSIONS; do - if ! echo "$supported_versions" | grep -qw "$ver"; then - echo "!!!! Error: Iceberg version $ver is not supported on Spark $SPARK_VER (supported: $supported_versions)" - return 1 - fi - done + local matrix_reader="$WORKSPACE/jenkins/get_iceberg_versions.py" + local supported_versions + supported_versions=$(python "$matrix_reader" --spark-version "$SPARK_VER") || return 1 + + local iceberg_versions + local user_specified_versions=false + if [[ -n "${ICEBERG_VERSIONS:-}" ]]; then + iceberg_versions=$(python "$matrix_reader" \ + --spark-version "$SPARK_VER" --requested-versions "$ICEBERG_VERSIONS") || return 1 + user_specified_versions=true echo "Using user-specified ICEBERG_VERSIONS=$ICEBERG_VERSIONS" + elif [[ -z "$supported_versions" ]]; then + echo "!!!! Skipping Iceberg tests. No supported Iceberg version for Spark $SPARK_VER" + return 0 + elif [[ "$test_type" == "default" ]]; then + # Exercise every supported local-catalog combination. Fast mode keeps the + # expanded matrix practical while retaining tests that require a local catalog. + iceberg_versions="$supported_versions" else - # Default: test one representative version per Spark patch range + # Remote catalogs keep one representative version per Spark patch range. if [[ "$ICEBERG_SPARK_VER" == "4.1" ]]; then - ICEBERG_VERSIONS="1.11.0" + iceberg_versions="1.11.0" elif [[ "$ICEBERG_SPARK_VER" == "4.0" ]]; then - ICEBERG_VERSIONS="1.10.1" + iceberg_versions="1.10.1" elif [[ "$SPARK_PATCH_VER" -le 3 ]]; then - ICEBERG_VERSIONS="1.6.1" + iceberg_versions="1.6.1" elif [[ "$SPARK_PATCH_VER" -le 6 ]]; then - ICEBERG_VERSIONS="1.9.2" + iceberg_versions="1.9.2" else - ICEBERG_VERSIONS="1.10.1" + iceberg_versions="1.10.1" fi + iceberg_versions=$(python "$matrix_reader" \ + --spark-version "$SPARK_VER" --requested-versions "$iceberg_versions") || return 1 fi - for ICEBERG_VERSION in $ICEBERG_VERSIONS; do + + local iceberg_test_fast_run=${ICEBERG_TEST_FAST_RUN:-} + if [[ "$test_type" == "default" && "$user_specified_versions" == "false" ]]; then + iceberg_test_fast_run=1 + fi + + for ICEBERG_VERSION in $iceberg_versions; do echo "Running Iceberg tests for Iceberg version $ICEBERG_VERSION" if [[ "$test_type" == "default" ]]; then echo "!!! Running iceberg tests" env \ HOST_NAME=$PROJECT_REPO_HOST \ EXPECTED_ICEBERG_VERSION=${ICEBERG_VERSION} \ + ICEBERG_TEST_FAST_RUN="${iceberg_test_fast_run}" \ PYSP_TEST_spark_driver_memory=1G \ PYSP_TEST_spark_executor_memory=2G \ PYSP_TEST_spark_jars_packages=org.apache.iceberg:iceberg-spark-runtime-${ICEBERG_SPARK_VER}_${SCALA_BINARY_VER}:${ICEBERG_VERSION} \ @@ -445,6 +435,7 @@ org.apache.iceberg:iceberg-aws-bundle:${ICEBERG_VERSION}" EXPECTED_ICEBERG_VERSION=${ICEBERG_VERSION} \ ICEBERG_EXTRA_CLASSPATH="${ICEBERG_REST_EXTRA_CLASSPATH}" \ ICEBERG_TEST_CATALOG_TYPE="rest" \ + ICEBERG_TEST_FAST_RUN="${iceberg_test_fast_run}" \ ICEBERG_TEST_REMOTE_CATALOG=1 \ PYSP_TEST_spark_driver_memory=1G \ PYSP_TEST_spark_executor_memory=2G \ @@ -503,6 +494,7 @@ com.amazonaws:aws-java-sdk-bundle:${AWS_SDK_BUNDLE_VERSION}" HOST_NAME=$PROJECT_REPO_HOST \ EXPECTED_ICEBERG_VERSION=${ICEBERG_VERSION} \ ICEBERG_EXTRA_CLASSPATH="${ICEBERG_S3TABLES_EXTRA_CLASSPATH}" \ + ICEBERG_TEST_FAST_RUN="${iceberg_test_fast_run}" \ ICEBERG_TEST_REMOTE_CATALOG=1 \ PYSP_TEST_spark_driver_memory=1G \ PYSP_TEST_spark_executor_memory=2G \ diff --git a/scripts/tests/test_get_iceberg_versions.py b/scripts/tests/test_get_iceberg_versions.py new file mode 100644 index 00000000000..5536abbed2c --- /dev/null +++ b/scripts/tests/test_get_iceberg_versions.py @@ -0,0 +1,118 @@ +#!/usr/bin/env python3 + +# 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 copy +import importlib.util +import tempfile +import unittest +from pathlib import Path + +import yaml + + +REPO_ROOT = Path(__file__).parents[2] +SCRIPT = REPO_ROOT / "jenkins" / "get_iceberg_versions.py" +MATRIX_PATH = REPO_ROOT / "jenkins" / "iceberg-test-matrix.yaml" +POM_PATH = REPO_ROOT / "pom.xml" +SPEC = importlib.util.spec_from_file_location("get_iceberg_versions", SCRIPT) +MATRIX = importlib.util.module_from_spec(SPEC) +assert SPEC.loader is not None +SPEC.loader.exec_module(MATRIX) + + +class IcebergVersionMatrixSuite(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.entries = MATRIX.load_matrix(MATRIX_PATH, POM_PATH) + + def _write_modified_matrix(self, modify): + with open(MATRIX_PATH, encoding="utf-8") as stream: + document = yaml.safe_load(stream) + document = copy.deepcopy(document) + modify(document) + temporary = tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) + with temporary: + yaml.safe_dump(document, temporary) + path = Path(temporary.name) + self.addCleanup(path.unlink) + return path + + def test_returns_all_supported_iceberg_versions(self): + expected = { + "3.5.1": ["1.6.1"], + "3.5.4": [], + "3.5.5": ["1.9.2"], + "3.5.6": ["1.9.2", "1.10.1"], + "3.5.8": ["1.9.2", "1.10.1"], + "4.0.2": ["1.10.1", "1.11.0"], + "4.1.3": ["1.11.0"], + } + for spark_version, iceberg_versions in expected.items(): + with self.subTest(spark_version=spark_version): + self.assertEqual( + iceberg_versions, + MATRIX.supported_iceberg_versions(self.entries, spark_version)) + + def test_accepts_supported_requested_versions(self): + requested = ["1.9.2", "1.10.1"] + self.assertEqual( + requested, + MATRIX.validate_requested_versions(self.entries, "3.5.8", requested)) + + def test_returns_no_versions_for_unlisted_spark(self): + self.assertEqual([], MATRIX.supported_iceberg_versions(self.entries, "3.6.0")) + + def test_reports_reason_for_known_unsupported_combination(self): + with self.assertRaisesRegex( + MATRIX.MatrixError, "not currently packaged for Spark 3.5.x"): + MATRIX.validate_requested_versions(self.entries, "3.5.8", ["1.11.0"]) + + def test_rejects_combination_below_upstream_minimum(self): + with self.assertRaisesRegex(MATRIX.MatrixError, "not upstream-compatible"): + MATRIX.validate_requested_versions(self.entries, "3.5.4", ["1.9.2"]) + + def test_rejects_unknown_iceberg_version(self): + with self.assertRaisesRegex(MATRIX.MatrixError, "not present in the test matrix"): + MATRIX.validate_requested_versions(self.entries, "4.0.2", ["2.0.0"]) + + def test_requires_reason_for_unsupported_combination(self): + def remove_reason(document): + del document["iceberg_versions"][0]["spark_versions"][0]["reason"] + + path = self._write_modified_matrix(remove_reason) + with self.assertRaisesRegex(MATRIX.MatrixError, "needs a reason"): + MATRIX.load_matrix(path, POM_PATH) + + def test_requires_every_eligible_spark_shim(self): + def remove_spark_version(document): + document["iceberg_versions"][-1]["spark_versions"].pop() + + path = self._write_modified_matrix(remove_spark_version) + with self.assertRaisesRegex(MATRIX.MatrixError, "missing 4.1.3"): + MATRIX.load_matrix(path, POM_PATH) + + def test_rejects_duplicate_spark_version(self): + def duplicate_spark_version(document): + spark_versions = document["iceberg_versions"][0]["spark_versions"] + spark_versions.append(copy.deepcopy(spark_versions[0])) + + path = self._write_modified_matrix(duplicate_spark_version) + with self.assertRaisesRegex(MATRIX.MatrixError, "duplicate Spark version 3.3.4"): + MATRIX.load_matrix(path, POM_PATH) + + +if __name__ == "__main__": + unittest.main() diff --git a/scripts/tests/test_iceberg_fast_test_selection.py b/scripts/tests/test_iceberg_fast_test_selection.py new file mode 100644 index 00000000000..05fb2e7b9b0 --- /dev/null +++ b/scripts/tests/test_iceberg_fast_test_selection.py @@ -0,0 +1,67 @@ +#!/usr/bin/env python3 + +# 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 ast +import unittest +from pathlib import Path + + +ICEBERG_TEST_DIR = Path(__file__).parents[2] / \ + "integration_tests" / "src" / "main" / "python" / "iceberg" + + +def _is_skipif(call): + return (isinstance(call.func, ast.Attribute) and call.func.attr == "skipif" and + isinstance(call.func.value, ast.Attribute) and call.func.value.attr == "mark") + + +def _called_names(expression): + return { + node.func.id + for node in ast.walk(expression) + if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) + } + + +def _reason(call): + reason = next((keyword.value for keyword in call.keywords if keyword.arg == "reason"), None) + return reason.value if isinstance(reason, ast.Constant) else "" + + +class IcebergFastTestSelectionSuite(unittest.TestCase): + def test_fast_mode_only_extends_runtime_reduction_skips(self): + runtime_reduction_skips = 0 + catalog_specific_skips = 0 + for path in ICEBERG_TEST_DIR.glob("*_test.py"): + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + for call in (node for node in ast.walk(tree) + if isinstance(node, ast.Call) and _is_skipif(node)): + names = _called_names(call.args[0]) + if "is_iceberg_remote_catalog" not in names: + continue + if "reduce test time" in _reason(call): + runtime_reduction_skips += 1 + self.assertIn("is_iceberg_test_fast_run", names, path) + else: + catalog_specific_skips += 1 + self.assertNotIn("is_iceberg_test_fast_run", names, path) + + self.assertGreater(runtime_reduction_skips, 0) + self.assertGreater(catalog_specific_skips, 0) + + +if __name__ == "__main__": + unittest.main() From dfd72acca719cf236a6743b3c3aca3569b0b4eac Mon Sep 17 00:00:00 2001 From: Ray Liu Date: Fri, 4 Sep 2026 12:13:57 +0800 Subject: [PATCH 2/3] Address Iceberg test matrix review feedback Signed-off-by: Ray Liu --- iceberg/README.md | 2 +- iceberg/iceberg-versions.yml | 135 +++++++++++ integration_tests/README.md | 2 +- jenkins/get_iceberg_versions.py | 209 +++++++++++------- jenkins/spark-tests.sh | 32 +-- scripts/tests/test_get_iceberg_versions.py | 118 ---------- .../tests/test_iceberg_fast_test_selection.py | 67 ------ 7 files changed, 262 insertions(+), 303 deletions(-) create mode 100644 iceberg/iceberg-versions.yml delete mode 100644 scripts/tests/test_get_iceberg_versions.py delete mode 100644 scripts/tests/test_iceberg_fast_test_selection.py diff --git a/iceberg/README.md b/iceberg/README.md index efc522aafe8..7e6d581beac 100644 --- a/iceberg/README.md +++ b/iceberg/README.md @@ -19,7 +19,7 @@ and the directory that contains the corresponding support code. Iceberg GPU acceleration is currently supported on Spark 3.5.x, 4.0.x, and 4.1.x. The authoritative integration-test compatibility list, including upstream-compatible combinations that are not currently packaged, is maintained in -[`jenkins/iceberg-test-matrix.yaml`](../jenkins/iceberg-test-matrix.yaml). +[`iceberg-versions.yml`](iceberg-versions.yml). For Spark 3.5.4+, both `iceberg-1-9-x` and `iceberg-1-10-x` modules are compiled into the build. The integration-test support baseline follows the Spark patch versions used to build diff --git a/iceberg/iceberg-versions.yml b/iceberg/iceberg-versions.yml new file mode 100644 index 00000000000..e7e295f03df --- /dev/null +++ b/iceberg/iceberg-versions.yml @@ -0,0 +1,135 @@ +# 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. + +# The Spark versions published in Apache Iceberg's gradle/libs.versions.toml are +# treated as the minimum supported patch for each Spark minor release. Every +# cudf-spark shim at or above that minimum is recorded below. Combinations that +# Iceberg supports but cudf-spark does not currently package stay in the matrix +# with an explanation so future support gaps are visible. +iceberg_versions: + - version: "1.6.1" + upstream_minimums: + "3.3": "3.3.4" + "3.4": "3.4.3" + "3.5": "3.5.1" + spark_versions: + - version: "3.3.4" + supported: false + reason: "Iceberg GPU acceleration is not currently packaged for Spark 3.3.x" + - version: "3.4.3" + supported: false + reason: "Iceberg GPU acceleration is not currently packaged for Spark 3.4.x" + - version: "3.4.4" + supported: false + reason: "Iceberg GPU acceleration is not currently packaged for Spark 3.4.x" + - version: "3.5.1" + supported: true + - version: "3.5.2" + supported: true + - version: "3.5.3" + supported: true + - version: "3.5.4" + supported: false + reason: "The Iceberg 1.6.x integration module is not packaged for Spark 3.5.4 and later" + - version: "3.5.5" + supported: false + reason: "The Iceberg 1.6.x integration module is not packaged for Spark 3.5.4 and later" + - version: "3.5.6" + supported: false + reason: "The Iceberg 1.6.x integration module is not packaged for Spark 3.5.4 and later" + - version: "3.5.7" + supported: false + reason: "The Iceberg 1.6.x integration module is not packaged for Spark 3.5.4 and later" + - version: "3.5.8" + supported: false + reason: "The Iceberg 1.6.x integration module is not packaged for Spark 3.5.4 and later" + - version: "3.5.9" + supported: false + reason: "The Iceberg 1.6.x integration module is not packaged for Spark 3.5.4 and later" + + - version: "1.9.2" + upstream_minimums: + "3.4": "3.4.4" + "3.5": "3.5.5" + spark_versions: + - version: "3.4.4" + supported: false + reason: "Iceberg GPU acceleration is not currently packaged for Spark 3.4.x" + - version: "3.5.5" + supported: true + - version: "3.5.6" + supported: true + - version: "3.5.7" + supported: true + - version: "3.5.8" + supported: true + - version: "3.5.9" + supported: true + + - version: "1.10.1" + upstream_minimums: + "3.4": "3.4.4" + "3.5": "3.5.6" + "4.0": "4.0.0" + spark_versions: + - version: "3.4.4" + supported: false + reason: "Iceberg GPU acceleration is not currently packaged for Spark 3.4.x" + - version: "3.5.6" + supported: true + - version: "3.5.7" + supported: true + - version: "3.5.8" + supported: true + - version: "3.5.9" + supported: true + - version: "4.0.0" + supported: true + - version: "4.0.1" + supported: true + - version: "4.0.2" + supported: true + - version: "4.0.3" + supported: true + - version: "4.0.4" + supported: true + + - version: "1.11.0" + upstream_minimums: + "3.4": "3.4.4" + "3.5": "3.5.8" + "4.0": "4.0.2" + "4.1": "4.1.1" + spark_versions: + - version: "3.4.4" + supported: false + reason: "Iceberg GPU acceleration is not currently packaged for Spark 3.4.x" + - version: "3.5.8" + supported: false + reason: "The Iceberg 1.11.x integration module is not currently packaged for Spark 3.5.x" + - version: "3.5.9" + supported: false + reason: "The Iceberg 1.11.x integration module is not currently packaged for Spark 3.5.x" + - version: "4.0.2" + supported: true + - version: "4.0.3" + supported: true + - version: "4.0.4" + supported: true + - version: "4.1.1" + supported: true + - version: "4.1.2" + supported: true + - version: "4.1.3" + supported: true diff --git a/integration_tests/README.md b/integration_tests/README.md index 3d781b5b45d..38d113253af 100644 --- a/integration_tests/README.md +++ b/integration_tests/README.md @@ -543,7 +543,7 @@ If Spark has been configured to support Iceberg then these tests can be enabled Set `ICEBERG_TEST_FAST_RUN=1` to skip redundant, high-cost cases while retaining tests that specifically require a local Hadoop catalog. CI uses this mode when it expands the supported -Iceberg and Spark combinations from `jenkins/iceberg-test-matrix.yaml`. +Iceberg and Spark combinations from `iceberg/iceberg-versions.yml`. When testing Iceberg package-private access paths, load the local Iceberg runtime jar with `ICEBERG_EXTRA_CLASSPATH` instead of `PYSP_TEST_spark_jars` or diff --git a/jenkins/get_iceberg_versions.py b/jenkins/get_iceberg_versions.py index 7b856f4f234..aa54e289d15 100644 --- a/jenkins/get_iceberg_versions.py +++ b/jenkins/get_iceberg_versions.py @@ -20,13 +20,15 @@ import re import sys import xml.etree.ElementTree as ET +from dataclasses import dataclass from pathlib import Path +from typing import Dict, List, Optional import yaml REPO_ROOT = Path(__file__).resolve().parents[1] -DEFAULT_MATRIX = Path(__file__).with_name("iceberg-test-matrix.yaml") +DEFAULT_MATRIX = REPO_ROOT / "iceberg" / "iceberg-versions.yml" DEFAULT_POM = REPO_ROOT / "pom.xml" VERSION_PATTERN = re.compile(r"^[0-9]+\.[0-9]+\.[0-9]+$") SPARK_PROPERTY_PATTERN = re.compile(r"^spark[0-9]+\.version$") @@ -62,72 +64,78 @@ def read_spark_shims(pom_path): return versions -def load_matrix(matrix_path=DEFAULT_MATRIX, pom_path=DEFAULT_POM): - with open(matrix_path, encoding="utf-8") as stream: - document = yaml.safe_load(stream) - if not isinstance(document, dict) or set(document) != {"iceberg_versions"}: - raise MatrixError("matrix must contain only an iceberg_versions list") - entries = document["iceberg_versions"] - if not isinstance(entries, list) or not entries: - raise MatrixError("iceberg_versions must be a non-empty list") +@dataclass(frozen=True) +class SparkSupport: + version: str + supported: bool + reason: Optional[str] = None - spark_shims = read_spark_shims(pom_path) - seen_iceberg_versions = set() - for entry in entries: + @classmethod + def from_dict(cls, entry, iceberg_version): + if not isinstance(entry, dict): + raise MatrixError(f"invalid Spark entry for Iceberg {iceberg_version}") + required_keys = {"version", "supported"} + if not required_keys.issubset(entry) or not set(entry).issubset( + required_keys | {"reason"}): + raise MatrixError( + f"Spark entries for Iceberg {iceberg_version} require version and supported") + + version = entry["version"] + _version_tuple(version) + supported = entry["supported"] + if type(supported) is not bool: + raise MatrixError( + f"supported must be boolean for Iceberg {iceberg_version}, Spark {version}") + reason = entry.get("reason") + if not supported and (not isinstance(reason, str) or not reason.strip()): + raise MatrixError( + f"unsupported Iceberg {iceberg_version}, Spark {version} needs a reason") + if supported and reason is not None: + raise MatrixError( + f"supported Iceberg {iceberg_version}, Spark {version} cannot have a reason") + return cls(version, supported, reason) + + +@dataclass(frozen=True) +class IcebergSupport: + version: str + spark_minor_to_patch: Dict[str, str] + spark_supports: List[SparkSupport] + + @classmethod + def from_dict(cls, entry, spark_shims): if not isinstance(entry, dict) or set(entry) != { "version", "upstream_minimums", "spark_versions"}: raise MatrixError( "each Iceberg entry requires version, upstream_minimums, and spark_versions") - iceberg_version = entry["version"] - _version_tuple(iceberg_version) - if iceberg_version in seen_iceberg_versions: - raise MatrixError(f"duplicate Iceberg version: {iceberg_version}") - seen_iceberg_versions.add(iceberg_version) - - minimums = entry["upstream_minimums"] - spark_versions = entry["spark_versions"] - if not isinstance(minimums, dict) or not minimums: - raise MatrixError(f"upstream_minimums for Iceberg {iceberg_version} must be a mapping") - if not isinstance(spark_versions, list) or not spark_versions: - raise MatrixError(f"spark_versions for Iceberg {iceberg_version} must be a list") + + version = entry["version"] + _version_tuple(version) + spark_minor_to_patch = entry["upstream_minimums"] + spark_entries = entry["spark_versions"] + if not isinstance(spark_minor_to_patch, dict) or not spark_minor_to_patch: + raise MatrixError(f"upstream_minimums for Iceberg {version} must be a mapping") + if not isinstance(spark_entries, list) or not spark_entries: + raise MatrixError(f"spark_versions for Iceberg {version} must be a list") expected_versions = set() - for family, minimum in minimums.items(): + for family, minimum in spark_minor_to_patch.items(): minimum_tuple = _version_tuple(minimum) if not isinstance(family, str) or _spark_family(minimum) != family: raise MatrixError( f"minimum {minimum!r} does not belong to Spark family {family!r}") expected_versions.update( - version for version in spark_shims - if _spark_family(version) == family and _version_tuple(version) >= minimum_tuple) + spark_version for spark_version in spark_shims + if _spark_family(spark_version) == family and + _version_tuple(spark_version) >= minimum_tuple) + spark_supports = [SparkSupport.from_dict(item, version) for item in spark_entries] actual_versions = set() - for spark_entry in spark_versions: - if not isinstance(spark_entry, dict): - raise MatrixError(f"invalid Spark entry for Iceberg {iceberg_version}") - required_keys = {"version", "supported"} - if not required_keys.issubset(spark_entry) or not set(spark_entry).issubset( - required_keys | {"reason"}): - raise MatrixError( - f"Spark entries for Iceberg {iceberg_version} require version and supported") - spark_version = spark_entry["version"] - _version_tuple(spark_version) - if spark_version in actual_versions: - raise MatrixError( - f"duplicate Spark version {spark_version} for Iceberg {iceberg_version}") - actual_versions.add(spark_version) - if type(spark_entry["supported"]) is not bool: - raise MatrixError( - f"supported must be boolean for Iceberg {iceberg_version}, " - f"Spark {spark_version}") - reason = spark_entry.get("reason") - if not spark_entry["supported"] and (not isinstance(reason, str) or not reason.strip()): + for spark_support in spark_supports: + if spark_support.version in actual_versions: raise MatrixError( - f"unsupported Iceberg {iceberg_version}, Spark {spark_version} needs a reason") - if spark_entry["supported"] and reason is not None: - raise MatrixError( - f"supported Iceberg {iceberg_version}, Spark {spark_version} " - "cannot have a reason") + f"duplicate Spark version {spark_support.version} for Iceberg {version}") + actual_versions.add(spark_support.version) missing = expected_versions - actual_versions extra = actual_versions - expected_versions @@ -137,37 +145,66 @@ def load_matrix(matrix_path=DEFAULT_MATRIX, pom_path=DEFAULT_POM): details.append(f"missing {', '.join(sorted(missing, key=_version_tuple))}") if extra: details.append(f"unexpected {', '.join(sorted(extra, key=_version_tuple))}") - raise MatrixError(f"Iceberg {iceberg_version} Spark entries: {'; '.join(details)}") - return entries - - -def supported_iceberg_versions(entries, spark_version): - _version_tuple(spark_version) - return [ - entry["version"] - for entry in entries - for spark_entry in entry["spark_versions"] - if spark_entry["version"] == spark_version and spark_entry["supported"] - ] - - -def validate_requested_versions(entries, spark_version, requested_versions): - _version_tuple(spark_version) - by_iceberg = {entry["version"]: entry for entry in entries} - for iceberg_version in requested_versions: - if iceberg_version not in by_iceberg: - raise MatrixError( - f"Iceberg version {iceberg_version} is not present in the test matrix") - spark_entry = next((item for item in by_iceberg[iceberg_version]["spark_versions"] - if item["version"] == spark_version), None) - if spark_entry is None: - raise MatrixError( - f"Iceberg {iceberg_version} is not upstream-compatible with Spark {spark_version}") - if not spark_entry["supported"]: - raise MatrixError( - f"Iceberg {iceberg_version} is not supported with Spark {spark_version}: " - f"{spark_entry['reason']}") - return requested_versions + raise MatrixError(f"Iceberg {version} Spark entries: {'; '.join(details)}") + return cls(version, spark_minor_to_patch, spark_supports) + + def support_for(self, spark_version): + return next( + (support for support in self.spark_supports if support.version == spark_version), None) + + def supports(self, spark_version): + spark_support = self.support_for(spark_version) + return spark_support is not None and spark_support.supported + + +@dataclass(frozen=True) +class IcebergVersionMatrix: + iceberg_supports: List[IcebergSupport] + + @classmethod + def load(cls, matrix_path=DEFAULT_MATRIX, pom_path=DEFAULT_POM): + with open(matrix_path, encoding="utf-8") as stream: + document = yaml.safe_load(stream) + if not isinstance(document, dict) or set(document) != {"iceberg_versions"}: + raise MatrixError("matrix must contain only an iceberg_versions list") + entries = document["iceberg_versions"] + if not isinstance(entries, list) or not entries: + raise MatrixError("iceberg_versions must be a non-empty list") + + spark_shims = read_spark_shims(pom_path) + iceberg_supports = [IcebergSupport.from_dict(entry, spark_shims) for entry in entries] + versions = [support.version for support in iceberg_supports] + if len(set(versions)) != len(versions): + duplicate = next(version for index, version in enumerate(versions) + if version in versions[:index]) + raise MatrixError(f"duplicate Iceberg version: {duplicate}") + return cls(iceberg_supports) + + def supported_iceberg_versions(self, spark_version): + _version_tuple(spark_version) + return [ + iceberg_support.version + for iceberg_support in self.iceberg_supports + if iceberg_support.supports(spark_version) + ] + + def validate_requested_versions(self, spark_version, requested_versions): + _version_tuple(spark_version) + by_version = {support.version: support for support in self.iceberg_supports} + for iceberg_version in requested_versions: + if iceberg_version not in by_version: + raise MatrixError( + f"Iceberg version {iceberg_version} is not present in the test matrix") + spark_support = by_version[iceberg_version].support_for(spark_version) + if spark_support is None: + raise MatrixError( + f"Iceberg {iceberg_version} is not upstream-compatible with " + f"Spark {spark_version}") + if not spark_support.supported: + raise MatrixError( + f"Iceberg {iceberg_version} is not supported with Spark {spark_version}: " + f"{spark_support.reason}") + return requested_versions def parse_args(arguments=None): @@ -192,15 +229,15 @@ def parse_args(arguments=None): def main(arguments=None): args = parse_args(arguments) try: - entries = load_matrix(args.matrix, args.pom) + matrix = IcebergVersionMatrix.load(args.matrix, args.pom) if args.validate: return 0 if args.requested_versions: requested = [ version for version in re.split(r"[\s,]+", args.requested_versions) if version] - versions = validate_requested_versions(entries, args.spark_version, requested) + versions = matrix.validate_requested_versions(args.spark_version, requested) else: - versions = supported_iceberg_versions(entries, args.spark_version) + versions = matrix.supported_iceberg_versions(args.spark_version) print(" ".join(versions)) return 0 except (MatrixError, ET.ParseError, OSError, yaml.YAMLError) as error: diff --git a/jenkins/spark-tests.sh b/jenkins/spark-tests.sh index 882b76f813d..bc56c642c82 100755 --- a/jenkins/spark-tests.sh +++ b/jenkins/spark-tests.sh @@ -355,8 +355,6 @@ run_delta_lake_tests() { run_iceberg_tests() { # get the major/minor version of Spark ICEBERG_SPARK_VER=$(echo "$SPARK_VER" | cut -d. -f1,2) - # get the patch version of Spark - SPARK_PATCH_VER=$(echo "$SPARK_VER" | cut -d. -f3) local test_type=${1:-'default'} if [[ "$ICEBERG_SPARK_VER" == "4.0" || "$ICEBERG_SPARK_VER" == "4.1" ]]; then @@ -371,39 +369,15 @@ run_iceberg_tests() { supported_versions=$(python "$matrix_reader" --spark-version "$SPARK_VER") || return 1 local iceberg_versions - local user_specified_versions=false if [[ -n "${ICEBERG_VERSIONS:-}" ]]; then iceberg_versions=$(python "$matrix_reader" \ --spark-version "$SPARK_VER" --requested-versions "$ICEBERG_VERSIONS") || return 1 - user_specified_versions=true echo "Using user-specified ICEBERG_VERSIONS=$ICEBERG_VERSIONS" elif [[ -z "$supported_versions" ]]; then echo "!!!! Skipping Iceberg tests. No supported Iceberg version for Spark $SPARK_VER" return 0 - elif [[ "$test_type" == "default" ]]; then - # Exercise every supported local-catalog combination. Fast mode keeps the - # expanded matrix practical while retaining tests that require a local catalog. - iceberg_versions="$supported_versions" else - # Remote catalogs keep one representative version per Spark patch range. - 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" - elif [[ "$SPARK_PATCH_VER" -le 6 ]]; then - iceberg_versions="1.9.2" - else - iceberg_versions="1.10.1" - fi - iceberg_versions=$(python "$matrix_reader" \ - --spark-version "$SPARK_VER" --requested-versions "$iceberg_versions") || return 1 - fi - - local iceberg_test_fast_run=${ICEBERG_TEST_FAST_RUN:-} - if [[ "$test_type" == "default" && "$user_specified_versions" == "false" ]]; then - iceberg_test_fast_run=1 + iceberg_versions="$supported_versions" fi for ICEBERG_VERSION in $iceberg_versions; do @@ -413,7 +387,7 @@ run_iceberg_tests() { env \ HOST_NAME=$PROJECT_REPO_HOST \ EXPECTED_ICEBERG_VERSION=${ICEBERG_VERSION} \ - ICEBERG_TEST_FAST_RUN="${iceberg_test_fast_run}" \ + ICEBERG_TEST_FAST_RUN=1 \ PYSP_TEST_spark_driver_memory=1G \ PYSP_TEST_spark_executor_memory=2G \ PYSP_TEST_spark_jars_packages=org.apache.iceberg:iceberg-spark-runtime-${ICEBERG_SPARK_VER}_${SCALA_BINARY_VER}:${ICEBERG_VERSION} \ @@ -435,7 +409,6 @@ org.apache.iceberg:iceberg-aws-bundle:${ICEBERG_VERSION}" EXPECTED_ICEBERG_VERSION=${ICEBERG_VERSION} \ ICEBERG_EXTRA_CLASSPATH="${ICEBERG_REST_EXTRA_CLASSPATH}" \ ICEBERG_TEST_CATALOG_TYPE="rest" \ - ICEBERG_TEST_FAST_RUN="${iceberg_test_fast_run}" \ ICEBERG_TEST_REMOTE_CATALOG=1 \ PYSP_TEST_spark_driver_memory=1G \ PYSP_TEST_spark_executor_memory=2G \ @@ -494,7 +467,6 @@ com.amazonaws:aws-java-sdk-bundle:${AWS_SDK_BUNDLE_VERSION}" HOST_NAME=$PROJECT_REPO_HOST \ EXPECTED_ICEBERG_VERSION=${ICEBERG_VERSION} \ ICEBERG_EXTRA_CLASSPATH="${ICEBERG_S3TABLES_EXTRA_CLASSPATH}" \ - ICEBERG_TEST_FAST_RUN="${iceberg_test_fast_run}" \ ICEBERG_TEST_REMOTE_CATALOG=1 \ PYSP_TEST_spark_driver_memory=1G \ PYSP_TEST_spark_executor_memory=2G \ diff --git a/scripts/tests/test_get_iceberg_versions.py b/scripts/tests/test_get_iceberg_versions.py deleted file mode 100644 index 5536abbed2c..00000000000 --- a/scripts/tests/test_get_iceberg_versions.py +++ /dev/null @@ -1,118 +0,0 @@ -#!/usr/bin/env python3 - -# 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 copy -import importlib.util -import tempfile -import unittest -from pathlib import Path - -import yaml - - -REPO_ROOT = Path(__file__).parents[2] -SCRIPT = REPO_ROOT / "jenkins" / "get_iceberg_versions.py" -MATRIX_PATH = REPO_ROOT / "jenkins" / "iceberg-test-matrix.yaml" -POM_PATH = REPO_ROOT / "pom.xml" -SPEC = importlib.util.spec_from_file_location("get_iceberg_versions", SCRIPT) -MATRIX = importlib.util.module_from_spec(SPEC) -assert SPEC.loader is not None -SPEC.loader.exec_module(MATRIX) - - -class IcebergVersionMatrixSuite(unittest.TestCase): - @classmethod - def setUpClass(cls): - cls.entries = MATRIX.load_matrix(MATRIX_PATH, POM_PATH) - - def _write_modified_matrix(self, modify): - with open(MATRIX_PATH, encoding="utf-8") as stream: - document = yaml.safe_load(stream) - document = copy.deepcopy(document) - modify(document) - temporary = tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) - with temporary: - yaml.safe_dump(document, temporary) - path = Path(temporary.name) - self.addCleanup(path.unlink) - return path - - def test_returns_all_supported_iceberg_versions(self): - expected = { - "3.5.1": ["1.6.1"], - "3.5.4": [], - "3.5.5": ["1.9.2"], - "3.5.6": ["1.9.2", "1.10.1"], - "3.5.8": ["1.9.2", "1.10.1"], - "4.0.2": ["1.10.1", "1.11.0"], - "4.1.3": ["1.11.0"], - } - for spark_version, iceberg_versions in expected.items(): - with self.subTest(spark_version=spark_version): - self.assertEqual( - iceberg_versions, - MATRIX.supported_iceberg_versions(self.entries, spark_version)) - - def test_accepts_supported_requested_versions(self): - requested = ["1.9.2", "1.10.1"] - self.assertEqual( - requested, - MATRIX.validate_requested_versions(self.entries, "3.5.8", requested)) - - def test_returns_no_versions_for_unlisted_spark(self): - self.assertEqual([], MATRIX.supported_iceberg_versions(self.entries, "3.6.0")) - - def test_reports_reason_for_known_unsupported_combination(self): - with self.assertRaisesRegex( - MATRIX.MatrixError, "not currently packaged for Spark 3.5.x"): - MATRIX.validate_requested_versions(self.entries, "3.5.8", ["1.11.0"]) - - def test_rejects_combination_below_upstream_minimum(self): - with self.assertRaisesRegex(MATRIX.MatrixError, "not upstream-compatible"): - MATRIX.validate_requested_versions(self.entries, "3.5.4", ["1.9.2"]) - - def test_rejects_unknown_iceberg_version(self): - with self.assertRaisesRegex(MATRIX.MatrixError, "not present in the test matrix"): - MATRIX.validate_requested_versions(self.entries, "4.0.2", ["2.0.0"]) - - def test_requires_reason_for_unsupported_combination(self): - def remove_reason(document): - del document["iceberg_versions"][0]["spark_versions"][0]["reason"] - - path = self._write_modified_matrix(remove_reason) - with self.assertRaisesRegex(MATRIX.MatrixError, "needs a reason"): - MATRIX.load_matrix(path, POM_PATH) - - def test_requires_every_eligible_spark_shim(self): - def remove_spark_version(document): - document["iceberg_versions"][-1]["spark_versions"].pop() - - path = self._write_modified_matrix(remove_spark_version) - with self.assertRaisesRegex(MATRIX.MatrixError, "missing 4.1.3"): - MATRIX.load_matrix(path, POM_PATH) - - def test_rejects_duplicate_spark_version(self): - def duplicate_spark_version(document): - spark_versions = document["iceberg_versions"][0]["spark_versions"] - spark_versions.append(copy.deepcopy(spark_versions[0])) - - path = self._write_modified_matrix(duplicate_spark_version) - with self.assertRaisesRegex(MATRIX.MatrixError, "duplicate Spark version 3.3.4"): - MATRIX.load_matrix(path, POM_PATH) - - -if __name__ == "__main__": - unittest.main() diff --git a/scripts/tests/test_iceberg_fast_test_selection.py b/scripts/tests/test_iceberg_fast_test_selection.py deleted file mode 100644 index 05fb2e7b9b0..00000000000 --- a/scripts/tests/test_iceberg_fast_test_selection.py +++ /dev/null @@ -1,67 +0,0 @@ -#!/usr/bin/env python3 - -# 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 ast -import unittest -from pathlib import Path - - -ICEBERG_TEST_DIR = Path(__file__).parents[2] / \ - "integration_tests" / "src" / "main" / "python" / "iceberg" - - -def _is_skipif(call): - return (isinstance(call.func, ast.Attribute) and call.func.attr == "skipif" and - isinstance(call.func.value, ast.Attribute) and call.func.value.attr == "mark") - - -def _called_names(expression): - return { - node.func.id - for node in ast.walk(expression) - if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) - } - - -def _reason(call): - reason = next((keyword.value for keyword in call.keywords if keyword.arg == "reason"), None) - return reason.value if isinstance(reason, ast.Constant) else "" - - -class IcebergFastTestSelectionSuite(unittest.TestCase): - def test_fast_mode_only_extends_runtime_reduction_skips(self): - runtime_reduction_skips = 0 - catalog_specific_skips = 0 - for path in ICEBERG_TEST_DIR.glob("*_test.py"): - tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) - for call in (node for node in ast.walk(tree) - if isinstance(node, ast.Call) and _is_skipif(node)): - names = _called_names(call.args[0]) - if "is_iceberg_remote_catalog" not in names: - continue - if "reduce test time" in _reason(call): - runtime_reduction_skips += 1 - self.assertIn("is_iceberg_test_fast_run", names, path) - else: - catalog_specific_skips += 1 - self.assertNotIn("is_iceberg_test_fast_run", names, path) - - self.assertGreater(runtime_reduction_skips, 0) - self.assertGreater(catalog_specific_skips, 0) - - -if __name__ == "__main__": - unittest.main() From 843ae926e7eba2370f127e9632058f87b44390cd Mon Sep 17 00:00:00 2001 From: Ray Liu Date: Fri, 4 Sep 2026 14:49:37 +0800 Subject: [PATCH 3/3] Clarify Iceberg matrix derivation Signed-off-by: Ray Liu --- iceberg/iceberg-versions.yml | 17 ++++++++++++----- 1 file changed, 12 insertions(+), 5 deletions(-) diff --git a/iceberg/iceberg-versions.yml b/iceberg/iceberg-versions.yml index e7e295f03df..06cc4b7281b 100644 --- a/iceberg/iceberg-versions.yml +++ b/iceberg/iceberg-versions.yml @@ -12,11 +12,18 @@ # See the License for the specific language governing permissions and # limitations under the License. -# The Spark versions published in Apache Iceberg's gradle/libs.versions.toml are -# treated as the minimum supported patch for each Spark minor release. Every -# cudf-spark shim at or above that minimum is recorded below. Combinations that -# Iceberg supports but cudf-spark does not currently package stay in the matrix -# with an explanation so future support gaps are visible. +# Each entry describes one Apache Iceberg runtime version tested by cudf-spark. +# For that Iceberg release, upstream_minimums is copied from the Spark versions +# in Apache Iceberg's gradle/libs.versions.toml. Each key is a Spark major/minor +# family and its value is the patch release that Iceberg builds and tests against; +# cudf-spark treats that patch as the minimum upstream-compatible version. +# +# spark_versions is computed from the spark*.version properties in the root +# pom.xml. For each family in upstream_minimums, it lists every cudf-spark shim +# whose patch version is greater than or equal to the upstream minimum. A shim is +# supported when cudf-spark packages the corresponding Iceberg integration module. +# Upstream-compatible shims that are not packaged remain listed with supported: +# false and a reason, making current support gaps explicit. iceberg_versions: - version: "1.6.1" upstream_minimums: