From 628e69e832f877664d716ded4e8545876db27ec3 Mon Sep 17 00:00:00 2001
From: Zouxxyy <zouxinyu.zxy@alibaba-inc.com>
Date: Sat, 21 Dec 2024 10:45:30 +0800
Subject: [PATCH] [hive] Ignore path comparison when creating external table
 (#4747)

---
 .../apache/paimon/schema/SchemaManager.java   |  2 ++
 .../sql/DDLWithHiveCatalogTestBase.scala      | 21 +++++++++++++++++++
 2 files changed, 23 insertions(+)

diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java
index 325e0c392d12..9edbe901b6fa 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java
@@ -262,7 +262,9 @@ private void checkSchemaForExternalTable(Schema existsSchema, Schema newSchema)
             Map<String, String> newOptions = newSchema.options();
             newOptions.forEach(
                     (key, value) -> {
+                        // ignore `owner` and `path`
                         if (!key.equals(Catalog.OWNER_PROP)
+                                && !key.equals(CoreOptions.PATH.key())
                                 && (!existsOptions.containsKey(key)
                                         || !existsOptions.get(key).equals(value))) {
                             throw new RuntimeException(
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DDLWithHiveCatalogTestBase.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DDLWithHiveCatalogTestBase.scala
index 5311c586a59c..179020ae5b74 100644
--- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DDLWithHiveCatalogTestBase.scala
+++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DDLWithHiveCatalogTestBase.scala
@@ -553,6 +553,27 @@ abstract class DDLWithHiveCatalogTestBase extends PaimonHiveTestBase {
     }
   }
 
+  test("Paimon DDL with hive catalog: create external table on managed table location") {
+    Seq(sparkCatalogName, paimonHiveCatalogName).foreach {
+      catalogName =>
+        spark.sql(s"USE $catalogName")
+        withDatabase("paimon_db") {
+          spark.sql(s"CREATE DATABASE IF NOT EXISTS paimon_db")
+          spark.sql(s"USE paimon_db")
+          withTable("external_tbl", "managed_tbl") {
+            spark.sql(s"CREATE TABLE managed_tbl (id INT) USING paimon")
+            spark.sql("INSERT INTO managed_tbl VALUES (1)")
+            checkAnswer(spark.sql("SELECT * FROM managed_tbl"), Row(1))
+
+            val tablePath = loadTable("paimon_db", "managed_tbl").location().toString
+            spark.sql(s"CREATE TABLE external_tbl (id INT) USING paimon LOCATION '$tablePath'")
+            checkAnswer(spark.sql("SELECT * FROM external_tbl"), Row(1))
+            assert(loadTable("paimon_db", "external_tbl").location().toString.equals(tablePath))
+          }
+        }
+    }
+  }
+
   test("Paimon DDL with hive catalog: case sensitive") {
     Seq(sparkCatalogName, paimonHiveCatalogName).foreach {
       catalogName =>