[SkipRecovery] Re-enable mixed-delete tests on Iceberg 1.10.2+ [reduced-it] [databricks]

integration_tests/README.md

@@ -539,13 +539,16 @@ Some tests require that Apache Iceberg has been configured in the Spark environm
 properly without it. These tests assume Iceberg is not configured and are disabled by default.
 If Spark has been configured to support Iceberg then these tests can be enabled by adding the
 `--iceberg` option to the command.
+Set `EXPECTED_ICEBERG_VERSION` to the exact Iceberg runtime version whenever `--iceberg` is used;
+pytest reports a configuration error when it is missing.
 
 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
 jars on `spark.driver.extraClassPath` and `spark.executor.extraClassPath`:
 
 ```shell
+EXPECTED_ICEBERG_VERSION=1.10.1 \
 ICEBERG_EXTRA_CLASSPATH=/path/to/iceberg-spark-runtime-3.5_2.12-1.10.1.jar \
 PYSP_TEST_spark_sql_extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions \
 PYSP_TEST_spark_sql_catalog_spark__catalog=org.apache.iceberg.spark.SparkSessionCatalog \

integration_tests/src/main/python/conftest.py

@@ -395,6 +395,9 @@ def pytest_runtest_setup(item):
             pytest.skip('tests for pyarrow not configured to run')
 
 def pytest_configure(config):
+    if config.getoption('iceberg') and not os.environ.get('EXPECTED_ICEBERG_VERSION'):
+        raise pytest.UsageError(
+            "EXPECTED_ICEBERG_VERSION must be set when running Iceberg tests")
     global _runtime_env
     _runtime_env = config.getoption('runtime_env')
     global _std_input_path

integration_tests/src/main/python/iceberg/iceberg_merge_on_read_test.py

@@ -27,14 +27,22 @@
     iceberg_base_table_cols, iceberg_gens_list, get_full_table_name, \
     supports_iceberg_v3, ICEBERG_V3_UNSUPPORTED_REASON, \
     supports_iceberg_row_lineage_inheritance, \
-    ICEBERG_ROW_LINEAGE_INHERITANCE_UNSUPPORTED_REASON
+    ICEBERG_ROW_LINEAGE_INHERITANCE_UNSUPPORTED_REASON, runtime_iceberg_version
 from data_gen import disable_parquet_field_id_write, gen_df, get_datagen_seed, int_gen, \
     long_gen, string_gen
 from marks import iceberg, ignore_order, validate_execs_in_gpu_plan
-from spark_session import with_gpu_session, with_cpu_session
+from spark_session import with_gpu_session, with_cpu_session, reset_spark_session_conf
 
 pytestmark = iceberg_unsupported_mark
 
+# Iceberg 1.10.2 and 1.11.0 fix cached equality-delete records being interpreted in the wrong
+# field order: https://github.com/apache/iceberg/pull/15514 (backport: #15605).
+# Keep the quarantine for older runtimes. Iceberg runs fail during pytest configuration when the
+# runtime version is unavailable.
+_iceberg_eq_delete_cache_bug = (
+    runtime_iceberg_version is not None and
+    tuple(int(part) for part in runtime_iceberg_version.split('.')[:3]) < (1, 10, 2))
+
 
 # Eq-delete pair coverage. All 14 eligible eq-delete columns of iceberg_table_gen
 # (_c0..c3, _c6..c15; _c4 float and _c5 double are excluded by can_be_eq_delete_col)
@@ -129,13 +137,10 @@ def test_iceberg_v2_position_delete_with_url_encoded_path(spark_tmp_table_factor
 @ignore_order(local=True)
 @pytest.mark.parametrize('reader_type', rapids_reader_types)
 @pytest.mark.skipif(is_iceberg_remote_catalog(), reason = "S3tables catalog is managed")
-@pytest.mark.xfail(reason = "https://github.com/NVIDIA/spark-rapids/issues/12885")
-# When using this datagen, local run is 784 rows
-@pytest.mark.datagen_overrides(seed=1749483297, permanent=True,
-                               reason="Debug https://github.com/NVIDIA/spark-rapids/issues/12885")
+@pytest.mark.skipif(condition=_iceberg_eq_delete_cache_bug,
+                    reason="https://github.com/NVIDIA/spark-rapids/issues/12885")
 def test_iceberg_v2_mixed_deletes(spark_tmp_table_factory, spark_tmp_path, reader_type,
                                   register_iceberg_add_eq_deletes_udf):
-    # We use a fixed seed here to ensure that data deletion vector has been generated
     table_name = setup_base_iceberg_table(spark_tmp_table_factory)
     # Position deletes
     _change_table(table_name,
@@ -316,10 +321,8 @@ def setup_table(spark):
 @pytest.mark.parametrize('reader_type', rapids_reader_types)
 @pytest.mark.skipif(is_iceberg_remote_catalog(), reason = "S3tables catalog is managed")
 @pytest.mark.skipif(not supports_iceberg_v3, reason=ICEBERG_V3_UNSUPPORTED_REASON)
-@pytest.mark.xfail(reason = "https://github.com/NVIDIA/spark-rapids/issues/12885")
-# When using this datagen, local run is 784 rows
-@pytest.mark.datagen_overrides(seed=1749483297, permanent=True,
-                               reason="Debug https://github.com/NVIDIA/spark-rapids/issues/12885")
+@pytest.mark.skipif(condition=_iceberg_eq_delete_cache_bug,
+                    reason="https://github.com/NVIDIA/spark-rapids/issues/12885")
 @validate_execs_in_gpu_plan('GpuBatchScanExec')
 def test_iceberg_v3_mixed_deletes(spark_tmp_table_factory, spark_tmp_path, reader_type,
                                   register_iceberg_add_eq_deletes_udf):
@@ -371,16 +374,23 @@ def add_deletion_vector(spark):
         'spark.rapids.sql.format.parquet.reader.type': reader_type,
     }
 
-    gpu_count = with_gpu_session(lambda spark: spark.table(table_name).count(),
-                                 conf=read_conf)
-    cpu_count = with_cpu_session(lambda spark: spark.table(table_name).count(),
-                                 conf=read_conf)
-    assert gpu_count == cpu_count, f"Result count diverges, cpu: {cpu_count}, gpu: {gpu_count}"
-    logging.info(f"Count is {cpu_count}")
-
-    assert_gpu_and_cpu_are_equal_collect(
-        lambda spark: spark.table(table_name),
-        conf=read_conf)
+    # The plan validator runs before AQE finalizes this aggregate. Keep the full-row read's
+    # AQE configuration unchanged while validating GpuBatchScanExec for both projections.
+    count_conf = {**read_conf, 'spark.sql.adaptive.enabled': 'false'}
+    try:
+        gpu_count = with_gpu_session(lambda spark: spark.table(table_name).count(),
+                                     conf=count_conf)
+        cpu_count = with_cpu_session(lambda spark: spark.table(table_name).count(),
+                                     conf=count_conf)
+        assert gpu_count == cpu_count, f"Result count diverges, cpu: {cpu_count}, gpu: {gpu_count}"
+        logging.info(f"Count is {cpu_count}")
+
+        assert_gpu_and_cpu_are_equal_collect(
+            lambda spark: spark.table(table_name),
+            conf=read_conf)
+    finally:
+        # The table fixture's SHOW TABLES cleanup must not inherit the scan-only plan assertion.
+        reset_spark_session_conf()
 
 
 def _normalize_position_delete_df(df):

jenkins/spark-premerge-build.sh

@@ -239,7 +239,8 @@ run_iceberg_extra_classpath_tests() {
 
     # Loading Iceberg from extraClassPath creates the app/shim classloader split.
     echo "!!! Running targeted Iceberg extraClassPath tests for Iceberg $iceberg_version"
-    ICEBERG_EXTRA_CLASSPATH="${iceberg_runtime_jar}" \
+    EXPECTED_ICEBERG_VERSION="${iceberg_version}" \
+        ICEBERG_EXTRA_CLASSPATH="${iceberg_runtime_jar}" \
         PYSP_TEST_spark_sql_extensions="org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions" \
         PYSP_TEST_spark_sql_catalog_spark__catalog="org.apache.iceberg.spark.SparkSessionCatalog" \
         PYSP_TEST_spark_sql_catalog_spark__catalog_type="hadoop" \