fix

Author: JingsongLi

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonCommand.scala

@@ -276,7 +276,7 @@ trait PaimonCommand extends WithFileStoreTable with ExpressionHelper with SQLCon
             relativeBucketPath,
             SerializationUtils.serializeBinaryRow(partition),
             bucket,
-            filePath,
+            new Path(filePath).getName,
             DeletionVector.serializeToBytes(dv)
           )
       }

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala

@@ -308,7 +308,7 @@ case class PaimonSparkWriter(table: FileStoreTable) {
             }
 
             dvIndexFileMaintainer.notifyNewDeletionVector(
-              new Path(sdv.dataFilePath).getName,
+              sdv.dataFileName,
               DeletionVector.deserializeFromBytes(sdv.deletionVector))
           }
           val indexEntries = dvIndexFileMaintainer.persist()

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/SparkDeletionVector.scala

@@ -32,16 +32,25 @@ case class SparkDeletionVector(
     partitionAndBucket: String,
     partition: Array[Byte],
     bucket: Int,
-    dataFilePath: String,
+    dataFileName: String,
     deletionVector: Array[Byte]
-)
+) {
+  def relativePath(pathFactory: FileStorePathFactory): String = {
+    val prefix = pathFactory
+      .relativeBucketPath(SerializationUtils.deserializeBinaryRow(partition), bucket)
+      .toUri
+      .toString + "/"
+    prefix + dataFileName
+  }
+}
 
 object SparkDeletionVector {
   def toDataSplit(
       deletionVector: SparkDeletionVector,
       root: Path,
+      pathFactory: FileStorePathFactory,
       dataFilePathToMeta: Map[String, SparkDataFileMeta]): DataSplit = {
-    val meta = dataFilePathToMeta(deletionVector.dataFilePath)
+    val meta = dataFilePathToMeta(deletionVector.relativePath(pathFactory))
     DataSplit
       .builder()
       .withBucketPath(root + "/" + deletionVector.partitionAndBucket)

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/UpdatePaimonTableCommand.scala

@@ -82,6 +82,7 @@ case class UpdatePaimonTableCommand(
       logDebug("No file need to rewrote. It's an empty Commit.")
       Seq.empty[CommitMessage]
     } else {
+      val pathFactory = fileStore.pathFactory()
       if (deletionVectorsEnabled) {
         // Step2: collect all the deletion vectors that marks the deleted rows.
         val deletionVectors = collectDeletionVectors(
@@ -95,7 +96,7 @@ case class UpdatePaimonTableCommand(
         try {
           // Step3: write these updated data
           val touchedDataSplits = deletionVectors.collect().map {
-            SparkDeletionVector.toDataSplit(_, root, dataFilePathToMeta)
+            SparkDeletionVector.toDataSplit(_, root, pathFactory, dataFilePathToMeta)
           }
           val addCommitMessage = writeOnlyUpdatedData(sparkSession, touchedDataSplits)