DV read tests with cdf should run with spark 353+ [databricks] (NVIDIA#15462)

Fixes NVIDIA#15461.

### Description

Two new integration tests are failing in the pre-release CI. These tests
test the deletion vector read with CDF enabled. Since these tests
require the DV scan to run on GPU, they should run with Spark 3.5.3+
which are the versions the plugin accelerates the DV scan.

One related issue is the confusing name of the
`supports_delta_lake_deletion_vectors` function. The "support" can mean
different things depending on the context. In the plugin context,
supporting DV means that the DV scan runs on GPU. In the Delta Lake
context, supporting DV means that the deletion vector feature is
supported. The function in question is confusing as it does not tell
what supports deletion vectors, and thus what is the meaning of
supporting deletion vectors.

This PR fixes the problems above by adding a new function
`gpu_supports_delta_dv_scan`, which returns true if the DV scan runs on
GPU in the given runtime. The failing tests now use
`gpu_supports_delta_dv_scan` to skip themselves for Spark 3.4.x. A new
test is also added to test the fallback with Spark 3.4.x. The failing
tests and the new tests are manually tested with Spark 3.5.5 and 3.4.4.

I will make another PR to rename the confusing
`supports_delta_lake_deletion_vectors` against the main branch.

### Checklists

Documentation
- [ ] Updated for new or modified user-facing features or behaviors
- [x] No user-facing change

Testing
- [x] Added or modified tests to cover new code paths
- [ ] Covered by existing tests
(Please provide the names of the existing tests in the PR description.)
- [ ] Not required

Performance
- [ ] Tests ran and results are added in the PR description
- [ ] Issue filed with a link in the PR description
- [x] Not required

---------

Signed-off-by: Jihoon Son <ghoonson@gmail.com>

Author: jihoonson

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()