(paimon) branch master updated: [spark] Close the commit after truncating a table or partitions (#9900)

[lzljs3620320]

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

JingsongLi 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 b38d2a81af [spark] Close the commit after truncating a table or 
partitions (#9900)
b38d2a81af is described below

commit b38d2a81af9cccbeca7983a48a292930a83e3718
Author: cxzl25 <3898450+cxz...@users.noreply.github.com>
AuthorDate: Fri Oct 16 13:08:22 2026 +0800

    [spark] Close the commit after truncating a table or partitions (#9900)
---
 .../apache/paimon/spark/PaimonSparkTableBase.scala |  6 ++-
 .../TruncatePaimonTableWithFilterExec.scala        | 53 ++++++++++++----------
 2 files changed, 33 insertions(+), 26 deletions(-)

diff --git 
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSparkTableBase.scala
 
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSparkTableBase.scala
index d42d129925..d139ed025b 100644
--- 
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSparkTableBase.scala
+++ 
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSparkTableBase.scala
@@ -186,7 +186,11 @@ abstract class PaimonSparkTableBase(val table: Table)
 
   def truncateTable: Boolean = {
     val commit = table.newBatchWriteBuilder().newCommit()
-    commit.truncateTable()
+    try {
+      commit.truncateTable()
+    } finally {
+      commit.close()
+    }
     true
   }
 }
diff --git 
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/TruncatePaimonTableWithFilterExec.scala
 
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/TruncatePaimonTableWithFilterExec.scala
index 41c2187ce3..3093d15e84 100644
--- 
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/TruncatePaimonTableWithFilterExec.scala
+++ 
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/TruncatePaimonTableWithFilterExec.scala
@@ -37,31 +37,34 @@ case class TruncatePaimonTableWithFilterExec(
 
   override def run(): Seq[InternalRow] = {
     val commit = table.newBatchWriteBuilder().newCommit()
-
-    partitionPredicate match {
-      case Some(p) =>
-        table match {
-          case fileStoreTable: FileStoreTable =>
-            val matchedPartitions =
-              
fileStoreTable.newSnapshotReader().withPartitionFilter(p).partitions().asScala
-            if (matchedPartitions.nonEmpty) {
-              val partitionComputer = new InternalRowPartitionComputer(
-                fileStoreTable.coreOptions().partitionDefaultName(),
-                fileStoreTable.schema().logicalPartitionType(),
-                fileStoreTable.partitionKeys.asScala.toArray,
-                fileStoreTable.coreOptions().legacyPartitionName()
-              )
-              val dropPartitions =
-                
matchedPartitions.map(partitionComputer.generatePartValues(_).asScala.asJava)
-              commit.truncatePartitions(dropPartitions.asJava)
-            } else {
-              commit.commit(JCollections.emptyList())
-            }
-          case _ =>
-            throw new UnsupportedOperationException("Unsupported truncate 
table")
-        }
-      case _ =>
-        commit.truncateTable()
+    try {
+      partitionPredicate match {
+        case Some(p) =>
+          table match {
+            case fileStoreTable: FileStoreTable =>
+              val matchedPartitions =
+                
fileStoreTable.newSnapshotReader().withPartitionFilter(p).partitions().asScala
+              if (matchedPartitions.nonEmpty) {
+                val partitionComputer = new InternalRowPartitionComputer(
+                  fileStoreTable.coreOptions().partitionDefaultName(),
+                  fileStoreTable.schema().logicalPartitionType(),
+                  fileStoreTable.partitionKeys.asScala.toArray,
+                  fileStoreTable.coreOptions().legacyPartitionName()
+                )
+                val dropPartitions =
+                  
matchedPartitions.map(partitionComputer.generatePartValues(_).asScala.asJava)
+                commit.truncatePartitions(dropPartitions.asJava)
+              } else {
+                commit.commit(JCollections.emptyList())
+              }
+            case _ =>
+              throw new UnsupportedOperationException("Unsupported truncate 
table")
+          }
+        case _ =>
+          commit.truncateTable()
+      }
+    } finally {
+      commit.close()
     }
     Nil
   }