(paimon) branch master updated: [spark] Delete with deletion vectors should enable AQE (#4171)

[lzljs3620320]

This is an automated email from the ASF dual-hosted git repository.

lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git


The following commit(s) were added to refs/heads/master by this push:
     new 2fa598dee [spark] Delete with deletion vectors should enable AQE 
(#4171)
2fa598dee is described below

commit 2fa598dee9fa54d3324be8fa5b8fa3eff669b0c1
Author: Xiduo You <ulyssesyo...@gmail.com>
AuthorDate: Thu Sep 12 10:18:34 2024 +0800

    [spark] Delete with deletion vectors should enable AQE (#4171)
---
 .../commands/DeleteFromPaimonTableCommand.scala    | 23 ++++++++++------------
 1 file changed, 10 insertions(+), 13 deletions(-)

diff --git 
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DeleteFromPaimonTableCommand.scala
 
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DeleteFromPaimonTableCommand.scala
index cc440dd5c..2b3888911 100644
--- 
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DeleteFromPaimonTableCommand.scala
+++ 
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DeleteFromPaimonTableCommand.scala
@@ -117,24 +117,21 @@ case class DeleteFromPaimonTableCommand(
   }
 
   def performNonPrimaryKeyDelete(sparkSession: SparkSession): 
Seq[CommitMessage] = {
-    val pathFactory = fileStore.pathFactory()
     // Step1: the candidate data splits which are filtered by Paimon Predicate.
     val candidateDataSplits = findCandidateDataSplits(condition, 
relation.output)
     val dataFilePathToMeta = candidateFileMap(candidateDataSplits)
 
     if (deletionVectorsEnabled) {
-      withSQLConf("spark.sql.adaptive.enabled" -> "false") {
-        // Step2: collect all the deletion vectors that marks the deleted rows.
-        val deletionVectors = collectDeletionVectors(
-          candidateDataSplits,
-          dataFilePathToMeta,
-          condition,
-          relation,
-          sparkSession)
-
-        // Step3: update the touched deletion vectors and index files
-        writer.persistDeletionVectors(deletionVectors)
-      }
+      // Step2: collect all the deletion vectors that marks the deleted rows.
+      val deletionVectors = collectDeletionVectors(
+        candidateDataSplits,
+        dataFilePathToMeta,
+        condition,
+        relation,
+        sparkSession)
+
+      // Step3: update the touched deletion vectors and index files
+      writer.persistDeletionVectors(deletionVectors)
     } else {
       // Step2: extract out the exactly files, which must have at least one 
record to be updated.
       val touchedFilePaths =