Spark 4.1: Fix view rename target namespace

spark/v4.1/spark-extensions/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ExtendedDataSourceV2Strategy.scala

@@ -119,12 +119,8 @@ case class ExtendedDataSourceV2Strategy(spark: SparkSession) extends Strategy wi
       OrderAwareCoalesceExec(numPartitions, coalescer, planLater(child)) :: Nil
 
     case RenameTable(ResolvedV2View(oldCatalog: ViewCatalog, oldIdent), newName, isView @ true) =>
-      val newIdent = Spark3Util.catalogAndIdentifier(spark, newName.toList.asJava)
-      if (oldCatalog.name != newIdent.catalog().name()) {
-        throw new IcebergAnalysisException(
-          s"Cannot move view between catalogs: from=${oldCatalog.name} and to=${newIdent.catalog().name()}")
-      }
-      RenameV2ViewExec(oldCatalog, oldIdent, newIdent.identifier()) :: Nil
+      val targetIdent = resolveViewRenameTarget(oldCatalog, newName)
+      RenameV2ViewExec(oldCatalog, oldIdent, targetIdent) :: Nil
 
     case DropIcebergView(ResolvedIdentifier(viewCatalog: ViewCatalog, ident), ifExists) =>
       DropV2ViewExec(viewCatalog, ident, ifExists) :: Nil
@@ -176,6 +172,22 @@ case class ExtendedDataSourceV2Strategy(spark: SparkSession) extends Strategy wi
     case _ => Nil
   }
 
+  private def resolveViewRenameTarget(
+      sourceCatalog: ViewCatalog,
+      targetName: Seq[String]): Identifier = {
+    if (targetName.length == 1) {
+      Identifier.of(Array.empty[String], targetName.head)
+    } else {
+      val target = Spark3Util.catalogAndIdentifier(spark, targetName.toList.asJava, sourceCatalog)
+      if (sourceCatalog.name != target.catalog().name()) {
+        throw new IcebergAnalysisException(
+          s"Cannot move view between catalogs: from=${sourceCatalog.name} and to=${target.catalog().name()}")
+      }
+
+      target.identifier()
+    }
+  }
+
   private object IcebergCatalogAndIdentifier {
     def unapply(identifier: Seq[String]): Option[(TableCatalog, Identifier)] = {
       val catalogAndIdentifier = Spark3Util.catalogAndIdentifier(spark, identifier.asJava)

spark/v4.1/spark-extensions/src/main/scala/org/apache/spark/sql/execution/datasources/v2/RenameV2ViewExec.scala

@@ -29,7 +29,14 @@ case class RenameV2ViewExec(catalog: ViewCatalog, oldIdent: Identifier, newIdent
   override lazy val output: Seq[Attribute] = Nil
 
   override protected def run(): Seq[InternalRow] = {
-    catalog.renameView(oldIdent, newIdent)
+    // Match Spark's v2 rename behavior: an unqualified target renames in place.
+    val qualifiedNewIdent =
+      if (newIdent.namespace().isEmpty) {
+        Identifier.of(oldIdent.namespace(), newIdent.name())
+      } else {
+        newIdent
+      }
+    catalog.renameView(oldIdent, qualifiedNewIdent)
 
     Seq.empty
   }

spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestViews.java

@@ -774,6 +774,34 @@ public void renameView() throws NoSuchTableException {
         .containsExactlyInAnyOrderElementsOf(expected);
   }
 
+  @TestTemplate
+  public void renameQualifiedViewToUnqualifiedTargetKeepsSourceNamespace() {
+    Namespace sourceNamespace = Namespace.of(viewName("rename_source_ns"));
+    Namespace currentNamespace = Namespace.of(viewName("rename_current_ns"));
+    String viewName = viewName("originalView");
+    String renamedView = viewName("renamedView");
+    String sql = String.format("SELECT id FROM %s.%s.%s", catalogName, NAMESPACE, tableName);
+
+    sql("CREATE NAMESPACE IF NOT EXISTS %s.%s", catalogName, sourceNamespace);
+    sql("CREATE NAMESPACE IF NOT EXISTS %s.%s", catalogName, currentNamespace);
+
+    ViewCatalog viewCatalog = viewCatalog();
+    viewCatalog
+        .buildView(TableIdentifier.of(sourceNamespace, viewName))
+        .withQuery("spark", sql)
+        .withDefaultNamespace(sourceNamespace)
+        .withDefaultCatalog(catalogName)
+        .withSchema(schema(sql))
+        .create();
+
+    sql("USE %s.%s", catalogName, currentNamespace);
+    sql("ALTER VIEW %s.%s RENAME TO %s", sourceNamespace, viewName, renamedView);
+
+    assertThat(viewCatalog.viewExists(TableIdentifier.of(sourceNamespace, viewName))).isFalse();
+    assertThat(viewCatalog.viewExists(TableIdentifier.of(sourceNamespace, renamedView))).isTrue();
+    assertThat(viewCatalog.viewExists(TableIdentifier.of(currentNamespace, renamedView))).isFalse();
+  }
+
   @TestTemplate
   public void renameViewHiddenByTempView() throws NoSuchTableException {
     insertRows(10);