diff --git a/iceberg/README.md b/iceberg/README.md index 191131c4cd2..7e6d581beac 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 +[`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 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/iceberg/iceberg-versions.yml b/iceberg/iceberg-versions.yml new file mode 100644 index 00000000000..06cc4b7281b --- /dev/null +++ b/iceberg/iceberg-versions.yml @@ -0,0 +1,142 @@ +# 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. + +# 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: + "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 35fd21cdcfc..38d113253af 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 `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 `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..aa54e289d15 --- /dev/null +++ b/jenkins/get_iceberg_versions.py @@ -0,0 +1,249 @@ +#!/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 dataclasses import dataclass +from pathlib import Path +from typing import Dict, List, Optional + +import yaml + + +REPO_ROOT = Path(__file__).resolve().parents[1] +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$") + + +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 + + +@dataclass(frozen=True) +class SparkSupport: + version: str + supported: bool + reason: Optional[str] = None + + @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") + + 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 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( + 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_support in spark_supports: + if spark_support.version in actual_versions: + raise MatrixError( + 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 + 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 {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): + 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: + 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 = matrix.validate_requested_versions(args.spark_version, requested) + else: + versions = matrix.supported_iceberg_versions(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/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..bc56c642c82 100755 --- a/jenkins/spark-tests.sh +++ b/jenkins/spark-tests.sh @@ -355,75 +355,39 @@ 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) - - 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 + local test_type=${1:-'default'} - # 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'} + local matrix_reader="$WORKSPACE/jenkins/get_iceberg_versions.py" + local supported_versions + supported_versions=$(python "$matrix_reader" --spark-version "$SPARK_VER") || return 1 - 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 iceberg_versions + if [[ -n "${ICEBERG_VERSIONS:-}" ]]; then + iceberg_versions=$(python "$matrix_reader" \ + --spark-version "$SPARK_VER" --requested-versions "$ICEBERG_VERSIONS") || return 1 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 else - # Default: test 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="$supported_versions" fi - for ICEBERG_VERSION in $ICEBERG_VERSIONS; do + + 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=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} \