Spark 3.5, 4.0, 4.1: Fix view rename target namespace

spark/v3.5/spark-extensions/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ExtendedDataSourceV2Strategy.scala

@@ -127,12 +127,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 AnalysisException(
-          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
@@ -192,6 +188,23 @@ case class ExtendedDataSourceV2Strategy(spark: SparkSession) extends Strategy wi
     new GenericInternalRow(values)
   }
 
+  private def resolveViewRenameTarget(
+      sourceCatalog: ViewCatalog,
+      targetName: Seq[String]): Identifier = {
+    if (targetName.length == 1) {
+      // An unqualified target keeps the source view's namespace
+      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 AnalysisException(
+          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/v3.5/spark-extensions/src/main/scala/org/apache/spark/sql/execution/datasources/v2/RenameV2ViewExec.scala

@@ -29,7 +29,13 @@ 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)
+    val qualifiedNewIdent = if (newIdent.namespace.isEmpty) {
+      Identifier.of(oldIdent.namespace, newIdent.name)
+    } else {
+      newIdent
+    }
+
+    catalog.renameView(oldIdent, qualifiedNewIdent)
 
     Seq.empty
   }

spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestViews.java

@@ -777,6 +777,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);

spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/Spark3Util.java

@@ -787,6 +787,9 @@ public static CatalogAndIdentifier catalogAndIdentifier(
    * A modified version of Spark's LookupCatalog.CatalogAndIdentifier.unapply Attempts to find the
    * catalog and identifier a multipart identifier represents
    *
+   * <p>A single-part name is resolved in the session's current namespace, or in the default
+   * namespace of {@code defaultCatalog} if it is not the current catalog.
+   *
    * @param spark Spark session to use for resolution
    * @param nameParts Multipart identifier representing a table
    * @param defaultCatalog Catalog to use if none is specified

spark/v4.0/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,23 @@ case class ExtendedDataSourceV2Strategy(spark: SparkSession) extends Strategy wi
     case _ => Nil
   }
 
+  private def resolveViewRenameTarget(
+      sourceCatalog: ViewCatalog,
+      targetName: Seq[String]): Identifier = {
+    if (targetName.length == 1) {
+      // An unqualified target keeps the source view's namespace
+      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.0/spark-extensions/src/main/scala/org/apache/spark/sql/execution/datasources/v2/RenameV2ViewExec.scala

@@ -29,7 +29,13 @@ 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)
+    val qualifiedNewIdent = if (newIdent.namespace.isEmpty) {
+      Identifier.of(oldIdent.namespace, newIdent.name)
+    } else {
+      newIdent
+    }
+
+    catalog.renameView(oldIdent, qualifiedNewIdent)
 
     Seq.empty
   }

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

@@ -776,6 +776,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);

spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/Spark3Util.java

@@ -791,6 +791,9 @@ public static CatalogAndIdentifier catalogAndIdentifier(
    * A modified version of Spark's LookupCatalog.CatalogAndIdentifier.unapply Attempts to find the
    * catalog and identifier a multipart identifier represents
    *
+   * <p>A single-part name is resolved in the session's current namespace, or in the default
+   * namespace of {@code defaultCatalog} if it is not the current catalog.
+   *
    * @param spark Spark session to use for resolution
    * @param nameParts Multipart identifier representing a table
    * @param defaultCatalog Catalog to use if none is specified

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,23 @@ case class ExtendedDataSourceV2Strategy(spark: SparkSession) extends Strategy wi
     case _ => Nil
   }
 
+  private def resolveViewRenameTarget(
+      sourceCatalog: ViewCatalog,
+      targetName: Seq[String]): Identifier = {
+    if (targetName.length == 1) {
+      // An unqualified target keeps the source view's namespace
+      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,13 @@ 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)
+    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

@@ -776,6 +776,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);

spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/Spark3Util.java

@@ -838,6 +838,9 @@ public static CatalogAndIdentifier catalogAndIdentifier(
    * A modified version of Spark's LookupCatalog.CatalogAndIdentifier.unapply Attempts to find the
    * catalog and identifier a multipart identifier represents
    *
+   * <p>A single-part name is resolved in the session's current namespace, or in the default
+   * namespace of {@code defaultCatalog} if it is not the current catalog.
+   *
    * @param spark Spark session to use for resolution
    * @param nameParts Multipart identifier representing a table
    * @param defaultCatalog Catalog to use if none is specified