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 887e5ed3fa [spark] Prevent table deletion for self referencing RTAS
(#9139)
887e5ed3fa is described below
commit 887e5ed3faf420192bb26e85f701c6dd95599d37
Author: Arnav Balyan <[email protected]>
AuthorDate: Thu Aug 13 19:51:29 2026 +0530
[spark] Prevent table deletion for self referencing RTAS (#9139)
---
.../shim/PaimonReplaceTableAsSelectStrategy.scala | 1 +
.../sql/execution/PaimonTableAsSelectHelper.scala | 18 ++++++++
.../shim/PaimonReplaceTableAsSelectStrategy.scala | 1 +
.../org/apache/paimon/spark/sql/DDLTestBase.scala | 51 ++++++++++++++++++++++
4 files changed, 71 insertions(+)
diff --git
a/paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/spark/sql/execution/shim/PaimonReplaceTableAsSelectStrategy.scala
b/paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/spark/sql/execution/shim/PaimonReplaceTableAsSelectStrategy.scala
index ee9064dcc3..9da1c75f69 100644
---
a/paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/spark/sql/execution/shim/PaimonReplaceTableAsSelectStrategy.scala
+++
b/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,
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/execution/PaimonTableAsSelectHelper.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/execution/PaimonTableAsSelectHelper.scala
index a6774200e9..0e6411dcfa 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/execution/PaimonTableAsSelectHelper.scala
+++
b/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.
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/execution/shim/PaimonReplaceTableAsSelectStrategy.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/execution/shim/PaimonReplaceTableAsSelectStrategy.scala
index 4bbcc0af25..8392750768 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/execution/shim/PaimonReplaceTableAsSelectStrategy.scala
+++
b/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,
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DDLTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DDLTestBase.scala
index 2b5b217837..70e38deecd 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DDLTestBase.scala
+++
b/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") {
