Delta table scan should be optimized when deletion vectors don't exist [databricks]

delta-lake/common/src/main/delta-33x-40x/scala/com/nvidia/spark/rapids/delta/common/GpuDeltaParquetFileFormatBase.scala

@@ -57,6 +57,14 @@ class GpuDeltaParquetFileFormatBase(
     isCDCRead: Boolean = false
   ) extends com.nvidia.spark.rapids.delta.GpuDeltaParquetFileFormat with Logging {
 
+  if (hasTablePath) {
+    logWarning("Input Delta table has deletion vectors. Optimizations such as file splitting " +
+      "and predicate pushdown are currently not supported for this table, which can lead to " +
+      "performance issues. To avoid this, consider disabling deletion vectors on the table and " +
+      "running the optimize command. See https://github.com/NVIDIA/spark-rapids/issues/13999 for " +
+      "more details about the issue.")
+  }
+
   // Validate either we have all arguments for DV enabled read or none of them.
 
   // disable optimizations
@@ -142,7 +150,11 @@ class GpuDeltaParquetFileFormatBase(
 
   override def isSplitable(sparkSession: SparkSession,
      options: Map[String, String],
-     path: Path): Boolean = optimizationsEnabled
+     path: Path): Boolean = {
+    // Disable the optimizations when there are DVs read.
+    // Note that tablePath is set only when DVs are read.
+    !hasTablePath && optimizationsEnabled
+  }
 
   def hasTablePath: Boolean = tablePath.isDefined
 

delta-lake/delta-33x/src/main/scala/com/nvidia/spark/rapids/delta/delta33x/Delta33xProvider.scala

@@ -74,8 +74,7 @@ object Delta33xProvider extends DeltaProviderBase {
 
   override protected def toGpuParquetFileFormat(fmt: DeltaParquetFileFormat): FileFormat =
     GpuDelta33xParquetFileFormat(fmt.protocol, fmt.metadata, fmt.nullableRowTrackingFields,
-      false, // we don't support splits and predicate pushdown yet
-      fmt.tablePath, fmt.isCDCRead)
+      fmt.optimizationsEnabled, fmt.tablePath, fmt.isCDCRead)
 
   override def convertToGpu(
       cpuExec: AppendDataExecV1,

delta-lake/delta-40x/src/main/scala/com/nvidia/spark/rapids/delta/delta40x/Delta40xProvider.scala

@@ -77,7 +77,7 @@ object Delta40xProvider extends DeltaProviderBase {
       protocol = fmt.protocol,
       metadata = fmt.metadata,
       nullableRowTrackingFields = false,
-      optimizationsEnabled = false, // we don't support splits and predicate pushdown yet
+      fmt.optimizationsEnabled,
       tablePath = fmt.tablePath,
       isCDCRead = fmt.isCDCRead)
 

integration_tests/src/main/python/delta_lake_delete_test.py

@@ -167,6 +167,7 @@ def read_parquet_sql(data_path):
 @pytest.mark.skipif(not supports_delta_lake_deletion_vectors() or is_before_spark_353(), \
                     reason="Deletion vectors new in Delta Lake 2.4 / Apache Spark 3.4")
 @pytest.mark.parametrize("reader_type", ["PERFILE", "COALESCING", "MULTITHREADED"])
+@pytest.mark.xfail(reason="https://github.com/NVIDIA/spark-rapids/issues/14004")
 def test_delta_deletion_vector_read_drop_row_group(spark_tmp_path, reader_type):
     data_path = spark_tmp_path + "/DELTA_DATA"
     def setup_tables(spark):

integration_tests/src/main/python/delta_lake_test.py

@@ -81,6 +81,7 @@ def setup_tables(spark):
     assert_gpu_and_cpu_are_equal_collect(
         lambda spark: spark.sql("SELECT * FROM delta.`{}`".format(data_path)))
 
+
 @allow_non_gpu("FileSourceScanExec", "ColumnarToRowExec", *delta_meta_allow)
 @delta_lake
 @ignore_order(local=True)
@@ -101,6 +102,100 @@ def setup_tables(spark):
         lambda spark: spark.sql("SELECT * FROM delta.`{}`".format(data_path)),
         conf=conf)
 
+
+def do_test_scan_split(spark_tmp_path, enable_deletion_vectors, expected_num_partitions, post_setup_table_func=None):
+    import os
+    import math
+
+    data_path = spark_tmp_path + "/DELTA_DATA"
+    num_rows = 2048
+    def setup_tables(spark):
+        setup_delta_dest_table(spark, data_path,
+                               dest_table_func=lambda spark: unary_op_df(spark, int_gen, length=num_rows, num_slices=1),
+                               use_cdf=False, enable_deletion_vectors=enable_deletion_vectors)
+        if post_setup_table_func:
+            post_setup_table_func(spark, data_path)
+    target_num_row_groups = 2
+    row_group_size = int(num_rows * 4 / target_num_row_groups) # num_rows * 4 bytes per int / target_num_row_groups
+    conf = {"parquet.block.size": str(row_group_size)}
+    with_cpu_session(setup_tables, conf)
+    # Verify that we have 1 file with 2 row groups
+    def verify_files_and_row_groups():
+        # list files in data_path
+        files = [f for f in os.listdir(data_path) if f.endswith(".parquet")]
+        files = [f"{data_path}/{f}" for f in files]
+        # find the most recently modified parquet file
+        most_recent_file = max(files, key=os.path.getmtime)
+        parquet_file = most_recent_file
+
+        import pyarrow.parquet as pq
+        metadata = pq.read_metadata(parquet_file)
+        assert metadata.num_row_groups == target_num_row_groups, f"Expected {target_num_row_groups} row groups in the parquet"
+        return parquet_file
+    data_file = verify_files_and_row_groups()
+    file_size = os.path.getsize(data_file)
+
+    conf = {"spark.sql.files.maxPartitionBytes": str(math.ceil(file_size/2.0))}
+
+    def get_num_partitions(spark):
+        df = spark.sql("SELECT * from delta.`{}`".format(data_path))
+        return df.rdd.getNumPartitions()
+    num_partitions = with_gpu_session(get_num_partitions, conf=conf)
+    assert num_partitions == expected_num_partitions, f"Expected {expected_num_partitions} partitions for split read"
+
+
+@allow_non_gpu(*delta_meta_allow)
+@delta_lake
+@pytest.mark.skipif(is_databricks_runtime(),
+                    reason="Scan split works differently on Databricks")
+def test_delta_scan_split_with_no_dv(spark_tmp_path):
+    do_test_scan_split(spark_tmp_path, enable_deletion_vectors=False, expected_num_partitions=2)
+
+
+@allow_non_gpu(*delta_meta_allow)
+@delta_lake
+@pytest.mark.skipif(is_databricks_runtime(),
+                    reason="Deletion vector scan is not supported on Databricks")
+def test_delta_scan_split_with_DV_enabled_with_no_DV(spark_tmp_path):
+    do_test_scan_split(spark_tmp_path, enable_deletion_vectors=True, expected_num_partitions=2)
+
+
+@allow_non_gpu(*delta_meta_allow)
+@delta_lake
+@pytest.mark.skipif(is_databricks_runtime(),
+                    reason="Deletion vector scan is not supported on Databricks")
+def test_delta_scan_split_with_DV_enabled_with_DVs(spark_tmp_path):
+    def do_delete(spark, data_path):
+        num_deleted = spark.sql(f"DELETE FROM delta.`{data_path}` WHERE a = 0").collect()[0][0]
+        assert num_deleted > 0, "Expected some rows to be deleted"
+    do_test_scan_split(spark_tmp_path, enable_deletion_vectors=True, expected_num_partitions=1, post_setup_table_func=do_delete)
+
+
+@allow_non_gpu(*delta_meta_allow)
+@delta_lake
+@pytest.mark.skipif(is_databricks_runtime(),
+                    reason="Deletion vector scan is not supported on Databricks")
+def test_delta_scan_split_with_DV_disabled_with_DVs(spark_tmp_path):
+    def do_delete_and_disable_DV(spark, data_path):
+        num_deleted = spark.sql(f"DELETE FROM delta.`{data_path}` WHERE a = 0").collect()[0][0]
+        assert num_deleted > 0, "Expected some rows to be deleted"
+        spark.sql(f"ALTER TABLE delta.`{data_path}` SET TBLPROPERTIES " +
+                  "('delta.enableDeletionVectors' = 'false')")
+    do_test_scan_split(spark_tmp_path, enable_deletion_vectors=True, expected_num_partitions=1, post_setup_table_func=do_delete_and_disable_DV)
+
+
+@allow_non_gpu(*delta_meta_allow)
+@delta_lake
+@pytest.mark.skipif(is_databricks_runtime(),
+                    reason="Deletion vector scan is not supported on Databricks")
+def test_delta_scan_split_with_DV_enabled_after_DVs_materialized(spark_tmp_path):
+    def do_delete_and_reorg(spark, data_path):
+        num_deleted = spark.sql(f"DELETE FROM delta.`{data_path}` WHERE a = 0").collect()[0][0]
+        assert num_deleted > 0, "Expected some rows to be deleted"
+        spark.sql(f"REORG table delta.`{data_path}` APPLY (PURGE)") # will rewrite files to purge soft-deleted data
+    do_test_scan_split(spark_tmp_path, enable_deletion_vectors=True, expected_num_partitions=2, post_setup_table_func=do_delete_and_reorg)
+
+
 # ID mapping is supported starting in Delta Lake 2.2, but currently cannot distinguish
 # Delta Lake 2.1 from 2.2 in tests. https://github.com/NVIDIA/spark-rapids/issues/9276
 column_mappings = ["name"]