(paimon) branch master updated: [spark] Fix remove_unexisting_files on empty table (#8697)

[lz...@apache.org]

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 7298f93ebb [spark] Fix remove_unexisting_files on empty table (#8697)
7298f93ebb is described below

commit 7298f93ebba87fd0f11d88f2b15ea5365276a663
Author: huangxiaoping <17...@qq.com>
AuthorDate: Thu Jul 16 22:05:28 2026 +0800

    [spark] Fix remove_unexisting_files on empty table (#8697)
---
 .../spark/procedure/SparkRemoveUnexistingFiles.scala     |  7 ++++++-
 .../procedure/RemoveUnexistingFilesProcedureTest.scala   | 16 ++++++++++++++++
 2 files changed, 22 insertions(+), 1 deletion(-)

diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/procedure/SparkRemoveUnexistingFiles.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/procedure/SparkRemoveUnexistingFiles.scala
index 1c27fc1d2a..cf55668737 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/procedure/SparkRemoveUnexistingFiles.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/procedure/SparkRemoveUnexistingFiles.scala
@@ -27,6 +27,7 @@ import org.apache.paimon.table.FileStoreTable
 import org.apache.paimon.table.sink.{CommitMessage, CommitMessageImpl, CommitMessageSerializer}
 
 import org.apache.spark.internal.Logging
+import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.{PaimonSparkSession, SparkSession}
 import org.apache.spark.sql.catalyst.SQLConfHelper
 
@@ -43,8 +44,12 @@ case class SparkRemoveUnexistingFiles(
   extends SQLConfHelper
   with Logging {
 
-  private def buildRDD() = {
+  private def buildRDD(): RDD[String] = {
     val binaryPartitions = table.newScan().listPartitions()
+    if (binaryPartitions.isEmpty) {
+      return spark.sparkContext.emptyRDD[String]
+    }
+
     val realParallelism = Math.min(binaryPartitions.size(), parallelism)
 
     val numPartitionFields = table.schema().partitionKeys().size()
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/RemoveUnexistingFilesProcedureTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/RemoveUnexistingFilesProcedureTest.scala
index b6b72806d1..afa7963182 100644
--- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/RemoveUnexistingFilesProcedureTest.scala
+++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/RemoveUnexistingFilesProcedureTest.scala
@@ -25,6 +25,21 @@ import java.util.UUID
 
 class RemoveUnexistingFilesProcedureTest extends PaimonSparkTestBase {
 
+  test("Paimon procedure: remove unexisting files on empty table") {
+    spark.sql("CREATE DATABASE IF NOT EXISTS mydb")
+    try {
+      spark.sql("CREATE TABLE mydb.t_empty (id INT) USING paimon")
+
+      checkAnswer(
+        spark.sql("CALL sys.remove_unexisting_files(table => 'mydb.t_empty', dry_run => true)"),
+        Nil)
+
+      checkAnswer(spark.sql("CALL sys.remove_unexisting_files(table => 'mydb.t_empty')"), Nil)
+    } finally {
+      spark.sql("DROP TABLE IF EXISTS mydb.t_empty")
+    }
+  }
+
   test("Paimon procedure: remove unexisting files, bucket = -1") {
     testImpl(-1)
   }
@@ -51,6 +66,7 @@ class RemoveUnexistingFilesProcedureTest extends PaimonSparkTestBase {
 
     val actual = new Array[Int](numPartitions)
     val pattern = "pt=(\\d+?)/".r
+    spark.sql("CREATE DATABASE IF NOT EXISTS mydb")
     spark.sql(s"USE mydb")
     spark
       .sql(s"CALL sys.remove_unexisting_files(table => '$tableName', dry_run => true)")