[spark] Prevent table deletion for self referencing RTAS

paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/spark/sql/execution/shim/PaimonReplaceTableAsSelectStrategy.scala

@@ -84,6 +84,7 @@ case class PaimonReplaceTableAsSelectStrategy(spark: SparkSession)
           invalidateCache
         ) :: Nil
       } else {
+        checkNonAtomicSelfReference(catalog, ident, analyzedQuery.get)
         ReplaceTableAsSelectExec(
           catalog,
           ident,

paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/execution/PaimonTableAsSelectHelper.scala

@@ -119,6 +119,24 @@ object PaimonTableAsSelectHelper {
     }
   }
 
+  def checkNonAtomicSelfReference(
+      catalog: TableCatalog,
+      ident: Identifier,
+      query: LogicalPlan): Unit = {
+    val referencesTarget = query.exists {
+      case r: DataSourceV2Relation =>
+        r.catalog.contains(catalog) && r.identifier.contains(ident)
+      case _ => false
+    }
+    if (referencesTarget) {
+      throw new UnsupportedOperationException(
+        s"Cannot replace table $ident because the replacement query reads from the same table " +
+          "and the requested table definition requires a non-atomic drop-and-create. " +
+          "Write the query result to a temporary table first, or keep the existing provider, " +
+          "table type, and partitioning.")
+    }
+  }
+
   /**
    * Rewrite to OverwriteByExpression or OverwritePartitionsDynamic for an existing table,
    * preserving table definition. Returns None if the table does not exist.

paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/execution/shim/PaimonReplaceTableAsSelectStrategy.scala

@@ -72,6 +72,7 @@ case class PaimonReplaceTableAsSelectStrategy(spark: SparkSession)
           finalWriteOptions,
           orCreate = orCreate) :: Nil
       } else {
+        checkNonAtomicSelfReference(catalog, ident, query)
         SparkShimLoader.shim.createReplaceTableAsSelectExec(
           catalog,
           ident,

paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DDLTestBase.scala

@@ -386,6 +386,57 @@ abstract class DDLTestBase extends PaimonSparkTestBase {
     }
   }
 
+  test("Paimon DDL: self-referencing RTAS on a partitioned table keeps the table") {
+    assume(gteqSpark3_4)
+    withTable("q") {
+      sql("CREATE TABLE q (id INT, dt STRING) USING paimon PARTITIONED BY (dt)")
+      sql("INSERT INTO q VALUES (1, 'a'), (2, 'b'), (3, 'c')")
+
+      val location = loadTable("q").location()
+      val fileIO = loadTable("q").fileIO()
+
+      val e = intercept[RuntimeException] {
+        sql("REPLACE TABLE q AS SELECT * FROM q WHERE dt = 'a'")
+      }
+      val messages = Iterator
+        .iterate(e: Throwable)(_.getCause)
+        .takeWhile(_ != null)
+        .map(t => String.valueOf(t.getMessage))
+        .mkString(" | ")
+      Assertions.assertTrue(messages.contains("Cannot replace table"), messages)
+
+      Assertions.assertTrue(sql("SHOW TABLES").collect().exists(_.getString(1) == "q"))
+      Assertions.assertTrue(fileIO.exists(location))
+      checkAnswer(
+        sql("SELECT * FROM q ORDER BY id"),
+        Row(1, "a") :: Row(2, "b") :: Row(3, "c") :: Nil)
+    }
+  }
+
+  test("Paimon DDL: self-referencing RTAS restating partitioning replaces in place") {
+    assume(gteqSpark3_4)
+    withTable("q") {
+      sql("CREATE TABLE q (id INT, dt STRING) USING paimon PARTITIONED BY (dt)")
+      sql("INSERT INTO q VALUES (1, 'a'), (2, 'b'), (3, 'c')")
+
+      sql("REPLACE TABLE q PARTITIONED BY (dt) AS SELECT * FROM q WHERE dt = 'a'")
+
+      checkAnswer(sql("SELECT * FROM q"), Row(1, "a") :: Nil)
+    }
+  }
+
+  test("Paimon DDL: self-referencing RTAS on an unpartitioned table replaces in place") {
+    assume(gteqSpark3_4)
+    withTable("q2") {
+      sql("CREATE TABLE q2 (id INT, dt STRING) USING paimon")
+      sql("INSERT INTO q2 VALUES (1, 'a'), (2, 'b'), (3, 'c')")
+
+      sql("REPLACE TABLE q2 AS SELECT * FROM q2 WHERE dt = 'a'")
+
+      checkAnswer(sql("SELECT * FROM q2"), Row(1, "a") :: Nil)
+    }
+  }
+
   test("Paimon DDL: CREATE OR REPLACE TABLE AS SELECT supports incompatible schema") {
     assume(gteqSpark3_4)
     withTable("t") {