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 ee9064dcc36b..9da1c75f69f9 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 a6774200e9ce..0e6411dcfab5 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 4bbcc0af25e2..83927507687c 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 2b5b21783759..70e38deecd80 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") {