[auto-merge] release/26.08 to main [skip ci] [bot] (NVIDIA#15482)

auto-merge triggered by github actions on `release/26.08` to create a PR
keeping `main` up-to-date. If this PR is unable to be merged due to
conflicts, it will remain open until manually fix.

Author: nvauto

integration_tests/src/main/python/delta_lake_test.py

@@ -23,7 +23,7 @@
 from parquet_test_utils import parquet_row_group_midpoints
 from spark_session import with_cpu_session, with_gpu_session, is_databricks_runtime, \
     is_spark_320_or_later, is_spark_340_or_later, supports_delta_lake_deletion_vectors, is_spark_401_or_later, \
-    is_before_spark_353, is_databricks173_or_later
+    gpu_supports_delta_dv_scan, is_before_spark_353, is_databricks173_or_later
 
 _conf = {'spark.rapids.sql.explain': 'ALL'}
 
@@ -198,15 +198,8 @@ def test_delta_deletion_vector_read(spark_tmp_path, chunk_size, use_cdf, dv_pred
 cdf_fallback = ["RowDataSourceScanExec"]
 
 
-@allow_non_gpu(*cdf_fallback, *delta_meta_allow)
-@delta_lake
-@ignore_order(local=True)
-@pytest.mark.parametrize("chunk_size", ["2000", "4000", None], ids=idfn)
-@pytest.mark.parametrize("parquet_reader_type", ["PERFILE", "COALESCING", "MULTITHREADED"], ids=idfn)
-@pytest.mark.skipif(not supports_delta_lake_deletion_vectors(),
-                    reason="Delta Lake deletion vector support is required")
-@pytest.mark.skipif(is_databricks_runtime(), reason="https://github.com/NVIDIA/cudf-spark/issues/15365")
-def test_delta_deletion_vector_read_with_cdf(spark_tmp_path, chunk_size, parquet_reader_type):
+def _test_delta_deletion_vector_read_with_cdf(
+        spark_tmp_path, chunk_size, parquet_reader_type, expect_fallback):
     data_path = spark_tmp_path + "/DELTA_DATA"
     conf = {"spark.databricks.delta.delete.deletionVectors.persistent": "true",
             "spark.rapids.sql.reader.chunked": f"{chunk_size is not None}",
@@ -220,10 +213,46 @@ def test_delta_deletion_vector_read_with_cdf(spark_tmp_path, chunk_size, parquet
         "DELETE FROM delta.`{}` WHERE a = 1".format(data_path)
     ])
 
-    assert_gpu_and_cpu_are_equal_collect(
-        lambda spark: read_delta_path_with_cdf(spark, data_path),
-        conf=conf
-    )
+    def read_cdf(spark):
+        return read_delta_path_with_cdf(spark, data_path)
+
+    if expect_fallback:
+        # DeltaCDFRelation hides its internal file scan behind this V1 CPU scan.
+        assert_gpu_fallback_collect(
+            read_cdf,
+            "RowDataSourceScanExec",
+            conf=conf)
+    else:
+        assert_gpu_and_cpu_are_equal_collect(read_cdf, conf=conf)
+
+
+@allow_non_gpu(*cdf_fallback, *delta_meta_allow)
+@delta_lake
+@ignore_order(local=True)
+@pytest.mark.parametrize("chunk_size", ["2000", "4000", None], ids=idfn)
+@pytest.mark.parametrize("parquet_reader_type", ["PERFILE", "COALESCING", "MULTITHREADED"], ids=idfn)
+@pytest.mark.skipif(not gpu_supports_delta_dv_scan(),
+                    reason="GPU Delta deletion vector scan support is required")
+@pytest.mark.skipif(is_databricks_runtime(), reason="https://github.com/NVIDIA/cudf-spark/issues/15365")
+def test_delta_deletion_vector_read_with_cdf(spark_tmp_path, chunk_size, parquet_reader_type):
+    _test_delta_deletion_vector_read_with_cdf(
+        spark_tmp_path, chunk_size, parquet_reader_type, expect_fallback=False)
+
+
+@allow_non_gpu("ColumnarToRowExec", *cdf_fallback, *delta_meta_allow)
+@delta_lake
+@ignore_order(local=True)
+@pytest.mark.parametrize("chunk_size", ["2000", "4000", None], ids=idfn)
+@pytest.mark.parametrize("parquet_reader_type", ["PERFILE", "COALESCING", "MULTITHREADED"], ids=idfn)
+@pytest.mark.skipif(not supports_delta_lake_deletion_vectors(),
+                    reason="Delta Lake deletion vector feature is required")
+@pytest.mark.skipif(gpu_supports_delta_dv_scan(),
+                    reason="GPU Delta deletion vector scans are supported")
+@pytest.mark.skipif(is_databricks_runtime(), reason="https://github.com/NVIDIA/cudf-spark/issues/15365")
+def test_delta_deletion_vector_read_with_cdf_fallback(
+        spark_tmp_path, chunk_size, parquet_reader_type):
+    _test_delta_deletion_vector_read_with_cdf(
+        spark_tmp_path, chunk_size, parquet_reader_type, expect_fallback=True)
 
 
 def _create_delta_cdf_mixed_filter_files(spark, data_path, second_file_partition):
@@ -591,10 +620,10 @@ def setup_tables(spark):
         conf=conf)
 
 
-@allow_non_gpu("FileSourceScanExec", "ColumnarToRowExec", *delta_meta_allow)
+@allow_non_gpu(*delta_meta_allow)
 @delta_lake
-@pytest.mark.skipif(not supports_delta_lake_deletion_vectors(),
-                    reason="Delta Lake deletion vector support is required")
+@pytest.mark.skipif(not gpu_supports_delta_dv_scan(),
+                    reason="GPU Delta deletion vector scan support is required")
 @pytest.mark.skipif(is_databricks_runtime(),
                     reason="This test targets the OSS multithreaded Delta reader")
 def test_delta_deletion_vector_multithreaded_combine_count_star_mixed_dv_no_dv(

integration_tests/src/main/python/spark_session.py

@@ -335,11 +335,19 @@ def is_databricks173_or_later():
     return is_databricks_version_or_later(17, 3)
 
 def supports_delta_lake_deletion_vectors():
+    """Whether the current Delta Lake runtime provides the deletion-vector feature."""
     if is_databricks_runtime():
         return is_databricks122_or_later()
     else:
         return is_spark_340_or_later()
 
+def gpu_supports_delta_dv_scan():
+    """Whether RAPIDS supports scanning Delta deletion vectors on the GPU."""
+    if is_databricks_runtime():
+        return is_databricks173_or_later()
+    else:
+        return is_spark_353_or_later()
+
 def is_support_default_values_in_schema():
     # Spark 340 + and Databricks 330 + support
     return is_spark_340_or_later() or is_databricks113_or_later()